feat: add purgeRuns API and retainRuns auto-cleanup option (#109)

* feat: add purgeRuns API and retainRuns auto-cleanup option

- `durably.purgeRuns({ olderThan, limit })` for manual batch deletion of
  terminal runs (completed, failed, cancelled) with cascading cleanup of
  steps, logs, and labels
- `retainRuns: '30d'` option on createDurably for automatic periodic purge
  (runs once per 60s during worker polling, batch size 100)
- Migration v2: adds (status, completed_at) index for efficient purge queries
- Duration parser supports 'd' (days), 'h' (hours), 'm' (minutes)
- 11 new tests covering all purge scenarios

Closes #88

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: simplify purge implementation per review

- Extract cascadeDeleteRuns() helper to eliminate duplication between
  deleteRun() and purgeRuns()
- Extract TERMINAL_STATUSES constant to module scope
- Move PURGE_INTERVAL_MS to module scope (was recreated per processOne call)
- Make auto-purge fire-and-forget (void) so it doesn't block job claiming
  or lease renewal in the worker polling loop

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* refactor: consolidate purge index into migration v1

No backward compatibility needed — merge the (status, completed_at) index
into the single v1 migration instead of keeping a separate v2.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address simplify review findings

- Move TERMINAL_STATUSES after RunStatus type definition (ordering)
- Replace void fire-and-forget with .catch() to prevent unhandled rejection
- Use single Date.now() variable in purge block instead of calling 3 times
- Add comment documenting intentional immediate purge on first cycle

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* perf: remove redundant getRuns({ status: 'leased' }) from processOne hot path

The claimNext SQL already contains an activeLeaseGuard subquery that
checks NOT EXISTS for active leases with the same concurrency key.
The JS-side getRuns → filter → excludeConcurrencyKeys flow was
redundant and added a full table scan + JOIN + JSON parse per poll cycle.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* docs: add purgeRuns and retainRuns to website API reference

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* docs: fix purgeRuns signature — olderThan is required Date, not optional string

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* fix: address CodeRabbit review findings

- Move auto-purge after claimNext so it never serializes with job claiming
- Add positive auto-purge test with backdated completed_at
- Clarify retainRuns runs only during worker polling in docs
- Fix purgeRuns(options?) → purgeRuns(options) in API quick reference
- Clarify olderThan matches completedAt, not just "completed" status

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: coji

CLAUDE.md

@@ -45,6 +45,7 @@ Five tables: `durably_runs`, `durably_run_labels`, `durably_steps`, `durably_log
 - `leaseRenewIntervalMs`: 5000ms
 - `leaseMs`: 30000ms (lease duration; expired leases are reclaimed)
 - `preserveSteps`: false (deletes step output data when runs reach terminal state)
+- `retainRuns`: undefined (no automatic cleanup; set e.g. `'30d'` to auto-delete terminal runs)
 
 ## Browser Constraints (by design)
 

packages/durably/docs/llms.md

@@ -36,6 +36,7 @@ const durably = createDurably({
   leaseRenewIntervalMs: 5000, // Lease renewal interval (ms)
   leaseMs: 30000, // Lease duration (ms); expired leases are reclaimed
   preserveSteps: false, // Set to true to keep step output data after terminal state (default: false = cleanup)
+  retainRuns: '30d', // Auto-delete terminal runs older than 30 days (runs during worker polling; supports 'd', 'h', 'm' units)
   // Optional: type-safe labels with Zod schema
   // labels: z.object({ organizationId: z.string(), env: z.string() }),
   jobs: {
@@ -229,6 +230,21 @@ await durably.cancel(runId)
 await durably.deleteRun(runId)
 ```
 
+### Purge Old Runs
+
+Batch-delete terminal runs (completed, failed, cancelled) older than a cutoff date.
+Pending and leased runs are never deleted.
+
+```ts
+// Delete terminal runs older than 30 days
+const deleted = await durably.purgeRuns({
+  olderThan: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000),
+  limit: 500, // optional batch size (default: 1000)
+})
+```
+
+For automatic cleanup, use the `retainRuns` option (see Quick Start). Cleanup runs during idle worker polling cycles, at most once per minute, in batches of 100.
+
 ## Events
 
 Subscribe to job execution events:

packages/durably/src/durably.ts

@@ -71,6 +71,12 @@ export interface DurablyOptions<
    * ```
    */
   jobs?: TJobs
+  /**
+   * Auto-delete terminal runs older than the specified duration.
+   * Only runs in terminal states (completed, failed, cancelled) are purged.
+   * @example '30d' (30 days), '24h' (24 hours), '60m' (60 minutes)
+   */
+  retainRuns?: string
 }
 
 /**
@@ -83,6 +89,25 @@ const DEFAULTS = {
   preserveSteps: false,
 } as const
 
+function parseDuration(value: string): number {
+  const match = value.match(/^(\d+)(d|h|m)$/)
+  if (!match) {
+    throw new Error(
+      `Invalid duration format: "${value}". Use e.g. '30d', '24h', '60m'`,
+    )
+  }
+  const num = Number.parseInt(match[1], 10)
+  const unit = match[2]
+  const multipliers: Record<string, number> = {
+    d: 86400000,
+    h: 3600000,
+    m: 60000,
+  }
+  return num * multipliers[unit]
+}
+
+const PURGE_INTERVAL_MS = 60_000
+
 const ulid = monotonicFactory()
 const BROWSER_SINGLETON_REGISTRY_KEY = '__durablyBrowserSingletonRegistry'
 const BROWSER_LOCAL_DIALECT_KEY = '__durablyBrowserLocalKey'
@@ -307,6 +332,13 @@ export interface Durably<
    */
   deleteRun(runId: string): Promise<void>
 
+  /**
+   * Delete terminal runs older than the specified cutoff.
+   * Only runs in terminal states (completed, failed, cancelled) are purged.
+   * @returns Number of deleted runs
+   */
+  purgeRuns(options: { olderThan: Date; limit?: number }): Promise<number>
+
   /**
    * Get a run by ID
    * @example
@@ -376,6 +408,8 @@ interface DurablyState<
   migrated: boolean
   leaseMs: number
   leaseRenewIntervalMs: number
+  retainRunsMs: number | null
+  lastPurgeAt: number
   releaseBrowserSingleton: () => void
 }
 
@@ -864,26 +898,38 @@ function createDurablyInstance<
       })
     },
 
+    async purgeRuns(options: {
+      olderThan: Date
+      limit?: number
+    }): Promise<number> {
+      return storage.purgeRuns({
+        olderThan: options.olderThan.toISOString(),
+        limit: options.limit,
+      })
+    },
+
     async processOne(options?: { workerId?: string }): Promise<boolean> {
       const workerId = options?.workerId ?? defaultWorkerId()
       const now = new Date().toISOString()
 
       await storage.releaseExpiredLeases(now)
 
-      const leasedRuns = await storage.getRuns({ status: 'leased' })
-      const excludeConcurrencyKeys = leasedRuns
-        .filter(
-          (entry): entry is Run<TLabels> & { concurrencyKey: string } =>
-            entry.concurrencyKey !== null &&
-            entry.leaseExpiresAt !== null &&
-            entry.leaseExpiresAt > now,
-        )
-        .map((entry) => entry.concurrencyKey)
-
-      const run = await storage.claimNext(workerId, now, state.leaseMs, {
-        excludeConcurrencyKeys,
-      })
+      const run = await storage.claimNext(workerId, now, state.leaseMs)
       if (!run) {
+        // Auto-purge old terminal runs if retainRuns is configured.
+        // Runs after claimNext so purge never serializes with job claiming.
+        // lastPurgeAt starts at 0, so the first idle cycle purges immediately.
+        if (
+          state.retainRunsMs !== null &&
+          Date.now() - state.lastPurgeAt >= PURGE_INTERVAL_MS
+        ) {
+          const purgeNow = Date.now()
+          state.lastPurgeAt = purgeNow
+          const cutoff = new Date(purgeNow - state.retainRunsMs).toISOString()
+          storage.purgeRuns({ olderThan: cutoff, limit: 100 }).catch(() => {
+            // Purge failure is non-fatal — will retry on next interval
+          })
+        }
         return false
       }
 
@@ -978,6 +1024,7 @@ export function createDurably<
       options.leaseRenewIntervalMs ?? DEFAULTS.leaseRenewIntervalMs,
     leaseMs: options.leaseMs ?? DEFAULTS.leaseMs,
     preserveSteps: options.preserveSteps ?? DEFAULTS.preserveSteps,
+    retainRunsMs: options.retainRuns ? parseDuration(options.retainRuns) : null,
   }
 
   const db = new Kysely<Database>({ dialect: options.dialect })
@@ -1023,6 +1070,8 @@ export function createDurably<
     migrated: false,
     leaseMs: config.leaseMs,
     leaseRenewIntervalMs: config.leaseRenewIntervalMs,
+    retainRunsMs: config.retainRunsMs,
+    lastPurgeAt: 0,
     releaseBrowserSingleton,
   }
 

packages/durably/src/migrations.ts

@@ -80,6 +80,13 @@ const migrations: Migration[] = [
         .columns(['job_name', 'created_at'])
         .execute()
 
+      await db.schema
+        .createIndex('idx_durably_runs_status_completed')
+        .ifNotExists()
+        .on('durably_runs')
+        .columns(['status', 'completed_at'])
+        .execute()
+
       // Create normalized labels table for indexed label filtering
       await db.schema
         .createTable('durably_run_labels')

packages/durably/src/storage.ts

@@ -11,6 +11,9 @@ export type RunStatus =
   | 'failed'
   | 'cancelled'
 
+/** Run statuses that represent terminal (non-active) states */
+const TERMINAL_STATUSES: RunStatus[] = ['completed', 'failed', 'cancelled']
+
 /**
  * Run data for creating a new run
  */
@@ -125,10 +128,6 @@ export interface ProgressData {
   message?: string
 }
 
-export interface ClaimOptions {
-  excludeConcurrencyKeys?: string[]
-}
-
 export type DatabaseBackend = 'generic' | 'postgres'
 
 /**
@@ -169,7 +168,6 @@ export interface Store<
     workerId: string,
     now: string,
     leaseMs: number,
-    options?: ClaimOptions,
   ): Promise<Run<TLabels> | null>
   renewLease(
     runId: string,
@@ -215,6 +213,9 @@ export interface Store<
     progress: ProgressData | null,
   ): Promise<void>
 
+  // Purge
+  purgeRuns(options: { olderThan: string; limit?: number }): Promise<number>
+
   // Logs
   createLog(input: CreateLogInput): Promise<Log>
   getLogs(runId: string): Promise<Log[]>
@@ -361,6 +362,20 @@ export function createKyselyStore(
 ): Store<Record<string, string>> {
   const withWriteLock = createWriteMutex()
 
+  /** Delete runs and all associated data (steps, logs, labels) in dependency order */
+  async function cascadeDeleteRuns(
+    trx: Kysely<Database>,
+    ids: string[],
+  ): Promise<void> {
+    await trx.deleteFrom('durably_steps').where('run_id', 'in', ids).execute()
+    await trx.deleteFrom('durably_logs').where('run_id', 'in', ids).execute()
+    await trx
+      .deleteFrom('durably_run_labels')
+      .where('run_id', 'in', ids)
+      .execute()
+    await trx.deleteFrom('durably_runs').where('id', 'in', ids).execute()
+  }
+
   async function insertLabelRows(
     executor: Kysely<Database>,
     runId: string,
@@ -648,30 +663,40 @@ export function createKyselyStore(
 
     async deleteRun(runId: string) {
       await db.transaction().execute(async (trx) => {
-        await trx
-          .deleteFrom('durably_steps')
-          .where('run_id', '=', runId)
-          .execute()
-        await trx
-          .deleteFrom('durably_logs')
-          .where('run_id', '=', runId)
-          .execute()
-        await trx
-          .deleteFrom('durably_run_labels')
-          .where('run_id', '=', runId)
+        await cascadeDeleteRuns(trx, [runId])
+      })
+    },
+
+    async purgeRuns(options: {
+      olderThan: string
+      limit?: number
+    }): Promise<number> {
+      const limit = options.limit ?? 1000
+
+      return await db.transaction().execute(async (trx) => {
+        const rows = await trx
+          .selectFrom('durably_runs')
+          .select('id')
+          .where('status', 'in', TERMINAL_STATUSES)
+          .where('completed_at', '<', options.olderThan)
+          .orderBy('completed_at', 'asc')
+          .limit(limit)
           .execute()
-        await trx.deleteFrom('durably_runs').where('id', '=', runId).execute()
+
+        if (rows.length === 0) return 0
+
+        const ids = rows.map((r) => r.id)
+        await cascadeDeleteRuns(trx, ids)
+        return ids.length
       })
     },
 
     async claimNext(
       workerId: string,
       now: string,
       leaseMs: number,
-      options?: ClaimOptions,
     ): Promise<Run | null> {
       const leaseExpiresAt = new Date(Date.parse(now) + leaseMs).toISOString()
-      const excludeConcurrencyKeys = options?.excludeConcurrencyKeys ?? []
       const activeLeaseGuard = sql<boolean>`
         (
           concurrency_key IS NULL
@@ -689,7 +714,7 @@ export function createKyselyStore(
 
       if (backend === 'postgres') {
         return await db.transaction().execute(async (trx) => {
-          const skipKeys = [...excludeConcurrencyKeys]
+          const skipKeys: string[] = []
 
           // Loop: on concurrency-key conflict, exclude that key and retry
           // to find the next eligible candidate in the same transaction.
@@ -790,15 +815,6 @@ export function createKyselyStore(
         .orderBy('id', 'asc')
         .limit(1)
 
-      if (excludeConcurrencyKeys.length > 0) {
-        subquery = subquery.where((eb) =>
-          eb.or([
-            eb('concurrency_key', 'is', null),
-            eb('concurrency_key', 'not in', excludeConcurrencyKeys),
-          ]),
-        )
-      }
-
       const row = await db
         .updateTable('durably_runs')
         .set({
@@ -1035,6 +1051,7 @@ export function createKyselyStore(
     'enqueueMany',
     'updateRun',
     'deleteRun',
+    'purgeRuns',
     'claimNext',
     'renewLease',
     'releaseExpiredLeases',

packages/durably/tests/node/migration-consolidated.test.ts

@@ -61,6 +61,7 @@ describe('migration consolidated schema', () => {
     expect(indexNames).toContain('idx_durably_runs_status_created')
     expect(indexNames).toContain('idx_durably_runs_status_lease_expires')
     expect(indexNames).toContain('idx_durably_runs_job_created')
+    expect(indexNames).toContain('idx_durably_runs_status_completed')
 
     // Steps indexes
     expect(indexNames).toContain('idx_durably_steps_run_index')

packages/durably/tests/node/purge.test.ts

@@ -0,0 +1,4 @@
+import { createNodeDialect } from '../helpers/node-dialect'
+import { createPurgeTests } from '../shared/purge.shared'
+
+createPurgeTests(createNodeDialect)