diff --git a/docs/book/src/reference/mcp-tools.md b/docs/book/src/reference/mcp-tools.md index 57b2f446e..d979a2c6b 100644 --- a/docs/book/src/reference/mcp-tools.md +++ b/docs/book/src/reference/mcp-tools.md @@ -5,8 +5,8 @@ integration, from its own MCP server: `quipu mcp --db ` over stdio, or `quipu-server` at `/mcp` over HTTP. (Bobbin embeds Quipu but serves its own `knowledge_*` tools, not these.) -The registry (`tool_definitions()`) exposes **79 tools** in a default build, or -**81** when built with the `owl` feature (which adds `quipu_load_ontology` and `quipu_explain`). +The registry (`tool_definitions()`) exposes **81 tools** in a default build, or +**83** when built with the `owl` feature (which adds `quipu_load_ontology` and `quipu_explain`). (The counts are pinned by tests in `src/mcp/tests.rs`, which also check this page and the README against the manifest.) @@ -1075,16 +1075,28 @@ Fields: `update` (required), `actor`, `source`. ### `quipu_events` -Read a bounded event batch without committing a consumer cursor. since overrides consumer. types is a comma-separated event-type filter; group is an episode group, not a graph selector. +Read a bounded batch of ROOT's events without committing a consumer cursor. since overrides consumer. types is a comma-separated event-type filter; group is an episode group, not a graph selector. A named graph's events are quipu_graph_events. Fields: `since`, `limit`, `types`, `group`, `consumer`. +### `quipu_graph_events` + +Read a bounded batch of ONE registered committed named graph's events (its own log and offsets) without committing a cursor. Same filters as quipu_events. + +Fields: `graph` (required), `since`, `limit`, `types`, `group`, `consumer`. + ### `quipu_events_commit` Durably set a consumer cursor. Lower offsets explicitly rewind; delivery alone never commits. Fields: `consumer_id` (required), `offset` (required). +### `quipu_graph_events_commit` + +Durably set a consumer cursor in one registered committed named graph's event log. Lower offsets explicitly rewind. + +Fields: `consumer_id` (required), `graph` (required), `offset` (required). + ### `quipu_changes` Read whole transaction change pages. graph scopes an exact graph; use next_tx for the next page. No cursor write. diff --git a/docs/book/src/reference/rest-api.md b/docs/book/src/reference/rest-api.md index ffbafd6f1..fb530c39a 100644 --- a/docs/book/src/reference/rest-api.md +++ b/docs/book/src/reference/rest-api.md @@ -2225,6 +2225,26 @@ Returns `{events, next_offset, lag, committed_offset?}`; pass `next_offset` back as `since` (or commit it) to page forward — polling is a fixpoint, not a rewind. +`/events` serves ROOT's events, as it always has, and refuses a `graph` +parameter (HTTP 400). + +### `GET /events/graph` + +`graph=` (required) plus the `/events` parameters: one registered +committed named graph's events. A commit to such a graph emits +`episode.ingested` and `entity.added` / `entity.updated` (one per entity per +transaction, "added" judged within that graph), each carrying a `graph` key. +Overlays and unregistered graphs emit nothing, and an unregistered or overlay +`graph` returns HTTP 400. The graph's events are a separate log with its own +offsets: `since`, `next_offset`, `consumer` resume and `lag` are all in that +log, and ROOT's offsets and cursors are untouched by board writes. + +Named graphs have their own routes, not a `graph` parameter on `/events`, +so that an older server cannot misread them. A server that predates +named-graph events ignores an unknown query parameter and would serve ROOT's +events; it answers an unknown route 404. So against an older server a graph +consumer gets a 404, never ROOT's events and never a ROOT cursor write. + ### `GET /events/stream` and `GET /changes/stream` Read-only Server-Sent Events replay of the corresponding durable feed. Supply a @@ -2233,8 +2253,11 @@ frames have event type `quipu.events`, a delivered event-offset ID, and the ordinary `/events` page as JSON data. Change frames use `quipu.changes`, a transaction-ID cursor and the ordinary `/changes` page as JSON data. -Event streams accept `types` and `group`; change streams accept `graph`, -including a registered named graph. Unrelated filter kinds and invalid resume +Event streams accept `types`, `group` and `graph` (one registered committed +named graph, read from its own log as on `/events/graph`; a server that +predates named-graph events refuses `graph` on an event stream with HTTP +400); change streams accept `graph`, including a +registered named graph. Unrelated filter kinds and invalid resume cursors return HTTP 400. Exhausted stream capacity returns HTTP 503. Delivery does not acknowledge a consumer. Commit applied data and its cursor @@ -2252,8 +2275,18 @@ before the output check; this is not a total heap bound. See the ### `POST /events/commit` -`{"consumer_id", "offset"}` — durably record a consumer's cursor. Any offset -≥ 0 is accepted, including a **lower** one: that is the explicit replay knob. +`{"consumer_id", "offset"}` — durably record a ROOT consumer's cursor. Any +offset ≥ 0 is accepted, including a **lower** one: that is the explicit replay +knob. A `graph` field is refused (HTTP 400): a server that predates +named-graph events would ignore it and upsert this consumer's ROOT cursor +before answering. + +### `POST /events/graph/commit` + +`{"consumer_id", "graph", "offset"}` — durably record a consumer's cursor in +one registered committed named graph's event log (the response echoes +`graph`). The same `consumer_id` may also hold a ROOT cursor; the two never +touch. An older server answers this route 404 and writes nothing. ### Refusal events (`write.refused`) diff --git a/docs/book/src/why-quipu.md b/docs/book/src/why-quipu.md index 7b8dd7e83..67d4d9a48 100644 --- a/docs/book/src/why-quipu.md +++ b/docs/book/src/why-quipu.md @@ -182,7 +182,7 @@ Quipu's thesis: **start strict, use agents to bear the cost of strictness.** - **Graph projection** — materialize subgraphs into petgraph for centrality, connected components, shortest path algorithms. - **Federation** — a `GraphProvider` trait for multi-source queries, with a `RemoteProvider` (behind the `remote` feature) built from `federation.remotes` config. The server health-checks every configured remote at startup, and `POST /query` with `"federated": true` fans out through the federated provider, reporting which members answered. Remotes carry declared trust labels at the federation edge, so a federated answer composes the labels of every member that contributed rather than silently inheriting the caller's. Federation config is read-side only: adding a remote never turns it into an outbound replication target or bypasses the share scrub/import boundary. - **Graph explorer** — the web UI draws the whole node-link view from a single `POST /graph` payload (nodes plus index-addressed edges), laid out with a Barnes-Hut force simulation on canvas. No CDN, so it renders on an air-gapped deploy. -- **Four interfaces** — Rust crate (embed), CLI (`quipu`), REST API (`quipu-server`), and built-in web UI with embeddable web components. Plus 79 MCP tools for agent integration (81 with the `owl` feature). +- **Four interfaces** — Rust crate (embed), CLI (`quipu`), REST API (`quipu-server`), and built-in web UI with embeddable web components. Plus 81 MCP tools for agent integration (83 with the `owl` feature). - **"SQLite energy"** — single process, no server required, inspect with `sqlite3`, back up with `cp`. - **Automated releases** — release-plz bumps versions from conventional commits, generates changelogs via git-cliff, and creates GitHub releases. Version discovery is `git_only = true`: the baseline comes from this repository's tags, never the unrelated `quipu` crate on crates.io. CI runs fmt, clippy, tests, and markdown lint on every push. `/version` also reports the deployed git SHA, which matters because a deployment can legitimately sit AHEAD of the newest tag — the SHA, not the version string, identifies what is actually running. @@ -305,7 +305,7 @@ The reasoner adds forward-chaining inference over the EAVT fact log: │ │ │ ┌─────┴─────┐ ┌─────┴─────┐ ┌──────┴──────┐ │ MCP Tools │ │ REST API │ │ Rust API │ - │ (79 tools) │ │ + Web UI │ │ (crate) │ + │ (81 tools) │ │ + Web UI │ │ (crate) │ └─────┬─────┘ └─────┬─────┘ └──────┬──────┘ └────────────────┼────────────────┘ │ @@ -334,7 +334,7 @@ Quipu is designed as a [Bobbin](https://github.com/scbrown/bobbin) subsystem. Bobbin holds the thread (code context); Quipu ties knots of structured meaning into it. Through Quipu's own MCP server (`quipu mcp`, or `quipu-server` at `/mcp`), -agents get 79 MCP tools (81 with the `owl` feature). The two most +agents get 81 MCP tools (83 with the `owl` feature). The two most commonly used for knowledge-aware context: **`quipu_context`** — unified knowledge discovery. Bobbin merges the result @@ -457,7 +457,7 @@ primitive only, not reachable from the shipped binaries · 🔜 planned. | Graph explorer | ✅ | Canvas + Barnes-Hut layout, one `POST /graph` payload, no CDN | | Web components | ✅ | Embeddable `` elements | | Semantic Web APIs | ✅ | Spotlight, TPF, OpenRefine reconciliation | -| MCP tools (79; 81 with `owl`) | ✅ | Agent integration | +| MCP tools (81; 83 with `owl`) | ✅ | Agent integration | | Python bindings | ✅ | `quipu-client` under `python/` — REST client, stdlib-only | | **Infrastructure** | | | | Graph projection (petgraph) | ✅ | Centrality, shortest path, etc. | diff --git a/docs/knowledge/repository-knowledge.json b/docs/knowledge/repository-knowledge.json index 1899d2385..b57285cd3 100644 --- a/docs/knowledge/repository-knowledge.json +++ b/docs/knowledge/repository-knowledge.json @@ -117,7 +117,7 @@ "source": "docs/book/src/reference/rest-api.md", "subject": "http://aegis.gastown.local/ontology/doc/quipu/docs%2Fbook%2Fsrc%2Freference%2Frest-api.md#rest-api", "summary": "The native server exposes graph operations over HTTP; authentication, request budgets and provenance headers have separate roles.", - "source_sha256": "44abde06edafbfaa99807e36a5bee20f69912c2525403d4f017a088cfe9c1942", + "source_sha256": "3a1d7d312e0ea77ac965b072abe7bd673f71898b1b39d7e4e608ceaecccf63b9", "properties": { "interfaceKind": "REST", "publicName": "quipu-server HTTP" diff --git a/src/http_auth.rs b/src/http_auth.rs index caa5bb6b5..fbd00cc6c 100644 --- a/src/http_auth.rs +++ b/src/http_auth.rs @@ -116,6 +116,7 @@ pub const WRITE_ENDPOINTS: &[&str] = &[ "/graph/freeze", // deep freeze: relocates rows, mutates registry + attachments "/graph/thaw", // restores rows, mutates registry + attachments "/events/commit", // durable consumer cursor upsert (event-log P1) + "/events/graph/commit", // a named graph's consumer cursor upsert (event-push §4) "/subscriptions", // push-subscription registry create/list/delete (event-log P2) "/datasets", // named-dataset registry create/remove (quipu #69) + meta-graph mirror "/update", // SPARQL 1.1 Update mutates default and named graphs @@ -213,10 +214,11 @@ pub const READ_ENDPOINTS: &[&str] = &[ "/entity/{iri}/html", "/entity_history", "/transactions", - "/events", // pull-batch event log read (event-log P1); the commit half is a write + "/events", // pull-batch event log read (event-log P1); the commit half is a write + "/events/graph", // one named graph's event log read (event-push §4) "/events/stream", // read-only replay; delivery never commits a consumer ACK "/changes/stream", // read-only transaction replay, including named graphs - "/changes", // fact-level change feed (quipu-2ae): pull-only, cursor is a tx id + "/changes", // fact-level change feed (quipu-2ae): pull-only, cursor is a tx id "/spotlight", "/fragments", "/reconcile", diff --git a/src/mcp/definitions/parity/workflow.rs b/src/mcp/definitions/parity/workflow.rs index 6dc111569..ffc8d22ad 100644 --- a/src/mcp/definitions/parity/workflow.rs +++ b/src/mcp/definitions/parity/workflow.rs @@ -50,7 +50,7 @@ pub(super) fn defs() -> Vec { }), serde_json::json!({ "name": "quipu_events", - "description": "Read a bounded event batch without committing a consumer cursor. since overrides consumer. types is a comma-separated event-type filter; group is an episode group, not a graph selector.", + "description": "Read a bounded batch of ROOT's events without committing a consumer cursor. since overrides consumer. types is a comma-separated event-type filter; group is an episode group, not a graph selector. A named graph's events are quipu_graph_events.", "inputSchema": { "type": "object", "properties": { @@ -74,6 +74,37 @@ pub(super) fn defs() -> Vec { "additionalProperties": false } }), + serde_json::json!({ + "name": "quipu_graph_events", + "description": "Read a bounded batch of ONE registered committed named graph's events (its own log and offsets) without committing a cursor. Same filters as quipu_events.", + "inputSchema": { + "type": "object", + "properties": { + "graph": { + "type": "string" + }, + "since": { + "type": "integer" + }, + "limit": { + "type": "integer" + }, + "types": { + "type": "string" + }, + "group": { + "type": "string" + }, + "consumer": { + "type": "string" + } + }, + "required": [ + "graph" + ], + "additionalProperties": false + } + }), serde_json::json!({ "name": "quipu_events_commit", "description": "Durably set a consumer cursor. Lower offsets explicitly rewind; delivery alone never commits.", @@ -95,6 +126,31 @@ pub(super) fn defs() -> Vec { "additionalProperties": false } }), + serde_json::json!({ + "name": "quipu_graph_events_commit", + "description": "Durably set a consumer cursor in one registered committed named graph's event log. Lower offsets explicitly rewind.", + "inputSchema": { + "type": "object", + "properties": { + "consumer_id": { + "type": "string" + }, + "graph": { + "type": "string" + }, + "offset": { + "type": "integer", + "minimum": 0 + } + }, + "required": [ + "consumer_id", + "graph", + "offset" + ], + "additionalProperties": false + } + }), serde_json::json!({ "name": "quipu_changes", "description": "Read whole transaction change pages. graph scopes an exact graph; use next_tx for the next page. No cursor write.", diff --git a/src/mcp/tests.rs b/src/mcp/tests.rs index 8d0d03176..d72525f69 100644 --- a/src/mcp/tests.rs +++ b/src/mcp/tests.rs @@ -1955,13 +1955,13 @@ fn test_tool_definitions() { assert!(names.contains(&"quipu_merge_apply")); #[cfg(feature = "owl")] { - assert_eq!(defs.len(), 81); + assert_eq!(defs.len(), 83); assert!(names.contains(&"quipu_load_ontology")); assert!(names.contains(&"quipu_explain")); } #[cfg(not(feature = "owl"))] { - assert_eq!(defs.len(), 79); + assert_eq!(defs.len(), 81); assert!(!names.contains(&"quipu_load_ontology")); assert!(!names.contains(&"quipu_explain")); } diff --git a/src/mcp_transport.rs b/src/mcp_transport.rs index de609646a..61c09e900 100644 --- a/src/mcp_transport.rs +++ b/src/mcp_transport.rs @@ -39,6 +39,7 @@ fn endpoint(name: &str) -> (Method, String) { "quipu_health" => return (Method::GET, "/health".into()), "quipu_version" => return (Method::GET, "/version".into()), "quipu_events" => return (Method::GET, "/events".into()), + "quipu_graph_events" => return (Method::GET, "/events/graph".into()), "quipu_changes" => return (Method::GET, "/changes".into()), "quipu_transactions" => return (Method::GET, "/transactions".into()), "quipu_entity" => return (Method::GET, "/entity".into()), @@ -47,6 +48,7 @@ fn endpoint(name: &str) -> (Method, String) { "quipu_import_promote" => "/import/promote", "quipu_acquire_composed" => "/acquire/composed", "quipu_events_commit" => "/events/commit", + "quipu_graph_events_commit" => "/events/graph/commit", "quipu_knot_stage" => "/knot/stage", "quipu_knot_promote" => "/knot/promote", "quipu_graph_create" => "/graph/create", diff --git a/src/mcp_transport/parity_tests.rs b/src/mcp_transport/parity_tests.rs index fdc31606d..af3970cb6 100644 --- a/src/mcp_transport/parity_tests.rs +++ b/src/mcp_transport/parity_tests.rs @@ -241,6 +241,8 @@ fn write_annotations_cover_new_workflows_and_reads_stay_read_only() { ("quipu_reason", true), ("quipu_events", false), ("quipu_events_commit", true), + ("quipu_graph_events", false), + ("quipu_graph_events_commit", true), ("quipu_changes", false), ("quipu_subscriptions", true), ("quipu_embed_backfill", true), diff --git a/src/server.rs b/src/server.rs index 6f50807ed..b2bceb147 100644 --- a/src/server.rs +++ b/src/server.rs @@ -509,6 +509,8 @@ async fn main() { .route("/events/stream", get(feed_stream::events_stream)) .route("/changes/stream", get(feed_stream::changes_stream)) .route("/events/commit", post(events_commit)) + .route("/events/graph", get(feed::events_graph_get)) + .route("/events/graph/commit", post(feed::events_graph_commit)) // Semantic web APIs (Phase 4) .route("/spotlight", post(spotlight_handler)) .route("/fragments", get(fragments_handler)) diff --git a/src/server/feed.rs b/src/server/feed.rs index 25d8c4c84..6d7117ebd 100644 --- a/src/server/feed.rs +++ b/src/server/feed.rs @@ -58,6 +58,9 @@ pub(crate) struct EventParams { types: Option, /// Filter to a single `group_id` (episode grouping, e.g. `aegis-ontology`). group: Option, + /// One registered named graph's events; absent means ROOT's, as before + /// graphs could emit. Combines with `types` and `group` by AND. + graph: Option, /// Resume from this consumer's durable committed offset. consumer: Option, } @@ -117,13 +120,46 @@ pub(crate) async fn changes_get( .await } -/// GET /events — pull a batch of graph-change events in offset order. +/// GET /events — pull a batch of ROOT's graph-change events in offset order. /// Response: `{events, next_offset, lag, committed_offset?}`; pass /// `next_offset` back as `since` (or POST /events/commit it) to page forward. +/// +/// A named graph's events are read at `/events/graph`, never here: a server +/// that predates named-graph events ignores an unknown `graph` parameter and +/// would answer with ROOT's events, while it answers an unknown route 404. +/// So `graph` is refused here, and a client never learns to depend on it. pub(crate) async fn events_get( State(store): State, headers: HeaderMap, Query(p): Query, +) -> Result, AppError> { + if p.graph.is_some() { + return Err(quipu::Error::InvalidValue( + "/events serves ROOT's events; read a named graph's at /events/graph".into(), + ) + .into()); + } + events_batch(store, headers, p, None).await +} + +/// GET /events/graph?graph= — one registered committed named graph's +/// events, its own log and offsets. Same parameters and response as `/events`. +pub(crate) async fn events_graph_get( + State(store): State, + headers: HeaderMap, + Query(mut p): Query, +) -> Result, AppError> { + let Some(graph) = p.graph.take() else { + return Err(quipu::Error::InvalidValue("/events/graph requires graph".into()).into()); + }; + events_batch(store, headers, p, Some(graph)).await +} + +async fn events_batch( + store: SharedStore, + headers: HeaderMap, + p: EventParams, + graph: Option, ) -> Result, AppError> { let client = super::query_usage::client(&headers); blocking(move || { @@ -135,7 +171,7 @@ pub(crate) async fn events_get( let store = store.read(); let _timing = StoreTiming::acquired(client, "/events", waiting.elapsed()); let committed: Option = match (&p.since, &p.consumer) { - (None, Some(c)) => Some(store.consumer_committed(c)?), + (None, Some(c)) => Some(store.consumer_committed_in(c, graph.as_deref())?), _ => None, }; let since = p.since.unwrap_or_else(|| committed.unwrap_or(0)); @@ -147,13 +183,28 @@ pub(crate) async fn events_get( .map(String::from) .collect() }); - let rows = store.events_after(since, limit, types.as_deref(), p.group.as_deref())?; + if let Some(iri) = graph.as_deref() + && store.committed_graph_id(iri)?.is_none() + { + return Err(quipu::Error::InvalidValue(format!( + "not a registered committed named graph: {iri}" + )) + .into()); + } + let rows = store.events_after_in( + since, + limit, + types.as_deref(), + p.group.as_deref(), + graph.as_deref(), + )?; // next_offset is the cursor for the NEXT call; when the batch is empty // it stays at `since` so polling is a fixpoint, not a rewind. let next_offset = rows.last().map_or(since, |r| r.offset); - // Lag counts ALL events beyond the cursor, unfiltered — it answers - // "how far behind the log am I", not "how many match my filter". - let latest = store.latest_event_offset()?; + // Lag counts ALL events of the reader's graph scope beyond the cursor, + // unfiltered by type or group — "how far behind my log am I", not "how + // many match my filter". Other graphs' events are not this log. + let latest = store.latest_event_offset_in(graph.as_deref())?; let lag = (latest - next_offset).max(0); let events: Vec = rows .iter() @@ -172,12 +223,43 @@ pub(crate) async fn events_get( .await } -/// POST /events/commit `{consumer_id, offset}` — durably record a consumer's -/// cursor. Any offset >= 0 is accepted, including a LOWER one (the explicit -/// replay knob; delivery is at-least-once and consumers dedup by offset). +/// POST /events/commit `{consumer_id, offset}` — durably record a ROOT +/// consumer's cursor. Any offset >= 0 is accepted, including a LOWER one (the +/// explicit replay knob; delivery is at-least-once and consumers dedup by +/// offset). +/// +/// A named graph's cursor is committed at `/events/graph/commit`, never here. +/// A server that predates named-graph events would ignore `graph` and upsert +/// the ROOT cursor of that `consumer_id` BEFORE answering, moving a ROOT +/// consumer past events it never received. It answers an unknown route 404 and +/// writes nothing. So `graph` is refused here. pub(crate) async fn events_commit( State(store): State, axum::Json(input): axum::Json, +) -> Result, AppError> { + if input.get("graph").is_some() { + return Err(quipu::Error::InvalidValue( + "/events/commit records ROOT cursors; commit a named graph's at /events/graph/commit" + .into(), + ) + .into()); + } + commit(store, input, false).await +} + +/// POST /events/graph/commit `{consumer_id, graph, offset}` — durably record a +/// consumer's cursor in one registered committed named graph's log. +pub(crate) async fn events_graph_commit( + State(store): State, + axum::Json(input): axum::Json, +) -> Result, AppError> { + commit(store, input, true).await +} + +async fn commit( + store: SharedStore, + input: JsonValue, + named: bool, ) -> Result, AppError> { blocking(move || { let consumer_id = input @@ -194,13 +276,35 @@ pub(crate) async fn events_commit( if offset < 0 { return Err(quipu::Error::InvalidValue("offset must be >= 0".into()).into()); } + let graph = match (named, input.get("graph")) { + (false, _) => None, + (true, Some(JsonValue::String(g))) => Some(g.clone()), + (true, _) => { + return Err(quipu::Error::InvalidValue( + "/events/graph/commit requires graph (a graph IRI)".into(), + ) + .into()); + } + }; let store = store.lock(); + if let Some(iri) = graph.as_deref() + && store.committed_graph_id(iri)?.is_none() + { + return Err(quipu::Error::InvalidValue(format!( + "not a registered committed named graph: {iri}" + )) + .into()); + } let now = quipu::time::now_iso(); - store.commit_consumer(&consumer_id, offset, &now)?; - Ok(axum::Json(json!({ + store.commit_consumer_in(&consumer_id, graph.as_deref(), offset, &now)?; + let mut answer = json!({ "consumer_id": consumer_id, "committed_offset": offset, - }))) + }); + if let Some(g) = graph { + answer["graph"] = json!(g); + } + Ok(axum::Json(answer)) }) .await } @@ -243,6 +347,9 @@ pub(crate) async fn transactions( .await } +#[cfg(test)] +#[path = "feed_graph_events_tests.rs"] +mod feed_graph_events_tests; #[cfg(test)] #[path = "feed_read_pool_tests.rs"] mod feed_read_pool_tests; diff --git a/src/server/feed_graph_events_tests.rs b/src/server/feed_graph_events_tests.rs new file mode 100644 index 000000000..15331b021 --- /dev/null +++ b/src/server/feed_graph_events_tests.rs @@ -0,0 +1,157 @@ +//! Named-graph events on their own routes (event-push design §4, §7). +//! +//! A server that predates named-graph events ignores an unknown `graph` +//! parameter: `/events?graph=G` would answer with ROOT's events, and +//! `/events/commit {graph}` would upsert the ROOT cursor of that consumer id +//! before answering. It answers an unknown ROUTE 404 and writes nothing. So +//! named graphs read and commit only at `/events/graph` and +//! `/events/graph/commit`, and the ROOT routes refuse `graph`, so no client +//! learns to depend on the unsafe form. + +use std::sync::Arc; + +use axum::extract::{Query, State}; +use axum::http::HeaderMap; +use serde_json::{Value, json}; + +use crate::feed; + +const BOARD: &str = "https://example.org/graph/board"; +const TS: &str = "2026-10-24T00:00:00Z"; + +fn store_with_root_and_board_events() -> crate::SharedStore { + let (_dir, handle) = crate::tests::pooled_handle(0); + std::mem::forget(_dir); + { + let mut s = handle.lock(); + quipu::rdf::ingest_rdf( + &mut s, + b" .".as_slice(), + oxrdfio::RdfFormat::Turtle, + None, + TS, + None, + None, + ) + .unwrap(); + s.graph_create_recorded(BOARD, TS, None).unwrap(); + let g = s.committed_graph_id(BOARD).unwrap().unwrap(); + let d = quipu::store::Datum { + entity: s.intern("urn:board:item").unwrap(), + attribute: s.intern("urn:p:x").unwrap(), + value: quipu::Value::Ref(s.intern("urn:o:1").unwrap()), + valid_from: TS.into(), + valid_to: None, + op: quipu::Op::Assert, + }; + s.transact_to_graph(&[d], TS, None, None, g).unwrap(); + } + Arc::new(handle) +} + +async fn get(s: &crate::SharedStore, graph_route: bool, q: Value) -> Result { + let p = serde_json::from_value(q).unwrap(); + let r = if graph_route { + feed::events_graph_get(State(s.clone()), HeaderMap::new(), Query(p)).await + } else { + feed::events_get(State(s.clone()), HeaderMap::new(), Query(p)).await + }; + r.map(|j| j.0).map_err(|e| format!("{e:?}")) +} + +async fn commit(s: &crate::SharedStore, graph_route: bool, body: Value) -> Result { + let r = if graph_route { + feed::events_graph_commit(State(s.clone()), axum::Json(body)).await + } else { + feed::events_commit(State(s.clone()), axum::Json(body)).await + }; + r.map(|j| j.0).map_err(|e| format!("{e:?}")) +} + +fn subjects(batch: &Value) -> Vec { + batch["events"] + .as_array() + .unwrap() + .iter() + .filter_map(|e| e["subject"].as_str().map(String::from)) + .collect() +} + +#[tokio::test] +async fn each_route_reads_only_its_own_log() { + let s = store_with_root_and_board_events(); + let root = get(&s, false, json!({"since": 0})).await.unwrap(); + assert!( + subjects(&root).iter().all(|x| !x.contains("urn:board")), + "{root}" + ); + let board = get(&s, true, json!({"graph": BOARD, "since": 0})) + .await + .unwrap(); + let board_subjects = subjects(&board); + assert!( + !board_subjects.is_empty() && board_subjects.iter().all(|x| x == "urn:board:item"), + "{board}" + ); + assert!( + board["events"] + .as_array() + .unwrap() + .iter() + .all(|e| e["graph"] == BOARD) + ); +} + +#[tokio::test] +async fn the_root_routes_refuse_graph_and_the_graph_routes_require_it() { + let s = store_with_root_and_board_events(); + let e = get(&s, false, json!({"graph": BOARD})).await.unwrap_err(); + assert!(e.contains("/events/graph"), "{e}"); + let e = get(&s, true, json!({"since": 0})).await.unwrap_err(); + assert!(e.contains("requires graph"), "{e}"); + let e = commit(&s, true, json!({"consumer_id": "c", "offset": 1})) + .await + .unwrap_err(); + assert!(e.contains("requires graph"), "{e}"); + let e = get( + &s, + true, + json!({"graph": "https://example.org/graph/unregistered"}), + ) + .await + .unwrap_err(); + assert!(e.contains("not a registered committed named graph"), "{e}"); +} + +/// ian's control, the new side: a ROOT reader at cursor 1 and a board reader +/// of the SAME consumer id. Neither the refused ROOT commit with `graph` nor +/// the graph commit may move the ROOT cursor. +#[tokio::test] +async fn a_graph_commit_never_moves_the_same_consumers_root_cursor() { + let s = store_with_root_and_board_events(); + commit(&s, false, json!({"consumer_id": "reader", "offset": 1})) + .await + .unwrap(); + let e = commit( + &s, + false, + json!({"consumer_id": "reader", "offset": 102, "graph": BOARD}), + ) + .await + .unwrap_err(); + assert!(e.contains("/events/graph/commit"), "{e}"); + let ok = commit( + &s, + true, + json!({"consumer_id": "reader", "graph": BOARD, "offset": 102}), + ) + .await + .unwrap(); + assert_eq!(ok["graph"], BOARD); + let root = get(&s, false, json!({"consumer": "reader"})).await.unwrap(); + assert_eq!(root["committed_offset"], 1, "ROOT cursor unmoved: {root}"); + let board = get(&s, true, json!({"graph": BOARD, "consumer": "reader"})) + .await + .unwrap(); + assert_eq!(board["committed_offset"], 102, "{board}"); +} diff --git a/src/server/feed_stream.rs b/src/server/feed_stream.rs index 73259585e..2339516b8 100644 --- a/src/server/feed_stream.rs +++ b/src/server/feed_stream.rs @@ -70,7 +70,6 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool Err(message) => return (StatusCode::BAD_REQUEST, message).into_response(), }; if changes && (params.types.is_some() || params.group.is_some()) - || !changes && params.graph.is_some() || params .graph .as_ref() @@ -158,9 +157,18 @@ fn stream(store: SharedStore, headers: &HeaderMap, params: Params, changes: bool } } else { let params = serde_json::from_value( - json!({"since": cursor.offset, "limit": 1, "types": cursor.params.types, "group": cursor.params.group}), + json!({"since": cursor.offset, "limit": 1, "types": cursor.params.types, "group": cursor.params.group, "graph": cursor.params.graph}), ); match params { + // A named graph streams from its own log; ROOT from `/events`. + Ok(params) if cursor.params.graph.is_some() => { + feed::events_graph_get( + State(cursor.store.clone()), + headers_for_feed(), + Query(params), + ) + .await + } Ok(params) => { feed::events_get( State(cursor.store.clone()), diff --git a/src/server/feed_stream_tests.rs b/src/server/feed_stream_tests.rs index 4fc48045d..382cc5036 100644 --- a/src/server/feed_stream_tests.rs +++ b/src/server/feed_stream_tests.rs @@ -59,13 +59,13 @@ async fn named_graph_changes_replay_without_event_offset_ack() { .transact_to_graph(&[first], "2026-01-01", None, None, graph) .unwrap(); assert_eq!( - store.latest_event_offset().unwrap(), + store.latest_event_offset_in(None).unwrap(), 0, "legacy ROOT event scope unchanged" ); println!( "DEMO legacy ROOT event offset: {}", - store.latest_event_offset().unwrap() + store.latest_event_offset_in(None).unwrap() ); let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); for replay in 0..2 { @@ -176,3 +176,47 @@ async fn inner_savepoint_hint_cannot_publish_speculative_rows() { 0 ); } + +#[tokio::test] +async fn an_event_stream_filtered_to_a_named_graph_carries_only_that_graph() { + let mut store = Store::open_in_memory().unwrap(); + let board = store.graph_create("urn:test:seeds-board").unwrap(); + let other = store.graph_create("urn:test:other-board").unwrap(); + let mut item = datum(&mut store, "urn:test:item-other"); + store + .transact_to_graph(&[item], "2026-01-01", None, None, other) + .unwrap(); + item = datum(&mut store, "urn:test:item-board"); + store + .transact_to_graph(&[item], "2026-01-01", None, None, board) + .unwrap(); + assert_eq!( + store.latest_event_offset_in(None).unwrap(), + 0, + "ROOT log untouched" + ); + let shared = Arc::new(super::super::handle::StoreHandle::writer_only(store)); + let response = events_stream( + State(shared), + HeaderMap::new(), + params(json!({"since":0,"graph":"urn:test:seeds-board"})), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let mut bytes = response.into_body().into_data_stream(); + let frame = tokio::time::timeout(Duration::from_secs(2), bytes.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + let text = String::from_utf8(frame.to_vec()).unwrap(); + assert!( + text.contains("quipu.events") && text.contains("entity.added"), + "{text}" + ); + assert!(text.contains("urn:test:item-board"), "{text}"); + assert!( + !text.contains("urn:test:item-other"), + "other graph leaked: {text}" + ); +} diff --git a/src/share_completeness.rs b/src/share_completeness.rs index d2913c4d4..d7d6fca9f 100644 --- a/src/share_completeness.rs +++ b/src/share_completeness.rs @@ -110,6 +110,8 @@ pub const DECLARED: &[(&str, Disposition)] = &[ ("projection_defs", Disposition::Regenerated), // -- log ---------------------------------------------------------------- ("events", Disposition::Log), + // Named-graph events (event-push design §4), the same kind of log. + ("graph_events", Disposition::Log), // -- excluded ----------------------------------------------------------- // A TRUST REGISTRY. Carrying it would undo aegis-tadzdf by the back door: // `share_attestation.rs` refuses to let a bundle register its own producer @@ -143,6 +145,8 @@ pub const DECLARED: &[(&str, Disposition)] = &[ ("deck_receipts", Disposition::Excluded), // A READER's cursor. Restoring it resumes someone else's position. ("consumers", Disposition::Excluded), + // The same, for named-graph event readers. + ("graph_consumers", Disposition::Excluded), // Receiver-local review decisions and notice routes. A foreign pack must // not install its producer's review position or silence local notices. ("import_reviews", Disposition::Excluded), diff --git a/src/store/events.rs b/src/store/events.rs index 44126f746..1fddecb6c 100644 --- a/src/store/events.rs +++ b/src/store/events.rs @@ -33,6 +33,15 @@ //! compose-only staging (#36); announcing their writes as graph-change events //! would double-fire when the content is promoted to ROOT. Recorded here so //! nobody reads the silence as a bug: only ROOT (g=0) commits are events. +//! +//! REGISTERED named graphs are the exception (event-push design §4): they are +//! durable, not staging, and their consumers otherwise poll. A commit to one +//! emits `episode.ingested` and `entity.added` / `entity.updated` — entity +//! level only, judged within that graph — into the separate `graph_events` +//! log, tagged with the graph IRI. No `edge.*` (the facts are on the change +//! feed), no `type.new` / `predicate.new` (store-wide, ROOT only). `events` +//! stays the ROOT log, byte for byte, so `events_after` and every older +//! binary see exactly what they always saw (see `migrate_graph_events`). use rusqlite::params; use serde_json::json; @@ -95,6 +104,8 @@ pub struct EventRow { pub tx_id: i64, /// Type-specific payload, stored as a JSON object string. pub payload: String, + /// The registered named graph this event belongs to; `None` for ROOT. + pub graph: Option, } impl EventRow { @@ -102,7 +113,7 @@ impl EventRow { pub fn to_json(&self) -> serde_json::Value { let payload: serde_json::Value = serde_json::from_str(&self.payload).unwrap_or_else(|_| json!({})); - json!({ + let mut json = json!({ "schema": EVENT_SCHEMA, "offset": self.offset, "type": self.event_type, @@ -111,7 +122,13 @@ impl EventRow { "group_id": self.group_id, "tx_id": self.tx_id, "payload": payload, - }) + }); + // Only named-graph events carry the key, so a ROOT event's wire form + // is byte-for-byte what it was before graphs could emit. + if let Some(graph) = &self.graph { + json["graph"] = json!(graph); + } + json } } @@ -149,11 +166,16 @@ impl Store { // even when this tx emits no semantic events (early returns below). let pending = std::mem::take(&mut *self.pending_write_events.borrow_mut()); - // ROOT-graph commits only (see module docs). Advisory events follow the - // same rule: an overlay write discards them (compose-only staging). - if graph != 0 || (asserts.is_empty() && retracts.is_empty() && pending.is_empty()) { + // ROOT and registered named graphs only (see module docs). Advisory + // events stay ROOT-only: a named-graph write discards them, as an + // overlay write always has. + if asserts.is_empty() && retracts.is_empty() && (graph != 0 || pending.is_empty()) { return Ok(()); } + if graph != 0 { + return self + .emit_named_graph_events(asserts, retracts, tx_id, timestamp, source, graph); + } let rdf_type_id: Option = self.lookup(RDF_TYPE)?; @@ -490,7 +512,18 @@ impl Store { "DELETE FROM events WHERE ts < ?1 AND \"offset\" <= ?2", params![older_than, safe], )?; - Ok(deleted as u64) + // Each graph's log is held back only by that graph's consumers. + let graph_deleted = if self.has_graph_events()? { + self.conn.execute( + "DELETE FROM graph_events WHERE ts < ?1 AND \"offset\" <= COALESCE(\ + (SELECT MIN(c.committed_offset) FROM graph_consumers c \ + WHERE c.graph = graph_events.graph), ?2)", + params![older_than, i64::MAX], + )? + } else { + 0 + }; + Ok((deleted + graph_deleted) as u64) } // -- Pull API --------------------------------------------------------- @@ -504,12 +537,43 @@ impl Store { types: Option<&[String]>, group: Option<&str>, ) -> Result> { - // Filters are dynamic; build the WHERE clause with positional params. - let mut sql = String::from( - "SELECT \"offset\", type, ts, subject, group_id, tx_id, payload \ - FROM events WHERE \"offset\" > ?1", - ); + self.events_after_in(since, limit, types, group, None) + } + + /// [`Self::events_after`] for one graph's events: `None` is ROOT (the log + /// every untagged reader has always seen), `Some(iri)` one registered + /// named graph's. There is no all-graphs form; a reader names each graph. + /// + /// # Errors + /// [`crate::Error::Sqlite`] if the read fails. + pub fn events_after_in( + &self, + since: i64, + limit: usize, + types: Option<&[String]>, + group: Option<&str>, + graph: Option<&str>, + ) -> Result> { + // ROOT reads `events` exactly as before named graphs could emit; one + // graph reads its rows of `graph_events`. A read-only handle on a store + // no writer has migrated has no `graph_events`: no graph has events. let mut bind: Vec> = vec![Box::new(since)]; + let mut sql = match graph { + None => String::from( + "SELECT \"offset\", type, ts, subject, group_id, tx_id, payload, NULL \ + FROM events WHERE \"offset\" > ?1", + ), + Some(g) => { + if !self.has_graph_events()? { + return Ok(Vec::new()); + } + bind.push(Box::new(g.to_string())); + String::from( + "SELECT \"offset\", type, ts, subject, group_id, tx_id, payload, graph \ + FROM graph_events WHERE \"offset\" > ?1 AND graph = ?2", + ) + } + }; if let Some(ts) = types && !ts.is_empty() { @@ -541,6 +605,7 @@ impl Store { group_id: r.get(4)?, tx_id: r.get(5)?, payload: r.get(6)?, + graph: r.get(7)?, }) })?; let mut out = Vec::new(); @@ -592,3 +657,7 @@ impl Store { #[cfg(test)] #[path = "events_tests.rs"] mod tests; + +#[cfg(test)] +#[path = "events_rollback_tests.rs"] +mod rollback_tests; diff --git a/src/store/events_rollback_tests.rs b/src/store/events_rollback_tests.rs new file mode 100644 index 000000000..cb0f88e4f --- /dev/null +++ b/src/store/events_rollback_tests.rs @@ -0,0 +1,176 @@ +//! The rollback acceptance pair (event-push design §7): a store written by +//! this code, read by the code that preceded named-graph events. +//! +//! The pre-change reader, resume and pruner are pinned below as the literal +//! SQL they ran, so this test keeps describing what an older binary does +//! even after the current functions change. + +use super::*; +use crate::types::Op; +use rusqlite::OptionalExtension; + +const BOARD: &str = "https://example.org/graph/board"; +const TS: &str = "2026-10-24T00:00:00Z"; +const LATER: &str = "2026-12-01T00:00:00Z"; + +/// `Store::events_after` before named graphs could emit, with no filter. +const PRE_CHANGE_READ: &str = "SELECT \"offset\", type, ts, subject, group_id, tx_id, payload \ + FROM events WHERE \"offset\" > ?1 ORDER BY \"offset\" LIMIT ?2"; +/// `Store::consumer_committed` before named graphs could emit. +const PRE_CHANGE_RESUME: &str = "SELECT committed_offset FROM consumers WHERE consumer_id = ?1"; +/// `Store::prune_events` before named graphs could emit. +const PRE_CHANGE_PRUNE_SAFE: &str = "SELECT COALESCE(MIN(committed_offset), ?1) FROM consumers"; +const PRE_CHANGE_PRUNE: &str = "DELETE FROM events WHERE ts < ?1 AND \"offset\" <= ?2"; + +fn datum(store: &Store, s: &str) -> Datum { + Datum { + entity: store.intern(s).unwrap(), + attribute: store.intern("urn:p:x").unwrap(), + value: Value::Ref(store.intern("urn:o:1").unwrap()), + valid_from: TS.into(), + valid_to: None, + op: Op::Assert, + } +} + +/// ROOT and board writes, with a ROOT consumer and a board consumer each +/// committed partway. +fn written() -> (Store, i64, i64) { + let mut store = Store::open_in_memory().unwrap(); + store + .transact(&[datum(&store, "urn:item:r1")], TS, None, None) + .unwrap(); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + let g = store.committed_graph_id(BOARD).unwrap().unwrap(); + store + .transact_to_graph(&[datum(&store, "urn:item:b1")], TS, None, None, g) + .unwrap(); + let root_cursor = store.latest_event_offset().unwrap(); + let board_cursor = store.latest_event_offset_in(Some(BOARD)).unwrap(); + store + .commit_consumer("root-reader", root_cursor, TS) + .unwrap(); + store + .commit_consumer_in("board-reader", Some(BOARD), board_cursor, TS) + .unwrap(); + // Writes after both cursors: what each reader resumes into. + store + .transact(&[datum(&store, "urn:item:r2")], TS, None, None) + .unwrap(); + store + .transact_to_graph(&[datum(&store, "urn:item:b2")], TS, None, None, g) + .unwrap(); + (store, root_cursor, board_cursor) +} + +fn pre_change_read(conn: &rusqlite::Connection, table: &str, since: i64) -> Vec<(i64, String)> { + let sql = PRE_CHANGE_READ.replace("FROM events", &format!("FROM {table}")); + let mut st = conn.prepare(&sql).unwrap(); + st.query_map(params![since, 10_000], |r| { + Ok(( + r.get(0)?, + r.get::<_, Option>(3)?.unwrap_or_default(), + )) + }) + .unwrap() + .collect::>() + .unwrap() +} + +#[test] +fn a_pre_change_reader_sees_exactly_the_root_log() { + let (store, root_cursor, _) = written(); + let current: Vec<(i64, String)> = store + .events_after(0, 10_000, None, None) + .unwrap() + .into_iter() + .map(|e| (e.offset, e.subject.unwrap_or_default())) + .collect(); + let old = pre_change_read(&store.conn, "events", 0); + assert_eq!( + old, current, + "the old reader and the new ROOT reader agree, offsets included" + ); + assert!( + old.iter().all(|(_, s)| !s.contains("urn:item:b")), + "no board event: {old:?}" + ); + assert!(old.iter().any(|(_, s)| s.contains("urn:item:r2"))); + // The old resume finds the ROOT reader where it committed, and the + // board reader's cursor is not in its table at all. + let resume = |id: &str| -> Option { + store + .conn + .query_row(PRE_CHANGE_RESUME, params![id], |r| r.get(0)) + .optional() + .unwrap() + }; + assert_eq!(resume("root-reader"), Some(root_cursor)); + assert_eq!(resume("board-reader"), None); + let after = pre_change_read(&store.conn, "events", root_cursor); + assert!(after.iter().all(|(o, _)| *o > root_cursor) && !after.is_empty()); +} + +#[test] +fn a_pre_change_prune_never_touches_the_board_log_and_the_board_resumes() { + let (store, _, board_cursor) = written(); + let board_before = store + .events_after_in(0, 10_000, None, None, Some(BOARD)) + .unwrap(); + let safe: i64 = store + .conn + .query_row(PRE_CHANGE_PRUNE_SAFE, params![i64::MAX], |r| r.get(0)) + .unwrap(); + store + .conn + .execute(PRE_CHANGE_PRUNE, params![LATER, safe]) + .unwrap(); + let board_after = store + .events_after_in(0, 10_000, None, None, Some(BOARD)) + .unwrap(); + assert_eq!( + board_before.iter().map(|e| e.offset).collect::>(), + board_after.iter().map(|e| e.offset).collect::>(), + "an older pruner leaves every board row" + ); + // Re-upgraded, the board reader resumes from its own cursor into only + // its graph's events. + let resume = store + .consumer_committed_in("board-reader", Some(BOARD)) + .unwrap(); + assert_eq!(resume, board_cursor); + let next = store + .events_after_in(resume, 10_000, None, None, Some(BOARD)) + .unwrap(); + assert!(!next.is_empty()); + assert!(next.iter().all(|e| { + e.graph.as_deref() == Some(BOARD) + && e.subject + .as_deref() + .is_some_and(|s| s.contains("urn:item:b2")) + })); +} + +/// The control: the rejected design, a `graph` column on `events`, leaks a +/// board event to the same pre-change reader. Without this, the test above +/// could pass on a reader that never saw any named-graph row at all. +#[test] +fn control_a_graph_column_on_events_would_leak_to_the_old_reader() { + let store = Store::open_in_memory().unwrap(); + store + .conn + .execute_batch( + "CREATE TABLE events_col (\"offset\" INTEGER PRIMARY KEY AUTOINCREMENT, type TEXT NOT NULL, \ + ts TEXT NOT NULL, subject TEXT, group_id TEXT, tx_id INTEGER NOT NULL, \ + payload TEXT NOT NULL DEFAULT '{}', graph TEXT); + INSERT INTO events_col (type, ts, subject, tx_id, graph) VALUES ('entity.added', 't', 'urn:item:r1', 1, NULL); + INSERT INTO events_col (type, ts, subject, tx_id, graph) VALUES ('entity.added', 't', 'urn:item:b1', 2, 'https://example.org/graph/board');", + ) + .unwrap(); + let old = pre_change_read(&store.conn, "events_col", 0); + assert_eq!( + old.len(), + 2, + "the old reader has no graph predicate: {old:?}" + ); +} diff --git a/src/store/events_tests.rs b/src/store/events_tests.rs index abdd511e8..3a952c7bf 100644 --- a/src/store/events_tests.rs +++ b/src/store/events_tests.rs @@ -1072,3 +1072,207 @@ fn write_provenance_counts_committed_request_writes_only() { assert_eq!(provenance_series(CLIENT, "/knot", "complete"), 1); assert_eq!(provenance_series(CLIENT, "/knot", "absent"), 1); } + +// -- Named-graph events (event-push design §4) -- + +const BOARD: &str = "https://example.org/graph/board"; +const OTHER: &str = "https://example.org/graph/other"; +const TS: &str = "2026-10-24T00:00:00Z"; + +fn ref_datum(store: &Store, s: &str, p: &str, o: &str) -> Datum { + Datum { + entity: store.intern(s).unwrap(), + attribute: store.intern(p).unwrap(), + value: Value::Ref(store.intern(o).unwrap()), + valid_from: TS.to_string(), + valid_to: None, + op: crate::types::Op::Assert, + } +} + +fn graph_events(store: &Store, graph: &str) -> Vec { + store + .events_after_in(0, 10_000, None, None, Some(graph)) + .unwrap() +} + +#[test] +fn a_committed_named_graph_emits_entity_events_tagged_with_its_iri() { + let mut store = Store::open_in_memory().unwrap(); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + let g = store.committed_graph_id(BOARD).unwrap().unwrap(); + let root_before = all_events(&store).len(); + + let first = [ref_datum( + &store, + "urn:item:1", + "urn:p:status", + "urn:s:open", + )]; + store.transact_to_graph(&first, TS, None, None, g).unwrap(); + let second = [ref_datum( + &store, + "urn:item:1", + "urn:p:status", + "urn:s:closed", + )]; + store.transact_to_graph(&second, TS, None, None, g).unwrap(); + + let evs = graph_events(&store, BOARD); + let kinds: Vec<&str> = evs.iter().map(|e| e.event_type.as_str()).collect(); + assert_eq!(kinds, ["entity.added", "entity.updated"]); + assert!(evs.iter().all(|e| e.graph.as_deref() == Some(BOARD))); + assert!( + evs.iter() + .all(|e| e.subject.as_deref() == Some("urn:item:1")) + ); + // Entity level only: a Ref assertion in a named graph is not an edge event. + assert!(!kinds.iter().any(|k| k.starts_with("edge."))); + // The untagged reader sees exactly the ROOT log it saw before. + assert_eq!(all_events(&store).len(), root_before); + assert!(all_events(&store).iter().all(|e| e.graph.is_none())); +} + +#[test] +fn added_is_judged_within_the_graph_not_the_store() { + let mut store = Store::open_in_memory().unwrap(); + store + .transact( + &[ref_datum(&store, "urn:item:1", "urn:p:x", "urn:o:1")], + TS, + None, + None, + ) + .unwrap(); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + let g = store.committed_graph_id(BOARD).unwrap().unwrap(); + store + .transact_to_graph( + &[ref_datum( + &store, + "urn:item:1", + "urn:p:status", + "urn:s:open", + )], + TS, + None, + None, + g, + ) + .unwrap(); + let kinds: Vec = graph_events(&store, BOARD) + .into_iter() + .map(|e| e.event_type) + .collect(); + assert_eq!(kinds, ["entity.added"]); +} + +#[test] +fn overlays_and_unregistered_graphs_stay_silent() { + let mut store = Store::open_in_memory().unwrap(); + let overlay = store + .overlay_create("https://example.org/overlay/o", 0) + .unwrap(); + let unregistered = store + .intern("https://example.org/graph/unregistered") + .unwrap(); + let before = store.latest_event_offset().unwrap(); + for g in [overlay, unregistered] { + let d = [ref_datum( + &store, + "urn:item:2", + "urn:p:status", + "urn:s:open", + )]; + // An unregistered target may itself be refused; either way no event. + let _ = store.transact_to_graph(&d, TS, None, None, g); + } + assert_eq!(store.latest_event_offset().unwrap(), before); + assert!(graph_events(&store, "https://example.org/overlay/o").is_empty()); +} + +#[test] +fn one_graphs_reader_never_sees_another_graphs_events() { + let mut store = Store::open_in_memory().unwrap(); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + store.graph_create_recorded(OTHER, TS, None).unwrap(); + for (iri, item) in [(BOARD, "urn:item:b"), (OTHER, "urn:item:o")] { + let g = store.committed_graph_id(iri).unwrap().unwrap(); + let d = [ref_datum(&store, item, "urn:p:status", "urn:s:open")]; + store.transact_to_graph(&d, TS, None, None, g).unwrap(); + } + let board = graph_events(&store, BOARD); + let other = graph_events(&store, OTHER); + assert_eq!(board.len(), 1); + assert_eq!(other.len(), 1); + assert_eq!(board[0].subject.as_deref(), Some("urn:item:b")); + assert_eq!(other[0].subject.as_deref(), Some("urn:item:o")); + // Filters combine by AND. + let typed = store + .events_after_in(0, 100, Some(&["entity.updated".into()]), None, Some(BOARD)) + .unwrap(); + assert!(typed.is_empty()); +} + +#[test] +fn only_named_graph_events_carry_the_graph_key_on_the_wire() { + let mut store = Store::open_in_memory().unwrap(); + store + .transact( + &[ref_datum(&store, "urn:item:r", "urn:p:x", "urn:o:1")], + TS, + None, + None, + ) + .unwrap(); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + let g = store.committed_graph_id(BOARD).unwrap().unwrap(); + store + .transact_to_graph( + &[ref_datum(&store, "urn:item:b", "urn:p:x", "urn:o:1")], + TS, + None, + None, + g, + ) + .unwrap(); + let root = all_events(&store); + assert!(!root.is_empty()); + assert!(root.iter().all(|e| e.to_json().get("graph").is_none())); + let named = graph_events(&store, BOARD); + assert_eq!(named[0].to_json()["graph"], BOARD); +} + +#[test] +fn lag_is_measured_within_the_readers_graph() { + let mut store = Store::open_in_memory().unwrap(); + store + .transact( + &[ref_datum(&store, "urn:item:r", "urn:p:x", "urn:o:1")], + TS, + None, + None, + ) + .unwrap(); + let root_head = store.latest_event_offset_in(None).unwrap(); + assert!(root_head > 0); + store.graph_create_recorded(BOARD, TS, None).unwrap(); + let root_head = store.latest_event_offset_in(None).unwrap(); // graph.created is ROOT + let g = store.committed_graph_id(BOARD).unwrap().unwrap(); + store + .transact_to_graph( + &[ref_datum(&store, "urn:item:b", "urn:p:x", "urn:o:1")], + TS, + None, + None, + g, + ) + .unwrap(); + // A ROOT reader caught up before the board write is still caught up. + assert_eq!(store.latest_event_offset_in(None).unwrap(), root_head); + // ROOT's log did not move at all: the board's events are their own log. + assert_eq!(store.latest_event_offset().unwrap(), root_head); + let board_head = store.latest_event_offset_in(Some(BOARD)).unwrap(); + assert!(board_head > 0); + assert_eq!(store.latest_event_offset_in(Some(OTHER)).unwrap(), 0); +} diff --git a/src/store/graph_events.rs b/src/store/graph_events.rs new file mode 100644 index 000000000..9cae3690e --- /dev/null +++ b/src/store/graph_events.rs @@ -0,0 +1,153 @@ +//! Named-graph semantic events (event-push design §4): their own log, +//! `graph_events`, and their readers' cursors, `graph_consumers`. Kept apart +//! from `events` so that a store written here reads the same to any older +//! binary (see `migrate_graph_events`). + +use std::collections::HashSet; + +use rusqlite::{OptionalExtension, params}; +use serde_json::json; + +use super::{Datum, Store}; +use crate::error::Result; + +impl Store { + /// Entity-level events for a commit to a registered COMMITTED named graph; + /// a write to an overlay or an unregistered graph emits nothing. Same savepoint, + /// same atomicity as the ROOT events. + pub(super) fn emit_named_graph_events( + &self, + asserts: &[&Datum], + retracts: &[&Datum], + tx_id: i64, + timestamp: &str, + source: Option<&str>, + graph: i64, + ) -> Result<()> { + let graph_iri = self.resolve(graph)?; + if self.committed_graph_id(&graph_iri)? != Some(graph) { + return Ok(()); + } + let insert_event = + |event_type: &str, subject: Option<&str>, payload: &serde_json::Value| { + self.conn.execute( + "INSERT INTO graph_events (type, ts, subject, group_id, tx_id, payload, graph) \ + VALUES (?1, ?2, ?3, NULL, ?4, ?5, ?6)", + params![ + event_type, + timestamp, + subject, + tx_id, + payload.to_string(), + graph_iri + ], + ) + }; + if let Some(name) = source.and_then(|s| s.strip_prefix("episode:")) { + let payload = json!({ "name": name, "facts": asserts.len() + retracts.len() }); + insert_event("episode.ingested", None, &payload)?; + } + // "Added" is judged within THIS graph: an item's first facts in the + // board are an addition to the board even if ROOT already knew the IRI. + let mut seen: HashSet = HashSet::new(); + for d in asserts.iter().chain(retracts.iter()) { + if !seen.insert(d.entity) { + continue; + } + let preexisting = self + .conn + .query_row( + "SELECT 1 FROM facts WHERE g = ?1 AND e = ?2 AND tx < ?3 LIMIT 1", + params![graph, d.entity, tx_id], + |_| Ok(true), + ) + .optional()? + .is_some(); + let event_type = if preexisting { + "entity.updated" + } else { + "entity.added" + }; + insert_event(event_type, Some(&self.resolve(d.entity)?), &json!({}))?; + } + Ok(()) + } + + /// Highest offset in ONE graph's log (`None` = ROOT's `events`), 0 when + /// empty. A reader's lag is measured against its own log, so a ROOT + /// consumer's lag never moves with board writes it does not receive. + /// + /// # Errors + /// [`crate::Error::Sqlite`] if the read fails. + pub fn latest_event_offset_in(&self, graph: Option<&str>) -> Result { + let Some(g) = graph else { + return self.latest_event_offset(); + }; + if !self.has_graph_events()? { + return Ok(0); + } + Ok(self.conn.query_row( + "SELECT COALESCE(MAX(\"offset\"), 0) FROM graph_events WHERE graph = ?1", + params![g], + |r| r.get(0), + )?) + } + + /// Whether this store has the named-graph event log. A read-only handle + /// on a store no writer has migrated yet does not. + pub(super) fn has_graph_events(&self) -> Result { + Ok(self + .conn + .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'graph_events'")? + .exists([])?) + } + + /// [`Self::consumer_committed`] in one graph's log (`None` = ROOT). A + /// graph consumer's cursor is an offset in `graph_events`, so it lives in + /// `graph_consumers`, never in `consumers` beside ROOT offsets. + /// + /// # Errors + /// [`crate::Error::Sqlite`] if the read fails. + pub fn consumer_committed_in(&self, consumer_id: &str, graph: Option<&str>) -> Result { + let Some(g) = graph else { + return self.consumer_committed(consumer_id); + }; + if !self.has_graph_events()? { + return Ok(0); + } + Ok(self + .conn + .query_row( + "SELECT committed_offset FROM graph_consumers WHERE consumer_id = ?1 AND graph = ?2", + params![consumer_id, g], + |r| r.get(0), + ) + .optional()? + .unwrap_or(0)) + } + + /// [`Self::commit_consumer`] in one graph's log (`None` = ROOT). + /// + /// # Errors + /// [`crate::Error::Sqlite`] if the write fails. + pub fn commit_consumer_in( + &self, + consumer_id: &str, + graph: Option<&str>, + offset: i64, + now: &str, + ) -> Result<()> { + let Some(g) = graph else { + return self.commit_consumer(consumer_id, offset, now); + }; + self.conn.execute( + "INSERT INTO graph_consumers (consumer_id, graph, committed_offset, updated_at) \ + VALUES (?1, ?2, ?3, ?4) \ + ON CONFLICT(consumer_id, graph) DO UPDATE SET \ + committed_offset = excluded.committed_offset, \ + updated_at = excluded.updated_at", + params![consumer_id, g, offset, now], + )?; + Ok(()) + } +} diff --git a/src/store/migrate.rs b/src/store/migrate.rs index 8879b6cae..36bed94f5 100644 --- a/src/store/migrate.rs +++ b/src/store/migrate.rs @@ -263,6 +263,43 @@ impl Store { Ok(()) } + /// Named-graph events and their consumers' cursors (event-push design §4). + /// + /// SEPARATE TABLES, deliberately, not a `graph` column on `events`. quipu + /// opens a store whatever its schema, so an older binary reads a newer + /// store, and its `events_after` has no graph predicate: a column would + /// hand board events to every ROOT consumer after a rollback. Older code + /// never reads these tables, so `events`, `consumers` and their pruning + /// stay exactly what every older binary expects, and the graph rows and + /// cursors survive a revert for re-upgrade. + /// + /// `graph` is the IRI as TEXT, never a term id (the #74 respace rule, like + /// `subject`). AUTOINCREMENT for the same never-reused-offset guarantee as + /// `events`; the offsets are their own sequence. + pub(super) fn migrate_graph_events(conn: &Connection) -> Result<()> { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS graph_events ( + \"offset\" INTEGER PRIMARY KEY AUTOINCREMENT, + graph TEXT NOT NULL, + type TEXT NOT NULL, + ts TEXT NOT NULL, + subject TEXT, + group_id TEXT, + tx_id INTEGER NOT NULL, + payload TEXT NOT NULL DEFAULT '{}' + ); + CREATE INDEX IF NOT EXISTS idx_graph_events_graph ON graph_events(graph, \"offset\"); + CREATE TABLE IF NOT EXISTS graph_consumers ( + consumer_id TEXT NOT NULL, + graph TEXT NOT NULL, + committed_offset INTEGER NOT NULL DEFAULT 0, + updated_at TEXT, + PRIMARY KEY (consumer_id, graph) + );", + )?; + Ok(()) + } + /// Stored named-query registry (quipu #79). /// /// Versioned in #71's close-don't-overwrite style **from day one** rather diff --git a/src/store/mod.rs b/src/store/mod.rs index b37689652..0e11031bd 100644 --- a/src/store/mod.rs +++ b/src/store/mod.rs @@ -30,6 +30,7 @@ pub mod forks; pub mod freeze; mod freeze_io; mod gate; +mod graph_events; mod identity; pub mod import; pub mod inferred; diff --git a/src/store/open.rs b/src/store/open.rs index 6367a1840..fa6765512 100644 --- a/src/store/open.rs +++ b/src/store/open.rs @@ -296,6 +296,7 @@ impl Store { Self::migrate_query_registry(&conn)?; Self::migrate_pending_share_queries(&conn)?; Self::migrate_retraction_tx(&conn)?; + Self::migrate_graph_events(&conn)?; // AFTER migrate_named_graphs: the fork registry references graphs(g). Self::migrate_forks(&conn)?; Self::migrate_frozen_packs(&conn)?; diff --git a/src/store/overlays.rs b/src/store/overlays.rs index 5f0f2aa6c..062584120 100644 --- a/src/store/overlays.rs +++ b/src/store/overlays.rs @@ -52,6 +52,25 @@ impl Store { Ok(g) } + /// The graph id of a registered COMMITTED named graph, or `None` for an + /// unregistered IRI and for an overlay (compose-only staging). + /// + /// # Errors + /// [`crate::Error::Sqlite`] if the lookup fails. + pub fn committed_graph_id(&self, graph_iri: &str) -> Result> { + let Some(g) = self.lookup(graph_iri)? else { + return Ok(None); + }; + Ok(self + .conn + .query_row( + "SELECT g FROM graphs WHERE g = ?1 AND class = 'committed'", + params![g], + |r| r.get(0), + ) + .optional()?) + } + /// The read-only half of [`Self::graph_create`]: `true` when `g` is already a /// committed graph, `false` when it is unregistered, and the same refusal /// when it is registered under any other class. A promotion dry run calls diff --git a/src/store/respace_map.rs b/src/store/respace_map.rs index 59817420a..c4ce1d960 100644 --- a/src/store/respace_map.rs +++ b/src/store/respace_map.rs @@ -232,6 +232,20 @@ pub const COLUMN_CLASSIFICATION: &[(&str, &str, TermIdKind)] = &[ ("pending_share_queries", "staged_at", TermIdKind::None), // -- Store::migrate_retraction_tx (quipu #83) -- ("facts", "retracted_tx", TermIdKind::None), + // -- Store::migrate_graph_events (event-push design §4) -- + // `graph` is the IRI as TEXT, like `events.subject`; never a term id. + ("graph_events", "offset", TermIdKind::None), + ("graph_events", "graph", TermIdKind::None), + ("graph_events", "type", TermIdKind::None), + ("graph_events", "ts", TermIdKind::None), + ("graph_events", "subject", TermIdKind::None), + ("graph_events", "group_id", TermIdKind::None), + ("graph_events", "tx_id", TermIdKind::None), + ("graph_events", "payload", TermIdKind::None), + ("graph_consumers", "consumer_id", TermIdKind::None), + ("graph_consumers", "graph", TermIdKind::None), + ("graph_consumers", "committed_offset", TermIdKind::None), + ("graph_consumers", "updated_at", TermIdKind::None), // -- Store::migrate_forks (quipu-gp5) -- // `g` is the fork graph's term id; `parent_branch` matches // `graphs.parent_branch` (v1 only writes ROOT `0`, the exempt sentinel). diff --git a/src/write_kind.rs b/src/write_kind.rs index 6ccbc8fe7..e4d47c075 100644 --- a/src/write_kind.rs +++ b/src/write_kind.rs @@ -102,8 +102,16 @@ impl WriteKind { "/retract" | "/retract/source" => Self::Retract, "/propose" | "/proposal/accept" | "/proposal/reject" => Self::Proposal, "/overlay/write" | "/overlay/create" => Self::Overlay, - "/graph/create" | "/graph/label" | "/graph/freeze" | "/graph/thaw" | "/datasets" - | "/queries" | "/shapes" | "/subscriptions" | "/events/commit" => Self::GraphAdmin, + "/graph/create" + | "/graph/label" + | "/graph/freeze" + | "/graph/thaw" + | "/datasets" + | "/queries" + | "/shapes" + | "/subscriptions" + | "/events/commit" + | "/events/graph/commit" => Self::GraphAdmin, "/project" | "/impact" | "/embed_backfill" diff --git a/tests/tool_schema_fields.rs b/tests/tool_schema_fields.rs index bd46d6853..0ce992272 100644 --- a/tests/tool_schema_fields.rs +++ b/tests/tool_schema_fields.rs @@ -165,6 +165,18 @@ fn audit_schema_fields() { if name == "quipu_update" { read.extend(["update", "actor", "source"].map(String::from)); } + for refused in parity::refused_fields(name) { + // A refused field must stay undeclared and must really be read + // (to be refused); otherwise the exemption is stale. + assert!( + !declared.contains(*refused), + "{name} declares refused {refused}" + ); + assert!( + read.remove(*refused), + "{name} no longer reads refused {refused}" + ); + } let missing: Vec<_> = read.difference(&declared).collect(); if !missing.is_empty() { gaps.insert(name.to_string(), serde_json::json!(missing)); diff --git a/tests/tool_schema_fields/parity.rs b/tests/tool_schema_fields/parity.rs index c72da6e21..f1e4dd225 100644 --- a/tests/tool_schema_fields/parity.rs +++ b/tests/tool_schema_fields/parity.rs @@ -42,6 +42,11 @@ pub(super) fn handler(name: &str) -> Option<(&'static str, Option<(&'static str, "quipu_reason" => ("reason", None), "quipu_update" => ("update_post", None), "quipu_events_commit" => ("events_commit", None), + "quipu_graph_events" => ( + "events_graph_get", + Some(("src/server/feed.rs", "EventParams")), + ), + "quipu_graph_events_commit" => ("events_graph_commit", None), "quipu_entity_history" => ("entity_history", None), "quipu_embed_backfill" => ("embed_backfill", None), "quipu_embed_backfill_graph" => ("embed_backfill_graph", None), @@ -53,6 +58,19 @@ pub(super) fn handler(name: &str) -> Option<(&'static str, Option<(&'static str, }) } +/// Fields a handler reads ONLY to refuse them, so its tool must NOT declare +/// them. `/events` and `/events/commit` refuse `graph`: a server that predates +/// named-graph events would ignore it (serving ROOT's events, or upserting the +/// ROOT cursor before answering), so named graphs have their own routes and +/// tools (`quipu_graph_events`, `quipu_graph_events_commit`). Declaring `graph` on +/// the ROOT tools would advertise exactly the form that is refused. +pub(super) fn refused_fields(name: &str) -> &'static [&'static str] { + match name { + "quipu_events" | "quipu_events_commit" => &["graph"], + _ => &[], + } +} + pub(super) fn wire_fields(file: &str, name: &str) -> BTreeSet { let source = std::fs::read_to_string(Path::new(env!("CARGO_MANIFEST_DIR")).join(file)).unwrap(); let ast = syn::parse_file(&source).unwrap();