[PR] Storage Layer Memory Governance: Five Fixes Consolidated into One (Streaming Fingerprint Saves / Batched Spot-Checks / B | 存储层内存治理：五个修复合成一个（保存指纹流式化 / 抽查分批 / 会话装载有界 / 增量车道只渲增量）

src/storage/backend.ts

@@ -33,6 +33,7 @@ export type {
     SessionSummaryRow,
     WebSummaryRow,
     IngestSessionInput,
+    SessionViewRows,
 } from "./store.js";
 import type {
     SessionRow,
@@ -44,6 +45,7 @@ import type {
     SessionSummaryRow,
     WebSummaryRow,
     IngestSessionInput,
+    SessionViewRows,
 } from "./store.js";
 
 /** Everything a storage backend must provide. Mirrors the operation surface
@@ -75,10 +77,15 @@ export interface StorageBackend {
     seqByRaw(sessionId: string): Promise<Map<string, number> | null>;
     /** #4316 PR3/3: fork adoption row clone (idempotent). */
     cloneSessionRows(parentId: string, childId: string, throughSeq: number, initial?: IngestSessionInput): Promise<boolean>;
-    getMessages(sessionId: string): Promise<(MessageRow & { contentHash: string })[]>;
-    getBlocks(sessionId: string): Promise<(BlockRow & { summaryHash: string; oneHash: string | null })[]>;
+    getMessages(sessionId: string, opts?: { includeRetired?: boolean; seqs?: readonly number[] }): Promise<(MessageRow & { contentHash: string })[]>;
+    getBlocks(sessionId: string, opts?: { limit: number; offset?: number }): Promise<(BlockRow & { summaryHash: string; oneHash: string | null })[]>;
+    /** #4739: the viewFor bundle from ONE consistent read snapshot (messages
+     *  fetched only at the tail/pluginSnapshot pointer seqs). */
+    sessionViewRows(sessionId: string): Promise<SessionViewRows>;
     getDeadRefs(sessionId: string): Promise<{ ord: number; ref: string; refNum: number; recordedAt: number }[]>;
     getRefs(sessionId: string): Promise<RefRow[]>;
+    /** All reads, including CAS hydration, use one short read-only snapshot. */
+    readSnapshot<T>(read: (reader: StorageReadView) => Promise<T>): Promise<T>;
     getCcr(sessionId: string): Promise<MessageContentStore | null>;
     getCcrEntries(sessionId: string): Promise<CcrEntryRow[]>;
 
@@ -141,6 +148,8 @@ export interface StorageBackend {
     init?(): Promise<void>;
 }
 
+export type StorageReadView = Pick<StorageBackend, "getSessionRow" | "getMessages" | "getBlocks" | "getDeadRefs" | "getRefs">;
+
 /** The sqlite backend: the existing UnifiedStore behind the async contract.
  *  Synchronous under the hood; `Promise.resolve` wrapping only. */
 export class SqliteBackend implements StorageBackend {
@@ -190,18 +199,30 @@ export class SqliteBackend implements StorageBackend {
     cloneSessionRows(parentId: string, childId: string, throughSeq: number, initial?: IngestSessionInput): Promise<boolean> {
         return Promise.resolve(this.store.cloneSessionRows(parentId, childId, throughSeq, initial));
     }
-    getMessages(sessionId: string): Promise<(MessageRow & { contentHash: string })[]> {
-        return Promise.resolve(this.store.getMessages(sessionId));
+    getMessages(sessionId: string, opts?: { includeRetired?: boolean; seqs?: readonly number[] }): Promise<(MessageRow & { contentHash: string })[]> {
+        return Promise.resolve(this.store.getMessages(sessionId, opts));
+    }
+    getBlocks(sessionId: string, opts?: { limit: number; offset?: number }): Promise<(BlockRow & { summaryHash: string; oneHash: string | null })[]> {
+        return Promise.resolve(this.store.getBlocks(sessionId, opts));
     }
-    getBlocks(sessionId: string): Promise<(BlockRow & { summaryHash: string; oneHash: string | null })[]> {
-        return Promise.resolve(this.store.getBlocks(sessionId));
+    sessionViewRows(sessionId: string): Promise<SessionViewRows> {
+        return Promise.resolve(this.store.sessionViewRows(sessionId));
     }
     getDeadRefs(sessionId: string): Promise<{ ord: number; ref: string; refNum: number; recordedAt: number }[]> {
         return Promise.resolve(this.store.getDeadRefs(sessionId));
     }
     getRefs(sessionId: string): Promise<RefRow[]> {
         return Promise.resolve(this.store.getRefs(sessionId));
     }
+    readSnapshot<T>(read: (reader: StorageReadView) => Promise<T>): Promise<T> {
+        return this.store.readSnapshotAsync((store) => read({
+            getSessionRow: async (id) => store.getSessionRow(id),
+            getMessages: async (id, opts) => store.getMessages(id, opts),
+            getBlocks: async (id, opts) => store.getBlocks(id, opts),
+            getDeadRefs: async (id) => store.getDeadRefs(id),
+            getRefs: async (id) => store.getRefs(id),
+        }));
+    }
     getCcr(sessionId: string): Promise<MessageContentStore | null> {
         return Promise.resolve(this.store.getCcr(sessionId));
     }

src/storage/canonical.ts

@@ -71,12 +71,108 @@ function canonicalizeWithContext(value: unknown, key: string | undefined): unkno
         }
         return out;
     }
+    return canonicalScalar(value, key);
+}
+
+function canonicalScalar(value: unknown, key: string | undefined): unknown {
     if (key === "biliVersion" && typeof value === "string") return "__VER__";
     if (typeof value === "number" && Number.isFinite(value) && value > 1_600_000_000_000) return "__TS__";
     if (typeof value === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value)) return "__UUID__";
     return value;
 }
 
+// Spec array index: "0" .. 2^32-2, no leading zeros.
+function isArrayIndexKey(k: string): boolean {
+    return /^(0|[1-9]\d*)$/.test(k) && Number(k) <= 4_294_967_294;
+}
+
 export function sha256(s: string): string {
     return createHash("sha256").update(s, "utf8").digest("hex");
 }
+
+/** Hash a storage view without materializing its canonical copy or JSON string. */
+export function canonicalDigest(value: unknown): string {
+    if (value === undefined || typeof value === "function" || typeof value === "symbol") return sha256(JSON.stringify(value));
+    const hash = createHash("sha256");
+    const chunkSize = 64 * 1024;
+    let pending = "";
+    function write(text: string): void {
+        pending += text;
+        if (pending.length >= chunkSize) {
+            hash.update(pending, "utf8");
+            pending = "";
+        }
+    }
+    function writeString(text: string): void {
+        write('"');
+        for (let start = 0; start < text.length;) {
+            let end = Math.min(start + chunkSize, text.length);
+            const last = text.charCodeAt(end - 1);
+            // Keep a surrogate pair together so JSON escaping and UTF-8 stay exact.
+            if (end < text.length && last >= 0xd800 && last <= 0xdbff) end--;
+            write(JSON.stringify(text.slice(start, end)).slice(1, -1));
+            start = end;
+        }
+        write('"');
+    }
+    function visit(current: unknown, key?: string): void {
+        if (Array.isArray(current)) {
+            write("[");
+            for (let i = 0; i < current.length; i++) {
+                if (i) write(",");
+                visit(current[i]);
+            }
+            write("]");
+        } else if (current instanceof Map) {
+            write('{"__map__":[');
+            let separator = "";
+            for (const [k, v] of current) {
+                write(separator + "[");
+                // canonicalize leaves Map keys untouched.
+                write(JSON.stringify(k) ?? "null");
+                write(",");
+                visit(v);
+                write("]");
+                separator = ",";
+            }
+            write("]}");
+        } else if (current instanceof Set) {
+            write('{"__set__":[');
+            let separator = "";
+            for (const v of current) {
+                write(separator);
+                visit(v);
+                separator = ",";
+            }
+            write("]}");
+        } else if (current && typeof current === "object") {
+            write("{");
+            let separator = "";
+            // Effective order of JSON.stringify(canonicalize(v)): array-index keys
+            // first, ascending numeric (spec enumeration overrides canonicalize's
+            // lexicographic sort), then the remaining keys lexicographic.
+            const keys = Object.keys(current);
+            const orderedKeys = keys.filter(isArrayIndexKey).sort((a, b) => Number(a) - Number(b))
+                .concat(keys.filter((k) => !isArrayIndexKey(k)).sort());
+            for (const k of orderedKeys) {
+                // Assigning __proto__ to canonicalize's plain output creates no JSON property.
+                if (Object.hasOwn(DIGEST_SKIPPED_KEYS, k) || k === "__proto__") continue;
+                const v = (current as Record<string, unknown>)[k];
+                if (v === undefined || typeof v === "function" || typeof v === "symbol") continue;
+                write(separator);
+                writeString(k);
+                write(":");
+                visit(v, k);
+                separator = ",";
+            }
+            write("}");
+        } else {
+            const normalized = canonicalScalar(current, key);
+            if (typeof normalized === "string") writeString(normalized);
+            else write(JSON.stringify(normalized) ?? "null");
+        }
+    }
+    visit(value);
+    hash.update(pending, "utf8");
+    return hash.digest("hex");
+}

src/storage/driver.ts

@@ -40,6 +40,8 @@ export interface SqliteDatabase {
     prepare(sql: string): SqliteStatement;
     /** Runs fn inside BEGIN/COMMIT; rolls back on throw (including nested calls via savepoints). */
     transaction<T>(fn: () => T): T;
+    /** Deferred snapshot: readers do not reserve the writer lock. */
+    readTransaction<T>(fn: () => T): T;
     close(): void;
     readonly open: boolean;
 }
@@ -87,6 +89,7 @@ function wrapBetterSqlite3(db: BetterSqlite3Database): SqliteDatabase {
         // cannot resolve. Taking the write lock up front makes concurrent
         // writers queue on busy_timeout instead (#2690 review blocking 5).
         transaction: (fn) => db.transaction(fn).immediate(),
+        readTransaction: (fn) => db.transaction(fn).deferred(),
         close: () => void db.close(),
         get open() {
             return db.open;
@@ -104,31 +107,33 @@ function openNodeSqlite(path: string): SqliteDatabase {
     };
     const db = new DatabaseSync(path);
     let depth = 0;
-    const begin = () => db.exec(depth === 0 ? "BEGIN IMMEDIATE" : `SAVEPOINT sp${depth}`); // IMMEDIATE at top level: multi-process SQLITE_BUSY_SNAPSHOT guard (see openBetterSqlite3)
+    const begin = (readOnly: boolean) => db.exec(depth === 0 ? (readOnly ? "BEGIN DEFERRED" : "BEGIN IMMEDIATE") : `SAVEPOINT sp${depth}`);
     const commit = () => db.exec(depth === 1 ? "COMMIT" : `RELEASE sp${depth - 1}`);
     const rollback = () => db.exec(depth === 1 ? "ROLLBACK" : `ROLLBACK TO sp${depth - 1}`);
+    const transaction = <T>(fn: () => T, readOnly: boolean): T => {
+        if (db.getAutoCommit?.() === false && depth === 0) {
+            // Already inside an outer transaction driven directly by
+            // DatabaseSync — just run inline; the outer COMMIT covers us.
+            return fn();
+        }
+        begin(readOnly);
+        depth++;
+        try {
+            const out = fn();
+            commit();
+            return out;
+        } catch (err) {
+            rollback();
+            throw err;
+        } finally {
+            depth--;
+        }
+    };
     return {
         exec: (sql) => void db.exec(sql),
         prepare: (sql) => adaptNodeSqliteStatement(db.prepare(sql)),
-        transaction: <T>(fn: () => T): T => {
-            if (db.getAutoCommit?.() === false && depth === 0) {
-                // Already inside an outer transaction driven directly by
-                // DatabaseSync — just run inline; the outer COMMIT covers us.
-                return fn();
-            }
-            begin();
-            depth++;
-            try {
-                const out = fn();
-                commit();
-                return out;
-            } catch (err) {
-                rollback();
-                throw err;
-            } finally {
-                depth--;
-            }
-        },
+        transaction: (fn) => transaction(fn, false),
+        readTransaction: (fn) => transaction(fn, true),
         close: () => void db.close(),
         get open() {
             return db.open;