diff --git a/api/server/controllers/agents/__tests__/jobReplacement.spec.js b/api/server/controllers/agents/__tests__/jobReplacement.spec.js index af1cedff20..7b4e5bc4dc 100644 --- a/api/server/controllers/agents/__tests__/jobReplacement.spec.js +++ b/api/server/controllers/agents/__tests__/jobReplacement.spec.js @@ -19,6 +19,7 @@ const mockGenerationJobManager = { // CAS whenever post-terminal title work is possible, so the facade mock must // mirror it or the turn stalls on an undefined call. registerUserFinalization: jest.fn(async () => undefined), + releaseOwnerLease: jest.fn(async () => undefined), holdUserFinalization: jest.fn(async () => ({ leaseId: 'held-lease', release: jest.fn() })), clearUserFinalization: jest.fn(async () => undefined), countUserFinalizations: jest.fn(async () => 0), diff --git a/api/server/controllers/agents/__tests__/request.partialDisconnect.spec.js b/api/server/controllers/agents/__tests__/request.partialDisconnect.spec.js index e7f4b22a4f..dceac3d2b5 100644 --- a/api/server/controllers/agents/__tests__/request.partialDisconnect.spec.js +++ b/api/server/controllers/agents/__tests__/request.partialDisconnect.spec.js @@ -26,6 +26,7 @@ const mockGenerationJobManager = { // CAS whenever post-terminal title work is possible, so the facade mock must // mirror it or the turn stalls on an undefined call. registerUserFinalization: jest.fn(async () => undefined), + releaseOwnerLease: jest.fn(async () => undefined), holdUserFinalization: jest.fn(async () => ({ leaseId: 'held-lease', release: jest.fn() })), clearUserFinalization: jest.fn(async () => undefined), countUserFinalizations: jest.fn(async () => 0), diff --git a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js index 5ec5c843d2..75fe1f915c 100644 --- a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js +++ b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js @@ -12,6 +12,7 @@ const mockGenerationJobManager = { // CAS whenever post-terminal title work is possible, so the facade mock must // mirror it or the turn stalls on an undefined call. registerUserFinalization: jest.fn(async () => undefined), + releaseOwnerLease: jest.fn(async () => undefined), holdUserFinalization: jest.fn(async (u, s2, t, g, leaseId) => ({ leaseId, release: mockLeaseRelease, diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 47b7a6c724..77d3f2e07a 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -53,6 +53,7 @@ const mockGenerationJobManager = { // CAS whenever post-terminal title work is possible, so the facade mock must // mirror it or the turn stalls on an undefined call. registerUserFinalization: jest.fn(async () => undefined), + releaseOwnerLease: jest.fn(async () => undefined), holdUserFinalization: jest.fn(async (u, s2, t, g, leaseId) => ({ leaseId, release: mockLeaseRelease, @@ -2223,12 +2224,11 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(mockLeaseRelease.mock.invocationCallOrder[0]).toBeGreaterThan( mockSaveMessage.mock.invocationCallOrder[0], ); - expect(mockGenerationJobManager.clearUserFinalization).toHaveBeenCalledWith( + expect(mockGenerationJobManager.releaseOwnerLease).toHaveBeenCalledWith( USER_ID, CONVO_ID, TENANT_ID, 1000, - 'owner', ); }); diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index db00bbdb9b..6806af4d48 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -1463,16 +1463,18 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit userFinalizationHold = null; hold.release(); // The OWNER-lifecycle lease is registered out-of-band (at abort-signal time, by - // the manager / transport pre-ACK fence — see USER_FINALIZATION_OWNER_LEASE) and - // this controller, as the generation owner, releases it with its own: by this - // point every owner-side write has landed or failed for good. - void GenerationJobManager.clearUserFinalization( - userId, - streamId, - req.user?.tenantId, - jobCreatedAt, - 'owner', - ).catch(() => undefined); + // the manager / transport pre-ACK fence) and this controller, as the generation + // owner, releases it with its own: by this point every owner-side write has + // landed or failed for good. releaseOwnerLease also stops the manager-held + // heartbeat when it lives on this process. + if (jobCreatedAt != null) { + void GenerationJobManager.releaseOwnerLease( + userId, + streamId, + req.user?.tenantId, + jobCreatedAt, + ).catch(() => undefined); + } }; const claimBeforeResponsePersistence = async () => { diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 3e47e3bdd5..59dfd06278 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -475,14 +475,12 @@ async function finalizeResumedTurn({ finalizationHold?.release(); finalizationHold = null; // Release the owner-lifecycle lease too: this resumed controller IS the - // generation owner, and by this point its writes have landed or failed for good - // (see USER_FINALIZATION_OWNER_LEASE in the manager). - void GenerationJobManager.clearUserFinalization( + // generation owner, and by this point its writes have landed or failed for good. + void GenerationJobManager.releaseOwnerLease( userId, streamId, req.user?.tenantId, job.createdAt, - 'owner', ).catch(() => undefined); }; @@ -1466,12 +1464,11 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) await settleAbortedScheduledResume(job, streamId, conversationId); // This resumed controller is the generation owner; its abort-path writes have // now landed, so the signal-time owner lease releases here. - void GenerationJobManager.clearUserFinalization( + void GenerationJobManager.releaseOwnerLease( userId, streamId, req.user?.tenantId, job.createdAt, - 'owner', ).catch(() => undefined); return; } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index fd39e6914f..dff095c6ff 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -1442,6 +1442,7 @@ class GenerationJobManagerClass { private async notifyReplacedGeneration( streamId: string, predecessorCreatedAt: number, + owner: { userId: string; tenantId?: string }, predecessorConversationId?: string, fallbackConversationId?: string, replacementCreationAttemptId?: string, @@ -1460,7 +1461,22 @@ class GenerationJobManagerClass { const ownsExactLocalProvider = localPredecessor?.createdAt === predecessorCreatedAt && this.ownedJobs.get(streamId) === predecessorCreatedAt; - const stoppedExactLocalProvider = stopProvider && ownsExactLocalProvider; + let stoppedExactLocalProvider = stopProvider && ownsExactLocalProvider; + if (stoppedExactLocalProvider) { + // A LOCAL replacement abort is a delivery path like any other: the + // predecessor's catch persists asynchronously after this trip, so its owner + // lease must be held first. FAIL CLOSED by reporting the receipt undelivered — + // the failed-handoff retention then keeps the user fenced. + try { + await this.acquireOwnerLease(owner.userId, streamId, owner.tenantId, predecessorCreatedAt); + } catch (leaseError) { + logger.error( + `[GenerationJobManager] Owner lease unavailable for replaced generation ${streamId}:`, + leaseError, + ); + stoppedExactLocalProvider = false; + } + } if (localPredecessor?.createdAt === predecessorCreatedAt) { localPredecessor.finalEvent = reconcileEvent; if (stopProvider) { @@ -1628,18 +1644,12 @@ class GenerationJobManagerClass { // The atomic replacement already removed this predecessor from active // storage, and an unconfirmed handoff terminalizes the replacement too — // leaving NOTHING a deletion quiesce could discover while the - // predecessor's provider may still be generating. Retain its owner lease; - // the owner replica renews it through the pre-ACK fence when the signal - // finally lands, and the TTL bounds a dead owner. - await this.jobStore - .registerUserFinalization?.( - job.userId, - streamId, - job.tenantId, - receipt.createdAt, - USER_FINALIZATION_OWNER_LEASE, - ) - .catch(() => undefined); + // predecessor's provider may still be generating. Retain its owner lease + // on a heartbeat whose renewal predicate is the durable acknowledgement + // proof: acquisition failures keep retrying rather than degrading to a + // one-shot, and the retainer stands down once the owner ACKs (and thereby + // holds its own lease through the pre-ACK fence). + this.retainPredecessorOwnerLease(job.userId, streamId, job.tenantId, receipt.createdAt); } } } @@ -2279,6 +2289,7 @@ class GenerationJobManagerClass { await this.notifyReplacedGeneration( streamId, predecessorCreatedAt, + { userId: jobData.userId, tenantId: jobData.tenantId }, predecessorConversationId, conversationId, ); @@ -3817,16 +3828,32 @@ class GenerationJobManagerClass { .catch(() => undefined); } else { // Delivery finally left (or this replica owns the generation): the retained - // stop lease has served its purpose — reap it instead of waiting out the TTL. - await this.jobStore - .clearUserFinalization?.( + // stop lease has served its purpose. A LOCAL delivery must hand off to the + // owner lease FIRST — the owner's catch persistence is still ahead — and the + // stop lease stays if that handoff fails (fail closed). A published delivery + // was fenced by the pre-ACK hook on the owner's replica. + let handedOff = true; + if (delivered) { + handedOff = await this.acquireOwnerLease( jobData.userId, streamId, jobData.tenantId, jobData.createdAt, - USER_FINALIZATION_STOP_LEASE, ) - .catch(() => undefined); + .then(() => true) + .catch(() => false); + } + if (handedOff) { + await this.jobStore + .clearUserFinalization?.( + jobData.userId, + streamId, + jobData.tenantId, + jobData.createdAt, + USER_FINALIZATION_STOP_LEASE, + ) + .catch(() => undefined); + } } return { delivered, published }; } @@ -4089,9 +4116,44 @@ class GenerationJobManagerClass { transitionFrom = currentAfterConflict.status; } } catch (transitionError) { - // A THROWN transition (store outage, not a lost CAS) aborted nothing: release - // the contender lease instead of leaking it to its TTL. - clearAbortFinalization(); + // A THROW is not proof the CAS did not commit: a Lua transaction can commit + // and lose its reply, leaving an aborted job invisible to active-set scans + // whose provider was never signalled. Disambiguate by re-reading the exact + // generation; only a job still live under this identity proves no commit. + let observedAfterThrow: SerializableJobData | null | undefined; + try { + observedAfterThrow = await this.jobStore.getJob(streamId); + } catch { + observedAfterThrow = undefined; + } + const provablyUncommitted = + observedAfterThrow != null && + observedAfterThrow.createdAt === jobData.createdAt && + (observedAfterThrow.status === 'running' || + observedAfterThrow.status === 'requires_action'); + if (provablyUncommitted) { + clearAbortFinalization(); + } else { + // Committed or ambiguous: hand the fence to the deterministic stop lease + // (renewed by every resignal attempt) so the retry path can re-drive + // delivery; keep the contender lease if even that handoff fails. + const handedOff = await this.jobStore + .registerUserFinalization?.( + jobData.userId, + streamId, + jobData.tenantId, + jobData.createdAt, + USER_FINALIZATION_STOP_LEASE, + ) + .then(() => true) + .catch(() => false); + if (handedOff) { + clearAbortFinalization(); + } + logger.error( + `[GenerationJobManager] Abort transition threw with committed/ambiguous outcome for ${streamId}; retaining fence`, + ); + } throw transitionError; } if (drainedSteers == null) { @@ -4144,7 +4206,7 @@ class GenerationJobManagerClass { // so a job observed `requires_action` can be aborted FROM `running`. Treating that // as a paused job made delivery vacuously true for a live generation this replica // does not own — reporting a stop no generating process ever received. - const abortSignalDelivered = + let abortSignalDelivered = transitionFrom === 'requires_action' || this.ownedJobs.get(streamId) === jobData.createdAt; let abortSignalPublished = true; @@ -4171,21 +4233,29 @@ class GenerationJobManagerClass { if (runtime) { this.releaseAbortSubscription(runtime); } - // OWNER-LIFECYCLE LEASE before the trip, mirroring the transport's pre-ACK - // fence: this abort's own lease clears when the call returns, but the owned - // generation's catch persists asynchronously after — the deterministic owner - // lease bridges that gap and the owner's catch releases it once its writes - // land. Best-effort: the contender lease still covers this call's window. + // OWNER-LIFECYCLE LEASE (held) before the trip, mirroring the transport's + // pre-ACK fence: this abort's own lease clears when the call returns, but the + // owned generation's catch persists asynchronously after — the owner lease + // bridges that gap and the owner's catch releases it once its writes land. + // FAIL CLOSED post-CAS: the trip itself cannot be withheld (the job is already + // terminal and its provider must stop), so an unacquirable lease instead + // downgrades delivery — the finally's retention handoff then keeps the user + // fenced and the Stop route stays retryable. if (abortSignalDelivered && runtime?.createdAt === jobData.createdAt) { - await this.jobStore - .registerUserFinalization?.( + try { + await this.acquireOwnerLease( jobData.userId, streamId, jobData.tenantId, jobData.createdAt, - USER_FINALIZATION_OWNER_LEASE, - ) - .catch(() => undefined); + ); + } catch (leaseError) { + logger.error( + `[GenerationJobManager] Owner lease unavailable for local abort of ${streamId}; downgrading delivery:`, + leaseError, + ); + abortSignalDelivered = false; + } } runtime?.abortController.abort(); diff --git a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts index f2a3ac74d3..532371f18d 100644 --- a/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts +++ b/packages/api/src/stream/__tests__/RedisEventTransport.spec.ts @@ -204,6 +204,37 @@ describe('RedisEventTransport', () => { transport.destroy(); }); + it('SUPPRESSES the abort acknowledgement when the pre-ACK fence rejects', async () => { + const mockPublisher = createMockPublisher(); + const mockSubscriber = createMockSubscriber(); + const transport = new RedisEventTransport( + mockPublisher as unknown as Redis, + mockSubscriber as unknown as Redis, + ); + const streamId = 'abort-ack-fence-reject'; + // FAIL CLOSED: an ACK without a durable owner lease lets the stopping side + // release its lease into an unfenced gap. The stopping side stays retryable and + // every resignal re-drives this handler until the fence holds. + transport.beforeAbortAcknowledged = jest.fn(async () => { + throw new Error('owner lease unavailable'); + }); + await transport.onAbort(streamId, () => true); + const messageHandler = getMessageHandler(mockSubscriber); + + messageHandler( + `stream:{${streamId}}:events`, + JSON.stringify({ type: 'abort', generationId: 7, abortRequestId: 'req-1' }), + ); + await new Promise((resolve) => setImmediate(resolve)); + + expect(mockPublisher.set).not.toHaveBeenCalled(); + const ackPublishes = mockPublisher.publish.mock.calls.filter(([, payload]) => + String(payload).includes('abort_ack'), + ); + expect(ackPublishes).toHaveLength(0); + transport.destroy(); + }); + it('ignores sequenced events while only internal abort listeners are attached', async () => { jest.useFakeTimers(); const mockPublisher = createMockPublisher(); diff --git a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts index afb0476d30..d9962fb73e 100644 --- a/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts +++ b/packages/api/src/stream/__tests__/abortSignalHonesty.spec.ts @@ -159,15 +159,18 @@ describe('abort signal honesty across replicas', () => { await owner.createJob('conv-7', 'user-1', 'conv-7'); const observed = await jobStore.getJob('conv-7'); // A replacement generation claims the stream before the peer's abort lands. + // (The replacement handoff itself holds the predecessor's owner lease — that + // baseline is the replacement path's designed fence, not this abort's.) await owner.createJob('conv-7', 'user-1', 'conv-7'); + const baseline = await jobStore.countUserFinalizations('user-1'); const result = await peer.abortJob('conv-7', { expectedCreatedAt: observed!.createdAt }); expect(result.success).toBe(false); - // Nothing will be persisted by this loser: holding the marker to its TTL would - // fence the user's deletion pointlessly. + // Nothing will be persisted by this loser: holding ITS marker to the TTL would + // fence the user's deletion pointlessly. Only the replacement baseline remains. await new Promise((resolve) => setImmediate(resolve)); - await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(baseline); }); it('RETAINS the abort lease while the stop is undelivered and unpublished', async () => { @@ -332,6 +335,143 @@ describe('abort signal honesty across replicas', () => { } }); + it('retains a fence when the abort transition commits but loses its reply', async () => { + const { owner, peer, jobStore } = await makeManagers(); + await owner.createJob('conv-14', 'user-1', 'conv-14'); + // The Lua CAS commits, then the reply is lost: the job is aborted (invisible to + // active-set scans) but no signal was ever sent. Clearing the contender lease on + // the throw — as if nothing happened — left NOTHING fencing the user. + const realTransition = jobStore.transitionStatusAndDrainSteers.bind(jobStore); + jest + .spyOn(jobStore, 'transitionStatusAndDrainSteers') + .mockImplementation(async (streamId, args) => { + await realTransition(streamId, args); + throw new Error('connection reset before reply'); + }); + + await expect(peer.abortJob('conv-14')).rejects.toThrow('connection reset'); + + expect((await jobStore.getJob('conv-14'))?.status).toBe('aborted'); + await new Promise((resolve) => setImmediate(resolve)); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBeGreaterThanOrEqual(1); + }); + + it('clears the contender lease only when the thrown transition provably did not commit', async () => { + const { owner, peer, jobStore } = await makeManagers(); + await owner.createJob('conv-15', 'user-1', 'conv-15'); + jest + .spyOn(jobStore, 'transitionStatusAndDrainSteers') + .mockRejectedValue(new Error('store outage')); + + await expect(peer.abortJob('conv-15')).rejects.toThrow('store outage'); + + // The re-read shows the generation still live: the CAS did not commit, so the + // contender lease releases instead of leaking to its TTL. + expect((await jobStore.getJob('conv-15'))?.status).toBe('running'); + await new Promise((resolve) => setImmediate(resolve)); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0); + }); + + it('hands a LOCAL resignal delivery to the owner lease before clearing the stop lease', async () => { + const { owner, peer, peerTransport, jobStore } = await makeManagers(); + await owner.createJob('conv-16', 'user-1', 'conv-16'); + Object.assign(peerTransport, { + emitAbort: jest.fn((): void => undefined), + emitAbortConfirmed: jest.fn(async () => false), + }); + await peer.abortJob('conv-16'); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(1); + + // The OWNER's replica retries the resignal: delivery is local, its catch + // persistence is still ahead — the stop lease may only clear once the owner + // lease holds. + const register = jest.spyOn(jobStore, 'registerUserFinalization'); + const resignal = await owner.resignalAbort('conv-16'); + + expect(resignal.delivered).toBe(true); + expect(register).toHaveBeenCalledWith( + 'user-1', + 'conv-16', + undefined, + expect.any(Number), + 'owner', + ); + await new Promise((resolve) => setImmediate(resolve)); + // Exactly the owner lease remains; the stop lease is reaped. + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(1); + }); + + it('holds the pre-ACK owner lease past the store TTL until the owner releases it', async () => { + const { GenerationJobManagerClass } = await import('../GenerationJobManager'); + const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore'); + const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport'); + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 0 }); + const transport = new InMemoryEventTransport(); + const manager = new GenerationJobManagerClass(); + manager.configure({ jobStore, eventTransport: transport, isRedis: false }); + manager.initialize(); + await manager.createJob('conv-17', 'user-1', 'conv-17'); + const createdAt = (await jobStore.getJob('conv-17'))!.createdAt; + + jest.useFakeTimers({ now: Date.now() }); + try { + // The transport hook is how the owner replica acquires the lease. + await ( + transport as { beforeAbortAcknowledged?: (s: string, g: number) => Promise } + ).beforeAbortAcknowledged?.('conv-17', createdAt); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(1); + + // Owner-side persistence can legitimately outlive the five-minute store TTL. + await jest.advanceTimersByTimeAsync(6 * 60 * 1000); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(1); + + await manager.releaseOwnerLease('user-1', 'conv-17', undefined, createdAt); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0); + // And the heartbeat is genuinely stopped, not just cleared once. + await jest.advanceTimersByTimeAsync(6 * 60 * 1000); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBe(0); + } finally { + jest.useRealTimers(); + } + }); + + it('retries failed-handoff retention on the heartbeat instead of degrading to one-shot', async () => { + const { GenerationJobManagerClass } = await import('../GenerationJobManager'); + const { InMemoryJobStore } = await import('../implementations/InMemoryJobStore'); + const { InMemoryEventTransport } = await import('../implementations/InMemoryEventTransport'); + const jobStore = new InMemoryJobStore({ ttlAfterComplete: 0 }); + const owner = new GenerationJobManagerClass(); + owner.configure({ jobStore, eventTransport: new InMemoryEventTransport(), isRedis: false }); + owner.initialize(); + const peer = new GenerationJobManagerClass(); + const peerTransport = new InMemoryEventTransport(); + Object.assign(peerTransport, { emitAbortConfirmed: jest.fn(async () => false) }); + peer.configure({ jobStore, eventTransport: peerTransport, isRedis: true }); + peer.initialize(); + await owner.createJob('conv-18', 'user-1', 'conv-18'); + + // Retention registration fails at first — the retainer must keep trying on its + // heartbeat, not silently give up while the predecessor is undiscoverable. + const realRegister = jobStore.registerUserFinalization.bind(jobStore); + let failures = 0; + jest.spyOn(jobStore, 'registerUserFinalization').mockImplementation(async (...args) => { + if (args[4] === 'owner' && failures++ < 1) { + throw new Error('marker store blip'); + } + return realRegister(...args); + }); + jest.useFakeTimers({ now: Date.now() }); + try { + await expect(peer.createJob('conv-18', 'user-1', 'conv-18')).rejects.toThrow( + 'predecessor handoff', + ); + await jest.advanceTimersByTimeAsync(61_000); + await expect(jobStore.countUserFinalizations('user-1')).resolves.toBeGreaterThanOrEqual(1); + } finally { + jest.useRealTimers(); + } + }); + it('reports republication as succeeded when the owner acknowledges', async () => { const { owner, peer, peerTransport, jobStore } = await makeManagers(); await owner.createJob('conv-4', 'user-1', 'conv-4'); diff --git a/packages/api/src/stream/implementations/RedisEventTransport.ts b/packages/api/src/stream/implementations/RedisEventTransport.ts index 38aeb4163f..44224bcf34 100644 --- a/packages/api/src/stream/implementations/RedisEventTransport.ts +++ b/packages/api/src/stream/implementations/RedisEventTransport.ts @@ -948,16 +948,18 @@ export class RedisEventTransport implements IEventTransport { void (async () => { // The owner-lifecycle lease must be durable BEFORE the acknowledgement: // the stopping side releases its own lease the moment this ACK arrives, - // and the owner's abort-catch persistence is still ahead. Best-effort — - // an unregistrable lease must not suppress the ACK forever (the stopping - // side's retention lease and resignal heartbeat cover that outage). + // and the owner's abort-catch persistence is still ahead. FAIL CLOSED — a + // rejected fence suppresses the acknowledgement entirely; the stopping + // side stays retryable and its retention lease keeps the user fenced, + // while every resignal re-drives this handler until the fence holds. try { await this.beforeAbortAcknowledged?.(streamId, generationId); } catch (error) { logger.error( - `[RedisEventTransport] Failed to fence abort acknowledgement for ${streamId}:`, + `[RedisEventTransport] Suppressing abort acknowledgement for ${streamId} — owner fence failed:`, error, ); + return; } await this.publishAbortAcknowledgement(streamId, generationId, abortRequestId); })(); @@ -1270,6 +1272,11 @@ export class RedisEventTransport implements IEventTransport { this.unsubscribeUnusedChannel(streamId, state); } + /** Public form of the durable owner-acknowledgement probe (see IEventTransport). */ + async hasAbortAcknowledgement(streamId: string, generationId: number): Promise { + return this.hasDurableAbortAck(streamId, generationId); + } + private async hasDurableAbortAck(streamId: string, generationId: number): Promise { return (await this.publisher.get(KEYS.abortAck(streamId, generationId))) === '1'; } diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 0657a72cb6..3c5cd56edd 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -1466,6 +1466,12 @@ export interface IEventTransport { providerExecutionId: string, ): Promise; + /** Whether the durable owner acknowledgement proof exists for this exact + * generation. Cross-process retention leases use it as their renewal predicate: + * once the owner has acknowledged (and therefore holds its own lease), a remote + * retainer stops renewing without clearing the shared owner field. */ + hasAbortAcknowledgement?(streamId: string, generationId: number): Promise; + /** Publish a predecessor DONE only while the current job's opaque creation * attempt still carries that predecessor in its durable receipt chain. */ emitReplacedDoneConfirmed?(