From 81628db1d7f45b60266128a8bc87580ace04a1da Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 21 Aug 2026 02:54:42 -0700 Subject: [PATCH] refactor(outbound): remove beta queue reply migration (#127133) * refactor(outbound): remove beta queue reply migration * test(outbound): retain legacy reply migration coverage * test(outbound): type legacy migration fixture --- .../outbound/delivery-queue-migration.test.ts | 111 +++--------------- .../outbound/delivery-queue-migration.ts | 47 -------- 2 files changed, 15 insertions(+), 143 deletions(-) diff --git a/src/infra/outbound/delivery-queue-migration.test.ts b/src/infra/outbound/delivery-queue-migration.test.ts index 57fb51752364..ebba8074544d 100644 --- a/src/infra/outbound/delivery-queue-migration.test.ts +++ b/src/infra/outbound/delivery-queue-migration.test.ts @@ -26,6 +26,7 @@ import { markDeliveryPlatformOutcomeUnknown, markDeliveryPlatformSendAttemptStarted, reserveDeliveryAttempt, + type LegacyQueuedDelivery, type LegacyQueuedDeliveryPreparation, type QueuedDelivery, } from "./delivery-queue-storage.js"; @@ -144,106 +145,17 @@ describe("outbound prepared queue migration", () => { ); }); - it("persists canonical reply facts across prepared namespaces exactly once", async () => { - const cases = [ - { - label: "first", - legacy: { replyToId: "legacy-first", replyToMode: "first" }, - expected: { source: "implicit", replyToId: "legacy-first", mode: "first" }, - }, - { - label: "batched", - legacy: { replyToId: "legacy-batched", replyToMode: "batched" }, - expected: { source: "implicit", replyToId: "legacy-batched", mode: "first" }, - }, - { - label: "all", - legacy: { replyToId: "legacy-all", replyToMode: "all" }, - expected: { source: "implicit", replyToId: "legacy-all", mode: "all" }, - }, - { - label: "off", - legacy: { replyToId: "legacy-off", replyToMode: "off" }, - expected: undefined, - }, - { - label: "explicit", - legacy: { - reply: { source: "explicit", replyToId: "explicit-root" }, - replyToId: "legacy-first", - replyToMode: "first", - }, - expected: { source: "explicit", replyToId: "explicit-root" }, - }, - ] as const; - const sourceNamespaces = [ - OUTBOUND_DELIVERY_QUEUE_NAME, - OUTBOUND_DELIVERY_MIGRATION_QUEUE_NAME, - ] as const; - const ids: string[] = []; - - for (const sourceQueueName of sourceNamespaces) { - for (const testCase of cases) { - const id = `${sourceQueueName}-${testCase.label}`; - ids.push(id); - const entry = { - id, - enqueuedAt: 100, - retryCount: 0, - attemptCount: 0, - channel: "matrix", - to: "!room:example", - queuePolicy: "required", - preparedBatch: { - schemaVersion: 1, - sourcePayloadCount: 1, - entries: [{ sourceIndex: 0, status: "accepted", payload: { text: id } }], - }, - ...testCase.legacy, - }; - upsertDeliveryQueueEntry({ - queueName: sourceQueueName, - entry, - stateDir: tmpDir(), - }); - } - } - - await migrateLegacyPendingOutboundDeliveries({ - cfg: {}, - log: createRecoveryLog(), - stateDir: tmpDir(), - }); - - const firstPass = new Map(); - for (const id of ids) { - const raw = readQueueEntryJson(OUTBOUND_DELIVERY_QUEUE_NAME, id, tmpDir()); - expect(raw).toBeDefined(); - firstPass.set(id, raw ?? ""); - const persisted = JSON.parse(raw ?? "{}") as Record; - const testCase = cases.find((candidate) => id.endsWith(`-${candidate.label}`)); - expect(persisted.reply).toEqual(testCase?.expected); - expect(persisted).not.toHaveProperty("replyToId"); - expect(persisted).not.toHaveProperty("replyToMode"); - } - - await migrateLegacyPendingOutboundDeliveries({ - cfg: {}, - log: createRecoveryLog(), - stateDir: tmpDir(), - }); - for (const id of ids) { - expect(readQueueEntryJson(OUTBOUND_DELIVERY_QUEUE_NAME, id, tmpDir())).toBe( - firstPass.get(id), - ); - } - }); - it("prepares a legacy row once, fences rollback, and never reruns modifiers", async () => { const id = "stable-legacy-delivery"; + const source = { + ...legacyEntry(id, "secret"), + completionRetention: "permanent", + replyToId: "root-message", + replyToMode: "batched", + } satisfies LegacyQueuedDelivery; upsertDeliveryQueueEntry({ queueName: LEGACY_OUTBOUND_DELIVERY_QUEUE_NAME, - entry: { ...legacyEntry(id, "secret"), completionRetention: "permanent" }, + entry: source, stateDir: tmpDir(), }); @@ -269,6 +181,13 @@ describe("outbound prepared queue migration", () => { expect( acceptedPreparedOutboundEntries(queued.preparedBatch).map((entry) => entry.payload), ).toEqual([{ text: "secret-prepared" }]); + expect(queued.reply).toEqual({ + source: "implicit", + replyToId: "root-message", + mode: "first", + }); + expect(queued).not.toHaveProperty("replyToId"); + expect(queued).not.toHaveProperty("replyToMode"); expect(queued).not.toHaveProperty("legacyPreparationOwnerId"); expect(queued).not.toHaveProperty("legacyPreparationLeaseExpiresAt"); diff --git a/src/infra/outbound/delivery-queue-migration.ts b/src/infra/outbound/delivery-queue-migration.ts index 1aad71aa2fa6..00a336893e60 100644 --- a/src/infra/outbound/delivery-queue-migration.ts +++ b/src/infra/outbound/delivery-queue-migration.ts @@ -10,8 +10,6 @@ import { } from "../delivery-queue-sqlite-namespace.js"; import { countPendingDeliveryQueueEntries, - loadDeliveryQueueEntries, - loadDeliveryQueueEntry, terminalizePendingDeliveryQueueEntry, } from "../delivery-queue-sqlite.js"; import { @@ -54,47 +52,6 @@ import { normalizeOutboundReplyFacts } from "./reply-policy.js"; const LEGACY_PREPARATION_LEASE_MS = 5 * 60_000; const LEGACY_PREPARATION_LEASE_RENEW_MS = 30_000; -type LegacyPreparedQueuedDelivery = QueuedDelivery & - Parameters[0]; - -function hasLegacyReplyFields(entry: LegacyPreparedQueuedDelivery): boolean { - return Object.hasOwn(entry, "replyToId") || Object.hasOwn(entry, "replyToMode"); -} - -function canonicalizePreparedReplyFields(entry: LegacyPreparedQueuedDelivery): QueuedDelivery { - const { replyToId, replyToMode, reply: storedReply, ...canonical } = entry; - const reply = normalizeOutboundReplyFacts({ reply: storedReply, replyToId, replyToMode }); - return { ...canonical, ...(reply ? { reply } : {}) }; -} - -function migratePreparedReplyFields(queueName: string, stateDir?: string): void { - const entries = loadDeliveryQueueEntries(queueName, stateDir) as LegacyPreparedQueuedDelivery[]; // SAFETY: callers pass prepared namespaces; beta rows add only legacy reply fields. - for (const entry of entries) { - if (!hasLegacyReplyFields(entry)) { - continue; - } - const migrated = canonicalizePreparedReplyFields(entry); - if ( - replacePendingDeliveryQueueEntry({ - queueName, - expectedEntry: entry, - replacementEntry: migrated, - stateDir, - }) - ) { - continue; - } - const current = loadDeliveryQueueEntry( - queueName, - entry.id, - stateDir, - ) as LegacyPreparedQueuedDelivery | null; // SAFETY: same prepared namespace/id; replacement keeps shape or removes row. - if (current && hasLegacyReplyFields(current)) { - throw new Error(`Prepared delivery ${entry.id} changed during reply migration`); - } - } -} - function withLegacyPreparationLease( entry: LegacyQueuedDeliveryPreparation, ownerId: string, @@ -537,10 +494,6 @@ async function migrateLegacyPendingOutboundDeliveriesOwned(params: { log: RecoveryLogger; stateDir?: string; }): Promise { - // Beta rows can exist in either prepared namespace. Canonicalize them before - // any recovery owner sees the row; interrupted migrations then resume normally. - migratePreparedReplyFields(OUTBOUND_DELIVERY_QUEUE_NAME, params.stateDir); - migratePreparedReplyFields(OUTBOUND_DELIVERY_MIGRATION_QUEUE_NAME, params.stateDir); let moved = 0; let skipped = 0; const ownerId = randomUUID();