feat: emit TTL warning events before sandbox expiry

Author: richiemcilroy

apps/api/src/index.ts

@@ -15,6 +15,7 @@ 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 { EventRecorderLive } from './services/event-recorder.live.js'
 import { IdempotencyRepoMemory } from './workers/idempotency-cleanup.memory.js'
 import { QuotaMemory } from './services/quota.memory.js'
 import { UsageMemory } from './services/usage.memory.js'
@@ -82,6 +83,7 @@ const GracefulShutdownLive = Layer.scopedDiscard(
 
 const ServerLive = Layer.mergeAll(AppLive, WorkersLive, GracefulShutdownLive).pipe(
   Layer.provide(ShutdownControllerLive),
+  Layer.provide(EventRecorderLive),
   Layer.provide(SandboxRepoMemory),
   Layer.provide(ExecRepoMemory),
   Layer.provide(SessionRepoMemory),

apps/api/src/services/events.ts

@@ -39,6 +39,10 @@ export function sandboxForked(params: { fork_sandbox_id: string }): EventPayload
   return { type: 'sandbox.forked', data: { fork_sandbox_id: params.fork_sandbox_id } }
 }
 
+export function sandboxTtlWarning(params: { seconds_remaining: number }): EventPayload {
+  return { type: 'sandbox.ttl_warning', data: { seconds_remaining: params.seconds_remaining } }
+}
+
 export function sandboxStopping(params: { reason: string }): EventPayload {
   return { type: 'sandbox.stopping', data: { reason: params.reason } }
 }

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

@@ -139,6 +139,12 @@ export function createIoRedisApi(client: Redis): RedisApi {
         return result === 1
       }),
 
+    markTtlWarned: (sandboxId, ttlSeconds) =>
+      Effect.promise(async () => {
+        const result = await client.set(`ttl_warned:${sandboxId}`, '1', 'EX', ttlSeconds, 'NX')
+        return result === 'OK'
+      }),
+
     ping: () =>
       Effect.promise(async () => {
         try {

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

@@ -23,6 +23,7 @@ export function createInMemoryRedisApi(): RedisApi {
   const artifactPaths = new Map<string, Set<string>>()
   const leaderLocks = new Map<string, { instanceId: string; expiresAt: number }>()
   const nodeHeartbeats = new Map<string, TtlEntry>()
+  const ttlWarned = new Map<string, TtlEntry>()
 
   function isExpired(entry: SlotEntry): boolean {
     return Date.now() >= entry.expiresAt
@@ -179,6 +180,16 @@ export function createInMemoryRedisApi(): RedisApi {
         return entry !== undefined && entry.expiresAt > Date.now()
       }),
 
+    markTtlWarned: (sandboxId, ttlSeconds) =>
+      Effect.sync(() => {
+        const key = `ttl_warned:${sandboxId}`
+        const now = Date.now()
+        const existing = ttlWarned.get(key)
+        if (existing && existing.expiresAt > now) return false
+        ttlWarned.set(key, { expiresAt: now + ttlSeconds * 1000 })
+        return true
+      }),
+
     ping: () => Effect.succeed(true),
   }
 }

apps/api/src/services/redis.ts

@@ -105,6 +105,12 @@ export interface RedisApi {
     nodeId: string,
   ) => Effect.Effect<boolean, never, never>
 
+  /** Atomically mark a sandbox as TTL-warned. Returns true if newly marked. */
+  readonly markTtlWarned: (
+    sandboxId: string,
+    ttlSeconds: number,
+  ) => Effect.Effect<boolean, never, never>
+
   /** Ping to check connectivity. */
   readonly ping: () => Effect.Effect<boolean, never, never>
 }

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

@@ -250,6 +250,17 @@ export function createInMemorySandboxRepo(): SandboxRepoApi {
         })
       }),
 
