From f73654470d34b06c677252d900c6eb2ebd50f145 Mon Sep 17 00:00:00 2001 From: Ben Jacobson Date: Thu, 29 Oct 2026 16:01:00 +0000 Subject: [PATCH 1/2] kv: answer entity-type matches at a temporal basis from the TYPE index Public Triples reads always resolve a temporal basis before reaching the KV layer, so the entity-type fast path (basis === undefined) never ran for them: every match({ entityType }) became a temporal scan of the whole store, journal included. ConfigStore.load issues five of these, so each ConfigRuntime.evaluate scanned every fact five times. Add getByEntityTypeTemporal (and its async twin) to KvTripleStore: read the TYPE index, then keep the datoms visible at the basis. Retraction only rewrites META, so the index still names every datom asserted with the type and visibleAt decides recorded and valid time as before. On an in-memory store with 16,000 facts, ConfigRuntime.evaluate drops from 480-985 ms to 1-3 ms. Found while running Triplex in the browser behind WorldVM's programmable-operations demo. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../core/src/kv/hexastore/KvTripleStore.ts | 37 ++++++++-- packages/core/src/kv/layers/KvTriplesLive.ts | 11 ++- .../test/kv/layers/KvTriples.store.test.ts | 71 ++++++++++++++++++- 3 files changed, 111 insertions(+), 8 deletions(-) diff --git a/packages/core/src/kv/hexastore/KvTripleStore.ts b/packages/core/src/kv/hexastore/KvTripleStore.ts index 3f2cd4a..7eeb409 100644 --- a/packages/core/src/kv/hexastore/KvTripleStore.ts +++ b/packages/core/src/kv/hexastore/KvTripleStore.ts @@ -755,6 +755,22 @@ export class KvTripleStore { * Returns null if sync KV operations are not available (caller should fall back). */ getByEntityType(entityType: string): Datom[] | null { + return this.getByEntityTypeWhere(entityType, (datom) => datom.retractedAt === null); + } + + /** + * Datoms of one entity type visible at a bitemporal basis, read from the TYPE + * index. Retraction only rewrites META, so the index still names every datom + * ever asserted with this type and the basis decides which are visible. + */ + getByEntityTypeTemporal(entityType: string, basis: ResolvedTemporalBasis): Datom[] | null { + return this.getByEntityTypeWhere(entityType, (datom) => visibleAt(datom, basis)); + } + + private getByEntityTypeWhere( + entityType: string, + keep: (datom: Datom) => boolean, + ): Datom[] | null { if (!this.kv.getRangeSync) return null; const { start, end } = typePrefixRange(entityType); @@ -764,8 +780,7 @@ export class KvTripleStore { for (let i = 0; i < entries.length; i++) { const tripleId = textDecoder.decode(entries[i]![1]); const datom = this.getByIdSync(tripleId); - if (datom === null) continue; - if (datom.retractedAt !== null) continue; + if (datom === null || !keep(datom)) continue; results.push(datom); } return results; @@ -874,6 +889,21 @@ export class KvTripleStore { * Scans the TYPE index, then batch-fetches all META records in one call. */ getByEntityTypeAsync(entityType: string): Effect.Effect { + return this.getByEntityTypeWhereAsync(entityType, (datom) => datom.retractedAt === null); + } + + /** Async version of getByEntityTypeTemporal using batched META fetches. */ + getByEntityTypeTemporalAsync( + entityType: string, + basis: ResolvedTemporalBasis, + ): Effect.Effect { + return this.getByEntityTypeWhereAsync(entityType, (datom) => visibleAt(datom, basis)); + } + + private getByEntityTypeWhereAsync( + entityType: string, + keep: (datom: Datom) => boolean, + ): Effect.Effect { const { start, end } = typePrefixRange(entityType); return Effect.gen({ self: this }, function* () { @@ -889,8 +919,7 @@ export class KvTripleStore { const results: Datom[] = []; for (const datom of datoms) { - if (datom === null) continue; - if (datom.retractedAt !== null) continue; + if (datom === null || !keep(datom)) continue; results.push(datom); } return results; diff --git a/packages/core/src/kv/layers/KvTriplesLive.ts b/packages/core/src/kv/layers/KvTriplesLive.ts index d617ebf..2ebd18c 100644 --- a/packages/core/src/kv/layers/KvTriplesLive.ts +++ b/packages/core/src/kv/layers/KvTriplesLive.ts @@ -281,16 +281,21 @@ const makeKvTriplesService = Effect.gen(function* () { basis?: ResolvedTemporalBasis, ): Effect.Effect => Effect.gen(function* () { - // Fast path: entityType-only query uses the KV TYPE index + // Fast path: entityType-only query uses the KV TYPE index, at any basis. + // Public reads always arrive with a resolved basis, so without the + // temporal variant every entity-type read scanned the whole store. if (pattern.entityType && !pattern.entityId && !pattern.attribute && !pattern.value) { - const datoms = basis === undefined ? hexaStore.getByEntityType(pattern.entityType) : null; + const datoms = + basis === undefined + ? hexaStore.getByEntityType(pattern.entityType) + : hexaStore.getByEntityTypeTemporal(pattern.entityType, basis); if (datoms !== null) { return datoms.map(datomToTriple); } const asyncDatoms = basis === undefined ? yield* hexaStore.getByEntityTypeAsync(pattern.entityType) - : yield* hexaStore.scanCollectTemporalAsync({}, basis); + : yield* hexaStore.getByEntityTypeTemporalAsync(pattern.entityType, basis); return asyncDatoms .filter((datom) => datom.entityType === pattern.entityType) .map(datomToTriple); diff --git a/packages/core/test/kv/layers/KvTriples.store.test.ts b/packages/core/test/kv/layers/KvTriples.store.test.ts index 3acd921..e781965 100644 --- a/packages/core/test/kv/layers/KvTriples.store.test.ts +++ b/packages/core/test/kv/layers/KvTriples.store.test.ts @@ -6,7 +6,7 @@ */ import { describe, it, expect } from "vitest"; -import { Effect, Layer, Option } from "effect"; +import { Effect, Layer, Option, Stream } from "effect"; import { Triples } from "../../../src/store/Triples.js"; import type { TripleId, EntityId } from "@triplex-build/triplex"; import { KvTriplesLive } from "../../../src/kv/layers/KvTriplesLive.js"; @@ -349,6 +349,75 @@ describe("KvTriplesLive - temporal match", () => { }); }); +describe("KvTriplesLive - entity type at a temporal basis", () => { + it("answers from the TYPE index, honouring recorded and valid time", async () => { + // Count the KV entries each range read returns, to see which index served the match. + let scanned = 0; + const backend = makeTestKvBackend(); + const counting = { + ...backend, + getRange: (options: Parameters[0]) => + backend.getRange(options).pipe(Stream.tap(() => Effect.sync(() => scanned++))), + getRangeSync: (options: Parameters>[0]) => { + const entries = backend.getRangeSync!(options); + scanned += entries.length; + return entries; + }, + }; + const layer = KvTriplesLive.pipe( + Layer.provide(TripleStoreRuntimeLayer), + Layer.provide(Layer.succeed(KvBackend, counting)), + ); + await Effect.runPromise( + Effect.gen(function* () { + const store = yield* Triples; + // Plenty of other facts, so a full scan would be visible in the count. + yield* store.transact( + Array.from({ length: 500 }, (_, i) => ({ + op: "assert" as const, + entityId: `other:${i}` as EntityId, + attribute: ":other/n", + value: num(i), + })), + ); + const early = yield* store.assert({ + entityId: "w:1", + attribute: ":widget/name", + value: str("first"), + entityType: "Widget", + validFrom: 1_000, + validTo: 2_000, + }); + yield* store.assert({ + entityId: "w:1", + attribute: ":widget/name", + value: str("second"), + entityType: "Widget", + validFrom: 2_000, + }); + const recorded = Date.now(); + yield* Effect.sleep("5 millis"); + yield* store.retract(early.id); + + const names = (triples: readonly { value: { value: unknown } }[]) => + triples.map((t) => t.value.value).sort(); + scanned = 0; + expect(names(yield* store.match({ entityType: "Widget" }, { validAt: 1_500 }))).toEqual([]); + expect( + names( + yield* store.match({ entityType: "Widget" }, { validAt: 1_500, recordedAt: recorded }), + ), + ).toEqual(["first"]); + expect(names(yield* store.match({ entityType: "Widget" }, { validAt: 2_500 }))).toEqual([ + "second", + ]); + // Three reads of a two-entry type index, not three scans of 500+ facts and the journal. + expect(scanned).toBe(6); + }).pipe(Effect.provide(layer)), + ); + }); +}); + // ─── History ─────────────────────────────────────────────────────────────── describe("KvTriplesLive - history", () => { From c75dddc383289d11ab4115c94217847bc8b301e0 Mon Sep 17 00:00:00 2001 From: Ben Jacobson Date: Fri, 30 Oct 2026 22:10:17 +0000 Subject: [PATCH 2/2] Read only what each ConfigStore operation needs Every `resolveRef`, `snapshotById`, `commit` and `setRef` called `load()`, which reads every config record and rebuilds every snapshot's node tree. `ConfigRuntime.evaluate` therefore grew with the whole config history, and on a large in-memory store each of its five entity-type reads scanned every fact. - `resolveRef` reads the ref entity, then `snapshotById` reads the snapshot record and only the objects reachable from its root, one batched `entities` read per tree level. - Object bodies, type expressions and snapshot records are content-addressed or immutable, so they are cached by identity for the life of the service. A decoded node's `type` comes from `validUnder`, which later releases can extend, so decoded trees are cached only until `:triplex/valid-under` changes. One `dependencyState` call checks that, which stays correct when another process commits. - `commit` gives the pure transition a partial store: the input subtrees' stored objects and schemas, those objects' revision histories, and the latest snapshot. It checks objects the transition mints outside the inputs (the snapshot root) before persisting them. - `setRef` and `reverseDependencies` read single records instead of loading. `ConfigStore.commit` no longer returns a `store`; `load()` is unchanged for whole-store views. Tests pin the new reads to the full-load semantics: a release sequence produces the same snapshot IDs and revisions as the pure store, resolved refs and snapshots equal the full load while a second instance commits, moves refs and extends `validUnder`, and resolving a ref issues no entity-type scans. Co-Authored-By: Claude Opus 5.5 (1M context) --- .changeset/config-store-targeted-reads.md | 9 + ARCHITECTURE.md | 6 +- packages/core/src/config/ConfigStore.ts | 352 ++++++++++++++++-- .../core/test/config/ConfigStoreReads.test.ts | 174 +++++++++ 4 files changed, 514 insertions(+), 27 deletions(-) create mode 100644 .changeset/config-store-targeted-reads.md create mode 100644 packages/core/test/config/ConfigStoreReads.test.ts diff --git a/.changeset/config-store-targeted-reads.md b/.changeset/config-store-targeted-reads.md new file mode 100644 index 0000000..e5149a6 --- /dev/null +++ b/.changeset/config-store-targeted-reads.md @@ -0,0 +1,9 @@ +--- +"@triplex-build/triplex": minor +--- + +Make the Triples-backed `ConfigStore` read only what each operation needs. `resolveRef` and +`snapshotById` read the ref and the objects reachable from the snapshot, cache immutable records +by identity, and keep decoded trees until `:triplex/valid-under` changes; `commit`, `setRef` and +`reverseDependencies` no longer load the whole store. `ConfigStore.commit` no longer returns a +`store`: call `load()` for the whole store. diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 36f12b8..318f82d 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -49,7 +49,11 @@ hosts typed configuration as a modular layer over the same core. - `ConfigStore` records reserved `_triplex/config/*` entities and `:triplex/*` attributes through the `Triples` transaction boundary. Release creation and an optional ref move are atomic; reverse-dependency and impact-candidate discovery use Datalog, while content identity and proof - verification remain Merkle operations. + verification remain Merkle operations. Reads are targeted: resolving a ref reads the ref and the + objects reachable from its snapshot, and a commit reads only the records its transition consults. + Immutable records are cached by identity; decoded snapshot trees are cached until + `:triplex/valid-under` changes, checked through `dependencyState`. `load()` remains the + whole-store view. - `ConfigRuntime` is the narrow execution bridge between the models. It resolves a config ref, derives a catalog from that immutable release, materializes only the current or historical Triple facts statically reachable by the rule, and invokes the pure evaluator. It rejects diff --git a/packages/core/src/config/ConfigStore.ts b/packages/core/src/config/ConfigStore.ts index 40b75ea..d3e34da 100644 --- a/packages/core/src/config/ConfigStore.ts +++ b/packages/core/src/config/ConfigStore.ts @@ -5,6 +5,13 @@ * Git-style refs are the only moving records. The pure InMemoryConfigStore is * used to calculate and validate each transition, keeping one semantic model * for dangling refs, duplicate objects, revision reuse, and `validUnder`. + * + * Reads are targeted rather than whole-store: resolving a ref reads the ref, + * then only the objects reachable from its snapshot, and commits read only + * what the transition consults. Immutable records are cached by identity for + * the life of the service. Decoded snapshot trees are cached until + * `:triplex/valid-under` changes, because a node's `type` is chosen from the + * schemas its body is known to satisfy, and later releases can add to them. */ import { Context, Data, Effect, Layer } from "effect"; @@ -23,9 +30,10 @@ import { transactSystem } from "../store/systemNamespace.js"; import type { Triple, TransactOp } from "../Triple.js"; import type { TripleValue } from "../Value.js"; import * as CanonicalJson from "../content/CanonicalJson.js"; -import type * as ConfigNode from "./ConfigNode.js"; +import * as ConfigNode from "./ConfigNode.js"; +import type { ContentId } from "../content/ContentId.js"; import * as InMemoryConfigStore from "./InMemoryConfigStore.js"; -import type * as TypeExpr from "./TypeExpr.js"; +import * as TypeExpr from "./TypeExpr.js"; import { unsafe, type EntityId } from "../Branded.js"; export type { @@ -96,7 +104,12 @@ export interface CommitInput { readonly ref?: string; } -export interface CommitResult extends InMemoryConfigStore.CommitResult { +/** + * The committed snapshot and the revisions it minted. Unlike the pure + * `InMemoryConfigStore.commit`, it carries no complete store: the commit reads + * only what it needs. Call `load()` for the whole store. + */ +export interface CommitResult extends Omit { readonly transaction: TransactionResult; } @@ -331,14 +344,302 @@ const makeService = Effect.gen(function* () { }), ]; + // ─── Targeted reads and caches ───────────────────────────────────────────── + + /** Object bodies and kinds by content ID: immutable. */ + const bodies = new Map< + ContentId, + { readonly kind: string; readonly body: CanonicalJson.CanonicalValue } + >(); + /** Type expressions by content ID: immutable. */ + const types = new Map(); + /** Snapshot records by snapshot ID: immutable. */ + const snapshotRecords = new Map(); + /** + * What depends on `validUnder`, valid while `:triplex/valid-under` hasn't + * changed since `position`: each object's known schemas, and the nodes and + * snapshots decoded from them. + */ + let decoded = { + position: -1, + validUnder: new Map>(), + nodes: new Map(), + snapshots: new Map(), + }; + + const refreshDecoded = Effect.gen(function* () { + const { sourcePosition } = yield* triples.dependencyState([System.attribute.validUnder]); + if (sourcePosition !== decoded.position) { + decoded = { + position: sourcePosition, + validUnder: new Map(), + nodes: new Map(), + snapshots: new Map(), + }; + } + return decoded; + }); + + /** Stored objects by content ID; absent IDs are not in the store. */ + const fetchObjects = (cids: ReadonlyArray, state: typeof decoded) => + Effect.gen(function* () { + const missing = cids.filter((cid) => !bodies.has(cid) || !state.validUnder.has(cid)); + if (missing.length > 0) { + const rows = yield* triples.entities(missing.map((cid) => entityId.object(cid))); + for (const [index, objectRows] of rows.entries()) { + if (objectRows.length === 0) continue; + const id = entityId.object(missing[index]!); + const cid = stringValue(latest(objectRows, System.attribute.contentId)); + const kind = stringValue(latest(objectRows, System.attribute.kind)); + const body = jsonValue( + latest(objectRows, System.attribute.data), + ); + if (cid !== missing[index] || !kind || body === undefined) { + return yield* corrupt(id, "missing object data"); + } + bodies.set(cid as ContentId, { kind, body }); + state.validUnder.set( + cid as ContentId, + objectRows + .filter((row) => row.attribute === System.attribute.validUnder) + .map((row) => stringValue(row)) + .filter((value): value is ContentId => value !== undefined) + .sort(), + ); + } + } + const found = new Map(); + for (const cid of cids) { + const stored = bodies.get(cid); + const validUnder = state.validUnder.get(cid); + if (stored && validUnder) found.set(cid, { cid, ...stored, validUnder }); + } + return found; + }); + + /** Type expressions by content ID; absent IDs are not in the store. */ + const fetchTypes = (cids: Iterable) => + Effect.gen(function* () { + const missing = [...new Set(cids)].filter((cid) => !types.has(cid)); + if (missing.length > 0) { + const rows = yield* triples.entities(missing.map((cid) => entityId.type(cid))); + for (const [index, typeRows] of rows.entries()) { + if (typeRows.length === 0) continue; + const id = entityId.type(missing[index]!); + const cid = stringValue(latest(typeRows, System.attribute.contentId)); + const type = jsonValue(latest(typeRows, System.attribute.data)); + if (cid !== missing[index] || type === undefined) { + return yield* corrupt(id, "missing type data"); + } + types.set(cid as ContentId, type); + } + } + return types; + }); + + type NodeBody = { + readonly kind: string; + readonly key: string; + readonly attrs: CanonicalJson.CanonicalValue; + readonly children: ReadonlyArray<{ readonly rel: string; readonly cid: ContentId }>; + readonly refs: ReadonlyArray; + }; + + /** Decode the tree under `rootCid`, reading only the objects it reaches. */ + const decodeTree = (rootCid: ContentId, state: typeof decoded) => + Effect.gen(function* () { + // Fetch level by level, one batched read per level. + let frontier = [rootCid]; + const seen = new Set(frontier); + const objects = new Map(); + while (frontier.length > 0) { + const fetched = yield* fetchObjects(frontier, state); + const next: ContentId[] = []; + for (const cid of frontier) { + const stored = fetched.get(cid); + if (!stored) + return yield* corrupt(entityId.object(cid), "snapshot references a missing node"); + objects.set(cid, stored); + for (const child of (stored.body as unknown as NodeBody).children) { + if (!seen.has(child.cid)) { + seen.add(child.cid); + next.push(child.cid); + } + } + } + frontier = next; + } + const schemas = yield* fetchTypes([...objects.values()].flatMap((o) => o.validUnder)); + + const build = (cid: ContentId, visiting: ReadonlySet): ConfigNode.ConfigNode => { + const cached = state.nodes.get(cid); + if (cached) return cached; + const stored = objects.get(cid)!; + if (visiting.has(cid)) throw corrupt(entityId.object(cid), "child graph contains a cycle"); + const body = stored.body as unknown as NodeBody; + const nested = new Set(visiting).add(cid); + const type = stored.validUnder + .map((id) => schemas.get(id)) + .find((value) => value !== undefined); + const node: ConfigNode.ConfigNode = { + kind: body.kind, + key: body.key, + attrs: body.attrs, + children: body.children.map((child) => ({ + rel: child.rel, + node: build(child.cid, nested), + })), + refs: body.refs, + cid, + ...(type !== undefined && { type }), + }; + state.nodes.set(cid, node); + return node; + }; + return yield* Effect.try({ + try: () => build(rootCid, new Set()), + catch: (error) => + error instanceof CorruptConfigStoreError ? error : corrupt("", String(error)), + }); + }); + + const snapshotRecord = (id: string) => + Effect.gen(function* () { + const cached = snapshotRecords.get(id); + if (cached) return cached; + const rows = yield* triples.match({ entityId: entityId.snapshot(id) }); + if (rows.length === 0) return undefined; + const record = jsonValue(latest(rows, System.attribute.data)); + if (!record || record.id !== id) { + return yield* corrupt(entityId.snapshot(id), "missing snapshot data"); + } + snapshotRecords.set(id, record); + return record; + }); + + const snapshotById: ConfigStoreService["snapshotById"] = (id) => + Effect.gen(function* () { + const state = yield* refreshDecoded; + const cached = state.snapshots.get(id); + if (cached) return cached; + const record = yield* snapshotRecord(id); + if (!record) return undefined; + const snapshot = { ...record, root: yield* decodeTree(record.rootCid, state) }; + state.snapshots.set(id, snapshot); + return snapshot; + }); + + const snapshotPrefix = entityId.snapshot(""); + + const resolveRef: ConfigStoreService["resolveRef"] = (name) => + Effect.gen(function* () { + const target = stringValue(yield* currentRefTriple(name)); + if (target === undefined) return undefined; + if (!target.startsWith(snapshotPrefix)) { + return yield* corrupt(entityId.ref(name), `points to unknown snapshot ${target}`); + } + const snapshot = yield* snapshotById( + decodeURIComponent(target.slice(snapshotPrefix.length)) as ContentId, + ); + if (!snapshot) { + return yield* corrupt(entityId.ref(name), `points to unknown snapshot ${target}`); + } + return snapshot; + }); + + /** The latest snapshot record, which also carries the store's sequence number. */ + const latestSnapshotRecord = Effect.gen(function* () { + let head: SnapshotRecord | undefined; + for (const [id, rows] of groupByEntity(yield* records(System.entityType.snapshot))) { + const record = jsonValue(latest(rows, System.attribute.data)); + if (!record) return yield* corrupt(id, "missing snapshot data"); + snapshotRecords.set(record.id, record); + if (!head || record.seq > head.seq || (record.seq === head.seq && record.id > head.id)) { + head = record; + } + } + return head; + }); + + /** Every revision of the given logical objects. */ + const revisionsOf = (objects: ReadonlyArray) => + Effect.gen(function* () { + const keys = [...new Set(objects.map((object) => object.key))]; + const matches = yield* Effect.forEach(keys, (key) => + triples.match({ + attribute: System.attribute.key, + value: { type: "string", value: key }, + entityType: System.entityType.revision, + }), + ); + const ids = [...new Set(matches.flat().map((row) => row.entityId))]; + const slots = new Set(objects.map((object) => `${object.kind} ${object.key}`)); + return (yield* revisionRecords(ids)).filter((revision) => + slots.has(`${revision.kind} ${revision.key}`), + ); + }); + + const revisionRecords = (ids: ReadonlyArray) => + Effect.gen(function* () { + const rows = yield* triples.entities(ids); + const revisions: InMemoryConfigStore.Revision[] = []; + for (const [index, revisionRows] of rows.entries()) { + const revision = jsonValue( + latest(revisionRows, System.attribute.data), + ); + if (!revision) return yield* corrupt(ids[index]!, "missing revision data"); + revisions.push(revision); + } + return revisions.sort((a, b) => a.seq - b.seq || a.id.localeCompare(b.id)); + }); + + /** + * The part of the store a commit of `objects` consults: their stored nodes + * and schemas, their revision histories, and the latest snapshot. + */ + const commitBase = (objects: ReadonlyArray) => + Effect.gen(function* () { + const state = yield* refreshDecoded; + const nodes = objects.flatMap((object) => + [...ConfigNode.walk(object)].map(({ node }) => node), + ); + const stored = yield* fetchObjects([...new Set(nodes.map((node) => node.cid))], state); + const schemaIds = [ + ...[...stored.values()].flatMap((object) => object.validUnder), + ...nodes.flatMap((node) => (node.type ? [TypeExpr.id(node.type)] : [])), + ]; + const known = yield* fetchTypes(schemaIds); + const schemas = new Map(); + for (const id of schemaIds) { + const type = known.get(id); + if (type) schemas.set(id, type); + } + const revisions = yield* revisionsOf(objects); + const head = yield* latestSnapshotRecord; + const parent = head ? yield* snapshotById(head.id) : undefined; + const store: InMemoryConfigStore.InMemoryConfigStore = { + objects: stored, + schemas, + revisions, + snapshots: parent ? [parent] : [], + refs: new Map(), + seq: Math.max(0, head?.seq ?? 0, ...revisions.map((revision) => revision.seq)), + }; + return { store, state }; + }); + const commit: ConfigStoreService["commit"] = (input) => Effect.gen(function* () { - const before = yield* load(); + const { store: before, state } = yield* commitBase(input.objects); const committed = yield* InMemoryConfigStore.commit(before, input); + // The commit may also mint objects outside the inputs (the snapshot root), + // which an earlier identical release may already have stored. + const minted = [...committed.store.objects.keys()].filter((cid) => !before.objects.has(cid)); + const existing = yield* fetchObjects(minted, state); const operations: TransactOp[] = []; for (const [cid, stored] of committed.store.objects) { - const previous = before.objects.get(cid); + const previous = before.objects.get(cid) ?? existing.get(cid); if (!previous) { operations.push( assertOp(entityId.object(cid), System.entityType.object, System.attribute.contentId, { @@ -428,32 +729,32 @@ const makeService = Effect.gen(function* () { configSnapshot: committed.snapshot.id, ...(current ? { preconditions: [{ _tag: "TripleLive" as const, id: current.id }] } : {}), }); - const store = yield* InMemoryConfigStore.setRef( - committed.store, - input.ref, - committed.snapshot.id, - ); - return { ...committed, store, transaction }; + return { snapshot: committed.snapshot, created: committed.created, transaction }; } const transaction = yield* transactSystem(triples, operations, { actor: "triplex/config-store", configSnapshot: committed.snapshot.id, }); - return { ...committed, store: committed.store, transaction }; + return { snapshot: committed.snapshot, created: committed.created, transaction }; }); const setRef: ConfigStoreService["setRef"] = (name, snapshotId) => Effect.gen(function* () { - const store = yield* load(); - const next = yield* InMemoryConfigStore.setRef(store, name, snapshotId); + const snapshot = yield* snapshotById(snapshotId); + if (!snapshot) { + return yield* new InMemoryConfigStore.UnknownSnapshotError({ + id: snapshotId, + message: `Cannot point ref "${name}" at unknown snapshot ${snapshotId}`, + }); + } const current = yield* currentRefTriple(name); yield* transactSystem(triples, refOps(name, snapshotId, current), { actor: "triplex/config-store", configSnapshot: snapshotId, ...(current ? { preconditions: [{ _tag: "TripleLive" as const, id: current.id }] } : {}), }); - return InMemoryConfigStore.resolveRef(next, name)!; + return snapshot; }); const reverseDependencies: ConfigStoreService["reverseDependencies"] = (object) => @@ -468,23 +769,22 @@ const makeService = Effect.gen(function* () { ], ], }); - const ids = new Set( - response.results - .map((row) => row["?revision"]) - .filter((value): value is string => typeof value === "string"), - ); - const store = yield* load(); - return store.revisions.filter((revision) => ids.has(entityId.revision(revision.id))); + const ids = [ + ...new Set( + response.results + .map((row) => row["?revision"]) + .filter((value): value is string => typeof value === "string"), + ), + ].map(unsafe.entityId); + return yield* revisionRecords(ids); }); return { load, commit, setRef, - resolveRef: (name: string) => - load().pipe(Effect.map((store) => InMemoryConfigStore.resolveRef(store, name))), - snapshotById: (id: string) => - load().pipe(Effect.map((store) => InMemoryConfigStore.snapshotById(store, id))), + resolveRef, + snapshotById, reverseDependencies, impactCandidates: reverseDependencies, recheck: (input: { readonly kind: string; readonly type: TypeExpr.TypeExpr }) => diff --git a/packages/core/test/config/ConfigStoreReads.test.ts b/packages/core/test/config/ConfigStoreReads.test.ts new file mode 100644 index 0000000..7127063 --- /dev/null +++ b/packages/core/test/config/ConfigStoreReads.test.ts @@ -0,0 +1,174 @@ +/** + * The Triples-backed ConfigStore reads only what each operation needs and caches immutable + * records. These tests pin it to the semantics of the pure store loaded in full. + */ + +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Layer } from "effect"; + +import { KvTriples } from "../../src/kv/layers/KvTriplesLive.js"; +import { Triples, type TriplesService } from "../../src/store/Triples.js"; +import * as ConfigNode from "../../src/config/ConfigNode.js"; +import * as ConfigStore from "../../src/config/ConfigStore.js"; +import * as InMemoryConfigStore from "../../src/config/InMemoryConfigStore.js"; +import * as TypeExpr from "../../src/config/TypeExpr.js"; + +const Narrow = TypeExpr.struct({ label: TypeExpr.required(TypeExpr.text) }); +// Accepts the same bodies, so committing a node under it extends `validUnder`. +const Wide = TypeExpr.struct({ + label: TypeExpr.required(TypeExpr.text), + hint: TypeExpr.optional(TypeExpr.text), +}); + +const field = (label: string, type: TypeExpr.TypeExpr = Narrow) => + ConfigNode.makeTyped({ kind: "field", key: "name", type, attrs: { label } }); + +const form = (fieldNode: ConfigNode.ConfigNode, title: string) => + ConfigNode.make({ + kind: "form", + key: "onboarding", + attrs: { title }, + children: [{ rel: "field", node: fieldNode }], + refs: [{ rel: "uses", kind: "policy", key: "p" }], + }); + +const policy = (message: string) => + ConfigNode.make({ kind: "policy", key: "p", attrs: { message } }); + +/** Two ConfigStore instances over one Triples store, like two processes sharing a database. */ +const makeStores = Effect.gen(function* () { + const triples = yield* Triples; + const instance = () => + Effect.service(ConfigStore.ConfigStore).pipe( + Effect.provide(ConfigStore.layer.pipe(Layer.provide(Layer.succeed(Triples, triples)))), + ); + return { triples, a: yield* instance(), b: yield* instance() }; +}); + +/** Every ref and snapshot, as the whole-store load reconstructs them. */ +const fromFullLoad = (store: ConfigStore.ConfigStoreService) => + Effect.gen(function* () { + const loaded = yield* store.load(); + return { + refs: [...loaded.refs.keys()].map((name) => [ + name, + InMemoryConfigStore.resolveRef(loaded, name), + ]), + snapshots: loaded.snapshots, + }; + }); + +const fromTargetedReads = (store: ConfigStore.ConfigStoreService, refs: readonly string[]) => + Effect.gen(function* () { + const loaded = yield* store.load(); + return { + refs: yield* Effect.forEach(refs, (name) => + store.resolveRef(name).pipe(Effect.map((snapshot) => [name, snapshot] as const)), + ), + snapshots: yield* Effect.forEach(loaded.snapshots, (snapshot) => + store.snapshotById(snapshot.id), + ), + }; + }); + +describe("ConfigStore targeted reads", () => { + it.effect("commit the same releases as the pure store", () => + Effect.gen(function* () { + const { a } = yield* makeStores; + let pure = InMemoryConfigStore.empty(); + const releases = [ + [yield* field("Name"), "Welcome", "v1"], + [yield* field("Name"), "Welcome!", "v1"], + [yield* field("Full name"), "Welcome!", "v2"], + [yield* field("Full name", Wide), "Welcome!", "v2"], + [yield* field("Name"), "Welcome", "v1"], + ] as const; + for (const [index, [fieldNode, title, message]] of releases.entries()) { + const objects = [yield* form(fieldNode, title), yield* policy(message)]; + const expected = yield* InMemoryConfigStore.commit(pure, { label: `r${index}`, objects }); + pure = expected.store; + const actual = yield* a.commit({ label: `r${index}`, objects, ref: "live" }); + expect(actual.snapshot.id).toBe(expected.snapshot.id); + expect(actual.snapshot.seq).toBe(expected.snapshot.seq); + expect(actual.snapshot.parentId).toBe(expected.snapshot.parentId); + expect(actual.created.map((revision) => revision.id)).toEqual( + expected.created.map((revision) => revision.id), + ); + } + const loaded = yield* a.load(); + expect(loaded.revisions.map((revision) => revision.id)).toEqual( + pure.revisions.map((revision) => revision.id), + ); + expect([...loaded.objects.keys()].sort()).toEqual([...pure.objects.keys()].sort()); + expect([...loaded.schemas.keys()].sort()).toEqual([...pure.schemas.keys()].sort()); + expect(loaded.seq).toBe(pure.seq); + }).pipe(Effect.provide(KvTriples.layer)), + ); + + it.effect("agree with the full load while another instance writes", () => + Effect.gen(function* () { + const { a, b } = yield* makeStores; + const first = yield* a.commit({ + label: "one", + objects: [yield* form(yield* field("Name"), "Hi"), yield* policy("v1")], + ref: "live", + }); + expect(yield* fromTargetedReads(a, ["live"])).toEqual(yield* fromFullLoad(a)); + + // Another writer: a new release and a ref move. + const second = yield* b.commit({ + label: "two", + objects: [yield* form(yield* field("Name"), "Hello"), yield* policy("v1")], + ref: "test", + }); + yield* b.setRef("live", second.snapshot.id); + expect((yield* a.resolveRef("live"))?.id).toBe(second.snapshot.id); + expect((yield* a.resolveRef("test"))?.id).toBe(second.snapshot.id); + + // Another writer proves an existing body valid under a new schema. Decoded nodes take + // their type from `validUnder`, so cached trees must follow, as the full load does. + yield* b.commit({ + label: "three", + objects: [yield* form(yield* field("Name", Wide), "Hello"), yield* policy("v1")], + }); + expect(yield* fromTargetedReads(a, ["live", "test"])).toEqual(yield* fromFullLoad(a)); + expect((yield* a.snapshotById(first.snapshot.id))?.root).toEqual( + InMemoryConfigStore.snapshotById(yield* a.load(), first.snapshot.id)?.root, + ); + + expect(yield* a.resolveRef("missing")).toBeUndefined(); + expect(yield* a.snapshotById("sha256-missing" as never)).toBeUndefined(); + const unknown = yield* a.setRef("live", "sha256-missing" as never).pipe(Effect.flip); + expect(unknown._tag).toBe("UnknownSnapshotError"); + }).pipe(Effect.provide(KvTriples.layer)), + ); + + it.effect("don't scan the store to resolve a ref", () => + Effect.gen(function* () { + const inner = yield* Triples; + const typeScans: string[] = []; + const counting: TriplesService = { + ...inner, + match: (pattern, basis) => { + if (pattern.entityType !== undefined && pattern.entityId === undefined) { + typeScans.push(pattern.entityType); + } + return inner.match(pattern, basis); + }, + }; + const store = yield* Effect.service(ConfigStore.ConfigStore).pipe( + Effect.provide(ConfigStore.layer.pipe(Layer.provide(Layer.succeed(Triples, counting)))), + ); + yield* store.commit({ + label: "one", + objects: [yield* form(yield* field("Name"), "Hi"), yield* policy("v1")], + ref: "live", + }); + typeScans.length = 0; + expect(yield* store.resolveRef("live")).toBeDefined(); + expect(yield* store.resolveRef("live")).toBeDefined(); + expect(yield* store.reverseDependencies({ kind: "policy", key: "p" })).toHaveLength(1); + expect(typeScans).toEqual([]); + }).pipe(Effect.provide(KvTriples.layer)), + ); +});