mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-23 10:36:09 +00:00
fix: one manager-owned owner-lease span across every abort delivery path
Seventh external review round; four lifecycle-fence gaps, closed by making the owner-lifecycle lease a single manager-owned, heartbeat-held span. - Fail-closed acknowledgements. The pre-ACK hook registered a one-shot lease and the transport ACKed even when it failed; a same-replica owned abort likewise swallowed registration failure. The hook now acquires a HELD owner lease (heartbeat until the owner's catch releases it via releaseOwnerLease) and a rejection SUPPRESSES the acknowledgement — the stopping side stays retryable behind its retention lease, and every resignal re-drives the handler. The hook also stopped gating on `job.createdAt === generationId`: during a replacement handoff the store holds the replacement while the abort targets the predecessor, and that gate silently skipped exactly the generation being acknowledged (the store job is owner identity, never a generation gate). A local owned abort acquires the same held lease before tripping its provider; post-CAS the trip cannot be withheld, so acquisition failure downgrades delivery and the retention handoff keeps the user fenced. - Committed-but-lost-reply disambiguation. A thrown abort transition released the contender lease as if nothing had happened, but a Lua CAS can commit and lose its reply — an aborted job invisible to active-set scans whose provider was never signalled, with no fence left. The throw path now re-reads the exact generation: only a job still live under the caller's identity proves no commit; committed or ambiguous outcomes hand the fence to the deterministic stop lease (kept on the contender lease if even that fails) before rethrowing. - Replacement handoff covered end to end. A LOCAL replacement abort acquires the predecessor's held owner lease before the trip (failure reports the receipt undelivered, engaging retention). Failed-handoff retention is no longer a swallowed one-shot: it heartbeats with the durable acknowledgement proof as its renewal predicate — acquisition failures keep retrying for as long as the fence is needed, and the retainer stands down (without clearing the shared field) once the remote owner ACKs and thereby holds its own lease. - A LOCAL resignal delivery hands off to the owner lease BEFORE clearing the retained stop lease, and keeps the stop lease when that handoff fails. Barrier tests: hook rejection suppressing the ACK, commit-then-lost-reply retention with its provably-uncommitted counterpart, local-resignal owner handoff ordering, pre-ACK owner lease held past the store TTL until release (and provably stopped after), and failed-handoff retention retrying on its heartbeat — verified fail-before/pass-after by stashing the fixes.
This commit is contained in:
Danny Avila
parent
291ed90765
commit
62ba45a3de
11 changed files with 313 additions and 57 deletions
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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',
|
||||
);
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -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 () => {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
|
|
|
|||
|
|
@ -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<void> }
|
||||
).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');
|
||||
|
|
|
|||
|
|
@ -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<boolean> {
|
||||
return this.hasDurableAbortAck(streamId, generationId);
|
||||
}
|
||||
|
||||
private async hasDurableAbortAck(streamId: string, generationId: number): Promise<boolean> {
|
||||
return (await this.publisher.get(KEYS.abortAck(streamId, generationId))) === '1';
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1466,6 +1466,12 @@ export interface IEventTransport {
|
|||
providerExecutionId: string,
|
||||
): Promise<boolean>;
|
||||
|
||||
/** 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<boolean>;
|
||||
|
||||
/** Publish a predecessor DONE only while the current job's opaque creation
|
||||
* attempt still carries that predecessor in its durable receipt chain. */
|
||||
emitReplacedDoneConfirmed?(
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue