From cbc9a22f5f44f7830e71ffaaeb45f2d9d49d2ef7 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Thu, 30 Jul 2026 08:27:10 -0400 Subject: [PATCH] test: cover the cross-replica preempt hop with two manager instances MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every other preempt test runs against a single manager, so the hop that actually carries an interrupt in production had no coverage: the steer POST lands on whichever replica the balancer picks, which is usually not the one generating. Non-owner publishes, owner arms, owner's level-triggered poll flips — none of that was exercised end to end. Two GenerationJobManagerClass instances are a faithful replica pair here. runtimeState and ownedJobs are private instance fields, there is no module-level mutable state between them, and createStreamServices duplicates a dedicated subscriber connection per call, so separate OS processes would exercise the same objects over the same Redis. Both assertions verified counterfactually against real Redis: - Deleting the preemptCapable deserialization in RedisJobStore fails this with 'Expected: true, Received: undefined' — the exact P1 that shipped past every in-memory test and would have made the feature a silent no-op on every Redis deployment. - Dropping the non-owner arm publish fails it with 'Received: false'. --- ...ationJobManager.stream_integration.spec.ts | 94 +++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts b/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts index 91f3d656d1..be72965156 100644 --- a/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/GenerationJobManager.stream_integration.spec.ts @@ -2416,4 +2416,98 @@ describe('GenerationJobManager Integration Tests', () => { expect(services.isRedis).toBe(false); }); }); + + /** + * The production topology for interrupt & steer: the steer POST is routed to + * whichever replica the load balancer picks, which is usually NOT the replica + * running the generation. Everything else in the preempt suite runs against a + * single manager, so the hop that actually carries the request — non-owner + * publishes, owner arms, owner's SDK poll flips — has no coverage. + * + * Two `GenerationJobManagerClass` instances are a faithful pair of replicas + * here: `runtimeState` and `ownedJobs` are private instance fields, there is + * no module-level mutable state between them, and `createStreamServices` + * duplicates a dedicated subscriber connection per call. Separate OS + * processes would exercise the same objects over the same Redis. + */ + describeRedis('Cross-Replica Preempt (Redis, two manager instances)', () => { + const services: Array<{ eventTransport: { destroy: () => void } }> = []; + + function createReplica(): GenerationJobManagerClass { + const manager = new GenerationJobManagerClass(); + const config = createStreamServices({ useRedis: true, redisClient: ioredisClient! }); + services.push(config); + manager.configure(config); + manager.initialize(); + return manager; + } + + afterEach(() => { + for (const service of services.splice(0)) { + service.eventTransport.destroy(); + } + }); + + test('a steer routed to a non-owning replica arms the owner and flips its poll', async () => { + const owner = createReplica(); + const router = createReplica(); + const streamId = `${testPrefix}-preempt-xreplica-${Date.now()}`; + + try { + const job = await owner.createJob(streamId, 'user-1', undefined, { + initialMetadata: { preemptCapable: true }, + }); + /** Let the owner's preempt SUBSCRIBE settle — it is fired detached. */ + await new Promise((resolve) => setTimeout(resolve, 300)); + + /** + * The routing replica reads the job, which installs a FACADE runtime + * entry on it. That facade must not make it look like an owner. + */ + const seenByRouter = await router.getJob(streamId); + expect(seenByRouter?.createdAt).toBe(job.createdAt); + expect(router.isPreemptRequested(streamId)).toBe(false); + + /** + * Guards the round-trip that shipped broken once: `preemptCapable` was + * serialized but never deserialized, so every Redis deployment read it + * back as undefined and the feature was a silent no-op. A same-replica + * read cannot catch that — this one crosses Redis. + */ + expect(seenByRouter?.metadata?.preemptCapable).toBe(true); + + await router.requestPreempt(streamId, 'steer-x', job.createdAt); + await new Promise((resolve) => setTimeout(resolve, 300)); + + /** The whole point: the owner's level-triggered poll now reads true. */ + expect(owner.isPreemptRequested(streamId)).toBe(true); + expect(owner.getArmedPreemptIds(streamId)).toContain('steer-x'); + + await router.noteSteersRemoved(streamId, ['steer-x'], job.createdAt); + await new Promise((resolve) => setTimeout(resolve, 300)); + + expect(owner.isPreemptRequested(streamId)).toBe(false); + } finally { + await owner.abortJob(streamId).catch(() => {}); + } + }, 30000); + + test('a stale generation id from another replica cannot arm the live job', async () => { + const owner = createReplica(); + const router = createReplica(); + const streamId = `${testPrefix}-preempt-xreplica-stale-${Date.now()}`; + + try { + const job = await owner.createJob(streamId, 'user-1'); + await new Promise((resolve) => setTimeout(resolve, 300)); + + await router.requestPreempt(streamId, 'steer-stale', job.createdAt - 1); + await new Promise((resolve) => setTimeout(resolve, 300)); + + expect(owner.isPreemptRequested(streamId)).toBe(false); + } finally { + await owner.abortJob(streamId).catch(() => {}); + } + }, 30000); + }); });