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/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/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)), + ); +}); 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", () => {