diff --git a/landing-zone-configurator/app/apps/api/src/plans/runner-pool.ts b/landing-zone-configurator/app/apps/api/src/plans/runner-pool.ts index 050d495e..797c4781 100644 --- a/landing-zone-configurator/app/apps/api/src/plans/runner-pool.ts +++ b/landing-zone-configurator/app/apps/api/src/plans/runner-pool.ts @@ -1,4 +1,5 @@ import { createHash, randomBytes, randomUUID } from "node:crypto"; +import { setTimeout as sleep } from "node:timers/promises"; import { type RunnerPoolAssignment, runnerPoolAssignmentSchema, @@ -174,23 +175,36 @@ export class RunnerPool implements WarmRunnerDispatch { if (origin !== this.config.origin) throw new Error("runner_pool_origin_mismatch"); const assignment = runnerPoolAssignmentSchema.parse({ id, ticket, broker }); - const slot = await this.transaction(async (client) => { - await client.query( - "SELECT pg_advisory_xact_lock(hashtextextended($1,0))", - [`runner-pool-job:${id}`], - ); - const existing = await client.query( - "SELECT id FROM lzc.runner_pool_slots WHERE job_id=$1", - [id], - ); - if (existing.rowCount) - throw new Error("runner_pool_job_already_dispatched"); - const reserved = await client.query( - `WITH candidate AS (SELECT id FROM lzc.runner_pool_slots WHERE pool_key=$1 AND status='ready' AND last_seen_at>now()-interval '10 seconds' AND expires_at>now()+interval '30 seconds' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1) UPDATE lzc.runner_pool_slots s SET status='reserved',job_id=$2,is_probe=$3,expires_at=now()+interval '30 seconds' FROM candidate WHERE s.id=candidate.id RETURNING s.id,s.app_id AS "appId",s.source_droplet_id AS "sourceDropletId"`, - [this.key, id, probe], + const reserve = () => + this.transaction(async (client) => { + await client.query( + "SELECT pg_advisory_xact_lock(hashtextextended($1,0))", + [`runner-pool-job:${id}`], + ); + const existing = await client.query( + "SELECT id FROM lzc.runner_pool_slots WHERE job_id=$1", + [id], + ); + if (existing.rowCount) + throw new Error("runner_pool_job_already_dispatched"); + const reserved = await client.query( + `WITH candidate AS (SELECT id FROM lzc.runner_pool_slots WHERE pool_key=$1 AND status='ready' AND last_seen_at>now()-interval '10 seconds' AND expires_at>now()+interval '30 seconds' ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 1) UPDATE lzc.runner_pool_slots s SET status='reserved',job_id=$2,is_probe=$3,expires_at=now()+interval '30 seconds' FROM candidate WHERE s.id=candidate.id RETURNING s.id,s.app_id AS "appId",s.source_droplet_id AS "sourceDropletId"`, + [this.key, id, probe], + ); + return reserved.rows[0]; + }); + let slot = await reserve(); + // A heartbeat briefly locks the same row. Retry only a known, still-ready + // candidate skipped before reservation; never replay an assignment or error. + for (let attempt = 0; !slot && attempt < 4; attempt++) { + const eligible = await this.database.query<{ ready: boolean }>( + "SELECT EXISTS(SELECT 1 FROM lzc.runner_pool_slots WHERE pool_key=$1 AND status='ready' AND last_seen_at>now()-interval '10 seconds' AND expires_at>now()+interval '30 seconds') AS ready", + [this.key], ); - return reserved.rows[0]; - }); + if (!eligible.rows[0]?.ready) break; + await sleep(50); + slot = await reserve(); + } if (!slot) return false; try { if (!slot.appId) throw new Error("runner_pool_app_missing"); diff --git a/landing-zone-configurator/app/integration/runner-pool.test.ts b/landing-zone-configurator/app/integration/runner-pool.test.ts index 6e86f2ad..c478ab8b 100644 --- a/landing-zone-configurator/app/integration/runner-pool.test.ts +++ b/landing-zone-configurator/app/integration/runner-pool.test.ts @@ -1,5 +1,6 @@ import { randomBytes, randomUUID } from "node:crypto"; import { readFile } from "node:fs/promises"; +import { setTimeout as sleep } from "node:timers/promises"; import { fileURLToPath } from "node:url"; import pg from "pg"; import { afterAll, beforeAll, beforeEach, expect, it, vi } from "vitest"; @@ -162,6 +163,81 @@ it("leases a standby to at most one concurrent job", async () => { expect(assigned?.id).toBe(assignments[results.indexOf(true)]?.id); }); +it("uses a ready standby after a brief concurrent claim lock", async () => { + const pool = create(); + const slot = await ready(pool); + const claimant = await database.connect(); + let release: Promise | undefined; + try { + await claimant.query("BEGIN"); + await claimant.query( + "SELECT id FROM lzc.runner_pool_slots WHERE id=$1 FOR UPDATE", + [slot.id], + ); + const assignment = job(); + const record = vi.fn(async () => {}); + release = sleep(100).then(() => claimant.query("COMMIT")); + expect( + await pool.start( + assignment.id, + assignment.ticket, + config.origin, + "application", + record, + ), + ).toBe(true); + expect(record).toHaveBeenCalledExactlyOnceWith( + slot.appId, + config.dropletId, + ); + expect(await pool.claim(slot.id, slot.ticket)).toEqual({ + ...assignment, + broker: "application", + }); + } finally { + await release; + await claimant.query("ROLLBACK"); + claimant.release(); + } +}); + +it("falls back without assigning a standby whose claim lock remains held", async () => { + const pool = create(); + const slot = await ready(pool); + const claimant = await database.connect(); + try { + await claimant.query("BEGIN"); + await claimant.query( + "SELECT id FROM lzc.runner_pool_slots WHERE id=$1 FOR UPDATE", + [slot.id], + ); + const assignment = job(); + const record = vi.fn(async () => {}); + expect( + await pool.start( + assignment.id, + assignment.ticket, + config.origin, + "application", + record, + ), + ).toBe(false); + expect(record).not.toHaveBeenCalled(); + const stored = await claimant.query( + "SELECT status,job_id,assignment FROM lzc.runner_pool_slots WHERE id=$1", + [slot.id], + ); + expect(stored.rows[0]).toEqual({ + status: "ready", + job_id: null, + assignment: null, + }); + } finally { + await claimant.query("ROLLBACK"); + claimant.release(); + } +}); + it("serializes pool capacity across API replicas", async () => { const first = create(); const second = create(); diff --git a/landing-zone-configurator/docs/plan-execution.de.md b/landing-zone-configurator/docs/plan-execution.de.md index a0466640..12252902 100644 --- a/landing-zone-configurator/docs/plan-execution.de.md +++ b/landing-zone-configurator/docs/plan-execution.de.md @@ -32,6 +32,13 @@ Planberechnung. Gespeicherte Vorbereitung und Apply-Freigabe bleiben unveränder Standard ist Kapazität null; siehe [Deployment-Steuerung](../deploy/cloud-foundry/README.de.md). +Eine gleichzeitige Bereitschaftsmeldung kann einen bereiten Eintrag kurz sperren. +Vor dem Kaltstart versucht der Dispatch die Reservierung höchstens vier weitere +Male im Abstand von 50 ms, solange ein frischer Kandidat im selben an die Quelle +gebundenen Pool sichtbar bleibt. Ohne Reserve sowie bei veralteter oder bereits +zugewiesener Reserve gilt weiterhin der Kaltstart. Dieses begrenzte Warten +wiederholt niemals eine gespeicherte Zuweisung oder einen unklaren Datenbankfehler. + Die Bereinigung eines Application-Runners erfordert einen dauerhaft gespeicherten Dispatch mit `succeeded` oder `failed` und die exakte Zuordnung von Job, App und Quell-Droplet. Die Pool-Wartung entfernt danach die einmalige CF-App, auch bei diff --git a/landing-zone-configurator/docs/plan-execution.md b/landing-zone-configurator/docs/plan-execution.md index 60c74932..7131ac37 100644 --- a/landing-zone-configurator/docs/plan-execution.md +++ b/landing-zone-configurator/docs/plan-execution.md @@ -24,6 +24,12 @@ validation or planning. It changes neither the saved preparation nor Apply approval. Pool capacity defaults to zero; see [the deployment controls](../deploy/cloud-foundry/README.md). +A concurrent readiness heartbeat can briefly lock a ready slot. Before falling +back, dispatch makes at most four further reservation attempts, 50 ms apart, and +only while a fresh candidate in the same source-bound pool remains visible. +A missing, stale or already assigned reserve still uses the cold path. This +bounded wait never replays a recorded assignment or an ambiguous database error. + Application Runner cleanup requires a durably saved `succeeded` or `failed` dispatch and the exact job/app/source-droplet binding. Pool maintenance then removes the one-shot CF app, including completed runners left by an older release. Running jobs and