Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
95 changes: 73 additions & 22 deletions src/storage/shadow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -494,7 +494,21 @@ export class UnifiedShadow {
const tIngestStart = performance.now();
let ingestPath = "delta";
phase = "ingest-delta";
if (!this.ingestDelta(legacy, hasRawTail)) {
// #5677: while drift-latched the sampled digest has PROVEN the
// stored view diverges from the live one — vouching for a trusted
// prefix is exactly the trust that sample just voided. A host that
// re-writes history slots below the watermark on every save (the
// #4614 producer class) makes the delta path re-diverge within ≤
// VERIFY_EVERY_INGESTS saves of each heal, so the samples drift in
// a row, the streak escalates and the envelope freezes — the
// chronic ESCALATED→FROZEN loop reported in #5677. Forcing the
// exact full path for every latched save keeps both legs
// same-generation at sample time: the next sample passes, the
// latch releases, the delta resumes. Bounded cost (full ingest per
// save) paid only by sessions whose views are already proven
// divergent; healthy sessions never see it.
const deltaApplied = !this.driftedSessions.has(legacy.id) && this.ingestDelta(legacy, hasRawTail);
if (!deltaApplied) {
ingestPath = "full";
phase = "plan-ingest";
const plan = planIngest(legacy);
Expand Down Expand Up @@ -1177,20 +1191,26 @@ export class UnifiedShadow {
// #3133 watermark delta.
phase = "ingest-delta";
let applied = false;
try {
const mark = await backend.ingestWatermark(legacy.id);
if (mark) {
const boundary = await backend.probeIngestBoundary(legacy.id, mark.seq);
if (boundary) {
const plan = planIngestDelta(legacy, mark.seq, mark.tailSeqs, { msgId: boundary.msgId, contentHash: boundary.contentHash }, hasRawTail);
if (plan) {
applied = await backend.ingestSession(plan, { trustedSeq: mark.seq, boundaryMsgId: boundary.msgId, boundaryHash: boundary.contentHash, trustedTailSeqs: mark.tailSeqs });
if (applied) this.stats.deltaIngests++;
// #5677: same latch gate as the sync path above — a drift-
// latched session's stored prefix is proven untrustworthy by
// definition; the exact full path keeps both legs same-
// generation so the next sample can re-admit and release it.
if (!this.driftedSessions.has(legacy.id)) {
try {
const mark = await backend.ingestWatermark(legacy.id);
if (mark) {
const boundary = await backend.probeIngestBoundary(legacy.id, mark.seq);
if (boundary) {
const plan = planIngestDelta(legacy, mark.seq, mark.tailSeqs, { msgId: boundary.msgId, contentHash: boundary.contentHash }, hasRawTail);
if (plan) {
applied = await backend.ingestSession(plan, { trustedSeq: mark.seq, boundaryMsgId: boundary.msgId, boundaryHash: boundary.contentHash, trustedTailSeqs: mark.tailSeqs });
if (applied) this.stats.deltaIngests++;
}
}
}
} catch {
applied = false;
}
} catch {
applied = false;
}
// #5412: mirror the sync path's latched-sweep promotion (#5408
// B1 / #4323). While the sampler has latched this session as
Expand Down Expand Up @@ -1508,7 +1528,7 @@ interface DriftDiagnostics {
* converges the tables it owns; the sub-key names the actual residue. */
stateKeys?: string[];
/** First divergent lastMessages slot (or the length-mismatch index). */
tailSlot?: { index: number; expect: DriftSlotFingerprint; got: DriftSlotFingerprint };
tailSlot?: { index: number; expect: DriftSlotFingerprint; got: DriftSlotFingerprint; /** #5677: top-level field names whose values disagree (names only). */ fields?: string[] };
}

/** Compare the replay by component, retaining only message fingerprints across batches. */
Expand All @@ -1517,7 +1537,10 @@ class BatchedViewVerification {
private readonly tail: { tailSeqs: number[] | null; pluginSnapshotSeqs: number[] | null };
private readonly remainingContents: LegacySessionLike["blockContents"];
private readonly orphans: LegacySessionLike["blockContents"];
private readonly messages = new Map<number, ReturnType<typeof messageSignature>>();
// #5677: the parsed row rides along with its signature so the FIRST
// divergent slot can be field-diffed without re-reading the store; the
// verifier is per-sample and short-lived, so the extra retention is bounded.
private readonly messages = new Map<number, { sig: ReturnType<typeof messageSignature>; raw: unknown }>();
private readonly keys = new Set<string>();
private headerDiag: DriftDiagnostics | null = null;
private drift = false;
Expand Down Expand Up @@ -1568,7 +1591,10 @@ class BatchedViewVerification {
}

addMessages(rows: MessageRow[]): void {
for (const row of rows) this.messages.set(row.seq, messageSignature(JSON.parse(row.content)));
for (const row of rows) {
const raw = JSON.parse(row.content);
this.messages.set(row.seq, { sig: messageSignature(raw), raw });
}
}

private changed(key: string): void {
Expand All @@ -1585,16 +1611,19 @@ class BatchedViewVerification {
const expected = this.legacy[key];
let index = 0;
for (const seq of seqs ?? []) {
const got = this.messages.get(seq);
if (!got) {
const rec = this.messages.get(seq);
if (!rec) {
this.tailGaps++;
continue;
}
const want = expected && index < expected.length ? messageSignature(expected[index]) : undefined;
if (!want || want.digest !== got.digest) {
if (!want || want.digest !== rec.sig.digest) {
this.changed(key);
if (key === "lastMessages" && expected && !this.tailSlot) {
this.tailSlot = { index, expect: want?.slot ?? missingSlot(), got: got.slot };
const fields = diffFieldNames(expected[index], rec.raw);
this.tailSlot = fields === undefined
? { index, expect: want?.slot ?? missingSlot(), got: rec.sig.slot }
: { index, expect: want?.slot ?? missingSlot(), got: rec.sig.slot, fields };
}
}
index++;
Expand Down Expand Up @@ -1637,6 +1666,23 @@ function messageSignature(value: unknown): { digest: string; slot: DriftSlotFing
};
}

/** #5677: top-level field names whose canonicalized values disagree between
* two slot copies. NAMES ONLY — values never leave this function (the log
* line goes to disk and mirrors publicly). undefined when either side is not
* a plain object; for two plain objects a digest mismatch always names ≥1
* field, so an empty result is impossible there. */
function diffFieldNames(a: unknown, b: unknown): string[] | undefined {
if (!a || typeof a !== "object" || Array.isArray(a) || !b || typeof b !== "object" || Array.isArray(b)) return undefined;
const ao = a as Record<string, unknown>;
const bo = b as Record<string, unknown>;
const ser = (v: unknown): string | null => (v === undefined ? null : JSON.stringify(canonicalize(v)));
const out: string[] = [];
for (const k of [...new Set([...Object.keys(ao), ...Object.keys(bo)])].sort()) {
if (ser(ao[k]) !== ser(bo[k])) out.push(k);
}
return out.length > 0 ? out : undefined;
}

// #3919: which dimensions disagree when the sampled verify detects drift —
// the digest is a single hash and leaves classification to guesswork. Pure;
// runs only on detected drift (rare). Fingerprints carry id/role/length/
Expand Down Expand Up @@ -1686,7 +1732,12 @@ export function describeViewDivergence(expectView: unknown, gotView: unknown): D
const s2 = ser(m) ?? "";
return { id: typeof s.id === "string" ? s.id : "-", role: typeof s.role === "string" ? s.role : "-", length: s2.length, sha8: sha256(s2).slice(0, 8) };
};
diagnostics.tailSlot = { index, expect: fp(index < et.length ? et[index] : null), got: fp(index < gt.length ? gt[index] : null) };
const ea = index < et.length ? et[index] : null;
const ga = index < gt.length ? gt[index] : null;
const fields = diffFieldNames(ea, ga);
diagnostics.tailSlot = fields === undefined
? { index, expect: fp(ea), got: fp(ga) }
: { index, expect: fp(ea), got: fp(ga), fields };
}
}
return diagnostics;
Expand All @@ -1697,8 +1748,8 @@ export function formatDriftDiagnostics(d: DriftDiagnostics): string {
const parts = [`diverged: ${d.keys.length > 0 ? d.keys.join(",") : "<unknown>"}`];
if (d.stateKeys) parts.push(`state: ${d.stateKeys.join(",")}`);
if (d.tailSlot) {
const { index, expect: ex, got: go } = d.tailSlot;
parts.push(`tail slot ${index}: expect ${ex.id}/${ex.role} len=${ex.length} sha=${ex.sha8}, got ${go.id}/${go.role} len=${go.length} sha=${go.sha8}`);
const { index, expect: ex, got: go, fields } = d.tailSlot;
parts.push(`tail slot ${index}: expect ${ex.id}/${ex.role} len=${ex.length} sha=${ex.sha8}, got ${go.id}/${go.role} len=${go.length} sha=${go.sha8}${fields ? `; fields: ${fields.join(",")}` : ""}`);
}
return " [" + parts.join("; ") + "]";
}
Loading
Loading