feat: implement background workers with Redis leader election

Author: richiemcilroy

apps/api/src/index.ts

@@ -1,6 +1,6 @@
 import { HttpServer } from '@effect/platform'
 import { NodeHttpServer, NodeRuntime } from '@effect/platform-node'
-import { Layer } from 'effect'
+import { Effect, Layer } from 'effect'
 import { createServer } from 'node:http'
 import { ApiRouter } from './server.js'
 import { withAuth, withRequestId } from './middleware.js'
@@ -13,6 +13,8 @@ import { NodeClientMemory } from './services/node-client.memory.js'
 import { ArtifactRepoMemory } from './services/artifact-repo.memory.js'
 import { createRedisLayer } from './services/redis.ioredis.js'
 import { RedisMemory } from './services/redis.memory.js'
+import { IdempotencyRepoMemory } from './workers/idempotency-cleanup.memory.js'
+import { startAllWorkers } from './workers/index.js'
 
 const PORT = Number(process.env.PORT ?? 3000)
 const REDIS_URL = process.env.REDIS_URL
@@ -22,13 +24,21 @@ const AppLive = ApiRouter.pipe(withRateLimit, withAuth, withRequestId, HttpServe
 
 const RedisLive = REDIS_URL ? createRedisLayer(REDIS_URL) : RedisMemory
 
-const ServerLive = AppLive.pipe(
+// Workers launch as scoped fibers alongside the HTTP server
+const WorkersLive = Layer.scopedDiscard(
+  Effect.gen(function* () {
+    yield* startAllWorkers()
+  }),
+)
+
+const ServerLive = Layer.mergeAll(AppLive, WorkersLive).pipe(
   Layer.provide(SandboxRepoMemory),
   Layer.provide(ExecRepoMemory),
   Layer.provide(SessionRepoMemory),
   Layer.provide(ObjectStorageMemory),
   Layer.provide(NodeClientMemory),
   Layer.provide(ArtifactRepoMemory),
+  Layer.provide(IdempotencyRepoMemory),
   Layer.provide(RedisLive),
   Layer.provide(NodeHttpServer.layer(() => createServer(), { port: PORT })),
 )

apps/api/src/services/artifact-repo.memory.ts

@@ -83,6 +83,23 @@ export function createInMemoryArtifactRepo(): ArtifactRepoApi {
         }
         return n
       }),
+
+    findExpiredRetention: (before) =>
+      Effect.sync(() =>
+        Array.from(store.values()).filter(
+          (r) => r.retentionUntil !== null && r.retentionUntil.getTime() < before.getTime(),
+        ),
+      ),
+
+    deleteByIds: (ids) =>
+      Effect.sync(() => {
+        let deleted = 0
+        for (const id of ids) {
+          const key = keyFor(id)
+          if (store.delete(key)) deleted++
+        }
+        return deleted
+      }),
   }
 }
 

apps/api/src/services/artifact-repo.ts

@@ -52,6 +52,16 @@ export interface ArtifactRepoApi {
     sandboxId: Uint8Array,
     orgId: string,
   ) => Effect.Effect<number, never, never>
+
+  /** Find artifacts past their retention date. */
+  readonly findExpiredRetention: (
+    before: Date,
+  ) => Effect.Effect<ArtifactRow[], never, never>
+
+  /** Delete artifacts by IDs. Returns count of deleted rows. */
+  readonly deleteByIds: (
+    ids: Uint8Array[],
+  ) => Effect.Effect<number, never, never>
 }
 
 export class ArtifactRepo extends Context.Tag('ArtifactRepo')<ArtifactRepo, ArtifactRepoApi>() {}

apps/api/src/services/object-storage.memory.ts

@@ -19,6 +19,11 @@ export function createInMemoryObjectStorage(): ObjectStorageApi {
       Effect.sync(() =>
         `https://s3.example.com/${key}?expires=${expiresInSeconds}`,
       ),
+
+    deleteObject: (key) =>
+      Effect.sync(() => {
+        store.delete(key)
+      }),
   }
 }
 

apps/api/src/services/object-storage.ts

@@ -18,6 +18,11 @@ export interface ObjectStorageApi {
     key: string,
     expiresInSeconds: number,
   ) => Effect.Effect<string, never, never>
+
+  /** Delete an object from a bucket. */
+  readonly deleteObject: (
+    key: string,
+  ) => Effect.Effect<void, never, never>
 }
 
 export class ObjectStorage extends Context.Tag('ObjectStorage')<ObjectStorage, ObjectStorageApi>() {}

