From 1569c6c99642e5e68e570fb06d446301f33690b2 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Fri, 30 Oct 2026 21:34:06 +0000 Subject: [PATCH] Prevent idle triggers before lifecycle delivery --- .../api/src/postgres/automation-scheduler.ts | 45 ++++++++++--- .../api/test/job-lifecycle.postgres.test.ts | 67 +++++++++++++++++++ 2 files changed, 104 insertions(+), 8 deletions(-) diff --git a/packages/api/src/postgres/automation-scheduler.ts b/packages/api/src/postgres/automation-scheduler.ts index df07b5b..efafda7 100644 --- a/packages/api/src/postgres/automation-scheduler.ts +++ b/packages/api/src/postgres/automation-scheduler.ts @@ -53,7 +53,11 @@ export class AutomationScheduler { } private async lifecycle() { - const rows = await this.database.pool.query(`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(`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) { diff --git a/packages/api/test/job-lifecycle.postgres.test.ts b/packages/api/test/job-lifecycle.postgres.test.ts index 3e6fa03..f00d329 100644 --- a/packages/api/test/job-lifecycle.postgres.test.ts +++ b/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"] });