Leaner storage: store only events the row can't say, 15 → 9 indexes, HOT renewals

docs/workers.md

@@ -27,6 +27,7 @@ const worker = new Worker({
   retention: {
     completedMs: 7 * 86_400_000,
     failedMs: 30 * 86_400_000,
+    eventsMs: 86_400_000,
     intervalMs: 600_000,
     batchSize: 100,
   },
@@ -42,7 +43,7 @@ By default a worker claims only its registered types. Separate workers with dist
 
 Worker events are local observations, emitted after corresponding commits where applicable. Listener exceptions and rejected Promises are contained. Synchronous expensive listeners still block the event loop. Reaper dead letters appear in persisted event history, not necessarily a local worker callback. Use durable activities for required follow-up work rather than relying on an event listener.
 
-Retention is disabled by default. It deletes complete terminal trees and their inputs, results, checkpoints, events, keys and dependencies atomically. Live consumer trees pin shared producer results. Completion/failure retention clocks are separate; zero keeps that class forever.
+Retention is disabled by default. It deletes complete terminal trees and their inputs, results, checkpoints, events, keys and dependencies atomically. Live consumer trees pin shared producer results. Completion/failure retention clocks are separate; zero keeps that class forever. `eventsMs` trims the events of finished activities sooner than their tree; zero keeps them with the tree.
 
 ### Executor snapshots
 

src/postgres/command.ts

@@ -634,7 +634,8 @@ export class PostgresCommands {
       return skipped(row, "still running: cancel it first");
     const live = await c.query(
       `SELECT EXISTS(SELECT 1 FROM runnerq_activities
-      WHERE queue_name=$1 AND root_activity_id=$2 AND status NOT IN ${terminalSQL}) AS live`,
+      WHERE queue_name=$1 AND root_activity_id=$2 AND parent_activity_id IS NOT NULL
+      AND status NOT IN ${terminalSQL}) AS live`,
       [queue, row.id],
     );
     if (live.rows[0].live)

src/postgres/query.ts

@@ -60,7 +60,7 @@ const activityFields: Record<string, Field> = {
   queue: { expr: "a.queue_name", kind: "string" },
   status: { expr: "a.status", kind: "status" },
   priority: { expr: "a.priority", kind: "int" },
-  root_id: { expr: "COALESCE(a.root_activity_id, a.id)", kind: "uuid" },
+  root_id: { expr: "a.root_activity_id", kind: "uuid" },
   parent_id: { expr: "a.parent_activity_id", kind: "uuid", nullable: true },
   depth: { expr: "a.depth", kind: "int" },
   idempotency_key: {
@@ -134,18 +134,12 @@ export function canonicalStatus(internal: string): RecordStatus {
 
 /** Internal event names (Go's and TypeScript's are the same) to canonical event types. */
 const canonicalEvents: Record<string, string> = {
-  Enqueued: RecordEvent.created,
-  Scheduled: RecordEvent.scheduled,
-  Dequeued: RecordEvent.attemptStarted,
-  Completed: RecordEvent.attemptSucceeded,
   Failed: RecordEvent.attemptFailed,
   Retrying: RecordEvent.attemptFailed,
   DeadLetter: RecordEvent.deadLetter,
   Requeued: RecordEvent.leaseExpired,
   Yielded: RecordEvent.waitParked,
   Signaled: RecordEvent.signalReceived,
-  LeaseExtended: RecordEvent.leaseExtended,
-  ResultStored: RecordEvent.resultStored,
   SpawnLinked: RecordEvent.childLinked,
   Cancelled: RecordEvent.cancelled,
   Retried: RecordEvent.retried,
@@ -270,7 +264,17 @@ export class SqlBuilder {
         const negate = f.op === "ne" || f.op === "nin";
         // Nothing can match (e.g. an id that is not a UUID).
         if (!vals) return String(negate);
-        const cond = `${fd.expr} = ANY(${this.arg(vals.values, vals.type + "[]")})`;
+        let cond: string;
+        if (fd.kind === "status")
+          // Literals, not a parameter: Postgres proves a partial index's predicate
+          // (idx_runnerq_query_status) only from constants. The values are canonicalStatuses'.
+          cond = `${fd.expr} IN (${(vals.values as string[]).map((v) => `'${v.replaceAll("'", "''")}'`).join(", ")})`;
+        else if (name === "root_id") {
+          // Roots aren't in idx_runnerq_root_children; they're found by id.
+          const p = this.arg(vals.values, "uuid[]");
+          cond = `(a.id = ANY(${p}) OR (a.root_activity_id = ANY(${p}) AND a.parent_activity_id IS NOT NULL))`;
+        } else
+          cond = `${fd.expr} = ANY(${this.arg(vals.values, vals.type + "[]")})`;
         return negate ? `(NOT COALESCE(${cond}, false))` : cond;
       }
       case "lt":
@@ -348,7 +352,7 @@ export class SqlBuilder {
     const vals = convertValues(name, "uuid", raw);
     if (!vals) return "false";
     const p = this.arg(vals.values, "uuid[]");
-    return `e.activity_id IN (SELECT x.id FROM runnerq_activities x WHERE x.id = ANY(${p}) OR x.root_activity_id = ANY(${p}))`;
+    return `e.activity_id IN (SELECT x.id FROM runnerq_activities x WHERE x.id = ANY(${p}) OR (x.root_activity_id = ANY(${p}) AND x.parent_activity_id IS NOT NULL))`;
   }
 }
 
@@ -934,7 +938,7 @@ export class PostgresQueries {
     const { cols, joins } = activitySelect(inc);
     const rows = await this.run(
       `SELECT ${cols} FROM runnerq_activities a${joins}
-      WHERE a.id = $1::uuid OR a.root_activity_id = $1::uuid
+      WHERE a.id = $1::uuid OR (a.root_activity_id = $1::uuid AND a.parent_activity_id IS NOT NULL)
       ORDER BY a.depth, a.created_at, a.id LIMIT ${max + 1}`,
       [root],
     );

src/postgres/trees.ts

@@ -5,6 +5,13 @@ import type { PoolClient } from "pg";
 /** The statuses an activity never leaves (bar a retry command). */
 export const terminalSQL = "('completed','failed','dead_letter','cancelled')";
 
+/**
+ * The workflow rooted at $2 in queue $1: the root by id, its descendants through
+ * idx_runnerq_root_children.
+ */
+export const treeSQL = `SELECT a.id,a.idempotency_key FROM runnerq_activities a WHERE a.queue_name=$1
+  AND (a.id=$2 OR (a.root_activity_id=$2 AND a.parent_activity_id IS NOT NULL))`;
+
 /**
  * Locks the tree's idempotency keys, then reports whether a live workflow in another tree
  * depends on a result this tree produced (the tree must then be kept). Key reuse registers
@@ -15,19 +22,20 @@ export async function lockTree(
   queue: string,
   root: string,
 ): Promise<boolean> {
+  // A key's activity carries the key, so each key is a primary-key probe.
   await c.query(
-    `SELECT 1 FROM runnerq_idempotency WHERE queue_name=$1 AND activity_id IN
-    (SELECT id FROM runnerq_activities WHERE queue_name=$1 AND (id=$2 OR root_activity_id=$2)) FOR UPDATE`,
+    `SELECT 1 FROM runnerq_idempotency k JOIN (${treeSQL}) t
+    ON k.queue_name=$1 AND k.idempotency_key=t.idempotency_key AND k.activity_id=t.id FOR UPDATE OF k`,
     [queue, root],
   );
   const r = await c.query(
-    `SELECT EXISTS(SELECT 1 FROM runnerq_dependencies d
-    JOIN runnerq_activities producer ON producer.id=d.producer_activity_id AND producer.queue_name=$1
+    `SELECT EXISTS(SELECT 1 FROM (${treeSQL}) t
+    JOIN runnerq_dependencies d ON d.queue_name=$1 AND d.result_id=t.id AND d.producer_activity_id=t.id
     JOIN runnerq_activities waiter ON waiter.id=d.waiter_activity_id AND waiter.queue_name=$1
-    WHERE d.queue_name=$1 AND COALESCE(producer.root_activity_id,producer.id)=$2
-    AND COALESCE(waiter.root_activity_id,waiter.id)<>$2
+    WHERE waiter.root_activity_id<>$2
     AND EXISTS(SELECT 1 FROM runnerq_activities live WHERE live.queue_name=$1
-      AND COALESCE(live.root_activity_id,live.id)=COALESCE(waiter.root_activity_id,waiter.id)
+      AND (live.id=waiter.root_activity_id
+        OR (live.root_activity_id=waiter.root_activity_id AND live.parent_activity_id IS NOT NULL))
       AND live.status NOT IN ${terminalSQL})) AS pinned`,
     [queue, root],
   );
@@ -41,16 +49,53 @@ export async function deleteTree(
   root: string,
 ): Promise<void> {
   await c.query(
-    `WITH tree AS (SELECT id FROM runnerq_activities WHERE queue_name=$1 AND (id=$2 OR root_activity_id=$2)),
+    `WITH tree AS (${treeSQL}),
     del_dependencies AS (DELETE FROM runnerq_dependencies WHERE queue_name=$1
-      AND (waiter_activity_id IN (SELECT id FROM tree) OR producer_activity_id IN (SELECT id FROM tree))),
+      AND (waiter_activity_id IN (SELECT id FROM tree)
+        OR (result_id IN (SELECT id FROM tree) AND producer_activity_id IS NOT NULL))),
     del_results AS (DELETE FROM runnerq_results WHERE queue_name=$1
       AND (activity_id IN (SELECT id FROM tree) OR owner_activity_id IN (SELECT id FROM tree)) RETURNING activity_id),
     del_events AS (DELETE FROM runnerq_events WHERE queue_name=$1
       AND (activity_id IN (SELECT id FROM tree) OR activity_id IN (SELECT activity_id FROM del_results))),
-    del_idem AS (DELETE FROM runnerq_idempotency WHERE queue_name=$1 AND activity_id IN (SELECT id FROM tree)),
+    del_idem AS (DELETE FROM runnerq_idempotency k USING tree
+      WHERE k.queue_name=$1 AND k.idempotency_key=tree.idempotency_key AND k.activity_id=tree.id),
     del_inputs AS (DELETE FROM runnerq_inputs WHERE activity_id IN (SELECT id FROM tree))
     DELETE FROM runnerq_activities WHERE queue_name=$1 AND id IN (SELECT id FROM tree)`,
     [queue, root],
   );
 }
+
+/** $2 and $3 are the completed and failed ages in milliseconds (0 keeps them). */
+const expiredCondSQL = `r.status IN ${terminalSQL}
+  AND r.completed_at<NOW()-(CASE WHEN r.status='completed' THEN $2::bigint ELSE $3::bigint END)*INTERVAL '1 millisecond'
+  AND CASE WHEN r.status='completed' THEN $2::bigint ELSE $3::bigint END>0`;
+
+/**
+ * The oldest expired root of queue $1 with no live descendant ($4 lists roots already passed
+ * over): one short walk of idx_runnerq_root_terminal per terminal status.
+ */
+export const expiredRootSQL = `SELECT r.id FROM unnest(ARRAY['completed','failed','dead_letter','cancelled']) s(status)
+  CROSS JOIN LATERAL (SELECT r.id,r.completed_at FROM runnerq_activities r
+    WHERE r.queue_name=$1 AND r.parent_activity_id IS NULL AND r.status IN ${terminalSQL} AND r.status=s.status
+    AND r.completed_at<NOW()-(CASE WHEN s.status='completed' THEN $2::bigint ELSE $3::bigint END)*INTERVAL '1 millisecond'
+    AND r.id<>ALL($4::uuid[])
+    AND NOT EXISTS(SELECT 1 FROM runnerq_activities c WHERE c.root_activity_id=r.id AND c.parent_activity_id IS NOT NULL
+      AND c.status NOT IN ${terminalSQL})
+    ORDER BY r.completed_at LIMIT 1) r
+  WHERE CASE WHEN s.status='completed' THEN $2::bigint ELSE $3::bigint END>0
+  ORDER BY r.completed_at LIMIT 1`;
+
+/** Locks and rechecks a candidate root ($4) that expiredRootSQL read unlocked. */
+export const lockExpiredRootSQL = `SELECT EXISTS(SELECT 1 FROM runnerq_activities r
+  WHERE r.id=$4 AND r.queue_name=$1 AND r.parent_activity_id IS NULL AND ${expiredCondSQL}
+  FOR UPDATE SKIP LOCKED) AS locked`;
+
+/**
+ * Deletes up to $3 events older than $2 milliseconds whose activity has finished (or is
+ * gone), oldest first along idx_runnerq_events_queue_seq. Events of unfinished activities
+ * stay: a waiting activity's latest Yielded is its waiting reason.
+ */
+export const trimEventsSQL = `DELETE FROM runnerq_events WHERE id IN (SELECT e.id FROM runnerq_events e
+  WHERE e.queue_name=$1 AND e.created_at<NOW()-$2::bigint*INTERVAL '1 millisecond'
+  AND NOT EXISTS(SELECT 1 FROM runnerq_activities a WHERE a.id=e.activity_id AND a.status NOT IN ${terminalSQL})
+  ORDER BY e.queue_name,e.id LIMIT $3)`;

src/query.ts

@@ -62,6 +62,9 @@ export type RecordStatus =
 /**
  * Canonical event types. Backends map their internal event names onto these; events with
  * no canonical equivalent keep a namespaced name of their own ("other.<name>").
+ *
+ * `created`, `scheduled`, `attemptStarted` and `attemptSucceeded` are never stored: an
+ * activity's createdAt, scheduledAt, startedAt and completedAt say them.
  */
 export const RecordEvent = {
   created: "activity.created",
@@ -72,10 +75,8 @@ export const RecordEvent = {
   attemptFailed: "attempt.failed",
   attemptTimedOut: "attempt.timed_out",
   leaseExpired: "attempt.lease_expired",
-  leaseExtended: "attempt.lease_extended",
   waitParked: "wait.parked",
   signalReceived: "signal.received",
-  resultStored: "result.stored",
   childLinked: "child.linked",
   deadLetter: "dead_letter.entered",
   redriven: "dead_letter.redriven",