akkudoktoreos.server.retentionmanager.RetentionManager

class akkudoktoreos.server.retentionmanager.RetentionManager(config_getter: Callable[[str], Any], *, shutdown_timeout: float = 30.0)

Bases: object

Orchestrates all periodic server-maintenance jobs.

The manager itself is driven by an external make_repeated_task heartbeat (the compaction tick). A config_getter callable — accepting a string key and returning the corresponding value — is supplied at initialisation and stored on every registered job, keeping the manager decoupled from any specific config implementation.

Jobs are launched as independent asyncio.Task objects so they run concurrently without blocking the tick. Call shutdown during application teardown to wait for any in-flight tasks to complete before the event loop closes. A configurable shutdown_timeout prevents the wait from blocking indefinitely; jobs still running after the timeout are reported by name but not cancelled.

__init__(config_getter: Callable[[str], Any], *, shutdown_timeout: float = 30.0) → None

Initialise the manager with a configuration accessor.

Parameters:

  • config_getter – Callable that accepts a string key and returns the corresponding configuration value. Used by each registered job to look up its interval in seconds.

  • shutdown_timeout – Maximum number of seconds to wait for in-flight jobs to finish during shutdown. If the timeout elapses before all tasks complete, an error is logged and the names of the still-running jobs are reported. The tasks are not cancelled so they may continue running until the event loop closes. Defaults to 30.0.

Example:

manager = RetentionManager(get_config().get_nested_value, shutdown_timeout=60.0)

Methods

__init__(config_getter, *[, shutdown_timeout])

Initialise the manager with a configuration accessor.

register(name, func, *, interval_attr[, ...])

Register a maintenance function with the manager.

shutdown()

Wait for all currently running job tasks to complete.

status()

Return a snapshot of every job's state for health or metrics endpoints.

tick()

Single compaction tick: check every job and fire those that are due.

unregister(name)

Remove a previously registered job from the manager.
__init__(config_getter: Callable[[str], Any], *, shutdown_timeout: float = 30.0) → None

Initialise the manager with a configuration accessor.

Parameters:

  • config_getter – Callable that accepts a string key and returns the corresponding configuration value. Used by each registered job to look up its interval in seconds.

  • shutdown_timeout – Maximum number of seconds to wait for in-flight jobs to finish during shutdown. If the timeout elapses before all tasks complete, an error is logged and the names of the still-running jobs are reported. The tasks are not cancelled so they may continue running until the event loop closes. Defaults to 30.0.

Example:

manager = RetentionManager(get_config().get_nested_value, shutdown_timeout=60.0)
register(name: str, func: Callable[[], None] | Callable[[], Coroutine[Any, Any, None]], *, interval_attr: str, fallback_interval: float = 300.0, on_exception: Callable[[Exception], None] | Callable[[Exception], Coroutine[Any, Any, None]] | None = None) → None

Register a maintenance function with the manager.

Parameters:

  • name – Unique human-readable job name used in logs and metrics.

  • func – The maintenance callable. Must accept no arguments.

  • interval_attr – Key passed to config_getter to retrieve the interval in seconds for this job. When the config value is None the job is treated as disabled and will never fire.

  • fallback_interval – Seconds to use when the config attribute is missing or zero. Defaults to 300.0.

  • on_exception – Optional callable invoked with the raised exception whenever func fails. Useful for cleanup or alerting. May be sync or async.

Raises:

ValueError – If a job with the given name is already registered.

unregister(name: str) → None

Remove a previously registered job from the manager.

If no job with the given name exists, this is a no-op.

Parameters:

name – The name of the job to remove.

async tick() → None

Single compaction tick: check every job and fire those that are due.

Each job resolves its own interval via the config_getter captured at registration time. Jobs whose interval is None are silently skipped (disabled). Due jobs are launched as independent asyncio.Task objects so they run concurrently without blocking the tick. Each task is tracked in _running_tasks and removed automatically on completion, allowing shutdown to await all of them gracefully.

Jobs that are still running from a previous tick are skipped to prevent overlapping executions.

Note

This is the function you pass to make_repeated_task.

async shutdown() → None

Wait for all currently running job tasks to complete.

Waits up to shutdown_timeout seconds (configured at initialisation) for in-flight tasks to finish. If the timeout elapses before all tasks complete, an error is logged listing the names of the jobs that are still running. Those tasks are not cancelled — they continue until the event loop closes — but shutdown returns so that application teardown is not blocked indefinitely.

Returns immediately if no tasks are running.

Example:

@asynccontextmanager
async def lifespan(app: FastAPI):
    tick_task = make_repeated_task(manager.tick, seconds=5, wait_first=2)
    await tick_task()
Yields:

await manager.shutdown()

status() → list[dict]

Return a snapshot of every job’s state for health or metrics endpoints.

Returns:

A list of dictionaries, one per registered job, each produced by JobState.summary.