apps/api/src/services/redis.ioredis.ts

@@ -120,6 +120,24 @@ export function createIoRedisApi(client: Redis): RedisApi {
         return client.scard(key)
       }),
 
+    acquireLeaderLock: (workerName, instanceId, ttlMs) =>
+      Effect.promise(async () => {
+        const key = `worker:${workerName}:leader`
+        const result = await client.set(key, instanceId, 'PX', ttlMs, 'NX')
+        return result === 'OK'
+      }),
+
+    registerNodeHeartbeat: (nodeId, ttlSeconds) =>
+      Effect.promise(async () => {
+        await client.set(`node_heartbeat:${nodeId}`, '1', 'EX', ttlSeconds)
+      }),
+
+    hasNodeHeartbeat: (nodeId) =>
+      Effect.promise(async () => {
+        const result = await client.exists(`node_heartbeat:${nodeId}`)
+        return result === 1
+      }),
+
     ping: () =>
       Effect.promise(async () => {
         try {

apps/api/src/services/redis.memory.ts

@@ -10,13 +10,19 @@ interface RateEntry {
   timestamps: number[]
 }
 
+interface TtlEntry {
+  expiresAt: number
+}
+
 /** In-memory Redis implementation for testing. */
 export function createInMemoryRedisApi(): RedisApi {
   const slots = new Map<string, SlotEntry>()
   const rateLimits = new Map<string, RateEntry>()
   const execEvents = new Map<string, BufferedEvent[]>()
   const replayEvents = new Map<string, BufferedEvent[]>()
   const artifactPaths = new Map<string, Set<string>>()
+  const leaderLocks = new Map<string, { instanceId: string; expiresAt: number }>()
+  const nodeHeartbeats = new Map<string, TtlEntry>()
 
   function isExpired(entry: SlotEntry): boolean {
     return Date.now() >= entry.expiresAt
@@ -145,6 +151,31 @@ export function createInMemoryRedisApi(): RedisApi {
         return artifactPaths.get(key)?.size ?? 0
       }),
 
+    acquireLeaderLock: (workerName, instanceId, ttlMs) =>
+      Effect.sync(() => {
+        const key = `worker:${workerName}:leader`
+        const now = Date.now()
+        const existing = leaderLocks.get(key)
+        if (existing && existing.expiresAt > now) {
+          return existing.instanceId === instanceId
+        }
+        leaderLocks.set(key, { instanceId, expiresAt: now + ttlMs })
+        return true
+      }),
+
+    registerNodeHeartbeat: (nodeId, ttlSeconds) =>
+      Effect.sync(() => {
+        nodeHeartbeats.set(`node_heartbeat:${nodeId}`, {
+          expiresAt: Date.now() + ttlSeconds * 1000,
+        })
+      }),
+
+    hasNodeHeartbeat: (nodeId) =>
+      Effect.sync(() => {
+        const entry = nodeHeartbeats.get(`node_heartbeat:${nodeId}`)
+        return entry !== undefined && entry.expiresAt > Date.now()
+      }),
+
     ping: () => Effect.succeed(true),
   }
 }

apps/api/src/services/redis.ts

@@ -86,6 +86,24 @@ export interface RedisApi {
     sandboxId: string,
   ) => Effect.Effect<number, never, never>
 
+  /** Acquire a worker leader lock. Returns true if this instance became the leader. */
+  readonly acquireLeaderLock: (
+    workerName: string,
+    instanceId: string,
+    ttlMs: number,
+  ) => Effect.Effect<boolean, never, never>
+
+  /** Register a node heartbeat with a TTL. */
+  readonly registerNodeHeartbeat: (
+    nodeId: string,
+    ttlSeconds: number,
+  ) => Effect.Effect<void, never, never>
+
+  /** Check if a node has an active heartbeat. */
+  readonly hasNodeHeartbeat: (
+    nodeId: string,
+  ) => Effect.Effect<boolean, never, never>
+
   /** Ping to check connectivity. */
   readonly ping: () => Effect.Effect<boolean, never, never>
 }

apps/api/src/services/sandbox-repo.memory.ts

@@ -49,6 +49,7 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
         const row: SandboxRow = {
           id: params.id,
           orgId: params.orgId,
+          nodeId: null,
           imageId: params.imageId,
           profileId: params.profileId,
           profileName: params.profileName,
@@ -59,6 +60,7 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
           forkCount: 0,
           ttlSeconds: params.ttlSeconds,
           failureReason: null,
+          lastActivityAt: null,
           createdAt: now,
           updatedAt: now,
           startedAt: null,
@@ -149,6 +151,7 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
         const row: SandboxRow = {
           id: params.id,
           orgId: params.orgId,
+          nodeId: params.source.nodeId,
           imageId: params.source.imageId,
           profileId: params.source.profileId,
           profileName: params.source.profileName,
@@ -159,6 +162,7 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
           forkCount: 0,
           ttlSeconds: params.ttlSeconds,
           failureReason: null,
+          lastActivityAt: now,
           createdAt: now,
           updatedAt: now,
           startedAt: now,
@@ -211,6 +215,51 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
 
         return result
       }),
+
+    findExpiredTtl: () =>
+      Effect.sync(() => {
+        const now = Date.now()
+        return Array.from(store.values()).filter((r) => {
+          if (r.status !== 'running' || !r.startedAt) return false
+          return r.startedAt.getTime() + r.ttlSeconds * 1000 < now
+        })
+      }),
+
+    findIdleSince: (cutoff) =>
+      Effect.sync(() =>
+        Array.from(store.values()).filter((r) => {
+          if (r.status !== 'running') return false
+          const activity = r.lastActivityAt ?? r.startedAt ?? r.createdAt
+          return activity.getTime() < cutoff.getTime()
+        }),
+      ),
+
+    findQueuedBefore: (cutoff) =>
+      Effect.sync(() =>
+        Array.from(store.values()).filter(
+          (r) => r.status === 'queued' && r.createdAt.getTime() < cutoff.getTime(),
+        ),
+      ),
+
+    getActiveNodeIds: () =>
+      Effect.sync(() => {
+        const nodeIds = new Map<string, Uint8Array>()
+        for (const row of store.values()) {
+          if (row.status === 'running' && row.nodeId) {
+            nodeIds.set(base62Encode(row.nodeId), row.nodeId)
+          }
+        }
+        return Array.from(nodeIds.values())
+      }),
+
+    findRunningOnNodes: (nodeIds) =>
+      Effect.sync(() => {
+        if (nodeIds.length === 0) return []
+        return Array.from(store.values()).filter((r) => {
+          if (r.status !== 'running' || !r.nodeId) return false
+          return nodeIds.some((nid) => bytesEqual(r.nodeId!, nid))
+        })
+      }),
   }
 }
 

