From a16b5d2e17d076c3508113fcc7d5067f3512d9fb Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 05:09:47 -0700 Subject: [PATCH 1/6] feat(messages): reconcile Mongo fallback messages into Postgres (TASK-313 part 2) While Postgres is down, non-reply messages land in Mongo. Once it is healthy, a bounded pass copies each unmoved row into messages and stamps the Mongo row with the PG id it became. The moved row keeps its Mongo id as the client id (messages.legacy_mongo_id), so clients holding it keep working. - schema.sql: additive nullable legacy_mongo_id. The UNIQUE index is built by the reconciler with CREATE UNIQUE INDEX CONCURRENTLY, outside schema.sql, because the schema script runs as one transaction and CONCURRENTLY cannot. - Idempotent: INSERT ... ON CONFLICT (legacy_mongo_id) DO NOTHING; the Mongo stamp is set-once (filtered on reconciledAt absent). Two replicas can run the pass at once. - Bounded per pass (default 200, oldest first). A row the database refuses permanently (FK 23503, pod not in PG) is stamped reconcileFailedAt and left out of the scan so it cannot block the rows behind it. - Id resolution: PGMessage.resolveId maps a 24-hex Mongo id to its PG row. Used by findById, update, updatePayload, delete, reaction add/remove, thread follow/unfollow/collapsed, the created-at backdate route, and the message delete route. A delete of a reconciled message during a PG outage returns 503 rather than deleting only the Mongo copy. - Reaction enrichment keys by the PG id (pgId), not the client id. - Removes the "NEVER RECONCILED" comment. Replies still 503 during an outage, because Mongo rows have no reply edge to carry. Scheduled once a minute per process from server.ts (in-process overlap guard). Tests: __tests__/service/pgMessageReconcile.test.js, real PostgreSQL + MongoDB. Co-authored-by: Swift Co-Authored-By: Claude Haiku 5.5 Co-authored-by: swift Agent-Model: claude/claude-haiku-5-5/default --- .../service/pgMessageReconcile.test.js | 175 ++++++++++++++++ .../controllers/messageController.test.js | 3 +- .../controllers/reactionController.test.js | 4 +- backend/config/schema.sql | 8 + backend/controllers/messageController.ts | 64 +++--- backend/controllers/reactionController.ts | 21 +- backend/controllers/threadStateController.ts | 7 +- backend/models/Message.ts | 13 ++ backend/models/pg/Message.ts | 51 ++++- backend/routes/messages.ts | 2 +- backend/server.ts | 2 + backend/services/pgMessageReconcileService.ts | 189 ++++++++++++++++++ 12 files changed, 491 insertions(+), 48 deletions(-) create mode 100644 backend/__tests__/service/pgMessageReconcile.test.js create mode 100644 backend/services/pgMessageReconcileService.ts diff --git a/backend/__tests__/service/pgMessageReconcile.test.js b/backend/__tests__/service/pgMessageReconcile.test.js new file mode 100644 index 0000000000..5b735a1841 --- /dev/null +++ b/backend/__tests__/service/pgMessageReconcile.test.js @@ -0,0 +1,175 @@ +/** + * TASK-313 part 2 — Mongo fallback messages reconciled into PostgreSQL. + * + * Runs against real PostgreSQL and real MongoDB (Tier 1). The claims need a + * database that enforces them: ON CONFLICT on a unique index, a foreign key + * from messages.pod_id to pods(id), and two concurrent passes racing on the + * same rows. pg-mem can do none of that faithfully. + */ +const fs = require('fs'); +const path = require('path'); +const mongoose = require('mongoose'); +const { Pool } = require('pg'); +const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../utils/testUtils'); + +const RUN = process.env.INTEGRATION_TEST === 'true'; +const d = RUN ? describe : describe.skip; +const MARKER = 'tier1-reconcile-313'; + +let pool; +let MongoMessage; +let reconcile; +let PGMessage; +let MessageReaction; + +const POD = new mongoose.Types.ObjectId().toHexString(); +const USER = new mongoose.Types.ObjectId().toHexString(); + +const connect = () => new Pool({ + host: process.env.PG_HOST, + port: Number(process.env.PG_PORT || 5432), + database: process.env.PG_DATABASE, + user: process.env.PG_USER, + password: process.env.PG_PASSWORD, + ssl: false, +}); + +const insertFallback = (content, extra = {}) => MongoMessage.create({ + podId: POD, + userId: USER, + content: `${MARKER} ${content}`, + messageType: 'text', + createdAt: new Date(), + ...extra, +}); + +const pgRowsFor = async (mongoId) => (await pool.query( + 'SELECT id, content, created_at FROM messages WHERE legacy_mongo_id = $1', + [String(mongoId)], +)).rows; + +d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { + beforeAll(async () => { + pool = connect(); + await pool.query(fs.readFileSync(path.join(__dirname, '../../config/schema.sql'), 'utf8')); + await pool.query( + `INSERT INTO pods (id, name, type, created_by) VALUES ($1, $2, 'chat', $3) + ON CONFLICT (id) DO NOTHING`, + [POD, 'reconcile-313', USER], + ); + await setupMongoDb(); + // The app's own pool and models: this is the code under test. + // eslint-disable-next-line global-require + MongoMessage = require('../../models/Message'); + // eslint-disable-next-line global-require + ({ reconcileMongoFallbackMessages: reconcile } = require('../../services/pgMessageReconcileService')); + // eslint-disable-next-line global-require + PGMessage = require('../../models/pg/Message'); + // eslint-disable-next-line global-require + MessageReaction = require('../../models/pg/MessageReaction').default; + }); + + beforeEach(async () => { + await pool.query('DELETE FROM messages WHERE content LIKE $1', [`${MARKER}%`]); + await clearMongoDb(); + }); + + afterAll(async () => { + await pool.query('DELETE FROM messages WHERE content LIKE $1', [`${MARKER}%`]); + await pool.query('DELETE FROM pods WHERE id = $1', [POD]); + await pool.end(); + await closeMongoDb(); + // The app's pool is module-scoped; close it so Jest does not hold the process. + // eslint-disable-next-line global-require + const { pool: appPool } = require('../../config/db-pg'); + if (appPool?.end) await appPool.end(); + }); + + test('moves a fallback message once and keeps its Mongo id as the legacy id', async () => { + const doc = await insertFallback('a'); + + const pass = await reconcile(); + + expect(pass.moved).toBe(1); + const rows = await pgRowsFor(doc._id); + expect(rows).toHaveLength(1); + expect(rows[0].content).toBe(`${MARKER} a`); + expect(new Date(rows[0].created_at).getTime()).toBe(doc.createdAt.getTime()); + const after = await MongoMessage.findById(doc._id).lean(); + expect(after.reconciledAt).toBeInstanceOf(Date); + expect(after.reconciledPgId).toBe(Number(rows[0].id)); + }); + + test('each message appears once in the Postgres read after a reconcile, and a second pass moves nothing', async () => { + await insertFallback('b1'); + await insertFallback('b2'); + + await reconcile(); + const second = await reconcile(); + + expect(second.moved).toBe(0); + const listed = await PGMessage.findByPodId(POD, 50); + const ours = listed.filter((m) => String(m.content).startsWith(`${MARKER} b`)); + expect(ours).toHaveLength(2); + expect(new Set(ours.map((m) => m._id)).size).toBe(2); + }); + + test('two concurrent passes leave exactly one Postgres row per message', async () => { + const doc = await insertFallback('c'); + + const [x, y] = await Promise.all([reconcile(), reconcile()]); + + expect(x.moved + y.moved).toBe(1); + expect(await pgRowsFor(doc._id)).toHaveLength(1); + const stamped = await MongoMessage.findById(doc._id).lean(); + expect(stamped.reconciledAt).toBeInstanceOf(Date); + }); + + test('a pass is bounded by maxPerPass and the remainder moves on the next pass', async () => { + await insertFallback('d1'); + await insertFallback('d2'); + await insertFallback('d3'); + + const first = await reconcile({ maxPerPass: 2 }); + const second = await reconcile({ maxPerPass: 2 }); + + expect(first.moved).toBe(2); + expect(second.moved).toBe(1); + }); + + test('a row whose pod is not in Postgres is refused once and does not block the pass behind it', async () => { + const orphan = await insertFallback('e-orphan', { podId: new mongoose.Types.ObjectId().toHexString() }); + const doc = await insertFallback('e-ok'); + + const pass = await reconcile({ maxPerPass: 1 }); + const next = await reconcile({ maxPerPass: 1 }); + + expect(pass.skipped).toBe(1); + const refused = await MongoMessage.findById(orphan._id).lean(); + expect(refused.reconcileFailedAt).toBeInstanceOf(Date); + expect(await pgRowsFor(doc._id)).toHaveLength(1); + expect(next.moved).toBe(1); + }); + + test('a moved message resolves its Mongo id for delete', async () => { + const doc = await insertFallback('f'); + await reconcile(); + const pgId = await PGMessage.resolveId(String(doc._id)); + expect(pgId).toBe(String((await pgRowsFor(doc._id))[0].id)); + + await PGMessage.delete(String(doc._id)); + + expect(await pgRowsFor(doc._id)).toHaveLength(0); + }); + + test('a moved message resolves its Mongo id for reactions', async () => { + const doc = await insertFallback('g'); + await reconcile(); + const pgId = await PGMessage.resolveId(String(doc._id)); + + await MessageReaction.add(pgId, USER, '👍'); + + const summaries = await MessageReaction.listForMessage(pgId, USER); + expect(summaries.map((s) => s.emoji)).toEqual(['👍']); + }); +}); diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 86ae61ab9a..d6556b8ae9 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -608,9 +608,10 @@ describe('messageController', () => { it('deletes message successfully for message owner', async () => { const mockMessage = { user_id: 'u1', pod_id: 'p1' }; + PGMessage.resolveId.mockResolvedValue('1'); PGMessage.findById.mockResolvedValue(mockMessage); PGMessage.delete.mockResolvedValue(true); - const req = { params: { id: 'm1' }, userId: 'u1', user: { id: 'u1' } }; + const req = { params: { id: '1' }, userId: 'u1', user: { id: 'u1' } }; const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; await messageController.deleteMessage(req, res); expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' }); diff --git a/backend/__tests__/unit/controllers/reactionController.test.js b/backend/__tests__/unit/controllers/reactionController.test.js index 99d8a6466f..2f1bc83e32 100644 --- a/backend/__tests__/unit/controllers/reactionController.test.js +++ b/backend/__tests__/unit/controllers/reactionController.test.js @@ -354,7 +354,7 @@ describe('reactionController.addReaction — agent runtime path', () => { }); await reactionController.addReaction({ - params: { messageId: 'suffix-44' }, + params: { messageId: '144' }, body: { emoji: '👍' }, user: { _id: 'human-reactor' }, }, buildRes()); @@ -383,7 +383,7 @@ describe('reactionController.addReaction — agent runtime path', () => { const warning = jest.spyOn(console, 'warn').mockImplementation(); await reactionController.addReaction({ - params: { messageId: 'legacy-44' }, + params: { messageId: '244' }, body: { emoji: '👍' }, user: { _id: 'human-reactor' }, }, buildRes()); diff --git a/backend/config/schema.sql b/backend/config/schema.sql index 93d142b6c5..c5a3a8df4c 100644 --- a/backend/config/schema.sql +++ b/backend/config/schema.sql @@ -109,6 +109,14 @@ ALTER TABLE messages ALTER TABLE messages ADD COLUMN IF NOT EXISTS freshness_revision BIGINT NOT NULL DEFAULT 0; +-- TASK-313 part 2. The Mongo id of a fallback message this row was reconciled +-- from. Nullable: ordinary rows carry NULL, and several NULLs never collide. +-- The UNIQUE index is NOT declared here: CONCURRENTLY cannot run inside this +-- multi-statement script, which executes as one transaction. The reconciler +-- builds it (services/pgMessageReconcileService.ensureLegacyIdIndex) before +-- its first insert, so it never blocks writes on messages. +ALTER TABLE messages ADD COLUMN IF NOT EXISTS legacy_mongo_id TEXT; + CREATE TABLE IF NOT EXISTS pod_message_revisions ( pod_id VARCHAR(24) PRIMARY KEY REFERENCES pods(id) ON DELETE CASCADE, revision BIGINT NOT NULL DEFAULT 0 diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index 10f955d977..25af4bda3d 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -129,16 +129,18 @@ exports.getMessages = async (req: AuthRequest, res: Response): Promise => try { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const MessageReaction = require('../models/pg/MessageReaction').default; - const ids = (messages as Array<{ id?: string; _id?: string }>) - .map((m) => m.id || m._id) + // Keyed by the PG row id: a reconciled message's client `id` is its + // Mongo id, which reaction rows do not know. + const ids = (messages as Array<{ pgId?: string }>) + .map((m) => m.pgId) .filter(Boolean); if (ids.length > 0) { const reactionsMap = await MessageReaction.listForMessages(ids, String(userId)); // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { decorateReactionMap } = require('../services/reactionAttributionService'); const decorated = await decorateReactionMap(reactionsMap); - for (const m of messages as Array<{ id?: string; _id?: string; reactions?: unknown }>) { - const key = String(m.id || m._id || ''); + for (const m of messages as Array<{ pgId?: string; reactions?: unknown }>) { + const key = String(m.pgId || ''); m.reactions = decorated.get(key) || []; } } @@ -315,24 +317,11 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = } catch (pgErr) { const e = pgErr as { message?: string }; console.warn('PG unavailable for createMessage, falling back to MongoDB:', e.message); - // Mongo messages have no reply_to_message_id column and are never - // reconciled into PG. A reply written here would look successful while - // permanently losing its parent edge, so only non-replies may use this - // availability fallback. - // - // "NEVER RECONCILED" IS A CURRENT FACT, NOT AN INVARIANT — and this - // guard is correct only while it holds (@sprint-review, pod 56881). - // Reconciling Mongo into PG was the other option considered on TASK-040; - // #1116 chose this one instead, so nothing reconciles today and the 503 - // is the resolution rather than a stopgap. - // - // If reconciliation is ever built, this stops being a data-loss guard - // and becomes a rejection with no remaining reason: a user told - // "Replies are temporarily unavailable" about a path that would now - // preserve their reply. Whoever builds it must remove or relax this in - // the same change. The dependency runs both ways and appears in neither - // diff, which is why it is written at both ends — TASK-040 carries the - // pointer back here. + // Mongo rows have no reply_to_message_id column, and the reconciler + // (services/pgMessageReconcileService, TASK-313 part 2) copies them into + // PG without a reply edge. A reply written here would look successful + // while losing its parent edge, so replies still wait for Postgres. + // Non-replies take this fallback and are moved once PG is back. if (replyToMessageId) { res.status(503).json({ error: 'Replies are temporarily unavailable. Please try again shortly.', @@ -455,11 +444,30 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = exports.deleteMessage = async (req: AuthRequest, res: Response): Promise => { try { + // A message the reconciler moved answers to its Mongo id, and the PG row + // is the one to delete. Deleting only the Mongo copy would leave the + // message visible in PG. + const messageId = req.params.id; let message: NormalizedMessage | null = null; + let pgId: string | null = null; + let pgUnavailable = false; try { - message = await PGMessage.findById(req.params.id) as NormalizedMessage | null; + pgId = await PGMessage.resolveId(messageId); + if (pgId) message = await PGMessage.findById(pgId) as NormalizedMessage | null; } catch { - const mongoMsg = await MongoMessage.findById(req.params.id) as Record | null; + pgUnavailable = true; + } + if (!pgId) { + const mongoMsg = await MongoMessage.findById(messageId) as Record | null; + if (mongoMsg?.reconciledAt && pgUnavailable) { + // The PG row exists and cannot be reached. Deleting the Mongo copy + // now would leave it visible once PG returns. + res.status(503).json({ + error: 'Message is temporarily unavailable. Please try again shortly.', + code: 'MESSAGE_REQUIRES_POSTGRES', + }); + return; + } if (mongoMsg) message = normalizeMongo(mongoMsg); } @@ -482,10 +490,10 @@ exports.deleteMessage = async (req: AuthRequest, res: Response): Promise = } } - try { - await PGMessage.delete(req.params.id); - } catch { - await MongoMessage.findByIdAndDelete(req.params.id); + if (pgId) { + await PGMessage.delete(pgId); + } else { + await MongoMessage.findByIdAndDelete(messageId); } res.json({ msg: 'Message deleted' }); diff --git a/backend/controllers/reactionController.ts b/backend/controllers/reactionController.ts index dc5f83f105..3ff202642e 100644 --- a/backend/controllers/reactionController.ts +++ b/backend/controllers/reactionController.ts @@ -79,18 +79,27 @@ async function emitReactionChange(messageId: string | number, podId: string, rea interface MessageContext { podId: string; authorUserId: string; + // The PG row id. The client's messageId may be a Mongo id for a message the + // reconciler moved (TASK-313 part 2); reaction rows are keyed by this one. + pgId: string; } async function loadMessageContext(messageId: string | number): Promise { + // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require + const PGMessage = require('../models/pg/Message'); + const pgId = await PGMessage.resolveId(String(messageId)); + if (!pgId) return null; // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { pool } = require('../config/db-pg'); const result = await pool.query( 'SELECT pod_id, user_id FROM messages WHERE id = $1 LIMIT 1', - [Number(messageId)], + [Number(pgId)], ); const row = result.rows[0]; if (!row?.pod_id || !row?.user_id) return null; - return { podId: String(row.pod_id), authorUserId: String(row.user_id) }; + return { + podId: String(row.pod_id), authorUserId: String(row.user_id), pgId, + }; } // A reaction is an acknowledgement of an already-posted agent message, not a @@ -253,8 +262,8 @@ export async function addReaction(req: AuthedReq, res: AuthedRes): Promise res.status(403).json({ msg: 'Not a member of this pod' }); return; } - const added = await MessageReaction.add(messageId, userId, emoji); - const rawSummaries = await MessageReaction.listForMessage(messageId, userId); + const added = await MessageReaction.add(message.pgId, userId, emoji); + const rawSummaries = await MessageReaction.listForMessage(message.pgId, userId); const reactions = await decorateReactionSummaries(rawSummaries); void emitReactionChange(messageId, podId, reactions); if (added) { @@ -300,8 +309,8 @@ export async function removeReaction(req: AuthedReq, res: AuthedRes): Promise({ @@ -48,6 +57,10 @@ const MessageSchema = new Schema({ // leaves pending=true, which makes the send-time hold fail closed. freshnessPending: { type: Boolean }, freshnessCommittedAt: { type: Date }, + reconciledAt: { type: Date }, + reconciledPgId: { type: Number }, + reconcileFailedAt: { type: Date }, + reconcileError: { type: String }, }); MessageSchema.pre('save', function rememberWhetherThisIsANewPost(next) { diff --git a/backend/models/pg/Message.ts b/backend/models/pg/Message.ts index 2c5210250d..1537325fde 100644 --- a/backend/models/pg/Message.ts +++ b/backend/models/pg/Message.ts @@ -20,6 +20,9 @@ interface MessageRow { payload?: unknown; reply_to_message_id?: string; thread_root_id?: string | number | null; + // TASK-313 part 2: the Mongo id this row was reconciled from. Clients keep + // holding that id, so it is what they see as `id`. + legacy_mongo_id?: string | null; created_at: unknown; updated_at?: unknown; username?: string; @@ -33,6 +36,9 @@ interface MessageRow { interface FormattedMessage extends MessageRow { _id: string; + // The PG row id. Reaction rows are keyed by it, so enrichment reads this + // field, never `id`, which is the client-facing id. + pgId: string; text: string; messageType: string; createdAt: unknown; @@ -58,12 +64,15 @@ interface PodActivityEntry { } function formatMessage(msg: MessageRow): FormattedMessage { - const messageId = msg.id ? msg.id.toString() : ''; + const pgId = msg.id ? msg.id.toString() : ''; + // A reconciled row answers to the Mongo id its clients already hold. + const messageId = msg.legacy_mongo_id || pgId; const userId = msg.user_id || ''; return { ...msg, _id: messageId, id: messageId, + pgId, content: msg.content || '', text: msg.content || '', messageType: msg.message_type || 'text', @@ -235,7 +244,7 @@ class Message { try { let query = ` SELECT - m.id, m.pod_id, m.user_id, m.content, m.message_type, m.author_kind, m.payload, + m.id, m.legacy_mongo_id, m.pod_id, m.user_id, m.content, m.message_type, m.author_kind, m.payload, m.reply_to_message_id, -- MUST be selected. Late columns are easy to miss in an explicit -- projection, and this one is read by the wake path. @@ -273,11 +282,31 @@ class Message { } } + /** + * Maps a message id a client sent to the PG row id. A numeric id is already + * one. A 24-hex id is a Mongo id: it resolves only if the reconciler moved + * that message. Returns null for anything else, including a Mongo id that + * was never moved, which is the caller's to look up in Mongo. + */ + static async resolveId(id: string | number): Promise { + const raw = String(id); + if (/^\d+$/.test(raw)) return raw; + if (!/^[0-9a-f]{24}$/i.test(raw)) return null; + const result = await (pool as PgPool).query( + 'SELECT id FROM messages WHERE legacy_mongo_id = $1 LIMIT 1', + [raw], + ); + const row = result.rows[0] as { id?: string | number } | undefined; + return row?.id != null ? String(row.id) : null; + } + static async findById(id: string): Promise { + const pgId = await Message.resolveId(id); + if (!pgId) return null; try { const query = ` SELECT - m.id, m.pod_id, m.user_id, m.content, m.message_type, m.author_kind, m.payload, + m.id, m.legacy_mongo_id, m.pod_id, m.user_id, m.content, m.message_type, m.author_kind, m.payload, m.reply_to_message_id, -- MUST be selected. The controller prefers THIS row over create()'s -- RETURNING * (message = populated || created) and hands it to @@ -297,7 +326,7 @@ class Message { LEFT JOIN users ru ON rm.user_id = ru._id WHERE m.id = $1 `; - const result = await (pool as PgPool).query(query, [id]); + const result = await (pool as PgPool).query(query, [pgId]); if (result.rows.length === 0) return null; return formatMessage(result.rows[0] as unknown as MessageRow); } catch (error) { @@ -395,19 +424,23 @@ class Message { } static async update(id: string, content: string): Promise { + const pgId = await Message.resolveId(id); + if (!pgId) return undefined; const query = ` UPDATE messages SET content = $1, updated_at = CURRENT_TIMESTAMP WHERE id = $2 RETURNING * `; - const result = await (pool as PgPool).query(query, [content, id]); + const result = await (pool as PgPool).query(query, [content, pgId]); return result.rows[0] as unknown as MessageRow | undefined; } // ADR-020 D3: card status transitions rewrite the message's payload so // every client (and late loaders) see the authoritative card face. static async updatePayload(id: string, payload: unknown): Promise { + const pgId = await Message.resolveId(id); + if (!pgId) return undefined; const query = ` UPDATE messages SET payload = $1, updated_at = CURRENT_TIMESTAMP @@ -415,18 +448,20 @@ class Message { RETURNING * `; const result = await (pool as PgPool).query(query, [ - payload == null ? null : JSON.stringify(payload), id, + payload == null ? null : JSON.stringify(payload), pgId, ]); return result.rows[0] as unknown as MessageRow | undefined; } static async delete(id: string): Promise { + const pgId = await Message.resolveId(id); + if (!pgId) return undefined; const query = `DELETE FROM messages WHERE id = $1 RETURNING *`; - const result = await (pool as PgPool).query(query, [id]); + const result = await (pool as PgPool).query(query, [pgId]); const deleted = result.rows[0] as unknown as MessageRow | undefined; // eslint-disable-next-line global-require const { resolve } = require('../../services/attentionItemService'); - await resolve('message', id); + await resolve('message', pgId); return deleted; } diff --git a/backend/routes/messages.ts b/backend/routes/messages.ts index fb766f9a44..688d4114c1 100644 --- a/backend/routes/messages.ts +++ b/backend/routes/messages.ts @@ -206,7 +206,7 @@ router.patch('/:id/created-at', auth, async (req: any, res: any) => { const { pool } = require('../config/db-pg'); const r = await pool.query( 'UPDATE messages SET created_at = $1, updated_at = $1 WHERE id = $2 RETURNING id, created_at', - [ts, id], + [ts, message.pgId], ); if (r.rowCount === 0) return res.status(404).json({ msg: 'message not found in PG' }); return res.json({ ok: true, id: r.rows[0].id, created_at: r.rows[0].created_at }); diff --git a/backend/server.ts b/backend/server.ts index 34ac77daa9..13a0e2de4b 100644 --- a/backend/server.ts +++ b/backend/server.ts @@ -451,6 +451,8 @@ if (process.env.PG_HOST) { retentionErr?.message || retentionErr, ); } + // TASK-313 part 2: moves Mongo fallback messages into PG once it is back. + require('./services/pgMessageReconcileService').initPgMessageReconcile(); try { require('./services/agentInstallationCleanupService').initInstallationCleanup(); } catch (cleanupErr: any) { diff --git a/backend/services/pgMessageReconcileService.ts b/backend/services/pgMessageReconcileService.ts new file mode 100644 index 0000000000..7218608cb9 --- /dev/null +++ b/backend/services/pgMessageReconcileService.ts @@ -0,0 +1,189 @@ +/** + * TASK-313 part 2 — moves Mongo fallback messages into PostgreSQL. + * + * While Postgres is down, createMessage writes non-replies to Mongo. Once it + * is healthy this pass copies each unmoved row into `messages`, stamping the + * Mongo row with the PG id it became. Three properties carry the design: + * + * - Idempotent across replicas. The PG insert is ON CONFLICT on + * legacy_mongo_id, so a second writer inserts nothing. The Mongo stamp is + * guarded on `reconciledAt` being absent, so a second stamp matches nothing. + * - Bounded. One pass reads at most `maxPerPass` rows, oldest first. + * - Refusals are recorded, not retried. A row the database will never accept + * (its pod is not in Postgres) is stamped `reconcileFailedAt` and left out + * of the scan, so it cannot hold back the rows behind it. + * + * Reads never merge the two stores: chat reads Postgres when it is healthy and + * Mongo only when it is not. A reconciled Mongo row therefore cannot be + * returned alongside its PG copy in one response. + */ +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { pool } = require('../config/db-pg') as { + pool: { + query: (sql: string, params?: unknown[]) => Promise<{ rows: Record[] }>; + }; +}; +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const MongoMessage = require('../models/Message') as { + find: (q: Record) => { + sort: (s: Record) => { + limit: (n: number) => { lean: () => Promise }; + }; + }; + updateOne: ( + filter: Record, + update: Record, + ) => Promise<{ matchedCount?: number; modifiedCount?: number }>; +}; + +interface MongoFallbackDoc { + _id: { toString(): string }; + podId: { toString(): string }; + userId: { toString(): string }; + content: string; + messageType?: string; + createdAt: Date; +} + +export interface ReconcilePassResult { + /** False when Postgres did not answer; the pass did nothing. */ + healthy: boolean; + /** Rows this pass inserted into Postgres. */ + moved: number; + /** Rows Postgres already had (a concurrent pass or an earlier crash). */ + alreadyPresent: number; + /** Rows that failed this pass; permanent refusals are also stamped. */ + skipped: number; +} + +export const LEGACY_ID_INDEX = 'messages_legacy_mongo_id_key'; +export const DEFAULT_MAX_PER_PASS = 200; + +// Postgres SQLSTATE 23503: foreign_key_violation. The row can never be +// accepted as written, so retrying it is pointless. +const PERMANENT_PG_CODES = new Set(['23503']); + +/** + * Builds the unique index ON CONFLICT needs, without blocking writes. + * + * CONCURRENTLY cannot run inside a transaction, so this issues the statement + * on its own autocommit connection. A failed concurrent build leaves an + * INVALID index behind; IF NOT EXISTS would keep that forever, so an invalid + * one is dropped and rebuilt. + */ +export async function ensureLegacyIdIndex(): Promise { + const state = await pool.query( + `SELECT i.indisvalid FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE c.relname = $1`, + [LEGACY_ID_INDEX], + ); + if (state.rows[0] && state.rows[0].indisvalid === false) { + await pool.query(`DROP INDEX CONCURRENTLY IF EXISTS ${LEGACY_ID_INDEX}`); + } + await pool.query( + `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ${LEGACY_ID_INDEX} + ON messages (legacy_mongo_id)`, + ); +} + +async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { + const id = doc._id.toString(); + const inserted = await pool.query( + `INSERT INTO messages (pod_id, user_id, content, message_type, created_at, legacy_mongo_id) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (legacy_mongo_id) DO NOTHING + RETURNING id`, + [ + doc.podId.toString(), + doc.userId.toString(), + doc.content || '', + doc.messageType || 'text', + doc.createdAt, + id, + ], + ); + let pgId = inserted.rows[0]?.id; + if (pgId === undefined) { + const existing = await pool.query( + 'SELECT id FROM messages WHERE legacy_mongo_id = $1', + [id], + ); + pgId = existing.rows[0]?.id; + } + // Stamp only after Postgres has the row. A crash between the two leaves an + // unstamped Mongo row that the next pass turns into `present`, then stamps. + await MongoMessage.updateOne( + { _id: doc._id, reconciledAt: { $exists: false } }, + { $set: { reconciledAt: new Date(), reconciledPgId: Number(pgId) } }, + ); + return inserted.rows[0] ? 'moved' : 'present'; +} + +export async function reconcileMongoFallbackMessages( + { maxPerPass = DEFAULT_MAX_PER_PASS }: { maxPerPass?: number } = {}, +): Promise { + const result: ReconcilePassResult = { + healthy: false, moved: 0, alreadyPresent: 0, skipped: 0, + }; + try { + await pool.query('SELECT 1'); + } catch { + return result; + } + result.healthy = true; + + await ensureLegacyIdIndex(); + + const batch = await MongoMessage.find({ + reconciledAt: { $exists: false }, + reconcileFailedAt: { $exists: false }, + }) + .sort({ createdAt: 1, _id: 1 }) + .limit(Math.max(1, Math.trunc(maxPerPass))) + .lean(); + + for (const doc of batch) { + try { + const outcome = await copyOne(doc); + if (outcome === 'moved') result.moved += 1; + else result.alreadyPresent += 1; + } catch (err) { + const e = err as { code?: string; message?: string }; + result.skipped += 1; + if (e.code && PERMANENT_PG_CODES.has(e.code)) { + await MongoMessage.updateOne( + { _id: doc._id, reconcileFailedAt: { $exists: false } }, + { $set: { reconcileFailedAt: new Date(), reconcileError: e.message || e.code } }, + ); + } + console.warn('[pgMessageReconcile] row not moved:', doc._id.toString(), e.message); + } + } + return result; +} + +export default { reconcileMongoFallbackMessages, ensureLegacyIdIndex }; + +// One pass a minute per process. Replicas may each run one: the database +// (ON CONFLICT) and the Mongo stamp (set-once) make overlapping passes harmless. +// The in-process flag only keeps a slow pass from piling up behind itself. +let timer: ReturnType | null = null; +let passInFlight = false; + +export function initPgMessageReconcile(intervalMs = 60_000): void { + if (timer) return; + timer = setInterval(() => { + if (passInFlight) return; + passInFlight = true; + reconcileMongoFallbackMessages() + .then((r) => { + if (r.moved || r.skipped) { + console.log(`[pgMessageReconcile] moved=${r.moved} present=${r.alreadyPresent} skipped=${r.skipped}`); + } + }) + .catch((err: Error) => console.warn('[pgMessageReconcile] pass failed:', err.message)) + .finally(() => { passInFlight = false; }); + }, intervalMs); + timer.unref?.(); +} From 1606f5259ec4bc1bbf3aefd20e1eb326bcfc70b9 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 05:57:37 -0700 Subject: [PATCH 2/6] fix(messages): reconciler keeps payload and kernel marker; backfills missing pods; replies accept moved ids (TASK-313 part 2) Review findings on #2624: - copyOne drops payload and authorKind. Both now move with the row, so a card keeps its face and a kernel seat post keeps its marker. - A 23503 (pod absent from PG) was stamped as a permanent refusal. The Mongo-only pod path leaves pods absent from PG until the next send backfills them, so the reconciler now backfills with syncPodFromMongo and retries once. A refusal is stamped only when the pod is absent from Mongo too. - Replies and thread posts that name a moved message by its Mongo id are mapped to the PG row before ThreadRootError validation, so they stop 400ing. Tests: payload and kernel marker survive; a Mongo-only pod is backfilled and its message moves. Co-authored-by: Swift Co-Authored-By: Claude Haiku 5.5 Co-authored-by: swift Agent-Model: claude/claude-haiku-5-5/default --- .../service/pgMessageReconcile.test.js | 33 +++++++++++++ backend/controllers/messageController.ts | 22 +++++++-- backend/services/pgMessageReconcileService.ts | 48 ++++++++++++++++--- 3 files changed, 91 insertions(+), 12 deletions(-) diff --git a/backend/__tests__/service/pgMessageReconcile.test.js b/backend/__tests__/service/pgMessageReconcile.test.js index 5b735a1841..8dc07d5d40 100644 --- a/backend/__tests__/service/pgMessageReconcile.test.js +++ b/backend/__tests__/service/pgMessageReconcile.test.js @@ -151,6 +151,39 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { expect(next.moved).toBe(1); }); + test('a card keeps its payload and a kernel post keeps its author marker', async () => { + const payload = { kind: 'approval', status: 'open' }; + const doc = await insertFallback('h', { messageType: 'card', payload, authorKind: 'kernel' }); + + await reconcile(); + + const row = (await pool.query( + 'SELECT payload, author_kind FROM messages WHERE legacy_mongo_id = $1', + [String(doc._id)], + )).rows[0]; + expect(row.payload).toEqual(payload); + expect(row.author_kind).toBe('kernel'); + }); + + test('a pod that exists only in Mongo is backfilled and its message then moves', async () => { + const orphanPod = new mongoose.Types.ObjectId().toHexString(); + await mongoose.connection.db.collection('pods').insertOne({ + _id: new mongoose.Types.ObjectId(orphanPod), + name: 'reconcile-313-backfill', + type: 'chat', + createdBy: new mongoose.Types.ObjectId(USER), + members: [new mongoose.Types.ObjectId(USER)], + }); + const doc = await insertFallback('i', { podId: orphanPod }); + + const pass = await reconcile(); + + expect(pass.moved).toBe(1); + expect(await pgRowsFor(doc._id)).toHaveLength(1); + expect((await MongoMessage.findById(doc._id).lean()).reconcileFailedAt).toBeUndefined(); + await pool.query('DELETE FROM pods WHERE id = $1', [orphanPod]); + }); + test('a moved message resolves its Mongo id for delete', async () => { const doc = await insertFallback('f'); await reconcile(); diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index 25af4bda3d..dc858e8f9f 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -260,18 +260,30 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = // message and change nothing. There is nothing to reconcile until there // are two claims to reconcile. let resolvedThreadRootId: number | null = null; + // A message the reconciler moved answers to its Mongo id, so a reply or a + // thread post names it that way. Map it to the PG row before validation. + // An id that resolves to nothing is passed through as sent, and the + // resolver rejects it as it always has. + const pgReplyToMessageId = replyToMessageId + ? (await PGMessage.resolveId(String(replyToMessageId))) ?? replyToMessageId + : replyToMessageId; + const pgThreadRootId = threadRootId + ? (await PGMessage.resolveId(String(threadRootId))) ?? threadRootId + : threadRootId; try { - if (threadRootId != null && threadRootId !== '') { + if (pgThreadRootId != null && pgThreadRootId !== '') { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { resolveThreadRoot } = require('../services/threadRootResolver'); - resolvedThreadRootId = await resolveThreadRoot({ podId, replyToMessageId, threadRootId }); - } else if (replyToMessageId) { + resolvedThreadRootId = await resolveThreadRoot({ + podId, replyToMessageId: pgReplyToMessageId, threadRootId: pgThreadRootId, + }); + } else if (pgReplyToMessageId) { // A reply's target must be in this pod (TASK-278). A store failure here // is left to the write below, which answers it and re-checks in SQL. // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { assertReplyTargetInPod } = require('../services/threadRootResolver'); try { - await assertReplyTargetInPod({ podId, replyToMessageId }); + await assertReplyTargetInPod({ podId, replyToMessageId: pgReplyToMessageId }); } catch (checkErr) { if ((checkErr as { name?: string })?.name === 'ThreadRootError') throw checkErr; } @@ -296,7 +308,7 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = userId, messageContent || '', 'text', - replyToMessageId || null, + pgReplyToMessageId || null, null, resolvedThreadRootId, ); diff --git a/backend/services/pgMessageReconcileService.ts b/backend/services/pgMessageReconcileService.ts index 7218608cb9..137503e384 100644 --- a/backend/services/pgMessageReconcileService.ts +++ b/backend/services/pgMessageReconcileService.ts @@ -24,6 +24,14 @@ const { pool } = require('../config/db-pg') as { }; }; // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { syncPodFromMongo } = require('./pgPodSyncService') as { + syncPodFromMongo: (podId: string, requestingUserId: string) => Promise; +}; +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const MongoPod = require('../models/Pod') as { + exists: (q: Record) => Promise; +}; +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const MongoMessage = require('../models/Message') as { find: (q: Record) => { sort: (s: Record) => { @@ -42,6 +50,10 @@ interface MongoFallbackDoc { userId: { toString(): string }; content: string; messageType?: string; + // Both must survive the move: a card loses its face without payload, and a + // kernel seat post loses its marker without authorKind. + authorKind?: string | null; + payload?: unknown; createdAt: Date; } @@ -59,9 +71,8 @@ export interface ReconcilePassResult { export const LEGACY_ID_INDEX = 'messages_legacy_mongo_id_key'; export const DEFAULT_MAX_PER_PASS = 200; -// Postgres SQLSTATE 23503: foreign_key_violation. The row can never be -// accepted as written, so retrying it is pointless. -const PERMANENT_PG_CODES = new Set(['23503']); +// Postgres SQLSTATE 23503: foreign_key_violation (messages.pod_id has no pod row). +const FK_VIOLATION = '23503'; /** * Builds the unique index ON CONFLICT needs, without blocking writes. @@ -87,11 +98,17 @@ export async function ensureLegacyIdIndex(): Promise { ); } +async function podExistsInMongo(podId: string): Promise { + return Boolean(await MongoPod.exists({ _id: podId })); +} + async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { const id = doc._id.toString(); const inserted = await pool.query( - `INSERT INTO messages (pod_id, user_id, content, message_type, created_at, legacy_mongo_id) - VALUES ($1, $2, $3, $4, $5, $6) + `INSERT INTO messages ( + pod_id, user_id, content, message_type, author_kind, payload, created_at, legacy_mongo_id + ) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (legacy_mongo_id) DO NOTHING RETURNING id`, [ @@ -99,6 +116,8 @@ async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { doc.userId.toString(), doc.content || '', doc.messageType || 'text', + doc.authorKind === 'kernel' ? 'kernel' : null, + doc.payload == null ? null : JSON.stringify(doc.payload), doc.createdAt, id, ], @@ -145,13 +164,28 @@ export async function reconcileMongoFallbackMessages( for (const doc of batch) { try { - const outcome = await copyOne(doc); + const outcome = await copyOne(doc).catch(async (err: { code?: string }) => { + // The pod is missing from PG. The Mongo-only pod path leaves it that + // way until the next send backfills it (createMessage does the same + // before its insert), so backfill and retry once. + if (err.code !== FK_VIOLATION) throw err; + try { + await syncPodFromMongo(doc.podId.toString(), doc.userId.toString()); + } catch { + // No Mongo pod to sync from. The retry below fails the same way and + // the outer handler decides whether that is permanent. + } + return copyOne(doc); + }); if (outcome === 'moved') result.moved += 1; else result.alreadyPresent += 1; } catch (err) { const e = err as { code?: string; message?: string }; result.skipped += 1; - if (e.code && PERMANENT_PG_CODES.has(e.code)) { + // Stamp a refusal only when the pod is absent from Mongo too: then no + // sync can ever place the message. Anything else stays in the scan and + // is retried on the next pass. + if (e.code === FK_VIOLATION && !(await podExistsInMongo(doc.podId.toString()))) { await MongoMessage.updateOne( { _id: doc._id, reconcileFailedAt: { $exists: false } }, { $set: { reconcileFailedAt: new Date(), reconcileError: e.message || e.code } }, From 9ad31e71b889c8b77dec17eaa83c1197e578db59 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:00:13 -0700 Subject: [PATCH 3/6] fix(messages): quarantine malformed reconciliation rows and preserve retry progress Co-authored-by: kai Agent-Model: codex/gpt-6.1-sol/medium --- .../service/pgMessageReconcile.test.js | 116 +++++++++++++++--- .../controllers/messageController.test.js | 34 +++++ backend/controllers/messageController.ts | 29 ++--- backend/models/Message.ts | 4 +- backend/models/pg/Message.ts | 20 +-- backend/routes/messages.ts | 2 +- backend/services/pgMessageId.ts | 13 ++ backend/services/pgMessageReconcileService.ts | 112 +++++++++-------- backend/services/threadRootResolver.ts | 9 +- 9 files changed, 245 insertions(+), 94 deletions(-) create mode 100644 backend/services/pgMessageId.ts diff --git a/backend/__tests__/service/pgMessageReconcile.test.js b/backend/__tests__/service/pgMessageReconcile.test.js index 8dc07d5d40..345f13b353 100644 --- a/backend/__tests__/service/pgMessageReconcile.test.js +++ b/backend/__tests__/service/pgMessageReconcile.test.js @@ -19,6 +19,8 @@ const MARKER = 'tier1-reconcile-313'; let pool; let MongoMessage; let reconcile; +let copyOne; +let ensureLegacyIdIndex; let PGMessage; let MessageReaction; @@ -44,7 +46,7 @@ const insertFallback = (content, extra = {}) => MongoMessage.create({ }); const pgRowsFor = async (mongoId) => (await pool.query( - 'SELECT id, content, created_at FROM messages WHERE legacy_mongo_id = $1', + 'SELECT id, content, created_at, payload, author_kind FROM messages WHERE legacy_mongo_id = $1', [String(mongoId)], )).rows; @@ -62,7 +64,7 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { // eslint-disable-next-line global-require MongoMessage = require('../../models/Message'); // eslint-disable-next-line global-require - ({ reconcileMongoFallbackMessages: reconcile } = require('../../services/pgMessageReconcileService')); + ({ reconcileMongoFallbackMessages: reconcile, copyOne, ensureLegacyIdIndex } = require('../../services/pgMessageReconcileService')); // eslint-disable-next-line global-require PGMessage = require('../../models/pg/Message'); // eslint-disable-next-line global-require @@ -117,14 +119,29 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { test('two concurrent passes leave exactly one Postgres row per message', async () => { const doc = await insertFallback('c'); - const [x, y] = await Promise.all([reconcile(), reconcile()]); + await ensureLegacyIdIndex(); + const [x, y] = await Promise.all([copyOne(doc), copyOne(doc)]); - expect(x.moved + y.moved).toBe(1); + expect([x, y].filter((outcome) => outcome === 'moved')).toHaveLength(1); expect(await pgRowsFor(doc._id)).toHaveLength(1); const stamped = await MongoMessage.findById(doc._id).lean(); expect(stamped.reconciledAt).toBeInstanceOf(Date); }); + test('another replica holding the index maintenance lock defers the pass', async () => { + const doc = await insertFallback('index-lock'); + const client = await pool.connect(); + try { + await client.query('SELECT pg_advisory_lock(313, 2)'); + expect((await reconcile()).moved).toBe(0); + expect(await pgRowsFor(doc._id)).toHaveLength(0); + } finally { + await client.query('SELECT pg_advisory_unlock(313, 2)'); + client.release(); + } + expect((await reconcile()).moved).toBe(1); + }); + test('a pass is bounded by maxPerPass and the remainder moves on the next pass', async () => { await insertFallback('d1'); await insertFallback('d2'); @@ -137,7 +154,7 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { expect(second.moved).toBe(1); }); - test('a row whose pod is not in Postgres is refused once and does not block the pass behind it', async () => { + test('a missing PG pod is deferred without blocking later rows, then retries after sync', async () => { const orphan = await insertFallback('e-orphan', { podId: new mongoose.Types.ObjectId().toHexString() }); const doc = await insertFallback('e-ok'); @@ -146,23 +163,92 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { expect(pass.skipped).toBe(1); const refused = await MongoMessage.findById(orphan._id).lean(); - expect(refused.reconcileFailedAt).toBeInstanceOf(Date); + expect(refused.reconcileFailedAt).toBeUndefined(); + expect(refused.reconcileRetryAt).toBeInstanceOf(Date); + await pool.query(`INSERT INTO pods (id, name, type, created_by) VALUES ($1, 'late-pod', 'chat', $2)`, [String(orphan.podId), USER]); + await MongoMessage.updateOne({ _id: orphan._id }, { $set: { reconcileRetryAt: new Date(0) } }); + expect((await reconcile()).moved).toBe(1); + expect(await pgRowsFor(orphan._id)).toHaveLength(1); + await pool.query('DELETE FROM messages WHERE pod_id = $1', [String(orphan.podId)]); + await pool.query('DELETE FROM pods WHERE id = $1', [String(orphan.podId)]); expect(await pgRowsFor(doc._id)).toHaveLength(1); expect(next.moved).toBe(1); }); - test('a card keeps its payload and a kernel post keeps its author marker', async () => { - const payload = { kind: 'approval', status: 'open' }; - const doc = await insertFallback('h', { messageType: 'card', payload, authorKind: 'kernel' }); + test('more than a full batch of poison rows cannot starve a valid message', async () => { + const bad = Array.from({ length: 201 }, () => ({ + podId: new mongoose.Types.ObjectId(POD), userId: new mongoose.Types.ObjectId(USER), + content: `${MARKER} bad\0`, createdAt: new Date(0), + })); + await MongoMessage.collection.insertMany([...bad, + { content: `${MARKER} missing-pod`, userId: new mongoose.Types.ObjectId(USER), createdAt: new Date(0) }, + { content: `${MARKER} missing-user`, podId: new mongoose.Types.ObjectId(POD), createdAt: new Date(0) }, + ]); + const good = await insertFallback('after-poison'); + expect((await reconcile()).skipped).toBe(200); + expect((await reconcile()).moved).toBe(1); + expect(await pgRowsFor(good._id)).toHaveLength(1); + expect(await MongoMessage.countDocuments({ reconcileFailedAt: { $exists: true } })).toBe(203); + expect((await reconcile()).skipped).toBe(0); + }); + + test('retry-ready orphan rows do not crowd out unattempted messages', async () => { + await MongoMessage.collection.insertMany(Array.from({ length: 201 }, () => ({ + podId: new mongoose.Types.ObjectId(), userId: new mongoose.Types.ObjectId(USER), + content: `${MARKER} retry-orphan`, createdAt: new Date(0), reconcileRetryAt: new Date(0), + }))); + const good = await insertFallback('new-after-retries'); + expect((await reconcile({ maxPerPass: 1 })).moved).toBe(1); + expect(await pgRowsFor(good._id)).toHaveLength(1); + }); + test('a second copy leaves the first reconciledAt and pgId unchanged', async () => { + const doc = await insertFallback('set-once'); + await copyOne(doc); + const first = await MongoMessage.findById(doc._id).lean(); + await new Promise((resolve) => setTimeout(resolve, 5)); + await copyOne(doc); + const second = await MongoMessage.findById(doc._id).lean(); + expect(second.reconciledAt).toEqual(first.reconciledAt); + expect(second.reconciledPgId).toBe(first.reconciledPgId); + expect(await pgRowsFor(doc._id)).toHaveLength(1); + }); + + test('a conflict row deleted before lookup is not stamped with a NaN pgId', async () => { + const doc = await insertFallback('deleted-during-copy'); + const { pool: appPool } = require('../../config/db-pg'); + const spy = jest.spyOn(appPool, 'query').mockResolvedValueOnce({ rows: [] }).mockResolvedValueOnce({ rows: [] }); + await expect(copyOne(doc)).rejects.toThrow('disappeared'); + spy.mockRestore(); + const after = await MongoMessage.findById(doc._id).lean(); + expect(after.reconciledAt).toBeUndefined(); + expect(after.reconciledPgId).toBeUndefined(); + }); + + test('card payload and kernel authorship survive reconciliation', async () => { + const payload = { type: 'decision', options: ['continue'] }; + const doc = await insertFallback('card', { messageType: 'card', payload, authorKind: 'kernel' }); + await reconcile(); + expect((await pgRowsFor(doc._id))[0]).toMatchObject({ payload, author_kind: 'kernel' }); + }); + + test('reconciled IDs work as replies and thread roots and reject missing or cross-pod targets', async () => { + const doc = await insertFallback('thread-root'); await reconcile(); + const { resolveThreadRoot, assertReplyTargetInPod } = require('../../services/threadRootResolver'); + const root = await resolveThreadRoot({ podId: POD, threadRootId: String(doc._id) }); + await expect(assertReplyTargetInPod({ podId: POD, replyToMessageId: String(doc._id) })).resolves.toBeUndefined(); + const reply = await PGMessage.create(POD, USER, `${MARKER} reply`, 'text', String(doc._id)); + expect(Number(reply.reply_to_message_id)).toBe(root); + expect(Number(reply.thread_root_id)).toBe(root); + await expect(resolveThreadRoot({ podId: POD, threadRootId: new mongoose.Types.ObjectId().toHexString() })).rejects.toMatchObject({ code: 'thread_root_invalid' }); + await expect(assertReplyTargetInPod({ podId: new mongoose.Types.ObjectId().toHexString(), replyToMessageId: String(doc._id) })).rejects.toMatchObject({ code: 'reply_target_unavailable' }); + }); - const row = (await pool.query( - 'SELECT payload, author_kind FROM messages WHERE legacy_mongo_id = $1', - [String(doc._id)], - )).rows[0]; - expect(row.payload).toEqual(payload); - expect(row.author_kind).toBe('kernel'); + test('all-digit ObjectIds resolve as Mongo IDs rather than PG integers', async () => { + const doc = await insertFallback('digit-id', { _id: new mongoose.Types.ObjectId('123456789012345678901234') }); + await reconcile(); + expect(await PGMessage.resolveId(String(doc._id))).toBe(String((await pgRowsFor(doc._id))[0].id)); }); test('a pod that exists only in Mongo is backfilled and its message then moves', async () => { diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index d6556b8ae9..4829ed5167 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -8,6 +8,8 @@ const { AgentInstallation } = require('../../../models/AgentRegistry'); jest.mock('../../../models/Pod'); jest.mock('../../../models/Message', () => ({ create: jest.fn(), + deleteOne: jest.fn(), + findByIdAndDelete: jest.fn(), findById: jest.fn(), find: jest.fn(), })); @@ -599,6 +601,7 @@ describe('messageController', () => { describe('deleteMessage', () => { it('returns 404 when message not found', async () => { + PGMessage.resolveId.mockResolvedValue(null); PGMessage.findById.mockResolvedValue(null); const req = { params: { id: 'm1' }, userId: 'u1', user: { id: 'u1' } }; const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; @@ -606,6 +609,37 @@ describe('messageController', () => { expect(res.status).toHaveBeenCalledWith(404); }); + it('deletes the resolved PG row and availability copy for a reconciled Mongo id', async () => { + PGMessage.resolveId.mockResolvedValue('41'); + PGMessage.findById.mockResolvedValue({ user_id: 'u1', pod_id: 'p1' }); + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + await messageController.deleteMessage({ params: { id: '123456789012345678901234' }, userId: 'u1' }, res); + expect(PGMessage.delete).toHaveBeenCalledWith('41'); + expect(MongoMessage.deleteOne).toHaveBeenCalledWith({ reconciledPgId: 41 }); + expect(MongoMessage.findByIdAndDelete).not.toHaveBeenCalled(); + expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' }); + }); + + it('refuses deletion of a reconciled row while PG is down without deleting either copy', async () => { + PGMessage.resolveId.mockRejectedValue(new Error('PG unavailable')); + MongoMessage.findById.mockResolvedValue({ reconciledAt: new Date(), userId: 'u1', podId: 'p1' }); + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + await messageController.deleteMessage({ params: { id: '123456789012345678901234' }, userId: 'u1' }, res); + expect(res.status).toHaveBeenCalledWith(503); + expect(PGMessage.delete).not.toHaveBeenCalled(); + expect(MongoMessage.deleteOne).not.toHaveBeenCalled(); + expect(MongoMessage.findByIdAndDelete).not.toHaveBeenCalled(); + }); + + it('deletes an unreconciled Mongo-only row', async () => { + PGMessage.resolveId.mockResolvedValue(null); + MongoMessage.findById.mockResolvedValue({ _id: 'mongo-only', userId: 'u1', podId: 'p1' }); + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + await messageController.deleteMessage({ params: { id: 'mongo-only' }, userId: 'u1' }, res); + expect(MongoMessage.findByIdAndDelete).toHaveBeenCalledWith('mongo-only'); + expect(PGMessage.delete).not.toHaveBeenCalled(); + }); + it('deletes message successfully for message owner', async () => { const mockMessage = { user_id: 'u1', pod_id: 'p1' }; PGMessage.resolveId.mockResolvedValue('1'); diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index dc858e8f9f..b52390553c 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -260,30 +260,18 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = // message and change nothing. There is nothing to reconcile until there // are two claims to reconcile. let resolvedThreadRootId: number | null = null; - // A message the reconciler moved answers to its Mongo id, so a reply or a - // thread post names it that way. Map it to the PG row before validation. - // An id that resolves to nothing is passed through as sent, and the - // resolver rejects it as it always has. - const pgReplyToMessageId = replyToMessageId - ? (await PGMessage.resolveId(String(replyToMessageId))) ?? replyToMessageId - : replyToMessageId; - const pgThreadRootId = threadRootId - ? (await PGMessage.resolveId(String(threadRootId))) ?? threadRootId - : threadRootId; try { - if (pgThreadRootId != null && pgThreadRootId !== '') { + if (threadRootId != null && threadRootId !== '') { // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { resolveThreadRoot } = require('../services/threadRootResolver'); - resolvedThreadRootId = await resolveThreadRoot({ - podId, replyToMessageId: pgReplyToMessageId, threadRootId: pgThreadRootId, - }); - } else if (pgReplyToMessageId) { + resolvedThreadRootId = await resolveThreadRoot({ podId, replyToMessageId, threadRootId }); + } else if (replyToMessageId) { // A reply's target must be in this pod (TASK-278). A store failure here // is left to the write below, which answers it and re-checks in SQL. // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { assertReplyTargetInPod } = require('../services/threadRootResolver'); try { - await assertReplyTargetInPod({ podId, replyToMessageId: pgReplyToMessageId }); + await assertReplyTargetInPod({ podId, replyToMessageId }); } catch (checkErr) { if ((checkErr as { name?: string })?.name === 'ThreadRootError') throw checkErr; } @@ -308,7 +296,7 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = userId, messageContent || '', 'text', - pgReplyToMessageId || null, + replyToMessageId || null, null, resolvedThreadRootId, ); @@ -504,6 +492,13 @@ exports.deleteMessage = async (req: AuthRequest, res: Response): Promise = if (pgId) { await PGMessage.delete(pgId); + // Keep the availability copy from resurrecting a deleted message during + // the next PG outage. Cleanup cannot turn a committed delete into a 500. + try { + await MongoMessage.deleteOne({ reconciledPgId: Number(pgId) }); + } catch (cleanupError) { + console.warn('[messageController] fallback delete cleanup failed:', (cleanupError as Error).message); + } } else { await MongoMessage.findByIdAndDelete(messageId); } diff --git a/backend/models/Message.ts b/backend/models/Message.ts index 4202b71960..9f18850205 100644 --- a/backend/models/Message.ts +++ b/backend/models/Message.ts @@ -30,10 +30,11 @@ export interface IMessage extends Document { // updateOne filters on its absence, so concurrent passes cannot re-stamp. reconciledAt?: Date; reconciledPgId?: number; - // A permanent refusal (e.g. the pod is not in Postgres). Excluded from the + // A permanent malformed-data refusal. Excluded from the // reconciler's scan so one such row cannot hold back the rows behind it. reconcileFailedAt?: Date; reconcileError?: string; + reconcileRetryAt?: Date; } const MessageSchema = new Schema({ @@ -61,6 +62,7 @@ const MessageSchema = new Schema({ reconciledPgId: { type: Number }, reconcileFailedAt: { type: Date }, reconcileError: { type: String }, + reconcileRetryAt: { type: Date }, }); MessageSchema.pre('save', function rememberWhetherThisIsANewPost(next) { diff --git a/backend/models/pg/Message.ts b/backend/models/pg/Message.ts index 1537325fde..b6cfbad0f8 100644 --- a/backend/models/pg/Message.ts +++ b/backend/models/pg/Message.ts @@ -1,3 +1,4 @@ +import { resolvePgMessageId } from '../../services/pgMessageId'; interface PgPool { query: (sql: string, params?: unknown[]) => Promise<{ rows: Record[]; @@ -119,6 +120,15 @@ class Message { answeringMessageId: string | null = null, authorKind: 'kernel' | null = null, ): Promise { + if (replyToMessageId != null && replyToMessageId !== '') { + const resolvedReply = await Message.resolveId(replyToMessageId); + if (!resolvedReply) { + // eslint-disable-next-line global-require, @typescript-eslint/no-require-imports + const { ThreadRootError } = require('../../services/threadRootResolver'); + throw new ThreadRootError('reply_target_unavailable', 'the message being replied to is not in this pod'); + } + replyToMessageId = resolvedReply; + } console.log('Creating message with params:', { podId, userId, content, messageType, podIdType: typeof podId, }); @@ -289,15 +299,7 @@ class Message { * was never moved, which is the caller's to look up in Mongo. */ static async resolveId(id: string | number): Promise { - const raw = String(id); - if (/^\d+$/.test(raw)) return raw; - if (!/^[0-9a-f]{24}$/i.test(raw)) return null; - const result = await (pool as PgPool).query( - 'SELECT id FROM messages WHERE legacy_mongo_id = $1 LIMIT 1', - [raw], - ); - const row = result.rows[0] as { id?: string | number } | undefined; - return row?.id != null ? String(row.id) : null; + return resolvePgMessageId(id); } static async findById(id: string): Promise { diff --git a/backend/routes/messages.ts b/backend/routes/messages.ts index 688d4114c1..f70b2d682e 100644 --- a/backend/routes/messages.ts +++ b/backend/routes/messages.ts @@ -85,7 +85,7 @@ const router: ReturnType = express.Router(); router.get('/:podId', readMessageRateLimit, auth, getMessages); router.post('/:podId', sendMessageRateLimit, auth, createMessage); -router.delete('/:id', auth, deleteMessage); +router.delete('/:id', sendMessageRateLimit, auth, deleteMessage); // Sprint B5 — reactions. Lives under /:messageId/reactions so the // `:podId` route above doesn't shadow it (Express routes by order + diff --git a/backend/services/pgMessageId.ts b/backend/services/pgMessageId.ts new file mode 100644 index 0000000000..c17513bf5b --- /dev/null +++ b/backend/services/pgMessageId.ts @@ -0,0 +1,13 @@ +// Shared by PG reads and thread validation so both accept reconciled client IDs. +// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require +const { pool } = require('../config/db-pg'); + +export async function resolvePgMessageId(id: string | number): Promise { + const raw = String(id); + // A Mongo ObjectId can contain only digits; test its shape first. + if (/^[0-9a-f]{24}$/i.test(raw)) { + const result = await pool.query('SELECT id FROM messages WHERE legacy_mongo_id = $1 LIMIT 1', [raw]); + return result.rows[0]?.id != null ? String(result.rows[0].id) : null; + } + return /^\d+$/.test(raw) ? raw : null; +} diff --git a/backend/services/pgMessageReconcileService.ts b/backend/services/pgMessageReconcileService.ts index 137503e384..c1f7dfabb4 100644 --- a/backend/services/pgMessageReconcileService.ts +++ b/backend/services/pgMessageReconcileService.ts @@ -8,18 +8,23 @@ * - Idempotent across replicas. The PG insert is ON CONFLICT on * legacy_mongo_id, so a second writer inserts nothing. The Mongo stamp is * guarded on `reconciledAt` being absent, so a second stamp matches nothing. - * - Bounded. One pass reads at most `maxPerPass` rows, oldest first. - * - Refusals are recorded, not retried. A row the database will never accept - * (its pod is not in Postgres) is stamped `reconcileFailedAt` and left out - * of the scan, so it cannot hold back the rows behind it. + * - Bounded. One pass reads at most `maxPerPass` rows, unattempted rows first, then the oldest deferred attempts. + * - Malformed rows are permanently refused. Recoverable failures (including + * missing PG pods) are deferred so later rows can progress, then retried. * * Reads never merge the two stores: chat reads Postgres when it is healthy and * Mongo only when it is not. A reconciled Mongo row therefore cannot be * returned alongside its PG copy in one response. */ +interface ReconcileClient { + query: (sql: string, params?: unknown[]) => Promise<{ rows: Record[] }>; + release: () => void; +} + // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const { pool } = require('../config/db-pg') as { pool: { + connect: () => Promise; query: (sql: string, params?: unknown[]) => Promise<{ rows: Record[] }>; }; }; @@ -28,10 +33,6 @@ const { syncPodFromMongo } = require('./pgPodSyncService') as { syncPodFromMongo: (podId: string, requestingUserId: string) => Promise; }; // eslint-disable-next-line @typescript-eslint/no-require-imports, global-require -const MongoPod = require('../models/Pod') as { - exists: (q: Record) => Promise; -}; -// eslint-disable-next-line @typescript-eslint/no-require-imports, global-require const MongoMessage = require('../models/Message') as { find: (q: Record) => { sort: (s: Record) => { @@ -50,9 +51,7 @@ interface MongoFallbackDoc { userId: { toString(): string }; content: string; messageType?: string; - // Both must survive the move: a card loses its face without payload, and a - // kernel seat post loses its marker without authorKind. - authorKind?: string | null; + authorKind?: 'kernel' | null; payload?: unknown; createdAt: Date; } @@ -71,8 +70,12 @@ export interface ReconcilePassResult { export const LEGACY_ID_INDEX = 'messages_legacy_mongo_id_key'; export const DEFAULT_MAX_PER_PASS = 200; -// Postgres SQLSTATE 23503: foreign_key_violation (messages.pod_id has no pod row). -const FK_VIOLATION = '23503'; +// FK failures can recover after pod synchronization; retry after a backoff. +export const RETRY_DELAY_MS = 60_000; +const isPermanentRowError = (err: unknown): boolean => { + const code = (err as { code?: string }).code; + return err instanceof TypeError || Boolean(code && code !== '23503' && /^(22|23)/.test(code)); +}; /** * Builds the unique index ON CONFLICT needs, without blocking writes. @@ -82,32 +85,40 @@ const FK_VIOLATION = '23503'; * INVALID index behind; IF NOT EXISTS would keep that forever, so an invalid * one is dropped and rebuilt. */ -export async function ensureLegacyIdIndex(): Promise { - const state = await pool.query( - `SELECT i.indisvalid FROM pg_index i - JOIN pg_class c ON c.oid = i.indexrelid - WHERE c.relname = $1`, - [LEGACY_ID_INDEX], - ); - if (state.rows[0] && state.rows[0].indisvalid === false) { - await pool.query(`DROP INDEX CONCURRENTLY IF EXISTS ${LEGACY_ID_INDEX}`); +export async function ensureLegacyIdIndex(): Promise { + const client = await pool.connect(); + let locked = false; + try { + const lock = await client.query('SELECT pg_try_advisory_lock(313, 2) AS acquired'); + locked = lock.rows[0]?.acquired === true; + if (!locked) return false; + const state = await client.query( + `SELECT i.indisvalid FROM pg_index i + JOIN pg_class c ON c.oid = i.indexrelid + WHERE c.relname = $1`, + [LEGACY_ID_INDEX], + ); + if (state.rows[0]?.indisvalid === false) { + await client.query(`DROP INDEX CONCURRENTLY IF EXISTS ${LEGACY_ID_INDEX}`); + } + await client.query( + `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ${LEGACY_ID_INDEX} + ON messages (legacy_mongo_id)`, + ); + return true; + } finally { + try { + if (locked) await client.query('SELECT pg_advisory_unlock(313, 2)'); + } finally { + client.release(); + } } - await pool.query( - `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS ${LEGACY_ID_INDEX} - ON messages (legacy_mongo_id)`, - ); -} - -async function podExistsInMongo(podId: string): Promise { - return Boolean(await MongoPod.exists({ _id: podId })); } -async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { +export async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { const id = doc._id.toString(); const inserted = await pool.query( - `INSERT INTO messages ( - pod_id, user_id, content, message_type, author_kind, payload, created_at, legacy_mongo_id - ) + `INSERT INTO messages (pod_id, user_id, content, message_type, created_at, legacy_mongo_id, author_kind, payload) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (legacy_mongo_id) DO NOTHING RETURNING id`, @@ -116,10 +127,10 @@ async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { doc.userId.toString(), doc.content || '', doc.messageType || 'text', - doc.authorKind === 'kernel' ? 'kernel' : null, - doc.payload == null ? null : JSON.stringify(doc.payload), doc.createdAt, id, + doc.authorKind === 'kernel' ? 'kernel' : null, + doc.payload == null ? null : JSON.stringify(doc.payload), ], ); let pgId = inserted.rows[0]?.id; @@ -130,6 +141,9 @@ async function copyOne(doc: MongoFallbackDoc): Promise<'moved' | 'present'> { ); pgId = existing.rows[0]?.id; } + if (pgId == null || !Number.isSafeInteger(Number(pgId)) || Number(pgId) <= 0) { + throw new Error('Reconciled PG row disappeared before stamping'); + } // Stamp only after Postgres has the row. A crash between the two leaves an // unstamped Mongo row that the next pass turns into `present`, then stamps. await MongoMessage.updateOne( @@ -152,28 +166,27 @@ export async function reconcileMongoFallbackMessages( } result.healthy = true; - await ensureLegacyIdIndex(); + if (!await ensureLegacyIdIndex()) return result; const batch = await MongoMessage.find({ reconciledAt: { $exists: false }, reconcileFailedAt: { $exists: false }, + $or: [{ reconcileRetryAt: { $exists: false } }, { reconcileRetryAt: { $lte: new Date() } }], }) - .sort({ createdAt: 1, _id: 1 }) + .sort({ reconcileRetryAt: 1, createdAt: 1, _id: 1 }) .limit(Math.max(1, Math.trunc(maxPerPass))) .lean(); for (const doc of batch) { try { const outcome = await copyOne(doc).catch(async (err: { code?: string }) => { - // The pod is missing from PG. The Mongo-only pod path leaves it that - // way until the next send backfills it (createMessage does the same - // before its insert), so backfill and retry once. - if (err.code !== FK_VIOLATION) throw err; + if (err.code !== '23503') throw err; + // A Mongo-only pod can be mirrored once PG returns. Keep failures + // retryable; even a missing Mongo pod may be restored later. try { await syncPodFromMongo(doc.podId.toString(), doc.userId.toString()); } catch { - // No Mongo pod to sync from. The retry below fails the same way and - // the outer handler decides whether that is permanent. + throw err; } return copyOne(doc); }); @@ -182,15 +195,18 @@ export async function reconcileMongoFallbackMessages( } catch (err) { const e = err as { code?: string; message?: string }; result.skipped += 1; - // Stamp a refusal only when the pod is absent from Mongo too: then no - // sync can ever place the message. Anything else stays in the scan and - // is retried on the next pass. - if (e.code === FK_VIOLATION && !(await podExistsInMongo(doc.podId.toString()))) { + if (isPermanentRowError(err)) { await MongoMessage.updateOne( { _id: doc._id, reconcileFailedAt: { $exists: false } }, { $set: { reconcileFailedAt: new Date(), reconcileError: e.message || e.code } }, ); } + else { + await MongoMessage.updateOne( + { _id: doc._id, reconciledAt: { $exists: false } }, + { $set: { reconcileRetryAt: new Date(Date.now() + RETRY_DELAY_MS) } }, + ); + } console.warn('[pgMessageReconcile] row not moved:', doc._id.toString(), e.message); } } diff --git a/backend/services/threadRootResolver.ts b/backend/services/threadRootResolver.ts index 92c8f0f54a..66024933ab 100644 --- a/backend/services/threadRootResolver.ts +++ b/backend/services/threadRootResolver.ts @@ -1,3 +1,4 @@ +import { resolvePgMessageId } from './pgMessageId'; /** * Resolve and validate a message's thread root (W-T, TASK-029). * @@ -57,7 +58,8 @@ export async function assertReplyTargetInPod({ replyToMessageId?: number | string | null; }): Promise { if (replyToMessageId == null || replyToMessageId === '') return; - const parentId = Number(replyToMessageId); + const resolved = await resolvePgMessageId(replyToMessageId); + const parentId = resolved == null ? NaN : Number(resolved); if (!Number.isInteger(parentId) || parentId <= 0) { throw new ThreadRootError('reply_target_invalid', 'replyToMessageId must be a positive integer'); } @@ -85,14 +87,15 @@ export async function resolveThreadRoot({ replyToMessageId?: number | string | null; threadRootId?: number | string | null; }): Promise { - const explicit = threadRootId == null || threadRootId === '' ? null : Number(threadRootId); + const rootId = threadRootId == null || threadRootId === '' ? null : await resolvePgMessageId(threadRootId); + const explicit = threadRootId == null || threadRootId === '' ? null : rootId == null ? NaN : Number(rootId); if (explicit !== null && (!Number.isInteger(explicit) || explicit <= 0)) { throw new ThreadRootError('thread_root_invalid', 'threadRootId must be a positive integer'); } await assertReplyTargetInPod({ podId, replyToMessageId }); const parentId = replyToMessageId == null || replyToMessageId === '' - ? null : Number(replyToMessageId); + ? null : Number(await resolvePgMessageId(replyToMessageId)); // Derive first when there is a reply edge — needed both as the answer (no // explicit root) and as the cross-check (explicit root present). From c8623d4e1cccde159c410a751bedce5ac6ff969b Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:02:34 -0700 Subject: [PATCH 4/6] fix(messages): reconcile only explicitly marked Mongo fallback writes Co-authored-by: kai Agent-Model: codex/gpt-6.1-sol/medium --- .../service/pgMessageReconcile.test.js | 20 +++++++++++++++---- .../controllers/messageController.test.js | 2 +- ...MessageService.threadRootBroadcast.test.js | 1 + backend/controllers/messageController.ts | 1 + backend/models/Message.ts | 9 +++++++++ backend/services/agentMessageService.ts | 1 + backend/services/pgMessageReconcileService.ts | 1 + 7 files changed, 30 insertions(+), 5 deletions(-) diff --git a/backend/__tests__/service/pgMessageReconcile.test.js b/backend/__tests__/service/pgMessageReconcile.test.js index 345f13b353..cee38caac5 100644 --- a/backend/__tests__/service/pgMessageReconcile.test.js +++ b/backend/__tests__/service/pgMessageReconcile.test.js @@ -37,6 +37,7 @@ const connect = () => new Pool({ }); const insertFallback = (content, extra = {}) => MongoMessage.create({ + pgFallback: true, podId: POD, userId: USER, content: `${MARKER} ${content}`, @@ -102,6 +103,17 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { expect(after.reconciledPgId).toBe(Number(rows[0].id)); }); + test('old unmarked history is never imported but an explicitly marked fallback moves', async () => { + const old = await insertFallback('old-history', { pgFallback: undefined, createdAt: new Date('2020-01-01') }); + const fallback = await insertFallback('marked-fallback'); + expect((await reconcile()).moved).toBe(1); + expect(await pgRowsFor(old._id)).toHaveLength(0); + expect(await pgRowsFor(fallback._id)).toHaveLength(1); + const unchanged = await MongoMessage.findById(old._id).lean(); + expect(unchanged.reconciledAt).toBeUndefined(); + expect(unchanged.reconcileFailedAt).toBeUndefined(); + }); + test('each message appears once in the Postgres read after a reconcile, and a second pass moves nothing', async () => { await insertFallback('b1'); await insertFallback('b2'); @@ -177,12 +189,12 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { test('more than a full batch of poison rows cannot starve a valid message', async () => { const bad = Array.from({ length: 201 }, () => ({ - podId: new mongoose.Types.ObjectId(POD), userId: new mongoose.Types.ObjectId(USER), + pgFallback: true, podId: new mongoose.Types.ObjectId(POD), userId: new mongoose.Types.ObjectId(USER), content: `${MARKER} bad\0`, createdAt: new Date(0), })); await MongoMessage.collection.insertMany([...bad, - { content: `${MARKER} missing-pod`, userId: new mongoose.Types.ObjectId(USER), createdAt: new Date(0) }, - { content: `${MARKER} missing-user`, podId: new mongoose.Types.ObjectId(POD), createdAt: new Date(0) }, + { pgFallback: true, content: `${MARKER} missing-pod`, userId: new mongoose.Types.ObjectId(USER), createdAt: new Date(0) }, + { pgFallback: true, content: `${MARKER} missing-user`, podId: new mongoose.Types.ObjectId(POD), createdAt: new Date(0) }, ]); const good = await insertFallback('after-poison'); expect((await reconcile()).skipped).toBe(200); @@ -194,7 +206,7 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { test('retry-ready orphan rows do not crowd out unattempted messages', async () => { await MongoMessage.collection.insertMany(Array.from({ length: 201 }, () => ({ - podId: new mongoose.Types.ObjectId(), userId: new mongoose.Types.ObjectId(USER), + pgFallback: true, podId: new mongoose.Types.ObjectId(), userId: new mongoose.Types.ObjectId(USER), content: `${MARKER} retry-orphan`, createdAt: new Date(0), reconcileRetryAt: new Date(0), }))); const good = await insertFallback('new-after-retries'); diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 4829ed5167..0ff988e13f 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -397,7 +397,7 @@ describe('messageController', () => { await messageController.createMessage(req, res); expect(MongoMessage.create).toHaveBeenCalledWith({ - podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text', + pgFallback: true, podId: 'p1', userId: 'u1', content: mongoFallbackRow.content, messageType: 'text', }); expect(res.status).not.toHaveBeenCalled(); expect(res.json).toHaveBeenCalledWith(expect.objectContaining({ diff --git a/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js index f5680780ae..9e77c8d5ce 100644 --- a/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js +++ b/backend/__tests__/unit/services/agentMessageService.threadRootBroadcast.test.js @@ -200,6 +200,7 @@ describe('agent broadcast carries author identity', () => { }); const msg = emitted.find((e) => e.event === 'newMessage'); + expect(Message).toHaveBeenCalledWith(expect.objectContaining({ pgFallback: true })); expect(msg.payload.userId).toEqual(expect.objectContaining({ _id: 'agent-user-1', isBot: true, diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index b52390553c..134362de81 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -330,6 +330,7 @@ exports.createMessage = async (req: AuthRequest, res: Response): Promise = return; } const mongoMsg = await MongoMessage.create({ + pgFallback: true, podId, userId, content: messageContent || '', diff --git a/backend/models/Message.ts b/backend/models/Message.ts index 9f18850205..9af8e046aa 100644 --- a/backend/models/Message.ts +++ b/backend/models/Message.ts @@ -35,6 +35,8 @@ export interface IMessage extends Document { reconcileFailedAt?: Date; reconcileError?: string; reconcileRetryAt?: Date; + /** Written only by the availability path, never inferred for legacy history. */ + pgFallback?: boolean; } const MessageSchema = new Schema({ @@ -63,8 +65,15 @@ const MessageSchema = new Schema({ reconcileFailedAt: { type: Date }, reconcileError: { type: String }, reconcileRetryAt: { type: Date }, + pgFallback: { type: Boolean }, }); +// Pending fallback work is small; keep the bounded scan off legacy history. +MessageSchema.index( + { pgFallback: 1, reconcileRetryAt: 1, createdAt: 1, _id: 1 }, + { partialFilterExpression: { pgFallback: true } }, +); + MessageSchema.pre('save', function rememberWhetherThisIsANewPost(next) { (this as any).$locals.wasNewPost = this.isNew; if (this.isNew) this.freshnessPending = true; diff --git a/backend/services/agentMessageService.ts b/backend/services/agentMessageService.ts index 59eb14d63b..1114c2079c 100644 --- a/backend/services/agentMessageService.ts +++ b/backend/services/agentMessageService.ts @@ -1804,6 +1804,7 @@ class AgentMessageService { if (!message) { const mongoMessage = new Message({ + pgFallback: true, content, userId: agentUser._id, podId, diff --git a/backend/services/pgMessageReconcileService.ts b/backend/services/pgMessageReconcileService.ts index c1f7dfabb4..575c20a4de 100644 --- a/backend/services/pgMessageReconcileService.ts +++ b/backend/services/pgMessageReconcileService.ts @@ -169,6 +169,7 @@ export async function reconcileMongoFallbackMessages( if (!await ensureLegacyIdIndex()) return result; const batch = await MongoMessage.find({ + pgFallback: true, reconciledAt: { $exists: false }, reconcileFailedAt: { $exists: false }, $or: [{ reconcileRetryAt: { $exists: false } }, { reconcileRetryAt: { $lte: new Date() } }], From b7fcb71f620acb65d210ab8c761c90b4dbf7aa73 Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:09:18 -0700 Subject: [PATCH 5/6] test: retire fixed message delete rate-limit exception Co-authored-by: kai Agent-Model: codex/gpt-6.1-sol/medium --- backend/__tests__/unit/routes/routeRateLimitGuard.baseline.json | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/__tests__/unit/routes/routeRateLimitGuard.baseline.json b/backend/__tests__/unit/routes/routeRateLimitGuard.baseline.json index f3cdee71d3..cb228732a3 100644 --- a/backend/__tests__/unit/routes/routeRateLimitGuard.baseline.json +++ b/backend/__tests__/unit/routes/routeRateLimitGuard.baseline.json @@ -105,7 +105,6 @@ "routes/marketplace-api.ts POST /publish [unlimited]", "routes/marketplace-api.ts POST /publish/:installableId(*)/deprecate [unlimited]", "routes/marketplace.ts GET /official [unlimited]", - "routes/messages.ts DELETE /:id [unlimited]", "routes/messages.ts PATCH /:id/created-at [unlimited]", "routes/pg-status.ts GET / [unlimited]", "routes/pg-status.ts POST /sync-user [unlimited]", From ab93e78d499411b5027865c82cf6a6708c79e25d Mon Sep 17 00:00:00 2001 From: Lily Shen <115414357+lilyshen0722@users.noreply.github.com> Date: Sat, 24 Oct 2026 06:17:46 -0700 Subject: [PATCH 6/6] fix(messages): remove unstamped fallback copies when deleting reconciled rows Co-authored-by: kai Agent-Model: codex/gpt-6.1-sol/medium --- .../__tests__/service/pgMessageReconcile.test.js | 15 +++++++++++++++ .../unit/controllers/messageController.test.js | 4 +++- backend/controllers/messageController.ts | 10 +++++++++- 3 files changed, 27 insertions(+), 2 deletions(-) diff --git a/backend/__tests__/service/pgMessageReconcile.test.js b/backend/__tests__/service/pgMessageReconcile.test.js index cee38caac5..a731be09ff 100644 --- a/backend/__tests__/service/pgMessageReconcile.test.js +++ b/backend/__tests__/service/pgMessageReconcile.test.js @@ -293,6 +293,21 @@ d('TASK-313 part 2: reconcile Mongo fallback messages into PostgreSQL', () => { expect(await pgRowsFor(doc._id)).toHaveLength(0); }); + test.each(['mongo', 'pg'])('delete by %s ID removes the unstamped availability copy before another pass', async (idKind) => { + const doc = await insertFallback('crashed-stamp-delete'); + await copyOne(doc); + const pgId = await PGMessage.resolveId(String(doc._id)); + await MongoMessage.updateOne({ _id: doc._id }, { $unset: { reconciledAt: 1, reconciledPgId: 1 } }); + const controller = require('../../controllers/messageController'); + const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; + await controller.deleteMessage({ params: { id: idKind === 'mongo' ? String(doc._id) : pgId }, userId: USER }, res); + expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' }); + expect(await MongoMessage.findById(doc._id).lean()).toBeNull(); + expect(await pgRowsFor(doc._id)).toHaveLength(0); + expect((await reconcile()).moved).toBe(0); + expect(await pgRowsFor(doc._id)).toHaveLength(0); + }); + test('a moved message resolves its Mongo id for reactions', async () => { const doc = await insertFallback('g'); await reconcile(); diff --git a/backend/__tests__/unit/controllers/messageController.test.js b/backend/__tests__/unit/controllers/messageController.test.js index 0ff988e13f..eaafe8393a 100644 --- a/backend/__tests__/unit/controllers/messageController.test.js +++ b/backend/__tests__/unit/controllers/messageController.test.js @@ -615,7 +615,9 @@ describe('messageController', () => { const res = { status: jest.fn().mockReturnThis(), json: jest.fn() }; await messageController.deleteMessage({ params: { id: '123456789012345678901234' }, userId: 'u1' }, res); expect(PGMessage.delete).toHaveBeenCalledWith('41'); - expect(MongoMessage.deleteOne).toHaveBeenCalledWith({ reconciledPgId: 41 }); + expect(MongoMessage.deleteOne).toHaveBeenCalledWith({ $or: [ + { reconciledPgId: 41 }, { _id: '123456789012345678901234' }, + ] }); expect(MongoMessage.findByIdAndDelete).not.toHaveBeenCalled(); expect(res.json).toHaveBeenCalledWith({ msg: 'Message deleted' }); }); diff --git a/backend/controllers/messageController.ts b/backend/controllers/messageController.ts index 134362de81..56230608ef 100644 --- a/backend/controllers/messageController.ts +++ b/backend/controllers/messageController.ts @@ -24,6 +24,7 @@ interface AuthRequest extends Request { } interface NormalizedMessage { + legacy_mongo_id?: string; id: string; pod_id: string; user_id: string; @@ -496,7 +497,14 @@ exports.deleteMessage = async (req: AuthRequest, res: Response): Promise = // Keep the availability copy from resurrecting a deleted message during // the next PG outage. Cleanup cannot turn a committed delete into a 500. try { - await MongoMessage.deleteOne({ reconciledPgId: Number(pgId) }); + const copyKeys: Array> = [{ reconciledPgId: Number(pgId) }]; + // PG can commit before its Mongo stamp. The stable legacy ID still + // identifies that unstamped copy, even when deletion used the PG ID. + const legacyId = message.legacy_mongo_id || messageId; + if (typeof legacyId === 'string' && /^[0-9a-f]{24}$/i.test(legacyId)) { + copyKeys.push({ _id: legacyId }); + } + await MongoMessage.deleteOne({ $or: copyKeys }); } catch (cleanupError) { console.warn('[messageController] fallback delete cleanup failed:', (cleanupError as Error).message); }