feat: schedule runtime housekeeping

Author: cardmagic

CHANGELOG.md

@@ -47,6 +47,9 @@
   timeout, fairness yield, lease loss, and shutdown.
 - Bound activation passes by configurable elapsed time as well as message
   count so slow hot actors yield workers fairly.
+- Add independent supervised schedulers for expired message and process
+  retention and stale process recovery, with disable switches and bounded
+  failure backoff.
 
 ## 0.1.0 - 2026-08-13
 

docs/operations.md

@@ -61,12 +61,21 @@ every repair through the actor's typed `send` dispatcher, optionally with a
 future `availableAt` to spread large repairs. Never update persisted actor
 state from reconciliation code.
 
-Retention is explicit through `runtime.retention.preview()` and
+The runtime automatically prunes expired message and stopped-process history
+once at startup and every `retentionIntervalMilliseconds`, which defaults to
+one hour. Stale process ownership is recovered independently every
+`deadProcessCleanupIntervalMilliseconds`, which defaults to one minute. Set an
+interval to zero to disable its scheduler. Failed passes emit metadata-only
+events and retry with bounded exponential backoff without stopping other
+runtime roles.
+
+Operators can also use `runtime.retention.preview()` and
 `runtime.retention.prune()`. Both calls require administration authorization;
 use preview first and alert on an unexpected count before executing deletion.
 Message history defaults to 30 days with optional per-actor overrides. Stopped
 process history defaults to 7 days. Instance expiration is disabled unless an
-actor type appears in `instanceRetentionByActorType`.
+actor type appears in `instanceRetentionByActorType`, and remains an explicit
+operator action because it deletes the entire actor incarnation.
 
 Pruning selects and rechecks at most `pruneBatchSize` rows per transaction. It
 preserves ready and claimed messages, dead-letter originals and replacements,

docs/parity.md

@@ -41,12 +41,12 @@ Reference: Ruby `solid_objects` 0.12.0 at commit `a01b6f5`.
 
 | Capability                                                                    | Status | TypeScript shape or remaining work                                                                                                                               |
 | ----------------------------------------------------------------------------- | ------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| Process registration, heartbeats, stale claim recovery, and graceful shutdown | Native | Runtime roles persist process state and release ownership on shutdown; authorized inspection and cleanup recover every stale claim type.                         |
+| Process registration, heartbeats, stale claim recovery, and graceful shutdown | Native | Runtime roles persist process state and release ownership on shutdown; scheduled and authorized manual cleanup recover every stale claim type.                   |
 | Failed-role replacement                                                       | Native | Built-in and registered roles are rebuilt through their factories with capped backoff; shutdown is the terminal replacement boundary.                            |
 | Additional supervised components                                              | Native | `registerComponent()` builds, validates, runs, and stops application components with the runtime.                                                                |
 | Dead-letter inspection and retry                                              | Native | `runtime.deadLetters` provides deny-by-default immutable inspection and idempotent durable retry linkage.                                                        |
 | Reconciliation reads                                                          | Native | Authorized cursor pages cover active, quiet, and orphaned instances; bounded state batches are migrated and deeply frozen.                                       |
-| Message, process, and opt-in instance retention                               | Native | Authorized preview and bounded prune APIs preserve live work and support default, per-actor, and opt-in policies.                                                |
+| Message, process, and opt-in instance retention                               | Native | Supervised scheduling bounds message and process growth; authorized manual APIs add preview and keep destructive instance expiration explicit.                   |
 | Doctor and schema verification                                                | Native | Structured checks cover configuration, schema/version shape, adapter server versions, policy posture, live roles, and a targeted round trip.                     |
 | CLI                                                                           | Native | The packaged executable loads an application runtime and exposes start, diagnostics, processes, dead letters, reminders, and explicit retention pruning as JSON. |
 | Structured instrumentation                                                    | Native | An isolated transport-neutral sink emits immutable lifecycle metadata and structurally excludes application payloads.                                            |

src/configuration.ts

