diff --git a/src/server/redis/__tests__/queues.integration.test.ts b/src/server/redis/__tests__/queues.integration.test.ts index e99ee740cb..717e00d3ee 100644 --- a/src/server/redis/__tests__/queues.integration.test.ts +++ b/src/server/redis/__tests__/queues.integration.test.ts @@ -1,8 +1,8 @@ import { PrismaClient } from '@prisma/client'; +import { REDIS_SYS_KEYS } from '@civitai/redis/client'; import { createClient } from 'redis'; import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; -// Top-level, not an inline `typeof import(...)` — that trips consistent-type-imports. -import type * as SysReadDeadline from '~/server/redis/sys-read-deadline'; +import { dbMock, redisMock } from '~/__tests__/mocks'; /** * End-to-end proof for the dropped-enqueue parking lot, against a REAL Postgres and a @@ -27,54 +27,39 @@ const redisUrl = process.env.QUEUES_IT_REDIS_URL; const SCHEMA = 'queues_it'; +const NS = `queues-it:${process.pid}`; + // `backing = null` is the outage: every command rejects fast, which is the DOWN mode // queues.ts fails open on. Swapping it back is the recovery the drain has to survive. -const { holder, NS } = vi.hoisted(() => ({ - holder: { backing: null as null | Record Promise> }, - NS: `queues-it:${process.pid}`, -})); +let backing: typeof redis | null = null; -vi.mock('~/server/redis/client', async () => { - const { withSysReadDeadline } = await vi.importActual( - '~/server/redis/sys-read-deadline' - ); - const call = - (fn: string) => - (...args: never[]) => - holder.backing - ? holder.backing[fn](...args) - : Promise.reject(new Error('sysRedis unavailable (test outage)')); - return { - sysRedis: { - hGet: call('hGet'), - hSet: call('hSet'), - sAdd: call('sAdd'), - sMembers: call('sMembers'), - del: call('del'), - exists: call('exists'), - set: call('set'), - }, - REDIS_SYS_KEYS: { QUEUES: { BUCKETS: `${NS}:buckets` } }, - REDIS_SUB_KEYS: { QUEUES: { MERGING: 'merging' } }, - withSysReadDeadline, - }; -}); +/** + * 🔴 This file drives the CANONICAL shared mocks rather than registering its own. + * + * `~/server/db/client`, `~/server/redis/client` and `~/server/logging/client` all have + * canonical mocks registered once in `src/__tests__/setup.ts`, and a per-file `vi.mock` of + * any of them freezes that shape into every later file in the same worker under + * `isolate: false` — which is what `no-direct-shared-module-mock` exists to stop, and why + * `gen-mock-allowlist.mjs` refuses to allowlist a new file. + * + * Being an INTEGRATION test is not an exemption from that, only a constraint on how the + * seam is used: the canonical nodes are `vi.fn()`s, so instead of declaring canned return + * values this file points them at the REAL Postgres and Redis clients it builds in + * `beforeAll`. That satisfies the guard and keeps the suite end-to-end. + * + * It also removes the hand-typed key constants this file used to carry — `setup.ts` spreads + * the real `@civitai/redis/client` into the canonical factory, so `REDIS_SYS_KEYS` and + * `REDIS_SUB_KEYS` are production's own values and cannot drift + * (`no-hand-typed-redis-key-constants`). `withSysReadDeadline` likewise defaults to the real + * implementation there, so queues.ts keeps running the real deadline wrapper. + */ +// `~/server/redis/fail-open-log` is a PENDING specifier: no canonical mock exists yet, so a +// direct mock is counted rather than enforced. Stubbed because the real logger is noise here. vi.mock('~/server/redis/fail-open-log', () => ({ logSysRedisFailOpen: vi.fn() })); -vi.mock('~/server/logging/client', () => ({ logToAxiom: vi.fn(() => Promise.resolve()) })); let prisma: PrismaClient; let bootstrap: PrismaClient; -// A getter, not a value: the client cannot exist until beforeAll has the URL, and -// queues.ts captures the binding at import. -vi.mock('~/server/db/client', () => ({ - get dbWrite() { - return prisma; - }, - get dbRead() { - return prisma; - }, -})); const { addToQueue, drainDroppedEnqueues } = await import('~/server/redis/queues'); @@ -100,6 +85,36 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres redis = createClient({ url: redisUrl }); await redis.connect(); + + // Point the canonical db mock at the REAL scratch-schema client. queues.ts calls these as + // TAGGED TEMPLATES, so the implementation forwards (strings, ...values) verbatim — which + // is exactly the binding this suite exists to prove (the `::jsonb` cast, and + // `jsonb_array_length` on what we actually store). + dbMock.dbWrite.$queryRaw.mockImplementation((sql: TemplateStringsArray, ...values: unknown[]) => + prisma.$queryRaw(sql, ...values) + ); + dbMock.dbWrite.$executeRaw.mockImplementation( + (sql: TemplateStringsArray, ...values: unknown[]) => prisma.$executeRaw(sql, ...values) + ); + + // Point the canonical sysRedis mock at the REAL redis. Every key this suite writes derives + // from BUCKETS, so prefixing arg 0 namespaces the whole run at the transport layer while + // leaving the PRODUCTION key names in play — the mirror of the scratch SCHEMA on the + // Postgres side, where the SQL stays unqualified and the isolation lives on the connection. + // `del` accepts an array of keys as well as one. + const nsKey = (key: unknown) => + Array.isArray(key) ? key.map((k) => `${NS}:${k}`) : `${NS}:${key}`; + const COMMANDS = ['hGet', 'hSet', 'sAdd', 'sMembers', 'del', 'exists', 'set'] as const; + for (const cmd of COMMANDS) { + redisMock.sysRedis[cmd].mockImplementation((key: unknown, ...rest: unknown[]) => + backing + ? (backing as unknown as Record Promise>)[cmd]( + nsKey(key), + ...rest + ) + : Promise.reject(new Error('sysRedis unavailable (test outage)')) + ); + } }, 30000); afterAll(async () => { @@ -119,7 +134,7 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres await prisma.$executeRawUnsafe(`TRUNCATE ${SCHEMA}."KeyValue"`); const keys = await redis.keys(`${NS}*`); if (keys.length) await redis.del(keys); - holder.backing = redis as never; + backing = redis; }); const parked = () => @@ -127,9 +142,13 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres `SELECT "key","value" FROM ${SCHEMA}."KeyValue" ORDER BY "key"` ); + // Reads through the same `${NS}:` prefix the sysRedis stub applies, against the REAL + // key names. The bucket NAME stored in the hash is a value, not a key, so it is + // unprefixed on the way out and has to be prefixed again to be read back. const bucketMembers = async () => { - const bucket = await redis.hGet(`${NS}:buckets`, 'images_v6:Delete'); - return bucket ? redis.sMembers(bucket) : []; + const bucketsKey = `${NS}:${REDIS_SYS_KEYS.QUEUES.BUCKETS}`; + const bucket = await redis.hGet(bucketsKey, 'images_v6:Delete'); + return bucket ? redis.sMembers(`${NS}:${bucket}`) : []; }; it('a healthy enqueue reaches redis and writes nothing to Postgres', async () => { @@ -140,7 +159,7 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres }); it('an outage parks the ids in Postgres instead of losing them', async () => { - holder.backing = null; + backing = null; await expect(addToQueue('images_v6:Delete', [10, 11])).resolves.toBe(false); @@ -148,7 +167,7 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres }); it('successive drops during one outage accumulate under the same key', async () => { - holder.backing = null; + backing = null; await addToQueue('images_v6:Delete', [10, 11]); await addToQueue('images_v6:Delete', [12]); @@ -156,9 +175,9 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres }); it('the drain replays them into redis once the outage ends, and clears the row', async () => { - holder.backing = null; + backing = null; await addToQueue('images_v6:Delete', [10, 11]); - holder.backing = redis as never; + backing = redis; await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 2, reparked: 0 }); @@ -167,22 +186,22 @@ describe.skipIf(!databaseUrl || !redisUrl)('queues parking lot — real Postgres }); it('the drain leaves the row parked while the outage continues, and never duplicates it', async () => { - holder.backing = null; + backing = null; await addToQueue('images_v6:Delete', [10, 11]); await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 0, reparked: 2 }); // Still exactly the two ids — a re-park would have appended a second copy. expect((await parked())[0].value).toEqual([10, 11]); - holder.backing = redis as never; + backing = redis; await expect(drainDroppedEnqueues()).resolves.toEqual({ keys: 1, replayed: 2, reparked: 0 }); expect(await parked()).toHaveLength(0); }); it('a drop landing between the drain read and its delete is not swallowed', async () => { - holder.backing = null; + backing = null; await addToQueue('images_v6:Delete', [10, 11]); - holder.backing = redis as never; + backing = redis; // The interleaving the length guard exists for: the drain has read [10,11] and is // about to delete when another pod parks id 12. The delete must refuse the grown row. diff --git a/tests/preview-engagement.spec.ts b/tests/preview-engagement.spec.ts index a4c171064b..06bccf3f3b 100644 --- a/tests/preview-engagement.spec.ts +++ b/tests/preview-engagement.spec.ts @@ -22,13 +22,19 @@ import { trpcMutation, uniqueToken } from './preview-trpc'; * - reaction.toggle (reaction.router.ts:9 guardedProcedure .input(toggleReactionSchema)) * — toggleReactionSchema (reaction.schema.ts:37): { entityId: number; * entityType: enum(reactableEntities incl. 'post'); reaction: enum(ReviewReactions) }. - * ReviewReactions.Like = 'Like' (enums.ts:355). The mutation is FIRE-AND-FORGET: - * the router calls toggleReactionHandler(...).catch(handleLogError) and returns - * nothing (reaction.router.ts:13-17), which superjson serializes as `json: null` - * over the wire (NOT undefined — confirmed against a live preview run). Success - * therefore = the call RESOLVES (the helper throws on any HTTP/tRPC error, so - * reaching past it means auth + rate-limit + handler-dispatch all accepted it); - * we assert the resolved value is nullish rather than a specific type. + * ReviewReactions.Like = 'Like' (enums.ts:355). The mutation is AWAITED and + * RESOLVES TO THE HANDLER RESULT: `.mutation(toggleReactionHandler)` + * (reaction.router.ts:15) returns toggleReaction()'s discriminator, one of + * 'created' | 'removed' | 'noop' (reaction.service.ts:25,30 → controller :286). + * This post is self-seeded in this same test and `tester` has never reacted to + * it, so getReaction finds nothing and the create branch runs — the value is + * deterministically 'created', which is what we pin. + * + * 🔴 It used to resolve to null, and this spec used to assert that. `ee376ea7d8` + * (2026-03) made the router fire-and-forget for latency; #4516 / `1e810875c8` + * reverted that because detaching the write from the request lifecycle dropped + * reactions on pod drain. `src/server/routers/__tests__/reaction.router.awaited.test.ts` + * is the unit-level contract test for the restored shape — keep the two in step. * - commentv2.upsert (commentv2.router.ts:62 guardedProcedure .input(upsertCommentv2Schema)) * — upsertCommentv2Schema (commentv2.schema.ts) extends commentConnectorSchema * ({ entityId: number; entityType: enum incl. 'post' }) with a non-empty sanitized @@ -60,18 +66,20 @@ test.describe('tester self-seeds a post, reacts to it, and comments on it (mutat }); expect(typeof post?.id, 'post.create should return a numeric post id').toBe('number'); - // 2. React to that exact post. reaction.toggle is fire-and-forget: it resolves - // to null over the wire (superjson encodes the void return as json:null), so - // success = the call resolves without the helper throwing. A broken - // auth/rate-limit/dispatch path would surface as an HTTP/tRPC error here. - const reaction = await trpcMutation(page.request, 'reaction.toggle', { - entityId: post!.id, - entityType: 'post', - reaction: 'Like', - }); - // `?? null` collapses null/undefined alike so the assertion is robust to the - // exact void encoding — the point is "it was accepted", not its serialized form. - expect(reaction ?? null, 'reaction.toggle should be accepted (resolve, not throw)').toBeNull(); + // 2. React to that exact post. reaction.toggle is awaited and resolves to the + // handler's result, so this asserts the WRITE happened, not merely that the call + // was accepted: a first Like on a post nobody has reacted to takes the create + // branch and resolves to 'created'. A broken auth/rate-limit/dispatch path would + // surface as an HTTP/tRPC error before we get here. + const reaction = await trpcMutation<'created' | 'removed' | 'noop'>( + page.request, + 'reaction.toggle', + { entityId: post!.id, entityType: 'post', reaction: 'Like' } + ); + // Pinned to the exact discriminator rather than a truthiness check: 'noop' and + // 'removed' are also non-null, and both would mean the reaction did NOT get + // created — the regression this spec exists to catch. + expect(reaction, 'reaction.toggle should create the reaction and say so').toBe('created'); // 3. Comment on that exact post. upsert returns the created comment row, so we // assert it came back with a numeric id and that our token survived sanitization