From 06341333e295f873ed5162abf1d893b22df15600 Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Thu, 13 Aug 2026 16:15:48 +0200 Subject: [PATCH 1/7] MILAB-6705: snapshot codec and restore path for the tree mirror Adds a length-prefixed binary format for a tree mirror, with capture and restore around it. Resource bodies store every reference as a global id and keep signatures in a side table the decoder rejoins, so the corpus outlives the signatures it was written with. The session witness (the root's signature bytes) sits outside the compressed payload, so a rotated session is detected without inflating the file. Restore goes through updateFromResourceData into a throwaway tree, which is what keeps a failed restore from invalidating a working one. A torn, foreign or corrupt file reports a reason instead of throwing, and PlTreeState grows an isValid getter so capturing an invalidated tree is refused rather than silently written. --- .changeset/disk-persist-tree-codec.md | 5 + lib/node/pl-tree/src/index.ts | 1 + lib/node/pl-tree/src/persisted_tree.test.ts | 281 ++++++++++ lib/node/pl-tree/src/persisted_tree.ts | 592 ++++++++++++++++++++ lib/node/pl-tree/src/state.ts | 8 + 5 files changed, 887 insertions(+) create mode 100644 .changeset/disk-persist-tree-codec.md create mode 100644 lib/node/pl-tree/src/persisted_tree.test.ts create mode 100644 lib/node/pl-tree/src/persisted_tree.ts diff --git a/.changeset/disk-persist-tree-codec.md b/.changeset/disk-persist-tree-codec.md new file mode 100644 index 0000000000..3722e81f3b --- /dev/null +++ b/.changeset/disk-persist-tree-codec.md @@ -0,0 +1,5 @@ +--- +"@milaboratories/pl-tree": minor +--- + +Snapshot codec for persisting a tree mirror to disk, plus capture and restore. Bodies are stored against global ids with signatures in a side table, so a snapshot survives the signatures it was taken with. diff --git a/lib/node/pl-tree/src/index.ts b/lib/node/pl-tree/src/index.ts index 714dea0929..edc511bef1 100644 --- a/lib/node/pl-tree/src/index.ts +++ b/lib/node/pl-tree/src/index.ts @@ -3,6 +3,7 @@ export * from "./state"; export * from "./sync"; export * from "./accessors"; export * from "./snapshot"; +export * from "./persisted_tree"; export * from "./synchronized_tree"; export * from "./value_and_error"; export * from "./value_or_error"; diff --git a/lib/node/pl-tree/src/persisted_tree.test.ts b/lib/node/pl-tree/src/persisted_tree.test.ts new file mode 100644 index 0000000000..b8bbe711fe --- /dev/null +++ b/lib/node/pl-tree/src/persisted_tree.test.ts @@ -0,0 +1,281 @@ +import { describe, expect, test } from "vitest"; +import type { FinalResourceDataPredicate } from "@milaboratories/pl-client"; +import { createSignedResourceId, toResourceSignature } from "@milaboratories/pl-client"; +import type { ExtendedResourceData } from "./state"; +import { PlTreeState } from "./state"; +import { constructTreeLoadingRequest } from "./sync"; +import { + captureTreeState, + decodePersistedTree, + encodePersistedTree, + PERSISTED_TREE_SCHEMA_VERSION, + readPersistedTreeHeader, + restoreTreeState, +} from "./persisted_tree"; +import { + dField, + iField, + TestDynamicRootState1, + TestStructuralResourceState1, + TestValueResourceState1, +} from "./test_utils"; + +const sig = (hex: string) => toResourceSignature(Buffer.from(hex, "hex")); + +/** Ids here carry real signature bytes, unlike the shared test fixtures, so the + * global-id / signature split is actually exercised rather than trivially satisfied. */ +const rid = (id: bigint, signature = "a1b2c3d4") => createSignedResourceId(id, sig(signature)); + +const RootSignature = sig("deadbeef"); +const RootId = createSignedResourceId(1000001n, RootSignature); + +/** The shared fixtures use resource types `DefaultFinalResourceDataPredicate` does not know, + * so it settles nothing and the tree has no final/non-final split to test. Trusting the + * backend's derived flag instead gives one, which is what the loading request is built from. */ +const finalByFlag: FinalResourceDataPredicate = (r) => r.final; + +/** A tree with a settled branch, an unsettled branch, data, kv, a dynamic field pointing at + * a value, and an unresolved field. Enough shape that a codec losing a distinction the tree + * cares about shows up as a different loading request. */ +function buildPopulatedTree(): PlTreeState { + const tree = new PlTreeState(RootId, finalByFlag); + tree.updateFromResourceData([ + { + ...TestDynamicRootState1, + id: RootId, + fields: [dField("settled", rid(10n)), dField("running", rid(20n)), dField("pending")], + }, + { + ...TestStructuralResourceState1, + id: rid(10n), + inputsLocked: true, + outputsLocked: true, + resourceReady: true, + final: true, + fields: [iField("payload", rid(11n))], + kv: [ + { key: "meta", value: Buffer.from('{"n":1}') }, + { key: "binary", value: Uint8Array.from([0, 1, 2, 255]) }, + ], + }, + { + ...TestValueResourceState1, + id: rid(11n), + data: Buffer.from("settled payload"), + }, + { + ...TestStructuralResourceState1, + id: rid(20n), + fields: [iField("payload"), dField("progress", rid(21n))], + }, + { + ...TestValueResourceState1, + id: rid(21n), + data: Buffer.from("in progress"), + }, + ]); + return tree; +} + +/** The claim the whole design rests on: what comes back addresses the backend the same way + * the original did. Compared as sorted arrays because neither the seed order nor the skip + * set's iteration order is part of the contract. */ +function loadingRequestOf(tree: PlTreeState) { + const req = constructTreeLoadingRequest(tree); + return { + seeds: [...req.seedResources].sort(), + skips: [...req.finalResources].sort(), + }; +} + +async function roundTrip(tree: PlTreeState, compress?: boolean): Promise { + const captured = captureTreeState(tree, RootSignature); + const bytes = await encodePersistedTree(captured, { compress }); + + const decoded = await decodePersistedTree(bytes); + expect(decoded.ok).toBe(true); + if (!decoded.ok) throw new Error("unreachable"); + + const restored = restoreTreeState(decoded.value, finalByFlag); + expect(restored).toBeDefined(); + return restored!; +} + +describe("the contract", () => { + test.for([true, false])( + "restored tree builds the same loading request (compress: %s)", + async (compress) => { + const original = buildPopulatedTree(); + const restored = await roundTrip(original, compress); + + const expected = loadingRequestOf(original); + // Guards against a vacuous pass: an empty tree would match trivially. + expect(expected.seeds.length).toBeGreaterThan(0); + expect(expected.skips.length).toBeGreaterThan(0); + + expect(loadingRequestOf(restored)).toStrictEqual(expected); + }, + ); + + test("restored tree holds the same resource states", async () => { + const original = buildPopulatedTree(); + const restored = await roundTrip(original); + + // Byte payloads are compared as plain arrays: the decoder hands back Uint8Array views + // while the fixtures were built from Buffers, and strict equality would read that + // prototype difference as a difference in state. + const asBytes = (b?: Uint8Array) => (b === undefined ? undefined : [...b]); + const normalize = (r: ExtendedResourceData) => ({ + ...r, + data: asBytes(r.data), + kv: r.kv.map((e) => ({ key: e.key, value: asBytes(e.value) })), + }); + const states = (tree: PlTreeState) => + tree + .dumpState() + .map(normalize) + .sort((a, b) => (a.id < b.id ? -1 : a.id > b.id ? 1 : 0)); + + expect(states(restored)).toStrictEqual(states(original)); + }); + + test("roots survive the round trip", async () => { + const original = buildPopulatedTree(); + const restored = await roundTrip(original); + expect([...restored.roots]).toStrictEqual([...original.roots]); + }); + + test("finality is recomputed, not read from the file", async () => { + const original = buildPopulatedTree(); + const captured = captureTreeState(original, RootSignature); + const decoded = await decodePersistedTree(await encodePersistedTree(captured)); + expect(decoded.ok).toBe(true); + if (!decoded.ok) throw new Error("unreachable"); + + // A predicate that settles nothing must yield a tree that skips nothing, even though + // the file says otherwise. The file's finality is an artefact of the predicate in force + // when it was written, and the two are allowed to disagree. + const restored = restoreTreeState(decoded.value, () => false); + expect(restored).toBeDefined(); + expect(constructTreeLoadingRequest(restored!).finalResources.size).toBe(0); + expect(constructTreeLoadingRequest(original).finalResources.size).toBeGreaterThan(0); + }); +}); + +describe("the witness", () => { + test("is readable without inflating the payload", async () => { + const captured = captureTreeState(buildPopulatedTree(), RootSignature); + const bytes = await encodePersistedTree(captured); + + const header = readPersistedTreeHeader(bytes); + expect(header.ok).toBe(true); + if (!header.ok) throw new Error("unreachable"); + + expect(header.value.schemaVersion).toBe(PERSISTED_TREE_SCHEMA_VERSION); + expect(Buffer.from(header.value.witness).equals(Buffer.from(RootSignature))).toBe(true); + }); +}); + +describe("a snapshot that cannot be read", () => { + const encoded = async () => + await encodePersistedTree(captureTreeState(buildPopulatedTree(), RootSignature)); + + test("a foreign file is not a snapshot", async () => { + const result = await decodePersistedTree(Buffer.from("this is not a tree snapshot at all")); + expect(result).toStrictEqual({ ok: false, reason: "not-a-snapshot" }); + }); + + test("an empty file is not a snapshot", async () => { + expect(await decodePersistedTree(Buffer.alloc(0))).toStrictEqual({ + ok: false, + reason: "not-a-snapshot", + }); + }); + + test("a truncated file is rejected rather than replayed", async () => { + const bytes = await encoded(); + const result = await decodePersistedTree(bytes.subarray(0, bytes.length - 32)); + expect(result.ok).toBe(false); + if (result.ok) throw new Error("unreachable"); + expect(["truncated", "checksum"]).toContain(result.reason); + }); + + test("a damaged payload fails its checksum", async () => { + const bytes = Buffer.from(await encoded()); + // Flip a bit inside the payload, leaving the header and the length trailer intact. + const target = Math.floor(bytes.length / 2); + bytes[target] = bytes[target] ^ 0xff; + + expect(await decodePersistedTree(bytes)).toStrictEqual({ ok: false, reason: "checksum" }); + }); + + test("an unknown schema version loads as absent", async () => { + const bytes = Buffer.from(await encoded()); + bytes.writeUInt16LE(PERSISTED_TREE_SCHEMA_VERSION + 1, 4); + + expect(await decodePersistedTree(bytes)).toStrictEqual({ ok: false, reason: "unknown-schema" }); + expect(readPersistedTreeHeader(bytes)).toStrictEqual({ ok: false, reason: "unknown-schema" }); + }); + + test("every truncation point is a clean failure, never a throw", async () => { + const bytes = await encoded(); + for (let length = 0; length < bytes.length; length++) { + const result = await decodePersistedTree(bytes.subarray(0, length)); + expect(result.ok).toBe(false); + } + }); +}); + +describe("a snapshot that decodes but cannot be applied", () => { + test("a dangling reference leaves no tree and does not throw", async () => { + const tree = buildPopulatedTree(); + const captured = captureTreeState(tree, RootSignature); + + // Drop a resource that others still point at. The codec has no opinion on this; the + // state-update call is what refuses it, which is the point of reusing that path. + const dangling = { + ...captured, + resources: captured.resources.filter((r) => r.id !== rid(11n)), + }; + const decoded = await decodePersistedTree(await encodePersistedTree(dangling)); + expect(decoded.ok).toBe(true); + if (!decoded.ok) throw new Error("unreachable"); + + expect(restoreTreeState(decoded.value, finalByFlag)).toBeUndefined(); + }); + + test("a live tree survives a failed restore", async () => { + const live = buildPopulatedTree(); + const before = loadingRequestOf(live); + + const captured = captureTreeState(live, RootSignature); + const decoded = await decodePersistedTree( + await encodePersistedTree({ + ...captured, + resources: captured.resources.filter((r) => r.id !== rid(11n)), + }), + ); + if (!decoded.ok) throw new Error("unreachable"); + expect(restoreTreeState(decoded.value, finalByFlag)).toBeUndefined(); + + // The throwaway tree absorbed the invalidation, so the working tree is untouched. + expect(live.isValid).toBe(true); + expect(loadingRequestOf(live)).toStrictEqual(before); + }); +}); + +describe("capture", () => { + test("refuses an invalidated tree", () => { + const tree = buildPopulatedTree(); + tree.invalidateTree("test"); + expect(() => captureTreeState(tree, RootSignature)).toThrow(/invalidated/); + }); + + test("an empty tree round trips", async () => { + const empty = new PlTreeState(RootId, finalByFlag); + const restored = await roundTrip(empty); + expect(loadingRequestOf(restored)).toStrictEqual(loadingRequestOf(empty)); + // An unmaterialized root is still seeded, which is how a cold tree starts. + expect(loadingRequestOf(restored).seeds).toStrictEqual([RootId]); + }); +}); diff --git a/lib/node/pl-tree/src/persisted_tree.ts b/lib/node/pl-tree/src/persisted_tree.ts new file mode 100644 index 0000000000..eec799b398 --- /dev/null +++ b/lib/node/pl-tree/src/persisted_tree.ts @@ -0,0 +1,592 @@ +import type { + FieldData, + FieldStatus, + FieldType, + FinalResourceDataPredicate, + KeyValue, + OptionalSignedResourceId, + ResourceKind, + ResourceSignature, + SignedResourceId, +} from "@milaboratories/pl-client"; +import { + createSignedResourceId, + isNotNullSignedResourceId, + NullSignedResourceId, + parseSignedResourceId, + toResourceSignature, +} from "@milaboratories/pl-client"; +import type { MiLogger } from "@milaboratories/ts-helpers"; +import { deflate, inflate } from "node:zlib"; +import { promisify } from "node:util"; +import type { ExtendedResourceData } from "./state"; +import { PlTreeState } from "./state"; + +const deflateAsync = promisify(deflate); +const inflateAsync = promisify(inflate); + +/** "PLTS", little-endian. Distinguishes our file from anything else that lands in the + * snapshot directory, so a foreign file is rejected instead of parsed as garbage. */ +const MAGIC = 0x53544c50; + +/** Bumped whenever the byte layout below changes in a way an older decoder would + * misread. Only an exact match is accepted: the decoder's job here is to recognise a + * format it cannot read, not to migrate it. Invalidation on rule changes is the cache + * key's job (the middle layer's build stamp), not this number's. */ +export const PERSISTED_TREE_SCHEMA_VERSION = 1; + +/** Payload is deflated. Absent means the payload is stored as-is, which is what a + * periodic write falls back to if compression CPU ever becomes a problem. */ +const FLAG_COMPRESSED = 1 << 0; + +const HEADER_FIXED_BYTES = 4 /* magic */ + 2 /* schema */ + 2 /* flags */ + 2 /* witness len */; +const TRAILER_BYTES = 4 /* payload length */ + 4 /* checksum */; + +/** Enum orderings are part of the on-disk format: append only, never reorder. An index + * the decoder does not know is malformed input, which is why decoding is bounds-checked + * rather than cast. */ +const KINDS: readonly ResourceKind[] = ["Structural", "Value"]; +const FIELD_TYPES: readonly FieldType[] = ["Input", "Output", "Service", "OTW", "Dynamic", "MTW"]; +const FIELD_STATUSES: readonly FieldStatus[] = ["Empty", "Assigned", "Resolved"]; + +const RES_HAS_DATA = 1 << 0; +const RES_INPUTS_LOCKED = 1 << 1; +const RES_OUTPUTS_LOCKED = 1 << 2; +const RES_READY = 1 << 3; +const RES_FINAL = 1 << 4; + +/** Global id 0 stands for "no reference". A real resource can never have it: + * `createSignedResourceId` rejects the null id, so the sentinel is unambiguous. */ +const NO_REFERENCE = 0n; + +/** + * A tree mirror as it sits on disk. + * + * Every reference inside {@link resources} is stored as a global id, with signatures held + * apart in a side table that {@link decodePersistedTree} rejoins. That split is what lets a + * snapshot outlive the signatures it was taken with: the bodies stay valid indefinitely, so + * a future signature refresh can replace the table and reuse the same corpus. + */ +export type PersistedTree = { + /** Session witness: the signature bytes of the tree's root at write time. + * A resource signature is an HMAC over the id, the session and the colour, so byte + * equality against a freshly resolved root signature means every other signature in the + * table still addresses something. Inequality means they are all dead. */ + readonly witness: ResourceSignature; + readonly roots: readonly SignedResourceId[]; + readonly resources: readonly ExtendedResourceData[]; +}; + +/** The part of a snapshot readable without inflating the payload. Kept outside the + * compressed section on purpose: a rotated session must be detectable without paying to + * decompress ten megabytes that are about to be discarded. */ +export type PersistedTreeHeader = { + readonly schemaVersion: number; + readonly witness: ResourceSignature; +}; + +/** Why a snapshot could not be read. Carried rather than thrown, because every one of + * these means "open cold" and the caller wants to count which happened. */ +export type PersistedTreeReadFailure = + /** Not our file at all: wrong magic, or too short to hold a header. */ + | "not-a-snapshot" + /** Written by a different schema version. */ + | "unknown-schema" + /** File ends early: the length trailer disagrees with the actual size. */ + | "truncated" + /** Checksum mismatch: the bytes are ours but damaged. */ + | "checksum" + /** Structurally decodable but internally nonsensical (bad enum index, local id, ...). */ + | "malformed"; + +export type PersistedTreeReadResult = + | { readonly ok: true; readonly value: T } + | { readonly ok: false; readonly reason: PersistedTreeReadFailure }; + +export type EncodePersistedTreeOps = { + /** Defaults to true. Turning it off trades roughly a factor of three in file size for + * the compression CPU, which is the documented escape hatch for periodic writes. */ + readonly compress?: boolean; +}; + +// +// Capture and restore +// + +/** + * Captures a live tree's state for persistence. `witness` is the signature of the root as + * currently held, which is what a later open compares against to decide whether the + * signatures in this snapshot are still live. + * + * Throws on an invalidated tree. {@link PlTreeState.dumpState} would happily read through + * one, and a terminated or inconsistent tree is exactly what must not reach disk, so the + * caller has to capture before it tears the tree down. + */ +export function captureTreeState(state: PlTreeState, witness: ResourceSignature): PersistedTree { + if (!state.isValid) throw new Error("refusing to capture an invalidated tree"); + return { witness, roots: [...state.roots], resources: state.dumpState() }; +} + +/** + * Rebuilds a tree from a snapshot, or returns undefined if the snapshot cannot be applied. + * + * The snapshot goes through {@link PlTreeState.updateFromResourceData}, the same call the + * live loading path uses, so every invariant that path enforces is enforced here and cannot + * drift from it. Reference counts are applied after the whole batch, so the order resources + * appear in the file does not matter, and finality is recomputed from `finalPredicate` + * rather than read from the file. + * + * The tree is built fresh and thrown away on failure: that call invalidates the tree it is + * given when it finds an inconsistency, so restoring into a live tree would destroy a + * working one instead of falling back to a cold open. + */ +export function restoreTreeState( + snapshot: PersistedTree, + finalPredicate: FinalResourceDataPredicate, + ops: { roots?: Set; logger?: MiLogger } = {}, +): PlTreeState | undefined { + const roots = ops.roots ?? new Set(snapshot.roots); + const restored = new PlTreeState(roots, finalPredicate); + try { + // allowOrphanInputs mirrors the live path. The check that matters for a snapshot is the + // orphan-reference one, which runs either way and catches a corpus referencing an id it + // does not carry. + restored.updateFromResourceData([...snapshot.resources], { allowOrphanInputs: true }); + return restored; + } catch (e: unknown) { + ops.logger?.warn( + `tree snapshot could not be restored, opening cold: ${e instanceof Error ? e.message : String(e)}`, + ); + return undefined; + } +} + +// +// Encoding +// + +/** Serializes a tree mirror to the on-disk format. */ +export async function encodePersistedTree( + tree: PersistedTree, + ops: EncodePersistedTreeOps = {}, +): Promise { + const compress = ops.compress ?? true; + + const payload = writePayload(tree); + const stored = compress ? await deflateAsync(payload) : payload; + + const header = new Writer(HEADER_FIXED_BYTES + tree.witness.length); + header.u32(MAGIC); + header.u16(PERSISTED_TREE_SCHEMA_VERSION); + header.u16(compress ? FLAG_COMPRESSED : 0); + header.shortBytes(tree.witness); + + const trailer = new Writer(TRAILER_BYTES); + trailer.u32(stored.length); + trailer.u32(crc32(stored)); + + return Buffer.concat([header.result(), stored, trailer.result()]); +} + +function writePayload(tree: PersistedTree): Buffer { + // Signatures are collected from every id mentioned anywhere, not just from resource + // bodies. originalResourceId is not refcounted by the tree, so a duplicate's original + // can be referenced without being held, and its signature would otherwise be lost. + const signatures = new Map(); + const collect = (id: OptionalSignedResourceId) => { + if (!isNotNullSignedResourceId(id)) return; + const { globalId, signature } = parseSignedResourceId(id); + signatures.set(globalId, signature); + }; + + for (const root of tree.roots) collect(root); + for (const res of tree.resources) { + collect(res.id); + collect(res.originalResourceId); + collect(res.error); + for (const f of res.fields) { + collect(f.value); + collect(f.error); + } + } + + const w = new Writer(); + + w.u32(tree.roots.length); + for (const root of tree.roots) w.u64(globalIdOf(root)); + + w.u32(signatures.size); + for (const [globalId, signature] of signatures) { + w.u64(globalId); + w.shortBytes(signature); + } + + w.u32(tree.resources.length); + for (const res of tree.resources) writeResource(w, res); + + return w.result(); +} + +function writeResource(w: Writer, res: ExtendedResourceData) { + w.u64(globalIdOf(res.id)); + w.u64(optionalGlobalIdOf(res.originalResourceId)); + w.u64(optionalGlobalIdOf(res.error)); + + w.u8(indexOfOrThrow(KINDS, res.kind, "resource kind")); + w.str(res.type.name); + w.str(res.type.version); + + w.u8( + (res.data !== undefined ? RES_HAS_DATA : 0) | + (res.inputsLocked ? RES_INPUTS_LOCKED : 0) | + (res.outputsLocked ? RES_OUTPUTS_LOCKED : 0) | + (res.resourceReady ? RES_READY : 0) | + (res.final ? RES_FINAL : 0), + ); + if (res.data !== undefined) w.bytes(res.data); + + w.u32(res.fields.length); + for (const f of res.fields) { + w.str(f.name); + w.u8(indexOfOrThrow(FIELD_TYPES, f.type, "field type")); + w.u8(indexOfOrThrow(FIELD_STATUSES, f.status, "field status")); + w.u64(optionalGlobalIdOf(f.value)); + w.u64(optionalGlobalIdOf(f.error)); + w.u8(f.valueIsFinal ? 1 : 0); + } + + w.u32(res.kv.length); + for (const kv of res.kv) { + w.str(kv.key); + w.bytes(kv.value); + } +} + +// +// Decoding +// + +/** Reads magic, schema version and witness without touching the payload. Cheap enough to + * run on every open, which is what makes a rotated-session miss cheap. */ +export function readPersistedTreeHeader( + bytes: Uint8Array, +): PersistedTreeReadResult { + try { + if (bytes.length < HEADER_FIXED_BYTES + TRAILER_BYTES) return failure("not-a-snapshot"); + + const r = new Reader(bytes); + if (r.u32() !== MAGIC) return failure("not-a-snapshot"); + + const schemaVersion = r.u16(); + r.u16(); // flags, only meaningful to the full decode + const witness = toResourceSignature(r.shortBytes()); + + if (schemaVersion !== PERSISTED_TREE_SCHEMA_VERSION) return failure("unknown-schema"); + + return { ok: true, value: { schemaVersion, witness } }; + } catch { + // Any bounds violation while reading a fixed-size header means the file is not one. + return failure("not-a-snapshot"); + } +} + +/** Reads a whole snapshot. Never throws: a torn, corrupt, foreign or unreadable file is + * reported as a failure reason, so the caller opens cold instead of replaying garbage. */ +export async function decodePersistedTree( + bytes: Uint8Array, +): Promise> { + const header = readPersistedTreeHeader(bytes); + if (!header.ok) return header; + + let payload: Uint8Array; + try { + const r = new Reader(bytes); + r.skip(4 + 2); // magic, schema + const flags = r.u16(); + r.shortBytes(); // witness, already read + const payloadStart = r.position; + + // The trailer's length is what says where the payload ends. Deriving it from the file + // size instead would accept a file with trailing garbage as intact. + const trailer = new Reader(bytes); + trailer.skip(bytes.length - TRAILER_BYTES); + const payloadLength = trailer.u32(); + const checksum = trailer.u32(); + + if (payloadStart + payloadLength !== bytes.length - TRAILER_BYTES) return failure("truncated"); + + const stored = bytes.subarray(payloadStart, payloadStart + payloadLength); + if (crc32(stored) !== checksum) return failure("checksum"); + + payload = (flags & FLAG_COMPRESSED) !== 0 ? await inflateAsync(stored) : stored; + } catch { + // Includes inflate failures: a payload that passes its checksum but will not + // decompress is damaged in a way we cannot distinguish from corruption. + return failure("checksum"); + } + + try { + return { ok: true, value: readPayload(payload, header.value.witness) }; + } catch { + return failure("malformed"); + } +} + +function readPayload(payload: Uint8Array, witness: ResourceSignature): PersistedTree { + const r = new Reader(payload); + + const rootCount = r.u32(); + const rootIds: bigint[] = []; + for (let i = 0; i < rootCount; i++) rootIds.push(r.u64()); + + const signatureCount = r.u32(); + const signatures = new Map(); + for (let i = 0; i < signatureCount; i++) { + const globalId = r.u64(); + signatures.set(globalId, toResourceSignature(r.shortBytes())); + } + + /** Rejoins a stored global id with its signature. A reference with no table entry is + * malformed rather than recoverable: an unsigned id addresses nothing. */ + const signed = (globalId: bigint): SignedResourceId => { + const signature = signatures.get(globalId); + if (signature === undefined) throw new Error(`no signature stored for global id ${globalId}`); + return createSignedResourceId(globalId, signature); + }; + const optionalSigned = (globalId: bigint): OptionalSignedResourceId => + globalId === NO_REFERENCE ? NullSignedResourceId : signed(globalId); + + const roots = rootIds.map(signed); + + const resourceCount = r.u32(); + const resources: ExtendedResourceData[] = []; + for (let i = 0; i < resourceCount; i++) resources.push(readResource(r, signed, optionalSigned)); + + if (!r.atEnd) throw new Error("trailing bytes in snapshot payload"); + + return { witness, roots, resources }; +} + +function readResource( + r: Reader, + signed: (globalId: bigint) => SignedResourceId, + optionalSigned: (globalId: bigint) => OptionalSignedResourceId, +): ExtendedResourceData { + const id = signed(r.u64()); + const originalResourceId = optionalSigned(r.u64()); + const error = optionalSigned(r.u64()); + + const kind = atOrThrow(KINDS, r.u8(), "resource kind"); + const type = { name: r.str(), version: r.str() }; + + const flags = r.u8(); + const data = (flags & RES_HAS_DATA) !== 0 ? r.bytes() : undefined; + + const fieldCount = r.u32(); + const fields: FieldData[] = []; + for (let i = 0; i < fieldCount; i++) { + const name = r.str(); + const fieldType = atOrThrow(FIELD_TYPES, r.u8(), "field type"); + const status = atOrThrow(FIELD_STATUSES, r.u8(), "field status"); + const value = optionalSigned(r.u64()); + const fieldError = optionalSigned(r.u64()); + const valueIsFinal = r.u8() !== 0; + fields.push({ name, type: fieldType, status, value, error: fieldError, valueIsFinal }); + } + + const kvCount = r.u32(); + const kv: KeyValue[] = []; + for (let i = 0; i < kvCount; i++) kv.push({ key: r.str(), value: r.bytes() }); + + return { + id, + originalResourceId, + error, + kind, + type, + data, + inputsLocked: (flags & RES_INPUTS_LOCKED) !== 0, + outputsLocked: (flags & RES_OUTPUTS_LOCKED) !== 0, + resourceReady: (flags & RES_READY) !== 0, + final: (flags & RES_FINAL) !== 0, + fields, + kv, + }; +} + +// +// Helpers +// + +function failure(reason: PersistedTreeReadFailure): { + ok: false; + reason: PersistedTreeReadFailure; +} { + return { ok: false, reason }; +} + +function globalIdOf(id: SignedResourceId): bigint { + return parseSignedResourceId(id).globalId; +} + +function optionalGlobalIdOf(id: OptionalSignedResourceId): bigint { + return isNotNullSignedResourceId(id) ? globalIdOf(id) : NO_REFERENCE; +} + +function indexOfOrThrow(values: readonly T[], value: T, what: string): number { + const idx = values.indexOf(value); + if (idx < 0) throw new Error(`unknown ${what}: ${String(value)}`); + return idx; +} + +function atOrThrow(values: readonly T[], idx: number, what: string): T { + if (idx < 0 || idx >= values.length) throw new Error(`unknown ${what} index: ${idx}`); + return values[idx]; +} + +/** Table-driven CRC-32 (IEEE). Node's `zlib.crc32` would do, but it landed in 22.2 and + * the repo's floor is 22, so this keeps the format readable on every supported runtime + * without adding a dependency. */ +const CRC_TABLE = (() => { + const table = new Int32Array(256); + for (let i = 0; i < 256; i++) { + let c = i; + for (let bit = 0; bit < 8; bit++) c = (c & 1) !== 0 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; + table[i] = c; + } + return table; +})(); + +function crc32(bytes: Uint8Array): number { + let c = 0xffffffff; + for (let i = 0; i < bytes.length; i++) c = CRC_TABLE[(c ^ bytes[i]) & 0xff] ^ (c >>> 8); + return (c ^ 0xffffffff) >>> 0; +} + +/** Growable little-endian writer. Doubling keeps a ten megabyte tree to a handful of + * reallocations. */ +class Writer { + private buf: Buffer; + private pos = 0; + + constructor(initialBytes = 1 << 16) { + this.buf = Buffer.allocUnsafe(Math.max(initialBytes, 16)); + } + + private ensure(extra: number) { + if (this.pos + extra <= this.buf.length) return; + let size = this.buf.length; + while (size < this.pos + extra) size *= 2; + const next = Buffer.allocUnsafe(size); + this.buf.copy(next, 0, 0, this.pos); + this.buf = next; + } + + u8(v: number) { + this.ensure(1); + this.pos = this.buf.writeUInt8(v, this.pos); + } + + u16(v: number) { + this.ensure(2); + this.pos = this.buf.writeUInt16LE(v, this.pos); + } + + u32(v: number) { + this.ensure(4); + this.pos = this.buf.writeUInt32LE(v, this.pos); + } + + u64(v: bigint) { + this.ensure(8); + this.pos = this.buf.writeBigUInt64LE(v, this.pos); + } + + /** u32-prefixed. For resource data and kv values, which have no small bound. */ + bytes(b: Uint8Array) { + this.u32(b.length); + this.ensure(b.length); + this.buf.set(b, this.pos); + this.pos += b.length; + } + + /** u16-prefixed. For signatures, which are a hash and cannot approach 64 KB. */ + shortBytes(b: Uint8Array) { + if (b.length > 0xffff) throw new Error(`value too long for a short field: ${b.length}`); + this.u16(b.length); + this.ensure(b.length); + this.buf.set(b, this.pos); + this.pos += b.length; + } + + str(s: string) { + this.bytes(Buffer.from(s, "utf8")); + } + + result(): Buffer { + return this.buf.subarray(0, this.pos); + } +} + +/** Little-endian reader. Every accessor bounds-checks, so a truncated payload throws + * rather than reading past its end; callers turn that into a failure reason. */ +class Reader { + private pos = 0; + private readonly view: DataView; + + constructor(private readonly src: Uint8Array) { + this.view = new DataView(src.buffer, src.byteOffset, src.byteLength); + } + + get position(): number { + return this.pos; + } + + get atEnd(): boolean { + return this.pos === this.src.length; + } + + private take(n: number): number { + if (n < 0 || this.pos + n > this.src.length) + throw new Error(`read past end of snapshot: need ${n} at ${this.pos}`); + const at = this.pos; + this.pos += n; + return at; + } + + skip(n: number) { + this.take(n); + } + + u8(): number { + return this.view.getUint8(this.take(1)); + } + + u16(): number { + return this.view.getUint16(this.take(2), true); + } + + u32(): number { + return this.view.getUint32(this.take(4), true); + } + + u64(): bigint { + return this.view.getBigUint64(this.take(8), true); + } + + bytes(): Uint8Array { + const length = this.u32(); + const at = this.take(length); + return this.src.subarray(at, at + length); + } + + shortBytes(): Uint8Array { + const length = this.u16(); + const at = this.take(length); + return this.src.subarray(at, at + length); + } + + str(): string { + return Buffer.from(this.bytes()).toString("utf8"); + } +} diff --git a/lib/node/pl-tree/src/state.ts b/lib/node/pl-tree/src/state.ts index b4856c3127..5f6b393ace 100644 --- a/lib/node/pl-tree/src/state.ts +++ b/lib/node/pl-tree/src/state.ts @@ -469,6 +469,14 @@ export class PlTreeState { this.resources.forEach((v) => cb(v)); } + /** False once the tree has been invalidated (an inconsistent update, or termination of + * its synchronization loop). {@link dumpState} deliberately reads through an invalid + * tree, so anything persisting that dump must check this first: the contents of an + * invalidated tree are not something to write to disk. */ + public get isValid(): boolean { + return this._isValid; + } + private checkValid() { if (!this._isValid) throw new Error(this.invalidationMessage ?? "tree is in invalid state"); } From 68f19fa191dfd36ba8741d186d85b5b9cffb27e6 Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Thu, 13 Aug 2026 16:35:55 +0200 Subject: [PATCH 2/7] MILAB-6705: restore project tree mirrors from disk on open Wires the snapshot codec into the tree and the middle layer. SynchronizedTreeState gains a restoreFrom option that seeds the tree before its first refresh, a capture method, and a change generation bumped on every cycle that brought something new, which is what the periodic write gates on. On the middle-layer side a filesystem store addresses one file per project by backend, user, root, build stamp and schema version, witnesses the session with the root's signature so a rotated session is a miss that keeps the file, evicts out-of-scope entries and trims to a ceiling at startup, and writes nothing at all for an impersonated client. Writes happen on the existing project maintenance loop, change-gated and once per interval, plus once at closeProject, deliberately not in the shutdown teardown. A restored tree whose first refresh is refused discards its snapshot and reopens cold, once. The build stamp is injected through rolldown's transform.define, so any change to the pruning, field filter, traversal stop or finality rules invalidates every snapshot. A dirty worktree stamps the build time too, so editing those rules locally cannot hit a stale mirror. --- .changeset/disk-persist-tree-middle-layer.md | 6 + lib/node/pl-middle-layer/build.node.config.js | 39 ++- .../src/middle_layer/build_stamp.ts | 31 +++ .../src/middle_layer/middle_layer.ts | 21 ++ .../pl-middle-layer/src/middle_layer/ops.ts | 47 +++- .../src/middle_layer/project.ts | 180 +++++++++++-- .../middle_layer/tree_snapshot_store.test.ts | 236 ++++++++++++++++++ .../src/middle_layer/tree_snapshot_store.ts | Bin 0 -> 11949 bytes lib/node/pl-tree/src/persisted_tree.ts | 62 +++++ lib/node/pl-tree/src/synchronized_tree.ts | 82 +++++- 10 files changed, 684 insertions(+), 20 deletions(-) create mode 100644 .changeset/disk-persist-tree-middle-layer.md create mode 100644 lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts create mode 100644 lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts create mode 100644 lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts diff --git a/.changeset/disk-persist-tree-middle-layer.md b/.changeset/disk-persist-tree-middle-layer.md new file mode 100644 index 0000000000..800dbe74a0 --- /dev/null +++ b/.changeset/disk-persist-tree-middle-layer.md @@ -0,0 +1,6 @@ +--- +"@milaboratories/pl-middle-layer": minor +"@milaboratories/pl-tree": minor +--- + +Persist project tree mirrors to disk and restore them on open, so reopening a project transfers what changed rather than the whole tree. On by default, with a kill switch in `treeSnapshotOps`. diff --git a/lib/node/pl-middle-layer/build.node.config.js b/lib/node/pl-middle-layer/build.node.config.js index 10888bc1ac..fd5c8f7158 100644 --- a/lib/node/pl-middle-layer/build.node.config.js +++ b/lib/node/pl-middle-layer/build.node.config.js @@ -1,5 +1,42 @@ import { createRolldownNodeConfig } from "@milaboratories/ts-builder/configs/utils/createRolldownNodeConfig.js"; +import { execFileSync } from "node:child_process"; + +/** + * Identifies this build for the persisted-tree cache key, see `src/middle_layer/build_stamp.ts`. + * + * A clean worktree stamps its commit, so every build of a given release shares a stamp and a + * reopen stays warm across restarts. A dirty worktree stamps the build time too, so editing + * the tree pruning or finality rules locally cannot hit a snapshot written under the old + * ones. `git status` covers the whole repo, which over-invalidates rather than under-. + */ +function buildStamp() { + const git = (args) => + execFileSync("git", args, { stdio: ["ignore", "pipe", "ignore"] }) + .toString() + .trim(); + try { + const sha = git(["rev-parse", "--short=12", "HEAD"]); + const dirty = git(["status", "--porcelain"]).length > 0; + return dirty ? `${sha}-dirty-${Date.now()}` : sha; + } catch { + // No git available (a published tarball being rebuilt, for instance). Falling back to the + // build time keeps the stamp honest: it cannot claim to be a commit it does not know. + return `nogit-${Date.now()}`; + } +} export default createRolldownNodeConfig({ entry: ["./src/index.ts", "./src/worker/worker.ts"], -}); +}).map((config) => ({ + ...config, + // Note `transform.define`, not a top-level `define`: rolldown ignores the latter without + // complaining, which leaves the identifier in the output and the cache permanently cold. + // The spread preserves `transform.target` from the shared config. + transform: { + ...config.transform, + define: { + ...config.transform?.define, + __PL_ML_BUILD_STAMP__: JSON.stringify(buildStamp()), + }, + }, +})); diff --git a/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts b/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts new file mode 100644 index 0000000000..d1c6ac2ce9 --- /dev/null +++ b/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts @@ -0,0 +1,31 @@ +import { randomUUID } from "node:crypto"; + +/** Injected by rolldown at build time, see `build.node.config.js`. Absent when the package is + * consumed straight from sources (`USE_SOURCES=1`), because no build step runs then. */ +declare const __PL_ML_BUILD_STAMP__: string | undefined; + +function injectedStamp(): string | undefined { + try { + // Read inside a try: with no build step the identifier is an undeclared global, and + // reading it throws a ReferenceError rather than yielding undefined. + return __PL_ML_BUILD_STAMP__; + } catch { + return undefined; + } +} + +/** + * Identifies the build of this package, and through it every rule that shapes what a + * persisted tree mirror contains: the pruning function, the field filter, the traversal stop + * rules and the finality predicate all live in this package or in one it pins. + * + * Used as a cache-key component, so any change to those rules invalidates every snapshot, + * costing one cold open. A build from a clean worktree stamps its commit, so released builds + * share a stamp and reopens stay warm across restarts. A build from a dirty worktree stamps + * the build time as well, so editing those rules locally can never hit a snapshot written + * under the old ones. + * + * With no build at all (sources mode) the value is unique per process, so nothing ever hits. + * That is the safe direction: an unbuilt tree has no way to say which rules were in force. + */ +export const ML_BUILD_STAMP: string = injectedStamp() ?? `unbuilt-${randomUUID()}`; diff --git a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts index 17ceaf4731..ccbc537982 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts @@ -94,6 +94,7 @@ import type { Dispatcher } from "undici"; import { RetryAgent } from "undici"; import { getDebugFlags } from "../debug"; import { ProjectHelper } from "../model/project_helper"; +import { TreeSnapshotStore } from "./tree_snapshot_store"; export interface MiddleLayerEnvironment { dispose(): Promise; @@ -112,6 +113,9 @@ export interface MiddleLayerEnvironment { readonly driverKit: MiddleLayerDriverKit; readonly serviceRegistry: ModelServiceRegistry; readonly projectHelper: ProjectHelper; + /** Persisted project tree mirrors. Undefined when snapshots are switched off, or when the + * client is impersonating another user, in which case nothing is read or written. */ + readonly treeSnapshots?: TreeSnapshotStore; } /** @@ -954,6 +958,12 @@ export class MiddleLayer { const prj = this.openedProjects.get(id); if (prj === undefined) throw new Error(`Project ${id} not found among opened projects`); this.openedProjects.delete(id); + + // Snapshot before destroy, and here rather than inside destroy(): destroy() is also what + // application shutdown runs, and quitting should perform no snapshot work. Terminating the + // tree invalidates it, so the state has to be taken first either way. + await prj.writeSnapshotOnClose(); + await prj.destroy(); this.openedProjectsList.setValue([...this.openedProjects.keys()]); } @@ -1092,6 +1102,16 @@ export class MiddleLayer { const serviceRegistry = createModelServiceRegistry({ logger }); + const treeSnapshots = TreeSnapshotStore.create(pl, { + dir: ops.treeSnapshotPath, + maxSizeBytes: ops.treeSnapshotOps.maxSizeBytes, + enabled: ops.treeSnapshotOps.enabled, + logger, + }); + // Housekeeping before any project opens: drop snapshots from other builds, backends and + // users, then trim to the ceiling. + await treeSnapshots?.evict(); + const env: MiddleLayerEnvironment = { pl, blockEventDispatcher: new BlockEventDispatcher(), @@ -1112,6 +1132,7 @@ export class MiddleLayer { serviceRegistry, quickJs, projectHelper: new ProjectHelper(quickJs, logger), + treeSnapshots, dispose: async () => { await serviceRegistry.dispose(); await retryHttpDispatcher.destroy(); diff --git a/lib/node/pl-middle-layer/src/middle_layer/ops.ts b/lib/node/pl-middle-layer/src/middle_layer/ops.ts index ff22690112..9ac1e95f27 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/ops.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/ops.ts @@ -222,6 +222,40 @@ export type DriverKitOpsConstructor = Omit< export type MiddleLayerOpsPaths = DriverKitOpsPaths & { /** Common root where to put frontend code. */ readonly frontendDownloadPath: string; + + /** + * Directory holding persisted project tree mirrors, one file per project. Like + * {@link DriverKitOpsPaths.parquetCachePath} and unlike the spill directories, it is NOT + * emptied on startup: surviving a restart is the entire point. It is pruned instead, see + * {@link TreeSnapshotOps.maxSizeBytes}. + */ + readonly treeSnapshotPath: string; +}; + +/** Tuning for the persisted project tree mirrors. Their directory is + * {@link MiddleLayerOpsPaths.treeSnapshotPath}; this carries the behaviour knobs. */ +export type TreeSnapshotOps = { + /** + * Whether project tree mirrors are persisted and restored at all. + * + * On by default. This is an operational kill switch, for a deployment where the cache + * directory turns out to be unwritable or otherwise troublesome, not a rollout gate: the + * floor of the feature is current behaviour, since a cache that never hits is a cold open. + */ + readonly enabled: boolean; + + /** + * Minimum wall-clock gap between periodic writes for one project. + * + * Can be generous, because a stale snapshot is less complete rather than wrong: final + * resources never change and are never refetched, so this only bounds how much recent work + * comes back from the non-final frontier on restore. + */ + readonly writeInterval: number; + + /** Total bytes the snapshot directory may occupy after startup eviction. Needed because a + * heavy project runs to roughly ten megabytes. */ + readonly maxSizeBytes: number; }; /** Debug options for middle layer. */ @@ -253,6 +287,10 @@ export type MiddleLayerOpsSettings = DriverKitOpsSettings & { * `sharedAt + envelopeTtlMs`. Share-with-everybody envelopes never expire * (`expiresAt: null`) and ignore this. */ readonly envelopeTtlMs: number; + + /** Settings for persisting project tree mirrors to disk, so reopening a project transfers + * what changed rather than the tree again. */ + readonly treeSnapshotOps: TreeSnapshotOps; }; export type MiddleLayerOps = MiddleLayerOpsSettings & MiddleLayerOpsPaths; @@ -266,6 +304,7 @@ export const DefaultMiddleLayerOpsSettings: Pick< | "devBlockUpdateRecheckInterval" | "debugOps" | "envelopeTtlMs" + | "treeSnapshotOps" > = { ...DefaultDriverKitOpsSettings, defaultTreeOptions: { @@ -279,17 +318,23 @@ export const DefaultMiddleLayerOpsSettings: Pick< devBlockUpdateRecheckInterval: 1000, projectRefreshInterval: 2000, envelopeTtlMs: 14 * 24 * 3600 * 1000, // 14 days + treeSnapshotOps: { + enabled: true, + writeInterval: 5 * 60 * 1000, // 5 minutes + maxSizeBytes: 256 * 1024 * 1024, // 256 MB, roughly 25 heavy projects + }, }; export function DefaultMiddleLayerOpsPaths( workDir: string, ): Pick< MiddleLayerOpsPaths, - keyof ReturnType | "frontendDownloadPath" + keyof ReturnType | "frontendDownloadPath" | "treeSnapshotPath" > { return { ...DefaultDriverKitOpsPaths(workDir), frontendDownloadPath: path.join(workDir, "frontend"), + treeSnapshotPath: path.join(workDir, "treeSnapshots"), }; } diff --git a/lib/node/pl-middle-layer/src/middle_layer/project.ts b/lib/node/pl-middle-layer/src/middle_layer/project.ts index 28874f6bc0..cb1b152e1b 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/project.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/project.ts @@ -11,7 +11,10 @@ import { ensureSignedResourceIdNotNull, field, isNotFoundError, + isPermissionDenied, isTimeoutOrCancelError, + isUnauthenticated, + parseSignedResourceId, Pl, resourceIdToString, ResourceTypeName, @@ -25,7 +28,12 @@ import type { BlockPackSpecAny } from "../model"; import { randomUUID } from "node:crypto"; import { withProject, withProjectAuthored } from "../mutator/project"; import type { ExtendedResourceData, PruningFunction } from "@milaboratories/pl-tree"; -import { SynchronizedTreeState, treeDumpStats } from "@milaboratories/pl-tree"; +import { + SynchronizedTreeState, + treeDumpStats, + TreeStateUpdateError, +} from "@milaboratories/pl-tree"; +import type { TreeSnapshotStore } from "./tree_snapshot_store"; import { setTimeout } from "node:timers/promises"; import { frontendData } from "./frontend_path"; import type { NavigationState } from "@milaboratories/pl-model-common"; @@ -102,6 +110,16 @@ export class Project { private readonly abortController = new AbortController(); + /** Tree change generation as of the snapshot currently on disk, or -1 when this session has + * not written one. Compared against the tree's current generation to skip writing a mirror + * that has not moved, which is what makes a project left open and idle go quiet. */ + private snapshotGeneration: number; + + /** When the last snapshot was written, for the periodic write's wall-clock gate. Seeded at + * construction so the first write lands one interval after the project opens rather than + * immediately. */ + private lastSnapshotAt = Date.now(); + private get destroyed() { return this.abortController.signal.aborted; } @@ -111,7 +129,11 @@ export class Project { public readonly id: ProjectId /* Project ID, exposed to outer consumers, who work with ML */, readonly rid: SignedResourceId /* Contains signature, not exposed outside middle layer. */, private readonly projectTree: SynchronizedTreeState, + /** Whether this tree was seeded from a snapshot. When it was, the file on disk already + * holds generation 0, so an idle warm reopen writes nothing at all. */ + restoredFromSnapshot: boolean = false, ) { + this.snapshotGeneration = restoredFromSnapshot ? 0 : -1; this.overview = projectOverview( projectTree.entry(), this.navigationStates, @@ -129,6 +151,76 @@ export class Project { return "project:" + this.id.toString(); } + /** + * Periodic snapshot write, carried on the maintenance loop rather than a timer of its own. + * + * Gated on the tree having changed since the last snapshot, so a project left open and idle + * writes once and then goes quiet, and on wall clock, so a project changing continuously + * writes at most once per interval. + */ + private async maybeWriteSnapshot(): Promise { + const store = this.env.treeSnapshots; + if (store === undefined) return; + + const generation = this.projectTree.changeGeneration; + if (generation === this.snapshotGeneration) return; + if (Date.now() - this.lastSnapshotAt < this.env.ops.treeSnapshotOps.writeInterval) return; + + await this.writeSnapshot(store, generation); + } + + /** + * Snapshot write at the close boundary, on top of the periodic one, since closing is a + * natural point to persist. Change-gated but not interval-gated: rewriting a mirror that has + * not moved is pure waste, but a mirror that has moved is worth keeping however recently the + * last write happened. + * + * Must run before {@link destroy}, which terminates the tree and thereby invalidates it. + */ + public async writeSnapshotOnClose(): Promise { + const store = this.env.treeSnapshots; + if (store === undefined) return; + + const generation = this.projectTree.changeGeneration; + if (generation === this.snapshotGeneration) return; + + await this.writeSnapshot(store, generation); + } + + /** In-flight snapshot write, if any. Both triggers can fire close together (the close write + * lands while the loop is mid-write), and encoding ten megabytes twice for the same mirror + * is worth avoiding. */ + private snapshotInFlight: Promise | undefined; + + /** Serializes writes, and skips one that the in-flight write has already made redundant. */ + private async writeSnapshot(store: TreeSnapshotStore, generation: number): Promise { + if (this.snapshotInFlight !== undefined) { + await this.snapshotInFlight; + if (this.snapshotGeneration >= generation) return; + } + + this.snapshotInFlight = this.captureAndWrite(store, generation).finally(() => { + this.snapshotInFlight = undefined; + }); + await this.snapshotInFlight; + } + + /** Captures and writes, never throwing: a snapshot is an optimisation and must not delay or + * fail whatever triggered it. */ + private async captureAndWrite(store: TreeSnapshotStore, generation: number): Promise { + try { + // The root's signature is the session witness a later open compares against. + const snapshot = this.projectTree.capture(parseSignedResourceId(this.rid).signature); + await store.write(this.rid, snapshot); + this.snapshotGeneration = generation; + this.lastSnapshotAt = Date.now(); + } catch (e: unknown) { + this.env.logger.warn( + new Error(`failed to capture tree snapshot for project ${this.id}`, { cause: e }), + ); + } + } + private async refreshLoop(): Promise { let retryState: InfiniteRetryState | undefined; while (!this.destroyed) { @@ -147,6 +239,8 @@ export class Project { signal: this.abortController.signal, }); + await this.maybeWriteSnapshot(); + // Block computables housekeeping const overviewLight = await this.overviewLight.getValue(); const existingBlocks = new Set(overviewLight.listOfBlocks); @@ -727,18 +821,8 @@ export class Project { // Doing a no-op mutation to apply all migration and schema fixes await withProject(env.projectHelper, env.pl, rid, (_) => {}, { name: "init" }); - // Loading project tree - const projectTree = await SynchronizedTreeState.init( - env.pl, - rid, - { - ...env.ops.defaultTreeOptions, - pruning: projectTreePruning(env.logger), - fieldFilter: projectTreeFieldFilter(), - traverseStopRules: projectTreeTraverseStopRules(), - }, - env.logger, - ); + // Loading project tree, warm from a persisted mirror when one is usable + const { tree: projectTree, restored } = await loadProjectTree(env, rid); if (env.ops.debugOps.dumpInitialTreeState) { const state = projectTree.dumpState(); @@ -748,10 +832,78 @@ export class Project { await fs.writeFile(`${resourceIdToString(rid)}.stats.json`, stringifyForDump(stats)); } - return new Project(env, id, rid, projectTree); + return new Project(env, id, rid, projectTree, restored); } } +/** + * Opens the project tree, seeded from a persisted mirror when there is a usable one. + * + * Carries the fail-safe: if the restored tree fails its first refresh on authentication, + * permission or an inconsistency, the snapshot is deleted and the open is retried cold. Once, + * and only for that first refresh, so a genuinely dead session still surfaces as itself rather + * than being masked as a slow open. + * + * The fail-safe is what bounds every case the cache key does not cover: a rotated master + * secret, a revoked grant, a snapshot valid in itself but no longer matching what the backend + * will serve. Without it, an explicit-root tree propagates the refresh failure rather than + * healing, so the project would fail to open on every attempt until someone deleted the cache + * directory by hand. + */ +async function loadProjectTree( + env: MiddleLayerEnvironment, + rid: SignedResourceId, +): Promise<{ tree: SynchronizedTreeState; restored: boolean }> { + const treeOps = { + ...env.ops.defaultTreeOptions, + pruning: projectTreePruning(env.logger), + fieldFilter: projectTreeFieldFilter(), + traverseStopRules: projectTreeTraverseStopRules(), + }; + const cold = async () => ({ + tree: await SynchronizedTreeState.init(env.pl, rid, treeOps, env.logger), + restored: false, + }); + + const store = env.treeSnapshots; + if (store === undefined) return await cold(); + + const snapshot = await store.read(rid); + if (!snapshot.ok) { + env.logger.info(`project tree opening cold, snapshot miss: ${snapshot.miss}`); + return await cold(); + } + + try { + return { + tree: await SynchronizedTreeState.init( + env.pl, + rid, + { ...treeOps, restoreFrom: snapshot.tree }, + env.logger, + ), + restored: true, + }; + } catch (e: unknown) { + if (!isSnapshotFailsafeError(e)) throw e; + + env.logger.warn( + new Error("restored project tree failed its first refresh, discarding it and opening cold", { + cause: e, + }), + ); + await store.discard(rid); + return await cold(); + } +} + +/** The failures that can mean a snapshot no longer matches what the backend will serve, as + * opposed to a client that has genuinely lost its session. Both look the same on one refresh, + * which is why the retry is spent only on the first. */ +function isSnapshotFailsafeError(e: unknown): boolean { + return isUnauthenticated(e) || isPermissionDenied(e) || e instanceof TreeStateUpdateError; +} + export function projectTreePruning(logger: MiLogger): PruningFunction { return (r: ExtendedResourceData): FieldData[] => { if (r.fields.length > 1000) diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts new file mode 100644 index 0000000000..a205c7f730 --- /dev/null +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts @@ -0,0 +1,236 @@ +import { beforeEach, describe, expect, test } from "vitest"; +import type { PlClient, SignedResourceId } from "@milaboratories/pl-client"; +import { createSignedResourceId, toResourceSignature } from "@milaboratories/pl-client"; +import type { PersistedTree } from "@milaboratories/pl-tree"; +import type { MiLogger } from "@milaboratories/ts-helpers"; +import fsp from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { TreeSnapshotStore } from "./tree_snapshot_store"; + +const silent: MiLogger = { info: () => {}, warn: () => {}, error: () => {} }; + +const sig = (hex: string) => toResourceSignature(Buffer.from(hex, "hex")); + +/** Only the three fields the store reads. */ +function fakeClient(ops: { host?: string; user?: string | null; asUser?: string } = {}): PlClient { + return { + conf: { hostAndPort: ops.host ?? "localhost:6345", asUser: ops.asUser }, + authUser: ops.user === undefined ? "someone@example.com" : ops.user, + } as unknown as PlClient; +} + +/** A snapshot with roots and no resources: enough to exercise the store, since the codec is + * tested against real trees in pl-tree. */ +function snapshotFor(root: SignedResourceId): PersistedTree { + return { + witness: toResourceSignature(Buffer.from(root.split("|")[1], "hex")), + roots: [root], + resources: [], + }; +} + +let dir: string; + +beforeEach(async () => { + dir = await fsp.mkdtemp(path.join(os.tmpdir(), "tree-snapshots-")); +}); + +function storeIn( + dirPath: string = dir, + ops: { maxSizeBytes?: number; enabled?: boolean; client?: PlClient } = {}, +): TreeSnapshotStore | undefined { + return TreeSnapshotStore.create(ops.client ?? fakeClient(), { + dir: dirPath, + maxSizeBytes: ops.maxSizeBytes ?? 256 * 1024 * 1024, + enabled: ops.enabled ?? true, + logger: silent, + }); +} + +const rootA = createSignedResourceId(1001n, sig("aaaa")); +const rootB = createSignedResourceId(1002n, sig("bbbb")); + +async function files(): Promise { + return (await fsp.readdir(dir)).sort(); +} + +describe("when the store should not exist at all", () => { + test("disabled by configuration", () => { + expect(storeIn(dir, { enabled: false })).toBeUndefined(); + }); + + test("client is impersonating another user", () => { + // Reading or writing here would leave another user's mirror at rest under the admin's + // identity, so nothing is persisted for an impersonated client. + const client = fakeClient({ asUser: "someone-else@example.com" }); + expect(storeIn(dir, { client })).toBeUndefined(); + }); +}); + +describe("round trip", () => { + test("a written snapshot reads back", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + + const read = await store.read(rootA); + expect(read.ok).toBe(true); + if (!read.ok) throw new Error("unreachable"); + expect(read.tree.roots).toStrictEqual([rootA]); + + expect(store.stats.writes).toBe(1); + expect(store.stats.hits).toBe(1); + }); + + test("nothing written means an absent miss", async () => { + const store = storeIn()!; + expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + expect(store.stats.misses.absent).toBe(1); + }); + + test("one file per project, rewritten in place", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + await store.write(rootA, snapshotFor(rootA)); + await store.write(rootB, snapshotFor(rootB)); + + expect(await files()).toHaveLength(2); + }); + + test("a successful write leaves no staging file behind", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + expect((await files()).filter((f) => f.includes(".tmp."))).toStrictEqual([]); + }); + + test("discard removes the file", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + await store.discard(rootA); + + expect(await files()).toStrictEqual([]); + expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + }); +}); + +describe("the session witness", () => { + test("a rotated signature is a miss, and the file is kept", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + + // Same resource, next session: same global id, different signature. The file is addressed + // by global id, so this is the same file, and only the witness distinguishes them. + const rotated = createSignedResourceId(1001n, sig("cccc")); + expect(await store.read(rotated)).toStrictEqual({ ok: false, miss: "session-rotated" }); + + // Kept on purpose: the bodies stay valid, only the signatures died, so a future signature + // refresh would have something to repair. + expect(await files()).toHaveLength(1); + }); +}); + +describe("a snapshot that cannot be read", () => { + test("a truncated file misses without raising", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + + const [name] = await files(); + const file = path.join(dir, name); + const bytes = await fsp.readFile(file); + await fsp.writeFile(file, bytes.subarray(0, bytes.length - 6)); + + const read = await store.read(rootA); + expect(read.ok).toBe(false); + if (read.ok) throw new Error("unreachable"); + expect(["truncated", "checksum"]).toContain(read.miss); + }); + + test("a foreign file in our own filename misses without raising", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + const [name] = await files(); + await fsp.writeFile(path.join(dir, name), "not a snapshot at all"); + + expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "not-a-snapshot" }); + }); +}); + +describe("the key", () => { + test("another backend does not see this one's snapshots", async () => { + await storeIn()!.write(rootA, snapshotFor(rootA)); + + const other = storeIn(dir, { client: fakeClient({ host: "elsewhere:6345" }) })!; + expect(await other.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + }); + + test("another user does not see this one's snapshots", async () => { + await storeIn()!.write(rootA, snapshotFor(rootA)); + + const other = storeIn(dir, { client: fakeClient({ user: "other@example.com" }) })!; + expect(await other.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + }); +}); + +describe("eviction", () => { + test("drops what is not addressed to the current scope", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + + // A snapshot from another build, and a staging file from a write that was killed before + // its rename. Neither can ever be read again. + await fsp.writeFile(path.join(dir, "tree.1.otherbuild.0123456789abcdef.99.plts"), "old"); + await fsp.writeFile(path.join(dir, "tree.1.thisbuild.0123456789abcdef.99.plts.tmp.ab"), "torn"); + + await store.evict(); + + expect(await files()).toHaveLength(1); + expect((await store.read(rootA)).ok).toBe(true); + expect(store.stats.evicted).toBe(2); + expect(store.stats.evictedForSize).toBe(0); + }); + + test("keeps everything when under the ceiling", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + await store.write(rootB, snapshotFor(rootB)); + + await store.evict(); + expect(await files()).toHaveLength(2); + expect(store.stats.evicted).toBe(0); + }); + + test("trims to the ceiling, least recently written first", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + await store.write(rootB, snapshotFor(rootB)); + + const names = await files(); + const sizes = await Promise.all(names.map((n) => fsp.stat(path.join(dir, n)))); + const perFile = Math.max(...sizes.map((s) => s.size)); + + // Age rootA's file so recency is unambiguous rather than dependent on write order timing. + const old = new Date(Date.now() - 60 * 60 * 1000); + const rootAFile = path.join(dir, names.find((n) => n.endsWith(".1001.plts"))!); + await fsp.utimes(rootAFile, old, old); + + // Room for one file only. + const tight = storeIn(dir, { maxSizeBytes: perFile })!; + await tight.evict(); + + expect((await tight.read(rootA)).ok).toBe(false); + expect((await tight.read(rootB)).ok).toBe(true); + expect(tight.stats.evictedForSize).toBe(1); + }); + + test("an unusable directory costs the cache, not the startup", async () => { + // A path that cannot be a directory, because a file already occupies it. + const occupied = path.join(dir, "occupied"); + await fsp.writeFile(occupied, "in the way"); + + const store = storeIn(occupied)!; + await expect(store.evict()).resolves.toBeUndefined(); + await expect(store.write(rootA, snapshotFor(rootA))).resolves.toBeUndefined(); + expect(store.stats.writeFailures).toBe(1); + expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + }); +}); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts new file mode 100644 index 0000000000000000000000000000000000000000..d3f6291d8dcf4712eff07aec61000ecf063c0c09 GIT binary patch literal 11949 zcmb_iYjfPjvCU`wipiGC*_D@|lB%1kOPO|Li!8HZJ&KgCQ*k*W7MSHC1TY95BsVnw z`#q<72ACx&OXWl#R9rlIdb&?{pYB0XHLmSVziq6!HE(U(6%0w0CV`yZxrz>+$C7`IAEB-aauVx0%cBu4r7g^`n>H9-W?@ zoV`1G`TpJM(b4;}Z@)Ww_2T{i@#>q`OT5=Yj~!v(;eu_KPd-_xu-Wq4vaof(G-t&{ zZSzyxxuMPMN&aj5%jBJCho(u}&i*GNygsjr@7=|PZ9lkr-|cN|*}zBf@)-=$+qY@I zdC|M7$kMXhE@9mcj(SlY2_nTyXmv)=+#Tq_5$l7hwyJ_`x*LWnz?KKbpToj#+ zG;E+*{1AUn5^l|_@85s(!^!tA-=DpE@#^ikUb4R$iZXxS^=Z}M`6v5NpPC;xTayX` zOkU)scD=!Z`H)#$Ch|=0aJvH>t*n`tZ{1J>&;}UYz_$G+txZv{ZQI&>X;wB%ht57R zPmP(?&UAI!bQ{;tOj_q=mW~(9wT0j{p3QL&(ywg0g%WE>0)`;Bs4qHmy|Fb^xlY>( zkGsa!B6^Lu>8RVv>OD&DTxDHt%_fyw*RH)ZZs;Lno0zBjPwdSUnvBOYuiLAl>&zh{ zffjk~%(^JCS;xHBz18A@=+O9WLU}@x*?^UED`Eh~_M@t|x0Y;_2qvHLvZ#V?r~O7%YM$$S7er!6xO7PE$VCHq6iJDsaX z={rf|bjoJH)`v`e1~tr~VHV>u-GQlxYGvE!xUj+2G*Y*+xU)H_UPdosCi}45kOcvenD^07dtoZ0I>B?XFY?V zsBt$f39R|zz{DsJuqo6KOtqkH#N9KNz5qJ^BOr?J@Sle2f$_JI(eB4nd=K{8OZUcv z7vCD{%i3MndtJ7%RXU4q^({8S(2TC(q3p67s_0UcmT*uNJ$9%12Vi>@@1fdX-G#Kj z7-7y|8J)TNd-P;1ZFHl*c+ZDBS4RiRRe102*AfIO&6@`KSlq>+L(R1r3eeVUTuF!s zl7pSQ#JoYymz<4C&hqt7Jnr$yB;)Fmg{3_}@U(cH`G9iYAt#myV1&C!#3O#4Dn&?T z+0dXkN;vzdGh!j2?!NRYy&>4Yp{y9dK}n@Bj3V7khtC_x}BC?|A3$px~x?j?@TVz~mazP611$ zbp{Td4oFLN4_=0-kzBMitwZs{OF}nUM?rL$DyxR#g&X)Z`{wOB?Tybo_fs51}CAKpl?gJl9h;ey&m{1;684(bK| z0GS{P1!5k_B_+auN{~!4$i?N=p#c0?cR4)KC5CUkUPi z+-Z|up%@nALoZ2;m^up3L6RNno})y!P7639J4XXW+PfY|y(SWdm8Dkei}=-~MN7Mm zj4~@zFv#7~L&D5D(zL}D%q+$cB>_uPpq;vfXrG>xtk4cjd2=a54@9^IS?0lV$eLOc znA}dVo4#1na3ea1mk=Bh6H<*5bH>EN`mgCEfD*5?FyB z(W>Can(7REOXla_CbrYODr!7HeDy_-R&wal6xETZFm&nsfx{9@#{~ik^ zUJ9TjYs>11meB<5rkl@twx=Q209=rV3@dB1a==XM;$lFKgCSV3xCPQbsc|F1^!lCL z<>F15@XbX1Zhn7wIO+Hsl|jLPyXharn+Y)o!t3+)@{;1{m-9g z3-8rrXmUjK{QR@q;6U_4{fjz(i)q~5@8mL>1B2Oo{k56R?#{=T^Wp;WGM{bi%_Oly zT-f=urTOEZ0UlXLHIj4(b+Fd2&erad&u)cq?G9uXydHLScdq#JsN}fdL0+FD96$=l zJ_N;*T7BG-=`nUY_CBYc?C!L}XSn;74HxdqI4^;yMj{>urb1&a|{3ynHRlGxFCZK1Ax^qx-U6_nMkAr4Dtn`8r1+C z5O|RjZvA>PTRePxWGLJ4f$z7@Sy}Q=SJZPc0uv2r|I(Z@nZPe<8k(A3(k-3jcc0xx zBptvR4JQnkHG#I*HX>vsDN5-`@n|Jq<2Z%lWJ{O_vKEL>sBp;nEFZ{?aWOmR%jbNK zHUa%X#F6cmoIOdIMrs)_fWzEutkx%3V;m$2m*dz-yg>P2K-Ak4Jl-FvTzUbb)j`{i zQzt(<3WE7$jv*7g zV)TIrx2X1E=TUJ=yKS9GebLJFI68%Rpk(R1^vhf0E&;lBuwdYDF75=nzzD-w^cNH@ zKC@2C4iAx$?y{;^oD1HC;PJuF`3saiG%vB41YP~yBW#Q+v`)an7hh;(_h`^=1-J&p zq1S%%0E~!(PJP5c!RaC6Thc?5UZ#0ZXRX;V-k0fj`GCU_PM?tx?m!*BFRa5J91PBn?Mj3V}V?E~Nq z%wNcd3{4eaWv($y32`!_?dM@KI_*fauJM&%1C^k~q3E{bOg*reODjLVElfh+N!;b) z;d=g_+5oS4Ct+uAIC2i1XM{} zk*IO<$f?+4i14pH4vBE{J|?Ev=8!ss_K4vr*DQUm34+9g2D2VI2Ix@$`j~{YgpAYZU9l;rdUM`g7Q?kJw1yQnIG*Z@1&) z%#SfK5_AqH2v=>hwy29}x!vH$8@?gz*opf7GcJIy*_i@9t-B^+MnRDat>vf>SsMpL zT#reH4Nik03_1Z4f1mNt)R!gO1Dpt<4Wv7yf8_q~RL4a@T`Lo~hD`JYu;2pPL*N$+ zC;}H}iY&Zf!eCbyQo2CJWkBc4s!WX}I(-#|{0xH>bBS3uUZ{$EPgRy?QiFap$w(n$ zdXMus*Vd}}V=bIkNfG+7GHFJy+^dS3sHD%k6+VEJ*L@vza)`Fu(9VTQd*zD!lZ#|j z=roLV{u+Y#ku&Uhu*MQlCP$%82W&>ssOc|rH)eg!1@{hXcgWuTJR*hf5Y7cD-l?pi zIIGCVPK77A+L$hduL5#HSmqeWHatdNn8RQnrU4X~+4ap#*;VFn_Mt;G@V!2&1j7C%%49@65x3d6xm9N_>dm_g%Mc5HaV zaRj=jajrhLaO^+LJPd~P09HM{Gs|Tu+da-$fPWAx0CvR=6CglIEbSU~OHVO698-l; z;3X%SU9XaH_60%)Zo+6{L<_@=i|{MV;nzHN;b2jYIT&5&deMP$xGEAdIgG}(QIuJU zAqb_9Wty6d=M%anDr*6u{tFv)n}U0!ETlDtXpeKQ%MNC|R^e2Sj|7R==!5x{2ZDx2 zIj2mQR18Nm(%brJG3WvNKp2bPRh06X1$F_d0hNZj#aW2Xt+LjkM5Ao?99bySsQB@L zPzun5*c@>b$!3Sz4|cN@gLRONaGdDJ))@v;qQ2aZrX-3MmHnXYl_N(^cRnrXK2rIZ zIYXzwTElbJC|Y4as6CZ6{FrKj)@u(^@=8{oF!p~A<3^|brlNgcp-$@;7v}D#pRv?W zKaUlLh?%QMDSF7!qR>!?z|SrM$kV{fQoP;&hzoLtr<|c*sZhX;t0* z*ny*Zz{a)2ll(*N-iw#FF}4H1CRy%L#_0kVEq<@FSgTc!$GL(T%t$aF41T)Zq!(ggaiC(`3HTtk85$z zEXP3u*b{)uq><)K8UQ$uLnZ)p;mlpK1zbDIlgcn{$a^M6In1K7yQCQ~+Sa&2+kma0 zztT90YNWU0Ju^hhs9mL$>W4#O+8|J-U89}H*VHrq^B`DLUS~y(KWzwhBMN%?dICdv zuRA&-d^I?qYMl zKa#TsCtJR($7F(~3$=i*;K3Jyi;tsIBIN?$@cHHy&KpCh1id>9W9Rb}+*ou@lVI^Z zvkHHYnny_s#S*;?BHRm#J^J_tSHR3=Fd;rJlIdRTCJ`w;2tPN4cyBotuTcf?VzK<* zVf0Lps>fIX)W3x)F+EK5G<@$OWRypG{~4=BPzd0{fp!?|@)T1*8mftcH53oDRpDbY zf28=yPuxBEe=tZymjD0& literal 0 HcmV?d00001 diff --git a/lib/node/pl-tree/src/persisted_tree.ts b/lib/node/pl-tree/src/persisted_tree.ts index eec799b398..808b4df929 100644 --- a/lib/node/pl-tree/src/persisted_tree.ts +++ b/lib/node/pl-tree/src/persisted_tree.ts @@ -1,3 +1,65 @@ +/** + * On-disk format for a tree mirror. + * + * Header and trailer are always plain bytes; only the payload is compressed. Every integer + * is little-endian and fixed-width. + * + * ```text + * +-- header (never compressed) -----------------------------+ + * | u32 magic 0x53544C50 ("PLTS", little-endian) | + * | u16 schemaVersion | + * | u16 flags bit0 = payload is deflated | + * | u16 witnessLen + witness bytes (the root's signature) | + * +-- payload (deflated, or raw if the flag is clear) -------+ + * | u32 rootCount, then u64 globalId per root | + * | | + * | u32 signatureCount, then per entry: | + * | u64 globalId | + * | u16 sigLen + signature bytes | + * | | + * | u32 resourceCount, then per resource: | + * | u64 own globalId | + * | u64 originalResourceId (0 = none) | + * | u64 error (0 = none) | + * | u8 kind index | + * | str type.name | + * | str type.version | + * | u8 flags (hasData, inputsLocked, outputsLocked, | + * | resourceReady, final) | + * | [u32 len + data] only if hasData | + * | u32 fieldCount, then per field: | + * | str name | + * | u8 field type index | + * | u8 field status index | + * | u64 value (0 = none) | + * | u64 error (0 = none) | + * | u8 valueIsFinal | + * | u32 kvCount, then per entry: | + * | str key | + * | u32 len + value bytes | + * +-- trailer (never compressed) ----------------------------+ + * | u32 payload length, as stored | + * | u32 crc32 of the payload, as stored | + * +----------------------------------------------------------+ + * ``` + * + * Notes on the encoding: + * + * - `str` is a u32 length followed by UTF-8. + * - A {@link SignedResourceId} is the string `"|"`. Bodies + * store only the global id; the signature comes from the side table. Global id 0 stands + * for "no reference". + * - `kind`, field type and field status are stored as indices into {@link KINDS}, + * {@link FIELD_TYPES} and {@link FIELD_STATUSES}. Those orderings are part of the format: + * append only, never reorder. + * - The payload length in the trailer, not the file size, delimits the payload. + * - The witness is the root's signature at write time, and is outside the compressed section + * so it can be read without inflating the payload ({@link readPersistedTreeHeader}). + * - Reference counts, resource and data versions, change sources and the derived final state + * are not stored. They are rebuilt on restore. The backend's `final` flag is stored, as + * part of the body. + */ + import type { FieldData, FieldStatus, diff --git a/lib/node/pl-tree/src/synchronized_tree.ts b/lib/node/pl-tree/src/synchronized_tree.ts index 6833b70794..bb63f0c56b 100644 --- a/lib/node/pl-tree/src/synchronized_tree.ts +++ b/lib/node/pl-tree/src/synchronized_tree.ts @@ -3,6 +3,7 @@ import { PlTreeEntry, PlTreeRootsEntry } from "./accessors"; import type { FinalResourceDataPredicate, PlClient, + ResourceSignature, ResourceType, SignedResourceId, TxOps, @@ -17,6 +18,8 @@ import type { ExtendedResourceData } from "./state"; import { PlTreeState, TreeStateUpdateError } from "./state"; import type { PruningFunction, TraversalMode, TreeLoadingStat } from "./sync"; import { constructTreeLoadingRequest, initialTreeLoadingStat, loadTreeState } from "./sync"; +import type { PersistedTree } from "./persisted_tree"; +import { captureTreeState, restoreTreeState } from "./persisted_tree"; import * as tp from "node:timers/promises"; import type { MiLogger } from "@milaboratories/ts-helpers"; @@ -74,6 +77,16 @@ export type SynchronizedTreeOps = { /** Controls which tree-loading path to use. Default `"auto"`. */ traversalMode?: TraversalMode; + + /** A previously persisted mirror to seed the tree with, before its first refresh, so that + * refresh transfers only what changed while the tree was gone. + * + * Advisory: a snapshot that cannot be applied, or does not belong to this tree, is logged + * and dropped, leaving an ordinary cold open. The caller is responsible for having + * established that the snapshot's signatures are still live (see + * {@link PersistedTree.witness}); this option does not check that. Ignored for trees with + * shared-type seeds, which rediscover their roots anyway. */ + restoreFrom?: PersistedTree; }; /** An explicit resource to serve as a tree root. Several explicit seeds may be passed. */ @@ -174,6 +187,11 @@ export class SynchronizedTreeState { /** Roots discovered for shared-type seeds on the last discovery poll. */ private discoveredRoots: SignedResourceId[] = []; + /** Bumped once per refresh cycle that brought something new: a resource appeared, changed, + * or became final. Lets a holder tell whether the tree has moved since it last persisted + * it, without diffing state. Read through {@link changeGeneration}. */ + private changeGenerationCounter = 0; + private constructor( private readonly pl: PlClient, seeds: TreeSeed[], @@ -218,6 +236,50 @@ export class SynchronizedTreeState { return new Set([...this.explicitRoots, ...this.discoveredRoots]); } + /** How many refresh cycles brought something new. Only ever increases. Equal values at two + * points in time mean nothing was added, changed or settled in between, which is what makes + * a periodic snapshot write skippable on an idle tree. */ + public get changeGeneration(): number { + return this.changeGenerationCounter; + } + + /** Captures the current mirror for persistence. + * + * Must be called before {@link terminate}: terminating invalidates the tree, and capturing + * an invalidated tree is refused rather than silently written. */ + public capture(witness: ResourceSignature): PersistedTree { + if (this.terminated) throw new Error("tree synchronization is terminated"); + return captureTreeState(this.state, witness); + } + + /** Installs a snapshot as this tree's state. Returns false if the snapshot was refused, in + * which case the tree is left as it was and the open proceeds cold. + * + * Only meaningful before the first refresh, which is why it is private and driven from + * {@link init}: replacing the state of a running tree would strand its observers. */ + private restore(snapshot: PersistedTree): boolean { + if (this.sharedSeeds.length > 0) { + this.logger?.warn("ignoring tree snapshot: trees with shared-type seeds are not restored"); + return false; + } + + const roots = this.currentRootSet(); + if (snapshot.roots.length !== roots.size || !snapshot.roots.every((r) => roots.has(r))) { + // A snapshot addressed to a different root is a mis-keyed file, not a stale one. + this.logger?.warn("ignoring tree snapshot: its roots are not this tree's roots"); + return false; + } + + const restored = restoreTreeState(snapshot, this.finalPredicate, { + roots, + logger: this.logger, + }); + if (restored === undefined) return false; + + this.state = restored; + return true; + } + /** Resolves the single root for the backward-compatible single-root accessors, throwing * if the tree does not have exactly one root (guards legacy callers against multi-root). */ private soleRoot(): SignedResourceId { @@ -439,7 +501,9 @@ export class SynchronizedTreeState { // actual tree synchronization await this.refresh(stat); - this.updatePollingInterval(countedChanges(stat) > changesBefore); + const changed = countedChanges(stat) > changesBefore; + if (changed) this.changeGenerationCounter++; + this.updatePollingInterval(changed); // logging stats if we were asked to if (this.logStat && this.logger) @@ -569,7 +633,13 @@ export class SynchronizedTreeState { ) { const tree = new SynchronizedTreeState(pl, normalizeSeeds(seeds), ops, logger); - const stat = ops.logStat ? initialTreeLoadingStat() : undefined; + // Seed from the snapshot before the first refresh, so that refresh is the one that + // transfers only what changed. A refused snapshot leaves an ordinary cold open. + const restored = ops.restoreFrom !== undefined && tree.restore(ops.restoreFrom); + + // Always collected, even when not logging: the initial load's change count is what seeds + // the change generation, so a holder can tell a populated tree from an untouched one. + const stat = initialTreeLoadingStat(); let ok = false; @@ -581,10 +651,14 @@ export class SynchronizedTreeState { }); ok = true; } finally { + if (countedChanges(stat) > 0) tree.changeGenerationCounter++; + // logging stats if we were asked to (even if error occured) - if (stat && logger) + if (ops.logStat && logger) logger.info( - `Tree stat (initial load, ${ok ? "success" : "failure"}): ${JSON.stringify(stat)}`, + `Tree stat (initial load, ${ok ? "success" : "failure"}, ${ + restored ? "restored from snapshot" : "cold" + }): ${JSON.stringify(stat)}`, ); } From 2d539f6a56241b7d9967c99de72f2b99356b4d6f Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Thu, 13 Aug 2026 17:26:24 +0200 Subject: [PATCH 3/7] MILAB-6705: acceptance scenarios against a live backend Covers warm reopen, project switching, killed process, open and idle, open and computing, rotated signature, poisoned snapshot and the kill switch. Two fixes the scenarios forced out. The tree now reports whether a snapshot was actually applied, via wasRestoredFromSnapshot, and the middle layer reads that instead of assuming a restore happened whenever init did not throw: a snapshot can be handed over and still be refused, and the old flag would then claim the file on disk described the tree we hold. The store gains a restores counter alongside hits, since a hit only means the bytes were read and a warm reopen needs the tree to have accepted them. MiddleLayer exposes treeSnapshotStats so that is observable at all. Verified the reopen assertions bite by breaking restore and watching them fail. --- .../pl-middle-layer/src/middle_layer/index.ts | 1 + .../src/middle_layer/middle_layer.ts | 8 + .../src/middle_layer/project.ts | 24 +- .../tree_snapshot_scenarios.test.ts | 309 ++++++++++++++++++ .../src/middle_layer/tree_snapshot_store.ts | Bin 11949 -> 12508 bytes lib/node/pl-tree/src/synchronized_tree.ts | 13 + 6 files changed, 346 insertions(+), 9 deletions(-) create mode 100644 lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts diff --git a/lib/node/pl-middle-layer/src/middle_layer/index.ts b/lib/node/pl-middle-layer/src/middle_layer/index.ts index b06e189d56..273d3e8480 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/index.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/index.ts @@ -2,5 +2,6 @@ export { MiddleLayer } from "./middle_layer"; export { Project } from "./project"; export * from "./driver_kit"; export * from "./ops"; +export type { TreeSnapshotMiss, TreeSnapshotStat } from "./tree_snapshot_store"; export { ProjectsField } from "./project_list"; export type { OutgoingShare, PendingShare } from "./sharing_list"; diff --git a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts index ccbc537982..09eaa0f9d2 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts @@ -94,6 +94,7 @@ import type { Dispatcher } from "undici"; import { RetryAgent } from "undici"; import { getDebugFlags } from "../debug"; import { ProjectHelper } from "../model/project_helper"; +import type { TreeSnapshotStat } from "./tree_snapshot_store"; import { TreeSnapshotStore } from "./tree_snapshot_store"; export interface MiddleLayerEnvironment { @@ -980,6 +981,13 @@ export class MiddleLayer { return this.openedProjects.has(id); } + /** Counters for the persisted project tree mirrors, or undefined when they are switched off. + * Reads and hits are what show whether a reopen was actually warm, and the miss breakdown + * says why it was not. */ + public get treeSnapshotStats(): Readonly | undefined { + return this.env.treeSnapshots?.stats; + } + /** * Deallocates all runtime resources consumed by this object and awaits * actual termination of event loops and other processes associated with diff --git a/lib/node/pl-middle-layer/src/middle_layer/project.ts b/lib/node/pl-middle-layer/src/middle_layer/project.ts index cb1b152e1b..d72cdbf160 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/project.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/project.ts @@ -875,15 +875,21 @@ async function loadProjectTree( } try { - return { - tree: await SynchronizedTreeState.init( - env.pl, - rid, - { ...treeOps, restoreFrom: snapshot.tree }, - env.logger, - ), - restored: true, - }; + const tree = await SynchronizedTreeState.init( + env.pl, + rid, + { ...treeOps, restoreFrom: snapshot.tree }, + env.logger, + ); + + // Read from the tree rather than assumed: a snapshot can be handed over and still be + // refused, in which case this open was cold and the file on disk does not describe the + // tree we now hold. + const restored = tree.wasRestoredFromSnapshot; + if (restored) store.noteRestored(); + else env.logger.info("project tree opening cold: the snapshot was not applied"); + + return { tree, restored }; } catch (e: unknown) { if (!isSnapshotFailsafeError(e)) throw e; diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts new file mode 100644 index 0000000000..452bc8c026 --- /dev/null +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts @@ -0,0 +1,309 @@ +import { describe, expect, test } from "vitest"; +import { TestHelpers } from "@milaboratories/pl-client"; +import { randomUUID } from "node:crypto"; +import fsp from "node:fs/promises"; +import path from "node:path"; +import * as tp from "node:timers/promises"; +import { MiddleLayer } from "./middle_layer"; +import type { ProjectId } from "../model/project_model"; +import type { TreeSnapshotOps } from "./ops"; + +/** + * The acceptance scenarios that need a live backend. Each one runs several middle layers in + * turn against one backend root and one work folder, which is what makes a reopen a reopen: + * the projects are the same projects and the snapshot directory is the same directory. + * + * `MiddleLayer.close()` closes the client it was given, so every middle layer here gets its + * own client. They share a session, because the test client reuses one cached token, and a + * shared session is exactly what a warm reopen needs. + */ + +const WORK_ROOT = path.resolve(import.meta.dirname, "..", "..", "work"); + +/** Short intervals so the periodic write is observable inside a test rather than in five + * minutes. Everything else is left at its default. */ +function fastSnapshots(overrides: Partial = {}): TreeSnapshotOps { + return { + enabled: true, + writeInterval: 250, + maxSizeBytes: 256 * 1024 * 1024, + ...overrides, + }; +} + +type Scenario = { + /** Opens another middle layer, on its own client, over the same root and work folder. */ + open: (treeSnapshotOps?: TreeSnapshotOps) => Promise; + /** Closes one, so it is not closed twice during cleanup. */ + close: (ml: MiddleLayer) => Promise; + /** Creates a project, opens it, and lets its tree settle so there is a mirror worth writing. + * Tracked for cleanup. */ + project: (ml: MiddleLayer, label: string) => Promise; + /** The shared snapshot directory. */ + snapshotDir: string; +}; + +/** + * Each middle layer gets its own client, because `MiddleLayer.close()` closes the client it + * was given, and a reopen has to survive that. + * + * The clients use the caller's own root rather than a temporary one: `PlClient.init` with an + * `alternativeRoot` name always creates a fresh ephemeral root and overwrites the field, so a + * second client asking for the same name gets an empty project list, which is precisely the + * state a reopen must not start from. The projects created here are deleted afterwards. + */ +async function withScenario(body: (scenario: Scenario) => Promise): Promise { + const workFolder = path.resolve(WORK_ROOT, randomUUID()); + const live = new Set(); + const projects = new Set(); + + const openMl = async (treeSnapshotOps: TreeSnapshotOps) => { + const client = await TestHelpers.getTestClient(); + const ml = await MiddleLayer.init(client, workFolder, { + defaultTreeOptions: { pollingInterval: 250, stopPollingDelay: 500 }, + devBlockUpdateRecheckInterval: 300, + projectRefreshInterval: 250, + localSecret: MiddleLayer.generateLocalSecret(), + localProjections: [], + openFileDialogCallback: () => { + throw new Error("Not implemented."); + }, + treeSnapshotOps, + }); + live.add(ml); + return ml; + }; + + const scenario: Scenario = { + snapshotDir: path.join(workFolder, "treeSnapshots"), + open: async (treeSnapshotOps = fastSnapshots()) => await openMl(treeSnapshotOps), + close: async (ml: MiddleLayer) => { + live.delete(ml); + await ml.close(); + }, + project: async (ml: MiddleLayer, label: string) => { + const id = await ml.createProject({ label: `${label} ${randomUUID()}` }); + projects.add(id); + await ml.openProject(id); + // Reading the overview forces the tree to load and the computables to resolve. + await ml.getOpenedProject(id).overview.awaitStableValue(); + return id; + }, + }; + + try { + await body(scenario); + } finally { + for (const ml of live) await ml.close().catch(() => {}); + + // The root outlives the test, so the projects have to be cleaned up explicitly. + if (projects.size > 0) { + const cleanup = await openMl({ ...fastSnapshots(), enabled: false }); + try { + for (const id of projects) await cleanup.deleteProject(id).catch(() => {}); + } finally { + await cleanup.close().catch(() => {}); + } + } + await fsp.rm(workFolder, { recursive: true, force: true }); + } +} + +async function snapshotFiles(dir: string): Promise { + try { + return (await fsp.readdir(dir)).sort(); + } catch { + return []; + } +} + +describe("reopening a project", () => { + test("the close write makes the next open warm", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(); + const id = await project(first, "warm reopen"); + await first.closeProject(id); + + // One file for the one project, written at the close boundary. + expect(await snapshotFiles(snapshotDir)).toHaveLength(1); + expect(first.treeSnapshotStats?.writes).toBeGreaterThanOrEqual(1); + await close(first); + + const second = await open(); + await second.openProject(id); + // The claim: the reopen read the snapshot and restored from it. + expect(second.treeSnapshotStats?.hits).toBe(1); + expect(second.treeSnapshotStats?.misses.absent).toBe(0); + // Read is not enough: this is the tree actually accepting the mirror. + expect(second.treeSnapshotStats?.restores).toBe(1); + + // And the project is genuinely usable, not merely restored. + const overview = await second.getOpenedProject(id).overview.awaitStableValue(); + expect(overview.meta.label).toContain("warm reopen"); + await close(second); + }); + }); + + test("project switching: both returns hit", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(); + const a = await project(first, "A"); + const b = await project(first, "B"); + await first.closeProject(a); + await first.closeProject(b); + expect(await snapshotFiles(snapshotDir)).toHaveLength(2); + await close(first); + + const second = await open(); + await second.openProject(a); + await second.openProject(b); + expect(second.treeSnapshotStats?.hits).toBe(2); + expect(second.treeSnapshotStats?.restores).toBe(2); + await close(second); + }); + }); + + test("a killed process is covered by the periodic write", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(fastSnapshots({ writeInterval: 250 })); + const id = await project(first, "killed"); + + // Never closed, standing in for a reboot, a lost connection or a kill. The periodic + // write on the maintenance loop is the only thing that can have saved this. + await tp.setTimeout(1500); + expect(first.treeSnapshotStats?.writes).toBeGreaterThanOrEqual(1); + expect(await snapshotFiles(snapshotDir)).toHaveLength(1); + + // Closing the middle layer without closing the project: close() must not snapshot, so + // whatever is on disk came from the periodic write. + const writesBefore = first.treeSnapshotStats!.writes; + await close(first); + expect(first.treeSnapshotStats?.writes).toBe(writesBefore); + + const second = await open(); + await second.openProject(id); + expect(second.treeSnapshotStats?.hits).toBe(1); + expect(second.treeSnapshotStats?.restores).toBe(1); + await close(second); + }); + }); +}); + +describe("write cadence", () => { + test("open and idle writes once, then goes quiet", async () => { + await withScenario(async ({ open, close, project }) => { + const ml = await open(fastSnapshots({ writeInterval: 250 })); + await project(ml, "idle"); + + // Several intervals of nothing happening. + await tp.setTimeout(2000); + const writes = ml.treeSnapshotStats!.writes; + + // At most one: the change gate stops the loop rewriting a mirror that has not moved. A + // cold open writes exactly once; a project whose tree never settled writes zero. + expect(writes).toBeLessThanOrEqual(1); + + await tp.setTimeout(1500); + expect(ml.treeSnapshotStats?.writes).toBe(writes); + await close(ml); + }); + }); + + test("a project that keeps changing writes at most once per interval", async () => { + await withScenario(async ({ open, close, project }) => { + const ml = await open(fastSnapshots({ writeInterval: 1000 })); + const id = await project(ml, "changing"); + + // Keep the tree moving for roughly three intervals. + const until = Date.now() + 3000; + let n = 0; + while (Date.now() < until) { + await ml.setProjectMeta(id, { label: `changing ${n++}` }); + await tp.setTimeout(150); + } + + // Bounded by wall clock, not by how often the tree changed. + expect(ml.treeSnapshotStats!.writes).toBeLessThanOrEqual(4); + expect(n).toBeGreaterThan(4); + await close(ml); + }); + }); +}); + +describe("when the snapshot cannot be used", () => { + test("a rotated signature is a miss, the file is kept, and the project still opens", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(); + const id = await project(first, "rotated"); + await first.closeProject(id); + await close(first); + + // Rewrite the witness in the header to stand in for a session that has ended. Its + // offset is fixed: magic (4) + schema (2) + flags (2), then a u16 length and the bytes. + const [name] = await snapshotFiles(snapshotDir); + const file = path.join(snapshotDir, name); + const bytes = await fsp.readFile(file); + const witnessLength = bytes.readUInt16LE(8); + expect(witnessLength).toBeGreaterThan(0); + bytes[10] = bytes[10] ^ 0xff; + await fsp.writeFile(file, bytes); + + const second = await open(); + await second.openProject(id); + expect(second.treeSnapshotStats?.hits).toBe(0); + expect(second.treeSnapshotStats?.restores).toBe(0); + expect(second.treeSnapshotStats?.misses["session-rotated"]).toBe(1); + + // Kept: the bodies are still good, only the signatures died. + expect(await snapshotFiles(snapshotDir)).toHaveLength(1); + + const overview = await second.getOpenedProject(id).overview.awaitStableValue(); + expect(overview.meta.label).toContain("rotated"); + await close(second); + }); + }); + + test("a truncated snapshot opens cold without raising", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(); + const id = await project(first, "poisoned"); + await first.closeProject(id); + await close(first); + + const [name] = await snapshotFiles(snapshotDir); + const file = path.join(snapshotDir, name); + const bytes = await fsp.readFile(file); + await fsp.writeFile(file, bytes.subarray(0, Math.floor(bytes.length / 2))); + + const second = await open(); + await second.openProject(id); + expect(second.treeSnapshotStats?.hits).toBe(0); + expect(second.treeSnapshotStats?.restores).toBe(0); + + const overview = await second.getOpenedProject(id).overview.awaitStableValue(); + expect(overview.meta.label).toContain("poisoned"); + await close(second); + }); + }); +}); + +describe("the kill switch", () => { + test("nothing is read or written when snapshots are off", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const first = await open(fastSnapshots({ enabled: false })); + const id = await project(first, "disabled"); + await tp.setTimeout(1000); + await first.closeProject(id); + + expect(first.treeSnapshotStats).toBeUndefined(); + expect(await snapshotFiles(snapshotDir)).toStrictEqual([]); + await close(first); + + // And a project created while off still opens once it is back on. + const second = await open(); + await second.openProject(id); + expect(second.treeSnapshotStats?.misses.absent).toBe(1); + await close(second); + }); + }); +}); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts index d3f6291d8dcf4712eff07aec61000ecf063c0c09..3dc246ee852df4da2963bb67ded4b41e1628a64c 100644 GIT binary patch delta 566 zcmXw$!D`z;6h*;c3|=&J(M_RncWoV`?A=ZB11*GZyBLjL&|rB+=siV<5d0DOj1UO* zck}}S`Iz3ZWe`Fm&D?w5(aWEA_qv%sz0E)DJ8>1X_g+-*`lcBR+@KDIpoqyZios!U z-43ahxPjFI>qOvPgj#4CbaK}#-=-yi72w*o84+xap$@*zQA6zv57$kITW(iNygkCF zi$0nbSY7cbTlMMR{PW&<>m5;!K|v}wB||R#st7Sya1E=SNrB%r(|KjA*zZc|_{c7G zBM(hLon&@SK@d(d^_@93)4?CyeYINQ@N*t#i|Om(@8Z{|J#NLPt|Ce~9;-b0Se2p| z4vK9|!-?Uj#!+W$+?}a3x>oCCnU$+NdUi*-6P*birpO#c8mO3owTl&3_v|P~+l#Kj zH`FMM%p=N(l1w|9%$yxemNMToX4{ud@cC!T?a6%g?fVAzDO7mK2uurFWS_-Z$>n91 P@whx0SsrFzo Date: Thu, 13 Aug 2026 17:57:49 +0200 Subject: [PATCH 4/7] MILAB-6705: fixes from code review Correctness: - The fail-safe now retries cold on any failure of a warm open, not only on the three classified ones. Rethrowing left the snapshot on disk, so the next open restored it and failed identically: a project that never opens again until someone deletes the cache directory, which is the outcome the fail-safe exists to prevent. Deletion stays reserved for failures that implicate the snapshot, so a timeout no longer destroys a good mirror. The classifier walks the cause chain and is now tested. - A failed write was recorded as a successful persist, so both triggers believed the tree was on disk and skipped it forever, close write included. One transient I/O error cost the whole session. write() now reports success. - captureTreeState copied nothing: it handed out the tree's live field objects, which the next update mutates in place. Any await between capture and encode produced a corpus whose fields point at resources it does not carry, i.e. an unrestorable snapshot. Only the synchronous call path made this safe. - The signature side table silently collapsed one global id carrying two signatures, rewriting one resource's references to another's. Unreachable for single-root trees; now refused rather than silently wrong. - changeGeneration missed dynamic-field removals, so a poll that dropped a field and collected its subtree read as idle and skipped the write. - purge() was an unguarded recursive delete of a caller-supplied path. It now removes only this class's files. - The cache key carries the backend instanceId, so a database reset at a fixed address is a miss rather than a hit against reused global ids. Also: eviction no longer stops early on an undeletable file; reads distinguish unreadable from absent; a hit touches the file so the size trim orders by use rather than by write; decode copies byte slices instead of pinning the whole payload; the in-flight guard loops; the first periodic write lands on the first maintenance pass rather than one interval in, so sessions shorter than the interval are covered; and the sources-mode stamp is stable, so running from sources exercises restore instead of guaranteeing a miss. Two build-stamp comments were wrong: the finality predicate lives in pl-client, not here, and release builds always take the dirty path because CI writes version bumps before building. --- lib/node/pl-middle-layer/build.node.config.js | 12 ++- .../src/middle_layer/build_stamp.ts | 31 ++++--- .../src/middle_layer/middle_layer.ts | 14 ++- .../src/middle_layer/project.ts | 73 ++++++++++----- .../src/middle_layer/project_failsafe.test.ts | 47 ++++++++++ .../tree_snapshot_scenarios.test.ts | 17 ++-- .../middle_layer/tree_snapshot_store.test.ts | 84 +++++++++++++++++- .../src/middle_layer/tree_snapshot_store.ts | Bin 12508 -> 16502 bytes lib/node/pl-tree/src/persisted_tree.ts | 39 +++++++- lib/node/pl-tree/src/synchronized_tree.ts | 24 +++-- 10 files changed, 282 insertions(+), 59 deletions(-) create mode 100644 lib/node/pl-middle-layer/src/middle_layer/project_failsafe.test.ts diff --git a/lib/node/pl-middle-layer/build.node.config.js b/lib/node/pl-middle-layer/build.node.config.js index fd5c8f7158..65a37a9d9e 100644 --- a/lib/node/pl-middle-layer/build.node.config.js +++ b/lib/node/pl-middle-layer/build.node.config.js @@ -4,10 +4,14 @@ import { execFileSync } from "node:child_process"; /** * Identifies this build for the persisted-tree cache key, see `src/middle_layer/build_stamp.ts`. * - * A clean worktree stamps its commit, so every build of a given release shares a stamp and a - * reopen stays warm across restarts. A dirty worktree stamps the build time too, so editing - * the tree pruning or finality rules locally cannot hit a snapshot written under the old - * ones. `git status` covers the whole repo, which over-invalidates rather than under-. + * A clean worktree stamps its commit; a dirty one stamps the build time too, so editing the + * tree pruning or traversal rules locally cannot hit a snapshot written under the old ones. + * `git status` covers the whole repo, which over-invalidates rather than under-. + * + * Note that release builds take the dirty path as well: CI runs `version-packages` before + * building and commits the bump afterwards, so the worktree always carries the version edits at + * build time. Harmless, because each published artifact bakes in one stamp and only has to be + * stable within itself, but it does mean a rebuild of identical code produces a different one. */ function buildStamp() { const git = (args) => diff --git a/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts b/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts index d1c6ac2ce9..af6d0365d6 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/build_stamp.ts @@ -1,5 +1,3 @@ -import { randomUUID } from "node:crypto"; - /** Injected by rolldown at build time, see `build.node.config.js`. Absent when the package is * consumed straight from sources (`USE_SOURCES=1`), because no build step runs then. */ declare const __PL_ML_BUILD_STAMP__: string | undefined; @@ -15,17 +13,24 @@ function injectedStamp(): string | undefined { } /** - * Identifies the build of this package, and through it every rule that shapes what a - * persisted tree mirror contains: the pruning function, the field filter, the traversal stop - * rules and the finality predicate all live in this package or in one it pins. + * Identifies the build of this package, and through it the rules that shape what a persisted + * tree mirror contains: the pruning function, the field filter and the traversal stop rules, + * all of which live in this package. (The finality predicate comes from pl-client and is NOT + * covered, which is harmless: finality is recomputed on restore, so it is the one rule that + * cannot poison a stored file.) * - * Used as a cache-key component, so any change to those rules invalidates every snapshot, - * costing one cold open. A build from a clean worktree stamps its commit, so released builds - * share a stamp and reopens stay warm across restarts. A build from a dirty worktree stamps - * the build time as well, so editing those rules locally can never hit a snapshot written - * under the old ones. + * Used as a cache-key component, so a change to those rules invalidates every snapshot, costing + * one cold open. Each built artifact bakes in one stamp, so reopens stay warm across restarts + * of an installed version. A build from a dirty worktree includes the build time, so editing + * those rules locally can never hit a snapshot written under the old ones. In practice release + * builds are dirty too, because CI writes version bumps into the worktree before building; that + * costs nothing, since the stamp only has to be stable within an artifact. * - * With no build at all (sources mode) the value is unique per process, so nothing ever hits. - * That is the safe direction: an unbuilt tree has no way to say which rules were in force. + * With no build at all (sources mode) the value is a constant. That deliberately trades away + * the dirty-worktree guarantee: it means someone running from sources exercises the restore + * path at all, rather than every snapshot being a guaranteed miss for the one audience most + * likely to find its defects. The exposure it reintroduces, editing pruning rules from sources + * and hitting a mirror written under the old ones, is the local-development gap the design + * already accepts, and `treeSnapshots: false` or deleting the directory clears it. */ -export const ML_BUILD_STAMP: string = injectedStamp() ?? `unbuilt-${randomUUID()}`; +export const ML_BUILD_STAMP: string = injectedStamp() ?? "sources"; diff --git a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts index 09eaa0f9d2..33bcf0e056 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts @@ -1116,9 +1116,17 @@ export class MiddleLayer { enabled: ops.treeSnapshotOps.enabled, logger, }); - // Housekeeping before any project opens: drop snapshots from other builds, backends and - // users, then trim to the ceiling. - await treeSnapshots?.evict(); + if (ops.treeSnapshotOps.enabled) { + // Housekeeping before any project opens: drop snapshots from other builds, backends and + // users, then trim to the ceiling. + await treeSnapshots?.evict(); + } else { + // Switched off, so reclaim what earlier sessions left on disk. The reason to reach for + // this switch is usually the disk itself, and leaving the files behind would answer the + // wrong half of that complaint. Keyed on the setting, not on the store being absent: it + // is also absent for an impersonated client, whose session must not delete anything. + await TreeSnapshotStore.purge(ops.treeSnapshotPath, logger); + } const env: MiddleLayerEnvironment = { pl, diff --git a/lib/node/pl-middle-layer/src/middle_layer/project.ts b/lib/node/pl-middle-layer/src/middle_layer/project.ts index d72cdbf160..c41d78854e 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/project.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/project.ts @@ -115,10 +115,15 @@ export class Project { * that has not moved, which is what makes a project left open and idle go quiet. */ private snapshotGeneration: number; - /** When the last snapshot was written, for the periodic write's wall-clock gate. Seeded at - * construction so the first write lands one interval after the project opens rather than - * immediately. */ - private lastSnapshotAt = Date.now(); + /** When a snapshot was last attempted, for the periodic write's wall-clock gate. + * + * Zero, not the construction time, so the first write lands on the first maintenance pass + * after the tree has settled rather than a full interval later. Sessions shorter than one + * interval are the common case for a desktop app that is quit with a project still open, and + * seeding this to now would leave every one of them with nothing on disk. Set on every + * attempt, successful or not, so a persistently failing write retries at the interval rather + * than on every pass of the loop. */ + private lastSnapshotAt = 0; private get destroyed() { return this.abortController.signal.aborted; @@ -194,7 +199,11 @@ export class Project { /** Serializes writes, and skips one that the in-flight write has already made redundant. */ private async writeSnapshot(store: TreeSnapshotStore, generation: number): Promise { - if (this.snapshotInFlight !== undefined) { + // A loop, not a single check: with three or more callers, re-checking only once would let + // a waiter install its own promise over another's and clear the field while that write is + // still running. Two callers is the most that can happen today, so this is a guard against + // the next caller rather than a live fix. + while (this.snapshotInFlight !== undefined) { await this.snapshotInFlight; if (this.snapshotGeneration >= generation) return; } @@ -205,15 +214,20 @@ export class Project { await this.snapshotInFlight; } - /** Captures and writes, never throwing: a snapshot is an optimisation and must not delay or - * fail whatever triggered it. */ + /** Captures and writes, never throwing: a snapshot is an optimisation and must not fail + * whatever triggered it. */ private async captureAndWrite(store: TreeSnapshotStore, generation: number): Promise { + // Recorded before the attempt and regardless of its outcome, so a failing disk is retried + // once per interval instead of on every pass of the maintenance loop. + this.lastSnapshotAt = Date.now(); try { // The root's signature is the session witness a later open compares against. const snapshot = this.projectTree.capture(parseSignedResourceId(this.rid).signature); - await store.write(this.rid, snapshot); - this.snapshotGeneration = generation; - this.lastSnapshotAt = Date.now(); + + // Only a real write advances the change gate. Marking the generation persisted after a + // failed write would tell both triggers the tree is already on disk, so one transient + // I/O error would cost the rest of the session, close write included. + if (await store.write(this.rid, snapshot)) this.snapshotGeneration = generation; } catch (e: unknown) { this.env.logger.warn( new Error(`failed to capture tree snapshot for project ${this.id}`, { cause: e }), @@ -891,23 +905,40 @@ async function loadProjectTree( return { tree, restored }; } catch (e: unknown) { - if (!isSnapshotFailsafeError(e)) throw e; - + // Retry cold on ANY failure of the warm open, not only on the classified ones. A cold open + // is exactly what this code did before snapshots existed, so the retry cannot regress + // anything, whereas rethrowing here leaves a project that fails to open on every attempt + // until someone deletes the cache directory by hand: the snapshot stays on disk and the + // next open restores it and fails the same way. That is the outcome this fail-safe exists + // to prevent, and the error classes that can reach here are not a closed set. env.logger.warn( - new Error("restored project tree failed its first refresh, discarding it and opening cold", { - cause: e, - }), + new Error("restored project tree failed its first refresh, opening cold", { cause: e }), ); - await store.discard(rid); + + // Deleting is reserved for failures that implicate the snapshot itself. Anything else (a + // timeout, a dropped connection) says nothing about the file, and throwing it away would + // destroy a mirror that is still good, along with the evidence a later signature refresh + // would repair. + if (isSnapshotFailsafeError(e)) await store.discard(rid); + return await cold(); } } -/** The failures that can mean a snapshot no longer matches what the backend will serve, as - * opposed to a client that has genuinely lost its session. Both look the same on one refresh, - * which is why the retry is spent only on the first. */ -function isSnapshotFailsafeError(e: unknown): boolean { - return isUnauthenticated(e) || isPermissionDenied(e) || e instanceof TreeStateUpdateError; +/** The failures that implicate the snapshot rather than the link or the session: a rotated + * master secret, a revoked grant, or state the tree cannot reconcile. Only these delete the + * file; every other failure still falls back to a cold open, it just keeps the file. + * + * The cause chain is walked because a wrapper anywhere between the tree update and here would + * otherwise silently disarm the inconsistency arm. `isUnauthenticated` and `isPermissionDenied` + * do their own one-level unwrapping. */ +export function isSnapshotFailsafeError(e: unknown): boolean { + if (isUnauthenticated(e) || isPermissionDenied(e)) return true; + for (let cause: unknown = e, depth = 0; cause !== undefined && depth < 8; depth++) { + if (cause instanceof TreeStateUpdateError) return true; + cause = (cause as { cause?: unknown } | null)?.cause; + } + return false; } export function projectTreePruning(logger: MiLogger): PruningFunction { diff --git a/lib/node/pl-middle-layer/src/middle_layer/project_failsafe.test.ts b/lib/node/pl-middle-layer/src/middle_layer/project_failsafe.test.ts new file mode 100644 index 0000000000..fa8b4154b0 --- /dev/null +++ b/lib/node/pl-middle-layer/src/middle_layer/project_failsafe.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, test } from "vitest"; +import { TreeStateUpdateError } from "@milaboratories/pl-tree"; +import { PermissionDeniedError, UnauthenticatedError } from "@milaboratories/pl-client"; +import { isSnapshotFailsafeError } from "./project"; + +/** + * Which first-refresh failures implicate the snapshot itself, and so delete it. + * + * Every failure retries cold regardless, so a mistake here cannot leave a project unopenable. + * What it can do is either destroy a good mirror over a transient link problem, or keep a bad + * one and pay a wasted warm attempt on every open. + */ +describe("the fail-safe classification", () => { + test("authentication and permission failures implicate the snapshot", () => { + expect(isSnapshotFailsafeError(new UnauthenticatedError("token expired"))).toBe(true); + expect(isSnapshotFailsafeError(new PermissionDeniedError("grant revoked"))).toBe(true); + }); + + test("a tree inconsistency implicates the snapshot", () => { + expect(isSnapshotFailsafeError(new TreeStateUpdateError("orphan resource"))).toBe(true); + }); + + test("a wrapped tree inconsistency still implicates it", () => { + // The cause chain is walked precisely so that a wrapper introduced anywhere between the + // tree update and the caller cannot silently disarm this arm of the fail-safe. + const wrapped = new Error("refresh failed", { + cause: new Error("while loading", { cause: new TreeStateUpdateError("orphan resource") }), + }); + expect(isSnapshotFailsafeError(wrapped)).toBe(true); + }); + + test("a link failure does not, so the mirror is kept", () => { + expect(isSnapshotFailsafeError(new Error("socket hang up"))).toBe(false); + expect(isSnapshotFailsafeError(new Error("deadline exceeded"))).toBe(false); + }); + + test("nothing exotic throws", () => { + // A cause chain that loops must not hang the classifier. + const looped: { cause?: unknown } = {}; + looped.cause = looped; + + expect(isSnapshotFailsafeError(looped)).toBe(false); + expect(isSnapshotFailsafeError(undefined)).toBe(false); + expect(isSnapshotFailsafeError(null)).toBe(false); + expect(isSnapshotFailsafeError("a string")).toBe(false); + }); +}); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts index 452bc8c026..7d9b3302a9 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts @@ -170,6 +170,12 @@ describe("reopening a project", () => { // Never closed, standing in for a reboot, a lost connection or a kill. The periodic // write on the maintenance loop is the only thing that can have saved this. + // + // What this does NOT reproduce is the relaunch: both middle layers here share a session, + // because the test client reuses one cached token. In production the equivalent is the + // desktop app reconnecting with the JWT it persisted, which keeps the session and so the + // signatures; a change that made relaunch re-login instead would break the warm reopen + // and no assertion here would notice. await tp.setTimeout(1500); expect(first.treeSnapshotStats?.writes).toBeGreaterThanOrEqual(1); expect(await snapshotFiles(snapshotDir)).toHaveLength(1); @@ -197,14 +203,15 @@ describe("write cadence", () => { // Several intervals of nothing happening. await tp.setTimeout(2000); - const writes = ml.treeSnapshotStats!.writes; - // At most one: the change gate stops the loop rewriting a mirror that has not moved. A - // cold open writes exactly once; a project whose tree never settled writes zero. - expect(writes).toBeLessThanOrEqual(1); + // Exactly one, not "at most one": a cold open loads a tree, so the change gate is open + // and the first maintenance pass writes. Asserting <= 1 would pass with zero writes and + // prove nothing about the periodic trigger existing at all. + expect(ml.treeSnapshotStats?.writes).toBe(1); + // And then quiet, because the gate closes on a mirror that has not moved. await tp.setTimeout(1500); - expect(ml.treeSnapshotStats?.writes).toBe(writes); + expect(ml.treeSnapshotStats?.writes).toBe(1); await close(ml); }); }); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts index a205c7f730..f493ee183f 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts @@ -12,10 +12,13 @@ const silent: MiLogger = { info: () => {}, warn: () => {}, error: () => {} }; const sig = (hex: string) => toResourceSignature(Buffer.from(hex, "hex")); -/** Only the three fields the store reads. */ -function fakeClient(ops: { host?: string; user?: string | null; asUser?: string } = {}): PlClient { +/** Only the fields the store reads. */ +function fakeClient( + ops: { host?: string; user?: string | null; asUser?: string; instanceId?: string } = {}, +): PlClient { return { conf: { hostAndPort: ops.host ?? "localhost:6345", asUser: ops.asUser }, + serverInfo: { instanceId: ops.instanceId ?? "instance-1" }, authUser: ops.user === undefined ? "someone@example.com" : ops.user, } as unknown as PlClient; } @@ -68,6 +71,68 @@ describe("when the store should not exist at all", () => { }); }); +describe("purge", () => { + test("removes our files, so turning the switch off reclaims the disk", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + await store.write(rootB, snapshotFor(rootB)); + expect(await files()).toHaveLength(2); + + await TreeSnapshotStore.purge(dir, silent); + await expect(fsp.stat(dir)).rejects.toThrow(); + }); + + test("leaves anything that is not ours, and the directory holding it", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + // The path is caller-supplied, so a misconfigured one must not take a stranger's files + // with it. + await fsp.writeFile(path.join(dir, "someone-elses.txt"), "not ours"); + + await TreeSnapshotStore.purge(dir, silent); + + expect(await files()).toStrictEqual(["someone-elses.txt"]); + }); + + test("is quiet about a directory that is not there", async () => { + await expect( + TreeSnapshotStore.purge(path.join(dir, "never-existed"), silent), + ).resolves.toBeUndefined(); + }); +}); + +describe("reporting failure", () => { + test("a failed write says so, rather than reporting a phantom success", async () => { + // A file where the directory should be, so every write fails. + const occupied = path.join(dir, "occupied"); + await fsp.writeFile(occupied, "in the way"); + const store = storeIn(occupied)!; + + expect(await store.write(rootA, snapshotFor(rootA))).toBe(false); + expect(store.stats.writeFailures).toBe(1); + }); + + test("a successful write says so", async () => { + const store = storeIn()!; + expect(await store.write(rootA, snapshotFor(rootA))).toBe(true); + }); + + test("an unreadable file is not reported as absent", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + + // Replace the file with a directory: present, but unopenable. + const [name] = await files(); + const file = path.join(dir, name); + await fsp.rm(file); + await fsp.mkdir(file); + + const read = await store.read(rootA); + expect(read).toStrictEqual({ ok: false, miss: "unreadable" }); + expect(store.stats.misses.absent).toBe(0); + }); +}); + describe("round trip", () => { test("a written snapshot reads back", async () => { const store = storeIn()!; @@ -169,6 +234,15 @@ describe("the key", () => { const other = storeIn(dir, { client: fakeClient({ user: "other@example.com" }) })!; expect(await other.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); }); + + test("a backend that reset its database does not see the old state's snapshots", async () => { + await storeIn()!.write(rootA, snapshotFor(rootA)); + + // Same address, same user, new instance: global ids are reused after a reset, so the + // address alone would be a hit against a tree that no longer exists. + const reset = storeIn(dir, { client: fakeClient({ instanceId: "instance-2" }) })!; + expect(await reset.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + }); }); describe("eviction", () => { @@ -229,8 +303,10 @@ describe("eviction", () => { const store = storeIn(occupied)!; await expect(store.evict()).resolves.toBeUndefined(); - await expect(store.write(rootA, snapshotFor(rootA))).resolves.toBeUndefined(); + await expect(store.write(rootA, snapshotFor(rootA))).resolves.toBe(false); expect(store.stats.writeFailures).toBe(1); - expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); + // "unreadable", not "absent": the directory is broken rather than empty, and that is the + // distinction someone reading the counters needs. + expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "unreadable" }); }); }); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts index 3dc246ee852df4da2963bb67ded4b41e1628a64c..2a768dfb67213413a1190e25c4bafcce670bb84f 100644 GIT binary patch delta 4039 zcmbVP&u=7071kOHh!a90FHzX6l;_zvnHf!b!T||vmer#5N?tAYvi52tV1=u@Yi26$ zu4+|Pdpy}#0}?010n|uH`~#doM04fHi8H4NkoX63&LH*moRYx+mMdiA~U zeedhPi+}(0%Abbo^V>91nIz@NPo$L;m!1++W+`doDVAgksbxySlgP3iDx}SobINEp zAVIN6DL19|xX`peJS1tYvC#$*^@;8%=ao)8jjYM(M8r;Ne}b@yD4oRm`AAvk!zKuT zZ78tj69Q;tQ2WD4WC1MJddb9 z#EOroQ)+%H;!JjMW7zL6qV<2RZ>(-F)?biouReIVcX$7Nx=9^xrHneeizhFB=gaeB zWE6`8>7~?H+IgW9*&hU?a^#iGNO?yhO|5iphuT;6Q?kYjFC9%MK$Eis+HEX83=M}! z+(|6w5UKDY7ETVZKqV6{k364?#A6Fyncx}6Lc#dbi5y_o+Y-{U2;rnXQ+iCI4Zj4B z-|&)3od&c}h?K^eiA6>#1z3-jpBT_UaZ&56Liy*Q$~jY^*{CZhGl|IB4H{@TXT#|k zd&&u{YM2@zz_iNEne5>OrWq(E=v2;V67~bMd?KU6NwL0-p2_*hlQ%fKXGHvJJujFl zuL1gH!1BV+5Qt*qC*)L$&2hz?txM=9A`pqOFRTRScmqW+rA>gT8f)Q8Tn94arRvw4 zdslSz&&~WLDu8j5S@rlTy-UZ);o#AIz>Y@(l*O53QS*>rcEDTYu^N}AtjepqM?WgE zC^32z0W5!8rw@z;-nhwylXhQ^OmvArw{B6V6P}CGPafeV3b&p_9~h;(ogcr~=~e%@ z`sX=F&58Btp=6jj1F$E_gmdqpQ_$>9zXalFQMd^zvMf?W4@(_rO0IbbMuTS#2%{y_ zI=C5)2ILIT1br-5oPd>fH0Q9VD$AM$gg2M{&d*ZiPMOc667K+}QctY{+cQ~52FXh( z&5Xv}3?a!vpdMVc0RdXD9=D-3nG}KHWx=!vZ>kUHXb!0|CZ6*ak^;i=k|w&sjgivI zqgiW_s&~C3GZjlj=OzXlu?k!XF{Ouuji1>Wk2)bJA4D7MWUTD$01fv#|K!^ubEY{_ zP`&D}ue>p5m4uAROoo_sjJgj%Pj5}#-q75@Ar#C^bsc3@ji9Aisx0tJl@^dk2kx^b zM|9B0i#5}lxk3AiGR{=;jnMikA>n2^p`x^7*+uDh7%LodfQBBAC9c1(?uTF1E892b z=#OXcIXFDVQK8JGb?OWy<@2CtM|6T`DCGID5740~yIk`#~!$M{w zQX@P>#uA#xAS)kL4|><;Vk&^w$Q2RKJuKsD_j>fS*@AdwcMHlSqk3*U#`(K<_wVoh zq#lBJ8K{D-rQH25i!ch^YRS=~yO^6jdi!k(F$oHmWF@HG?L?9@u{SnH{m7r4i5JYjypYYnYFS#PW6Xv+Dag1wJBh_E~R%x%<9GA{^ECE zJG}DpTTd^Q=VI~cYbR^fCtL4S2Vei*3rwhhQ}xGaYaY7mHaCXjvj(Hb?blGhHbQ7d zh%1lCT;hR1xiOX5^$mg<1nFjB04)8(WmF4m!08-jhr_V!>7^*KUuex;ATMDTh&R9W zIe!-l!6vU|e@I?D62&XiL~u`M4lo#8lZ zD+-JE{}r(3 z-fXhO?5}@yXy>wtUASw%Sz~!VHf2zu;BR20STPx!+yD)QK4MCG1(%4Kn4CH&6fo#R z%3xoY7IVrH!pofcFw9FNg5VepIU~su4r7OTYPiOzWITggoHurnWsD6YDDeSIJ6PU) zI`BY(42oGea3a|59u)z=U3GZ@sqm6>9VVIF}OI(ygA{`0dNf zyCWD0x%l)(e|-*ThSouG+2O9aTwe0)JL8xHf)7X}q*~))Pb4aRkd^5~ zMIR+g!1JJ5O?~vAp6lsCFPB#dV?>jRZ{e@n`ZpB~B^eB3_D&>?L_8ts-zbVnizQJ^9HfH~hON$mQLKI9+Kz{&z8)t ({ + ...r, + fields: r.fields.map((f) => ({ + name: f.name, + type: f.type, + status: f.status, + value: f.value, + error: f.error, + valueIsFinal: f.valueIsFinal, + })), + })); + + return { witness, roots: [...state.roots], resources }; } /** @@ -258,6 +277,19 @@ function writePayload(tree: PersistedTree): Buffer { const collect = (id: OptionalSignedResourceId) => { if (!isNotNullSignedResourceId(id)) return; const { globalId, signature } = parseSignedResourceId(id); + + // The tree keys its heap by the whole signed string, so one global id carrying two + // different signatures is two distinct resources to the tree and one entry here. Refusing + // is the only safe answer: last-write-wins would silently rewrite one resource's + // references to point at the other. Unreachable for a single-root tree, where every + // resource is signed under one colour, but a tree with several explicit seeds can legally + // be served the same resource under two colours. + const existing = signatures.get(globalId); + if (existing !== undefined && !Buffer.from(existing).equals(Buffer.from(signature))) + throw new Error( + `cannot persist a tree holding global id ${globalId} under two different signatures`, + ); + signatures.set(globalId, signature); }; @@ -636,10 +668,13 @@ class Reader { return this.view.getBigUint64(this.take(8), true); } + /** Copies rather than returning a view. A view would keep the whole inflated payload alive + * for as long as the restored tree holds any one byte payload, including all the structure + * bytes it will never read again. The copy is transient; the retention would not be. */ bytes(): Uint8Array { const length = this.u32(); const at = this.take(length); - return this.src.subarray(at, at + length); + return Uint8Array.prototype.slice.call(this.src, at, at + length); } shortBytes(): Uint8Array { diff --git a/lib/node/pl-tree/src/synchronized_tree.ts b/lib/node/pl-tree/src/synchronized_tree.ts index 97b15db3a0..30e9b5f0fe 100644 --- a/lib/node/pl-tree/src/synchronized_tree.ts +++ b/lib/node/pl-tree/src/synchronized_tree.ts @@ -81,11 +81,14 @@ export type SynchronizedTreeOps = { /** A previously persisted mirror to seed the tree with, before its first refresh, so that * refresh transfers only what changed while the tree was gone. * - * Advisory: a snapshot that cannot be applied, or does not belong to this tree, is logged - * and dropped, leaving an ordinary cold open. The caller is responsible for having - * established that the snapshot's signatures are still live (see - * {@link PersistedTree.witness}); this option does not check that. Ignored for trees with - * shared-type seeds, which rediscover their roots anyway. */ + * A snapshot that cannot be applied, or does not belong to this tree, is logged and dropped, + * leaving an ordinary cold open. + * + * A snapshot that applies but whose ids are dead is NOT handled here: its resources become + * this tree's seeds, so the first refresh fails and {@link init} rejects, where a cold open + * would have succeeded. Establishing that the signatures are still live is the caller's job + * (see {@link PersistedTree.witness}), as is deciding what to do when the first refresh is + * refused anyway. Ignored for trees with shared-type seeds, which rediscover their roots. */ restoreFrom?: PersistedTree; }; @@ -129,7 +132,13 @@ const DISCOVERY_INTERVAL_MS = 3_000; * `resourcesUnchanged` is excluded by design, since a cycle that only re-fetched unchanged * state is exactly the idle case the backoff exists for. */ function countedChanges(stat: TreeLoadingStat): number { - return stat.resourcesNew + stat.resourcesChanged + stat.resourcesMarkedFinal; + // `fieldsRemoved` is included despite being a per-field count, because it is the one change + // that never shows up in `resourcesChanged`: the removed-dynamic-field branch in + // `updateFromResourceData` does not set its `changed` flag, so a cycle that only dropped a + // field (and garbage-collected whatever it pointed at) otherwise reads as an idle cycle. + // That double-counts a resource that both changed and lost a field, which is harmless here: + // every caller compares this against an earlier value rather than reading it as a total. + return stat.resourcesNew + stat.resourcesChanged + stat.resourcesMarkedFinal + stat.fieldsRemoved; } /** The poll-cadence policy, as a pure function of the last cycle's outcome. @@ -276,7 +285,8 @@ export class SynchronizedTreeState { } const roots = this.currentRootSet(); - if (snapshot.roots.length !== roots.size || !snapshot.roots.every((r) => roots.has(r))) { + const snapshotRoots = new Set(snapshot.roots); + if (snapshotRoots.size !== roots.size || ![...snapshotRoots].every((r) => roots.has(r))) { // A snapshot addressed to a different root is a mis-keyed file, not a stale one. this.logger?.warn("ignoring tree snapshot: its roots are not this tree's roots"); return false; From 03f9d56c1aeb0bfd11f1025031a78c56f67d729c Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Fri, 14 Aug 2026 10:52:26 +0200 Subject: [PATCH 5/7] MILAB-6705: do not block a project close on its snapshot write closeProject captured and encoded up to ten megabytes before returning, so project switching sat behind the write. Only the capture needs the tree alive, and a capture is a copy rather than a view, so the encode and write can run after the tree is gone. closeProject now starts the write and returns. MiddleLayer keeps the promise so close() can drain it, bounded at five seconds: quitting still starts no snapshot work of its own, it only lets one already in flight finish, and a wedged filesystem cannot hold the quit open. The close write is queued behind any in-flight periodic write rather than racing it, so the same mirror is not encoded twice. Measured first: moving the encode to the middle layer's worker thread was the obvious alternative and is a bad trade, because structured-cloning the snapshot across the boundary costs 32ms of the 54ms it would save on the reference project's shape. --- .../src/middle_layer/middle_layer.ts | 42 +++++++++++++++- .../src/middle_layer/project.ts | 49 +++++++++++++++---- .../tree_snapshot_scenarios.test.ts | 25 +++++++++- 3 files changed, 104 insertions(+), 12 deletions(-) diff --git a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts index 33bcf0e056..91e3c70e11 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts @@ -97,6 +97,11 @@ import { ProjectHelper } from "../model/project_helper"; import type { TreeSnapshotStat } from "./tree_snapshot_store"; import { TreeSnapshotStore } from "./tree_snapshot_store"; +/** How long shutdown waits for close-boundary snapshot writes that are already running. Long + * enough for a ten-megabyte encode and write on ordinary storage, short enough that a wedged + * filesystem does not hold the quit open. */ +const SNAPSHOT_DRAIN_TIMEOUT_MS = 5_000; + export interface MiddleLayerEnvironment { dispose(): Promise; readonly pl: PlClient; @@ -946,6 +951,36 @@ export class MiddleLayer { private readonly openedProjects = new Map(); + /** Snapshot writes started by {@link closeProject} and not yet finished. Held only so + * {@link close} can give them a bounded chance to land. */ + private readonly pendingSnapshotWrites = new Set>(); + + private trackSnapshotWrite(write: Promise): void { + this.pendingSnapshotWrites.add(write); + void write.finally(() => this.pendingSnapshotWrites.delete(write)); + } + + /** Waits for close-boundary snapshot writes that are already running, up to `timeoutMs`. + * + * This starts no work: quitting still performs no snapshot of its own. It only lets a write + * that a project close already began finish, so closing a project and immediately quitting + * does not routinely lose it. Bounded, because a wedged filesystem must not hang the quit, + * and losing the write costs one cold open rather than any correctness. */ + private async drainSnapshotWrites(timeoutMs: number): Promise { + if (this.pendingSnapshotWrites.size === 0) return; + + let timer: NodeJS.Timeout | undefined; + const expiry = new Promise((resolve) => { + timer = setTimeout(resolve, timeoutMs); + timer.unref?.(); + }); + try { + await Promise.race([Promise.allSettled(this.pendingSnapshotWrites), expiry]); + } finally { + if (timer !== undefined) clearTimeout(timer); + } + } + /** Opens a project, and starts corresponding project maintenance loop. */ public async openProject(id: ProjectId): Promise { if (this.openedProjects.has(id)) throw new Error(`Project ${id} already opened`); @@ -963,7 +998,11 @@ export class MiddleLayer { // Snapshot before destroy, and here rather than inside destroy(): destroy() is also what // application shutdown runs, and quitting should perform no snapshot work. Terminating the // tree invalidates it, so the state has to be taken first either way. - await prj.writeSnapshotOnClose(); + // + // Started, not awaited. The capture happens synchronously inside, which is the part that + // needs the tree alive; the encode and write are up to ten megabytes of work that closing a + // project should not sit behind. Kept so shutdown can drain it. + this.trackSnapshotWrite(prj.snapshotOnClose()); await prj.destroy(); this.openedProjectsList.setValue([...this.openedProjects.keys()]); @@ -1003,6 +1042,7 @@ export class MiddleLayer { this.sharingStateTree.terminate(), this.pendingSharesTree.terminate(), ]); + await this.drainSnapshotWrites(SNAPSHOT_DRAIN_TIMEOUT_MS); await this.env.dispose(); await this.pl.close(); } diff --git a/lib/node/pl-middle-layer/src/middle_layer/project.ts b/lib/node/pl-middle-layer/src/middle_layer/project.ts index c41d78854e..b79e3e54af 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/project.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/project.ts @@ -175,21 +175,52 @@ export class Project { } /** - * Snapshot write at the close boundary, on top of the periodic one, since closing is a - * natural point to persist. Change-gated but not interval-gated: rewriting a mirror that has - * not moved is pure waste, but a mirror that has moved is worth keeping however recently the - * last write happened. + * Starts the close-boundary snapshot and returns without waiting for the write. * - * Must run before {@link destroy}, which terminates the tree and thereby invalidates it. + * On top of the periodic write, since closing is a natural point to persist. Change-gated but + * not interval-gated: rewriting a mirror that has not moved is pure waste, but a mirror that + * has moved is worth keeping however recently the last write happened. + * + * The **capture is synchronous and happens here**, before the caller destroys the tree, + * because destroying it invalidates it and a later capture would be refused. Only the encode + * and the write are deferred: they are up to ten megabytes of work, and project switching + * should not wait for them. Deferring is safe only because a capture is a copy rather than a + * view of the tree. + * + * The returned promise never rejects. The caller is expected to keep it so it can be drained + * at shutdown, not to await it here. */ - public async writeSnapshotOnClose(): Promise { + public snapshotOnClose(): Promise { const store = this.env.treeSnapshots; - if (store === undefined) return; + if (store === undefined) return Promise.resolve(); const generation = this.projectTree.changeGeneration; - if (generation === this.snapshotGeneration) return; + if (generation === this.snapshotGeneration) return Promise.resolve(); - await this.writeSnapshot(store, generation); + let snapshot; + try { + snapshot = this.projectTree.capture(parseSignedResourceId(this.rid).signature); + } catch (e: unknown) { + this.env.logger.warn( + new Error(`failed to capture tree snapshot for project ${this.id} on close`, { cause: e }), + ); + return Promise.resolve(); + } + + this.lastSnapshotAt = Date.now(); + + // Queued behind any in-flight periodic write rather than racing it. Both would land + // atomically, but the loser would be a wasted encode of the same mirror. + const previous = this.snapshotInFlight ?? Promise.resolve(); + const write = previous.then(async () => { + if (this.snapshotGeneration >= generation) return; // the in-flight write covered it + if (await store.write(this.rid, snapshot)) this.snapshotGeneration = generation; + }); + + this.snapshotInFlight = write.finally(() => { + this.snapshotInFlight = undefined; + }); + return this.snapshotInFlight; } /** In-flight snapshot write, if any. Both triggers can fire close together (the close write diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts index 7d9b3302a9..a18645dbf2 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_scenarios.test.ts @@ -124,10 +124,13 @@ describe("reopening a project", () => { const id = await project(first, "warm reopen"); await first.closeProject(id); + // The close write is started, not awaited, so that closing a project does not sit behind + // an encode. Shutdown drains it, which is what makes it observable here. + await close(first); + // One file for the one project, written at the close boundary. expect(await snapshotFiles(snapshotDir)).toHaveLength(1); expect(first.treeSnapshotStats?.writes).toBeGreaterThanOrEqual(1); - await close(first); const second = await open(); await second.openProject(id); @@ -144,6 +147,24 @@ describe("reopening a project", () => { }); }); + test("closing a project does not wait for its snapshot to be written", async () => { + await withScenario(async ({ open, close, project, snapshotDir }) => { + const ml = await open(); + const id = await project(ml, "unblocked close"); + + await ml.closeProject(id); + // The capture happened synchronously inside closeProject, but the encode and write are + // deferred, so the file is normally not there yet. Asserted as "not blocked on it" + // rather than "definitely absent": a tiny mirror can beat us to the assertion, and the + // point is that close does not await, not that the write is slow. + const writesRightAfterClose = ml.treeSnapshotStats!.writes; + + await close(ml); // drains + expect(ml.treeSnapshotStats!.writes).toBeGreaterThanOrEqual(writesRightAfterClose); + expect(await snapshotFiles(snapshotDir)).toHaveLength(1); + }); + }); + test("project switching: both returns hit", async () => { await withScenario(async ({ open, close, project, snapshotDir }) => { const first = await open(); @@ -151,8 +172,8 @@ describe("reopening a project", () => { const b = await project(first, "B"); await first.closeProject(a); await first.closeProject(b); + await close(first); // drains both deferred close writes expect(await snapshotFiles(snapshotDir)).toHaveLength(2); - await close(first); const second = await open(); await second.openProject(a); From f517a5ec311194ae010a13fbbb92e787391a75c7 Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:53:52 +0200 Subject: [PATCH 6/7] MILAB-6705: keep eviction to our own files, bound snapshot inflation Startup eviction deleted every file in the snapshot directory that was not addressed to the current scope, so a `treeSnapshotPath` pointed at a shared or pre-existing directory lost unrelated files. It now applies the same ownership rule `purge` already documented, shared as `isOurFile`. Decoding also inflates under a 512 MB ceiling, so a replaced file whose checksum-consistent payload has a huge compression ratio costs a cold open instead of the process's memory. --- .../middle_layer/tree_snapshot_store.test.ts | 15 +++++++++++ .../src/middle_layer/tree_snapshot_store.ts | 22 ++++++++++++---- lib/node/pl-tree/src/persisted_tree.test.ts | 10 ++++++++ lib/node/pl-tree/src/persisted_tree.ts | 25 ++++++++++++++++--- 4 files changed, 64 insertions(+), 8 deletions(-) diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts index f493ee183f..905d1f7b94 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts @@ -263,6 +263,21 @@ describe("eviction", () => { expect(store.stats.evictedForSize).toBe(0); }); + test("leaves files that are not ours, whatever the directory holds", async () => { + const store = storeIn()!; + await store.write(rootA, snapshotFor(rootA)); + // `treeSnapshotPath` is caller-supplied: pointed at an existing or shared directory, + // startup housekeeping must not take a stranger's files with it. + await fsp.writeFile(path.join(dir, "someone-elses.txt"), "not ours"); + await fsp.writeFile(path.join(dir, "tree.txt"), "shares our prefix, not our suffix"); + + await store.evict(); + + expect(await files()).toContain("someone-elses.txt"); + expect(await files()).toContain("tree.txt"); + expect(store.stats.evicted).toBe(0); + }); + test("keeps everything when under the ceiling", async () => { const store = storeIn()!; await store.write(rootA, snapshotFor(rootA)); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts index 2a768dfb67..de43dd44cf 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts @@ -96,6 +96,14 @@ function safe(part: string): string { return part.replace(/[^A-Za-z0-9_-]/g, "_"); } +/** Names this class writes: a finished snapshot, or the staging file of a write killed before + * its rename. The directory is caller-supplied and only defaults to one of ours, so nothing + * failing this is ever deleted, by the purge or by the startup eviction. */ +function isOurFile(name: string): boolean { + if (!name.startsWith(FILE_PREFIX)) return false; + return name.endsWith(FILE_SUFFIX) || name.includes(`${FILE_SUFFIX}.tmp.`); +} + /** * Snapshots of project tree mirrors on the local filesystem. * @@ -177,8 +185,7 @@ export class TreeSnapshotStore { // reached for a switch labelled "my disk is troublesome". Only files this class writes // are removed, then the directory itself if that emptied it. for (const name of await fsp.readdir(dir)) { - if (!name.startsWith(FILE_PREFIX)) continue; - if (!name.endsWith(FILE_SUFFIX) && !name.includes(".tmp.")) continue; + if (!isOurFile(name)) continue; await fsp.rm(path.join(dir, name), { force: true }).catch(() => {}); } await fsp.rmdir(dir).catch(() => { @@ -346,12 +353,17 @@ export class TreeSnapshotStore { for (const name of names) { const file = path.join(this.ops.dir, name); - // Anything not addressed to the current scope goes: another build, backend, user or - // schema version, and also the staging files of a write that was killed before its - // rename, which end in `.tmp.` rather than the suffix. + // Anything of ours not addressed to the current scope goes: another build, backend, + // user or schema version, and also the staging files of a write that was killed + // before its rename, which end in `.tmp.` rather than the suffix. const inScope = name.startsWith(`${FILE_PREFIX}${this.scope}.`) && name.endsWith(FILE_SUFFIX); + // Out of scope is not the same as ours to delete: a `treeSnapshotPath` pointed at an + // existing or shared directory would otherwise have every file in it removed at + // startup. Same rule as `purge`, for the same reason. + if (!inScope && !isOurFile(name)) continue; + let size = 0; let mtimeMs = 0; try { diff --git a/lib/node/pl-tree/src/persisted_tree.test.ts b/lib/node/pl-tree/src/persisted_tree.test.ts index b8bbe711fe..3074826f9f 100644 --- a/lib/node/pl-tree/src/persisted_tree.test.ts +++ b/lib/node/pl-tree/src/persisted_tree.test.ts @@ -209,6 +209,16 @@ describe("a snapshot that cannot be read", () => { expect(await decodePersistedTree(bytes)).toStrictEqual({ ok: false, reason: "checksum" }); }); + test("a payload that inflates past the ceiling is refused, not inflated", async () => { + // The checksum only says the compressed bytes are the ones that were written, so a + // replaced file can be consistent and still inflate to far more than a snapshot can be. + const bytes = await encoded(); + expect(await decodePersistedTree(bytes, { maxPayloadBytes: 16 })).toStrictEqual({ + ok: false, + reason: "checksum", + }); + }); + test("an unknown schema version loads as absent", async () => { const bytes = Buffer.from(await encoded()); bytes.writeUInt16LE(PERSISTED_TREE_SCHEMA_VERSION + 1, 4); diff --git a/lib/node/pl-tree/src/persisted_tree.ts b/lib/node/pl-tree/src/persisted_tree.ts index 80e6f6bf59..a744582b38 100644 --- a/lib/node/pl-tree/src/persisted_tree.ts +++ b/lib/node/pl-tree/src/persisted_tree.ts @@ -104,6 +104,13 @@ const FLAG_COMPRESSED = 1 << 0; const HEADER_FIXED_BYTES = 4 /* magic */ + 2 /* schema */ + 2 /* flags */ + 2 /* witness len */; const TRAILER_BYTES = 4 /* payload length */ + 4 /* checksum */; +/** Ceiling on the inflated payload. The checksum only says the bytes are the ones that were + * written, so a replaced file can pair a small, consistent payload with a ratio that + * inflates to gigabytes: unbounded, that costs the process its memory instead of costing + * one cold open. Far above anything real, the heaviest reference project being 10 MB + * compressed against a 256 MB cap on the whole directory. */ +const DEFAULT_MAX_PAYLOAD_BYTES = 512 * 1024 * 1024; + /** Enum orderings are part of the on-disk format: append only, never reorder. An index * the decoder does not know is malformed input, which is why decoding is bounds-checked * rather than cast. */ @@ -165,6 +172,11 @@ export type PersistedTreeReadResult = | { readonly ok: true; readonly value: T } | { readonly ok: false; readonly reason: PersistedTreeReadFailure }; +export type DecodePersistedTreeOps = { + /** Refuses a payload that inflates past this. Defaults to 512 MB. */ + readonly maxPayloadBytes?: number; +}; + export type EncodePersistedTreeOps = { /** Defaults to true. Turning it off trades roughly a factor of three in file size for * the compression CPU, which is the documented escape hatch for periodic writes. */ @@ -388,6 +400,7 @@ export function readPersistedTreeHeader( * reported as a failure reason, so the caller opens cold instead of replaying garbage. */ export async function decodePersistedTree( bytes: Uint8Array, + ops: DecodePersistedTreeOps = {}, ): Promise> { const header = readPersistedTreeHeader(bytes); if (!header.ok) return header; @@ -412,10 +425,16 @@ export async function decodePersistedTree( const stored = bytes.subarray(payloadStart, payloadStart + payloadLength); if (crc32(stored) !== checksum) return failure("checksum"); - payload = (flags & FLAG_COMPRESSED) !== 0 ? await inflateAsync(stored) : stored; + payload = + (flags & FLAG_COMPRESSED) !== 0 + ? await inflateAsync(stored, { + maxOutputLength: ops.maxPayloadBytes ?? DEFAULT_MAX_PAYLOAD_BYTES, + }) + : stored; } catch { - // Includes inflate failures: a payload that passes its checksum but will not - // decompress is damaged in a way we cannot distinguish from corruption. + // Includes inflate failures: a payload that passes its checksum but will not decompress, + // or inflates past the ceiling, is damaged in a way we cannot distinguish from + // corruption. Either way the answer is the same, open cold. return failure("checksum"); } From d25c4b088f29c96c006cbb534ea8ea245405c0cf Mon Sep 17 00:00:00 2001 From: xnacly <47723417+xnacly@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:40:12 +0200 Subject: [PATCH 7/7] MILAB-6705: review fixes, getStats and no hand-parsed resource id `stats` becomes `getStats()`, and the test builds its witness with `parseSignedResourceId` instead of splitting the id on the pipe itself. --- .../src/middle_layer/middle_layer.ts | 2 +- .../middle_layer/tree_snapshot_store.test.ts | 30 +++++++++++-------- .../src/middle_layer/tree_snapshot_store.ts | 2 +- 3 files changed, 19 insertions(+), 15 deletions(-) diff --git a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts index 91e3c70e11..c25580716b 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/middle_layer.ts @@ -1024,7 +1024,7 @@ export class MiddleLayer { * Reads and hits are what show whether a reopen was actually warm, and the miss breakdown * says why it was not. */ public get treeSnapshotStats(): Readonly | undefined { - return this.env.treeSnapshots?.stats; + return this.env.treeSnapshots?.getStats(); } /** diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts index 905d1f7b94..7bcfec1b19 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.test.ts @@ -1,6 +1,10 @@ import { beforeEach, describe, expect, test } from "vitest"; import type { PlClient, SignedResourceId } from "@milaboratories/pl-client"; -import { createSignedResourceId, toResourceSignature } from "@milaboratories/pl-client"; +import { + createSignedResourceId, + parseSignedResourceId, + toResourceSignature, +} from "@milaboratories/pl-client"; import type { PersistedTree } from "@milaboratories/pl-tree"; import type { MiLogger } from "@milaboratories/ts-helpers"; import fsp from "node:fs/promises"; @@ -27,7 +31,7 @@ function fakeClient( * tested against real trees in pl-tree. */ function snapshotFor(root: SignedResourceId): PersistedTree { return { - witness: toResourceSignature(Buffer.from(root.split("|")[1], "hex")), + witness: parseSignedResourceId(root).signature, roots: [root], resources: [], }; @@ -109,7 +113,7 @@ describe("reporting failure", () => { const store = storeIn(occupied)!; expect(await store.write(rootA, snapshotFor(rootA))).toBe(false); - expect(store.stats.writeFailures).toBe(1); + expect(store.getStats().writeFailures).toBe(1); }); test("a successful write says so", async () => { @@ -129,7 +133,7 @@ describe("reporting failure", () => { const read = await store.read(rootA); expect(read).toStrictEqual({ ok: false, miss: "unreadable" }); - expect(store.stats.misses.absent).toBe(0); + expect(store.getStats().misses.absent).toBe(0); }); }); @@ -143,14 +147,14 @@ describe("round trip", () => { if (!read.ok) throw new Error("unreachable"); expect(read.tree.roots).toStrictEqual([rootA]); - expect(store.stats.writes).toBe(1); - expect(store.stats.hits).toBe(1); + expect(store.getStats().writes).toBe(1); + expect(store.getStats().hits).toBe(1); }); test("nothing written means an absent miss", async () => { const store = storeIn()!; expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "absent" }); - expect(store.stats.misses.absent).toBe(1); + expect(store.getStats().misses.absent).toBe(1); }); test("one file per project, rewritten in place", async () => { @@ -259,8 +263,8 @@ describe("eviction", () => { expect(await files()).toHaveLength(1); expect((await store.read(rootA)).ok).toBe(true); - expect(store.stats.evicted).toBe(2); - expect(store.stats.evictedForSize).toBe(0); + expect(store.getStats().evicted).toBe(2); + expect(store.getStats().evictedForSize).toBe(0); }); test("leaves files that are not ours, whatever the directory holds", async () => { @@ -275,7 +279,7 @@ describe("eviction", () => { expect(await files()).toContain("someone-elses.txt"); expect(await files()).toContain("tree.txt"); - expect(store.stats.evicted).toBe(0); + expect(store.getStats().evicted).toBe(0); }); test("keeps everything when under the ceiling", async () => { @@ -285,7 +289,7 @@ describe("eviction", () => { await store.evict(); expect(await files()).toHaveLength(2); - expect(store.stats.evicted).toBe(0); + expect(store.getStats().evicted).toBe(0); }); test("trims to the ceiling, least recently written first", async () => { @@ -308,7 +312,7 @@ describe("eviction", () => { expect((await tight.read(rootA)).ok).toBe(false); expect((await tight.read(rootB)).ok).toBe(true); - expect(tight.stats.evictedForSize).toBe(1); + expect(tight.getStats().evictedForSize).toBe(1); }); test("an unusable directory costs the cache, not the startup", async () => { @@ -319,7 +323,7 @@ describe("eviction", () => { const store = storeIn(occupied)!; await expect(store.evict()).resolves.toBeUndefined(); await expect(store.write(rootA, snapshotFor(rootA))).resolves.toBe(false); - expect(store.stats.writeFailures).toBe(1); + expect(store.getStats().writeFailures).toBe(1); // "unreadable", not "absent": the directory is broken rather than empty, and that is the // distinction someone reading the counters needs. expect(await store.read(rootA)).toStrictEqual({ ok: false, miss: "unreadable" }); diff --git a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts index de43dd44cf..d7cd8e5f5b 100644 --- a/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts +++ b/lib/node/pl-middle-layer/src/middle_layer/tree_snapshot_store.ts @@ -199,7 +199,7 @@ export class TreeSnapshotStore { } } - public get stats(): Readonly { + public getStats(): Readonly { return this.stat; }