@@ -60,6 +60,8 @@ export interface SolidObjectsConfiguration {
   effectWorkerCount?: number
   broadcastWorkerCount?: number
   reminderSchedulerCount?: number
+  retentionIntervalMilliseconds?: number
+  deadProcessCleanupIntervalMilliseconds?: number
   messageRetentionMilliseconds?: number
   messageRetentionByActorType?: Readonly<Record<string, number>>
   instanceRetentionByActorType?: Readonly<Record<string, number>>
@@ -131,6 +133,9 @@ export function buildSettings(configuration: SolidObjectsConfiguration): Runtime
     effectWorkerCount: configuration.effectWorkerCount ?? 1,
     broadcastWorkerCount: configuration.broadcastWorkerCount ?? 1,
     reminderSchedulerCount: configuration.reminderSchedulerCount ?? 1,
+    retentionIntervalMilliseconds: configuration.retentionIntervalMilliseconds ?? 3_600_000,
+    deadProcessCleanupIntervalMilliseconds:
+      configuration.deadProcessCleanupIntervalMilliseconds ?? 60_000,
     messageRetentionMilliseconds: configuration.messageRetentionMilliseconds ?? 30 * 86_400_000,
     messageRetentionByActorType: Object.freeze({
       ...(configuration.messageRetentionByActorType ?? {}),
@@ -224,6 +229,12 @@ function validateSettings(settings: RuntimeSettings): void {
   ) {
     throw new TypeError("idleDeactivationTimeoutMilliseconds must be non-negative")
   }
+  for (const [name, value] of Object.entries({
+    retentionIntervalMilliseconds: settings.retentionIntervalMilliseconds,
+    deadProcessCleanupIntervalMilliseconds: settings.deadProcessCleanupIntervalMilliseconds,
+  })) {
+    if (!Number.isFinite(value) || value < 0) throw new TypeError(`${name} must be non-negative`)
+  }
   if (
     settings.supervisorMaximumRestartDelayMilliseconds < settings.supervisorRestartDelayMilliseconds
   ) {
@@ -253,7 +264,9 @@ function validateSettings(settings: RuntimeSettings): void {
     settings.workerCount +
     settings.effectWorkerCount +
     settings.reminderSchedulerCount +
-    (broadcastsEnabled(settings) ? settings.broadcastWorkerCount : 0)
+    (broadcastsEnabled(settings) ? settings.broadcastWorkerCount : 0) +
+    (settings.retentionIntervalMilliseconds > 0 ? 1 : 0) +
+    (settings.deadProcessCleanupIntervalMilliseconds > 0 ? 1 : 0)
   if (roleCount === 0) throw new TypeError("at least one runtime role must be configured")
 }
 

src/maintenance-scheduler.ts

@@ -0,0 +1,54 @@
+import type { SolidObjectsRuntime } from "./runtime.js"
+import { waitFor } from "./worker.js"
+
+export class MaintenanceScheduler {
+  private stopping = false
+
+  constructor(
+    private readonly options: {
+      runtime: SolidObjectsRuntime
+      intervalMilliseconds: number
+      failureEvent: string
+      operation: () => Promise<void>
+    },
+  ) {}
+
+  async run(signal: AbortSignal): Promise<void> {
+    let failureCount = 0
+    while (!signal.aborted && !this.stopping) {
+      try {
+        await this.options.operation()
+        failureCount = 0
+      } catch (error) {
+        failureCount += 1
+        this.options.runtime.emitInstrumentation(this.options.failureEvent, {
+          errorName: error instanceof Error ? error.name : "Error",
+          failureCount,
+        })
+      }
+      await waitFor(this.pauseMilliseconds(failureCount), signal)
+    }
+  }
+
+  requestShutdown(): void {
+    this.stopping = true
+  }
+
+  stopped(): boolean {
+    return this.stopping
+  }
+
+  stop(): void {
+    this.stopping = true
+  }
+
+  private pauseMilliseconds(failureCount: number): number {
+    if (failureCount === 0) return this.options.intervalMilliseconds
+    const exponent = Math.min(failureCount - 1, 16)
+    return Math.min(
+      this.options.runtime.settings.supervisorRestartDelayMilliseconds * 2 ** exponent,
+      this.options.runtime.settings.supervisorMaximumRestartDelayMilliseconds,
+      this.options.intervalMilliseconds,
+    )
+  }
+}

src/runtime.ts

@@ -66,6 +66,7 @@ import {
 } from "./process-administration.js"
 import { RealtimeManager } from "./realtime.js"
 import type { PayloadEnvelope } from "./browser/index.js"
+import { MaintenanceScheduler } from "./maintenance-scheduler.js"
 import type {
   BroadcastRow,
   ClaimedTurn,
@@ -253,6 +254,8 @@ export class SolidObjectsRuntime {
       broadcastWorkerCount: broadcastsEnabled(this.settings)
         ? this.settings.broadcastWorkerCount
         : 0,
+      retentionEnabled: this.settings.retentionIntervalMilliseconds > 0,
+      deadProcessCleanupEnabled: this.settings.deadProcessCleanupIntervalMilliseconds > 0,
     })
     const controller = new AbortController()
     const abort = () => controller.abort(signal.reason)
@@ -607,6 +610,11 @@ export class SolidObjectsRuntime {
       resource: "processes",
       authorizationContext: options.authorizationContext,
     })
+    const cleaned = await this.cleanupStaleProcesses()
+    return Object.freeze({ cleaned })
+  }
+
+  private async cleanupStaleProcesses(): Promise<number> {
     const cleaned = await this.repository.cleanupStaleProcesses()
     if (cleaned > 0) {
       this.wakeUp("actors")
@@ -615,7 +623,7 @@ export class SolidObjectsRuntime {
       this.wakeUp("broadcasts")
       this.emitInstrumentation("processes.cleaned", { count: cleaned })
     }
-    return Object.freeze({ cleaned })
+    return cleaned
   }
 
   async instancesWithoutPendingWork(
@@ -1158,6 +1166,28 @@ export class SolidObjectsRuntime {
             () => () => this.broadcastWorker(),
           )
         : []),
+      ...(this.settings.retentionIntervalMilliseconds > 0
+        ? [
+            () =>
+              new MaintenanceScheduler({
+                runtime: this,
+                intervalMilliseconds: this.settings.retentionIntervalMilliseconds,
+                failureEvent: "supervisor.retention_failed",
+                operation: () => this.pruneExpiredRecords(),
+              }),
+          ]
+        : []),
+      ...(this.settings.deadProcessCleanupIntervalMilliseconds > 0
+        ? [
+            () =>
+              new MaintenanceScheduler({
+                runtime: this,
+                intervalMilliseconds: this.settings.deadProcessCleanupIntervalMilliseconds,
+                failureEvent: "supervisor.process_cleanup_failed",
+                operation: () => this.cleanupStaleProcesses().then(() => undefined),
+              }),
+          ]
+        : []),
     ]
 
     for (const { count, factory } of this.additionalComponents) {
@@ -1179,6 +1209,13 @@ export class SolidObjectsRuntime {
     }
   }
 
+  private async pruneExpiredRecords(): Promise<void> {
+    for (const target of ["messages", "processes"] as const) {
+      const count = await this.repository.pruneRetention(target)
+      this.emitInstrumentation(`${target}.pruned`, { count })
+    }
+  }
+
   private async supervise(slot: ComponentSlot, signal: AbortSignal): Promise<void> {
     let failureCount = 0
     let replacementErrorName: string | null = null

test/definition.test.ts

@@ -219,8 +219,16 @@ describe("runtime configuration", () => {
         workerCount: 0,
         effectWorkerCount: 0,
         reminderSchedulerCount: 0,
+        retentionIntervalMilliseconds: 0,
+        deadProcessCleanupIntervalMilliseconds: 0,
       }),
     ).toThrow("at least one runtime role")
+    expect(() => buildSettings({ database, retentionIntervalMilliseconds: -1 })).toThrow(
+      "retentionIntervalMilliseconds must be non-negative",
+    )
+    expect(() => buildSettings({ database, deadProcessCleanupIntervalMilliseconds: -1 })).toThrow(
+      "deadProcessCleanupIntervalMilliseconds must be non-negative",
+    )
     expect(() => validateComponent({} as never)).toThrow("must implement run")
 
     await database.close()