Index instance retention lookups

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## 0.15.1 - 2026-09-16
+
+- Index opt-in instance expiration by actor type and update time so pruning
+  can find expired instances without scanning retained history. Install schema
+  migration 10 before running this version. Retention policies, timestamps,
+  pending-work protections, and deletion order are unchanged.
+
 ## 0.15.0 - 2026-09-15
 
 - Report transient process-heartbeat errors and keep retrying while effects run,

docs/parity.md

@@ -86,7 +86,7 @@ Ruby field names; this does not change runtime delivery semantics.
 | 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 | Supervised scheduling bounds message and process growth; authorized manual APIs add preview and keep destructive instance expiration explicit.                                                                                                                                                                                                                                                                                                                                                                                                               |
+| 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. SQL instance pruning uses an index on actor type and update time.                                                                                                                                                                                                                                                                                                                                             |
 | Doctor and schema verification                                                | Native | Structured checks cover configuration, schema/version shape, adapter server versions, neutral-context policy probes, 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.                                                                                                                                                                                                                                                                                                                                                                                             |
 | Operator dashboard                                                            | Native | The opt-in `solid-objects/web` export provides Fetch and Node/Connect mounting, authorized runtime views and actions, session-backed CSRF, filtering, paging, charts, and immutable extension hooks. Matches the Ruby dashboard's own documented limits: no audit trail of admin actions, dead-letter retry is one at a time, and pause sets a flag rather than interrupting an in-flight turn.                                                                                                                                                              |

package.json

@@ -1,6 +1,6 @@
 {
   "name": "solid-objects",
-  "version": "0.15.0",
+  "version": "0.15.1",
   "description": "Race-free realtime state per application identity, backed by your SQL database",
   "type": "module",
   "license": "MIT",
@@ -115,8 +115,8 @@
     "test": "vitest run",
     "test:browser": "pnpm run build && playwright test",
     "test:coverage": "vitest run --coverage",
-    "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts",
-    "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts",
+    "test:postgresql": "vitest run test/postgresql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts",
+    "test:mysql": "vitest run test/mysql.test.ts test/effect-recovery.test.ts test/instance-retention.test.ts",
     "test:package": "node scripts/release-artifact-smoke.mjs",
     "test:recovery": "pnpm run build && node examples/failure-recovery/demo.ts",
     "test:at-least-once": "pnpm run build && node examples/at-least-once/demo.ts",

src/doctor.ts

@@ -210,11 +210,11 @@ export class Doctor {
           message: `incompatible schema identity ${wrongIdentity.schema_identity}`,
         })
       }
