[PR] fix(storage): refs ledger diffs instead of full-replace per save (#5744)

src/server.ts

@@ -4622,11 +4622,15 @@ function formatBytes(n: number): string {
 // bytes are counted with Buffer.byteLength since undici sends UTF-8 bytes even
 // for string bodies (.length would count UTF-16 chars and undercount any
 // non-ASCII injection). Omitted on paths that fail fast without forwarding.
-// #4528: inbound histories past this size pay an O(messages) synchronous scan
-// pass per turn on the main thread, large enough to starve the event loop and
+// #4528: bili's own per-request processing past this WALL-CLOCK budget means
+// synchronous main-thread work large enough to starve the event loop and
 // freeze every endpoint on the same listener — warn so a pathological session
 // shows up in the log instead of being found through a frozen UI.
-const SLOW_SESSION_MSG_WARN = 1000;
+// #5744: the trigger used to be message COUNT (>1000), which cannot separate
+// cost from size — one reporter session saved at 0.8s AND 14.5s at the same
+// ~3k msgs. Wall clock IS the cost; the info line above keeps reporting the
+// count for correlation.
+const SLOW_SESSION_MS_WARN = 1000;
 export function logRequestCost(log: (level: string, msg: string) => void, sessionId: string, msgs: number | null, inboundBytes: number, t0: number, outbound?: string | Buffer): void {
     const ms = Math.max(0, Math.round(performance.now() - t0));
     const outboundField = outbound !== undefined ? `, outbound=${formatBytes(Buffer.byteLength(outbound))}` : "";
@@ -4635,8 +4639,8 @@ export function logRequestCost(log: (level: string, msg: string) => void, sessio
     // #5408: this line is THE observation point — a hot-at-attach session that
     // never produces one is silent compression loss (the incident signature).
     noteRequest(sessionId);
-    if (msgs !== null && msgs > SLOW_SESSION_MSG_WARN) {
-        log("warn", `[${sessionId}] slow-session: ${msgs} msgs / inbound=${formatBytes(inboundBytes)} took local=${ms}ms — per-turn O(messages) synchronous scans are starving the event loop (#4528); consider closing/restarting the session or enabling compression`);
+    if (ms > SLOW_SESSION_MS_WARN) {
+        log("warn", `[${sessionId}] slow-session: ${msgs ?? "?"} msgs / inbound=${formatBytes(inboundBytes)} took local=${ms}ms — synchronous per-turn work is starving the event loop (#4528/#5744); consider closing/restarting the session or enabling compression`);
     }
 }
 

src/storage/ingest.ts

@@ -403,6 +403,8 @@ function finishPlan(
         const views = completeBlockContents(session.blockContents, session.state.blocks).get(b.blockId);
         const full = views?.full ?? { text: b.summary, count: 0 };
         const one = views?.one ?? null;
+        const summaryJson = JSON.stringify(full);
+        const oneJson = one === null ? null : JSON.stringify(one);
         const covered = { effectiveMessageIds: b.effectiveMessageIds, directMessageIds: b.directMessageIds };
         let startSeq = -1;
         let endSeq = -1;
@@ -420,8 +422,13 @@ function finishPlan(
             startRef: b.startRef !== undefined ? refToIndex(b.startRef) : null,
             endRef: b.endRef !== undefined ? refToIndex(b.endRef) : null,
             tier: b.tier,
-            summary: JSON.stringify(full),
-            one: one === null ? null : JSON.stringify(one),
+            summary: summaryJson,
+            one: oneJson,
+            // #5744: carry the digests so ingestSession can compare against the
+            // stored row BEFORE any CAS op (putContent with a precomputed hash
+            // recomputes nothing — same bytes, same utf8, same sha256).
+            summaryHash: sha256Hex(summaryJson),
+            oneHash: oneJson === null ? null : sha256Hex(oneJson),
             supersedesBlockId: null,
             active: b.active,
             createdAt: b.createdAt,
@@ -451,11 +458,10 @@ function finishPlan(
     }));
 
     const ccrEntries: CcrEntryRow[] = session.contentStore
-        ? Object.entries(session.contentStore.byRef).map(([key, value], ord) => ({
-              ord,
-              key,
-              content: JSON.stringify(value),
-          }))
+        ? Object.entries(session.contentStore.byRef).map(([key, value], ord) => {
+              const content = JSON.stringify(value);
+              return { ord, key, content, contentHash: sha256Hex(content) };
+          })
         : [];
 
     const kstate: Record<string, unknown> = {};

src/storage/mysql.ts

@@ -92,6 +92,10 @@ const MIGRATION_LOCK_POLL_S = 30;
 /** PR #4725 N3 (mirror of store.ts): covered-seq span point queries chunk the
  *  msg_id IN (…) list at the same 500 batch size for cross-backend parity. */
 const SEQ_SPAN_QUERY_BATCH = 500;
+/** Max bound parameters per generated statement (refs/ccr batching, #5744) —
+ *  mirror of the sqlite twin's constant; well under MySQL's max_allowed_packet
+ *  and driver placeholder limits. */
+const REF_STMT_MAX_PARAMS = 700;
 
 /** #4796: rows per batch of gc()'s zero-ref CAS sweep (mysql counterpart of
  *  GC_ZERO_REF_BATCH_SIZE in store.ts): one giant DELETE holds every row
@@ -800,16 +804,28 @@ export class MysqlBackend implements StorageBackend {
                     }
                 }
             }
-            const incomingBlockIds = new Set(input.blocks.map((b) => b.blockId));
-            const staleBlocks = (await c.query(
-                "SELECT block_id, summary_hash, one_hash FROM blocks WHERE session_id = ?",
+            // #5744 (mirror of store.ts): ONE full-row read serves both the
+            // stale sweep and the unchanged-row skip below.
+            type StoredBlockRow = {
+                block_id: string | number; ord: number | string; start_seq: number | string; end_seq: number | string;
+                start_ref: number | null; end_ref: number | null; tier: number | string;
+                summary_hash: string | number; one_hash: string | null; supersedes_block_id: string | null;
+                active: number | string; created_at: number | string; covered_hash: string | number; payload_json: string | number;
+            };
+            const storedBlocks = new Map<string, StoredBlockRow>();
+            const storedBlockRows = (await c.query(
+                "SELECT block_id, ord, start_seq, end_seq, start_ref, end_ref, tier, " +
+                "summary_hash, one_hash, supersedes_block_id, active, created_at, covered_hash, payload_json " +
+                "FROM blocks WHERE session_id = ?",
                 [s.sessionId],
-            )) as unknown as Row[][];
-            for (const stale of staleBlocks[0]) {
-                if (incomingBlockIds.has(str(stale.block_id))) continue;
+            )) as unknown as StoredBlockRow[][];
+            for (const row of storedBlockRows[0]) storedBlocks.set(str(row.block_id), row);
+            for (const [blockId, stale] of storedBlocks) {
+                if (input.blocks.some((b) => b.blockId === blockId)) continue;
+                // referencing row first, then release (#4242 delete-on-zero)
+                await c.query("DELETE FROM blocks WHERE session_id = ? AND block_id = ?", [s.sessionId, blockId]);
                 await this.releaseContent(c, str(stale.summary_hash));
                 if (stale.one_hash !== null && stale.one_hash !== undefined) await this.releaseContent(c, str(stale.one_hash));
-                await c.query("DELETE FROM blocks WHERE session_id = ? AND block_id = ?", [s.sessionId, stale.block_id]);
             }
             // #4316 layer 3 (mirror of store.ts): resolve coverage to canonical
             // seq bounds from the append-only transcript; unresolvable stores
@@ -866,18 +882,35 @@ export class MysqlBackend implements StorageBackend {
             };
             for (const b of input.blocks) {
                 const [bStartSeq, bEndSeq] = boundsForCovered(b.payloadJson);
-                const summaryHash = await this.putContent(c, b.summary);
-                const oneHash = b.one === null ? null : await this.putContent(c, b.one);
-                const [existingRows] = (await c.query(
-                    "SELECT summary_hash, one_hash FROM blocks WHERE session_id = ? AND block_id = ?",
-                    [s.sessionId, b.blockId],
-                )) as unknown as Row[][];
-                const existing = existingRows[0];
-                if (existing) {
-                    if (str(existing.summary_hash) === summaryHash) await this.releaseContent(c, summaryHash);
-                    else await this.releaseContent(c, str(existing.summary_hash));
-                    if (existing.one_hash !== null && existing.one_hash !== undefined) {
-                        const exOne = str(existing.one_hash);
+                // #5744 (mirror of store.ts): skip the whole write + CAS dance
+                // when the effective post-upsert row equals the stored row —
+                // steady-state saves re-send every block byte-identical. The
+                // -1/unknown bound keeps its stored value via the IF(>=0)
+                // guard below, so "incoming < 0" counts as unchanged.
+                const ex = storedBlocks.get(b.blockId);
+                if (ex &&
+                    num(ex.ord) === b.ord &&
+                    (bStartSeq < 0 || bStartSeq === num(ex.start_seq)) &&
+                    (bEndSeq < 0 || bEndSeq === num(ex.end_seq)) &&
+                    (ex.start_ref === null ? null : num(ex.start_ref)) === b.startRef &&
+                    (ex.end_ref === null ? null : num(ex.end_ref)) === b.endRef &&
+                    num(ex.tier) === b.tier &&
+                    str(ex.summary_hash) === b.summaryHash &&
+                    (ex.one_hash === null || ex.one_hash === undefined ? null : str(ex.one_hash)) === b.oneHash &&
+                    (ex.supersedes_block_id === null || ex.supersedes_block_id === undefined ? null : str(ex.supersedes_block_id)) === b.supersedesBlockId &&
+                    num(ex.active) === (b.active ? 1 : 0) &&
+                    num(ex.created_at) === b.createdAt &&
+                    str(ex.covered_hash) === b.coveredHash &&
+                    str(ex.payload_json) === b.payloadJson) {
+                    continue;
+                }
+                const summaryHash = await this.putContent(c, b.summary, b.summaryHash);
+                const oneHash = b.one === null ? null : await this.putContent(c, b.one, b.oneHash ?? undefined);
+                if (ex) {
+                    if (str(ex.summary_hash) === summaryHash) await this.releaseContent(c, summaryHash);
+                    else await this.releaseContent(c, str(ex.summary_hash));
+                    if (ex.one_hash !== null && ex.one_hash !== undefined) {
+                        const exOne = str(ex.one_hash);
                         if (exOne === oneHash && oneHash !== null) await this.releaseContent(c, oneHash);
                         else await this.releaseContent(c, exOne);
                     }
@@ -924,22 +957,88 @@ export class MysqlBackend implements StorageBackend {
                     [s.sessionId, d.ord, d.ref, d.refNum, d.recordedAt],
                 );
             }
-            await c.query("DELETE FROM refs WHERE session_id = ?", [s.sessionId]);
+            // #5744 (mirror of store.ts): diff against one indexed read of the
+            // stored set instead of DELETE-all + per-row INSERT — the network
+            // round-trip cost of a full-history rewrite is worse than sqlite's.
+            const oldRefs = new Map<string, { ord: number; ref: string; refNum: number | null }>();
+            const storedRefRows = (await c.query(
+                "SELECT raw_id, ord, ref, ref_num FROM refs WHERE session_id = ?",
+                [s.sessionId],
+            )) as unknown as Row[][];
+            for (const row of storedRefRows[0]) {
+                oldRefs.set(str(row.raw_id), { ord: num(row.ord), ref: str(row.ref), refNum: row.ref_num === null || row.ref_num === undefined ? null : num(row.ref_num) });
+            }
+            const goneRefIds: string[] = [];
+            const touchedRefs: RefRow[] = [];
             for (const r of input.refs) {
-                await c.query(
-                    "INSERT INTO refs (session_id, ord, raw_id, ref, ref_num) VALUES (?, ?, ?, ?, ?) " +
-                        "ON DUPLICATE KEY UPDATE ord = VALUES(ord), ref = VALUES(ref), ref_num = VALUES(ref_num)",
-                    [s.sessionId, r.ord, r.rawId, r.ref, r.refNum],
-                );
+                const o = oldRefs.get(r.rawId);
+                if (!o || o.ord !== r.ord || o.ref !== r.ref || o.refNum !== r.refNum) {
+                    touchedRefs.push(r);
+                } else {
+                    oldRefs.delete(r.rawId);
+                }
             }
-            const oldCcr = (await c.query(
-                "SELECT content_hash FROM ccr_entries WHERE session_id = ?",
+            for (const rawId of oldRefs.keys()) goneRefIds.push(rawId);
+            if (goneRefIds.length > 0) {
+                const perChunk = REF_STMT_MAX_PARAMS - 1;
+                for (let i = 0; i < goneRefIds.length; i += perChunk) {
+                    const ids = goneRefIds.slice(i, i + perChunk);
+                    await c.query(
+                        "DELETE FROM refs WHERE session_id = ? AND raw_id IN (" + ids.map(() => "?").join(", ") + ")",
+                        [s.sessionId, ...ids],
+                    );
+                }
+            }
+            if (touchedRefs.length > 0) {
+                const colsPerRow = 5;
+                const perChunk = Math.floor(REF_STMT_MAX_PARAMS / colsPerRow);
+                for (let i = 0; i < touchedRefs.length; i += perChunk) {
+                    const rows = touchedRefs.slice(i, i + perChunk);
+                    const params: unknown[] = [];
+                    for (const r of rows) params.push(s.sessionId, r.ord, r.rawId, r.ref, r.refNum);
+                    await c.query(
+                        "INSERT INTO refs (session_id, ord, raw_id, ref, ref_num) VALUES " +
+                            rows.map(() => "(?, ?, ?, ?, ?)").join(", ") +
+                            " ON DUPLICATE KEY UPDATE ord = VALUES(ord), ref = VALUES(ref), ref_num = VALUES(ref_num)",
+                        params,
+                    );
+                }
+            }
+            // #5744 (mirror of store.ts): diff instead of DELETE-all +
+            // re-INSERT-all; unchanged entries skip the CAS round-trips.
+            const incomingCcr = new Map<string, CcrEntryRow>();
+            for (const e of input.ccrEntries) incomingCcr.set(e.key, e);
+            const storedCcrRows = (await c.query(
+                "SELECT ord, ccr_key, content_hash FROM ccr_entries WHERE session_id = ?",
                 [s.sessionId],
             )) as unknown as Row[][];
-            for (const row of oldCcr[0]) await this.releaseContent(c, str(row.content_hash));
-            await c.query("DELETE FROM ccr_entries WHERE session_id = ?", [s.sessionId]);
+            const changedCcrKeys: string[] = [];
+            const unchangedCcrKeys = new Set<string>();
+            const releasedOldCcrHashes: string[] = [];
+            for (const row of storedCcrRows[0]) {
+                const key = str(row.ccr_key);
+                const inc = incomingCcr.get(key);
+                if (!inc) { changedCcrKeys.push(key); releasedOldCcrHashes.push(str(row.content_hash)); continue; }
+                if (num(row.ord) !== inc.ord || str(row.content_hash) !== inc.contentHash) {
+                    changedCcrKeys.push(key);
+                    releasedOldCcrHashes.push(str(row.content_hash));
+                } else {
+                    unchangedCcrKeys.add(key);
+                }
+            }
+            if (changedCcrKeys.length > 0) {
+                for (let i = 0; i < changedCcrKeys.length; i += SEQ_SPAN_QUERY_BATCH) {
+                    const keys = changedCcrKeys.slice(i, i + SEQ_SPAN_QUERY_BATCH);
+                    await c.query(
+                        "DELETE FROM ccr_entries WHERE session_id = ? AND ccr_key IN (" + keys.map(() => "?").join(", ") + ")",
+                        [s.sessionId, ...keys],
+                    );
+                }
+            }
+            for (const hash of releasedOldCcrHashes) await this.releaseContent(c, hash);
             for (const e of input.ccrEntries) {
-                const hash = await this.putContent(c, e.content);
+                if (unchangedCcrKeys.has(e.key)) continue;
+                const hash = await this.putContent(c, e.content, e.contentHash);
                 await c.query(
                     "INSERT INTO ccr_entries (session_id, ord, ccr_key, content_hash) VALUES (?, ?, ?, ?)",
                     [s.sessionId, e.ord, e.key, hash],
@@ -1650,7 +1749,7 @@ export class MysqlBackend implements StorageBackend {
         )) as unknown as Row[][];
         const out: CcrEntryRow[] = [];
         for (const r of rows) {
-            out.push({ ord: num(r.ord), key: str(r.ccr_key), content: (await this.getContent(this.pool, str(r.content_hash))) ?? "" });
+            out.push({ ord: num(r.ord), key: str(r.ccr_key), content: (await this.getContent(this.pool, str(r.content_hash))) ?? "", contentHash: str(r.content_hash) });
         }
         return out;
     }

src/storage/shadow.ts

@@ -64,6 +64,11 @@ interface ResidentFields {
  *  session always verifies (cold-path full check), then every N-th. */
 export const VERIFY_EVERY_INGESTS = 20;
 
+/** #5744: wall-clock threshold for the slow-ingest warn (ms). The old
+ *  request-level warn keyed on message count, which cannot separate a fast
+ *  big session from a slow small one. */
+const SLOW_INGEST_WARN_MS = 500;
+
 /** #4323: consecutive sampled drifts required before escalating. A one-off
  *  drift is transient external churn (the watermark clear below heals it);
  *  N in a row proves the views are STRUCTURALLY divergent — the multi-writer
@@ -482,8 +487,15 @@ export class UnifiedShadow {
             // full path below; the delta itself re-verifies the boundary
             // inside the ingest transaction, so a racing writer degrades to
             // a full ingest, never a corrupt one.
+            // #5744: wall-clock guard on the ingest itself — message count is
+            // a bad proxy (one reporter session: 0.8s and 14.5s saves at the
+            // SAME ~3k msgs). Names the path so a spike is attributable
+            // (delta fast path vs full fallback) without a profiler.
+            const tIngestStart = performance.now();
+            let ingestPath = "delta";
             phase = "ingest-delta";
             if (!this.ingestDelta(legacy, hasRawTail)) {
+                ingestPath = "full";
                 phase = "plan-ingest";
                 const plan = planIngest(legacy);
                 phase = "ingest-session";
@@ -501,6 +513,14 @@ export class UnifiedShadow {
                 this.sync.ingestSession(plan, undefined, { completeLog });
             }
             this.stats.ingests++;
+            const ingestMs = Math.round(performance.now() - tIngestStart);
+            if (ingestMs > SLOW_INGEST_WARN_MS) {
+                // #5744: attach WAL forensics so a stall spike can be attributed
+                // to a checkpoint burst (multi-second tick inside the window)
+                // instead of per-row statement cost — read-only, warn path only.
+                const walSuffix = this.sync.walDiagnosticsSuffix() ?? "";
+                loggerLog("warn", `[storage] slow-ingest: ${legacy.id} path=${ingestPath} took=${ingestMs}ms${walSuffix} — synchronous per-save storage work is starving the event loop (#5744); path=full means the delta fast path fell back (watermark missing/cleared or tail shape changed)`);
+            }
             // #3133 sampled verification: ingest integrity is already guarded
             // by the SQLite write path; the read-back digest is a migration
             // parity oracle, so it runs on the first ingest of a session and