Prevent idle triggers from passing pending lifecycle deliveries

packages/api/src/postgres/automation-scheduler.ts

@@ -53,7 +53,11 @@ export class AutomationScheduler {
   }
 
   private async lifecycle() {
-    const rows = await this.database.pool.query<any>(`SELECT t.tenant_id,t.id trigger_id,t.job_id,t.slug,t.config,r.id source_run_id,r.job_id source_job_id,r.state,r.updated_at
+    // Drain the backlog before checking idle triggers. The idle query below
+    // also guards against eligible unacknowledged deliveries so concurrent
+    // scheduler passes cannot cross this ordering boundary.
+    for (;;) {
+      const rows = await this.database.pool.query<any>(`SELECT t.tenant_id,t.id trigger_id,t.job_id,t.slug,t.config,r.id source_run_id,r.job_id source_job_id,r.state,r.updated_at
       FROM app.triggers t
       JOIN app.jobs j ON j.tenant_id=t.tenant_id AND j.id=t.job_id
       JOIN app.runs r ON r.tenant_id=t.tenant_id
@@ -63,13 +67,17 @@ export class AutomationScheduler {
         AND r.state IN ('succeeded','failed','stopped')
         AND COALESCE(t.config->'states',t.config->'terminalStates','["succeeded","failed","stopped"]'::jsonb) ? r.state
         AND d.source_run_id IS NULL ORDER BY r.updated_at LIMIT 100`);
-    for (const row of rows.rows) {
-      const context = { sourceRunId: row.source_run_id, sourceJobId: row.source_job_id, state: row.state, completedAt: row.updated_at.toISOString() }, occurrence = { occurredAt: context.completedAt, externalId: `${row.source_run_id}:${row.state}`, metadata: context };
-      // Acknowledge after invocation so transient failures remain retryable.
-      // The stable claim key deduplicates concurrent passes and retries after
-      // an invocation commits but its delivery acknowledgement fails.
-      await new InvocationService(this.repository(row.tenant_id), value => encrypt(value, this.env.CREDENTIAL_ENCRYPTION_KEY)).invoke(row.job_id, { source: "jobLifecycle", triggerId: row.trigger_id, claimKey: `lifecycle:${row.trigger_id}:${row.source_run_id}:${row.state}`, context: { [row.slug]: context }, occurrence }).catch(skipFullQueue);
-      await this.database.pool.query("INSERT INTO app.lifecycle_deliveries(tenant_id,trigger_id,source_run_id,terminal_state) VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING", [row.tenant_id, row.trigger_id, row.source_run_id, row.state]);
+      if (!rows.rows.length) break;
+      for (const row of rows.rows) {
+        const context = { sourceRunId: row.source_run_id, sourceJobId: row.source_job_id, state: row.state, completedAt: row.updated_at.toISOString() }, occurrence = { occurredAt: context.completedAt, externalId: `${row.source_run_id}:${row.state}`, metadata: context };
+        // Acknowledge after invocation so transient failures remain retryable.
+        // Queue-full is an intentional terminal outcome: no run is queued, so
+        // acknowledging it permits idle triggers to proceed.
+        // The stable claim key deduplicates retries after an invocation commits
+        // but its delivery acknowledgement fails.
+        await new InvocationService(this.repository(row.tenant_id), value => encrypt(value, this.env.CREDENTIAL_ENCRYPTION_KEY)).invoke(row.job_id, { source: "jobLifecycle", triggerId: row.trigger_id, claimKey: `lifecycle:${row.trigger_id}:${row.source_run_id}:${row.state}`, context: { [row.slug]: context }, occurrence }).catch(skipFullQueue);
+        await this.database.pool.query("INSERT INTO app.lifecycle_deliveries(tenant_id,trigger_id,source_run_id,terminal_state) VALUES ($1,$2,$3,$4) ON CONFLICT DO NOTHING", [row.tenant_id, row.trigger_id, row.source_run_id, row.state]);
+      }
     }
   }
 
@@ -91,6 +99,27 @@ export class AutomationScheduler {
           WHERE newer.state IN ('succeeded','failed','stopped')
             AND (newer.updated_at,newer.id) > (r.updated_at,r.id)
         )
+        AND NOT EXISTS (
+          SELECT 1
+          FROM jsonb_array_elements_text(t.config->'sourceJobIds') watched(value)
+          JOIN app.runs pending_source ON pending_source.tenant_id=t.tenant_id
+            AND pending_source.job_id=watched.value::uuid
+            AND pending_source.state IN ('succeeded','failed','stopped')
+          JOIN app.triggers lifecycle_trigger ON lifecycle_trigger.tenant_id=pending_source.tenant_id
+            AND lifecycle_trigger.kind='jobLifecycle' AND lifecycle_trigger.enabled
+            AND lifecycle_trigger.removed_at IS NULL
+            AND COALESCE(lifecycle_trigger.config->'sourceJobIds',jsonb_build_array(lifecycle_trigger.config->>'sourceJobId')) ? pending_source.job_id::text
+            AND COALESCE(lifecycle_trigger.config->'states',lifecycle_trigger.config->'terminalStates','["succeeded","failed","stopped"]'::jsonb) ? pending_source.state
+          JOIN app.jobs lifecycle_job ON lifecycle_job.tenant_id=lifecycle_trigger.tenant_id
+            AND lifecycle_job.id=lifecycle_trigger.job_id AND lifecycle_job.enabled
+          WHERE NOT EXISTS (
+            SELECT 1 FROM app.lifecycle_deliveries pending_delivery
+            WHERE pending_delivery.tenant_id=lifecycle_trigger.tenant_id
+              AND pending_delivery.trigger_id=lifecycle_trigger.id
+              AND pending_delivery.source_run_id=pending_source.id
+              AND pending_delivery.terminal_state=pending_source.state
+          )
+        )
         AND NOT EXISTS (SELECT 1 FROM app.lifecycle_deliveries d WHERE d.tenant_id=t.tenant_id AND d.trigger_id=t.id AND d.source_run_id=r.id AND d.terminal_state='runsZero')
         ORDER BY r.updated_at LIMIT 100`);
     for (const row of rows.rows) {

packages/api/test/job-lifecycle.postgres.test.ts

@@ -141,6 +141,73 @@ describe.skipIf(!process.env.AUTH_TEST_DATABASE_URL)("job lifecycle delivery (Po
     expect(await invocations(a, target.id)).toMatchObject([{ source: "jobIdle" }]);
   });
 
+  it("holds an all-runs-complete trigger behind a pending lifecycle delivery", async () => {
+    const a = await tenant(), source = await job(a), downstream = await job(a, { sourceJobIds: [source.id], states: ["succeeded"] });
+    const idle = await job(a, { sourceJobIds: [source.id, downstream.id] }, true, "jobIdle");
+    await complete(a, source);
+
+    // The database guard must work even when an idle pass races ahead of the
+    // lifecycle pass (or a previous lifecycle pass has been delayed).
+    await deliverIdle();
+    expect(await invocations(a, idle.id)).toHaveLength(0);
+    await deliver();
+    await deliverIdle();
+    expect(await invocations(a, downstream.id)).toHaveLength(1);
+    expect(await invocations(a, idle.id)).toHaveLength(0);
+
+    const downstreamRun = await runs.claimNext();
+    expect(downstreamRun?.jobId).toBe(downstream.id);
+    await runs.consumeWakeHint();
+    await runs.terminal(downstreamRun!, "succeeded", "stored", undefined, true);
+    await deliverIdle();
+    expect(await invocations(a, idle.id)).toHaveLength(1);
+  });
+
+  it("drains lifecycle deliveries beyond the first 100 before firing all-runs-complete", async () => {
+    const a = await tenant(), source = await job(a), downstream = await job(a, { sourceJobIds: [source.id], states: ["succeeded"] });
+    const idle = await job(a, { sourceJobIds: [source.id, downstream.id] }, true, "jobIdle");
+    for (let index = 0; index < 101; index++) await complete(a, source);
+
+    await deliver();
+    expect((await database.pool.query("SELECT * FROM app.lifecycle_deliveries WHERE tenant_id=$1 AND trigger_id=$2", [a, downstream.triggers[0]!.id])).rows).toHaveLength(101);
+    expect(await invocations(a, downstream.id)).toHaveLength(2);
+    await deliverIdle();
+    expect(await invocations(a, idle.id)).toHaveLength(0);
+
+    let downstreamRun = await runs.claimNext();
+    expect(downstreamRun?.jobId).toBe(downstream.id);
+    await runs.consumeWakeHint();
+    await runs.terminal(downstreamRun!, "succeeded", "stored", undefined, true);
+    downstreamRun = await runs.claimNext();
+    expect(downstreamRun?.jobId).toBe(downstream.id);
+    await runs.consumeWakeHint();
+    await runs.terminal(downstreamRun!, "succeeded", "stored", undefined, true);
+    await deliverIdle();
+    expect(await invocations(a, idle.id)).toHaveLength(1);
+  }, 20_000);
+
+  it("acknowledges a queue-full lifecycle delivery and lets idle proceed after existing work ends", async () => {
+    const a = await tenant(), source = await job(a), downstream = await job(a, { sourceJobIds: [source.id], states: ["succeeded"] });
+    const idle = await job(a, { sourceJobIds: [source.id, downstream.id] }, true, "jobIdle");
+    const invoker = new InvocationService(repository(a), value => encrypt(value, key));
+    const occupied = await invoker.invoke(downstream.id, { source: "manual", triggerId: downstream.triggers[0]!.id, claimKey: crypto.randomUUID(), context: {} });
+    const waiter = await invoker.invoke(downstream.id, { source: "manual", triggerId: downstream.triggers[0]!.id, claimKey: crypto.randomUUID(), context: {} });
+    const occupiedRun = await runs.claimNext();
+    expect(occupiedRun?.id).toBe(occupied.run.id);
+    await complete(a, source);
+    await deliver();
+    expect(await invocations(a, downstream.id)).toHaveLength(2);
+    expect(await deliveries()).toHaveLength(1);
+    await runs.consumeWakeHint();
+    await runs.terminal(occupiedRun!, "succeeded", "stored", undefined, true);
+    const waitingRun = await runs.claimNext();
+    expect(waitingRun?.id).toBe(waiter.run.id);
+    await runs.consumeWakeHint();
+    await runs.terminal(waitingRun!, "succeeded", "stored", undefined, true);
+    await deliverIdle();
+    expect(await invocations(a, idle.id)).toHaveLength(1);
+  });
+
   it("matches each of multiple source IDs and legacy singular config with terminalStates", async () => {
     const a = await tenant(), first = await job(a), second = await job(a);
     const target = await job(a, { sourceJobIds: [first.id, second.id], states: ["succeeded", "stopped"] });