-      if (versions.join(",") !== "1,2,3,4,5,6,7,8,9") {
+      if (versions.join(",") !== "1,2,3,4,5,6,7,8,9,10") {
         return check({
           name: "schema",
           status: "fail",
-          message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9; found ${versions.join(", ")}`,
+          message: `expected schema migrations 1, 2, 3, 4, 5, 6, 7, 8, 9, 10; found ${versions.join(", ")}`,
         })
       }
       return check({

src/schema.ts

@@ -10,7 +10,8 @@ const OBSERVABLE_INVALIDATIONS_VERSION = 6
 const KEYED_REMINDERS_VERSION = 7
 const POLLING_INDEXES_VERSION = 8
 const EFFECT_RECOVERY_VERSION = 9
-const LATEST_VERSION = EFFECT_RECOVERY_VERSION
+const INSTANCE_RETENTION_INDEX_VERSION = 10
+const LATEST_VERSION = INSTANCE_RETENTION_INDEX_VERSION
 
 export async function installSchema(options: {
   connection: DatabaseConnection
@@ -333,6 +334,22 @@ export async function installSchema(options: {
     })
   }
 
+  if (!installedVersions.has(INSTANCE_RETENTION_INDEX_VERSION)) {
+    await createIndex({
+      connection,
+      family,
+      table: table("instances"),
+      name: `${prefix}instances_retention`,
+      columns: "actor_type, updated_at_ms, id",
+    })
+    await recordMigration({
+      connection,
+      table: table("schema_migrations"),
+      version: INSTANCE_RETENTION_INDEX_VERSION,
+      schemaIdentity,
+    })
+  }
+
   if (installedVersions.has(POLLING_INDEXES_VERSION)) return
   const pollingIndexes = [
     ["effects", `${prefix}effects_poll`, "status, available_at_ms, id"],

src/version.ts

@@ -1 +1 @@
-export const VERSION = "0.15.0"
+export const VERSION = "0.15.1"

test/dead-letters.test.ts

@@ -151,7 +151,7 @@ describe("schema migrations", () => {
     const broadcastColumns = await runtime.settings.database.connection((connection) =>
       connection.all<{ name: string }>("PRAGMA table_info(solid_objects_broadcasts)"),
     )
-    expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9])
+    expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
     expect(deadLetterColumns.map(({ name }) => name)).toContain("retried_message_id")
     expect(broadcastColumns.map(({ name }) => name)).toContain("invalidations")
     expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS)
@@ -184,7 +184,7 @@ describe("schema migrations", () => {
         "SELECT version FROM solid_objects_schema_migrations ORDER BY version",
       ),
     )
-    expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9])
+    expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
     expect(await installedPollingIndexes(runtime)).toEqual(POLLING_INDEX_COLUMNS)
   })
 

test/doctor.test.ts

@@ -27,7 +27,7 @@ describe("runtime doctor", () => {
     expect(check(report, "configuration").status).toBe("pass")
     expect(check(report, "schema")).toMatchObject({
       status: "pass",
-      details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9] },
+      details: { versions: [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] },
     })
     expect(check(report, "authorization").status).toBe("pass")
     expect(check(report, "database").status).toBe("pass")

test/instance-retention.test.ts

@@ -0,0 +1,225 @@
+import { afterEach, expect, it } from "vitest"
+import { mysql } from "../src/database/mysql.js"
+import { postgresql } from "../src/database/postgresql.js"
+import { sqlite } from "../src/database/sqlite.js"
+import type {
+  Database,
+  DatabaseConnection,
+  DatabaseTransactionOptions,
+} from "../src/database/types.js"
+import { createRuntime, type SolidObjectsRuntime } from "../src/runtime.js"
+
+const PREFIX = "retention_index_test_"
+const DAY = 24 * 60 * 60 * 1_000
+type RetentionQueryPlanRow = { key?: string | null; detail?: string; "QUERY PLAN"?: string }
+let runtime: SolidObjectsRuntime | undefined
+
+afterEach(async () => {
+  try {
+    await runtime?.repository.resetForTesting()
+  } finally {
+    await runtime?.close()
+    runtime = undefined
+  }
+})
+
+it.each(["fresh installation", "version-nine upgrade", "interrupted upgrade"])(
+  "uses the retention index to find rare expired instances after %s",
+  async (scenario) => {
+    const database = new RetentionPlanDatabase(testDatabase())
+    runtime = createRuntime({
+      database,
+      tableNamePrefix: PREFIX,
+      authorizeAdministration: () => true,
+      instanceRetentionByActorType: { RetentionActor: DAY },
+    })
+    await runtime.install()
+    await database.transaction(async (connection) => {
+      const now = await connection.nowMilliseconds()
+      for (let offset = 0; offset < 2_000; offset += 250) {
+        const parameters = Array.from({ length: 250 }, (_, index) => {
+          const identity = String(offset + index).padStart(8, "0")
+          return [identity, "RetentionActor", identity, "{}", 1, now - 3 * DAY, now]
+        })
+        await connection.run(
+          `INSERT INTO ${PREFIX}instances
+         (id, actor_type, actor_id, state, state_version, created_at_ms, updated_at_ms)
+         VALUES ${parameters.map(() => "(?, ?, ?, ?, ?, ?, ?)").join(", ")}`,
+          parameters.flat(),
+        )
+      }
+      await connection.run(`UPDATE ${PREFIX}instances SET updated_at_ms = ? WHERE id = ?`, [
+        now - 2 * DAY,
+        "00001999",
+      ])
+    })
+    if (scenario === "version-nine upgrade") {
+      await database.connection(async (connection) => {
+        const tableClause = database.family === "mysql" ? ` ON ${PREFIX}instances` : ""
+        await connection.run(`DROP INDEX ${PREFIX}instances_retention${tableClause}`)
+      })
+    }
+    if (scenario === "version-nine upgrade" || scenario === "interrupted upgrade") {
+      await database.connection((connection) =>
+        connection.run(`DELETE FROM ${PREFIX}schema_migrations WHERE version = 10`),
+      )
+      await runtime.install()
+      await runtime.install()
+    }
+    await database.connection((connection) =>
+      connection.run(
+        `${database.family === "mysql" ? "ANALYZE TABLE" : "ANALYZE"} ${PREFIX}instances`,
+      ),
+    )
+
+    expect(await runtime.retention.preview({ target: "instances" })).toEqual({
+      target: "instances",
+      count: 1,
+    })
+    expect(await runtime.retention.prune({ target: "instances" })).toEqual({
+      target: "instances",
+      count: 1,
+    })
+    expect(database.previewPlans).toHaveLength(1)
+    expect(database.pruningPlans).toHaveLength(2)
+    const planColumn =
+      database.family === "mysql" ? "key" : database.family === "sqlite" ? "detail" : "QUERY PLAN"
+    for (const plan of [...database.previewPlans, ...database.pruningPlans]) {
+      expect(plan.map((row) => String(row[planColumn])).join("\n")).toContain(
+        `${PREFIX}instances_retention`,
+      )
+    }
+    const remaining = await database.connection((connection) =>
+      connection.get<{ count: number | bigint }>(
+        `SELECT COUNT(*) AS count FROM ${PREFIX}instances`,
+      ),
+    )
+    expect(Number(remaining?.count)).toBe(1_999)
+    const versions = await database.connection((connection) =>
+      connection.all<{ version: number | bigint }>(
+        `SELECT version FROM ${PREFIX}schema_migrations ORDER BY version`,
+      ),
+    )
+    expect(versions.map(({ version }) => Number(version))).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
+  },
+  30_000,
+)
+
+it("preserves distinct actor policies and unconfigured instances with the retention index", async () => {
+  const database = testDatabase()
+  runtime = createRuntime({
+    database,
+    tableNamePrefix: PREFIX,
+    authorizeAdministration: () => true,
+    instanceRetentionByActorType: { RetentionActor: DAY, OtherRetentionActor: 2 * DAY },
+  })
+  await runtime.install()
+  await database.transaction(async (connection) => {
+    const now = await connection.nowMilliseconds()
+    const cases = [
+      { identity: "expired", actorType: "RetentionActor", age: 2 * DAY },
+      { identity: "recent", actorType: "RetentionActor", age: DAY / 2 },
+      { identity: "other-expired", actorType: "OtherRetentionActor", age: 3 * DAY },
+      { identity: "other-retained", actorType: "OtherRetentionActor", age: 1.5 * DAY },
+      { identity: "unconfigured", actorType: "UnconfiguredActor", age: 3 * DAY },
+    ]
+    for (const { identity, actorType, age } of cases) {
+      await connection.run(
+        `INSERT INTO ${PREFIX}instances
+         (id, actor_type, actor_id, state, state_version, created_at_ms, updated_at_ms)
+         VALUES (?, ?, ?, ?, ?, ?, ?)`,
+        [identity, actorType, identity, "{}", 1, now - 3 * DAY, now - age],
+      )
+    }
+  })
+
+  expect(await runtime.retention.preview({ target: "instances" })).toEqual({
+    target: "instances",
+    count: 2,
+  })
+  expect(await runtime.retention.prune({ target: "instances" })).toEqual({
+    target: "instances",
+    count: 2,
+  })
+  const remaining = await database.connection((connection) =>
+    connection.all<{ id: string }>(`SELECT id FROM ${PREFIX}instances ORDER BY id`),
+  )
+  expect(remaining.map(({ id }) => id)).toEqual(["other-retained", "recent", "unconfigured"])
+}, 30_000)
+
+function testDatabase(): Database {
+  const connectionString = process.env.SOLID_OBJECTS_DATABASE_URL
+  if (!connectionString) return sqlite({ path: ":memory:" })
+  if (connectionString.startsWith("mysql:")) return mysql({ connectionString })
+  if (connectionString.startsWith("postgresql:")) return postgresql({ connectionString })
+  throw new Error("instance retention tests require a mysql: or postgresql: database URL")
+}
+
+class RetentionPlanDatabase implements Database {
+  readonly family: Database["family"]
+  readonly schemaIdentity: string
+  readonly previewPlans: RetentionQueryPlanRow[][] = []
+  readonly pruningPlans: RetentionQueryPlanRow[][] = []
+
+  constructor(private readonly database: Database) {
+    this.family = database.family
+    this.schemaIdentity = database.schemaIdentity
+  }
+
+  connection<Result>(
+    callback: (connection: DatabaseConnection) => Promise<Result>,
+  ): Promise<Result> {
+    return this.database.connection((connection) => callback(this.recordingConnection(connection)))
+  }
+
+  transaction<Result>(
+    callback: (connection: DatabaseConnection) => Promise<Result>,
+    options?: DatabaseTransactionOptions,
+  ): Promise<Result> {
+    return this.database.transaction(
+      (connection) => callback(this.recordingConnection(connection)),
+      options,
+    )
+  }
+
+  close(): Promise<void> {
+    return this.database.close()
+  }
+
+  private recordingConnection(connection: DatabaseConnection): DatabaseConnection {
+    const explainStatement = {
+      sqlite: "EXPLAIN QUERY PLAN",
+      mysql: "EXPLAIN FORMAT=TRADITIONAL",
+      postgresql: "EXPLAIN",
+    }[this.family]
+    const explain = (options: {
+      sql: string
+      parameters: Parameters<DatabaseConnection["all"]>[1]
+    }) =>
+      connection.all<RetentionQueryPlanRow>(
+        `${explainStatement} ${options.sql}`,
+        options.parameters,
+      )
+
+    return {
+      run: (sql, parameters) => connection.run(sql, parameters),
+      get: async <Row extends object>(
+        ...[sql, parameters]: Parameters<DatabaseConnection["get"]>
+      ) => {
+        if (sql.startsWith(`SELECT COUNT(*) AS count FROM ${PREFIX}instances WHERE`)) {
+          this.previewPlans.push(await explain({ sql, parameters }))
+        }
+        return connection.get<Row>(sql, parameters)
+      },
+      all: async <Row extends object>(
+        ...[sql, parameters]: Parameters<DatabaseConnection["all"]>
+      ) => {
+        if (sql.startsWith(`SELECT id FROM ${PREFIX}instances WHERE`)) {
+          this.pruningPlans.push(await explain({ sql, parameters }))
+        }
+        return connection.all<Row>(sql, parameters)
+      },
+      nowMilliseconds: () => connection.nowMilliseconds(),
+    }
+  }
+}