From 02189ebb2e262f4fddcef8b2355999269daafedd Mon Sep 17 00:00:00 2001 From: root Date: Wed, 16 Sep 2026 14:20:23 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=8A=A0=E5=9B=BA=E8=81=8A=E5=A4=A9?= =?UTF-8?q?=E9=99=90=E6=B5=81=E4=B8=8E=E5=AE=89=E5=85=A8=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E4=B8=B2=E8=A1=8C=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../conversation/chat-rate-limiter.spec.ts | 56 +++++++++++ .../api/src/conversation/chat-rate-limiter.ts | 9 +- apps/api/src/conversation/chat.gateway.ts | 8 +- .../src/conversation/conversation.e2e-spec.ts | 97 +++++++++++++++++++ .../src/conversation/conversation.module.ts | 3 +- .../src/conversation/conversation.service.ts | 76 ++++++++++----- .../src/conversation/outbox-message-relay.ts | 36 ++++--- apps/api/src/redis/redis.service.ts | 21 ++++ apps/api/src/safety/safety-lock.service.ts | 28 ++++++ apps/api/src/safety/safety.module.ts | 8 ++ 10 files changed, 299 insertions(+), 43 deletions(-) create mode 100644 apps/api/src/conversation/chat-rate-limiter.spec.ts create mode 100644 apps/api/src/safety/safety-lock.service.ts create mode 100644 apps/api/src/safety/safety.module.ts diff --git a/apps/api/src/conversation/chat-rate-limiter.spec.ts b/apps/api/src/conversation/chat-rate-limiter.spec.ts new file mode 100644 index 0000000..6880f50 --- /dev/null +++ b/apps/api/src/conversation/chat-rate-limiter.spec.ts @@ -0,0 +1,56 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { RedisService } from "../redis/redis.service.js"; +import { ChatRateLimiter } from "./chat-rate-limiter.js"; + +type FakeRedisClient = { + isReady: boolean; + isOpen: boolean; + on: ReturnType; + connect: ReturnType; + eval: ReturnType; + destroy: ReturnType; + quit: ReturnType; +}; + +function fakeClient(evalResult: Promise): FakeRedisClient { + return { + isReady: true, + isOpen: true, + on: vi.fn(), + connect: vi.fn(() => Promise.resolve()), + eval: vi.fn(() => evalResult), + destroy: vi.fn(), + quit: vi.fn(() => Promise.resolve()), + }; +} + +describe("ChatRateLimiter Redis command deadline", () => { + const previousTimeout = process.env.REDIS_OPERATION_TIMEOUT_MS; + + afterEach(() => { + if (previousTimeout === undefined) + delete process.env.REDIS_OPERATION_TIMEOUT_MS; + else process.env.REDIS_OPERATION_TIMEOUT_MS = previousTimeout; + }); + + it("destroys a client whose eval never settles and recovers with a new client", async () => { + process.env.REDIS_OPERATION_TIMEOUT_MS = "20"; + const stuck = fakeClient(new Promise(() => undefined)); + const healthy = fakeClient(Promise.resolve(1)); + const factory = vi + .fn() + .mockReturnValueOnce(stuck) + .mockReturnValueOnce(healthy); + const redis = new RedisService(factory as never); + const limiter = new ChatRateLimiter(redis); + + await expect(limiter.consume("account", "session")).rejects.toThrow( + "Auth Redis operation timed out", + ); + expect(stuck.destroy).toHaveBeenCalledOnce(); + + await expect(limiter.consume("account", "session")).resolves.toBe(true); + expect(healthy.eval).toHaveBeenCalledOnce(); + expect(factory).toHaveBeenCalledTimes(2); + }); +}); diff --git a/apps/api/src/conversation/chat-rate-limiter.ts b/apps/api/src/conversation/chat-rate-limiter.ts index 9a83a20..149bea6 100644 --- a/apps/api/src/conversation/chat-rate-limiter.ts +++ b/apps/api/src/conversation/chat-rate-limiter.ts @@ -5,7 +5,6 @@ import { RedisService } from "../redis/redis.service.js"; export class ChatRateLimiter { constructor(@Inject(RedisService) private readonly redis: RedisService) {} async consume(accountId: string, sessionId: string): Promise { - await this.redis.ensureConnected(); const prefix = process.env.REDIS_KEY_PREFIX ?? "drift:auth:"; const max = Number(process.env.CHAT_MESSAGE_RATE_LIMIT ?? 30); const ttl = Number(process.env.CHAT_MESSAGE_RATE_WINDOW_SECONDS ?? 60); @@ -13,9 +12,11 @@ export class ChatRateLimiter { `${prefix}chat:account:${accountId}`, `${prefix}chat:session:${sessionId}`, ]; - const result = await this.redis.client.eval( - `for _,k in ipairs(KEYS) do if tonumber(redis.call('GET',k) or '0')+1>tonumber(ARGV[1]) then return 0 end end; for _,k in ipairs(KEYS) do local n=redis.call('INCR',k); if n==1 then redis.call('EXPIRE',k,ARGV[2]) end end; return 1`, - { keys, arguments: [String(max), String(ttl)] }, + const result = await this.redis.executeWithDeadline((client) => + client.eval( + `for _,k in ipairs(KEYS) do if tonumber(redis.call('GET',k) or '0')+1>tonumber(ARGV[1]) then return 0 end end; for _,k in ipairs(KEYS) do local n=redis.call('INCR',k); if n==1 then redis.call('EXPIRE',k,ARGV[2]) end end; return 1`, + { keys, arguments: [String(max), String(ttl)] }, + ), ); return Number(result) === 1; } diff --git a/apps/api/src/conversation/chat.gateway.ts b/apps/api/src/conversation/chat.gateway.ts index 0353f20..2c6e916 100644 --- a/apps/api/src/conversation/chat.gateway.ts +++ b/apps/api/src/conversation/chat.gateway.ts @@ -99,7 +99,6 @@ export class ChatGateway const user = await this.auth.validateAccess(token); client.data.token = token; client.data.user = user; - await client.join(`account:${user.sub}`); next(); } catch { next(new Error("Unauthorized")); @@ -108,11 +107,14 @@ export class ChatGateway }); } - handleConnection(client: ChatSocket): void { + async handleConnection(client: ChatSocket): Promise { // Single-instance MVP: require a Redis adapter before horizontal scaling. const user = client.data.user; - if (!user || !client.rooms.has(`account:${user.sub}`)) + if (!user) { client.disconnect(true); + return; + } + await client.join(`account:${user.sub}`); } @SubscribeMessage("message:send") diff --git a/apps/api/src/conversation/conversation.e2e-spec.ts b/apps/api/src/conversation/conversation.e2e-spec.ts index b28996b..9e877f7 100644 --- a/apps/api/src/conversation/conversation.e2e-spec.ts +++ b/apps/api/src/conversation/conversation.e2e-spec.ts @@ -14,6 +14,8 @@ import { leaseHmac, encryptLeaseToken } from "../auth/auth.config.js"; import { configureApp } from "../main.js"; import { OutboxMessageRelay } from "./outbox-message-relay.js"; import { ChatRateLimiter } from "./chat-rate-limiter.js"; +import { ConversationService } from "./conversation.service.js"; +import { SafetyLockService } from "../safety/safety-lock.service.js"; process.env.NODE_ENV = "test"; process.env.WEB_ORIGIN = "http://localhost:3000"; @@ -281,6 +283,101 @@ describe("conversation API with real PostgreSQL", () => { } }); + it("linearizes a send before a later block using canonical subject locks", async () => { + const author = await actor("lock-send-author"); + const picker = await actor("lock-send-picker"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const conversationId = created.body.data.conversationId as string; + const locks = app.get(SafetyLockService); + const originalLock = locks.lockAccounts.bind(locks); + let announceLocked!: () => void; + let releaseSend!: () => void; + const sendLocked = new Promise( + (resolve) => (announceLocked = resolve), + ); + const sendRelease = new Promise((resolve) => (releaseSend = resolve)); + let pause = true; + locks.lockAccounts = async (tx, accountIds) => { + await originalLock(tx, accountIds); + if (pause) { + pause = false; + announceLocked(); + await sendRelease; + } + }; + + const send = app + .get(ConversationService) + .send(picker.id, conversationId, randomUUID(), "before block"); + await sendLocked; + let blockCommitted = false; + const block = prisma + .$transaction(async (tx) => { + await locks.lockAccounts(tx, [author.id, picker.id]); + await tx.block.create({ + data: { blockerId: author.id, blockedId: picker.id }, + }); + }) + .then(() => { + blockCommitted = true; + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(blockCommitted).toBe(false); + releaseSend(); + await expect(send).resolves.toMatchObject({ deduplicated: false }); + await block; + locks.lockAccounts = originalLock; + }); + + it("waits for an earlier block and then rejects the message", async () => { + const author = await actor("lock-block-author"); + const picker = await actor("lock-block-picker"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const conversationId = created.body.data.conversationId as string; + const locks = app.get(SafetyLockService); + let announceLocked!: () => void; + let releaseBlock!: () => void; + const blockLocked = new Promise( + (resolve) => (announceLocked = resolve), + ); + const blockRelease = new Promise( + (resolve) => (releaseBlock = resolve), + ); + const block = prisma.$transaction(async (tx) => { + await locks.lockAccounts(tx, [picker.id, author.id]); + await tx.block.create({ + data: { blockerId: author.id, blockedId: picker.id }, + }); + announceLocked(); + await blockRelease; + }); + await blockLocked; + let sendSettled = false; + const send = app + .get(ConversationService) + .send(picker.id, conversationId, randomUUID(), "after block") + .finally(() => { + sendSettled = true; + }); + await new Promise((resolve) => setTimeout(resolve, 30)); + expect(sendSettled).toBe(false); + releaseBlock(); + await block; + await expect(send).rejects.toMatchObject({ code: "USER_BLOCKED" }); + }); + it("authorizes before idempotency lookup and never exposes account ids", async () => { const author = await actor("author"); const picker = await actor("picker"); diff --git a/apps/api/src/conversation/conversation.module.ts b/apps/api/src/conversation/conversation.module.ts index 6c8d70e..f945e15 100644 --- a/apps/api/src/conversation/conversation.module.ts +++ b/apps/api/src/conversation/conversation.module.ts @@ -2,6 +2,7 @@ import { Module } from "@nestjs/common"; import { AuthModule } from "../auth/auth.module.js"; import { DatabaseModule } from "../database/database.module.js"; import { RedisModule } from "../redis/redis.module.js"; +import { SafetyModule } from "../safety/safety.module.js"; import { ChatGateway } from "./chat.gateway.js"; import { ChatRateLimiter } from "./chat-rate-limiter.js"; import { ConversationController } from "./conversation.controller.js"; @@ -10,7 +11,7 @@ import { CHAT_PUBLISHER, OutboxMessageRelay } from "./outbox-message-relay.js"; import { StateChangingOriginGuard } from "./state-changing-origin.guard.js"; @Module({ - imports: [DatabaseModule, RedisModule, AuthModule], + imports: [DatabaseModule, RedisModule, AuthModule, SafetyModule], controllers: [ConversationController], providers: [ ConversationService, diff --git a/apps/api/src/conversation/conversation.service.ts b/apps/api/src/conversation/conversation.service.ts index 02d11e1..7eea09c 100644 --- a/apps/api/src/conversation/conversation.service.ts +++ b/apps/api/src/conversation/conversation.service.ts @@ -4,6 +4,7 @@ import { Prisma, type PrismaClient } from "@prisma/client"; import { leaseHmac } from "../auth/auth.config.js"; import { DomainException } from "../common/domain.exception.js"; import { PrismaService } from "../database/prisma.service.js"; +import { SafetyLockService } from "../safety/safety-lock.service.js"; type Db = Prisma.TransactionClient | PrismaClient; type MessageRow = { @@ -42,7 +43,10 @@ const firstMessageInclude = { @Injectable() export class ConversationService { - constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {} + constructor( + @Inject(PrismaService) private readonly prisma: PrismaService, + @Inject(SafetyLockService) private readonly safetyLocks: SafetyLockService, + ) {} async reply( accountId: string, @@ -59,8 +63,7 @@ export class ConversationService { include: firstMessageInclude, }); if (existing) { - await this.assertConversationMember(accountId, existing.id); - return this.replyResult(existing, clientMsgId, false); + return this.resolveExistingReply(accountId, existing, clientMsgId); } try { return await this.prisma.$transaction( @@ -71,8 +74,12 @@ export class ConversationService { include: firstMessageInclude, }); if (winner) { - await this.assertConversationMember(accountId, winner.id, tx); - return this.replyResult(winner, clientMsgId, false); + return this.resolveExistingReply( + accountId, + winner, + clientMsgId, + tx, + ); } const lease = await tx.bottlePickLease.findFirst({ where: { @@ -99,6 +106,7 @@ export class ConversationService { ); } const authorId = lease.bottle.authorId; + await this.safetyLocks.lockAccounts(tx, [accountId, authorId]); await this.assertAllowed(tx, accountId, authorId); const profiles = await tx.anonymousProfile.findMany({ where: { accountId: { in: [accountId, authorId] } }, @@ -177,8 +185,7 @@ export class ConversationService { include: firstMessageInclude, }); if (raced) { - await this.assertConversationMember(accountId, raced.id); - return this.replyResult(raced, clientMsgId, false); + return this.resolveExistingReply(accountId, raced, clientMsgId); } } } @@ -189,6 +196,26 @@ export class ConversationService { ); } + private async resolveExistingReply( + accountId: string, + conversation: ConversationWithFirst, + clientMsgId: string, + existingTx?: Prisma.TransactionClient, + ) { + const resolve = async (tx: Prisma.TransactionClient) => { + const authorized = await this.authorizeConversationMembership( + tx, + accountId, + conversation.id, + ); + const peer = this.activePeer(authorized.members, accountId); + await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]); + await this.assertAllowed(tx, accountId, peer.accountId); + return this.replyResult(conversation, clientMsgId, false); + }; + return existingTx ? resolve(existingTx) : this.prisma.$transaction(resolve); + } + private replyResult( conversation: ConversationWithFirst, clientMsgId: string, @@ -214,11 +241,14 @@ export class ConversationService { this.moderate(text); try { return await this.prisma.$transaction(async (tx) => { - const conversation = await this.authorizeConversation( + const conversation = await this.authorizeConversationMembership( tx, accountId, conversationId, ); + const peer = this.activePeer(conversation.members, accountId); + await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]); + await this.assertAllowed(tx, accountId, peer.accountId); const previous = await tx.message.findUnique({ where: { conversationId_clientMsgId: { conversationId, clientMsgId }, @@ -267,11 +297,14 @@ export class ConversationService { } catch (error) { if (this.isMessageIdempotencyRace(error)) { return this.prisma.$transaction(async (tx) => { - const conversation = await this.authorizeConversation( + const conversation = await this.authorizeConversationMembership( tx, accountId, conversationId, ); + const peer = this.activePeer(conversation.members, accountId); + await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]); + await this.assertAllowed(tx, accountId, peer.accountId); const message = await tx.message.findUniqueOrThrow({ where: { conversationId_clientMsgId: { conversationId, clientMsgId }, @@ -396,7 +429,7 @@ export class ConversationService { return { conversationId: id, lastReadSeq: member.lastReadSeq.toString() }; } - private async authorizeConversation( + private async authorizeConversationMembership( tx: Db, accountId: string, conversationId: string, @@ -414,12 +447,18 @@ export class ConversationService { ) { throw this.forbidden(); } - const peer = conversation.members.find( + return conversation; + } + + private activePeer( + members: T[], + accountId: string, + ): T { + const peer = members.find( (member) => member.accountId !== accountId && !member.leftAt, ); if (!peer) throw this.forbidden(); - await this.assertAllowed(tx, accountId, peer.accountId); - return conversation; + return peer; } private async createMessageOutbox( @@ -524,17 +563,6 @@ export class ConversationService { return value; } - private async assertConversationMember( - accountId: string, - id: string, - tx: Db = this.prisma, - ) { - const found = await tx.conversationMember.findUnique({ - where: { conversationId_accountId: { conversationId: id, accountId } }, - }); - if (!found || found.leftAt) throw this.hidden(); - } - async assertMember(accountId: string, id: string) { const found = await this.prisma.conversationMember.findUnique({ where: { conversationId_accountId: { conversationId: id, accountId } }, diff --git a/apps/api/src/conversation/outbox-message-relay.ts b/apps/api/src/conversation/outbox-message-relay.ts index ba93cc5..3df8826 100644 --- a/apps/api/src/conversation/outbox-message-relay.ts +++ b/apps/api/src/conversation/outbox-message-relay.ts @@ -1,6 +1,7 @@ import { Inject, Injectable, + Logger, type OnModuleDestroy, type OnModuleInit, } from "@nestjs/common"; @@ -26,13 +27,22 @@ type ClaimedEvent = { const NEVER = new Date("9999-12-31T23:59:59.999Z"); +function positiveInt(value: string | undefined, fallback: number): number { + const parsed = Number(value); + return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback; +} + @Injectable() export class OutboxMessageRelay implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(OutboxMessageRelay.name); private stopped = false; private loopPromise: Promise | undefined; private wakeLoop: (() => void) | undefined; - private readonly leaseMs = Number(process.env.OUTBOX_LEASE_MS ?? 30_000); - private readonly maxAttempts = Number(process.env.OUTBOX_MAX_ATTEMPTS ?? 5); + private readonly leaseMs = positiveInt(process.env.OUTBOX_LEASE_MS, 30_000); + private readonly maxAttempts = positiveInt( + process.env.OUTBOX_MAX_ATTEMPTS, + 5, + ); constructor( @Inject(PrismaService) private readonly prisma: PrismaService, @@ -112,21 +122,25 @@ export class OutboxMessageRelay implements OnModuleInit, OnModuleDestroy { nextRetryAt: exhausted ? NEVER : new Date(Date.now() + delayMs), }, }); - console.error( - JSON.stringify({ - eventId: event.id, - eventType: "MESSAGE_CREATED", - attempt: event.attempts, - errorClass: error instanceof Error ? error.name : "UnknownError", - }), - ); + this.logger.error({ + eventId: event.id, + eventType: "MESSAGE_CREATED", + attempt: event.attempts, + errorClass: error instanceof Error ? error.name : "UnknownError", + }); } return true; } private async loop(): Promise { while (!this.stopped) { - const handled = await this.runOnce().catch(() => false); + const handled = await this.runOnce().catch((error: unknown) => { + this.logger.error({ + event: "Message outbox relay iteration failed", + errorClass: error instanceof Error ? error.name : "UnknownError", + }); + return false; + }); if (!handled && !this.stopped) await this.waitForWork(); } } diff --git a/apps/api/src/redis/redis.service.ts b/apps/api/src/redis/redis.service.ts index c582e3a..6f44287 100644 --- a/apps/api/src/redis/redis.service.ts +++ b/apps/api/src/redis/redis.service.ts @@ -74,6 +74,27 @@ export class RedisService implements OnModuleInit, OnModuleDestroy { } } + /** + * Runs an already-connected Redis command under the same hard deadline as + * connect/quit. On timeout the exact client generation is destroyed, which + * rejects queued commands and lets the next call create a fresh client. + * Callers must create the command from the provided client: retaining + * `redis.client` across calls would bypass generation-safe replacement. + */ + async executeWithDeadline( + operation: (client: RedisClient) => Promise, + ): Promise { + await this.ensureConnected(); + const client = this.client; + const generation = this.generation; + try { + return await this.withDeadline(client, generation, operation(client)); + } catch (error) { + this.destroyClient(client, generation); + throw error; + } + } + async onModuleInit(): Promise { /* Auth Redis connects lazily so unrelated modules remain isolated. */ } diff --git a/apps/api/src/safety/safety-lock.service.ts b/apps/api/src/safety/safety-lock.service.ts new file mode 100644 index 0000000..575b0f6 --- /dev/null +++ b/apps/api/src/safety/safety-lock.service.ts @@ -0,0 +1,28 @@ +import { Injectable } from "@nestjs/common"; +import { type Prisma } from "@prisma/client"; + +/** + * Transaction-scoped serialization protocol for policy subjects. + * + * Every transaction that reads or writes block, sanction, or account-status + * policy MUST call this method first for every affected account. IDs are + * de-duplicated and sorted so overlapping multi-account operations cannot + * deadlock. The lock is released by PostgreSQL only when the transaction ends; + * callers must therefore pass the transaction client, never PrismaService. + */ +@Injectable() +export class SafetyLockService { + async lockAccounts( + tx: Prisma.TransactionClient, + accountIds: readonly string[], + ): Promise { + const canonicalIds = Array.from(new Set(accountIds)).sort(); + for (const accountId of canonicalIds) { + await tx.$executeRaw` + SELECT pg_advisory_xact_lock( + hashtextextended(${`safety:${accountId}`}, 0) + ) + `; + } + } +} diff --git a/apps/api/src/safety/safety.module.ts b/apps/api/src/safety/safety.module.ts new file mode 100644 index 0000000..7c2a0fe --- /dev/null +++ b/apps/api/src/safety/safety.module.ts @@ -0,0 +1,8 @@ +import { Module } from "@nestjs/common"; +import { SafetyLockService } from "./safety-lock.service.js"; + +@Module({ + providers: [SafetyLockService], + exports: [SafetyLockService], +}) +export class SafetyModule {}