apps/api/src/services/sandbox-repo.ts

@@ -10,6 +10,7 @@ import type {
 export interface SandboxRow {
   readonly id: Uint8Array
   readonly orgId: string
+  readonly nodeId: Uint8Array | null
   readonly imageId: Uint8Array
   readonly profileId: Uint8Array
   readonly profileName: ProfileName
@@ -20,6 +21,7 @@ export interface SandboxRow {
   readonly forkCount: number
   readonly ttlSeconds: number
   readonly failureReason: FailureReason | null
+  readonly lastActivityAt: Date | null
   readonly createdAt: Date
   readonly updatedAt: Date
   readonly startedAt: Date | null
@@ -99,6 +101,27 @@ export interface SandboxRepoApi {
     id: Uint8Array,
     orgId: string,
   ) => Effect.Effect<SandboxRow[], never, never>
+
+  /** Find running sandboxes that have exceeded their TTL. */
+  readonly findExpiredTtl: () => Effect.Effect<SandboxRow[], never, never>
+
+  /** Find running sandboxes with lastActivityAt before the given cutoff. */
+  readonly findIdleSince: (
+    cutoff: Date,
+  ) => Effect.Effect<SandboxRow[], never, never>
+
+  /** Find queued sandboxes created before the given cutoff. */
+  readonly findQueuedBefore: (
+    cutoff: Date,
+  ) => Effect.Effect<SandboxRow[], never, never>
+
+  /** Get distinct nodeIds from running sandboxes. */
+  readonly getActiveNodeIds: () => Effect.Effect<Uint8Array[], never, never>
+
+  /** Find running sandboxes assigned to any of the given nodeIds. */
+  readonly findRunningOnNodes: (
+    nodeIds: Uint8Array[],
+  ) => Effect.Effect<SandboxRow[], never, never>
 }
 
 export class SandboxRepo extends Context.Tag('SandboxRepo')<SandboxRepo, SandboxRepoApi>() {}