+    findNearTtlExpiry: (warningThresholdSeconds) =>
+      Effect.sync(() => {
+        const now = Date.now()
+        return Array.from(store.values()).filter((r) => {
+          if (r.status !== 'running' || !r.startedAt) return false
+          const expiresAt = r.startedAt.getTime() + r.ttlSeconds * 1000
+          const warningAt = expiresAt - warningThresholdSeconds * 1000
+          return warningAt <= now && expiresAt > now
+        })
+      }),
+
     findIdleSince: (cutoff) =>
       Effect.sync(() =>
         Array.from(store.values()).filter((r) => {

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

@@ -119,6 +119,11 @@ export interface SandboxRepoApi {
   /** Find running sandboxes that have exceeded their TTL. */
   readonly findExpiredTtl: () => Effect.Effect<SandboxRow[], never, never>
 
+  /** Find running sandboxes within the warning threshold of TTL expiry (not yet expired). */
+  readonly findNearTtlExpiry: (
+    warningThresholdSeconds: number,
+  ) => Effect.Effect<SandboxRow[], never, never>
+
   /** Find running sandboxes with lastActivityAt before the given cutoff. */
   readonly findIdleSince: (
     cutoff: Date,

apps/api/src/workers/index.ts

@@ -4,9 +4,11 @@ import type { SandboxRepo } from '../services/sandbox-repo.js'
 import type { ArtifactRepo } from '../services/artifact-repo.js'
 import type { ObjectStorage } from '../services/object-storage.js'
 import type { QuotaService } from '../services/quota.js'
+import type { EventRecorder } from '../services/event-recorder.js'
 import type { IdempotencyRepo } from './idempotency-cleanup.js'
 import { startWorkers, type WorkerConfig } from './runner.js'
 import { ttlEnforcementWorker } from './ttl-enforcement.js'
+import { ttlWarningWorker } from './ttl-warning.js'
 import { idleShutdownWorker } from './idle-shutdown.js'
 import { orphanReconciliationWorker } from './orphan-reconciliation.js'
 import { queueTimeoutWorker } from './queue-timeout.js'
@@ -21,10 +23,12 @@ export type WorkerDeps =
   | ArtifactRepo
   | ObjectStorage
   | QuotaService
+  | EventRecorder
   | IdempotencyRepo
 
 const allWorkers: ReadonlyArray<WorkerConfig<WorkerDeps>> = [
   ttlEnforcementWorker,
+  ttlWarningWorker,
   idleShutdownWorker,
   orphanReconciliationWorker,
   queueTimeoutWorker,

apps/api/src/workers/runner.test.ts

@@ -4,14 +4,17 @@ import { createInMemoryRedisApi } from '../services/redis.memory.js'
 import { createInMemorySandboxRepo } from '../services/sandbox-repo.memory.js'
 import { createInMemoryArtifactRepo } from '../services/artifact-repo.memory.js'
 import { createInMemoryObjectStorage } from '../services/object-storage.memory.js'
+import { createLiveEventRecorder } from '../services/event-recorder.live.js'
 import { RedisService, type RedisApi } from '../services/redis.js'
 import { SandboxRepo, type SandboxRepoApi } from '../services/sandbox-repo.js'
 import { ArtifactRepo, type ArtifactRepoApi } from '../services/artifact-repo.js'
 import { ObjectStorage, type ObjectStorageApi } from '../services/object-storage.js'
+import { EventRecorder, type EventRecorderApi } from '../services/event-recorder.js'
 import { IdempotencyRepo, type IdempotencyRepoApi } from './idempotency-cleanup.js'
 import { createTestableIdempotencyRepo } from './idempotency-cleanup.memory.js'
 import { runWorkerTick, type WorkerConfig } from './runner.js'
 import { ttlEnforcementWorker } from './ttl-enforcement.js'
+import { ttlWarningWorker } from './ttl-warning.js'
 import { idleShutdownWorker } from './idle-shutdown.js'
 import { orphanReconciliationWorker } from './orphan-reconciliation.js'
 import { queueTimeoutWorker } from './queue-timeout.js'
@@ -28,6 +31,7 @@ let redis: RedisApi
 let sandboxRepo: SandboxRepoApi
 let artifactRepo: ArtifactRepoApi
 let objectStorage: ObjectStorageApi
+let eventRecorder: EventRecorderApi
 let idempotencyApi: IdempotencyRepoApi
 let idempotencyStore: Map<string, { createdAt: Date }>
 let quotaApi: QuotaApi & { setOrgQuota: (orgId: string, quota: Record<string, unknown>) => void }
@@ -45,6 +49,7 @@ beforeEach(() => {
   sandboxRepo = createInMemorySandboxRepo()
   artifactRepo = createInMemoryArtifactRepo()
   objectStorage = createInMemoryObjectStorage()
+  eventRecorder = createLiveEventRecorder(objectStorage, redis)
   const testable = createTestableIdempotencyRepo()
   idempotencyApi = testable.api
   idempotencyStore = testable.store
@@ -55,6 +60,7 @@ beforeEach(() => {
     Layer.succeed(SandboxRepo, sandboxRepo),
     Layer.succeed(ArtifactRepo, artifactRepo),
     Layer.succeed(ObjectStorage, objectStorage),
+    Layer.succeed(EventRecorder, eventRecorder),
     Layer.succeed(IdempotencyRepo, idempotencyApi),
     Layer.succeed(QuotaService, quotaApi),
   )
@@ -746,3 +752,116 @@ describe('replay-retention', () => {
     expect(longRow!.replayExpiresAt!.getTime()).toBeGreaterThan(Date.now())
   })
 })
+
+// ---------------------------------------------------------------------------
+// TTL warning worker
+// ---------------------------------------------------------------------------
+
+describe('ttl-warning', () => {
+  async function createRunningSandbox(orgId: string, ttlSeconds: number, startedSecondsAgo: number) {
+    const id = generateUUIDv7()
+    await Effect.runPromise(
+      sandboxRepo.create({
+        id,
+        orgId,
+        imageId: SEED_IMAGE_ID,
+        profileId: SEED_PROFILE_ID,
+        profileName: 'small',
+        env: null,
+        ttlSeconds,
+        imageRef: 'sandchest://ubuntu-22.04',
+      }),
+    )
+    // assignNode sets status to 'running' and startedAt
+    const nodeId = generateUUIDv7()
+    await Effect.runPromise(redis.registerNodeHeartbeat(base62Encode(nodeId), 300))
+    await Effect.runPromise(sandboxRepo.assignNode(id, orgId, nodeId))
+
+    // Backdate startedAt to simulate time passing
+    const row = await Effect.runPromise(sandboxRepo.findById(id, orgId))
+    if (row) {
+      const backdatedStart = new Date(Date.now() - startedSecondsAgo * 1000)
+      // Use updateStatus to persist the sandbox, then manually adjust startedAt
+      // The memory repo stores mutable objects, so we can rely on the reference
+      ;(row as { startedAt: Date }).startedAt = backdatedStart
+    }
+
+    return id
+  }
+
+  test('returns 0 when no sandboxes exist', async () => {
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(0)
+  })
+
+  test('does not warn sandboxes far from TTL expiry', async () => {
+    // TTL 3600s, started 100s ago → 3500s remaining (well above 60s threshold)
+    await createRunningSandbox('org_test', 3600, 100)
+
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(0)
+  })
+
+  test('warns sandbox near TTL expiry', async () => {
+    // TTL 100s, started 70s ago → 30s remaining (within 60s threshold)
+    const id = await createRunningSandbox('org_test', 100, 70)
+    const sandboxId = bytesToId(SANDBOX_PREFIX, id)
+
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(1)
+
+    // Verify event was recorded
+    const events = await Effect.runPromise(eventRecorder.getEvents({ sandboxId, orgId: 'org_test' }))
+    const warning = events.find((e) => e.type === 'sandbox.ttl_warning')
+    expect(warning).toBeDefined()
+    expect(warning!.data.seconds_remaining).toBeNumber()
+    expect(warning!.data.seconds_remaining as number).toBeLessThanOrEqual(60)
+  })
+
+  test('does not warn the same sandbox twice', async () => {
+    // TTL 100s, started 70s ago → 30s remaining
+    await createRunningSandbox('org_test', 100, 70)
+
+    const count1 = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count1).toBe(1)
+
+    const count2 = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count2).toBe(0)
+  })
+
+  test('does not warn already-expired sandboxes', async () => {
+    // TTL 100s, started 110s ago → already expired (handled by ttl-enforcement)
+    await createRunningSandbox('org_test', 100, 110)
+
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(0)
+  })
+
+  test('does not warn non-running sandboxes', async () => {
+    const id = generateUUIDv7()
+    await Effect.runPromise(
+      sandboxRepo.create({
+        id,
+        orgId: 'org_test',
+        imageId: SEED_IMAGE_ID,
+        profileId: SEED_PROFILE_ID,
+        profileName: 'small',
+        env: null,
+        ttlSeconds: 0,
+        imageRef: 'sandchest://ubuntu-22.04',
+      }),
+    )
+    // Leave as 'queued'
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(0)
+  })
+
+  test('warns multiple sandboxes near expiry', async () => {
+    // Both within warning threshold
+    await createRunningSandbox('org_test', 100, 50)
+    await createRunningSandbox('org_test', 200, 160)
+
+    const count = await run(runWorkerTick(ttlWarningWorker, 'inst-1'))
+    expect(count).toBe(2)
+  })
+})

apps/api/src/workers/ttl-warning.ts

@@ -0,0 +1,45 @@
+import { Effect } from 'effect'
+import { bytesToId, SANDBOX_PREFIX } from '@sandchest/contract'
+import { SandboxRepo } from '../services/sandbox-repo.js'
+import { RedisService } from '../services/redis.js'
+import { EventRecorder } from '../services/event-recorder.js'
+import { sandboxTtlWarning } from '../services/events.js'
+import type { WorkerConfig } from './runner.js'
+
+/** Warn 60 seconds before TTL expiry. */
+const WARNING_THRESHOLD_SECONDS = 60
+
+/** Dedup key TTL — prevents re-warning the same sandbox for 120 seconds. */
+const DEDUP_TTL_SECONDS = 120
+
+export const ttlWarningWorker: WorkerConfig<SandboxRepo | RedisService | EventRecorder> = {
+  name: 'ttl-warning',
+  intervalMs: 10_000,
+  handler: Effect.gen(function* () {
+    const repo = yield* SandboxRepo
+    const redis = yield* RedisService
+    const recorder = yield* EventRecorder
+
+    const nearExpiry = yield* repo.findNearTtlExpiry(WARNING_THRESHOLD_SECONDS)
+
+    let warned = 0
+    for (const sandbox of nearExpiry) {
+      const sandboxId = bytesToId(SANDBOX_PREFIX, sandbox.id)
+      const isNew = yield* redis.markTtlWarned(sandboxId, DEDUP_TTL_SECONDS)
+      if (!isNew) continue
+
+      const expiresAt = sandbox.startedAt!.getTime() + sandbox.ttlSeconds * 1000
+      const secondsRemaining = Math.max(0, Math.round((expiresAt - Date.now()) / 1000))
+
+      yield* recorder.record({
+        sandboxId,
+        orgId: sandbox.orgId,
+        event: sandboxTtlWarning({ seconds_remaining: secondsRemaining }),
+      })
+
+      warned++
+    }
+
+    return warned
+  }),
+}