From b7efbca5592940618fc570d38e7f55644b4d2674 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 10 Oct 2026 16:35:26 +0100 Subject: [PATCH 1/4] feat: add store-contract package The persistence port for process models and their run overlays as a versioned oRPC contract (v1): save, load, list and delete models, set run state, upsert items read from a source, structural ready, blocked-chain and impact queries, and a change feed as an event iterator. Inputs and outputs reuse graph-schema's model and overlay schemas, so the port carries the model as it is. createGraphStoreRouter builds the router on a plain GraphStore implementation and owns abort handling, error mapping and eager subscription, so a caller holding the change iterator cannot miss its own next write. Every write is checked with validate and validateOverlay and takes the next store-wide revision. The queries are structural (transitions and run state only, decision loop edges excluded, control elements passed through), so a SQL store can answer them without evaluating conditions; on models without conditions or artefacts ready agrees with graph-domain's readySet. An in-memory fake and a conformance kit (separate ./conformance entry, optional vitest peer) ship with it; the fake passes the kit in process and over a MessageChannel. --- README.md | 1 + packages/core/graph/store-contract/AGENTS.md | 1 + packages/core/graph/store-contract/CLAUDE.md | 1 + packages/core/graph/store-contract/README.md | 94 +++ .../graph/store-contract/eslint.config.ts | 9 + .../core/graph/store-contract/package.json | 91 +++ .../graph/store-contract/src/conformance.ts | 654 ++++++++++++++++++ .../core/graph/store-contract/src/contract.ts | 125 ++++ .../core/graph/store-contract/src/errors.ts | 38 + .../core/graph/store-contract/src/fake.ts | 163 +++++ .../store-contract/src/fake.unit.test.ts | 45 ++ .../core/graph/store-contract/src/feed.ts | 101 +++ .../store-contract/src/feed.unit.test.ts | 67 ++ .../core/graph/store-contract/src/fixtures.ts | 145 ++++ .../core/graph/store-contract/src/index.ts | 45 ++ .../core/graph/store-contract/src/merge.ts | 89 +++ .../core/graph/store-contract/src/router.ts | 154 +++++ .../core/graph/store-contract/src/schema.ts | 96 +++ .../store-contract/src/schema.unit.test.ts | 48 ++ .../core/graph/store-contract/src/store.ts | 91 +++ .../graph/store-contract/src/structure.ts | 153 ++++ .../store-contract/src/structure.unit.test.ts | 94 +++ .../graph/store-contract/stryker.config.ts | 3 + .../store-contract/test/package.smoke.test.ts | 28 + .../graph/store-contract/tsconfig.build.json | 20 + .../core/graph/store-contract/tsconfig.json | 9 + .../graph/store-contract/tsconfig.node.json | 8 + .../graph/store-contract/vitest.config.ts | 16 + pnpm-lock.yaml | 55 ++ tsconfig.build.json | 3 + 30 files changed, 2447 insertions(+) create mode 120000 packages/core/graph/store-contract/AGENTS.md create mode 120000 packages/core/graph/store-contract/CLAUDE.md create mode 100644 packages/core/graph/store-contract/README.md create mode 100644 packages/core/graph/store-contract/eslint.config.ts create mode 100644 packages/core/graph/store-contract/package.json create mode 100644 packages/core/graph/store-contract/src/conformance.ts create mode 100644 packages/core/graph/store-contract/src/contract.ts create mode 100644 packages/core/graph/store-contract/src/errors.ts create mode 100644 packages/core/graph/store-contract/src/fake.ts create mode 100644 packages/core/graph/store-contract/src/fake.unit.test.ts create mode 100644 packages/core/graph/store-contract/src/feed.ts create mode 100644 packages/core/graph/store-contract/src/feed.unit.test.ts create mode 100644 packages/core/graph/store-contract/src/fixtures.ts create mode 100644 packages/core/graph/store-contract/src/index.ts create mode 100644 packages/core/graph/store-contract/src/merge.ts create mode 100644 packages/core/graph/store-contract/src/router.ts create mode 100644 packages/core/graph/store-contract/src/schema.ts create mode 100644 packages/core/graph/store-contract/src/schema.unit.test.ts create mode 100644 packages/core/graph/store-contract/src/store.ts create mode 100644 packages/core/graph/store-contract/src/structure.ts create mode 100644 packages/core/graph/store-contract/src/structure.unit.test.ts create mode 100644 packages/core/graph/store-contract/stryker.config.ts create mode 100644 packages/core/graph/store-contract/test/package.smoke.test.ts create mode 100644 packages/core/graph/store-contract/tsconfig.build.json create mode 100644 packages/core/graph/store-contract/tsconfig.json create mode 100644 packages/core/graph/store-contract/tsconfig.node.json create mode 100644 packages/core/graph/store-contract/vitest.config.ts diff --git a/README.md b/README.md index 34421b9..47007d5 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,7 @@ Configuration lives at exactly one of two levels: in the package (`eslint.config | [`graph-layout`](packages/core/graph/layout/README.md) | Layered layout of a projected flowchart, swimlane or state machine into positions in the model's `views` shape. | | [`graph-source-contract`](packages/core/graph/source-contract/README.md) | The `GraphSource` port as an oRPC contract: versioned, paged and streamed sync of bound activities, participants, transitions and run state, resumable by cursor, with an in-memory fake and a conformance kit. | | [`graph-source-github-issues`](packages/adapters/graph/source-github-issues/README.md) | The GitHub issues adapter for the `GraphSource` port: issues as bound activities, `blockedBy` as transitions, sub-issues, repositories, organisations and assignees, across organisations and repositories. | +| [`graph-store-contract`](packages/core/graph/store-contract/README.md) | The persistence port as an oRPC contract: save, load and delete models, set run state, upsert items from a source, structural ready, blocked-chain and impact queries, and a change feed; with an in-memory fake and a conformance kit. | | [`process-graph`](packages/targets/process-graph/README.md) | The primary entry point: re-exports the schema, projection and domain APIs flat and the DAG API as the `dag` namespace. | ## Conventions diff --git a/packages/core/graph/store-contract/AGENTS.md b/packages/core/graph/store-contract/AGENTS.md new file mode 120000 index 0000000..42061c0 --- /dev/null +++ b/packages/core/graph/store-contract/AGENTS.md @@ -0,0 +1 @@ +README.md \ No newline at end of file diff --git a/packages/core/graph/store-contract/CLAUDE.md b/packages/core/graph/store-contract/CLAUDE.md new file mode 120000 index 0000000..42061c0 --- /dev/null +++ b/packages/core/graph/store-contract/CLAUDE.md @@ -0,0 +1 @@ +README.md \ No newline at end of file diff --git a/packages/core/graph/store-contract/README.md b/packages/core/graph/store-contract/README.md new file mode 100644 index 0000000..570a13e --- /dev/null +++ b/packages/core/graph/store-contract/README.md @@ -0,0 +1,94 @@ +# @exadev/process-graph-store-contract + +[![GitHub](https://img.shields.io/badge/GitHub-181717?logo=github&logoColor=white)](https://github.com/ExaDev/process-graph) [![npm](https://img.shields.io/badge/npm-CB3837?logo=npm&logoColor=white)](https://www.npmjs.com/package/@exadev/process-graph-store-contract) [![Release](https://img.shields.io/github/v/release/ExaDev/process-graph)](https://github.com/ExaDev/process-graph/releases/latest) [![CI](https://img.shields.io/github/actions/workflow/status/ExaDev/process-graph/ci.yml?branch=main)](https://github.com/ExaDev/process-graph/actions) + +The persistence port for process models and their run overlays, as an [oRPC](https://orpc.dev) router contract. One definition runs in process (a router client called directly), across a `MessagePort` to a worker, or over HTTP, with no change to its consumers. Inputs and outputs are Zod schemas built from `@exadev/process-graph-schema`'s `modelSchema` and `overlaySchema`, so the port carries the model as it is and storage layout never reaches it. `src/` imports nothing Node-only. + +```ts +import { createRouterClient } from "@orpc/server"; +import { + createFakeGraphStore, + createGraphStoreRouter, +} from "@exadev/process-graph-store-contract"; + +const store = createRouterClient( + createGraphStoreRouter(createFakeGraphStore()), +); + +await store.v1.models.save({ modelId: "release", model }); +await store.v1.overlay.set({ modelId: "release", overlay: { plan: "done" } }); +await store.v1.query.ready({ modelId: "release" }); // { elements: ["build"] } + +for await (const change of await store.v1.changes({ modelId: "release" })) { + // { modelId: "release", kind: "overlay", revision: 3 } +} +``` + +## Procedures + +The contract is keyed by version: `graphStoreContract.v1`. A breaking change adds a `v2` beside it, so one server serves both while clients move across. + +| Procedure | Input | Output | +| -------------------- | ----------------------------- | ----------------------------------------- | +| `models.list` | none | Each model's id and revision, by id | +| `models.save` | `modelId`, `model` | The revision the save took | +| `models.load` | `modelId` | `modelId`, `revision`, `model`, `overlay` | +| `models.delete` | `modelId` | Whether there was a model to delete | +| `overlay.set` | `modelId`, `overlay` | The revision the change took | +| `items.upsert` | `modelId`, `items` | The revision the change took | +| `query.ready` | `modelId` | The ready elements, in model order | +| `query.blockedChain` | `modelId`, `element` | The unfinished work upstream | +| `query.impact` | `modelId`, `element` | The unfinished work downstream | +| `changes` | `modelId`, optionally `since` | An event iterator of changes | + +Every call takes the caller's abort signal through the oRPC client options. Failures are the contract's typed errors: `NOT_FOUND` (with `modelId`, and `element` for a query about an element the model lacks) and `UNPROCESSABLE_CONTENT` (with the `issues` that `validate` and `validateOverlay` report). Input that does not match the schemas is oRPC's `BAD_REQUEST`. + +## Writes + +Every write is atomic and checked: a change that would leave the model invalid, or the overlay naming an element the model does not declare, is rejected whole and the store is unchanged. A write whose signal has fired before it commits changes nothing. + +- `models.save` replaces the model and keeps the run state only of elements the new model still declares. +- `overlay.set` sets the run state of the elements it names and leaves the others. +- `items.upsert` merges items read from a source: participants, bound activities and transitions replace the record with the same id in its place, or are appended in their given order; overlay entries replace the run state of the elements they name. It creates the model from an empty one when there is none. A bound activity's `source` is required and equals its id. + +Each write takes the next revision from one store-wide sequence, so revisions only grow, a model's revision is the revision of its last change, and a model deleted and saved again never reuses an earlier one. + +## Queries + +The queries are structural: they read the stored transitions and run state, and nothing else, so a store can answer them in its own query language without loading the model or evaluating anything. For reasoning that follows conditions, artefacts and three-valued status, load the model and use `@exadev/process-graph-domain`. + +- Transitions that close a legal decision loop (`loopTransitions` from `graph-domain`) are not dependencies. +- Decisions and pseudo-states are control elements: traversals pass through them and never report them. +- An element is done when its run state is `done`. Traversal stops at a done element and never reports it. +- `query.ready` is every element whose run state is absent or `open`, that is not a control element, and whose blocked chain is empty. On a model with no conditions and no artefacts it agrees with `graph-domain`'s `readySet`. +- `query.blockedChain` is everything not done upstream of the element, each with the fewest transitions to reach it, ordered by that distance and then by model order. A done element waits on nothing. +- `query.impact` is the same downstream, whatever the element's own state. + +## Changes + +`changes` yields every later change to one model as `{ modelId, kind, revision }`, with `kind` one of `saved`, `overlay`, `items` or `deleted`. The router subscribes before it returns, so a caller holding the iterator is already subscribed and misses nothing it writes next. Given `since`, a revision the caller last saw, the first event is `{ modelId, kind: "missed" }` when the model is no longer at that revision (it changed, or it has gone), and the caller should load it again. + +A store's `ChangeFeed` only sees the writes made through that store. Where several processes or tabs write one database, each bridges the others' changes into its own feed. + +## Implementing a store + +An adapter implements `GraphStore` (plain async functions taking a request object with the abort signal) and serves it with `createGraphStoreRouter`, which owns abort handling, error mapping and subscriptions. `checkStored`, `mergeBoundItems` and `overlayWithin` give every store the same write rules, and `structureOf`, `readyOf`, `blockedChainOf` and `impactOf` are the reference queries `createFakeGraphStore` uses. + +Every adapter runs the conformance kit, through a real oRPC client: + +```ts +import { createRouterClient } from "@orpc/server"; +import { graphStoreConformance } from "@exadev/process-graph-store-contract/conformance"; + +graphStoreConformance({ + name: "my store, in process", + createClient: async () => ({ + client: createRouterClient(createGraphStoreRouter(await openEmptyStore())), + close: () => Promise.resolve(), + }), +}); +``` + +The kit is a separate entry point with an optional `vitest` peer dependency, so the runtime entry carries no test framework. Run it over each transport the adapter is served on; this package runs it against the fake both in process and over a `MessageChannel`. + +oRPC clients are proxies over functions, so a procedure named `apply`, `call` or `bind` would be shadowed by `Function.prototype` on the client; the contract avoids those names. diff --git a/packages/core/graph/store-contract/eslint.config.ts b/packages/core/graph/store-contract/eslint.config.ts new file mode 100644 index 0000000..4c43201 --- /dev/null +++ b/packages/core/graph/store-contract/eslint.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from "eslint/config"; +import { packageLintConfig } from "../../../../eslint.shared.ts"; + +export default defineConfig( + packageLintConfig({ + tsconfigRootDir: import.meta.dirname, + isomorphic: true, + }), +); diff --git a/packages/core/graph/store-contract/package.json b/packages/core/graph/store-contract/package.json new file mode 100644 index 0000000..74b313b --- /dev/null +++ b/packages/core/graph/store-contract/package.json @@ -0,0 +1,91 @@ +{ + "name": "@exadev/process-graph-store-contract", + "version": "0.0.0", + "description": "The persistence port for process models and run overlays as an oRPC contract: versioned procedures with Zod inputs and outputs, an in-memory fake and a conformance kit every store adapter runs.", + "type": "module", + "repository": { + "type": "git", + "url": "git+https://github.com/ExaDev/process-graph.git", + "directory": "packages/core/graph/store-contract" + }, + "homepage": "https://github.com/ExaDev/process-graph/tree/main/packages/core/graph/store-contract#readme", + "bugs": { + "url": "https://github.com/ExaDev/process-graph/issues" + }, + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + }, + "./conformance": { + "types": "./dist/conformance.d.ts", + "import": "./dist/conformance.js" + } + }, + "types": "./dist/index.d.ts", + "files": [ + "dist" + ], + "publishConfig": { + "access": "public", + "provenance": true, + "registry": "https://registry.npmjs.org/" + }, + "sideEffects": false, + "license": "MIT", + "scripts": { + "build": "turbo run _build", + "_build": "tsc -b tsconfig.build.json", + "lint": "turbo run _lint", + "_lint": "eslint . --cache --max-warnings 0", + "_lint:fix": "eslint . --fix --cache --max-warnings 0", + "typecheck": "turbo run _typecheck _typecheck:attw", + "_typecheck": "tsc -p tsconfig.json --noEmit && tsc -p tsconfig.node.json --noEmit", + "_typecheck:attw": "attw --pack --profile esm-only", + "test": "turbo run _test", + "_test": "vitest run --project unit", + "test:coverage": "turbo run _test:coverage", + "_test:coverage": "vitest run --project unit --coverage", + "test:smoke": "turbo run _test:smoke", + "_test:smoke": "vitest run --project smoke", + "test:mutation": "turbo run _test:mutation", + "_test:mutation": "stryker run stryker.config.ts", + "prepublishOnly": "pnpm run lint && pnpm run typecheck && pnpm run test && pnpm run build && publint && attw --pack --profile esm-only" + }, + "keywords": [ + "process-model", + "store", + "orpc", + "contract", + "conformance" + ], + "devDependencies": { + "@arethetypeswrong/cli": "0.18.5", + "@orpc/client": "2.0.0-beta.41", + "@stryker-mutator/core": "10.0.0", + "@stryker-mutator/typescript-checker": "10.0.0", + "@stryker-mutator/vitest-runner": "10.0.0", + "@types/node": "26.6.4", + "@vitest/coverage-v8": "5.0.3", + "eslint": "10.12.0", + "publint": "0.3.25", + "typescript": "6.0.3", + "vitest": "5.0.3" + }, + "dependencies": { + "@exadev/process-graph-domain": "1.0.2", + "@exadev/process-graph-schema": "1.0.2", + "@orpc/contract": "2.0.0-beta.41", + "@orpc/server": "2.0.0-beta.41", + "trilean": "1.7.2", + "zod": "4.6.5" + }, + "peerDependencies": { + "vitest": "5.0.3" + }, + "peerDependenciesMeta": { + "vitest": { + "optional": true + } + } +} diff --git a/packages/core/graph/store-contract/src/conformance.ts b/packages/core/graph/store-contract/src/conformance.ts new file mode 100644 index 0000000..6ef28c1 --- /dev/null +++ b/packages/core/graph/store-contract/src/conformance.ts @@ -0,0 +1,654 @@ +import type { Model, OpenIds } from "@exadev/process-graph-schema"; +import { describe, expect, test } from "vitest"; +import type { GraphStoreClient } from "./contract.ts"; +import { everyFieldModel, queryModel } from "./fixtures.ts"; +import type { BoundItems, StoreChange } from "./schema.ts"; + +/** What the conformance kit needs from an adapter's test. */ +export interface GraphStoreConformanceOptions { + /** A name for the `describe` block, such as the adapter and transport under test. */ + readonly name: string; + /** + * Makes a client for a store with no models in it. Called once per test, so every test starts empty and any transport can be used, in process or over a link. `close` is called when the test ends. + */ + readonly createClient: () => Promise<{ + readonly client: GraphStoreClient; + readonly close: () => Promise; + }>; +} + +/** Runs a test body against a fresh client and closes it afterwards, pass or fail. */ +const withClient = + ( + options: GraphStoreConformanceOptions, + body: (client: GraphStoreClient) => Promise, + ) => + async (): Promise => { + const { client, close } = await options.createClient(); + try { + await body(client); + } finally { + await close(); + } + }; + +/** The settled error of a call that must fail, so its code and data can be asserted. */ +const failureOf = async ( + call: Readonly>, +): Promise => { + try { + await call; + } catch (error) { + return error; + } + throw new Error("expected the call to fail"); +}; + +/** A two-step model whose only transition can be broken by an upsert. */ +const pairModel: Model = { + participants: [{ id: "team", label: "Team", kind: "role" }], + events: [], + actions: [], + artefacts: [], + elements: [ + { kind: "activity", id: "plan", label: "Plan", actor: "team" }, + { kind: "activity", id: "build", label: "Build", actor: "team" }, + ], + transitions: [{ id: "plan-build", source: "plan", target: "build" }], + fragments: [], + projections: [], +}; + +/** Items as a source would send them: one new repository participant, two issues, one replacing an existing activity, and a dependency. */ +const issueItems: BoundItems = { + participants: [ + { id: "github:org/repo", label: "org/repo", kind: "system" }, + { id: "team", label: "Platform team", kind: "role" }, + ], + activities: [ + { + kind: "activity", + id: "github:org/repo#1", + label: "Design the schema", + actor: "github:org/repo", + source: "github:org/repo#1", + }, + { + kind: "activity", + id: "github:org/repo#2", + label: "Write the migration", + actor: "team", + source: "github:org/repo#2", + }, + ], + transitions: [ + { id: "dep-1-2", source: "github:org/repo#1", target: "github:org/repo#2" }, + ], + overlay: { "github:org/repo#1": "done" }, +}; + +/** Reads a subscription up to and including the first event `last` accepts, then stops it. */ +const takeUntil = async ( + events: Readonly>, + last: (change: StoreChange) => boolean, +): Promise => { + const received: StoreChange[] = []; + for await (const event of events) { + received.push(event); + if (last(event)) { + break; + } + } + + return received; +}; + +/** Reads the first event of a subscription, then stops it. */ +const firstOf = async ( + events: Readonly>, +): Promise => takeUntil(events, () => true); + +/** + * The behaviour every `GraphStore` adapter must show, as Vitest tests. Call it at the top level of a test file; every test runs through the oRPC client the options create, so the kit checks the port as consumers see it, transport included. + * @param options - How to make a client for an empty store under test. + */ +export const graphStoreConformance = ( + options: GraphStoreConformanceOptions, +): void => { + describe(`GraphStore conformance: ${options.name}`, () => { + test( + "a new store lists no models", + withClient(options, async (client) => { + await expect(client.v1.models.list()).resolves.toEqual([]); + }), + ); + + test( + "a saved model loads back unchanged, every field and order kept", + withClient(options, async (client) => { + const { revision } = await client.v1.models.save({ + modelId: "orders", + model: everyFieldModel, + }); + + await expect( + client.v1.models.load({ modelId: "orders" }), + ).resolves.toStrictEqual({ + modelId: "orders", + revision, + model: everyFieldModel, + overlay: {}, + }); + }), + ); + + test( + "loading a model that was never saved is NOT_FOUND", + withClient(options, async (client) => { + await expect( + failureOf(client.v1.models.load({ modelId: "missing" })), + ).resolves.toMatchObject({ + code: "NOT_FOUND", + data: { modelId: "missing" }, + }); + }), + ); + + test( + "an invalid model is rejected whole with its issues", + withClient(options, async (client) => { + const duplicated: Model = { + ...pairModel, + elements: [...pairModel.elements, ...pairModel.elements], + }; + + await expect( + failureOf( + client.v1.models.save({ modelId: "pair", model: duplicated }), + ), + ).resolves.toMatchObject({ + code: "UNPROCESSABLE_CONTENT", + data: { + issues: [ + { code: "duplicate-id", message: 'Duplicate element id "plan"' }, + { code: "duplicate-id", message: 'Duplicate element id "build"' }, + ], + }, + }); + await expect(client.v1.models.list()).resolves.toEqual([]); + }), + ); + + test( + "every write takes a later revision, store-wide, and list reports each model's last", + withClient(options, async (client) => { + const first = await client.v1.models.save({ + modelId: "b", + model: pairModel, + }); + const second = await client.v1.models.save({ + modelId: "a", + model: pairModel, + }); + const third = await client.v1.overlay.set({ + modelId: "b", + overlay: { plan: "done" }, + }); + + expect(second.revision).toBeGreaterThan(first.revision); + expect(third.revision).toBeGreaterThan(second.revision); + await expect(client.v1.models.list()).resolves.toEqual([ + { modelId: "a", revision: second.revision }, + { modelId: "b", revision: third.revision }, + ]); + }), + ); + + test( + "saving over a model replaces it and drops the run state of elements it no longer declares", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "pair", model: pairModel }); + await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "done", build: "unknown" }, + }); + const smaller: Model = { + ...pairModel, + elements: pairModel.elements.filter(({ id }) => id === "plan"), + transitions: [], + }; + const { revision } = await client.v1.models.save({ + modelId: "pair", + model: smaller, + }); + + await expect( + client.v1.models.load({ modelId: "pair" }), + ).resolves.toStrictEqual({ + modelId: "pair", + revision, + model: smaller, + overlay: { plan: "done" }, + }); + }), + ); + + test( + "an overlay sets the elements it names and leaves the rest", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "pair", model: pairModel }); + await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "open", build: "unknown" }, + }); + const { revision } = await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "done" }, + }); + + await expect( + client.v1.models.load({ modelId: "pair" }), + ).resolves.toMatchObject({ + revision, + overlay: { plan: "done", build: "unknown" }, + }); + }), + ); + + test( + "an overlay naming an element the model lacks is rejected and changes nothing", + withClient(options, async (client) => { + const { revision } = await client.v1.models.save({ + modelId: "pair", + model: pairModel, + }); + + await expect( + failureOf( + client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "done", ghost: "done" }, + }), + ), + ).resolves.toMatchObject({ + code: "UNPROCESSABLE_CONTENT", + data: { + issues: [ + { + code: "overlay-unknown-element", + message: 'overlay refers to unknown element id "ghost"', + }, + ], + }, + }); + await expect( + client.v1.models.load({ modelId: "pair" }), + ).resolves.toMatchObject({ revision, overlay: {} }); + }), + ); + + test( + "an overlay for a model that does not exist is NOT_FOUND", + withClient(options, async (client) => { + await expect( + failureOf( + client.v1.overlay.set({ + modelId: "missing", + overlay: { plan: "done" }, + }), + ), + ).resolves.toMatchObject({ + code: "NOT_FOUND", + data: { modelId: "missing" }, + }); + }), + ); + + test( + "deleting removes a model once, and it no longer loads", + withClient(options, async (client) => { + const saved = await client.v1.models.save({ + modelId: "pair", + model: pairModel, + }); + const deleted = await client.v1.models.delete({ modelId: "pair" }); + + expect(deleted.deleted).toBe(true); + expect(deleted.deleted && deleted.revision).toBeGreaterThan( + saved.revision, + ); + await expect( + client.v1.models.delete({ modelId: "pair" }), + ).resolves.toEqual({ deleted: false }); + await expect( + failureOf(client.v1.models.load({ modelId: "pair" })), + ).resolves.toMatchObject({ code: "NOT_FOUND" }); + }), + ); + + test( + "upserting items into a missing model creates it from the items", + withClient(options, async (client) => { + const { revision } = await client.v1.items.upsert({ + modelId: "issues", + items: issueItems, + }); + + await expect( + client.v1.models.load({ modelId: "issues" }), + ).resolves.toStrictEqual({ + modelId: "issues", + revision, + model: { + participants: issueItems.participants, + events: [], + actions: [], + artefacts: [], + elements: issueItems.activities, + transitions: issueItems.transitions, + fragments: [], + projections: [], + }, + overlay: issueItems.overlay, + }); + }), + ); + + test( + "upserting replaces records with the same id in place and appends the rest", + withClient(options, async (client) => { + const model: Model = { + ...pairModel, + elements: [ + ...pairModel.elements, + { + kind: "activity", + id: "github:org/repo#1", + label: "Old title", + source: "github:org/repo#1", + }, + ], + }; + await client.v1.models.save({ modelId: "mixed", model }); + await client.v1.overlay.set({ + modelId: "mixed", + overlay: { plan: "done" }, + }); + await client.v1.items.upsert({ modelId: "mixed", items: issueItems }); + + await expect( + client.v1.models.load({ modelId: "mixed" }), + ).resolves.toMatchObject({ + model: { + participants: [ + { id: "team", label: "Platform team", kind: "role" }, + { id: "github:org/repo", label: "org/repo", kind: "system" }, + ], + elements: [ + pairModel.elements[0], + pairModel.elements[1], + issueItems.activities[0], + issueItems.activities[1], + ], + transitions: [...pairModel.transitions, ...issueItems.transitions], + }, + overlay: { plan: "done", "github:org/repo#1": "done" }, + }); + }), + ); + + test( + "an upsert that would break the model is rejected and changes nothing", + withClient(options, async (client) => { + const { revision } = await client.v1.models.save({ + modelId: "pair", + model: pairModel, + }); + + await expect( + failureOf( + client.v1.items.upsert({ + modelId: "pair", + items: { + participants: [], + activities: [], + transitions: [ + { id: "dangling", source: "plan", target: "github:x/y#9" }, + ], + overlay: {}, + }, + }), + ), + ).resolves.toMatchObject({ + code: "UNPROCESSABLE_CONTENT", + data: { + issues: [ + { + code: "unknown-reference", + message: + 'transition "dangling" target refers to unknown id "github:x/y#9"', + }, + ], + }, + }); + await expect( + client.v1.models.load({ modelId: "pair" }), + ).resolves.toStrictEqual({ + modelId: "pair", + revision, + model: pairModel, + overlay: {}, + }); + }), + ); + + test( + "a bound activity whose id is not its source is refused before it reaches the store", + withClient(options, async (client) => { + await expect( + failureOf( + client.v1.items.upsert({ + modelId: "issues", + items: { + ...issueItems, + activities: [ + { + kind: "activity", + id: "renamed", + label: "Renamed", + source: "github:org/repo#1", + }, + ], + }, + }), + ), + ).resolves.toMatchObject({ code: "BAD_REQUEST" }); + await expect(client.v1.models.list()).resolves.toEqual([]); + }), + ); + + test( + "ready lists open work with nothing unfinished upstream, passing through control elements and loop edges", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "q", model: queryModel }); + await client.v1.overlay.set({ + modelId: "q", + overlay: { a: "done" }, + }); + + await expect(client.v1.query.ready({ modelId: "q" })).resolves.toEqual({ + elements: ["b", "x"], + }); + + await client.v1.overlay.set({ + modelId: "q", + overlay: { c: "done", x: "unknown" }, + }); + + await expect(client.v1.query.ready({ modelId: "q" })).resolves.toEqual({ + elements: ["b", "e"], + }); + }), + ); + + test( + "the blocked chain is the unfinished work upstream, nearest first, stopping at done work", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "q", model: queryModel }); + await client.v1.overlay.set({ + modelId: "q", + overlay: { a: "done" }, + }); + + await expect( + client.v1.query.blockedChain({ modelId: "q", element: "f" }), + ).resolves.toEqual({ + chain: [ + { element: "e", distance: 2 }, + { element: "x", distance: 2 }, + { element: "c", distance: 4 }, + { element: "b", distance: 5 }, + ], + }); + await expect( + client.v1.query.blockedChain({ modelId: "q", element: "b" }), + ).resolves.toEqual({ chain: [] }); + await expect( + client.v1.query.blockedChain({ modelId: "q", element: "a" }), + ).resolves.toEqual({ chain: [] }); + }), + ); + + test( + "impact is the unfinished work downstream, nearest first, whatever the element's own state", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "q", model: queryModel }); + await client.v1.overlay.set({ + modelId: "q", + overlay: { a: "done", e: "done" }, + }); + + await expect( + client.v1.query.impact({ modelId: "q", element: "b" }), + ).resolves.toEqual({ impacted: [{ element: "c", distance: 1 }] }); + await expect( + client.v1.query.impact({ modelId: "q", element: "a" }), + ).resolves.toEqual({ + impacted: [ + { element: "b", distance: 1 }, + { element: "c", distance: 2 }, + ], + }); + await expect( + client.v1.query.impact({ modelId: "q", element: "x" }), + ).resolves.toEqual({ impacted: [{ element: "f", distance: 2 }] }); + }), + ); + + test( + "a query about an element the model lacks is NOT_FOUND naming it", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "q", model: queryModel }); + + await expect( + failureOf(client.v1.query.impact({ modelId: "q", element: "ghost" })), + ).resolves.toMatchObject({ + code: "NOT_FOUND", + data: { modelId: "q", element: "ghost" }, + }); + await expect( + failureOf(client.v1.query.ready({ modelId: "missing" })), + ).resolves.toMatchObject({ + code: "NOT_FOUND", + data: { modelId: "missing" }, + }); + }), + ); + + test( + "a write whose signal has already fired is refused and changes nothing", + withClient(options, async (client) => { + const controller = new AbortController(); + controller.abort(); + + await expect( + client.v1.models.save( + { modelId: "pair", model: pairModel }, + { signal: controller.signal }, + ), + ).rejects.toThrow(); + await expect(client.v1.models.list()).resolves.toEqual([]); + }), + ); + + test( + "a subscriber receives every later change to its model, in order, and none of another's", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "pair", model: pairModel }); + const events = await client.v1.changes({ modelId: "pair" }); + const received = takeUntil(events, ({ kind }) => kind === "deleted"); + + const overlay = await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "done" }, + }); + await client.v1.models.save({ modelId: "other", model: pairModel }); + const items = await client.v1.items.upsert({ + modelId: "pair", + items: issueItems, + }); + const saved = await client.v1.models.save({ + modelId: "pair", + model: pairModel, + }); + const deleted = await client.v1.models.delete({ modelId: "pair" }); + + await expect(received).resolves.toEqual([ + { modelId: "pair", kind: "overlay", revision: overlay.revision }, + { modelId: "pair", kind: "items", revision: items.revision }, + { modelId: "pair", kind: "saved", revision: saved.revision }, + { + modelId: "pair", + kind: "deleted", + revision: deleted.deleted ? deleted.revision : undefined, + }, + ]); + }), + ); + + test( + "a subscriber that names a revision the model has moved past is told it missed changes", + withClient(options, async (client) => { + const first = await client.v1.models.save({ + modelId: "pair", + model: pairModel, + }); + const second = await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: "done" }, + }); + + const behind = await client.v1.changes({ + modelId: "pair", + since: first.revision, + }); + await expect(firstOf(behind)).resolves.toEqual([ + { modelId: "pair", kind: "missed" }, + ]); + + const current = await client.v1.changes({ + modelId: "pair", + since: second.revision, + }); + const next = firstOf(current); + const third = await client.v1.overlay.set({ + modelId: "pair", + overlay: { build: "done" }, + }); + await expect(next).resolves.toEqual([ + { modelId: "pair", kind: "overlay", revision: third.revision }, + ]); + }), + ); + }); +}; diff --git a/packages/core/graph/store-contract/src/contract.ts b/packages/core/graph/store-contract/src/contract.ts new file mode 100644 index 0000000..e02488c --- /dev/null +++ b/packages/core/graph/store-contract/src/contract.ts @@ -0,0 +1,125 @@ +import { type ContractRouterClient, eventIterator, oc } from "@orpc/contract"; +import { modelSchema, overlaySchema } from "@exadev/process-graph-schema"; +import { z } from "zod"; +import { + boundItemsSchema, + issueSchema, + modelIdSchema, + modelSummarySchema, + reachSchema, + revisionSchema, + storedModelSchema, + storeChangeSchema, +} from "./schema.ts"; + +/** + * The typed errors a procedure can end with, each carrying its detail as data. The codes are oRPC's common codes, so an HTTP transport answers with the matching status (404, 422) without any per-transport mapping. + */ +const graphStoreErrors = { + NOT_FOUND: { + message: "No such model, or no such element in it.", + data: z.strictObject({ + modelId: z.string(), + element: z.string().exactOptional(), + }), + }, + UNPROCESSABLE_CONTENT: { + message: "The change would leave the model or its overlay invalid.", + data: z.strictObject({ issues: z.array(issueSchema).readonly() }), + }, +}; + +const modelRef = z.strictObject({ modelId: modelIdSchema }).readonly(); +const elementRef = z + .strictObject({ modelId: modelIdSchema, element: z.string() }) + .readonly(); +const written = z.strictObject({ revision: revisionSchema }).readonly(); + +/** + * Version 1 of the `GraphStore` port. A breaking change adds a sibling `v2` beside it, so one server can serve both while clients move across. + * + * - `models.list`, `models.save`, `models.load` and `models.delete` keep whole models. A save replaces the model and drops the run state of elements it no longer declares. + * - `overlay.set` sets the run state of the elements it names. + * - `items.upsert` merges items read from a source into a model by identifier, creating the model when it does not exist. + * - `query.ready`, `query.blockedChain` and `query.impact` answer structural questions from the stored transitions and run state alone (see the package README for exactly what they read), so a store can answer them without loading the model or evaluating conditions. + * - `changes` yields every later change to one model as an event iterator. + * + * Every write is atomic and checked: a change that would leave the model or overlay invalid (`validate`, `validateOverlay`) is rejected whole with `UNPROCESSABLE_CONTENT`. Every call takes the caller's abort signal through the oRPC client options; a call aborted before it writes leaves the store unchanged. + */ +const graphStoreV1Contract = oc.errors(graphStoreErrors).router({ + models: { + list: oc.output(z.array(modelSummarySchema).readonly()), + save: oc + .input( + z + .strictObject({ modelId: modelIdSchema, model: modelSchema }) + .readonly(), + ) + .output(written), + load: oc.input(modelRef).output(storedModelSchema), + delete: oc.input(modelRef).output( + z.union([ + z + .strictObject({ + deleted: z.literal(true), + revision: revisionSchema, + }) + .readonly(), + z.strictObject({ deleted: z.literal(false) }).readonly(), + ]), + ), + }, + overlay: { + set: oc + .input( + z + .strictObject({ modelId: modelIdSchema, overlay: overlaySchema }) + .readonly(), + ) + .output(written), + }, + items: { + upsert: oc + .input( + z + .strictObject({ modelId: modelIdSchema, items: boundItemsSchema }) + .readonly(), + ) + .output(written), + }, + query: { + ready: oc + .input(modelRef) + .output( + z.strictObject({ elements: z.array(z.string()).readonly() }).readonly(), + ), + blockedChain: oc + .input(elementRef) + .output( + z.strictObject({ chain: z.array(reachSchema).readonly() }).readonly(), + ), + impact: oc + .input(elementRef) + .output( + z + .strictObject({ impacted: z.array(reachSchema).readonly() }) + .readonly(), + ), + }, + changes: oc + .input( + z + .strictObject({ + modelId: modelIdSchema, + since: revisionSchema.exactOptional(), + }) + .readonly(), + ) + .output(eventIterator(storeChangeSchema)), +}); + +/** The `GraphStore` port: an oRPC router contract, keyed by version. */ +export const graphStoreContract = { v1: graphStoreV1Contract }; + +/** A client for the `GraphStore` port over any transport: an in-process router client or one built on an oRPC link. */ +export type GraphStoreClient = ContractRouterClient; diff --git a/packages/core/graph/store-contract/src/errors.ts b/packages/core/graph/store-contract/src/errors.ts new file mode 100644 index 0000000..fa84989 --- /dev/null +++ b/packages/core/graph/store-contract/src/errors.ts @@ -0,0 +1,38 @@ +import type { Issue } from "@exadev/process-graph-schema"; + +/** Why a store operation failed, as data. */ +export type GraphStoreFailure = + | { + /** The model does not exist, or `element` is not one of its elements. */ + readonly kind: "not-found"; + readonly modelId: string; + readonly element?: string; + } + | { + /** The change would leave the model or its overlay invalid, as `validate` and `validateOverlay` report it; nothing was written. */ + readonly kind: "invalid"; + readonly issues: readonly Issue[]; + }; + +/** + * The one error a `GraphStore` implementation throws for an expected failure. `createGraphStoreRouter` turns it into the contract's typed oRPC error, so every store reports failures the same way over every transport. + */ +export class GraphStoreError extends Error { + /** What went wrong. */ + readonly failure: GraphStoreFailure; + + /** + * @param failure - What went wrong. + */ + constructor(failure: GraphStoreFailure) { + super( + failure.kind === "not-found" + ? failure.element === undefined + ? `No model "${failure.modelId}"` + : `Model "${failure.modelId}" has no element "${failure.element}"` + : `Invalid change: ${failure.issues.map((issue) => issue.message).join("; ")}`, + ); + this.name = "GraphStoreError"; + this.failure = failure; + } +} diff --git a/packages/core/graph/store-contract/src/fake.ts b/packages/core/graph/store-contract/src/fake.ts new file mode 100644 index 0000000..eee03e8 --- /dev/null +++ b/packages/core/graph/store-contract/src/fake.ts @@ -0,0 +1,163 @@ +import type { Model, OpenIds, Overlay } from "@exadev/process-graph-schema"; +import { GraphStoreError } from "./errors.ts"; +import { ChangeFeed } from "./feed.ts"; +import { + checkStored, + emptyModel, + mergeBoundItems, + overlayWithin, +} from "./merge.ts"; +import type { ChangeKind, StoredModel } from "./schema.ts"; +import type { GraphStore } from "./store.ts"; +import { blockedChainOf, impactOf, readyOf, structureOf } from "./structure.ts"; + +/** Runs a synchronous store operation as the async one the port expects; a throw becomes a rejection. */ +const later = + (operation: (request: A) => R) => + async (request: A): Promise => + Promise.resolve(request).then(operation); + +/** One stored entry: everything `models.load` returns. */ +interface Entry { + readonly model: Model; + readonly overlay: Overlay; + readonly revision: number; +} + +const encoder = new TextEncoder(); + +/** + * Orders strings by their UTF-8 bytes, which is Unicode code point order and the order SQLite's default `BINARY` collation gives text. JavaScript's own `<` compares UTF-16 code units, which puts characters outside the Basic Multilingual Plane before those from U+E000 up. + */ +const byCodePoint = (left: string, right: string): number => { + const theirs = encoder.encode(right).values(); + for (const byte of encoder.encode(left)) { + const other = theirs.next(); + if (other.done === true) { + return 1; + } + if (byte !== other.value) { + return byte - other.value; + } + } + + return theirs.next().done === true ? 0 : -1; +}; + +/** + * The reference in-memory store: the behaviour every adapter's conformance run is compared with. Models are held as values and replaced on every write, so a failed check leaves the previous value in place. + * @returns A store with no models, revisions starting after zero. + */ +export const createFakeGraphStore = (): GraphStore => { + const feed = new ChangeFeed(); + const entries = new Map(); + let sequence = 0; + + const entryOf = (modelId: string): Entry => { + const entry = entries.get(modelId); + if (entry === undefined) { + throw new GraphStoreError({ kind: "not-found", modelId }); + } + + return entry; + }; + + const elementOf = (modelId: string, element: string): Entry => { + const entry = entryOf(modelId); + if (!entry.model.elements.some(({ id }) => id === element)) { + throw new GraphStoreError({ kind: "not-found", modelId, element }); + } + + return entry; + }; + + /** What one write stores and reports. */ + interface Write { + readonly modelId: string; + readonly kind: ChangeKind; + readonly model: Model; + readonly overlay: Overlay; + readonly signal: AbortSignal | undefined; + } + + const write = ({ modelId, kind, model, overlay, signal }: Write): number => { + checkStored(model, overlay); + signal?.throwIfAborted(); + sequence += 1; + entries.set(modelId, { model, overlay, revision: sequence }); + feed.publish({ modelId, kind, revision: sequence }); + + return sequence; + }; + + return { + feed, + listModels: later(() => + [...entries] + .map(([modelId, { revision }]) => ({ modelId, revision })) + .sort((left, right) => byCodePoint(left.modelId, right.modelId)), + ), + saveModel: later(({ modelId, model, signal }) => + write({ + modelId, + kind: "saved", + model, + overlay: overlayWithin(model, entries.get(modelId)?.overlay ?? {}), + signal, + }), + ), + loadModel: later(({ modelId }): StoredModel => { + const { model, overlay, revision } = entryOf(modelId); + + return { modelId, revision, model, overlay }; + }), + deleteModel: later(({ modelId, signal }) => { + if (!entries.has(modelId)) { + return undefined; + } + signal?.throwIfAborted(); + entries.delete(modelId); + sequence += 1; + feed.publish({ modelId, kind: "deleted", revision: sequence }); + + return sequence; + }), + applyOverlay: later(({ modelId, overlay, signal }) => { + const entry = entryOf(modelId); + + return write({ + modelId, + kind: "overlay", + model: entry.model, + overlay: { ...entry.overlay, ...overlay }, + signal, + }); + }), + upsertItems: later(({ modelId, items, signal }) => { + const entry = entries.get(modelId) ?? { model: emptyModel, overlay: {} }; + + return write({ + modelId, + kind: "items", + ...mergeBoundItems(entry.model, entry.overlay, items), + signal, + }); + }), + ready: later(({ modelId }) => { + const { model, overlay } = entryOf(modelId); + + return readyOf(structureOf(model, overlay)); + }), + blockedChain: later(({ modelId, element }) => { + const { model, overlay } = elementOf(modelId, element); + + return blockedChainOf(structureOf(model, overlay), element); + }), + impact: later(({ modelId, element }) => { + const { model, overlay } = elementOf(modelId, element); + + return impactOf(structureOf(model, overlay), element); + }), + revisionOf: later(({ modelId }) => entries.get(modelId)?.revision), + }; +}; diff --git a/packages/core/graph/store-contract/src/fake.unit.test.ts b/packages/core/graph/store-contract/src/fake.unit.test.ts new file mode 100644 index 0000000..72ac5e5 --- /dev/null +++ b/packages/core/graph/store-contract/src/fake.unit.test.ts @@ -0,0 +1,45 @@ +import { createORPCClient } from "@orpc/client"; +import { RPCLink } from "@orpc/client/message-port"; +import { createRouterClient } from "@orpc/server"; +import { RPCHandler } from "@orpc/server/message-port"; +import { MessageChannel } from "node:worker_threads"; +import { graphStoreConformance } from "./conformance.ts"; +import type { GraphStoreClient } from "./contract.ts"; +import { createFakeGraphStore } from "./fake.ts"; +import { createGraphStoreRouter } from "./router.ts"; + +graphStoreConformance({ + name: "fake, in process", + createClient: async () => + Promise.resolve({ + client: createRouterClient( + createGraphStoreRouter(createFakeGraphStore()), + ), + close: async () => Promise.resolve(), + }), +}); + +graphStoreConformance({ + name: "fake, over a message port", + createClient: async () => { + const { port1, port2 } = new MessageChannel(); + new RPCHandler(createGraphStoreRouter(createFakeGraphStore())).upgrade( + port1, + ); + port1.start(); + const client: GraphStoreClient = createORPCClient( + new RPCLink({ port: port2 }), + ); + port2.start(); + + return Promise.resolve({ + client, + close: async () => { + port1.close(); + port2.close(); + + return Promise.resolve(); + }, + }); + }, +}); diff --git a/packages/core/graph/store-contract/src/feed.ts b/packages/core/graph/store-contract/src/feed.ts new file mode 100644 index 0000000..6b2f26d --- /dev/null +++ b/packages/core/graph/store-contract/src/feed.ts @@ -0,0 +1,101 @@ +import type { StoreChange } from "./schema.ts"; + +/** A live subscription to one model's changes. Iterate it to receive them; `close` ends it. */ +export interface ChangeSubscription extends AsyncIterable { + /** Stops delivery and ends the iteration. Idempotent. */ + readonly close: () => void; +} + +/** + * Fans the changes one store makes out to every subscriber of the model they touch. A store publishes after each committed write; the router reads a subscription for `changes`. + * + * A subscription buffers changes that arrive while its consumer is busy, so none is dropped between two pulls, and it is registered the moment `subscribe` returns, so a caller can subscribe first and then read the current revision without missing a change made in between. + * + * A feed only sees the writes made through the store that owns it. Several processes or tabs writing to one database each need their own bridge (a `BroadcastChannel`, a database notification) that publishes the others' changes into their feed. + */ +export class ChangeFeed { + readonly #listeners = new Map void>>(); + + /** + * Delivers a change to every current subscriber of its model. + * @param change - The change, after it is committed. + */ + publish(change: StoreChange): void { + for (const listener of this.#listeners.get(change.modelId) ?? []) { + listener(change); + } + } + + /** + * Subscribes to one model's changes from now on. + * @param modelId - The model to follow. + * @param signal - Ends the subscription when it fires. + * @returns The live subscription. + */ + subscribe(modelId: string, signal?: AbortSignal): ChangeSubscription { + const buffered: StoreChange[] = []; + // The pull waiting for the next change, when the buffer was empty as it asked. + let waiting: + ((result: IteratorResult) => void) | undefined; + let closed = false; + const listeners = this.#listeners.get(modelId) ?? new Set(); + this.#listeners.set(modelId, listeners); + const finished: IteratorResult = { + done: true, + value: undefined, + }; + + const listener = (change: StoreChange): void => { + if (waiting === undefined) { + buffered.push(change); + } else { + waiting({ done: false, value: change }); + waiting = undefined; + } + }; + const close = (): void => { + if (closed) { + return; + } + closed = true; + listeners.delete(listener); + if (listeners.size === 0) { + this.#listeners.delete(modelId); + } + signal?.removeEventListener("abort", close); + waiting?.(finished); + waiting = undefined; + }; + listeners.add(listener); + signal?.addEventListener("abort", close); + if (signal?.aborted === true) { + close(); + } + + const iterator: AsyncIterator = { + next: async () => { + const change = buffered.shift(); + if (change !== undefined) { + return { done: false, value: change }; + } + if (closed) { + return finished; + } + + return new Promise((resolve) => { + waiting = resolve; + }); + }, + return: async () => { + close(); + + return Promise.resolve(finished); + }, + }; + + return { + close, + [Symbol.asyncIterator]: () => iterator, + }; + } +} diff --git a/packages/core/graph/store-contract/src/feed.unit.test.ts b/packages/core/graph/store-contract/src/feed.unit.test.ts new file mode 100644 index 0000000..dfc3f49 --- /dev/null +++ b/packages/core/graph/store-contract/src/feed.unit.test.ts @@ -0,0 +1,67 @@ +import { describe, expect, test } from "vitest"; +import { ChangeFeed } from "./feed.ts"; + +const saved = (modelId: string, revision: number) => + ({ modelId, kind: "saved", revision }) as const; + +describe("ChangeFeed", () => { + test("buffers changes published before the subscriber pulls, in order", async () => { + const feed = new ChangeFeed(); + const subscription = feed.subscribe("m"); + const [mine, theirs, later] = ["m", "other", "m"].map((modelId, index) => + saved(modelId, index + 1), + ); + for (const change of [mine, theirs, later]) { + if (change !== undefined) { + feed.publish(change); + } + } + subscription.close(); + + const received = []; + for await (const change of subscription) { + received.push(change); + } + + expect(received).toEqual([mine, later]); + }); + + test("delivers a change published while the subscriber waits", async () => { + const feed = new ChangeFeed(); + const iterator = feed.subscribe("m")[Symbol.asyncIterator](); + const next = iterator.next(); + feed.publish(saved("m", 1)); + + await expect(next).resolves.toEqual({ done: false, value: saved("m", 1) }); + }); + + test("ends a waiting subscriber when its signal fires, and stops delivering to it", async () => { + const feed = new ChangeFeed(); + const controller = new AbortController(); + const iterator = feed + .subscribe("m", controller.signal) + [Symbol.asyncIterator](); + const next = iterator.next(); + controller.abort(); + + await expect(next).resolves.toEqual({ done: true, value: undefined }); + feed.publish(saved("m", 1)); + await expect(iterator.next()).resolves.toEqual({ + done: true, + value: undefined, + }); + }); + + test("ends at once for a signal that has already fired", async () => { + const controller = new AbortController(); + controller.abort(); + const feed = new ChangeFeed(); + const subscription = feed.subscribe("m", controller.signal); + feed.publish(saved("m", 1)); + + await expect(subscription[Symbol.asyncIterator]().next()).resolves.toEqual({ + done: true, + value: undefined, + }); + }); +}); diff --git a/packages/core/graph/store-contract/src/fixtures.ts b/packages/core/graph/store-contract/src/fixtures.ts new file mode 100644 index 0000000..ce447e5 --- /dev/null +++ b/packages/core/graph/store-contract/src/fixtures.ts @@ -0,0 +1,145 @@ +import type { Model, OpenIds } from "@exadev/process-graph-schema"; + +/** + * A model that uses every field the model schema has, each optional field present on some record and absent on another, so a store that drops, renames or reorders anything fails the round trip. + */ +export const everyFieldModel: Model = { + participants: [ + { id: "customer", label: "Customer", kind: "role" }, + { id: "shop", label: "Shop", kind: "system" }, + { id: "order", label: "Order", kind: "entity" }, + ], + events: [ + { id: "placed", name: "Order placed", payload: { total: "number" } }, + { id: "paid", name: "Payment received" }, + ], + actions: [ + { id: "notify", name: "Notify customer" }, + { id: "log", name: "Write audit log" }, + ], + artefacts: [ + { id: "basket", name: "Basket" }, + { id: "invoice", name: "Invoice", deliverable: true }, + { id: "receipt", name: "Receipt", deliverable: false }, + ], + elements: [ + { kind: "pseudo", id: "start", role: "initial" }, + { + kind: "activity", + id: "github:shop/orders#1", + label: "Place order", + actor: "customer", + action: "notify", + source: "github:shop/orders#1", + affects: ["order"], + consumes: ["basket"], + produces: ["invoice"], + extensions: { bpmn: { pool: "Customer", lane: 1, marker: null } }, + }, + { kind: "decision", id: "approve", label: "Approve?", actor: "shop" }, + { + kind: "state", + id: "pending", + label: "Pending", + parent: "start", + entry: "log", + exit: "notify", + }, + { kind: "state", id: "closed", label: "Closed" }, + { + kind: "message", + id: "confirm", + label: "Confirm", + from: "shop", + to: "customer", + mode: "async", + event: "paid", + }, + { + kind: "message", + id: "ack", + label: "Acknowledge", + from: "customer", + to: "shop", + mode: "reply", + }, + { kind: "activity", id: "ship", label: "Ship" }, + { kind: "pseudo", id: "end", role: "final", parent: "pending" }, + ], + transitions: [ + { id: "t-start", source: "start", target: "github:shop/orders#1" }, + { + id: "t-review", + source: "github:shop/orders#1", + target: "approve", + trigger: "placed", + effect: "log", + }, + { + id: "t-yes", + source: "approve", + target: "ship", + guard: "approved", + condition: { + kind: "compare", + op: "eq", + left: { kind: "reference", key: { $fact: "approved" } }, + right: { kind: "booleanLiteral", value: true }, + }, + }, + { id: "t-no", source: "approve", target: "closed", guard: "rejected" }, + { id: "t-end", source: "ship", target: "end" }, + ], + fragments: [ + { + id: "decide", + operator: "alt", + operands: [{ guard: "approved", body: ["ship"] }, { body: ["closed"] }], + }, + { id: "chat", operator: "par", operands: [] }, + ], + projections: [ + { id: "flow", notation: "flowchart" }, + { id: "lanes", notation: "swimlane" }, + { id: "lifecycle", notation: "stateMachine", subject: "order" }, + { id: "happy", notation: "sequence", scenario: ["t-review", "t-yes"] }, + ], + views: { + flowchart: { start: { x: 0, y: 0 }, ship: { x: 120.5, y: -40 } }, + sequence: {}, + }, +}; + +/** + * A model for the structural queries: a done first step, a chain through a decision whose retry edge loops back, a join waiting on two branches, and an unrelated step. + * + * `a` (done) leads to `b`, `b` to `c`, `c` to the decision `d`, which either approves into `e` or retries back to `b` (a loop edge, so it is no dependency). `e` and the independent `x` meet at the join `j`, which leads to `f`. + */ +export const queryModel: Model = { + participants: [{ id: "team", label: "Team", kind: "role" }], + events: [], + actions: [], + artefacts: [], + elements: [ + { kind: "activity", id: "a", label: "A", actor: "team" }, + { kind: "activity", id: "b", label: "B", actor: "team" }, + { kind: "activity", id: "c", label: "C", actor: "team" }, + { kind: "decision", id: "d", label: "D?" }, + { kind: "activity", id: "e", label: "E", actor: "team" }, + { kind: "activity", id: "x", label: "X", actor: "team" }, + { kind: "pseudo", id: "j", role: "join" }, + { kind: "activity", id: "f", label: "F", actor: "team" }, + ], + transitions: [ + { id: "ab", source: "a", target: "b" }, + { id: "bc", source: "b", target: "c" }, + { id: "cd", source: "c", target: "d" }, + { id: "de", source: "d", target: "e", guard: "approved" }, + { id: "db", source: "d", target: "b", guard: "retry" }, + { id: "ej", source: "e", target: "j" }, + { id: "xj", source: "x", target: "j" }, + { id: "jf", source: "j", target: "f" }, + ], + fragments: [], + projections: [], +}; diff --git a/packages/core/graph/store-contract/src/index.ts b/packages/core/graph/store-contract/src/index.ts new file mode 100644 index 0000000..df64f95 --- /dev/null +++ b/packages/core/graph/store-contract/src/index.ts @@ -0,0 +1,45 @@ +/** + * The `GraphStore` port: how process models and their run state are kept. It is an oRPC router contract, so one definition is served in process, across a message port to a worker, or over HTTP with no change to its consumers. Adapters implement `GraphStore`; `createGraphStoreRouter` builds the port on it. + * + * The conformance kit every adapter must pass is the separate `./conformance` entry point, so the runtime entry carries no test framework. + */ +export { graphStoreContract, type GraphStoreClient } from "./contract.ts"; +export { + boundActivitySchema, + boundItemsSchema, + changeKinds, + issueSchema, + modelIdSchema, + modelSummarySchema, + reachSchema, + revisionSchema, + storeChangeSchema, + storedModelSchema, + type BoundActivity, + type BoundItems, + type ChangeKind, + type ModelId, + type ModelSummary, + type Reach, + type Revision, + type StoreChange, + type StoredModel, +} from "./schema.ts"; +export { GraphStoreError, type GraphStoreFailure } from "./errors.ts"; +export { ChangeFeed, type ChangeSubscription } from "./feed.ts"; +export type { ElementRequest, GraphStore, ModelRequest } from "./store.ts"; +export { + checkStored, + emptyModel, + mergeBoundItems, + overlayWithin, +} from "./merge.ts"; +export { + blockedChainOf, + impactOf, + readyOf, + structureOf, + type Structure, +} from "./structure.ts"; +export { createGraphStoreRouter, type GraphStoreRouter } from "./router.ts"; +export { createFakeGraphStore } from "./fake.ts"; diff --git a/packages/core/graph/store-contract/src/merge.ts b/packages/core/graph/store-contract/src/merge.ts new file mode 100644 index 0000000..9c20ee9 --- /dev/null +++ b/packages/core/graph/store-contract/src/merge.ts @@ -0,0 +1,89 @@ +import { + validate, + validateOverlay, + type Model, + type OpenIds, + type Overlay, +} from "@exadev/process-graph-schema"; +import { GraphStoreError } from "./errors.ts"; +import type { BoundItems } from "./schema.ts"; + +/** A model with nothing in it, which `items.upsert` starts from when the model does not exist yet. */ +export const emptyModel: Model = { + participants: [], + events: [], + actions: [], + artefacts: [], + elements: [], + transitions: [], + fragments: [], + projections: [], +}; + +/** Replaces each record of `current` that an incoming record shares an id with, in place, and appends the rest in their incoming order. */ +const upsertById = ( + current: readonly T[], + incoming: readonly T[], +): readonly T[] => { + const replacements = new Map(incoming.map((record) => [record.id, record])); + const known = new Set(current.map(({ id }) => id)); + + return [ + ...current.map((record) => replacements.get(record.id) ?? record), + ...incoming.filter(({ id }) => !known.has(id)), + ]; +}; + +/** + * Merges items read from a source into a model and its overlay: participants, activities and transitions by id, replacing a record in its place or appending a new one; overlay entries replace the run state of the elements they name. The result is not checked; see `checkStored`. + * @param model - The stored model. + * @param overlay - The stored overlay. + * @param items - The items to merge. + * @returns The merged model and overlay. + */ +export const mergeBoundItems = ( + model: Model, + overlay: Overlay, + items: BoundItems, +): { readonly model: Model; readonly overlay: Overlay } => ({ + model: { + ...model, + participants: upsertById(model.participants, items.participants), + elements: upsertById(model.elements, items.activities), + transitions: upsertById(model.transitions, items.transitions), + }, + overlay: { ...overlay, ...items.overlay }, +}); + +/** + * Keeps only the overlay entries that name an element of the model, so replacing a model drops the run state of elements it no longer declares. + * @param model - The model the overlay is for. + * @param overlay - The overlay to narrow. + * @returns The entries whose key is an element of `model`. + */ +export const overlayWithin = ( + model: Model, + overlay: Overlay, +): Overlay => { + const elements = new Set(model.elements.map(({ id }) => id)); + + return Object.fromEntries( + Object.entries(overlay).filter(([id]) => elements.has(id)), + ); +}; + +/** + * Checks a model and overlay a write is about to store. Every store runs this before committing, so a store never holds a model `validate` rejects or an overlay entry for an element the model does not declare. + * @param model - The model to store. + * @param overlay - The overlay to store with it. + * @throws GraphStoreError with `invalid` and every issue found, when there is any. + */ +export const checkStored = ( + model: Model, + overlay: Overlay, +): void => { + const issues = [...validate(model), ...validateOverlay(model, overlay)]; + if (issues.length > 0) { + throw new GraphStoreError({ kind: "invalid", issues }); + } +}; diff --git a/packages/core/graph/store-contract/src/router.ts b/packages/core/graph/store-contract/src/router.ts new file mode 100644 index 0000000..b04c262 --- /dev/null +++ b/packages/core/graph/store-contract/src/router.ts @@ -0,0 +1,154 @@ +import type { ORPCErrorConstructorMap } from "@orpc/contract"; +import { type ContractedRouter, implement } from "@orpc/server"; +import { graphStoreContract } from "./contract.ts"; +import { GraphStoreError } from "./errors.ts"; +import type { GraphStore } from "./store.ts"; + +const os = implement(graphStoreContract); + +type V1Errors = ORPCErrorConstructorMap< + (typeof graphStoreContract.v1.models.load)["~orpc"]["errorMap"] +>; + +/** Rethrows a store's `GraphStoreError` as the contract's typed oRPC error; anything else propagates unchanged. */ +const toContractError = ( + error: unknown, + errors: Readonly, +): unknown => { + if (!(error instanceof GraphStoreError)) { + return error; + } + const { failure } = error; + if (failure.kind === "not-found") { + return errors.NOT_FOUND({ + message: error.message, + data: + failure.element === undefined + ? { modelId: failure.modelId } + : { modelId: failure.modelId, element: failure.element }, + cause: error, + }); + } + + return errors.UNPROCESSABLE_CONTENT({ + message: error.message, + data: { issues: failure.issues }, + cause: error, + }); +}; + +/** Runs one store call: refuses an already-aborted signal before touching the store, and maps the store's failures to the contract's errors. */ +const call = async ( + signal: AbortSignal | undefined, + errors: Readonly, + operation: () => Promise, +): Promise => { + signal?.throwIfAborted(); + try { + return await operation(); + } catch (error) { + throw toContractError(error, errors); + } +}; + +/** + * Builds the `GraphStore` router over a store. Serve it over any oRPC transport, or call it in process with `createRouterClient`. + * + * `changes` subscribes to the store's feed before it reads the model's revision and before it returns, so no change is lost between the two, and a caller holding the iterator is already subscribed. When the caller names a revision `since` and the model is no longer at it, the first event is `missed`; every later change follows as it commits. + * @param store - The adapter. + * @returns The implemented router for `graphStoreContract`. + */ +export const createGraphStoreRouter = (store: GraphStore): GraphStoreRouter => + os.router({ + v1: { + models: { + list: os.v1.models.list.handler(async ({ signal, errors }) => + call(signal, errors, async () => store.listModels({ signal })), + ), + save: os.v1.models.save.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + revision: await store.saveModel({ ...input, signal }), + })), + ), + load: os.v1.models.load.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => + store.loadModel({ ...input, signal }), + ), + ), + delete: os.v1.models.delete.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => { + const revision = await store.deleteModel({ ...input, signal }); + + return revision === undefined + ? { deleted: false as const } + : { deleted: true as const, revision }; + }), + ), + }, + overlay: { + set: os.v1.overlay.set.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + revision: await store.applyOverlay({ ...input, signal }), + })), + ), + }, + items: { + upsert: os.v1.items.upsert.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + revision: await store.upsertItems({ ...input, signal }), + })), + ), + }, + query: { + ready: os.v1.query.ready.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + elements: await store.ready({ ...input, signal }), + })), + ), + blockedChain: os.v1.query.blockedChain.handler( + async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + chain: await store.blockedChain({ ...input, signal }), + })), + ), + impact: os.v1.query.impact.handler(async ({ input, signal, errors }) => + call(signal, errors, async () => ({ + impacted: await store.impact({ ...input, signal }), + })), + ), + }, + changes: os.v1.changes.handler(async ({ input, signal, errors }) => { + // Subscribe before anything is awaited and before the iterator is returned: a caller that has the iterator in hand is already subscribed, over any transport, so a write it makes next is never missed. + const subscription = store.feed.subscribe(input.modelId, signal); + let missed = false; + try { + if (input.since !== undefined) { + const current = await call(signal, errors, async () => + store.revisionOf({ modelId: input.modelId, signal }), + ); + missed = current !== input.since; + } + } catch (error) { + subscription.close(); + throw error; + } + + return (async function* () { + try { + if (missed) { + yield { modelId: input.modelId, kind: "missed" as const }; + } + yield* subscription; + } finally { + subscription.close(); + } + })(); + }), + }, + }); + +/** The router `createGraphStoreRouter` builds: every procedure of `graphStoreContract`, needing no initial context. */ +export type GraphStoreRouter = ContractedRouter< + typeof graphStoreContract, + object +>; diff --git a/packages/core/graph/store-contract/src/schema.ts b/packages/core/graph/store-contract/src/schema.ts new file mode 100644 index 0000000..b235ad6 --- /dev/null +++ b/packages/core/graph/store-contract/src/schema.ts @@ -0,0 +1,96 @@ +import { modelSchema, overlaySchema } from "@exadev/process-graph-schema"; +import { z } from "zod"; + +/** + * The model's own record schemas, taken from `modelSchema` rather than restated, so what the store accepts is exactly a model or a slice of one and can never drift from it. + */ +const modelShape = modelSchema.unwrap().shape; + +/** The model's `activity` element: the first variant of its element union. A type test pins the order, so reordering the union is a compile error. */ +const [activityElementSchema] = modelShape.elements.unwrap().element.options; + +/** The name a store files a model under. Any non-empty string; the store gives it no structure. */ +export const modelIdSchema = z.string().min(1); + +/** + * A position in the store's history. Every change to any model takes the next revision from one store-wide sequence, so revisions only grow, a model's revision is the revision of its last change, and a model deleted and saved again never reuses an earlier one. + */ +export const revisionSchema = z.int().positive(); + +/** + * An activity bound to the external record it stands for. `source` is required here, where the model leaves it optional: a bound item's identity is its source record, and the source-qualified identifier is its element id. + */ +export const boundActivitySchema = activityElementSchema + .unwrap() + .extend({ source: z.string().min(1) }) + .refine((activity) => activity.id === activity.source, { + message: "a bound activity's id must equal its source", + path: ["id"], + }) + .readonly(); + +/** + * Items read from a source, to be merged into a stored model by identifier. Each participant, activity and transition replaces the one with the same id or is appended after the existing ones; the overlay's entries replace the run state of the elements they name. Everything else in the model is left as it is. + */ +export const boundItemsSchema = z + .strictObject({ + participants: z.array(modelShape.participants.unwrap().element).readonly(), + activities: z.array(boundActivitySchema).readonly(), + transitions: z.array(modelShape.transitions.unwrap().element).readonly(), + overlay: overlaySchema, + }) + .readonly(); + +/** One stored model and the run state laid over it, at the revision of its last change. */ +export const storedModelSchema = z + .strictObject({ + modelId: modelIdSchema, + revision: revisionSchema, + model: modelSchema, + overlay: overlaySchema, + }) + .readonly(); + +/** A stored model's name and the revision of its last change. */ +export const modelSummarySchema = z + .strictObject({ modelId: modelIdSchema, revision: revisionSchema }) + .readonly(); + +/** An element reached by a traversal and its distance from the start, in transitions. */ +export const reachSchema = z + .strictObject({ element: z.string(), distance: z.int().positive() }) + .readonly(); + +/** The kinds of change a store reports. */ +export const changeKinds = ["saved", "overlay", "items", "deleted"] as const; + +/** + * A change to a model. `revision` is the revision the change took. `missed` is sent only at the start of a subscription that names a revision the model is no longer at: the subscriber missed one or more changes, or the model has gone, and should load it again. + */ +export const storeChangeSchema = z.union([ + z + .strictObject({ + modelId: modelIdSchema, + kind: z.enum(changeKinds), + revision: revisionSchema, + }) + .readonly(), + z + .strictObject({ modelId: modelIdSchema, kind: z.literal("missed") }) + .readonly(), +]); + +/** A structural issue in a model or overlay, as `validate` and `validateOverlay` report it. */ +export const issueSchema = z + .strictObject({ code: z.string(), message: z.string() }) + .readonly(); + +export type ModelId = z.infer; +export type Revision = z.infer; +export type BoundActivity = z.infer; +export type BoundItems = z.infer; +export type StoredModel = z.infer; +export type ModelSummary = z.infer; +export type Reach = z.infer; +export type ChangeKind = (typeof changeKinds)[number]; +export type StoreChange = z.infer; diff --git a/packages/core/graph/store-contract/src/schema.unit.test.ts b/packages/core/graph/store-contract/src/schema.unit.test.ts new file mode 100644 index 0000000..fde8997 --- /dev/null +++ b/packages/core/graph/store-contract/src/schema.unit.test.ts @@ -0,0 +1,48 @@ +import { modelSchema, validate } from "@exadev/process-graph-schema"; +import { describe, expect, expectTypeOf, test } from "vitest"; +import { everyFieldModel, queryModel } from "./fixtures.ts"; +import { boundActivitySchema, type BoundActivity } from "./schema.ts"; + +describe("fixtures", () => { + test.each([ + { name: "everyFieldModel", model: everyFieldModel }, + { name: "queryModel", model: queryModel }, + ])("$name is a valid model", ({ model }) => { + expect(modelSchema.parse(model)).toStrictEqual(model); + expect(validate(model)).toEqual([]); + }); +}); + +describe("boundActivitySchema", () => { + test("is the model's activity with a required source", () => { + expectTypeOf().toMatchObjectType<{ + readonly kind: "activity"; + readonly source: string; + }>(); + }); + + test("accepts an activity whose id is its source", () => { + const activity = { + kind: "activity", + id: "github:o/r#1", + label: "One", + source: "github:o/r#1", + }; + + expect(boundActivitySchema.parse(activity)).toEqual(activity); + }); + + test.each([ + { name: "no source", value: { kind: "activity", id: "a", label: "A" } }, + { + name: "an id that is not its source", + value: { kind: "activity", id: "a", label: "A", source: "github:o/r#1" }, + }, + { + name: "another element kind", + value: { kind: "decision", id: "d", label: "D" }, + }, + ])("rejects $name", ({ value }) => { + expect(boundActivitySchema.safeParse(value).success).toBe(false); + }); +}); diff --git a/packages/core/graph/store-contract/src/store.ts b/packages/core/graph/store-contract/src/store.ts new file mode 100644 index 0000000..500b828 --- /dev/null +++ b/packages/core/graph/store-contract/src/store.ts @@ -0,0 +1,91 @@ +import type { Model, OpenIds, Overlay } from "@exadev/process-graph-schema"; +import type { ChangeFeed } from "./feed.ts"; +import type { + BoundItems, + ModelSummary, + Reach, + Revision, + StoredModel, +} from "./schema.ts"; + +/** What every store request carries: the model it is about and the caller's abort signal. */ +export interface ModelRequest { + readonly modelId: string; + /** Fires when the caller gives up. A write checks it before committing, so an aborted write changes nothing. */ + readonly signal: AbortSignal | undefined; +} + +/** A request about one element of a model. */ +export interface ElementRequest extends ModelRequest { + readonly element: string; +} + +/** + * What a store adapter implements. `createGraphStoreRouter` builds the whole port on it, so validation, error mapping, abort handling and change subscriptions behave identically for every store. + * + * Every write is atomic: it checks the result with `checkStored` and throws `GraphStoreError` with `invalid` before anything is visible, takes the next store-wide revision, and publishes the change to `feed` after it commits. A read or query of a model that does not exist, or of an element the model does not declare, throws `GraphStoreError` with `not-found`. + */ +export interface GraphStore { + /** The feed this store publishes its committed changes to. */ + readonly feed: ChangeFeed; + /** + * Lists the stored models. + * @returns Every model's id and revision, ordered by id. + */ + readonly listModels: (request: { + readonly signal: AbortSignal | undefined; + }) => Promise; + /** + * Stores a model under an id, replacing any model stored there and dropping the run state of elements the new model does not declare. + * @returns The revision the save took. + */ + readonly saveModel: ( + request: ModelRequest & { readonly model: Model }, + ) => Promise; + /** + * Reads a model back exactly as saved or merged, in its declared order, with its run state. + * @returns The model, its overlay and its revision. + */ + readonly loadModel: (request: ModelRequest) => Promise; + /** + * Removes a model and its run state. + * @returns The revision the deletion took, or `undefined` when there was no such model and nothing changed. + */ + readonly deleteModel: ( + request: ModelRequest, + ) => Promise; + /** + * Sets the run state of the elements an overlay names, leaving the others as they are. + * @returns The revision the change took. + */ + readonly applyOverlay: ( + request: ModelRequest & { readonly overlay: Overlay }, + ) => Promise; + /** + * Merges items read from a source into a model by id (see `mergeBoundItems`), creating the model from an empty one when it does not exist. + * @returns The revision the change took. + */ + readonly upsertItems: ( + request: ModelRequest & { readonly items: BoundItems }, + ) => Promise; + /** + * The work that can start now (see `readyOf`). + * @returns The ready elements, in model order. + */ + readonly ready: (request: ModelRequest) => Promise; + /** + * The unfinished work upstream of an element (see `blockedChainOf`). + * @returns The chain, by distance and then model order. + */ + readonly blockedChain: (request: ElementRequest) => Promise; + /** + * The unfinished work downstream of an element (see `impactOf`). + * @returns The impacted elements, by distance and then model order. + */ + readonly impact: (request: ElementRequest) => Promise; + /** + * The revision of a model's last change. + * @returns The revision, or `undefined` when there is no such model. + */ + readonly revisionOf: (request: ModelRequest) => Promise; +} diff --git a/packages/core/graph/store-contract/src/structure.ts b/packages/core/graph/store-contract/src/structure.ts new file mode 100644 index 0000000..7519c41 --- /dev/null +++ b/packages/core/graph/store-contract/src/structure.ts @@ -0,0 +1,153 @@ +import { loopTransitions } from "@exadev/process-graph-domain"; +import type { Model, OpenIds, Overlay } from "@exadev/process-graph-schema"; +import type { Reach } from "./schema.ts"; + +/** + * Whether an element is a pure control element: a `decision` or a pseudo-state. Control elements have no work of their own, so the structural queries pass through them and never report them. + */ +const isControl = (kind: Model["elements"][number]["kind"]): boolean => + kind === "decision" || kind === "pseudo"; + +/** + * Reads a key every caller knows is present: the structural view has an entry for every element of a valid model. + * @throws Error when the key is absent, which only an invalid model can cause. + */ +const lookup = (map: ReadonlyMap, key: string): V => { + const value = map.get(key); + if (value === undefined) { + throw new Error(`No element "${key}" in the structural view`); + } + + return value; +}; + +/** Which way a traversal follows transitions: towards what an element waits on, or towards what waits on it. */ +type Direction = "upstream" | "downstream"; + +/** + * The structural view of a stored model that the store's queries read: its elements in model order, its transitions without the loop edges `graph-domain` identifies, and its run state. + */ +export interface Structure { + readonly order: ReadonlyMap; + readonly control: ReadonlySet; + readonly done: ReadonlySet; + readonly open: ReadonlySet; + readonly upstream: ReadonlyMap; + readonly downstream: ReadonlyMap; +} + +/** + * Builds the structural view of a model and its overlay. + * @param model - A valid model. + * @param overlay - Its run state. + * @returns The view the queries below read. + */ +export const structureOf = ( + model: Model, + overlay: Overlay, +): Structure => { + const loops = new Set(loopTransitions(model)); + const upstream = new Map( + model.elements.map(({ id }) => [id, []]), + ); + const downstream = new Map( + model.elements.map(({ id }) => [id, []]), + ); + for (const { id, source, target } of model.transitions) { + if (!loops.has(id)) { + lookup(upstream, target).push(source); + lookup(downstream, source).push(target); + } + } + + return { + order: new Map(model.elements.map(({ id }, index) => [id, index])), + control: new Set( + model.elements.flatMap(({ id, kind }) => (isControl(kind) ? [id] : [])), + ), + done: new Set( + model.elements.flatMap(({ id }) => (overlay[id] === "done" ? [id] : [])), + ), + open: new Set( + model.elements.flatMap(({ id }) => + overlay[id] === undefined || overlay[id] === "open" ? [id] : [], + ), + ), + upstream, + downstream, + }; +}; + +/** + * Breadth-first traversal from `start` along non-loop transitions in one direction, expanding only elements that are not done. Returns every element reached that is neither done nor a control element, with the fewest transitions it takes to reach it, ordered by that distance and then by model order. `start` itself is not reported. + */ +const reach = ( + structure: Structure, + start: string, + direction: Direction, +): readonly Reach[] => { + const next = structure[direction]; + const distance = new Map([[start, 0]]); + let frontier = [start]; + for (let depth = 1; frontier.length > 0; depth += 1) { + const reached = frontier + .flatMap((element) => lookup(next, element)) + .filter((neighbour) => !distance.has(neighbour)); + for (const neighbour of reached) { + distance.set(neighbour, depth); + } + frontier = [...new Set(reached)].filter( + (neighbour) => !structure.done.has(neighbour), + ); + } + distance.delete(start); + + return [...distance] + .filter( + ([element]) => + !structure.done.has(element) && !structure.control.has(element), + ) + .map(([element, steps]) => ({ element, distance: steps })) + .sort( + (left, right) => + left.distance - right.distance || + lookup(structure.order, left.element) - + lookup(structure.order, right.element), + ); +}; + +/** + * What still stands before an element: the unfinished work upstream of it. An element that is done waits on nothing. + * @param structure - The structural view. + * @param element - An element of the model. + * @returns The upstream elements that are not done and are not control elements, by distance and then model order. + */ +export const blockedChainOf = ( + structure: Structure, + element: string, +): readonly Reach[] => + structure.done.has(element) ? [] : reach(structure, element, "upstream"); + +/** + * What depends on an element: the unfinished work downstream of it, whatever the element's own state. + * @param structure - The structural view. + * @param element - An element of the model. + * @returns The downstream elements that are not done and are not control elements, by distance and then model order. + */ +export const impactOf = ( + structure: Structure, + element: string, +): readonly Reach[] => reach(structure, element, "downstream"); + +/** + * The work that can start now: every open element (no run state, or `open`) that is not a control element and has nothing unfinished upstream. + * @param structure - The structural view. + * @returns The ready elements, in model order. + */ +export const readyOf = (structure: Structure): readonly string[] => + [...structure.order.keys()].filter( + (element) => + structure.open.has(element) && + !structure.control.has(element) && + blockedChainOf(structure, element).length === 0, + ); diff --git a/packages/core/graph/store-contract/src/structure.unit.test.ts b/packages/core/graph/store-contract/src/structure.unit.test.ts new file mode 100644 index 0000000..5ac778c --- /dev/null +++ b/packages/core/graph/store-contract/src/structure.unit.test.ts @@ -0,0 +1,94 @@ +import { readySet } from "@exadev/process-graph-domain"; +import type { Model, OpenIds, Overlay } from "@exadev/process-graph-schema"; +import { describe, expect, test } from "vitest"; +import { queryModel } from "./fixtures.ts"; +import { blockedChainOf, impactOf, readyOf, structureOf } from "./structure.ts"; + +const noFacts = () => ({ found: false }) as const; + +/** A fork into two branches that a join waits on, as a pseudo-state pair. */ +const forkJoin: Model = { + participants: [{ id: "team", label: "Team", kind: "role" }], + events: [], + actions: [], + artefacts: [], + elements: [ + { kind: "pseudo", id: "start", role: "initial" }, + { kind: "activity", id: "design", label: "Design", actor: "team" }, + { kind: "pseudo", id: "fork", role: "fork" }, + { kind: "activity", id: "api", label: "API", actor: "team" }, + { kind: "activity", id: "ui", label: "UI", actor: "team" }, + { kind: "pseudo", id: "join", role: "join" }, + { kind: "activity", id: "release", label: "Release", actor: "team" }, + ], + transitions: [ + { id: "s", source: "start", target: "design" }, + { id: "f", source: "design", target: "fork" }, + { id: "fa", source: "fork", target: "api" }, + { id: "fu", source: "fork", target: "ui" }, + { id: "aj", source: "api", target: "join" }, + { id: "uj", source: "ui", target: "join" }, + { id: "r", source: "join", target: "release" }, + ], + fragments: [], + projections: [], +}; + +const overlays: readonly Overlay[] = [ + {}, + { a: "done" }, + { a: "done", b: "done" }, + { a: "done", b: "done", c: "done" }, + { a: "done", b: "done", c: "done", e: "done", x: "done" }, + { a: "done", x: "unknown" }, + { design: "done" }, + { design: "done", api: "done" }, + { design: "done", api: "done", ui: "done" }, +]; + +describe("readyOf", () => { + test.each( + [queryModel, forkJoin].flatMap((model) => + overlays.map((overlay) => ({ model, overlay })), + ), + )( + "agrees with graph-domain's readySet on a model without conditions or artefacts ($overlay)", + ({ model, overlay }) => { + expect(readyOf(structureOf(model, overlay))).toEqual( + readySet(model, overlay, noFacts), + ); + }, + ); + + test("leaves out work whose run state is unknown, and everything waiting on it", () => { + expect(readyOf(structureOf(queryModel, { a: "unknown" }))).toEqual(["x"]); + }); +}); + +describe("blockedChainOf and impactOf", () => { + test("pass through control elements without reporting them", () => { + const structure = structureOf(forkJoin, {}); + + expect(blockedChainOf(structure, "release")).toEqual([ + { element: "api", distance: 2 }, + { element: "ui", distance: 2 }, + { element: "design", distance: 4 }, + ]); + expect(impactOf(structure, "design")).toEqual([ + { element: "api", distance: 2 }, + { element: "ui", distance: 2 }, + { element: "release", distance: 4 }, + ]); + }); + + test("ignore a decision's loop edge, so a retry is no dependency", () => { + const structure = structureOf(queryModel, {}); + + expect(blockedChainOf(structure, "b")).toEqual([ + { element: "a", distance: 1 }, + ]); + expect(impactOf(structure, "e").map(({ element }) => element)).toEqual([ + "f", + ]); + }); +}); diff --git a/packages/core/graph/store-contract/stryker.config.ts b/packages/core/graph/store-contract/stryker.config.ts new file mode 100644 index 0000000..6429b27 --- /dev/null +++ b/packages/core/graph/store-contract/stryker.config.ts @@ -0,0 +1,3 @@ +import { packageStrykerConfig } from "../../../../stryker.shared.ts"; + +export default packageStrykerConfig(); diff --git a/packages/core/graph/store-contract/test/package.smoke.test.ts b/packages/core/graph/store-contract/test/package.smoke.test.ts new file mode 100644 index 0000000..f76915a --- /dev/null +++ b/packages/core/graph/store-contract/test/package.smoke.test.ts @@ -0,0 +1,28 @@ +import { createRouterClient } from "@orpc/server"; +import { expect, test } from "vitest"; +// Deliberately imports the built dist/, not src/: this is the check that catches a wrong exports map or a build misconfiguration that a unit test importing straight from src/ never touches. +import { + createFakeGraphStore, + createGraphStoreRouter, + emptyModel, +} from "../dist/index.js"; + +test("the built package serves a store through an oRPC client", async () => { + const client = createRouterClient( + createGraphStoreRouter(createFakeGraphStore()), + ); + const { revision } = await client.v1.models.save({ + modelId: "empty", + model: emptyModel, + }); + + await expect(client.v1.models.list()).resolves.toEqual([ + { modelId: "empty", revision }, + ]); +}); + +test("the conformance kit is its own entry point", async () => { + const kit = await import("../dist/conformance.js"); + + expect(typeof kit.graphStoreConformance).toBe("function"); +}); diff --git a/packages/core/graph/store-contract/tsconfig.build.json b/packages/core/graph/store-contract/tsconfig.build.json new file mode 100644 index 0000000..63c536a --- /dev/null +++ b/packages/core/graph/store-contract/tsconfig.build.json @@ -0,0 +1,20 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "composite": true, + "noEmit": false, + "declaration": true, + "declarationMap": false, + "rootDir": "src", + "outDir": "dist", + "rewriteRelativeImportExtensions": true + }, + "references": [ + { + "path": "../schema/tsconfig.build.json" + }, + { + "path": "../domain/tsconfig.build.json" + } + ] +} diff --git a/packages/core/graph/store-contract/tsconfig.json b/packages/core/graph/store-contract/tsconfig.json new file mode 100644 index 0000000..8895dd1 --- /dev/null +++ b/packages/core/graph/store-contract/tsconfig.json @@ -0,0 +1,9 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2024", "WebWorker"], + "types": [] + }, + "include": ["src/**/*.ts"], + "exclude": ["src/**/*.test.ts", "src/test-support/**"] +} diff --git a/packages/core/graph/store-contract/tsconfig.node.json b/packages/core/graph/store-contract/tsconfig.node.json new file mode 100644 index 0000000..8a460f4 --- /dev/null +++ b/packages/core/graph/store-contract/tsconfig.node.json @@ -0,0 +1,8 @@ +{ + "extends": "../../../../tsconfig.base.json", + "compilerOptions": { + "lib": ["ES2024"], + "types": ["node"] + }, + "include": ["src/**/*.ts", "test/**/*.ts", "scripts/**/*.ts", "*.config.ts"] +} diff --git a/packages/core/graph/store-contract/vitest.config.ts b/packages/core/graph/store-contract/vitest.config.ts new file mode 100644 index 0000000..01b9391 --- /dev/null +++ b/packages/core/graph/store-contract/vitest.config.ts @@ -0,0 +1,16 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + projects: [ + { + extends: true, + test: { name: "unit", include: ["src/**/*.test.ts"] }, + }, + { + extends: true, + test: { name: "smoke", include: ["test/package.smoke.test.ts"] }, + }, + ], + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 4c65dce..d7ea062 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -554,6 +554,61 @@ importers: specifier: 5.0.3 version: 5.0.3(@types/node@26.6.4)(@vitest/coverage-v8@5.0.3)(vite@8.3.2(@types/node@26.6.4)(jiti@2.7.0)(yaml@2.9.1)) + packages/core/graph/store-contract: + dependencies: + '@exadev/process-graph-domain': + specifier: 1.0.2 + version: link:../domain + '@exadev/process-graph-schema': + specifier: 1.0.2 + version: link:../schema + '@orpc/contract': + specifier: 2.0.0-beta.41 + version: 2.0.0-beta.41 + '@orpc/server': + specifier: 2.0.0-beta.41 + version: 2.0.0-beta.41 + trilean: + specifier: 1.7.2 + version: 1.7.2 + zod: + specifier: 4.6.5 + version: 4.6.5 + devDependencies: + '@arethetypeswrong/cli': + specifier: 0.18.5 + version: 0.18.5 + '@orpc/client': + specifier: 2.0.0-beta.41 + version: 2.0.0-beta.41 + '@stryker-mutator/core': + specifier: 10.0.0 + version: 10.0.0(@types/node@26.6.4) + '@stryker-mutator/typescript-checker': + specifier: 10.0.0 + version: 10.0.0(@stryker-mutator/core@10.0.0(@types/node@26.6.4))(typescript@6.0.3) + '@stryker-mutator/vitest-runner': + specifier: 10.0.0 + version: 10.0.0(@stryker-mutator/core@10.0.0(@types/node@26.6.4))(vitest@5.0.3) + '@types/node': + specifier: 26.6.4 + version: 26.6.4 + '@vitest/coverage-v8': + specifier: 5.0.3 + version: 5.0.3(vitest@5.0.3) + eslint: + specifier: 10.12.0 + version: 10.12.0(jiti@2.7.0)(supports-color@7.2.0) + publint: + specifier: 0.3.25 + version: 0.3.25 + typescript: + specifier: 6.0.3 + version: 6.0.3 + vitest: + specifier: 5.0.3 + version: 5.0.3(@types/node@26.6.4)(@vitest/coverage-v8@5.0.3)(vite@8.3.2(@types/node@26.6.4)(jiti@2.7.0)(yaml@2.9.1)) + packages/targets/process-graph: dependencies: '@exadev/process-graph-dag': diff --git a/tsconfig.build.json b/tsconfig.build.json index a843570..d9f7bd8 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -22,6 +22,9 @@ { "path": "packages/core/graph/layout/tsconfig.build.json" }, + { + "path": "packages/core/graph/store-contract/tsconfig.build.json" + }, { "path": "packages/targets/process-graph/tsconfig.build.json" } From e1985e7eb6c0a783eb9e6e0168fbd0c80859798d Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 10 Oct 2026 17:04:55 +0100 Subject: [PATCH 2/4] feat(store-contract): ship the conformance fixtures and cover a cycle without a decision The model fixtures move into the ./conformance entry so adapters can reuse everyFieldModel and queryModel in their own tests, for example to check a database file round-trips every field. A new conformance case saves a model whose transitions cycle without a decision: the queries must still terminate and never report the element asked about. It caught a SQL walk that reported the start element when a cycle led back to it. --- .../graph/store-contract/src/conformance.ts | 187 +++++++++++++++++- .../core/graph/store-contract/src/fixtures.ts | 145 -------------- .../store-contract/src/schema.unit.test.ts | 2 +- .../store-contract/src/structure.unit.test.ts | 2 +- 4 files changed, 188 insertions(+), 148 deletions(-) delete mode 100644 packages/core/graph/store-contract/src/fixtures.ts diff --git a/packages/core/graph/store-contract/src/conformance.ts b/packages/core/graph/store-contract/src/conformance.ts index 6ef28c1..c158446 100644 --- a/packages/core/graph/store-contract/src/conformance.ts +++ b/packages/core/graph/store-contract/src/conformance.ts @@ -1,9 +1,152 @@ import type { Model, OpenIds } from "@exadev/process-graph-schema"; import { describe, expect, test } from "vitest"; import type { GraphStoreClient } from "./contract.ts"; -import { everyFieldModel, queryModel } from "./fixtures.ts"; import type { BoundItems, StoreChange } from "./schema.ts"; +/** + * A model that uses every field the model schema has, each optional field present on some record and absent on another, so a store that drops, renames or reorders anything fails the round trip. + */ +export const everyFieldModel: Model = { + participants: [ + { id: "customer", label: "Customer", kind: "role" }, + { id: "shop", label: "Shop", kind: "system" }, + { id: "order", label: "Order", kind: "entity" }, + ], + events: [ + { id: "placed", name: "Order placed", payload: { total: "number" } }, + { id: "paid", name: "Payment received" }, + ], + actions: [ + { id: "notify", name: "Notify customer" }, + { id: "log", name: "Write audit log" }, + ], + artefacts: [ + { id: "basket", name: "Basket" }, + { id: "invoice", name: "Invoice", deliverable: true }, + { id: "receipt", name: "Receipt", deliverable: false }, + ], + elements: [ + { kind: "pseudo", id: "start", role: "initial" }, + { + kind: "activity", + id: "github:shop/orders#1", + label: "Place order", + actor: "customer", + action: "notify", + source: "github:shop/orders#1", + affects: ["order"], + consumes: ["basket"], + produces: ["invoice"], + extensions: { bpmn: { pool: "Customer", lane: 1, marker: null } }, + }, + { kind: "decision", id: "approve", label: "Approve?", actor: "shop" }, + { + kind: "state", + id: "pending", + label: "Pending", + parent: "start", + entry: "log", + exit: "notify", + }, + { kind: "state", id: "closed", label: "Closed" }, + { + kind: "message", + id: "confirm", + label: "Confirm", + from: "shop", + to: "customer", + mode: "async", + event: "paid", + }, + { + kind: "message", + id: "ack", + label: "Acknowledge", + from: "customer", + to: "shop", + mode: "reply", + }, + { kind: "activity", id: "ship", label: "Ship" }, + { kind: "pseudo", id: "end", role: "final", parent: "pending" }, + ], + transitions: [ + { id: "t-start", source: "start", target: "github:shop/orders#1" }, + { + id: "t-review", + source: "github:shop/orders#1", + target: "approve", + trigger: "placed", + effect: "log", + }, + { + id: "t-yes", + source: "approve", + target: "ship", + guard: "approved", + condition: { + kind: "compare", + op: "eq", + left: { kind: "reference", key: { $fact: "approved" } }, + right: { kind: "booleanLiteral", value: true }, + }, + }, + { id: "t-no", source: "approve", target: "closed", guard: "rejected" }, + { id: "t-end", source: "ship", target: "end" }, + ], + fragments: [ + { + id: "decide", + operator: "alt", + operands: [{ guard: "approved", body: ["ship"] }, { body: ["closed"] }], + }, + { id: "chat", operator: "par", operands: [] }, + ], + projections: [ + { id: "flow", notation: "flowchart" }, + { id: "lanes", notation: "swimlane" }, + { id: "lifecycle", notation: "stateMachine", subject: "order" }, + { id: "happy", notation: "sequence", scenario: ["t-review", "t-yes"] }, + ], + views: { + flowchart: { start: { x: 0, y: 0 }, ship: { x: 120.5, y: -40 } }, + sequence: {}, + }, +}; + +/** + * A model for the structural queries: a done first step, a chain through a decision whose retry edge loops back, a join waiting on two branches, and an unrelated step. + * + * `a` (done) leads to `b`, `b` to `c`, `c` to the decision `d`, which either approves into `e` or retries back to `b` (a loop edge, so it is no dependency). `e` and the independent `x` meet at the join `j`, which leads to `f`. + */ +export const queryModel: Model = { + participants: [{ id: "team", label: "Team", kind: "role" }], + events: [], + actions: [], + artefacts: [], + elements: [ + { kind: "activity", id: "a", label: "A", actor: "team" }, + { kind: "activity", id: "b", label: "B", actor: "team" }, + { kind: "activity", id: "c", label: "C", actor: "team" }, + { kind: "decision", id: "d", label: "D?" }, + { kind: "activity", id: "e", label: "E", actor: "team" }, + { kind: "activity", id: "x", label: "X", actor: "team" }, + { kind: "pseudo", id: "j", role: "join" }, + { kind: "activity", id: "f", label: "F", actor: "team" }, + ], + transitions: [ + { id: "ab", source: "a", target: "b" }, + { id: "bc", source: "b", target: "c" }, + { id: "cd", source: "c", target: "d" }, + { id: "de", source: "d", target: "e", guard: "approved" }, + { id: "db", source: "d", target: "b", guard: "retry" }, + { id: "ej", source: "e", target: "j" }, + { id: "xj", source: "x", target: "j" }, + { id: "jf", source: "j", target: "f" }, + ], + fragments: [], + projections: [], +}; + /** What the conformance kit needs from an adapter's test. */ export interface GraphStoreConformanceOptions { /** A name for the `describe` block, such as the adapter and transport under test. */ @@ -545,6 +688,48 @@ export const graphStoreConformance = ( }), ); + test( + "queries terminate on a cycle no decision closes, and never report the element asked about", + withClient(options, async (client) => { + const cycle: Model = { + ...pairModel, + elements: [ + ...pairModel.elements, + { kind: "activity", id: "ship", label: "Ship", actor: "team" }, + ], + transitions: [ + ...pairModel.transitions, + { id: "build-plan", source: "build", target: "plan" }, + { id: "build-ship", source: "build", target: "ship" }, + ], + }; + await client.v1.models.save({ modelId: "cycle", model: cycle }); + + await expect( + client.v1.query.blockedChain({ modelId: "cycle", element: "plan" }), + ).resolves.toEqual({ chain: [{ element: "build", distance: 1 }] }); + await expect( + client.v1.query.blockedChain({ modelId: "cycle", element: "ship" }), + ).resolves.toEqual({ + chain: [ + { element: "build", distance: 1 }, + { element: "plan", distance: 2 }, + ], + }); + await expect( + client.v1.query.impact({ modelId: "cycle", element: "plan" }), + ).resolves.toEqual({ + impacted: [ + { element: "build", distance: 1 }, + { element: "ship", distance: 2 }, + ], + }); + await expect( + client.v1.query.ready({ modelId: "cycle" }), + ).resolves.toEqual({ elements: [] }); + }), + ); + test( "a query about an element the model lacks is NOT_FOUND naming it", withClient(options, async (client) => { diff --git a/packages/core/graph/store-contract/src/fixtures.ts b/packages/core/graph/store-contract/src/fixtures.ts deleted file mode 100644 index ce447e5..0000000 --- a/packages/core/graph/store-contract/src/fixtures.ts +++ /dev/null @@ -1,145 +0,0 @@ -import type { Model, OpenIds } from "@exadev/process-graph-schema"; - -/** - * A model that uses every field the model schema has, each optional field present on some record and absent on another, so a store that drops, renames or reorders anything fails the round trip. - */ -export const everyFieldModel: Model = { - participants: [ - { id: "customer", label: "Customer", kind: "role" }, - { id: "shop", label: "Shop", kind: "system" }, - { id: "order", label: "Order", kind: "entity" }, - ], - events: [ - { id: "placed", name: "Order placed", payload: { total: "number" } }, - { id: "paid", name: "Payment received" }, - ], - actions: [ - { id: "notify", name: "Notify customer" }, - { id: "log", name: "Write audit log" }, - ], - artefacts: [ - { id: "basket", name: "Basket" }, - { id: "invoice", name: "Invoice", deliverable: true }, - { id: "receipt", name: "Receipt", deliverable: false }, - ], - elements: [ - { kind: "pseudo", id: "start", role: "initial" }, - { - kind: "activity", - id: "github:shop/orders#1", - label: "Place order", - actor: "customer", - action: "notify", - source: "github:shop/orders#1", - affects: ["order"], - consumes: ["basket"], - produces: ["invoice"], - extensions: { bpmn: { pool: "Customer", lane: 1, marker: null } }, - }, - { kind: "decision", id: "approve", label: "Approve?", actor: "shop" }, - { - kind: "state", - id: "pending", - label: "Pending", - parent: "start", - entry: "log", - exit: "notify", - }, - { kind: "state", id: "closed", label: "Closed" }, - { - kind: "message", - id: "confirm", - label: "Confirm", - from: "shop", - to: "customer", - mode: "async", - event: "paid", - }, - { - kind: "message", - id: "ack", - label: "Acknowledge", - from: "customer", - to: "shop", - mode: "reply", - }, - { kind: "activity", id: "ship", label: "Ship" }, - { kind: "pseudo", id: "end", role: "final", parent: "pending" }, - ], - transitions: [ - { id: "t-start", source: "start", target: "github:shop/orders#1" }, - { - id: "t-review", - source: "github:shop/orders#1", - target: "approve", - trigger: "placed", - effect: "log", - }, - { - id: "t-yes", - source: "approve", - target: "ship", - guard: "approved", - condition: { - kind: "compare", - op: "eq", - left: { kind: "reference", key: { $fact: "approved" } }, - right: { kind: "booleanLiteral", value: true }, - }, - }, - { id: "t-no", source: "approve", target: "closed", guard: "rejected" }, - { id: "t-end", source: "ship", target: "end" }, - ], - fragments: [ - { - id: "decide", - operator: "alt", - operands: [{ guard: "approved", body: ["ship"] }, { body: ["closed"] }], - }, - { id: "chat", operator: "par", operands: [] }, - ], - projections: [ - { id: "flow", notation: "flowchart" }, - { id: "lanes", notation: "swimlane" }, - { id: "lifecycle", notation: "stateMachine", subject: "order" }, - { id: "happy", notation: "sequence", scenario: ["t-review", "t-yes"] }, - ], - views: { - flowchart: { start: { x: 0, y: 0 }, ship: { x: 120.5, y: -40 } }, - sequence: {}, - }, -}; - -/** - * A model for the structural queries: a done first step, a chain through a decision whose retry edge loops back, a join waiting on two branches, and an unrelated step. - * - * `a` (done) leads to `b`, `b` to `c`, `c` to the decision `d`, which either approves into `e` or retries back to `b` (a loop edge, so it is no dependency). `e` and the independent `x` meet at the join `j`, which leads to `f`. - */ -export const queryModel: Model = { - participants: [{ id: "team", label: "Team", kind: "role" }], - events: [], - actions: [], - artefacts: [], - elements: [ - { kind: "activity", id: "a", label: "A", actor: "team" }, - { kind: "activity", id: "b", label: "B", actor: "team" }, - { kind: "activity", id: "c", label: "C", actor: "team" }, - { kind: "decision", id: "d", label: "D?" }, - { kind: "activity", id: "e", label: "E", actor: "team" }, - { kind: "activity", id: "x", label: "X", actor: "team" }, - { kind: "pseudo", id: "j", role: "join" }, - { kind: "activity", id: "f", label: "F", actor: "team" }, - ], - transitions: [ - { id: "ab", source: "a", target: "b" }, - { id: "bc", source: "b", target: "c" }, - { id: "cd", source: "c", target: "d" }, - { id: "de", source: "d", target: "e", guard: "approved" }, - { id: "db", source: "d", target: "b", guard: "retry" }, - { id: "ej", source: "e", target: "j" }, - { id: "xj", source: "x", target: "j" }, - { id: "jf", source: "j", target: "f" }, - ], - fragments: [], - projections: [], -}; diff --git a/packages/core/graph/store-contract/src/schema.unit.test.ts b/packages/core/graph/store-contract/src/schema.unit.test.ts index fde8997..b08540b 100644 --- a/packages/core/graph/store-contract/src/schema.unit.test.ts +++ b/packages/core/graph/store-contract/src/schema.unit.test.ts @@ -1,6 +1,6 @@ import { modelSchema, validate } from "@exadev/process-graph-schema"; import { describe, expect, expectTypeOf, test } from "vitest"; -import { everyFieldModel, queryModel } from "./fixtures.ts"; +import { everyFieldModel, queryModel } from "./conformance.ts"; import { boundActivitySchema, type BoundActivity } from "./schema.ts"; describe("fixtures", () => { diff --git a/packages/core/graph/store-contract/src/structure.unit.test.ts b/packages/core/graph/store-contract/src/structure.unit.test.ts index 5ac778c..eb992f4 100644 --- a/packages/core/graph/store-contract/src/structure.unit.test.ts +++ b/packages/core/graph/store-contract/src/structure.unit.test.ts @@ -1,7 +1,7 @@ import { readySet } from "@exadev/process-graph-domain"; import type { Model, OpenIds, Overlay } from "@exadev/process-graph-schema"; import { describe, expect, test } from "vitest"; -import { queryModel } from "./fixtures.ts"; +import { queryModel } from "./conformance.ts"; import { blockedChainOf, impactOf, readyOf, structureOf } from "./structure.ts"; const noFacts = () => ({ found: false }) as const; From f1d117b563469c6d5f2ec2144d40ab7f6a746250 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 10 Oct 2026 19:41:40 +0100 Subject: [PATCH 3/4] docs: align the package table --- README.md | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 47007d5..d6136d9 100644 --- a/README.md +++ b/README.md @@ -45,17 +45,17 @@ Configuration lives at exactly one of two levels: in the package (`eslint.config ## Packages -| Package | What it is | -| -------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| [`graph-dag`](packages/core/graph/dag/README.md) | Model-independent dependency-graph primitives over a node set and a `needs` accessor: layers, ancestors, descendants, cycles, transitive reduction, critical path. | -| [`graph-schema`](packages/core/graph/schema/README.md) | The process model: typed identifier spaces, `defineModel` with compile-time reference checking, structural validation. | -| [`graph-projection`](packages/core/graph/projection/README.md) | Projections of the model into a flowchart, swimlane, state machine and sequence diagram, plus derived hand-offs. | -| [`graph-domain`](packages/core/graph/domain/README.md) | Reasoning over the model and a run overlay: control graph, illegal cycles, three-valued status, ready set, blocked-by, impact, what-if and critical path. | -| [`graph-layout`](packages/core/graph/layout/README.md) | Layered layout of a projected flowchart, swimlane or state machine into positions in the model's `views` shape. | -| [`graph-source-contract`](packages/core/graph/source-contract/README.md) | The `GraphSource` port as an oRPC contract: versioned, paged and streamed sync of bound activities, participants, transitions and run state, resumable by cursor, with an in-memory fake and a conformance kit. | -| [`graph-source-github-issues`](packages/adapters/graph/source-github-issues/README.md) | The GitHub issues adapter for the `GraphSource` port: issues as bound activities, `blockedBy` as transitions, sub-issues, repositories, organisations and assignees, across organisations and repositories. | -| [`graph-store-contract`](packages/core/graph/store-contract/README.md) | The persistence port as an oRPC contract: save, load and delete models, set run state, upsert items from a source, structural ready, blocked-chain and impact queries, and a change feed; with an in-memory fake and a conformance kit. | -| [`process-graph`](packages/targets/process-graph/README.md) | The primary entry point: re-exports the schema, projection and domain APIs flat and the DAG API as the `dag` namespace. | +| Package | What it is | +| -------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| [`graph-dag`](packages/core/graph/dag/README.md) | Model-independent dependency-graph primitives over a node set and a `needs` accessor: layers, ancestors, descendants, cycles, transitive reduction, critical path. | +| [`graph-schema`](packages/core/graph/schema/README.md) | The process model: typed identifier spaces, `defineModel` with compile-time reference checking, structural validation. | +| [`graph-projection`](packages/core/graph/projection/README.md) | Projections of the model into a flowchart, swimlane, state machine and sequence diagram, plus derived hand-offs. | +| [`graph-domain`](packages/core/graph/domain/README.md) | Reasoning over the model and a run overlay: control graph, illegal cycles, three-valued status, ready set, blocked-by, impact, what-if and critical path. | +| [`graph-layout`](packages/core/graph/layout/README.md) | Layered layout of a projected flowchart, swimlane or state machine into positions in the model's `views` shape. | +| [`graph-source-contract`](packages/core/graph/source-contract/README.md) | The `GraphSource` port as an oRPC contract: versioned, paged and streamed sync of bound activities, participants, transitions and run state, resumable by cursor, with an in-memory fake and a conformance kit. | +| [`graph-source-github-issues`](packages/adapters/graph/source-github-issues/README.md) | The GitHub issues adapter for the `GraphSource` port: issues as bound activities, `blockedBy` as transitions, sub-issues, repositories, organisations and assignees, across organisations and repositories. | +| [`graph-store-contract`](packages/core/graph/store-contract/README.md) | The persistence port as an oRPC contract: save, load and delete models, set run state, upsert items from a source, structural ready, blocked-chain and impact queries, and a change feed; with an in-memory fake and a conformance kit. | +| [`process-graph`](packages/targets/process-graph/README.md) | The primary entry point: re-exports the schema, projection and domain APIs flat and the DAG API as the `dag` namespace. | ## Conventions From 4ff1ed9d049b0c3bd8f18fd7768d49d8c771d45b Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Sat, 10 Oct 2026 19:48:28 +0100 Subject: [PATCH 4/4] fix(store-contract): bound each change subscription's buffer A subscriber that stopped reading made every later change pile up in its subscription without limit. A subscription now holds at most its capacity of undelivered changes; when another arrives, the held changes are replaced by one missed event, so memory stays bounded and the subscriber knows to load the model again. A subscriber that keeps reading receives every change directly and misses nothing. The capacity is a ChangeFeed option, overridable per subscription. The default, defaultSubscriptionCapacity, holds one change: the only thing a change asks of a subscriber is to read the model again, and the model read after one pending change already reflects every later one. A conformance case checks that a subscriber that stops reading under sustained writes is never left unaware of a change it lost, and the kit's helpers move to their own module. --- packages/core/graph/store-contract/README.md | 2 + .../store-contract/src/conformance-support.ts | 127 ++++++++++++++++++ .../graph/store-contract/src/conformance.ts | 118 ++++++---------- .../core/graph/store-contract/src/contract.ts | 2 +- .../core/graph/store-contract/src/feed.ts | 68 +++++++++- .../store-contract/src/feed.unit.test.ts | 103 +++++++++++++- .../core/graph/store-contract/src/index.ts | 8 +- .../core/graph/store-contract/src/router.ts | 2 +- .../core/graph/store-contract/src/schema.ts | 2 +- 9 files changed, 342 insertions(+), 90 deletions(-) create mode 100644 packages/core/graph/store-contract/src/conformance-support.ts diff --git a/packages/core/graph/store-contract/README.md b/packages/core/graph/store-contract/README.md index 570a13e..b3b7bfa 100644 --- a/packages/core/graph/store-contract/README.md +++ b/packages/core/graph/store-contract/README.md @@ -68,6 +68,8 @@ The queries are structural: they read the stored transitions and run state, and `changes` yields every later change to one model as `{ modelId, kind, revision }`, with `kind` one of `saved`, `overlay`, `items` or `deleted`. The router subscribes before it returns, so a caller holding the iterator is already subscribed and misses nothing it writes next. Given `since`, a revision the caller last saw, the first event is `{ modelId, kind: "missed" }` when the model is no longer at that revision (it changed, or it has gone), and the caller should load it again. +A subscription's memory is bounded. A change that commits while the subscriber is waiting goes straight to it; one that commits while the subscriber is busy is held, up to the subscription's capacity. When a change arrives and the subscription already holds that many, the held changes are replaced by one `{ modelId, kind: "missed" }`, and later changes follow it in order. A subscriber that keeps reading therefore misses nothing, and one that stops reading costs the store a fixed amount of memory and learns that it must load the model again. The capacity is a `ChangeFeed` option (`new ChangeFeed({ capacity })`, or `subscribe(modelId, { capacity })` for one subscription); the default, `defaultSubscriptionCapacity`, holds one change, because the only thing a change asks of a subscriber is to read the model again, and the model it reads after one pending change already reflects every later one. The conformance kit checks that a subscriber that stops reading under sustained writes is told it missed whatever it did not receive. + A store's `ChangeFeed` only sees the writes made through that store. Where several processes or tabs write one database, each bridges the others' changes into its own feed. ## Implementing a store diff --git a/packages/core/graph/store-contract/src/conformance-support.ts b/packages/core/graph/store-contract/src/conformance-support.ts new file mode 100644 index 0000000..814e898 --- /dev/null +++ b/packages/core/graph/store-contract/src/conformance-support.ts @@ -0,0 +1,127 @@ +import type { Model, OpenIds } from "@exadev/process-graph-schema"; +import type { BoundItems, StoreChange } from "./schema.ts"; + +/** The settled error of a call that must fail, so its code and data can be asserted. */ +export const failureOf = async ( + call: Readonly>, +): Promise => { + try { + await call; + } catch (error) { + return error; + } + throw new Error("expected the call to fail"); +}; + +/** A two-step model whose only transition can be broken by an upsert. */ +export const pairModel: Model = { + participants: [{ id: "team", label: "Team", kind: "role" }], + events: [], + actions: [], + artefacts: [], + elements: [ + { kind: "activity", id: "plan", label: "Plan", actor: "team" }, + { kind: "activity", id: "build", label: "Build", actor: "team" }, + ], + transitions: [{ id: "plan-build", source: "plan", target: "build" }], + fragments: [], + projections: [], +}; + +/** Items as a source would send them: one new repository participant, two issues, one replacing an existing activity, and a dependency. */ +export const issueItems: BoundItems = { + participants: [ + { id: "github:org/repo", label: "org/repo", kind: "system" }, + { id: "team", label: "Platform team", kind: "role" }, + ], + activities: [ + { + kind: "activity", + id: "github:org/repo#1", + label: "Design the schema", + actor: "github:org/repo", + source: "github:org/repo#1", + }, + { + kind: "activity", + id: "github:org/repo#2", + label: "Write the migration", + actor: "team", + source: "github:org/repo#2", + }, + ], + transitions: [ + { id: "dep-1-2", source: "github:org/repo#1", target: "github:org/repo#2" }, + ], + overlay: { "github:org/repo#1": "done" }, +}; + +/** Reads a subscription up to and including the first event `last` accepts, then stops it. */ +export const takeUntil = async ( + events: Readonly>, + last: (change: StoreChange) => boolean, +): Promise => { + const received: StoreChange[] = []; + for await (const event of events) { + received.push(event); + if (last(event)) { + break; + } + } + + return received; +}; + +/** Reads the first event of a subscription, then stops it. */ +export const firstOf = async ( + events: Readonly>, +): Promise => takeUntil(events, () => true); + +/** + * Runs steps one after another, each starting once the one before has settled, as consecutive writes from one client do. + * @param steps - The steps, in order. + * @returns Each step's result, in order. + */ +export const inSequence = async ( + steps: readonly (() => Promise)[], +): Promise => + steps.reduce>( + async (done, step) => [...(await done), await step()], + Promise.resolve([]), + ); + +/** + * The revisions a subscriber lost without being told: every change must arrive in commit order, except that a `missed` event may stand in for any run of changes before the next one that arrives. + * @param received - What the subscriber received, in order. + * @param written - The revisions committed, in order. + * @returns The revisions skipped with no `missed` in their place, plus any received out of order. Empty when the subscription kept its contract. + */ +export const silentLosses = ( + received: readonly StoreChange[], + written: readonly number[], +): readonly number[] => { + const lost: number[] = []; + let expected = 0; + let excused = false; + for (const change of received) { + if (change.kind === "missed") { + excused = true; + continue; + } + const position = written.indexOf(change.revision); + if (position < expected) { + lost.push(change.revision); + continue; + } + if (!excused) { + lost.push(...written.slice(expected, position)); + } + expected = position + 1; + excused = false; + } + if (!excused) { + lost.push(...written.slice(expected)); + } + + return lost; +}; diff --git a/packages/core/graph/store-contract/src/conformance.ts b/packages/core/graph/store-contract/src/conformance.ts index c158446..7d299e5 100644 --- a/packages/core/graph/store-contract/src/conformance.ts +++ b/packages/core/graph/store-contract/src/conformance.ts @@ -1,7 +1,16 @@ import type { Model, OpenIds } from "@exadev/process-graph-schema"; import { describe, expect, test } from "vitest"; import type { GraphStoreClient } from "./contract.ts"; -import type { BoundItems, StoreChange } from "./schema.ts"; +import { + failureOf, + firstOf, + inSequence, + issueItems, + pairModel, + silentLosses, + takeUntil, +} from "./conformance-support.ts"; +import { defaultSubscriptionCapacity } from "./feed.ts"; /** * A model that uses every field the model schema has, each optional field present on some record and absent on another, so a store that drops, renames or reorders anything fails the round trip. @@ -175,82 +184,6 @@ const withClient = } }; -/** The settled error of a call that must fail, so its code and data can be asserted. */ -const failureOf = async ( - call: Readonly>, -): Promise => { - try { - await call; - } catch (error) { - return error; - } - throw new Error("expected the call to fail"); -}; - -/** A two-step model whose only transition can be broken by an upsert. */ -const pairModel: Model = { - participants: [{ id: "team", label: "Team", kind: "role" }], - events: [], - actions: [], - artefacts: [], - elements: [ - { kind: "activity", id: "plan", label: "Plan", actor: "team" }, - { kind: "activity", id: "build", label: "Build", actor: "team" }, - ], - transitions: [{ id: "plan-build", source: "plan", target: "build" }], - fragments: [], - projections: [], -}; - -/** Items as a source would send them: one new repository participant, two issues, one replacing an existing activity, and a dependency. */ -const issueItems: BoundItems = { - participants: [ - { id: "github:org/repo", label: "org/repo", kind: "system" }, - { id: "team", label: "Platform team", kind: "role" }, - ], - activities: [ - { - kind: "activity", - id: "github:org/repo#1", - label: "Design the schema", - actor: "github:org/repo", - source: "github:org/repo#1", - }, - { - kind: "activity", - id: "github:org/repo#2", - label: "Write the migration", - actor: "team", - source: "github:org/repo#2", - }, - ], - transitions: [ - { id: "dep-1-2", source: "github:org/repo#1", target: "github:org/repo#2" }, - ], - overlay: { "github:org/repo#1": "done" }, -}; - -/** Reads a subscription up to and including the first event `last` accepts, then stops it. */ -const takeUntil = async ( - events: Readonly>, - last: (change: StoreChange) => boolean, -): Promise => { - const received: StoreChange[] = []; - for await (const event of events) { - received.push(event); - if (last(event)) { - break; - } - } - - return received; -}; - -/** Reads the first event of a subscription, then stops it. */ -const firstOf = async ( - events: Readonly>, -): Promise => takeUntil(events, () => true); - /** * The behaviour every `GraphStore` adapter must show, as Vitest tests. Call it at the top level of a test file; every test runs through the oRPC client the options create, so the kit checks the port as consumers see it, transport included. * @param options - How to make a client for an empty store under test. @@ -801,6 +734,37 @@ export const graphStoreConformance = ( }), ); + test( + "a subscriber that stops reading under sustained writes is never left unaware of a change it lost", + withClient(options, async (client) => { + await client.v1.models.save({ modelId: "pair", model: pairModel }); + const events = await client.v1.changes({ modelId: "pair" }); + // More writes than a subscription holds by default, so a store that buffers per subscription has to overflow; the property checked holds for any capacity. + const burst = await inSequence( + Array.from( + { + length: + defaultSubscriptionCapacity + defaultSubscriptionCapacity + 1, + }, + (_, write) => async () => + ( + await client.v1.overlay.set({ + modelId: "pair", + overlay: { plan: write % 2 === 0 ? "done" : "open" }, + }) + ).revision, + ), + ); + // Reading starts only now; the final write, the only save after subscribing, commits while the subscriber reads, so it must arrive. It is found by its kind because its event can arrive before the write's own reply does. + const received = takeUntil(events, ({ kind }) => kind === "saved"); + const final = ( + await client.v1.models.save({ modelId: "pair", model: pairModel }) + ).revision; + + expect(silentLosses(await received, [...burst, final])).toEqual([]); + }), + ); + test( "a subscriber that names a revision the model has moved past is told it missed changes", withClient(options, async (client) => { diff --git a/packages/core/graph/store-contract/src/contract.ts b/packages/core/graph/store-contract/src/contract.ts index e02488c..09224b7 100644 --- a/packages/core/graph/store-contract/src/contract.ts +++ b/packages/core/graph/store-contract/src/contract.ts @@ -42,7 +42,7 @@ const written = z.strictObject({ revision: revisionSchema }).readonly(); * - `overlay.set` sets the run state of the elements it names. * - `items.upsert` merges items read from a source into a model by identifier, creating the model when it does not exist. * - `query.ready`, `query.blockedChain` and `query.impact` answer structural questions from the stored transitions and run state alone (see the package README for exactly what they read), so a store can answer them without loading the model or evaluating conditions. - * - `changes` yields every later change to one model as an event iterator. + * - `changes` yields every later change to one model as an event iterator. A subscriber that falls behind by more than the store's subscription capacity gets one `missed` in place of the changes it let pile up, so a store's memory stays bounded and the subscriber knows to load the model again. * * Every write is atomic and checked: a change that would leave the model or overlay invalid (`validate`, `validateOverlay`) is rejected whole with `UNPROCESSABLE_CONTENT`. Every call takes the caller's abort signal through the oRPC client options; a call aborted before it writes leaves the store unchanged. */ diff --git a/packages/core/graph/store-contract/src/feed.ts b/packages/core/graph/store-contract/src/feed.ts index 6b2f26d..961fd3c 100644 --- a/packages/core/graph/store-contract/src/feed.ts +++ b/packages/core/graph/store-contract/src/feed.ts @@ -6,16 +6,62 @@ export interface ChangeSubscription extends AsyncIterable { readonly close: () => void; } +/** + * How many undelivered changes a subscription holds by default before it collapses them into `missed`. + * + * A change tells a subscriber only that its model has moved on to a later revision, and the one thing a subscriber can do about it is read the model again. A subscriber that has not yet taken one pending change will read the model after that change, and the model it reads already reflects every later one, so a second pending change gives it nothing it could act on: one held change is all a lagging subscriber can use. A subscriber that is waiting when a change commits receives it directly and never touches the buffer, so a subscriber that keeps reading misses nothing at this capacity. + */ +export const defaultSubscriptionCapacity = 1; + +/** Options for one subscription. */ +export interface SubscribeOptions { + /** Ends the subscription when it fires. */ + readonly signal?: AbortSignal | undefined; + /** Overrides the feed's capacity for this subscription; see `ChangeFeedOptions.capacity`. */ + readonly capacity?: number; +} + +/** Options for a feed. */ +export interface ChangeFeedOptions { + /** + * The most undelivered changes a subscription holds; `defaultSubscriptionCapacity` when absent. When a change arrives and the subscription already holds this many, every held change is dropped and replaced by one `missed` event, so the buffer never grows past the capacity and the subscriber knows to load the model again. A positive integer. + */ + readonly capacity?: number; +} + +/** Checks a capacity is a positive integer, since a subscription must be able to hold the `missed` it reports. */ +const checkedCapacity = (capacity: number): number => { + if (!Number.isInteger(capacity) || capacity < 1) { + throw new RangeError( + `A subscription capacity must be a positive integer, not ${capacity}`, + ); + } + + return capacity; +}; + /** * Fans the changes one store makes out to every subscriber of the model they touch. A store publishes after each committed write; the router reads a subscription for `changes`. * - * A subscription buffers changes that arrive while its consumer is busy, so none is dropped between two pulls, and it is registered the moment `subscribe` returns, so a caller can subscribe first and then read the current revision without missing a change made in between. + * A subscription is registered the moment `subscribe` returns, so a caller can subscribe first and then read the current revision without missing a change made in between. A change that arrives while the subscriber is waiting is handed straight to it; one that arrives while it is busy is held, up to the subscription's capacity. Past that, the held changes are replaced by a single `missed` event: memory stays bounded however slowly a subscriber reads, and a subscriber is never left unaware that it fell behind. * * A feed only sees the writes made through the store that owns it. Several processes or tabs writing to one database each need their own bridge (a `BroadcastChannel`, a database notification) that publishes the others' changes into their feed. */ export class ChangeFeed { readonly #listeners = new Map void>>(); + readonly #capacity: number; + + /** + * @param options - The capacity every subscription gets unless it sets its own. + * @throws RangeError when the capacity is not a positive integer. + */ + constructor(options: ChangeFeedOptions = {}) { + this.#capacity = checkedCapacity( + options.capacity ?? defaultSubscriptionCapacity, + ); + } + /** * Delivers a change to every current subscriber of its model. * @param change - The change, after it is committed. @@ -29,11 +75,17 @@ export class ChangeFeed { /** * Subscribes to one model's changes from now on. * @param modelId - The model to follow. - * @param signal - Ends the subscription when it fires. + * @param options - The signal that ends it, and a capacity overriding the feed's. * @returns The live subscription. + * @throws RangeError when the capacity is not a positive integer. */ - subscribe(modelId: string, signal?: AbortSignal): ChangeSubscription { - const buffered: StoreChange[] = []; + subscribe( + modelId: string, + options: SubscribeOptions = {}, + ): ChangeSubscription { + const { signal } = options; + const capacity = checkedCapacity(options.capacity ?? this.#capacity); + let buffered: StoreChange[] = []; // The pull waiting for the next change, when the buffer was empty as it asked. let waiting: ((result: IteratorResult) => void) | undefined; @@ -47,7 +99,10 @@ export class ChangeFeed { const listener = (change: StoreChange): void => { if (waiting === undefined) { - buffered.push(change); + buffered = + buffered.length < capacity + ? [...buffered, change] + : [{ modelId, kind: "missed" }]; } else { waiting({ done: false, value: change }); waiting = undefined; @@ -74,7 +129,8 @@ export class ChangeFeed { const iterator: AsyncIterator = { next: async () => { - const change = buffered.shift(); + const [change, ...rest] = buffered; + buffered = rest; if (change !== undefined) { return { done: false, value: change }; } diff --git a/packages/core/graph/store-contract/src/feed.unit.test.ts b/packages/core/graph/store-contract/src/feed.unit.test.ts index dfc3f49..c08cdb1 100644 --- a/packages/core/graph/store-contract/src/feed.unit.test.ts +++ b/packages/core/graph/store-contract/src/feed.unit.test.ts @@ -1,12 +1,14 @@ import { describe, expect, test } from "vitest"; import { ChangeFeed } from "./feed.ts"; +import type { StoreChange } from "./schema.ts"; const saved = (modelId: string, revision: number) => ({ modelId, kind: "saved", revision }) as const; describe("ChangeFeed", () => { test("buffers changes published before the subscriber pulls, in order", async () => { - const feed = new ChangeFeed(); + // Room for both changes this subscriber is sent, so the test is about order, not capacity. + const feed = new ChangeFeed({ capacity: 2 }); const subscription = feed.subscribe("m"); const [mine, theirs, later] = ["m", "other", "m"].map((modelId, index) => saved(modelId, index + 1), @@ -39,7 +41,7 @@ describe("ChangeFeed", () => { const feed = new ChangeFeed(); const controller = new AbortController(); const iterator = feed - .subscribe("m", controller.signal) + .subscribe("m", { signal: controller.signal }) [Symbol.asyncIterator](); const next = iterator.next(); controller.abort(); @@ -56,7 +58,7 @@ describe("ChangeFeed", () => { const controller = new AbortController(); controller.abort(); const feed = new ChangeFeed(); - const subscription = feed.subscribe("m", controller.signal); + const subscription = feed.subscribe("m", { signal: controller.signal }); feed.publish(saved("m", 1)); await expect(subscription[Symbol.asyncIterator]().next()).resolves.toEqual({ @@ -65,3 +67,98 @@ describe("ChangeFeed", () => { }); }); }); + +describe("ChangeFeed capacity", () => { + const capacity = 3; + /** Enough writes to overflow the buffer more than once. */ + const writes = capacity + capacity + 1; + + /** Publishes `count` changes to a subscriber that reads nothing, then reads everything it held. */ + const heldAfter = async (count: number) => { + const feed = new ChangeFeed({ capacity }); + const subscription = feed.subscribe("m"); + for (let revision = 1; revision <= count; revision += 1) { + feed.publish(saved("m", revision)); + } + subscription.close(); + const held = []; + for await (const change of subscription) { + held.push(change); + } + + return held; + }; + + test("a subscriber that reads nothing never holds more than its capacity, and is told it missed changes once it overflows", async () => { + const counts = Array.from({ length: writes }, (_, index) => index + 1); + const helds = await Promise.all(counts.map(heldAfter)); + + for (const [index, held] of helds.entries()) { + const count = index + 1; + expect(held.length).toBeLessThanOrEqual(capacity); + expect(count > capacity ? held[0] : held).toEqual( + count > capacity + ? { modelId: "m", kind: "missed" } + : counts.slice(0, count).map((revision) => saved("m", revision)), + ); + } + }); + + test("after a missed, the changes that follow are delivered in order", async () => { + const held = await heldAfter(capacity + 2); + + expect(held).toEqual([ + { modelId: "m", kind: "missed" }, + saved("m", capacity + 2), + ]); + }); + + test("a subscriber that keeps reading misses nothing, however many changes are published", async () => { + const feed = new ChangeFeed({ capacity: 1 }); + const iterator = feed.subscribe("m")[Symbol.asyncIterator](); + // Each change is published while the subscriber waits for it, then taken before the next is published. + const received = await Array.from( + { length: writes }, + (_, index) => index + 1, + ).reduce>( + async (done, revision) => { + const taken = await done; + const next = iterator.next(); + feed.publish(saved("m", revision)); + + const result = await next; + + return [...taken, result.done === true ? undefined : result.value]; + }, + Promise.resolve([]), + ); + + expect(received).toEqual( + Array.from({ length: writes }, (_, index) => saved("m", index + 1)), + ); + }); + + test("a subscription's own capacity overrides the feed's", async () => { + const feed = new ChangeFeed({ capacity }); + const subscription = feed.subscribe("m", { capacity: 1 }); + feed.publish(saved("m", 1)); + feed.publish(saved("m", 2)); + subscription.close(); + const held = []; + for await (const change of subscription) { + held.push(change); + } + + expect(held).toEqual([{ modelId: "m", kind: "missed" }]); + }); + + test.each([0, -1, Number.EPSILON, Number.NaN])( + "refuses a capacity of %s", + (bad) => { + expect(() => new ChangeFeed({ capacity: bad })).toThrow(RangeError); + expect(() => new ChangeFeed().subscribe("m", { capacity: bad })).toThrow( + RangeError, + ); + }, + ); +}); diff --git a/packages/core/graph/store-contract/src/index.ts b/packages/core/graph/store-contract/src/index.ts index df64f95..44ed2e4 100644 --- a/packages/core/graph/store-contract/src/index.ts +++ b/packages/core/graph/store-contract/src/index.ts @@ -26,7 +26,13 @@ export { type StoredModel, } from "./schema.ts"; export { GraphStoreError, type GraphStoreFailure } from "./errors.ts"; -export { ChangeFeed, type ChangeSubscription } from "./feed.ts"; +export { + ChangeFeed, + defaultSubscriptionCapacity, + type ChangeFeedOptions, + type ChangeSubscription, + type SubscribeOptions, +} from "./feed.ts"; export type { ElementRequest, GraphStore, ModelRequest } from "./store.ts"; export { checkStored, diff --git a/packages/core/graph/store-contract/src/router.ts b/packages/core/graph/store-contract/src/router.ts index b04c262..adb24b4 100644 --- a/packages/core/graph/store-contract/src/router.ts +++ b/packages/core/graph/store-contract/src/router.ts @@ -119,7 +119,7 @@ export const createGraphStoreRouter = (store: GraphStore): GraphStoreRouter => }, changes: os.v1.changes.handler(async ({ input, signal, errors }) => { // Subscribe before anything is awaited and before the iterator is returned: a caller that has the iterator in hand is already subscribed, over any transport, so a write it makes next is never missed. - const subscription = store.feed.subscribe(input.modelId, signal); + const subscription = store.feed.subscribe(input.modelId, { signal }); let missed = false; try { if (input.since !== undefined) { diff --git a/packages/core/graph/store-contract/src/schema.ts b/packages/core/graph/store-contract/src/schema.ts index b235ad6..45ebf37 100644 --- a/packages/core/graph/store-contract/src/schema.ts +++ b/packages/core/graph/store-contract/src/schema.ts @@ -65,7 +65,7 @@ export const reachSchema = z export const changeKinds = ["saved", "overlay", "items", "deleted"] as const; /** - * A change to a model. `revision` is the revision the change took. `missed` is sent only at the start of a subscription that names a revision the model is no longer at: the subscriber missed one or more changes, or the model has gone, and should load it again. + * A change to a model. `revision` is the revision the change took. `missed` says the subscriber missed one or more changes and should load the model again: it is sent at the start of a subscription that names a revision the model is no longer at (the model changed, or has gone), and in place of the changes a subscriber that reads too slowly has let pile up beyond its subscription's capacity. */ export const storeChangeSchema = z.union([ z