feat: option to disable cache and increase worker count

README.md

@@ -57,6 +57,27 @@ To use PostgreSQL, set `db_uri` or the `GOSSEBIT_DB_URI` environment variable to
 postgres://user:password@host:5432/db_name
 ```
 
+### Multiple Workers
+
+By default, gooseBit runs with a single worker process. Running multiple workers (e.g. to utilize more
+CPU cores) is supported if the following requirements are met:
+
+1. The in-memory object cache is disabled, since each worker would otherwise serve stale data
+   (set `cache: enabled: false` or `GOOSEBIT_CACHE__ENABLED=false`).
+2. `secret_key` is set explicitly. Otherwise, each worker generates its own random key and user sessions
+   signed by one worker fail on the others.
+3. PostgreSQL is recommended as the database. SQLite serializes writers via file locks, which is fine for
+   testing but not for concurrent load.
+4. All workers share the same artifacts storage. Workers on one host share the `artifacts_dir` volume;
+   deployments spanning multiple hosts should use the S3 storage backend.
+
+With the Docker image, the worker count can then be set at runtime, for example:
+
+```txt
+docker run -e GOOSEBIT_CACHE__ENABLED=false -e GOOSEBIT_SECRET_KEY=<key> \
+    -e GUNICORN_CMD_ARGS="--workers 4 --enable-stdio-inheritance" upstreamdata/goosebit
+```
+
 ### Artifact Storage
 
 The software packages managed by gooseBit are either stored on the local filesystem (`artifacts_dir` setting) or an S3-compatible object storage.

conftest.py

@@ -5,7 +5,6 @@
 from typing import Any, AsyncGenerator, Dict
 
 import pytest_asyncio
-from aiocache import caches
 from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
 from tortoise import Tortoise
@@ -17,7 +16,9 @@
 
 from goosebit import app  # noqa: E402
 from goosebit.auth.permissions import GOOSEBIT_PERMISSIONS  # noqa: E402
+from goosebit.cache import cache  # noqa: E402
 from goosebit.db.models import UpdateModeEnum, UpdateStateEnum  # noqa: E402
+from goosebit.device_manager import DeviceManager  # noqa: E402
 from goosebit.settings import PWD_CXT  # type: ignore[attr-defined]  # noqa: E402
 
 # Configure logging
@@ -35,7 +36,9 @@
 
 @pytest_asyncio.fixture(scope="function", autouse=True)
 async def clear_cache() -> AsyncGenerator[None, None]:
-    await caches.get("default").clear()
+    await cache.clear()
+    # class-level cache holding a row from the previous test's database
+    DeviceManager._hardware_default = None
     yield
 
 

docker/Dockerfile

@@ -18,7 +18,9 @@ EXPOSE 60053
 
 USER goosebit
 
-# We currently do not fully support multiple workers. For more information, see:
+# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false)
+# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the
+# worker count. For more information, see:
 # https://github.com/UpstreamDataInc/goosebit/issues/125
 ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance"
 

docker/dev.dockerfile

@@ -20,8 +20,9 @@ EXPOSE 60053
 
 USER goosebit
 
-# We currently do not fully support multiple workers. For more information, see:
-# https://github.com/UpstreamDataInc/goosebit/issues/125
+# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false)
+# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the
+# worker count.
 ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance"
 
 SHELL ["/bin/sh", "-c"]

goosebit.yaml

@@ -29,6 +29,11 @@ poll_time: 00:01:00
 # Limit the number of updates that run at the same time. Can be used to avoid overloading the update server.
 #max_concurrent_updates: 1000
 
+# In-memory cache for device/user objects. Must be disabled when running more than one worker
+# process, otherwise workers serve stale data. See https://github.com/UpstreamDataInc/goosebit/issues/125
+#cache:
+#  enabled: true
+
 # Whether to track the IP of the device when it polls.  Useful for debugging, but can be turned off for privacy.
 #track_device_ip: true
 

goosebit/cache.py

@@ -0,0 +1,45 @@
+from typing import Any
+
+from aiocache import SimpleMemoryCache
+from aiocache.serializers import PickleSerializer
+
+from goosebit.settings import config
+
+CACHE_TTL = 600
+
+
+class Cache:
+    """Object cache honoring the `cache.enabled` setting.
+
+    With caching disabled every lookup is a miss and writes are dropped, leaving the
+    database as the only source of truth. This is required when running multiple
+    workers, see https://github.com/UpstreamDataInc/goosebit/issues/125.
+    """
+
+    def __init__(self, enabled: bool):
+        self.enabled = enabled
+        self._backend = SimpleMemoryCache(serializer=PickleSerializer())
+
+    async def get(self, key: str) -> Any:
+        if not self.enabled:
+            return None
+        return await self._backend.get(key)
+
+    async def set(self, key: str, value: Any) -> None:
+        if not self.enabled:
+            return
+        await self._backend.set(key, value, ttl=CACHE_TTL)
+
+    async def delete(self, key: str) -> None:
+        if not self.enabled:
+            return
+        # missing keys are fine, the entry may have expired or was never cached by this worker
+        await self._backend.delete(key)
+
+    async def clear(self) -> None:
+        if not self.enabled:
+            return
+        await self._backend.clear()
+
+
+cache = Cache(enabled=config.cache.enabled)

goosebit/device_manager.py

@@ -5,9 +5,9 @@
 from enum import StrEnum
 from typing import Any, Awaitable, Callable, Optional
 
-from aiocache import caches
 from fastapi.requests import Request
 
+from goosebit.cache import cache
 from goosebit.db.models import (
     Device,
     Hardware,
@@ -18,16 +18,6 @@
 )
 from goosebit.schema.updates import UpdateChunk
 
-caches.set_config(
-    {
-        "default": {
-            "cache": "aiocache.SimpleMemoryCache",
-            "serializer": {"class": "aiocache.serializers.PickleSerializer"},
-            "ttl": 600,
-        },
-    }
-)
-
 
 class HandlingType(StrEnum):
     SKIP = "skip"
@@ -43,7 +33,6 @@ class DeviceManager:
 
     @staticmethod
     async def get_device(dev_id: str) -> Device:
-        cache = caches.get("default")
         device = await cache.get(dev_id)
         if device:
             return device  # type: ignore[no-any-return]
@@ -54,8 +43,7 @@ async def get_device(dev_id: str) -> Device:
             DeviceManager._hardware_default = hardware
 
         device = (await Device.get_or_create(id=dev_id, defaults={"hardware": hardware}))[0]
-        result = await cache.set(device.id, device, ttl=600)
-        assert result, "device being cached"
+        await cache.set(device.id, device)
 
         return device  # type: ignore[no-any-return]
 
@@ -64,8 +52,7 @@ async def save_device(device: Device, update_fields: list[str]) -> None:
         await device.save(update_fields=update_fields)
 
         # only update cache after a successful database save
-        result = await caches.get("default").set(device.id, device, ttl=600)
-        assert result, "device being cached"
+        await cache.set(device.id, device)
 
     @staticmethod
     async def update_auth_token(device: Device, auth_token: str) -> None:
@@ -254,8 +241,7 @@ async def update_log(device: Device, log_data: str) -> None:
     async def delete_devices(ids: list[str]) -> None:
         await Device.filter(id__in=ids).delete()
         for dev_id in ids:
-            result = await caches.get("default").delete(dev_id)
-            assert result == 1, "device has been cached"
+            await cache.delete(dev_id)
 
 
 async def get_device(dev_id: str) -> Device:

goosebit/settings/schema.py

@@ -1,5 +1,6 @@
 import os
 from enum import StrEnum
+from logging import getLogger
 from pathlib import Path
 from typing import Any
 
@@ -15,6 +16,14 @@
 from .const import CURRENT_DIR, GOOSEBIT_ROOT_DIR, LOGGING_DEFAULT
 
 
+def _generate_secret_key() -> OctKey:
+    getLogger(__name__).warning(
+        "No secret_key configured, generated a random one. User sessions will not survive restarts "
+        "and logins will fail intermittently when running multiple workers. Set GOOSEBIT_SECRET_KEY."
+    )
+    return OctKey.generate_key()
+
+
 class DeviceAuthMode(StrEnum):
     SETUP = "setup"  # setup mode, any devices polling with an auth token that don't have one will save it
     STRICT = "strict"  # all devices must have keys, and all keys must be set up with the API
@@ -70,6 +79,12 @@ class StorageSettings(BaseModel):
     s3: S3StorageSettings | None = None
 
 
+class CacheSettings(BaseModel):
+    # in-memory cache for device/user objects; must be disabled when running multiple workers,
+    # see https://github.com/UpstreamDataInc/goosebit/issues/125
+    enabled: bool = True
+
+
 class GooseBitSettings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="GOOSEBIT_", extra="ignore", env_nested_delimiter="__")
 
@@ -84,7 +99,7 @@ class GooseBitSettings(BaseSettings):
 
     device_auth: DeviceAuthSettings = DeviceAuthSettings()
 
-    secret_key: OctKey = Field(default_factory=OctKey.generate_key)
+    secret_key: OctKey = Field(default_factory=_generate_secret_key)
 
     plugins: list[str] = Field(default_factory=list)
 
@@ -95,6 +110,8 @@ class GooseBitSettings(BaseSettings):
 
     storage: StorageSettings = StorageSettings()
 
+    cache: CacheSettings = CacheSettings()
+
     metrics: MetricsSettings = MetricsSettings()
 
     logging: dict[str, Any] = LOGGING_DEFAULT

goosebit/updater/controller/v1/routes.py

@@ -9,8 +9,9 @@
     Response,
     StreamingResponse,
 )
+from tortoise.expressions import F
 
-from goosebit.db.models import Device, Software, UpdateStateEnum
+from goosebit.db.models import Device, Rollout, Software, UpdateStateEnum
 from goosebit.device_manager import DeviceManager, HandlingType, get_device
 from goosebit.settings import config
 from goosebit.storage import storage
@@ -159,8 +160,8 @@ async def deployment_feedback(
             rollout = await DeviceManager.get_rollout(device)
             if rollout:
                 if rollout.software == reported_software:
-                    rollout.success_count += 1
-                    await rollout.save()
+                    # atomic DB-side increment
+                    await Rollout.filter(id=rollout.id).update(success_count=F("success_count") + 1)
                 else:
                     # edge case where device update mode got changed while update was running
                     logging.warning(
@@ -180,8 +181,8 @@ async def deployment_feedback(
             rollout = await DeviceManager.get_rollout(device)
             if rollout:
                 if rollout.software == reported_software:
-                    rollout.failure_count += 1
-                    await rollout.save()
+                    # atomic DB-side increment
+                    await Rollout.filter(id=rollout.id).update(failure_count=F("failure_count") + 1)
                 else:
                     # edge case where device update mode got changed while update was running
                     logging.warning(

goosebit/users/__init__.py

@@ -1,6 +1,5 @@
-from aiocache import caches
-
 from goosebit.api.telemetry.metrics import users_count
+from goosebit.cache import cache
 from goosebit.db.models import User
 from goosebit.settings import PWD_CXT  # type: ignore[attr-defined]
 
@@ -19,8 +18,7 @@ async def save_user(user: User, update_fields: list[str]) -> None:
         await user.save(update_fields=update_fields)
 
         # only update cache after a successful database save
-        result = await caches.get("default").set(user.username, user, ttl=600)
-        assert result, "user being cached"
+        await cache.set(user.username, user)
 
     @staticmethod
     async def update_enabled(user: User, enabled: bool) -> None:
@@ -43,21 +41,19 @@ async def setup_user(cls, username: str, hashed_pwd: str, permissions: list[str]
 
     @staticmethod
     async def get_user(username: str) -> User:
-        cache = caches.get("default")
         user = await cache.get(username)
         if user:
             return user  # type: ignore[no-any-return]
 
         user = await User.get_or_none(username=username)
         if user is not None:
-            result = await cache.set(user.username, user, ttl=600)
-            assert result, "user being cached"
+            await cache.set(user.username, user)
 
         return user  # type: ignore[no-any-return]
 
     @staticmethod
     async def delete_users(usernames: list[str]) -> None:
         await User.filter(username__in=usernames).delete()
         for username in usernames:
-            await caches.get("default").delete(username)
+            await cache.delete(username)
         users_count.set(await User.all().count())

tests/e2e/docker/goosebit/Dockerfile

@@ -58,7 +58,9 @@ EXPOSE 60053
 ENV PYTHONPATH=/app
 ENV PYTHONUNBUFFERED=1
 
-# We currently do not fully support multiple workers. For more information, see:
+# Multiple workers require the in-memory cache to be disabled (GOOSEBIT_CACHE__ENABLED=false)
+# and an explicit GOOSEBIT_SECRET_KEY. Override GUNICORN_CMD_ARGS at runtime to raise the
+# worker count. For more information, see:
 # https://github.com/UpstreamDataInc/goosebit/issues/125
 ENV GUNICORN_CMD_ARGS="--workers 1 --enable-stdio-inheritance"