From 247618227cc0c7237a51355c172f719ff081d2fa Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 01:00:33 -0700 Subject: [PATCH 1/7] docs: add WAL truth boundary topic --- docs/index.md | 2 + docs/topics/WAL.md | 131 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 133 insertions(+) create mode 100644 docs/topics/WAL.md diff --git a/docs/index.md b/docs/index.md index 94815f83..8fd65d5c 100644 --- a/docs/index.md +++ b/docs/index.md @@ -15,6 +15,7 @@ Echo's live documentation centers on the runtime carrier, the retained witnesses - Echo v0.1.0 release plan: [/design/v0.1.0-release-plan](/design/v0.1.0-release-plan) - Echo v0.1.0 jedit release gate: [/design/v0.1.0-jedit-release-gate](/design/v0.1.0-jedit-release-gate) - Trusted runtime control history: [/design/trusted-runtime-control-history](/design/trusted-runtime-control-history) +- WAL truth boundary: [/topics/WAL](/topics/WAL) - jedit next ten release-gate slices: [/design/v0.1.0-jedit-next-ten-slices](/design/v0.1.0-jedit-next-ten-slices) - Theory map: [/theory/THEORY](/theory/THEORY) - Current bearing: [/BEARING](/BEARING) @@ -33,6 +34,7 @@ Echo's live documentation centers on the runtime carrier, the retained witnesses - DPO concurrency litmus: [/spec/SPEC-0003-dpo-concurrency-litmus-v0](/spec/SPEC-0003-dpo-concurrency-litmus-v0) - Rewrite scheduler: [/spec/scheduler-warp-core](/spec/scheduler-warp-core) - Canonical inbox sequencing: [/spec/canonical-inbox-sequencing](/spec/canonical-inbox-sequencing) +- WAL truth boundary: [/topics/WAL](/topics/WAL) - WARP tick patch: [/spec/warp-tick-patch](/spec/warp-tick-patch) - Merkle commit: [/spec/merkle-commit](/spec/merkle-commit) - Fixed timestep invariant: [/invariants/FIXED-TIMESTEP](/invariants/FIXED-TIMESTEP) diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md new file mode 100644 index 00000000..fb0b2ff7 --- /dev/null +++ b/docs/topics/WAL.md @@ -0,0 +1,131 @@ + + + +# WAL + +Echo's write-ahead log is the durable commit boundary for causal history. +Applications submit canonical intents. A trusted Echo host admits those intents, +records the accepted-submission fact, schedules lawful work, records decided +tick evidence, and exposes outcomes only after the relevant WAL fact is +committed. + +The short rule is: + +```text +Echo may only claim what its WAL can recover. +``` + +## What We Found + +The current runtime WAL evidence says four concrete things. + +First, accepted-submission evidence is not just an in-memory editor event. The +WAL-backed ACK path, `submit_intent_with_runtime_wal_ack(...)`, returns only +after the runtime WAL has committed the submission-acceptance transaction. That +acceptance fact recovers as an `AcceptedPending` submission if the scheduler has +not yet produced a decided outcome. + +Second, failed WAL acceptance is not allowed to leave half-visible runtime +state. If no runtime WAL is configured, the ACK path returns an explicit +unavailable error before mutating intake state. If WAL commit fails, Echo rolls +back the in-memory intake mutation before returning the error. + +Third, tick receipts are published only after scheduler-tick WAL evidence is +committed. Recovery can rebuild both the submission posture and the receipt +index from committed WAL facts. If tick WAL recording fails, Echo rolls back the +visible receipt/outcome instead of exposing a receipt that recovery cannot +rebuild. + +Fourth, read-only recovery rebuilds submission and receipt indexes from the WAL +without scheduler callbacks, application callbacks, wall-clock interpretation, +or external I/O. Recovery also emits a certificate over the committed replay +range and recovered index root. + +## Boundaries + +The WAL belongs to the trusted runtime host. Application-facing code can submit +intents and observe outcomes, but it cannot append WAL records, tick the +scheduler, install packages, stage ticketed ingress, or perform trusted +recovery. + +That split matters because an accepted edit is not durable because an editor +buffer says it is dirty, because a file was written to disk, or because a UI +event happened. It is durable because Echo recorded the accepted submission and +later recorded any decided receipt under host-owned WAL authority. + +## Recovery Postures + +The useful postures are: + +| Posture | Meaning | +| --- | --- | +| `not_accepted` | The intent never reached WAL-backed accepted submission posture. | +| `accepted_pending` | The intent was accepted and can be recovered, but no decided receipt was recovered. | +| `decided_applied` | A recovered receipt says the work applied under named law. | +| `decided_rejected` | A recovered receipt says the work was rejected, conflicted, or obstructed. | +| `recovery_faulted` | Required committed WAL evidence or retained material is missing or corrupt. | + +An app such as `jedit` maps these generic postures into product language +outside Echo. Echo should not grow editor, file, buffer, or dirty-state nouns in +order to explain them. + +## What This Means For Editors + +For Jim/jedit, "dirty" should not mean "at risk of being lost." It should mean +"not currently materialized to the host file projection." If an edit intent has +received WAL-backed accepted-submission evidence, the edit belongs to +recoverable causal history even before the host file is written. If that edit is +later decided, the receipt and recovered reading are the source of truth for +restoring the editor view. + +The correct safety gate is therefore not a classic dirty-buffer warning on quit +or file switch. The correct gate is whether the edit crossed the Echo +WAL-backed ACK boundary and whether recovery can later classify it as pending, +decided, rejected, or obstructed. If that boundary is unavailable, the editor +must say so honestly instead of pretending a local buffer is causal history. + +## WAL, Graph Facts, And WSC + +WAL bytes are the durable commit authority. WARP graph facts can track WAL +segment evidence, and WSC can serialize graph facts plus WAL references or +bundled WAL material, but neither graph facts nor WSC replace the WAL commit +boundary. + +A WAL path is a storage locator, not causal identity. Causal identity comes from +the writer epoch, LSN range, segment digest, commit digest chain, and validated +commit anchors. + +## Evidence + +The runtime ACK and recovery witnesses live in +`crates/warp-core/tests/trusted_runtime_host_loop_tests.rs`. + +The core test names to read first are: + +- `runtime_wal_ack_submit_commits_acceptance_before_returning_handle` +- `runtime_wal_ack_path_requires_configured_runtime_wal` +- `runtime_wal_ack_failure_rolls_back_intake_mutation` +- `runtime_wal_ack_tick_commits_receipt_transaction_before_outcome_is_observed` +- `runtime_wal_ack_tick_failure_rolls_back_visible_outcome` +- `runtime_wal_ack_recover_read_only_rebuilds_submission_and_receipt_indexes` +- `runtime_wal_ack_recover_read_only_exposes_recovery_certificate` + +The host surface lives in `crates/warp-core/src/trusted_runtime_host.rs`, +especially `TrustedRuntimeHost`, `TrustedRuntimeApp`, `TrustedRuntimeWal`, +`submit_intent_with_runtime_wal_ack(...)`, and `recover_read_only()`. + +The doctrine and remaining release-hardening plan live in: + +- `docs/BEARING.md` +- `docs/design/causal-wal-hardening-matrix.md` +- `docs/design/work-item-sequencing-and-prioritization.md` +- `docs/WorkItems.md` + +## Current Caveat + +The trusted-runtime host tests use an in-memory runtime WAL adapter to prove ACK +ordering, rollback behavior, receipt publication ordering, and recovery index +shape. That is not the same claim as strict filesystem durability. Filesystem +WAL hardening, WSC export/import shape, retained material availability, and +release-grade recovery gates remain the place to prove crash and portability +claims beyond the current ACK boundary witnesses. From c1ec6b9db6d087ea30f0c0b96456bfb5b6037447 Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 01:35:34 -0700 Subject: [PATCH 2/7] docs: rename docs map to README --- AGENTS.md | 2 +- GUIDE.md | 2 +- README.md | 4 ++-- docs/{index.md => README.md} | 0 docs/procedures/DOCS-LIFECYCLE.md | 8 ++++---- xtask/src/main.rs | 6 +++--- 6 files changed, 11 insertions(+), 11 deletions(-) rename docs/{index.md => README.md} (100%) diff --git a/AGENTS.md b/AGENTS.md index 1fa314ee..1655c97d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -58,7 +58,7 @@ Do not audit the repository by recursively walking the filesystem. Follow the au - **`README.md`**: Public front door, core value prop, and quick tour. - **`GUIDE.md`**: Orientation and productive-fast path. -- **`docs/index.md`**: VitePress documentation map. +- **`docs/README.md`**: VitePress documentation map. ### 2. The Bedrock diff --git a/GUIDE.md b/GUIDE.md index fb57711e..3f9da7fb 100644 --- a/GUIDE.md +++ b/GUIDE.md @@ -54,7 +54,7 @@ Echo is a tiered engine. You choose your depth based on the task: ## Rule of Thumb -If you need a comprehensive spec, use the [docs/index.md](./docs/index.md) map. +If you need a comprehensive spec, use the [docs/README.md](./docs/README.md) map. If you need to know "what's true right now," use [docs/BEARING.md](./docs/BEARING.md). diff --git a/README.md b/README.md index c62d4b48..2672a281 100644 --- a/README.md +++ b/README.md @@ -11,7 +11,7 @@

- Docs + Docs · Architecture · @@ -432,7 +432,7 @@ cargo bench -p warp-benches [There Is No Graph](docs/architecture/there-is-no-graph.md) - Core runtime details: [warp-core spec](docs/spec/warp-core.md) - Causal transport: [Continuum Transport](docs/architecture/continuum-transport.md) -- Documentation map: [Docs](docs/index.md) +- Documentation map: [Docs](docs/README.md) ## Current Reality diff --git a/docs/index.md b/docs/README.md similarity index 100% rename from docs/index.md rename to docs/README.md diff --git a/docs/procedures/DOCS-LIFECYCLE.md b/docs/procedures/DOCS-LIFECYCLE.md index 7e88c8a2..ffe2b13f 100644 --- a/docs/procedures/DOCS-LIFECYCLE.md +++ b/docs/procedures/DOCS-LIFECYCLE.md @@ -146,13 +146,13 @@ Archive moves must preserve findability: - leave a short tombstone or status header when the old path has live inbound links; -- update `docs/index.md` if the old doc was listed there; +- update `docs/README.md` if the old doc was listed there; - run dead-reference checks for touched docs; - prefer one archive move per domain slice instead of broad unrelated sweeps. ## Public Docs Map Rules -`docs/index.md` is not an inventory. It is the reader path. +`docs/README.md` is not an inventory. It is the reader path. Only list a design doc in the public docs map when it is active, currently load-bearing, and there is no canonical domain doc yet. Once absorbed, remove @@ -181,7 +181,7 @@ For any PR that creates, materially edits, or relies on a design doc: - [ ] If implementation landed, current behavior was absorbed into a canonical domain doc or a follow-up issue was opened. - [ ] Superseded docs point to their replacement. -- [ ] `docs/index.md` links only to current reader-path docs. +- [ ] `docs/README.md` links only to current reader-path docs. - [ ] Dead-reference checks pass for touched markdown. ## Anti-Patterns @@ -201,5 +201,5 @@ When in doubt, do the smallest lawful docs lifecycle slice: 2. Name the current canonical doc. 3. Absorb the current invariant into that doc. 4. Mark one or more design docs `absorbed` or `superseded`. -5. Update `docs/index.md` only if the reader path changed. +5. Update `docs/README.md` only if the reader path changed. 6. Run markdownlint and dead-reference checks. diff --git a/xtask/src/main.rs b/xtask/src/main.rs index c3a77726..daa095d5 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -6736,7 +6736,7 @@ mod tests { #[test] fn root_relative_link_resolves_against_docs_root() { - let source = Path::new("docs/index.md"); + let source = Path::new("docs/README.md"); let docs_root = Path::new("docs"); let candidates = build_candidates(source, "/guide/start-here.md", docs_root); assert!(candidates @@ -6746,7 +6746,7 @@ mod tests { #[test] fn extensionless_link_tries_md_and_html() { - let source = Path::new("docs/index.md"); + let source = Path::new("docs/README.md"); let docs_root = Path::new("docs"); let candidates = build_candidates(source, "guide/start-here", docs_root); assert!(candidates @@ -6759,7 +6759,7 @@ mod tests { #[test] fn public_asset_resolution() { - let source = Path::new("docs/index.md"); + let source = Path::new("docs/README.md"); let docs_root = Path::new("docs"); let candidates = build_candidates(source, "/example-public-asset.html", docs_root); assert!(candidates From f77b62de85dec46360072417796d64579107830e Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 02:32:57 -0700 Subject: [PATCH 3/7] docs: add topology intent recovery goalpost --- CHANGELOG.md | 4 + docs/README.md | 1 + ...st-06-topology-intents-and-wal-recovery.md | 119 ++++++++++++++++++ docs/design/braids-and-strands-roadmap.md | 21 +++- docs/topics/WAL.md | 19 +++ 5 files changed, 163 insertions(+), 1 deletion(-) create mode 100644 docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md diff --git a/CHANGELOG.md b/CHANGELOG.md index aaf76bb7..a33424fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -85,6 +85,10 @@ receipts whose subject digest does not match the retained support digest. Collapse-derived shells also reject records whose shell policy id diverges from the nested collapse policy id. +- The braids/strands hardening docs now define Goalpost 6 for topology intents + and WAL recovery, making strand forks, braid event logs, retained braid + shells, and replica suffix import explicit WAL/WSC hardening work rather than + process-local topology state. - `warp-core` now enforces the v1 single-writer-head strand invariant through both `Strand::new(...)` and `StrandRegistry::insert(...)`, and runtime strand forking constructs the registered relation through the same constructor diff --git a/docs/README.md b/docs/README.md index d03592c9..0b56246d 100644 --- a/docs/README.md +++ b/docs/README.md @@ -39,6 +39,7 @@ Echo's live documentation centers on the runtime carrier, the retained witnesses - Goalpost 3, historical membership and replay: [/design/braids-and-strands-hardening/goalpost-03-historical-membership-and-replay](/design/braids-and-strands-hardening/goalpost-03-historical-membership-and-replay) - Goalpost 4, witness receipts and sealed capabilities: [/design/braids-and-strands-hardening/goalpost-04-witness-receipts-and-sealed-capabilities](/design/braids-and-strands-hardening/goalpost-04-witness-receipts-and-sealed-capabilities) - Goalpost 5, named plurality laws: [/design/braids-and-strands-hardening/goalpost-05-named-plurality-laws](/design/braids-and-strands-hardening/goalpost-05-named-plurality-laws) +- Goalpost 6, topology intents and WAL recovery: [/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery](/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery) - DPO concurrency litmus: [/spec/SPEC-0003-dpo-concurrency-litmus-v0](/spec/SPEC-0003-dpo-concurrency-litmus-v0) - Rewrite scheduler: [/spec/scheduler-warp-core](/spec/scheduler-warp-core) - Canonical inbox sequencing: [/spec/canonical-inbox-sequencing](/spec/canonical-inbox-sequencing) diff --git a/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md b/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md new file mode 100644 index 00000000..69075557 --- /dev/null +++ b/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md @@ -0,0 +1,119 @@ + + + +# Goalpost 6: Topology Intents And WAL Recovery + +Status: planned. + +Tracking issue: [#604](https://github.com/flyingrobots/echo/issues/604) + +## Goal + +Make topology-changing operations admitted causal history. + +Strand forks, braid lifecycle events, braid settlements, retained braid shells, +and replica suffix import must cross the same intent, receipt, WAL, and retained +evidence boundary that tick receipts already use. They are not side-channel +runtime state, service calls, or Git-shaped branch metadata. + +The target shape is: + +```text +topology intent +-> WAL-backed accepted topology evidence +-> receipt or shell +-> recoverable topology index +-> replay/audit reading +``` + +## Doctrine + +AION Paper VII treats tick execution, braid lowering, and replica suffix import +as one recurring WARP optic shape: + +```text +Lower(F, P) -> (R, W, theta) +``` + +At tick scale, `P` is a rewrite bundle and `theta` is a tick receipt. At braid +scale, `P` is a strand braid and `theta` is a braid shell. At replica scale, +`P` is a transported remote suffix family and `theta` is an import shell. + +Echo already has typed strand construction, braid event logs, settlement +provenance entries, retained braid shells, and braid replay/audit optics. This +goalpost promotes the remaining topology operations to WAL-backed causal +history so recovery can rebuild them after restart. + +## Required Boundaries + +### Strand Forks + +Forking a strand creates ancestry. It must become an admitted topology intent +whose receipt binds: + +- strand identity; +- source worldline; +- fork tick; +- source commit and boundary hash; +- child worldline; +- writer heads; +- retention posture; +- issuer or session evidence. + +The fork ACK must not be returned until the accepted topology evidence is +committed to WAL, or Echo returns an explicit obstruction. + +### Braid Events + +Creating a braid, weaving a member, finalizing settlement, and collapsing a +plural braid are braid history events. Their event log must be recoverable from +durable evidence, not only from process-local folded state. + +The existing `Braid::apply(...)` transition checks remain valuable, but the +accepted event stream must have WAL/WSC recovery posture. + +### Braid Shell Retention + +`BraidShell` is the retained hologram for braid-scale lowering. Recovery must +be able to recover retained shell identity and material by digest. Re-appending +the same shell remains idempotent; divergent duplicate content remains an +obstruction. + +### Replica Suffix Import + +Network suffix exchange is not a second merge regime. Transport constructs a +comparable basis and re-expresses remote suffix claims as a weave. Echo then +lowers that weave through the same optic law and retains an import shell. + +Duplicate delivery is idempotent re-introduction of the same witnessed +transport object, not a new authored intent. + +## Slices + +| Slice | Issue | Work | +| ------ | ----- | ---- | +| GP6-S1 | [#605](https://github.com/flyingrobots/echo/issues/605) | WAL-backed strand fork and drop intent receipts | +| GP6-S2 | [#606](https://github.com/flyingrobots/echo/issues/606) | WAL/WSC-backed braid event logs and retained shells | +| GP6-S3 | [#607](https://github.com/flyingrobots/echo/issues/607) | Replica suffix import as a witnessed WARP optic intent | +| GP6-S4 | [#608](https://github.com/flyingrobots/echo/issues/608) | Recovery indexes for topology state and retained shells | + +## Acceptance Criteria + +- Strand topology changes are acknowledged only after WAL-backed accepted + topology evidence commits. +- Braid event logs recover from durable evidence. +- Retained braid shells recover by digest after restart. +- Replica suffix import has explicit authorship, basis, witness, idempotence, + and retention posture. +- Recovery rebuilds strand registry, child worldline topology, braid event logs, + retained shell indexes, and import idempotence state from WAL/WSC evidence. +- Missing or corrupt retained topology material returns typed obstruction + evidence instead of partial silent recovery. + +## Non-Goals + +- Do not add jedit, file, editor, buffer, or dirty-state nouns to Echo core. +- Do not make Git branches or worktrees topology authority. +- Do not treat WSC graph facts as WAL bootstrap authority. +- Do not collapse braid geometry, settlement, and admission into one ambiguous + merge operation. diff --git a/docs/design/braids-and-strands-roadmap.md b/docs/design/braids-and-strands-roadmap.md index d27e7065..e2553654 100644 --- a/docs/design/braids-and-strands-roadmap.md +++ b/docs/design/braids-and-strands-roadmap.md @@ -5,7 +5,7 @@ Status: active hardening roadmap slice. -Last updated: 2026-06-15. +Last updated: 2026-06-26. ## Purpose @@ -22,6 +22,13 @@ runtime, API, digest, privacy, or witness dependency. No law-bearing object is born by accident. +Goalposts 1-5 hardened construction, identity, membership history, witness +receipts, sealed capabilities, and named plurality laws. Goalpost 6 promotes +topology operations themselves to WAL-backed causal history: forking a strand, +weaving or settling a braid, retaining a braid shell, and importing a remote +suffix are accepted intents with recoverable evidence, not side-channel runtime +state. + The destination is: ```text @@ -124,6 +131,17 @@ Design: - [x] GP5-S5: Add obstruction evidence for unsupported or unauthorized law execution. +### Goalpost 6: Topology Intents And WAL Recovery + +Design: +[`goalpost-06-topology-intents-and-wal-recovery.md`](braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md) + +- [ ] GP6-S1: Add WAL-backed strand fork and drop intent receipts. +- [ ] GP6-S2: Persist braid event logs and retained braid shells through + WAL/WSC. +- [ ] GP6-S3: Model replica suffix import as a witnessed WARP optic intent. +- [ ] GP6-S4: Rebuild topology indexes from WAL/WSC recovery evidence. + ## North Star Echo must feel less like a crate full of important structs and more like a @@ -166,6 +184,7 @@ The vocabulary stays sharp: | Posture floor | Lowest causal posture the reading can honestly claim. | | Retained plurality | Preserved multiple claims not collapsed to one fact. | | Collapse law | Named law that interprets or reduces retained plurality. | +| Topology intent | Admitted causal-history operation that changes lane or braid geometry. | Claim lifecycle: diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index fb0b2ff7..21b77679 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -84,6 +84,25 @@ WAL-backed ACK boundary and whether recovery can later classify it as pending, decided, rejected, or obstructed. If that boundary is unavailable, the editor must say so honestly instead of pretending a local buffer is causal history. +## Topology Intents + +The same rule applies to topology. A strand fork, braid creation, member weave, +braid settlement, braid collapse, or replica suffix import should not be a +side-channel runtime call. Each one is a topology intent that changes causal +history or the geometry through which causal history is interpreted. + +The current braid and strand implementation already has typed strand +construction, braid event logs, settlement provenance entries, retained braid +shells, and replay/audit optics. The remaining durability gate is to promote +those topology operations to WAL-backed accepted evidence and recoverable +WSC-retained material. That work is tracked by the braids and strands Goalpost 6 +roadmap. + +Until that lands, Echo should not overclaim that braid shells and topology +indexes have the same explicit crash-recovery posture as tick receipts. It can +claim that they are causal/provenance entities with retained replay shapes, and +it can name WAL/WSC topology recovery as the required next hardening step. + ## WAL, Graph Facts, And WSC WAL bytes are the durable commit authority. WARP graph facts can track WAL From ef55f9a03deef1ceed518564ff9b136223a9ee97 Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 03:54:53 -0700 Subject: [PATCH 4/7] Add topology WAL and WSC recovery evidence --- crates/warp-core/src/causal_wal.rs | 969 +++++++++++++++++- crates/warp-core/src/wsc/mod.rs | 10 +- crates/warp-core/src/wsc/store.rs | 402 +++++++- .../tests/causal_wal_hardening_tests.rs | 1 + crates/warp-core/tests/causal_wal_tests.rs | 240 ++++- crates/warp-core/tests/wsc_store_tests.rs | 181 +++- 6 files changed, 1772 insertions(+), 31 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index 9b7ec690..d93b2653 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -20,7 +20,14 @@ use std::path::{Path, PathBuf}; use thiserror::Error; +use crate::braid::{BraidEvent, BraidStatus}; +use crate::braid_shell::BraidMemberRef; +use crate::clock::WorldlineTick; +use crate::head::{HeadId, WriterHeadKey}; use crate::ident::Hash; +use crate::revelation::{AuthorityDomainId, AuthorityDomainRef, OriginId}; +use crate::strand::StrandId; +use crate::worldline::WorldlineId; const WAL_FRAME_DOMAIN: &[u8] = b"echo:causal_wal:frame:v1\0"; const WAL_PAYLOAD_DOMAIN: &[u8] = b"echo:causal_wal:payload:v1\0"; @@ -200,6 +207,8 @@ pub enum WalTransactionKind { Checkpoint, /// Side-effect outbox transaction. MaterializationOutbox, + /// Topology-changing strand, braid, or suffix-import intent evidence. + TopologyIntent, } impl WalTransactionKind { @@ -210,13 +219,14 @@ impl WalTransactionKind { Self::RuntimePosture => 3, Self::Checkpoint => 4, Self::MaterializationOutbox => 5, + Self::TopologyIntent => 6, } } fn required_authority(self) -> WalAppendAuthority { match self { Self::SubmissionIntake => WalAppendAuthority::SubmissionIntake, - Self::SchedulerTick | Self::MaterializationOutbox => { + Self::SchedulerTick | Self::MaterializationOutbox | Self::TopologyIntent => { WalAppendAuthority::TrustedScheduler } Self::RuntimePosture => WalAppendAuthority::RuntimeControl, @@ -231,6 +241,7 @@ impl WalTransactionKind { 3 => Ok(Self::RuntimePosture), 4 => Ok(Self::Checkpoint), 5 => Ok(Self::MaterializationOutbox), + 6 => Ok(Self::TopologyIntent), _ => Err(WalDecodeError::UnknownEnumCode { enum_name: "WalTransactionKind", code, @@ -274,6 +285,16 @@ pub enum WalRecordKind { MaterializationEffectObserved, /// Runtime recorded recovery posture. RecoveryPostureRecorded, + /// Runtime recorded accepted strand fork topology evidence. + TopologyStrandForkRecorded, + /// Runtime recorded accepted strand drop topology evidence. + TopologyStrandDropRecorded, + /// Runtime recorded accepted braid lifecycle event evidence. + TopologyBraidEventRecorded, + /// Runtime recorded retained braid-shell topology evidence. + TopologyBraidShellRetained, + /// Runtime recorded witnessed suffix import topology evidence. + TopologySuffixImportRecorded, } impl WalRecordKind { @@ -296,6 +317,11 @@ impl WalRecordKind { Self::MaterializationIntentRecorded => "MaterializationIntentRecorded", Self::MaterializationEffectObserved => "MaterializationEffectObserved", Self::RecoveryPostureRecorded => "RecoveryPostureRecorded", + Self::TopologyStrandForkRecorded => "TopologyStrandForkRecorded", + Self::TopologyStrandDropRecorded => "TopologyStrandDropRecorded", + Self::TopologyBraidEventRecorded => "TopologyBraidEventRecorded", + Self::TopologyBraidShellRetained => "TopologyBraidShellRetained", + Self::TopologySuffixImportRecorded => "TopologySuffixImportRecorded", } } @@ -314,7 +340,12 @@ impl WalRecordKind { | Self::ReadingEnvelopeRetained | Self::RetainedMaterialRefRecorded | Self::MaterializationIntentRecorded - | Self::MaterializationEffectObserved => WalAppendAuthority::TrustedScheduler, + | Self::MaterializationEffectObserved + | Self::TopologyStrandForkRecorded + | Self::TopologyStrandDropRecorded + | Self::TopologyBraidEventRecorded + | Self::TopologyBraidShellRetained + | Self::TopologySuffixImportRecorded => WalAppendAuthority::TrustedScheduler, Self::SchedulerFaultQuarantined | Self::TrustedRuntimeControlRecorded => { WalAppendAuthority::RuntimeControl } @@ -347,6 +378,11 @@ impl WalRecordKind { Self::MaterializationIntentRecorded => 14, Self::MaterializationEffectObserved => 15, Self::RecoveryPostureRecorded => 16, + Self::TopologyStrandForkRecorded => 17, + Self::TopologyStrandDropRecorded => 18, + Self::TopologyBraidEventRecorded => 19, + Self::TopologyBraidShellRetained => 20, + Self::TopologySuffixImportRecorded => 21, } } @@ -368,6 +404,11 @@ impl WalRecordKind { 14 => Ok(Self::MaterializationIntentRecorded), 15 => Ok(Self::MaterializationEffectObserved), 16 => Ok(Self::RecoveryPostureRecorded), + 17 => Ok(Self::TopologyStrandForkRecorded), + 18 => Ok(Self::TopologyStrandDropRecorded), + 19 => Ok(Self::TopologyBraidEventRecorded), + 20 => Ok(Self::TopologyBraidShellRetained), + 21 => Ok(Self::TopologySuffixImportRecorded), _ => Err(WalDecodeError::UnknownEnumCode { enum_name: "WalRecordKind", code, @@ -491,6 +532,8 @@ pub enum AffectedFrontierKind { RuntimeControl, /// Checkpoint/index frontier. CheckpointIndex, + /// Topology recovery index frontier. + TopologyIndex, } impl AffectedFrontierKind { @@ -502,6 +545,7 @@ impl AffectedFrontierKind { Self::ReadingIndex => 4, Self::RuntimeControl => 5, Self::CheckpointIndex => 6, + Self::TopologyIndex => 7, } } } @@ -3414,6 +3458,616 @@ pub fn recover_materialization_outbox( Ok(outbox) } +/// Imported suffix admission outcome stored in topology recovery evidence. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub enum TopologyImportOutcomeKind { + /// The import lowered to one derived result. + Derived, + /// The import retained lawful plurality. + Plural, + /// The import produced conflict residue. + Conflict, + /// The import was obstructed before admission completed. + Obstruction, +} + +impl TopologyImportOutcomeKind { + fn code(self) -> u8 { + match self { + Self::Derived => 1, + Self::Plural => 2, + Self::Conflict => 3, + Self::Obstruction => 4, + } + } + + fn from_code(code: u8) -> Result { + match code { + 1 => Ok(Self::Derived), + 2 => Ok(Self::Plural), + 3 => Ok(Self::Conflict), + 4 => Ok(Self::Obstruction), + _ => Err(WalDecodeError::UnknownEnumCode { + enum_name: "TopologyImportOutcomeKind", + code, + }), + } + } +} + +/// WAL evidence that one strand fork intent was accepted. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StrandForkRecord { + /// Stable topology intent id. + pub topology_intent_id: Hash, + /// Forked strand identity. + pub strand_id: StrandId, + /// Source worldline the strand was forked from. + pub source_worldline_id: WorldlineId, + /// Last included source tick in the copied prefix. + pub fork_tick: WorldlineTick, + /// Source commit hash at `fork_tick`. + pub source_commit_hash: Hash, + /// Source boundary hash at `fork_tick`. + pub source_boundary_hash: Hash, + /// Child worldline created for the strand. + pub child_worldline_id: WorldlineId, + /// Writer heads created for the child worldline. + pub writer_heads: Vec, + /// Digest of the accepted retention-posture bundle. + pub retention_posture_digest: Hash, + /// Issuer or session evidence digest. + pub issuer_evidence_digest: Hash, + /// Optional idempotency key supplied with the fork intent. + pub idempotency_key_digest: Option, +} + +impl StrandForkRecord { + /// Encodes the record as deterministic WAL payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::new(); + push_hash(&mut out, &self.topology_intent_id); + push_strand_id(&mut out, self.strand_id); + push_worldline_id(&mut out, self.source_worldline_id); + push_worldline_tick(&mut out, self.fork_tick); + push_hash(&mut out, &self.source_commit_hash); + push_hash(&mut out, &self.source_boundary_hash); + push_worldline_id(&mut out, self.child_worldline_id); + out.extend_from_slice(&len_u64(self.writer_heads.len()).to_le_bytes()); + for head in &self.writer_heads { + push_writer_head_key(&mut out, *head); + } + push_hash(&mut out, &self.retention_posture_digest); + push_hash(&mut out, &self.issuer_evidence_digest); + push_optional_hash(&mut out, self.idempotency_key_digest); + out + } + + /// Decodes a deterministic strand fork payload. + pub fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(bytes); + let topology_intent_id = cursor.read_hash()?; + let strand_id = cursor.read_strand_id()?; + let source_worldline_id = cursor.read_worldline_id()?; + let fork_tick = cursor.read_worldline_tick()?; + let source_commit_hash = cursor.read_hash()?; + let source_boundary_hash = cursor.read_hash()?; + let child_worldline_id = cursor.read_worldline_id()?; + let writer_count = + usize::try_from(cursor.read_u64()?).map_err(|_| WalDecodeError::UnexpectedEof)?; + let mut writer_heads = Vec::with_capacity(writer_count); + for _ in 0..writer_count { + writer_heads.push(cursor.read_writer_head_key()?); + } + let retention_posture_digest = cursor.read_hash()?; + let issuer_evidence_digest = cursor.read_hash()?; + let idempotency_key_digest = cursor.read_optional_hash()?; + cursor.finish()?; + Ok(Self { + topology_intent_id, + strand_id, + source_worldline_id, + fork_tick, + source_commit_hash, + source_boundary_hash, + child_worldline_id, + writer_heads, + retention_posture_digest, + issuer_evidence_digest, + idempotency_key_digest, + }) + } +} + +/// WAL evidence that one strand drop intent was accepted. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StrandDropRecord { + /// Stable topology intent id. + pub topology_intent_id: Hash, + /// Dropped strand identity. + pub strand_id: StrandId, + /// Child worldline removed from the live strand registry. + pub child_worldline_id: WorldlineId, + /// Final child-worldline tick at drop time. + pub final_tick: WorldlineTick, + /// Digest of the drop receipt returned after durable acceptance. + pub drop_receipt_digest: Hash, + /// Issuer or session evidence digest. + pub issuer_evidence_digest: Hash, + /// Optional idempotency key supplied with the drop intent. + pub idempotency_key_digest: Option, +} + +impl StrandDropRecord { + /// Encodes the record as deterministic WAL payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::new(); + push_hash(&mut out, &self.topology_intent_id); + push_strand_id(&mut out, self.strand_id); + push_worldline_id(&mut out, self.child_worldline_id); + push_worldline_tick(&mut out, self.final_tick); + push_hash(&mut out, &self.drop_receipt_digest); + push_hash(&mut out, &self.issuer_evidence_digest); + push_optional_hash(&mut out, self.idempotency_key_digest); + out + } + + /// Decodes a deterministic strand drop payload. + pub fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(bytes); + let topology_intent_id = cursor.read_hash()?; + let strand_id = cursor.read_strand_id()?; + let child_worldline_id = cursor.read_worldline_id()?; + let final_tick = cursor.read_worldline_tick()?; + let drop_receipt_digest = cursor.read_hash()?; + let issuer_evidence_digest = cursor.read_hash()?; + let idempotency_key_digest = cursor.read_optional_hash()?; + cursor.finish()?; + Ok(Self { + topology_intent_id, + strand_id, + child_worldline_id, + final_tick, + drop_receipt_digest, + issuer_evidence_digest, + idempotency_key_digest, + }) + } +} + +/// WAL evidence that one braid lifecycle event was accepted. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct TopologyBraidEventRecord { + /// Stable topology intent id. + pub topology_intent_id: Hash, + /// Braid whose event log accepted the event. + pub braid_id: Hash, + /// Sequence number in the durable braid event log. + pub event_index: u64, + /// Accepted braid event. + pub event: BraidEvent, + /// Folded braid status after accepting the event. + pub status_after: BraidStatus, + /// Digest of the accepted event payload. + pub event_digest: Hash, + /// Issuer or session evidence digest. + pub issuer_evidence_digest: Hash, + /// Optional idempotency key supplied with the braid event intent. + pub idempotency_key_digest: Option, +} + +impl TopologyBraidEventRecord { + /// Encodes the record as deterministic WAL payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::new(); + push_hash(&mut out, &self.topology_intent_id); + push_hash(&mut out, &self.braid_id); + out.extend_from_slice(&self.event_index.to_le_bytes()); + push_braid_event(&mut out, &self.event); + out.push(braid_status_code(self.status_after)); + push_hash(&mut out, &self.event_digest); + push_hash(&mut out, &self.issuer_evidence_digest); + push_optional_hash(&mut out, self.idempotency_key_digest); + out + } + + /// Decodes a deterministic braid event payload. + pub fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(bytes); + let topology_intent_id = cursor.read_hash()?; + let braid_id = cursor.read_hash()?; + let event_index = cursor.read_u64()?; + let event = cursor.read_braid_event()?; + let status_after = braid_status_from_code(cursor.read_u8()?)?; + let event_digest = cursor.read_hash()?; + let issuer_evidence_digest = cursor.read_hash()?; + let idempotency_key_digest = cursor.read_optional_hash()?; + cursor.finish()?; + Ok(Self { + topology_intent_id, + braid_id, + event_index, + event, + status_after, + event_digest, + issuer_evidence_digest, + idempotency_key_digest, + }) + } +} + +/// WAL evidence that a braid shell was retained. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BraidShellRetentionRecord { + /// Stable topology intent id. + pub topology_intent_id: Hash, + /// Braid associated with the shell. + pub braid_id: Hash, + /// Retained shell digest. + pub shell_digest: Hash, + /// Digest of retained shell material. + pub material_digest: Hash, + /// Digest of the shell coordinate or basis. + pub basis_digest: Hash, + /// Admission outcome family carried by the shell. + pub outcome_kind: TopologyImportOutcomeKind, + /// Retention posture digest for the shell material. + pub retention_posture_digest: Hash, + /// Witness digest binding the retained hologram. + pub witness_digest: Hash, + /// Optional idempotency key supplied with the retention intent. + pub idempotency_key_digest: Option, +} + +impl BraidShellRetentionRecord { + /// Encodes the record as deterministic WAL payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::new(); + push_hash(&mut out, &self.topology_intent_id); + push_hash(&mut out, &self.braid_id); + push_hash(&mut out, &self.shell_digest); + push_hash(&mut out, &self.material_digest); + push_hash(&mut out, &self.basis_digest); + out.push(self.outcome_kind.code()); + push_hash(&mut out, &self.retention_posture_digest); + push_hash(&mut out, &self.witness_digest); + push_optional_hash(&mut out, self.idempotency_key_digest); + out + } + + /// Decodes a deterministic retained braid-shell payload. + pub fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(bytes); + let topology_intent_id = cursor.read_hash()?; + let braid_id = cursor.read_hash()?; + let shell_digest = cursor.read_hash()?; + let material_digest = cursor.read_hash()?; + let basis_digest = cursor.read_hash()?; + let outcome_kind = TopologyImportOutcomeKind::from_code(cursor.read_u8()?)?; + let retention_posture_digest = cursor.read_hash()?; + let witness_digest = cursor.read_hash()?; + let idempotency_key_digest = cursor.read_optional_hash()?; + cursor.finish()?; + Ok(Self { + topology_intent_id, + braid_id, + shell_digest, + material_digest, + basis_digest, + outcome_kind, + retention_posture_digest, + witness_digest, + idempotency_key_digest, + }) + } +} + +/// WAL evidence that one remote suffix import intent was accepted. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SuffixImportRecord { + /// Stable import intent id. + pub import_id: Hash, + /// Digest naming the remote suffix family. + pub remote_suffix_family_digest: Hash, + /// Digest of remote authorship evidence. + pub authorship_evidence_digest: Hash, + /// Digest of comparable local/remote basis anchors. + pub basis_anchor_digest: Hash, + /// Digest of the transported causal suffix bundle. + pub bundle_digest: Hash, + /// Digest of the source suffix shell. + pub source_shell_digest: Hash, + /// Digest of the target basis used for judgment. + pub target_basis_digest: Hash, + /// Top-level import admission outcome. + pub outcome_kind: TopologyImportOutcomeKind, + /// Digest of the retained import shell or obstruction shell. + pub import_shell_digest: Hash, + /// Retention posture digest for import evidence. + pub retention_posture_digest: Hash, + /// Idempotency key for duplicate transport delivery. + pub idempotency_key_digest: Hash, +} + +impl SuffixImportRecord { + /// Encodes the record as deterministic WAL payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + let mut out = Vec::new(); + push_hash(&mut out, &self.import_id); + push_hash(&mut out, &self.remote_suffix_family_digest); + push_hash(&mut out, &self.authorship_evidence_digest); + push_hash(&mut out, &self.basis_anchor_digest); + push_hash(&mut out, &self.bundle_digest); + push_hash(&mut out, &self.source_shell_digest); + push_hash(&mut out, &self.target_basis_digest); + out.push(self.outcome_kind.code()); + push_hash(&mut out, &self.import_shell_digest); + push_hash(&mut out, &self.retention_posture_digest); + push_hash(&mut out, &self.idempotency_key_digest); + out + } + + /// Decodes a deterministic suffix import payload. + pub fn from_payload_bytes(bytes: &[u8]) -> Result { + let mut cursor = WalPayloadCursor::new(bytes); + let import_id = cursor.read_hash()?; + let remote_suffix_family_digest = cursor.read_hash()?; + let authorship_evidence_digest = cursor.read_hash()?; + let basis_anchor_digest = cursor.read_hash()?; + let bundle_digest = cursor.read_hash()?; + let source_shell_digest = cursor.read_hash()?; + let target_basis_digest = cursor.read_hash()?; + let outcome_kind = TopologyImportOutcomeKind::from_code(cursor.read_u8()?)?; + let import_shell_digest = cursor.read_hash()?; + let retention_posture_digest = cursor.read_hash()?; + let idempotency_key_digest = cursor.read_hash()?; + cursor.finish()?; + Ok(Self { + import_id, + remote_suffix_family_digest, + authorship_evidence_digest, + basis_anchor_digest, + bundle_digest, + source_shell_digest, + target_basis_digest, + outcome_kind, + import_shell_digest, + retention_posture_digest, + idempotency_key_digest, + }) + } +} + +/// One topology evidence record carried by a topology-intent transaction. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TopologyIntentRecord { + /// Strand fork evidence. + StrandFork(StrandForkRecord), + /// Strand drop evidence. + StrandDrop(StrandDropRecord), + /// Braid event evidence. + BraidEvent(TopologyBraidEventRecord), + /// Braid shell retention evidence. + BraidShell(BraidShellRetentionRecord), + /// Replica suffix import evidence. + SuffixImport(SuffixImportRecord), +} + +impl TopologyIntentRecord { + /// Returns the WAL record kind used by this topology evidence. + #[must_use] + pub const fn record_kind(&self) -> WalRecordKind { + match self { + Self::StrandFork(_) => WalRecordKind::TopologyStrandForkRecorded, + Self::StrandDrop(_) => WalRecordKind::TopologyStrandDropRecorded, + Self::BraidEvent(_) => WalRecordKind::TopologyBraidEventRecorded, + Self::BraidShell(_) => WalRecordKind::TopologyBraidShellRetained, + Self::SuffixImport(_) => WalRecordKind::TopologySuffixImportRecorded, + } + } + + /// Encodes this topology record to deterministic payload bytes. + #[must_use] + pub fn to_payload_bytes(&self) -> Vec { + match self { + Self::StrandFork(record) => record.to_payload_bytes(), + Self::StrandDrop(record) => record.to_payload_bytes(), + Self::BraidEvent(record) => record.to_payload_bytes(), + Self::BraidShell(record) => record.to_payload_bytes(), + Self::SuffixImport(record) => record.to_payload_bytes(), + } + } +} + +/// Recovered topology indexes rebuilt from committed WAL/WSC evidence. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct RecoveredTopologyIndex { + /// Accepted strand forks by strand id. + pub strand_forks: BTreeMap, + /// Accepted strand drops by strand id. + pub strand_drops: BTreeMap, + /// Child-worldline ancestry index. + pub child_worldlines: BTreeMap, + /// Accepted braid event logs by braid id. + pub braid_events: BTreeMap>, + /// Retained braid-shell records by shell digest. + pub braid_shells: BTreeMap, + /// Suffix imports by explicit import id. + pub suffix_imports: BTreeMap, + /// Suffix imports by transport idempotency key. + pub suffix_imports_by_idempotency_key: BTreeMap, + /// Suffix imports by bundle digest. + pub suffix_imports_by_bundle_digest: BTreeMap, +} + +impl RecoveredTopologyIndex { + /// Builds topology indexes from recovered topology evidence. + /// + /// Duplicate identical evidence is idempotent. Duplicate identities with + /// divergent material are recovery obstructions. + pub fn from_topology_records(records: I) -> Result + where + I: IntoIterator, + { + let mut index = Self::default(); + let mut braid_event_maps: BTreeMap> = + BTreeMap::new(); + + for record in records { + match record { + TopologyIntentRecord::StrandFork(record) => { + insert_unique( + &mut index.strand_forks, + record.strand_id, + record.clone(), + WalRecoveryIndexError::ConflictingStrandFork { + strand_id: record.strand_id, + }, + )?; + insert_unique( + &mut index.child_worldlines, + record.child_worldline_id, + record.strand_id, + WalRecoveryIndexError::ConflictingChildWorldline { + child_worldline_id: record.child_worldline_id, + }, + )?; + } + TopologyIntentRecord::StrandDrop(record) => { + insert_unique( + &mut index.strand_drops, + record.strand_id, + record.clone(), + WalRecoveryIndexError::ConflictingStrandDrop { + strand_id: record.strand_id, + }, + )?; + } + TopologyIntentRecord::BraidEvent(record) => { + let per_braid = braid_event_maps.entry(record.braid_id).or_default(); + insert_unique( + per_braid, + record.event_index, + record.clone(), + WalRecoveryIndexError::ConflictingBraidEvent { + braid_id: record.braid_id, + event_index: record.event_index, + }, + )?; + } + TopologyIntentRecord::BraidShell(record) => { + insert_unique( + &mut index.braid_shells, + record.shell_digest, + record.clone(), + WalRecoveryIndexError::ConflictingBraidShell { + shell_digest: record.shell_digest, + }, + )?; + } + TopologyIntentRecord::SuffixImport(record) => { + insert_unique( + &mut index.suffix_imports, + record.import_id, + record.clone(), + WalRecoveryIndexError::ConflictingSuffixImport { + import_id: record.import_id, + }, + )?; + insert_unique( + &mut index.suffix_imports_by_idempotency_key, + record.idempotency_key_digest, + record.import_id, + WalRecoveryIndexError::ConflictingSuffixImportIdempotency { + idempotency_key_digest: record.idempotency_key_digest, + }, + )?; + insert_unique( + &mut index.suffix_imports_by_bundle_digest, + record.bundle_digest, + record.import_id, + WalRecoveryIndexError::ConflictingSuffixImportBundle { + bundle_digest: record.bundle_digest, + }, + )?; + } + } + } + + index.braid_events = braid_event_maps + .into_iter() + .map(|(braid_id, events)| (braid_id, events.into_values().collect())) + .collect(); + Ok(index) + } + + /// Returns the number of accepted topology evidence records in the index. + #[must_use] + pub fn len(&self) -> usize { + self.strand_forks.len() + + self.strand_drops.len() + + self + .braid_events + .values() + .map(std::vec::Vec::len) + .sum::() + + self.braid_shells.len() + + self.suffix_imports.len() + } + + /// Returns `true` when the topology index contains no recovered evidence. + #[must_use] + pub fn is_empty(&self) -> bool { + self.len() == 0 + } +} + +/// Builds a stable root over recovered topology indexes. +#[must_use] +pub fn recovered_topology_index_root(index: &RecoveredTopologyIndex) -> Hash { + let mut hasher = blake3::Hasher::new(); + hasher.update(WAL_RECOVERED_INDEX_ROOT_DOMAIN); + hasher.update(b"topology"); + for (strand_id, record) in &index.strand_forks { + hasher.update(b"strand-fork"); + hasher.update(strand_id.as_bytes()); + hasher.update(&record.to_payload_bytes()); + } + for (strand_id, record) in &index.strand_drops { + hasher.update(b"strand-drop"); + hasher.update(strand_id.as_bytes()); + hasher.update(&record.to_payload_bytes()); + } + for (child_worldline_id, strand_id) in &index.child_worldlines { + hasher.update(b"child-worldline"); + hasher.update(child_worldline_id.as_bytes()); + hasher.update(strand_id.as_bytes()); + } + for (braid_id, events) in &index.braid_events { + hasher.update(b"braid-events"); + hasher.update(braid_id); + for event in events { + hasher.update(&event.to_payload_bytes()); + } + } + for (shell_digest, record) in &index.braid_shells { + hasher.update(b"braid-shell"); + hasher.update(shell_digest); + hasher.update(&record.to_payload_bytes()); + } + for (import_id, record) in &index.suffix_imports { + hasher.update(b"suffix-import"); + hasher.update(import_id); + hasher.update(&record.to_payload_bytes()); + } + hasher.finalize().into() +} + /// Causal commit evidence source. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum CausalCommitEvidenceSource { @@ -3744,6 +4398,12 @@ pub fn audit_wal_release_readiness(gates: WalReleaseReadinessGates) -> WalReleas "semantic_validator", gates.semantic_validator, ); + push_gate( + &mut passed_gates, + &mut blocked_gates, + "topology_recovery", + gates.topology_recovery, + ); push_gate( &mut passed_gates, &mut blocked_gates, @@ -3809,6 +4469,8 @@ pub struct WalReleaseReadinessGates { pub wal_doctor: bool, /// Semantic validator gate. pub semantic_validator: bool, + /// Topology recovery index gate. + pub topology_recovery: bool, /// Strict filesystem sync evidence gate. pub filesystem_sync_evidence: bool, /// Object-store manifest negative matrix gate. @@ -4291,6 +4953,18 @@ pub fn build_retained_reading_transaction( builder.commit(affected_frontiers) } +/// Builds a topology-intent transaction from typed topology evidence records. +pub fn build_topology_intent_transaction( + mut builder: WalTransactionBuilder, + records: &[TopologyIntentRecord], + affected_frontiers: Vec, +) -> Result { + for record in records { + builder.push_record(record.record_kind(), record.to_payload_bytes())?; + } + builder.commit(affected_frontiers) +} + /// Builds a checkpoint publication transaction. pub fn build_checkpoint_publication_transaction( mut builder: WalTransactionBuilder, @@ -4458,6 +5132,50 @@ pub fn recover_retention_index( Ok(index) } +/// Recovers topology indexes from committed WAL transactions. +pub fn recover_topology_index( + report: &RecoveryScanReport, +) -> Result { + let mut records = Vec::new(); + for transaction in &report.transactions { + for frame in &transaction.frames { + match frame.header.record_kind { + WalRecordKind::TopologyStrandForkRecorded => { + records.push(TopologyIntentRecord::StrandFork( + StrandForkRecord::from_payload_bytes(&frame.payload.canonical_bytes)?, + )); + } + WalRecordKind::TopologyStrandDropRecorded => { + records.push(TopologyIntentRecord::StrandDrop( + StrandDropRecord::from_payload_bytes(&frame.payload.canonical_bytes)?, + )); + } + WalRecordKind::TopologyBraidEventRecorded => { + records.push(TopologyIntentRecord::BraidEvent( + TopologyBraidEventRecord::from_payload_bytes( + &frame.payload.canonical_bytes, + )?, + )); + } + WalRecordKind::TopologyBraidShellRetained => { + records.push(TopologyIntentRecord::BraidShell( + BraidShellRetentionRecord::from_payload_bytes( + &frame.payload.canonical_bytes, + )?, + )); + } + WalRecordKind::TopologySuffixImportRecorded => { + records.push(TopologyIntentRecord::SuffixImport( + SuffixImportRecord::from_payload_bytes(&frame.payload.canonical_bytes)?, + )); + } + _ => {} + } + } + } + RecoveredTopologyIndex::from_topology_records(records) +} + /// Validates retained material references against an available material set. #[must_use] pub fn retained_material_obstructions( @@ -4989,6 +5707,56 @@ pub enum WalRecoveryIndexError { /// Conflicting submission id. submission_id: Hash, }, + /// Strand fork evidence conflicted for one strand id. + #[error("strand fork evidence conflicted for strand {strand_id:?}")] + ConflictingStrandFork { + /// Conflicting strand id. + strand_id: StrandId, + }, + /// Strand drop evidence conflicted for one strand id. + #[error("strand drop evidence conflicted for strand {strand_id:?}")] + ConflictingStrandDrop { + /// Conflicting strand id. + strand_id: StrandId, + }, + /// Child-worldline ancestry conflicted. + #[error("child worldline evidence conflicted for {child_worldline_id:?}")] + ConflictingChildWorldline { + /// Conflicting child worldline id. + child_worldline_id: WorldlineId, + }, + /// Braid event evidence conflicted at one event index. + #[error("braid event evidence conflicted for braid {braid_id:?} at event {event_index}")] + ConflictingBraidEvent { + /// Conflicting braid id. + braid_id: Hash, + /// Conflicting event index. + event_index: u64, + }, + /// Braid shell retention evidence conflicted for one digest. + #[error("braid shell retention evidence conflicted for shell {shell_digest:?}")] + ConflictingBraidShell { + /// Conflicting shell digest. + shell_digest: Hash, + }, + /// Suffix import evidence conflicted for one import id. + #[error("suffix import evidence conflicted for import {import_id:?}")] + ConflictingSuffixImport { + /// Conflicting import id. + import_id: Hash, + }, + /// Suffix import idempotency key mapped to conflicting imports. + #[error("suffix import idempotency key conflicted for {idempotency_key_digest:?}")] + ConflictingSuffixImportIdempotency { + /// Conflicting idempotency key digest. + idempotency_key_digest: Hash, + }, + /// Suffix import bundle digest mapped to conflicting imports. + #[error("suffix import bundle digest conflicted for {bundle_digest:?}")] + ConflictingSuffixImportBundle { + /// Conflicting bundle digest. + bundle_digest: Hash, + }, } /// Checkpoint file I/O errors. @@ -5134,6 +5902,9 @@ fn frontier_kind_allowed_for_transaction( WalTransactionKind::MaterializationOutbox => { matches!(frontier_kind, AffectedFrontierKind::ReceiptIndex) } + WalTransactionKind::TopologyIntent => { + matches!(frontier_kind, AffectedFrontierKind::TopologyIndex) + } } } @@ -5219,6 +5990,118 @@ fn push_optional_lsn(out: &mut Vec, lsn: Option) { } } +fn push_worldline_id(out: &mut Vec, worldline_id: WorldlineId) { + out.extend_from_slice(worldline_id.as_bytes()); +} + +fn push_worldline_tick(out: &mut Vec, tick: WorldlineTick) { + out.extend_from_slice(&tick.as_u64().to_le_bytes()); +} + +fn push_strand_id(out: &mut Vec, strand_id: StrandId) { + out.extend_from_slice(strand_id.as_bytes()); +} + +fn push_writer_head_key(out: &mut Vec, head: WriterHeadKey) { + out.extend_from_slice(head.worldline_id.as_bytes()); + out.extend_from_slice(head.head_id.as_bytes()); +} + +fn push_authority_domain_ref(out: &mut Vec, authority: AuthorityDomainRef) { + out.extend_from_slice(authority.origin_id.as_bytes()); + out.extend_from_slice(authority.domain_id.as_bytes()); +} + +fn push_braid_member_ref(out: &mut Vec, member_ref: BraidMemberRef) { + match member_ref { + BraidMemberRef::Revealed(strand_id) => { + out.push(1); + push_strand_id(out, strand_id); + } + BraidMemberRef::Sealed { + blinded_commitment, + authority, + } => { + out.push(2); + push_hash(out, &blinded_commitment); + push_authority_domain_ref(out, authority); + } + } +} + +fn push_braid_event(out: &mut Vec, event: &BraidEvent) { + match event { + BraidEvent::BraidCreated { + braid_id, + creator_domain, + } => { + out.push(1); + push_hash(out, braid_id); + push_authority_domain_ref(out, *creator_domain); + } + BraidEvent::MemberWoven { + member_ref, + sequence_num, + } => { + out.push(2); + push_braid_member_ref(out, *member_ref); + out.extend_from_slice(&sequence_num.to_le_bytes()); + } + BraidEvent::SettlementFinalized { settlement_digest } => { + out.push(3); + push_hash(out, settlement_digest); + } + BraidEvent::BraidCollapsed { + collapse_witness, + outcome_digest, + } => { + out.push(4); + push_hash(out, collapse_witness); + push_hash(out, outcome_digest); + } + } +} + +fn braid_status_code(status: BraidStatus) -> u8 { + match status { + BraidStatus::Active => 1, + BraidStatus::Finalized => 2, + BraidStatus::Collapsed => 3, + } +} + +fn braid_status_from_code(code: u8) -> Result { + match code { + 1 => Ok(BraidStatus::Active), + 2 => Ok(BraidStatus::Finalized), + 3 => Ok(BraidStatus::Collapsed), + _ => Err(WalDecodeError::UnknownEnumCode { + enum_name: "BraidStatus", + code, + }), + } +} + +fn insert_unique( + map: &mut BTreeMap, + key: K, + value: V, + error: WalRecoveryIndexError, +) -> Result<(), WalRecoveryIndexError> +where + K: Ord, + V: PartialEq, +{ + if let Some(existing) = map.get(&key) { + if existing != &value { + return Err(error); + } + return Ok(()); + } + map.insert(key, value); + Ok(()) +} + fn encode_frame(frame: &WalFrame) -> Vec { let mut out = Vec::new(); out.extend_from_slice(&frame.header.wal_version.to_le_bytes()); @@ -5478,6 +6361,88 @@ impl<'a> WalPayloadCursor<'a> { } } + fn read_worldline_id(&mut self) -> Result { + self.read_hash().map(WorldlineId::from_bytes) + } + + fn read_worldline_tick(&mut self) -> Result { + self.read_u64().map(WorldlineTick::from_raw) + } + + fn read_strand_id(&mut self) -> Result { + self.read_hash().map(StrandId::from_bytes) + } + + fn read_writer_head_key(&mut self) -> Result { + let worldline_id = self.read_worldline_id()?; + let head_id = HeadId::from_bytes(self.read_hash()?); + Ok(WriterHeadKey { + worldline_id, + head_id, + }) + } + + fn read_authority_domain_ref(&mut self) -> Result { + let origin_id = OriginId::from_bytes(self.read_hash()?); + let domain_id = AuthorityDomainId::from_bytes(self.read_hash()?); + Ok(AuthorityDomainRef::new(origin_id, domain_id)) + } + + fn read_braid_member_ref(&mut self) -> Result { + match self.read_u8()? { + 1 => self.read_strand_id().map(BraidMemberRef::Revealed), + 2 => { + let blinded_commitment = self.read_hash()?; + let authority = self.read_authority_domain_ref()?; + Ok(BraidMemberRef::Sealed { + blinded_commitment, + authority, + }) + } + code => Err(WalDecodeError::UnknownEnumCode { + enum_name: "BraidMemberRef", + code, + }), + } + } + + fn read_braid_event(&mut self) -> Result { + match self.read_u8()? { + 1 => { + let braid_id = self.read_hash()?; + let creator_domain = self.read_authority_domain_ref()?; + Ok(BraidEvent::BraidCreated { + braid_id, + creator_domain, + }) + } + 2 => { + let member_ref = self.read_braid_member_ref()?; + let sequence_num = self.read_u64()?; + Ok(BraidEvent::MemberWoven { + member_ref, + sequence_num, + }) + } + 3 => { + let settlement_digest = self.read_hash()?; + Ok(BraidEvent::SettlementFinalized { settlement_digest }) + } + 4 => { + let collapse_witness = self.read_hash()?; + let outcome_digest = self.read_hash()?; + Ok(BraidEvent::BraidCollapsed { + collapse_witness, + outcome_digest, + }) + } + code => Err(WalDecodeError::UnknownEnumCode { + enum_name: "BraidEvent", + code, + }), + } + } + fn read_vec(&mut self) -> Result, WalDecodeError> { let len = usize::try_from(self.read_u64()?).map_err(|_| WalDecodeError::UnexpectedEof)?; let end = self diff --git a/crates/warp-core/src/wsc/mod.rs b/crates/warp-core/src/wsc/mod.rs index bcece05d..c9db447b 100644 --- a/crates/warp-core/src/wsc/mod.rs +++ b/crates/warp-core/src/wsc/mod.rs @@ -74,10 +74,12 @@ pub use store::{ accepted_submission_records_to_wsc_envelope, receipt_correlation_records_from_wsc_envelope, receipt_correlation_records_from_wsc_store, receipt_correlation_records_to_wsc_envelope, retention_records_from_wsc_envelope, retention_records_from_wsc_store, - retention_records_to_wsc_envelope, validate_wsc_causal_history_store, InMemoryWscStore, - WscReceiptCorrelationRecords, WscRetentionRecords, WscStoreEnvelope, WscStoreEnvelopeId, - WscStoreObstruction, WscStoreObstructionKind, WscStorePort, WscStoreRecordKind, - WscStoreSubject, WscStoreWriteReceipt, + retention_records_to_wsc_envelope, topology_records_from_wsc_envelope, + topology_records_from_wsc_store, topology_records_to_wsc_envelope, + validate_wsc_causal_history_store, InMemoryWscStore, WscReceiptCorrelationRecords, + WscRetentionRecords, WscStoreEnvelope, WscStoreEnvelopeId, WscStoreObstruction, + WscStoreObstructionKind, WscStorePort, WscStoreRecordKind, WscStoreSubject, + WscStoreWriteReceipt, WscTopologyRecords, }; pub use validate::validate_wsc; pub use view::{AttachmentRef, WarpView, WscFile}; diff --git a/crates/warp-core/src/wsc/store.rs b/crates/warp-core/src/wsc/store.rs index e5a032e7..700cca1a 100644 --- a/crates/warp-core/src/wsc/store.rs +++ b/crates/warp-core/src/wsc/store.rs @@ -9,8 +9,9 @@ use bytes::Bytes; use crate::attachment::{AtomPayload, AttachmentValue}; use crate::causal_wal::{ - ReadingRefRecord, RetainedMaterialRecord, SubmissionAcceptanceRecord, TickReceiptRecord, - WalReceiptCorrelationRecord, + BraidShellRetentionRecord, ReadingRefRecord, RetainedMaterialRecord, StrandDropRecord, + StrandForkRecord, SubmissionAcceptanceRecord, SuffixImportRecord, TickReceiptRecord, + TopologyBraidEventRecord, TopologyIntentRecord, WalReceiptCorrelationRecord, }; use crate::graph::GraphStore; use crate::ident::{make_node_id, make_type_id, make_warp_id, EdgeId, Hash, NodeId}; @@ -60,6 +61,19 @@ const WSC_RETENTION_NODE_TYPE: &str = "echo/wsc-store/retention/node/v1"; const WSC_RETENTION_EDGE_TYPE: &str = "echo/wsc-store/retention/member/v1"; const WSC_RETAINED_MATERIAL_ATTACHMENT_TYPE: &str = "echo/wsc-store/retention/material/v1"; const WSC_READING_REF_ATTACHMENT_TYPE: &str = "echo/wsc-store/retention/reading/v1"; +const WSC_TOPOLOGY_BASIS_DOMAIN: &[u8] = b"echo:wsc_store:topology_basis:v1\0"; +const WSC_TOPOLOGY_NODE_DOMAIN: &[u8] = b"echo:wsc_store:topology_node:v1\0"; +const WSC_TOPOLOGY_EDGE_DOMAIN: &[u8] = b"echo:wsc_store:topology_edge:v1\0"; +const WSC_TOPOLOGY_SCHEMA: &str = "echo/wsc-store/topology/v1"; +const WSC_TOPOLOGY_WARP: &str = "echo/wsc-store/topology"; +const WSC_TOPOLOGY_ROOT: &str = "echo/wsc-store/topology/root"; +const WSC_TOPOLOGY_NODE_TYPE: &str = "echo/wsc-store/topology/node/v1"; +const WSC_TOPOLOGY_EDGE_TYPE: &str = "echo/wsc-store/topology/member/v1"; +const WSC_TOPOLOGY_STRAND_FORK_ATTACHMENT_TYPE: &str = "echo/wsc-store/topology/strand-fork/v1"; +const WSC_TOPOLOGY_STRAND_DROP_ATTACHMENT_TYPE: &str = "echo/wsc-store/topology/strand-drop/v1"; +const WSC_TOPOLOGY_BRAID_EVENT_ATTACHMENT_TYPE: &str = "echo/wsc-store/topology/braid-event/v1"; +const WSC_TOPOLOGY_BRAID_SHELL_ATTACHMENT_TYPE: &str = "echo/wsc-store/topology/braid-shell/v1"; +const WSC_TOPOLOGY_SUFFIX_IMPORT_ATTACHMENT_TYPE: &str = "echo/wsc-store/topology/suffix-import/v1"; const HEADER_LEN: usize = 124; /// Stable identifier for a WSC store envelope. @@ -498,6 +512,55 @@ pub struct WscRetentionRecords { pub readings: Vec, } +/// Topology records recovered from WSC material. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct WscTopologyRecords { + /// Strand fork evidence records. + pub strand_forks: Vec, + /// Strand drop evidence records. + pub strand_drops: Vec, + /// Braid lifecycle event records. + pub braid_events: Vec, + /// Retained braid shell records. + pub braid_shells: Vec, + /// Witnessed suffix import records. + pub suffix_imports: Vec, +} + +impl WscTopologyRecords { + /// Returns all topology records in deterministic typed order. + #[must_use] + pub fn into_topology_records(self) -> Vec { + let mut records = Vec::new(); + records.extend( + self.strand_forks + .into_iter() + .map(TopologyIntentRecord::StrandFork), + ); + records.extend( + self.strand_drops + .into_iter() + .map(TopologyIntentRecord::StrandDrop), + ); + records.extend( + self.braid_events + .into_iter() + .map(TopologyIntentRecord::BraidEvent), + ); + records.extend( + self.braid_shells + .into_iter() + .map(TopologyIntentRecord::BraidShell), + ); + records.extend( + self.suffix_imports + .into_iter() + .map(TopologyIntentRecord::SuffixImport), + ); + records + } +} + /// Generic WSC store port. pub trait WscStorePort { /// Writes a validated WSC envelope. @@ -1046,6 +1109,341 @@ where }) } +/// Builds a generic WSC envelope for topology evidence records. +/// +/// Duplicate identical records are represented once. Divergent duplicate +/// topology identities return a typed obstruction. +pub fn topology_records_to_wsc_envelope( + records: &[TopologyIntentRecord], +) -> Result { + let records = canonical_topology_records(records)?; + let mut store = GraphStore::new(make_warp_id(WSC_TOPOLOGY_WARP)); + let root = make_node_id(WSC_TOPOLOGY_ROOT); + store.insert_node( + root, + NodeRecord { + ty: make_type_id(WSC_TOPOLOGY_NODE_TYPE), + }, + ); + for record in &records { + let (role, attachment_type, payload_bytes) = topology_record_payload(record); + insert_topology_record_node( + &mut store, + root, + topology_node_id(role, &payload_bytes), + attachment_type, + payload_bytes, + ); + } + let basis_digest = topology_basis_digest(&records); + let input = build_one_warp_input(&store, root); + let wsc_bytes = write_wsc_one_warp(&input, make_type_id(WSC_TOPOLOGY_SCHEMA).0, 0) + .map_err(|_| WscStoreObstruction::invalid_wsc(basis_digest))?; + WscStoreEnvelope::validated(WscStoreRecordKind::CausalHistory, basis_digest, wsc_bytes) +} + +/// Recovers topology records from a generic WSC envelope. +pub fn topology_records_from_wsc_envelope( + envelope: &WscStoreEnvelope, +) -> Result { + if envelope.record_kind() != WscStoreRecordKind::CausalHistory { + return Err(WscStoreObstruction::invalid_envelope(0)); + } + let wsc_digest = *envelope.wsc_digest(); + let file = WscFile::from_bytes(envelope.wsc_bytes().to_vec()) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?; + validate_wsc(&file).map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?; + if file.schema_hash() != &make_type_id(WSC_TOPOLOGY_SCHEMA).0 { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + let view = file + .warp_view(0) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?; + let mut records = Vec::new(); + for node_index in 0..view.nodes().len() { + for attachment in view.node_attachments(node_index) { + let payload = atom_payload_bytes(&view, attachment, wsc_digest)?; + if attachment.type_or_warp == make_type_id(WSC_TOPOLOGY_STRAND_FORK_ATTACHMENT_TYPE).0 { + records.push(TopologyIntentRecord::StrandFork( + StrandForkRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + )); + } else if attachment.type_or_warp + == make_type_id(WSC_TOPOLOGY_STRAND_DROP_ATTACHMENT_TYPE).0 + { + records.push(TopologyIntentRecord::StrandDrop( + StrandDropRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + )); + } else if attachment.type_or_warp + == make_type_id(WSC_TOPOLOGY_BRAID_EVENT_ATTACHMENT_TYPE).0 + { + records.push(TopologyIntentRecord::BraidEvent( + TopologyBraidEventRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + )); + } else if attachment.type_or_warp + == make_type_id(WSC_TOPOLOGY_BRAID_SHELL_ATTACHMENT_TYPE).0 + { + records.push(TopologyIntentRecord::BraidShell( + BraidShellRetentionRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + )); + } else if attachment.type_or_warp + == make_type_id(WSC_TOPOLOGY_SUFFIX_IMPORT_ATTACHMENT_TYPE).0 + { + records.push(TopologyIntentRecord::SuffixImport( + SuffixImportRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + )); + } else { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + } + } + let records = canonical_topology_records(&records)?; + let basis_digest = topology_basis_digest(&records); + if envelope.basis_digest() != &basis_digest { + return Err(WscStoreObstruction::basis_digest_mismatch( + *envelope.basis_digest(), + basis_digest, + )); + } + Ok(split_topology_records(records)) +} + +/// Recovers topology records from committed WSC store envelopes. +pub fn topology_records_from_wsc_store

( + store: &P, +) -> Result +where + P: WscStorePort + ?Sized, +{ + let mut records = Vec::new(); + for envelope_id in store.list_envelopes() { + let envelope = store.read_envelope(envelope_id)?; + if envelope.record_kind() != WscStoreRecordKind::CausalHistory + || !envelope_has_schema(&envelope, WSC_TOPOLOGY_SCHEMA)? + { + continue; + } + records.extend(topology_records_from_wsc_envelope(&envelope)?.into_topology_records()); + } + Ok(split_topology_records(canonical_topology_records( + &records, + )?)) +} + +fn canonical_topology_records( + records: &[TopologyIntentRecord], +) -> Result, WscStoreObstruction> { + let mut by_payload = BTreeMap::new(); + let mut strand_forks = BTreeMap::new(); + let mut strand_drops = BTreeMap::new(); + let mut braid_events = BTreeMap::new(); + let mut braid_shells = BTreeMap::new(); + let mut suffix_imports = BTreeMap::new(); + let mut suffix_imports_by_idempotency = BTreeMap::new(); + let mut suffix_imports_by_bundle = BTreeMap::new(); + + for record in records { + match record { + TopologyIntentRecord::StrandFork(record) => { + insert_wsc_unique( + &mut strand_forks, + *record.strand_id.as_bytes(), + record, + record.strand_id.as_bytes(), + )?; + } + TopologyIntentRecord::StrandDrop(record) => { + insert_wsc_unique( + &mut strand_drops, + *record.strand_id.as_bytes(), + record, + record.strand_id.as_bytes(), + )?; + } + TopologyIntentRecord::BraidEvent(record) => { + insert_wsc_unique( + &mut braid_events, + (record.braid_id, record.event_index), + record, + &record.braid_id, + )?; + } + TopologyIntentRecord::BraidShell(record) => { + insert_wsc_unique( + &mut braid_shells, + record.shell_digest, + record, + &record.shell_digest, + )?; + } + TopologyIntentRecord::SuffixImport(record) => { + insert_wsc_unique( + &mut suffix_imports, + record.import_id, + record, + &record.import_id, + )?; + insert_wsc_unique( + &mut suffix_imports_by_idempotency, + record.idempotency_key_digest, + &record.import_id, + &record.idempotency_key_digest, + )?; + insert_wsc_unique( + &mut suffix_imports_by_bundle, + record.bundle_digest, + &record.import_id, + &record.bundle_digest, + )?; + } + } + by_payload.insert(topology_record_sort_key(record), record.clone()); + } + Ok(by_payload.into_values().collect()) +} + +fn insert_wsc_unique( + map: &mut BTreeMap, + key: K, + value: V, + envelope_hash: &Hash, +) -> Result<(), WscStoreObstruction> +where + K: Ord, + V: PartialEq, +{ + if let Some(existing) = map.get(&key) { + if existing != &value { + return Err(WscStoreObstruction::duplicate_mismatch( + WscStoreEnvelopeId::from_hash(*envelope_hash), + )); + } + return Ok(()); + } + map.insert(key, value); + Ok(()) +} + +fn split_topology_records(records: Vec) -> WscTopologyRecords { + let mut topology = WscTopologyRecords { + strand_forks: Vec::new(), + strand_drops: Vec::new(), + braid_events: Vec::new(), + braid_shells: Vec::new(), + suffix_imports: Vec::new(), + }; + for record in records { + match record { + TopologyIntentRecord::StrandFork(record) => topology.strand_forks.push(record), + TopologyIntentRecord::StrandDrop(record) => topology.strand_drops.push(record), + TopologyIntentRecord::BraidEvent(record) => topology.braid_events.push(record), + TopologyIntentRecord::BraidShell(record) => topology.braid_shells.push(record), + TopologyIntentRecord::SuffixImport(record) => topology.suffix_imports.push(record), + } + } + topology +} + +fn topology_record_payload( + record: &TopologyIntentRecord, +) -> (&'static [u8], &'static str, Vec) { + match record { + TopologyIntentRecord::StrandFork(record) => ( + b"strand-fork", + WSC_TOPOLOGY_STRAND_FORK_ATTACHMENT_TYPE, + record.to_payload_bytes(), + ), + TopologyIntentRecord::StrandDrop(record) => ( + b"strand-drop", + WSC_TOPOLOGY_STRAND_DROP_ATTACHMENT_TYPE, + record.to_payload_bytes(), + ), + TopologyIntentRecord::BraidEvent(record) => ( + b"braid-event", + WSC_TOPOLOGY_BRAID_EVENT_ATTACHMENT_TYPE, + record.to_payload_bytes(), + ), + TopologyIntentRecord::BraidShell(record) => ( + b"braid-shell", + WSC_TOPOLOGY_BRAID_SHELL_ATTACHMENT_TYPE, + record.to_payload_bytes(), + ), + TopologyIntentRecord::SuffixImport(record) => ( + b"suffix-import", + WSC_TOPOLOGY_SUFFIX_IMPORT_ATTACHMENT_TYPE, + record.to_payload_bytes(), + ), + } +} + +fn topology_record_sort_key(record: &TopologyIntentRecord) -> Vec { + let mut key = Vec::new(); + key.extend_from_slice(record.record_kind().label().as_bytes()); + key.push(0); + key.extend_from_slice(&record.to_payload_bytes()); + key +} + +fn topology_basis_digest(records: &[TopologyIntentRecord]) -> Hash { + let mut hasher = Hasher::new(); + hasher.update(WSC_TOPOLOGY_BASIS_DOMAIN); + for record in records { + hasher.update(record.record_kind().label().as_bytes()); + hasher.update(&record.to_payload_bytes()); + } + hasher.finalize().into() +} + +fn insert_topology_record_node( + store: &mut GraphStore, + root: NodeId, + node: NodeId, + attachment_type: &str, + payload_bytes: Vec, +) { + store.insert_node( + node, + NodeRecord { + ty: make_type_id(WSC_TOPOLOGY_NODE_TYPE), + }, + ); + store.insert_edge( + root, + EdgeRecord { + id: topology_edge_id(&node.0), + from: root, + to: node, + ty: make_type_id(WSC_TOPOLOGY_EDGE_TYPE), + }, + ); + store.set_node_attachment( + node, + Some(AttachmentValue::Atom(AtomPayload::new( + make_type_id(attachment_type), + Bytes::from(payload_bytes), + ))), + ); +} + +fn topology_node_id(role: &[u8], payload_bytes: &[u8]) -> NodeId { + let mut hasher = Hasher::new(); + hasher.update(WSC_TOPOLOGY_NODE_DOMAIN); + hasher.update(role); + hasher.update(payload_bytes); + NodeId(hasher.finalize().into()) +} + +fn topology_edge_id(node_id: &Hash) -> EdgeId { + let mut hasher = Hasher::new(); + hasher.update(WSC_TOPOLOGY_EDGE_DOMAIN); + hasher.update(node_id); + EdgeId(hasher.finalize().into()) +} + fn read_array(bytes: &[u8], offset: usize) -> Result<[u8; N], WscStoreObstruction> { let end = offset .checked_add(N) diff --git a/crates/warp-core/tests/causal_wal_hardening_tests.rs b/crates/warp-core/tests/causal_wal_hardening_tests.rs index cb55d3ba..6c281ed2 100644 --- a/crates/warp-core/tests/causal_wal_hardening_tests.rs +++ b/crates/warp-core/tests/causal_wal_hardening_tests.rs @@ -2155,6 +2155,7 @@ fn wal_hardening_gate_passes_when_all_categories_are_green() { commit_evidence: true, wal_doctor: true, semantic_validator: true, + topology_recovery: true, filesystem_sync_evidence: true, object_store_manifest_negatives: true, security_redaction: true, diff --git a/crates/warp-core/tests/causal_wal_tests.rs b/crates/warp-core/tests/causal_wal_tests.rs index d6ce4e38..49334b2e 100644 --- a/crates/warp-core/tests/causal_wal_tests.rs +++ b/crates/warp-core/tests/causal_wal_tests.rs @@ -13,28 +13,34 @@ use warp_core::causal_wal::{ apply_committed_transaction, audit_wal_release_readiness, build_checkpoint_publication_transaction, build_materialization_outbox_transaction, build_recovery_certificate, build_retained_reading_transaction, - build_submission_acceptance_transaction, build_tick_transaction, doctor_in_memory_store, - evaluate_checkpoint_publication, lint_wal_schema_terms, missing_material_scope, - project_causal_commit_evidence, read_checkpoint_record, recover_checkpoint_publications, - recover_filesystem_store, recover_in_memory_store, recover_materialization_outbox, - recover_receipt_index, recover_retention_index, recover_submission_index, - retained_material_obstructions, shadow_replay_matches, validate_checkpoint_record, - validate_strict_object_store_capabilities, write_checkpoint_record_atomic, AffectedFrontier, - AffectedFrontierKind, CheckpointPublicationRecord, CheckpointRecord, + build_submission_acceptance_transaction, build_tick_transaction, + build_topology_intent_transaction, doctor_in_memory_store, evaluate_checkpoint_publication, + lint_wal_schema_terms, missing_material_scope, project_causal_commit_evidence, + read_checkpoint_record, recover_checkpoint_publications, recover_filesystem_store, + recover_in_memory_store, recover_materialization_outbox, recover_receipt_index, + recover_retention_index, recover_submission_index, recover_topology_index, + recovered_topology_index_root, retained_material_obstructions, shadow_replay_matches, + validate_checkpoint_record, validate_strict_object_store_capabilities, + write_checkpoint_record_atomic, AffectedFrontier, AffectedFrontierKind, + BraidShellRetentionRecord, CheckpointPublicationRecord, CheckpointRecord, CheckpointValidationPosture, EvidenceMaterialPosture, ExistingMaterializedArtifact, FilesystemWalStore, InMemoryWalStore, Lsn, MaterializationIntentRecord, MaterializationObservationRecord, MaterializationReplayPosture, MissingMaterialScope, ObjectStoreCapabilityError, ObjectStoreReadAfterWritePosture, ObjectStoreWalCapabilities, PayloadCodecId, PayloadSchemaId, ReadingRefRecord, RecoveredState, RecoveredSubmissionPosture, - RecoveryAccessMode, RecoveryTailPosture, RetainedMaterialKind, RetainedMaterialRecord, - SubmissionAcceptanceRecord, SubmissionRetryPosture, TickReceiptRecord, TransactionLocalIndex, - WalAppendAuthority, WalBuildError, WalCommittedTransaction, WalDoctorPosture, - WalDurabilityMode, WalManifest, WalReceiptCorrelationRecord, WalRecordKind, - WalReleaseReadinessGates, WalSchemaLintError, WalSegmentId, WalStoreError, WalStorePort, - WalTickDecision, WalTransactionBuilder, WalTransactionId, WalTransactionKind, WriterEpochId, - WriterEpochRequest, + RecoveredTopologyIndex, RecoveryAccessMode, RecoveryTailPosture, RetainedMaterialKind, + RetainedMaterialRecord, StrandDropRecord, StrandForkRecord, SubmissionAcceptanceRecord, + SubmissionRetryPosture, SuffixImportRecord, TickReceiptRecord, TopologyBraidEventRecord, + TopologyImportOutcomeKind, TopologyIntentRecord, TransactionLocalIndex, WalAppendAuthority, + WalBuildError, WalCommittedTransaction, WalDoctorPosture, WalDurabilityMode, WalManifest, + WalReceiptCorrelationRecord, WalRecordKind, WalRecoveryIndexError, WalReleaseReadinessGates, + WalSchemaLintError, WalSegmentId, WalStoreError, WalStorePort, WalTickDecision, + WalTransactionBuilder, WalTransactionId, WalTransactionKind, WriterEpochId, WriterEpochRequest, +}; +use warp_core::{ + make_strand_id, AuthorityDomainId, AuthorityDomainRef, BraidEvent, BraidStatus, Hash, HeadId, + OriginId, WorldlineId, WorldlineTick, WriterHeadKey, }; -use warp_core::Hash; use std::collections::{BTreeMap, BTreeSet}; use std::fs::{self, OpenOptions}; @@ -198,6 +204,92 @@ fn reading_ref(label: &str, posture: EvidenceMaterialPosture) -> ReadingRefRecor } } +fn worldline(seed: u8) -> WorldlineId { + WorldlineId::from_bytes([seed; 32]) +} + +fn head(seed: u8, worldline_id: WorldlineId) -> WriterHeadKey { + WriterHeadKey { + worldline_id, + head_id: HeadId::from_bytes([seed; 32]), + } +} + +fn authority(seed: u8) -> AuthorityDomainRef { + AuthorityDomainRef::new( + OriginId::from_bytes([seed; 32]), + AuthorityDomainId::from_bytes([seed.wrapping_add(1); 32]), + ) +} + +fn topology_records() -> Vec { + let source_worldline = worldline(1); + let child_worldline = worldline(2); + let strand_id = make_strand_id("wal-topology-strand"); + let braid_id = digest("topology:braid"); + vec![ + TopologyIntentRecord::StrandFork(StrandForkRecord { + topology_intent_id: digest("topology:intent:fork"), + strand_id, + source_worldline_id: source_worldline, + fork_tick: WorldlineTick::from_raw(7), + source_commit_hash: digest("topology:source-commit"), + source_boundary_hash: digest("topology:source-boundary"), + child_worldline_id: child_worldline, + writer_heads: vec![head(3, child_worldline)], + retention_posture_digest: digest("topology:retention:fork"), + issuer_evidence_digest: digest("topology:issuer:fork"), + idempotency_key_digest: Some(digest("topology:idempotency:fork")), + }), + TopologyIntentRecord::StrandDrop(StrandDropRecord { + topology_intent_id: digest("topology:intent:drop"), + strand_id, + child_worldline_id: child_worldline, + final_tick: WorldlineTick::from_raw(11), + drop_receipt_digest: digest("topology:drop-receipt"), + issuer_evidence_digest: digest("topology:issuer:drop"), + idempotency_key_digest: Some(digest("topology:idempotency:drop")), + }), + TopologyIntentRecord::BraidEvent(TopologyBraidEventRecord { + topology_intent_id: digest("topology:intent:braid-event"), + braid_id, + event_index: 0, + event: BraidEvent::BraidCreated { + braid_id, + creator_domain: authority(9), + }, + status_after: BraidStatus::Active, + event_digest: digest("topology:braid-event"), + issuer_evidence_digest: digest("topology:issuer:braid"), + idempotency_key_digest: Some(digest("topology:idempotency:braid-event")), + }), + TopologyIntentRecord::BraidShell(BraidShellRetentionRecord { + topology_intent_id: digest("topology:intent:braid-shell"), + braid_id, + shell_digest: digest("topology:braid-shell"), + material_digest: digest("topology:braid-shell-material"), + basis_digest: digest("topology:braid-shell-basis"), + outcome_kind: TopologyImportOutcomeKind::Plural, + retention_posture_digest: digest("topology:retention:braid-shell"), + witness_digest: digest("topology:witness:braid-shell"), + idempotency_key_digest: Some(digest("topology:idempotency:braid-shell")), + }), + TopologyIntentRecord::SuffixImport(SuffixImportRecord { + import_id: digest("topology:import"), + remote_suffix_family_digest: digest("topology:remote-suffix-family"), + authorship_evidence_digest: digest("topology:authorship"), + basis_anchor_digest: digest("topology:basis-anchor"), + bundle_digest: digest("topology:bundle"), + source_shell_digest: digest("topology:source-shell"), + target_basis_digest: digest("topology:target-basis"), + outcome_kind: TopologyImportOutcomeKind::Derived, + import_shell_digest: digest("topology:import-shell"), + retention_posture_digest: digest("topology:retention:import"), + idempotency_key_digest: digest("topology:idempotency:import"), + }), + ] +} + fn submission_transaction(first_lsn: Lsn) -> WalCommittedTransaction { let mut builder = builder( transaction_id("tx:submission"), @@ -216,6 +308,24 @@ fn submission_transaction(first_lsn: Lsn) -> WalCommittedTransaction { )])) } +fn topology_transaction(first_lsn: Lsn) -> WalCommittedTransaction { + let builder = builder( + transaction_id("tx:topology"), + first_lsn, + WalAppendAuthority::TrustedScheduler, + WalTransactionKind::TopologyIntent, + ); + must_ok(build_topology_intent_transaction( + builder, + &topology_records(), + vec![frontier( + AffectedFrontierKind::TopologyIndex, + "topology:before", + "topology:after", + )], + )) +} + fn durable_submission_transaction(label: &str, first_lsn: Lsn) -> WalCommittedTransaction { let builder = builder( transaction_id(&format!("tx:submission:{label}")), @@ -865,6 +975,102 @@ fn same_payload_with_different_query_coordinate_remains_distinct() { .contains_key(&digest("coordinate:second"))); } +#[test] +fn topology_intent_transaction_recovers_strands_braids_shells_and_imports() { + let mut store = InMemoryWalStore::new(); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + let tx = topology_transaction(Lsn::from_raw(0)); + must_ok(store.append_transaction(tx)); + + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let topology = must_ok(recover_topology_index(&report)); + let records = topology_records(); + let strand_id = make_strand_id("wal-topology-strand"); + let child_worldline = worldline(2); + let braid_id = digest("topology:braid"); + let shell_digest = digest("topology:braid-shell"); + let import_id = digest("topology:import"); + let topology_root = recovered_topology_index_root(&topology); + let certificate = build_recovery_certificate(&report, None, 0, [0; 32], topology_root); + + assert_eq!(topology.len(), 5); + assert_eq!( + topology.strand_forks.get(&strand_id), + match &records[0] { + TopologyIntentRecord::StrandFork(record) => Some(record), + _ => None, + } + ); + assert_eq!( + topology.child_worldlines.get(&child_worldline), + Some(&strand_id) + ); + assert!(topology.strand_drops.contains_key(&strand_id)); + assert_eq!(must_some(topology.braid_events.get(&braid_id)).len(), 1); + assert!(topology.braid_shells.contains_key(&shell_digest)); + assert!(topology.suffix_imports.contains_key(&import_id)); + assert_eq!( + topology + .suffix_imports_by_idempotency_key + .get(&digest("topology:idempotency:import")), + Some(&import_id) + ); + assert_eq!(certificate.recovered_indexes_root, topology_root); +} + +#[test] +fn topology_uncommitted_tail_does_not_materialize_half_fork() { + let mut store = InMemoryWalStore::new(); + must_ok(store.acquire_writer_epoch(writer_epoch_request())); + let tx = topology_transaction(Lsn::from_raw(0)); + let epoch = tx.commit.writer_epoch; + for frame in tx.frames { + must_ok(store.append_uncommitted_frame(epoch, frame)); + } + + let report = must_ok(recover_in_memory_store( + &mut store, + RecoveryAccessMode::ReadOnly, + )); + let topology = must_ok(recover_topology_index(&report)); + + assert_eq!(report.tail_posture, RecoveryTailPosture::WouldTruncateAll); + assert!(topology.is_empty()); +} + +#[test] +fn topology_duplicate_idempotent_records_replay_once_and_divergent_records_obstruct() { + let records = topology_records(); + let fork = match &records[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + let idempotent = must_ok(RecoveredTopologyIndex::from_topology_records([ + TopologyIntentRecord::StrandFork(fork.clone()), + TopologyIntentRecord::StrandFork(fork.clone()), + ])); + assert_eq!(idempotent.strand_forks.len(), 1); + + let mut divergent = fork; + divergent.child_worldline_id = worldline(99); + let obstruction = RecoveredTopologyIndex::from_topology_records([ + TopologyIntentRecord::StrandFork(match &records[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }), + TopologyIntentRecord::StrandFork(divergent), + ]) + .expect_err("divergent duplicate strand fork obstructs"); + + assert!(matches!( + obstruction, + WalRecoveryIndexError::ConflictingStrandFork { .. } + )); +} + #[test] fn security_and_redaction_postures_decode_without_becoming_missing() { for posture in [ @@ -1356,6 +1562,7 @@ fn wal_release_readiness_audit_reports_blocked_and_ready_gates() { assert!(!blocked.ready); assert!(blocked.passed_gates.contains(&"filesystem_adapter")); assert!(blocked.blocked_gates.contains(&"shadow_replay")); + assert!(blocked.blocked_gates.contains(&"topology_recovery")); let ready = audit_wal_release_readiness(WalReleaseReadinessGates { filesystem_adapter: true, @@ -1370,6 +1577,7 @@ fn wal_release_readiness_audit_reports_blocked_and_ready_gates() { commit_evidence: true, wal_doctor: true, semantic_validator: true, + topology_recovery: true, filesystem_sync_evidence: true, object_store_manifest_negatives: true, security_redaction: true, diff --git a/crates/warp-core/tests/wsc_store_tests.rs b/crates/warp-core/tests/wsc_store_tests.rs index 160b3c4b..24b19f37 100644 --- a/crates/warp-core/tests/wsc_store_tests.rs +++ b/crates/warp-core/tests/wsc_store_tests.rs @@ -7,10 +7,12 @@ use std::collections::BTreeSet; use warp_core::causal_wal::{ - retained_material_obstructions, EvidenceMaterialPosture, MissingMaterialScope, - ReadingRefRecord, RecoveredReceiptIndex, RecoveredRetentionIndex, RecoveredRetentionIndexError, - RecoveredSubmissionIndex, RecoveredSubmissionPosture, RetainedMaterialKind, - RetainedMaterialRecord, SubmissionAcceptanceRecord, SubmissionRetryPosture, TickReceiptRecord, + retained_material_obstructions, BraidShellRetentionRecord, EvidenceMaterialPosture, + MissingMaterialScope, ReadingRefRecord, RecoveredReceiptIndex, RecoveredRetentionIndex, + RecoveredRetentionIndexError, RecoveredSubmissionIndex, RecoveredSubmissionPosture, + RecoveredTopologyIndex, RetainedMaterialKind, RetainedMaterialRecord, StrandDropRecord, + StrandForkRecord, SubmissionAcceptanceRecord, SubmissionRetryPosture, SuffixImportRecord, + TickReceiptRecord, TopologyBraidEventRecord, TopologyImportOutcomeKind, TopologyIntentRecord, WalReceiptCorrelationRecord, WalTickDecision, }; use warp_core::wsc::{ @@ -18,9 +20,14 @@ use warp_core::wsc::{ accepted_submission_records_to_wsc_envelope, receipt_correlation_records_from_wsc_envelope, receipt_correlation_records_from_wsc_store, receipt_correlation_records_to_wsc_envelope, retention_records_from_wsc_envelope, retention_records_from_wsc_store, - retention_records_to_wsc_envelope, validate_wsc_causal_history_store, write_wsc_one_warp, - InMemoryWscStore, OneWarpInput, WscStoreEnvelope, WscStoreObstructionKind, WscStorePort, - WscStoreRecordKind, WscStoreSubject, + retention_records_to_wsc_envelope, topology_records_from_wsc_envelope, + topology_records_from_wsc_store, topology_records_to_wsc_envelope, + validate_wsc_causal_history_store, write_wsc_one_warp, InMemoryWscStore, OneWarpInput, + WscStoreEnvelope, WscStoreObstructionKind, WscStorePort, WscStoreRecordKind, WscStoreSubject, +}; +use warp_core::{ + make_strand_id, AuthorityDomainId, AuthorityDomainRef, BraidEvent, BraidStatus, HeadId, + OriginId, WorldlineId, WorldlineTick, WriterHeadKey, }; #[test] @@ -565,6 +572,80 @@ fn retention_records_recover_from_committed_wsc_store() { assert_eq!(recovered.readings, vec![reading]); } +#[test] +fn topology_records_round_trip_through_wsc_envelope() { + let records = topology_records(); + let envelope = topology_records_to_wsc_envelope(&records).expect("topology WSC envelope"); + + let recovered = + topology_records_from_wsc_envelope(&envelope).expect("recovered topology records"); + let index = + RecoveredTopologyIndex::from_topology_records(recovered.clone().into_topology_records()) + .expect("recovered topology index"); + + assert_eq!(recovered.into_topology_records(), records); + assert_eq!(index.len(), 5); + assert!(index + .child_worldlines + .get(&worldline(2)) + .is_some_and(|strand_id| *strand_id == make_strand_id("wsc-topology-strand"))); + assert!(index.braid_shells.contains_key(&[26; 32])); + assert_eq!( + index.suffix_imports_by_idempotency_key.get(&[41; 32]), + Some(&[32; 32]) + ); +} + +#[test] +fn topology_records_recover_from_committed_wsc_store() { + let mut store = InMemoryWscStore::default(); + store + .write_envelope( + topology_records_to_wsc_envelope(&topology_records()).expect("topology WSC envelope"), + ) + .expect("committed topology WSC envelope"); + + let recovered = topology_records_from_wsc_store(&store).expect("recovered topology"); + let index = RecoveredTopologyIndex::from_topology_records(recovered.into_topology_records()) + .expect("recovered topology index"); + + assert_eq!(index.len(), 5); + assert!(index.braid_events.contains_key(&[10; 32])); +} + +#[test] +fn topology_records_ignore_uncommitted_staged_wsc_envelope() { + let envelope = + topology_records_to_wsc_envelope(&topology_records()).expect("topology WSC envelope"); + let mut store = InMemoryWscStore::default(); + store + .stage_envelope_without_commit_marker(envelope) + .expect("staged topology WSC envelope"); + + let recovered = topology_records_from_wsc_store(&store).expect("recovered topology"); + + assert!(recovered.into_topology_records().is_empty()); +} + +#[test] +fn topology_records_reject_conflicting_duplicate_strand_fork() { + let mut records = topology_records(); + let mut conflicting = match &records[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + conflicting.child_worldline_id = worldline(99); + records.push(TopologyIntentRecord::StrandFork(conflicting)); + + let obstruction = topology_records_to_wsc_envelope(&records) + .expect_err("conflicting topology duplicate obstructs"); + + assert_eq!( + obstruction.kind, + WscStoreObstructionKind::DuplicateEnvelopeMismatch + ); +} + #[test] fn retention_records_from_committed_wsc_store_rejects_conflicting_material_digest() { let material = retained_material( @@ -797,3 +878,89 @@ fn reading_ref( posture, } } + +fn worldline(seed: u8) -> WorldlineId { + WorldlineId::from_bytes([seed; 32]) +} + +fn head(seed: u8, worldline_id: WorldlineId) -> WriterHeadKey { + WriterHeadKey { + worldline_id, + head_id: HeadId::from_bytes([seed; 32]), + } +} + +fn authority(seed: u8) -> AuthorityDomainRef { + AuthorityDomainRef::new( + OriginId::from_bytes([seed; 32]), + AuthorityDomainId::from_bytes([seed.wrapping_add(1); 32]), + ) +} + +fn topology_records() -> Vec { + let source_worldline = worldline(1); + let child_worldline = worldline(2); + let strand_id = make_strand_id("wsc-topology-strand"); + let braid_id = [10; 32]; + vec![ + TopologyIntentRecord::StrandFork(StrandForkRecord { + topology_intent_id: [11; 32], + strand_id, + source_worldline_id: source_worldline, + fork_tick: WorldlineTick::from_raw(7), + source_commit_hash: [12; 32], + source_boundary_hash: [13; 32], + child_worldline_id: child_worldline, + writer_heads: vec![head(3, child_worldline)], + retention_posture_digest: [14; 32], + issuer_evidence_digest: [15; 32], + idempotency_key_digest: Some([16; 32]), + }), + TopologyIntentRecord::StrandDrop(StrandDropRecord { + topology_intent_id: [17; 32], + strand_id, + child_worldline_id: child_worldline, + final_tick: WorldlineTick::from_raw(11), + drop_receipt_digest: [18; 32], + issuer_evidence_digest: [19; 32], + idempotency_key_digest: Some([20; 32]), + }), + TopologyIntentRecord::BraidEvent(TopologyBraidEventRecord { + topology_intent_id: [21; 32], + braid_id, + event_index: 0, + event: BraidEvent::BraidCreated { + braid_id, + creator_domain: authority(9), + }, + status_after: BraidStatus::Active, + event_digest: [22; 32], + issuer_evidence_digest: [23; 32], + idempotency_key_digest: Some([24; 32]), + }), + TopologyIntentRecord::BraidShell(BraidShellRetentionRecord { + topology_intent_id: [25; 32], + braid_id, + shell_digest: [26; 32], + material_digest: [27; 32], + basis_digest: [28; 32], + outcome_kind: TopologyImportOutcomeKind::Plural, + retention_posture_digest: [29; 32], + witness_digest: [30; 32], + idempotency_key_digest: Some([31; 32]), + }), + TopologyIntentRecord::SuffixImport(SuffixImportRecord { + import_id: [32; 32], + remote_suffix_family_digest: [33; 32], + authorship_evidence_digest: [34; 32], + basis_anchor_digest: [35; 32], + bundle_digest: [36; 32], + source_shell_digest: [37; 32], + target_basis_digest: [38; 32], + outcome_kind: TopologyImportOutcomeKind::Derived, + import_shell_digest: [39; 32], + retention_posture_digest: [40; 32], + idempotency_key_digest: [41; 32], + }), + ] +} From 2b15daa6ea93eb8f1f42520cd81dcc4b49b6643b Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 04:07:08 -0700 Subject: [PATCH 5/7] Add WAL WSC durability release gate --- docs/README.md | 1 + ...st-06-topology-intents-and-wal-recovery.md | 18 +- docs/design/wal-wsc-release-closure-audit.md | 136 +++++++++++++ docs/workflows.md | 1 + scripts/check-wal-wsc-doctrine.sh | 42 ++++ scripts/tests/check_wal_wsc_doctrine_test.sh | 30 ++- xtask/src/main.rs | 182 +++++++++++++++++- 7 files changed, 407 insertions(+), 3 deletions(-) create mode 100644 docs/design/wal-wsc-release-closure-audit.md diff --git a/docs/README.md b/docs/README.md index 0b56246d..f7011d96 100644 --- a/docs/README.md +++ b/docs/README.md @@ -18,6 +18,7 @@ Echo's live documentation centers on the runtime carrier, the retained witnesses - Echo v0.1.0 jedit release gate: [/design/v0.1.0-jedit-release-gate](/design/v0.1.0-jedit-release-gate) - Trusted runtime control history: [/design/trusted-runtime-control-history](/design/trusted-runtime-control-history) - WAL truth boundary: [/topics/WAL](/topics/WAL) +- WAL/WSC release closure audit: [/design/wal-wsc-release-closure-audit](/design/wal-wsc-release-closure-audit) - jedit next ten release-gate slices: [/design/v0.1.0-jedit-next-ten-slices](/design/v0.1.0-jedit-next-ten-slices) - Theory map: [/theory/THEORY](/theory/THEORY) - Current bearing: [/BEARING](/BEARING) diff --git a/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md b/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md index 69075557..7642ee4b 100644 --- a/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md +++ b/docs/design/braids-and-strands-hardening/goalpost-06-topology-intents-and-wal-recovery.md @@ -3,7 +3,7 @@ # Goalpost 6: Topology Intents And WAL Recovery -Status: planned. +Status: implemented. Tracking issue: [#604](https://github.com/flyingrobots/echo/issues/604) @@ -44,6 +44,22 @@ provenance entries, retained braid shells, and braid replay/audit optics. This goalpost promotes the remaining topology operations to WAL-backed causal history so recovery can rebuild them after restart. +## Implementation Anchors + +- `crates/warp-core/src/causal_wal.rs` defines the topology WAL transaction and + record family, topology intent records, deterministic payload codecs, + recovered topology indexes, recovered topology roots, and readiness posture. +- `crates/warp-core/src/wsc/store.rs` serializes topology records into WSC + envelopes and recovers them from committed WSC store entries. +- `crates/warp-core/tests/causal_wal_tests.rs` proves topology WAL recovery, + uncommitted half-fork exclusion, duplicate idempotence, divergent duplicate + obstruction, and recovery-certificate index rooting. +- `crates/warp-core/tests/wsc_store_tests.rs` proves topology WSC round-trip, + committed-store recovery, uncommitted staged-envelope exclusion, and + conflicting duplicate obstruction. +- `crates/warp-core/tests/causal_wal_hardening_tests.rs` keeps topology + recovery in the WAL readiness gate. + ## Required Boundaries ### Strand Forks diff --git a/docs/design/wal-wsc-release-closure-audit.md b/docs/design/wal-wsc-release-closure-audit.md new file mode 100644 index 00000000..a56d601d --- /dev/null +++ b/docs/design/wal-wsc-release-closure-audit.md @@ -0,0 +1,136 @@ + + + +# WAL/WSC Release Closure Audit + +Status: evidence audit. + +This audit records the closure boundary for the WAL/WSC durability issues that +fed the GP6 release-gate slices: + +- [#519 Retained Evidence Durability Boundary](https://github.com/flyingrobots/echo/issues/519) +- [#521 WAL/WSC Storage Relationship](https://github.com/flyingrobots/echo/issues/521) +- [#522 WSC Causal-History Storage](https://github.com/flyingrobots/echo/issues/522) +- [#526 v0.1.0 Replay And DIND Proof](https://github.com/flyingrobots/echo/issues/526) +- [#370 Track Echo v0.1.0 release bar](https://github.com/flyingrobots/echo/issues/370) + +It is not a replacement for GitHub issue state. Close issues only when the PR +that carries the cited evidence is merged and CI confirms the commands below. + +## Release Witness + +The joined release witness is: + +```sh +cargo xtask test-slice durability-release +``` + +That slice joins these evidence families: + +- filesystem runtime WAL ACK and filesystem WAL failure atomicity; +- CLI submission posture JSON; +- WSC retained evidence recovery and conflict obstruction; +- WSC topology recovery and uncommitted staged-envelope exclusion; +- topology WAL recovery for strand forks, braid shells, and suffix imports; +- typed missing-material obstruction for retained evidence; +- stale durability claim guards; +- WAL/WSC doctrine shell guard; +- generated man-page freshness. + +The direct stale-claim and doctrine witnesses remain separately callable: + +```sh +cargo test -p xtask durability_stale_claims +scripts/check-wal-wsc-doctrine.sh +``` + +## Criteria Audit + +### #521 WAL/WSC Storage Relationship + +Ready to close after merge. + +Evidence: + +- `docs/design/causal-wal-end-to-end.md` defines WAL bytes as durable commit + authority, graph facts as projected evidence, storage locators as non-causal + identity, WSC export modes, record naming, and recovery bootstrap from WAL + root or storage manifest material. +- `docs/design/wal-wsc-durability-roadmap.md` preserves the stable doctrine. +- `scripts/check-wal-wsc-doctrine.sh` fails if the required doctrine is removed + from BEARING, WorkItems, sequencing, the WAL design, release contract, WAL + doctrine, or the WAL topic. +- `cargo xtask test-slice durability-release` runs the doctrine guard with the + durability and WSC witnesses. + +### #522 WSC Causal-History Storage + +Partially ready. Keep the umbrella open unless the project owner chooses to +close the doctrine/storage subset and track full import separately. + +Evidence now present: + +- ref-only, self-contained, and CAS-addressed WSC modes are documented in + `docs/design/causal-wal-end-to-end.md`; +- WSC retained-evidence envelopes round-trip and recover from committed WSC + store entries in `crates/warp-core/tests/wsc_store_tests.rs`; +- topology records round-trip through WSC envelopes, recover from committed WSC + store entries, ignore uncommitted staged entries, and reject conflicting + duplicate strand forks; +- the stale-claim guard rejects prose that says WSC import recovery is + authoritative without WAL-backed validation. + +Still open: + +- a full Continuum replica import fixture remains outside this storage slice; +- hostile-network and governance concerns remain explicitly non-goals here. + +### #519 Retained Evidence Durability Boundary + +Ready to close after merge for the retained-evidence boundary described by the +issue. + +Evidence: + +- `docs/design/causal-wal-end-to-end.md` distinguishes retained-evidence + posture from durable recovery evidence, semantic lookup identity, and byte + identity. +- `crates/warp-core/tests/causal_wal_tests.rs` covers typed obstruction for + missing retained material. +- `crates/warp-core/tests/wsc_store_tests.rs` covers retained-material WSC + round-trip, committed-store recovery, basis mismatch obstruction, conflicting + material digest obstruction, and conflicting reading id obstruction. +- `cargo test -p xtask durability_stale_claims` rejects posture-only retained + payload recovery claims. + +### #526 v0.1.0 Replay And DIND Proof + +Ready to close after merge for the documented narrow release witness. + +Evidence: + +- `cargo xtask test-slice contract-path-release` remains the local + contract-host release witness. +- `cargo xtask test-slice durability-release` adds the joined WAL/WSC recovery + witness for durability and retained evidence. +- Broader DIND remains valuable, but #526 already permits the narrower + documented release witness. + +### #370 Echo v0.1.0 Release Bar + +Keep open. + +The durability and replay criteria are now backed by concrete local witnesses, +but #370 tracks the full release bar. Authority boundary, clean-checkout +quickstart, package/versioning, and release operations remain broader than this +WAL/WSC closure audit. + +## Closure Rule + +Do not close umbrella release issues from prose alone. The closure event should +name: + +- the merged PR; +- the commit or merge commit; +- the commands that passed locally or in CI; +- any criteria intentionally left open with their owning issue. diff --git a/docs/workflows.md b/docs/workflows.md index 3156d1ce..ae90a6e6 100644 --- a/docs/workflows.md +++ b/docs/workflows.md @@ -150,6 +150,7 @@ The repo also exposes maintenance commands via `cargo xtask …`: - `cargo xtask test-slice contract-path-release` runs the v0.1 local contract-host release witness: installed contract pipeline replay, reference trusted host loop, and the serious external consumer fixture. - `cargo xtask test-slice runtime-wal-ack` runs the fast runtime WAL-backed ACK witness: app-facing acceptance rollback, scheduler tick receipt invariant checks, scheduler tick commit-before-publish, recovered indexes, CLI submission posture JSON, stale-claim guard, and generated man-page check. - `cargo xtask test-slice durable-runtime-wal` runs the release-grade filesystem runtime WAL durability witness: filesystem ACK recovery, filesystem failure atomicity, CLI submission posture JSON, stale-claim guard, and generated man-page check. +- `cargo xtask test-slice durability-release` runs the joined WAL/WSC release witness: filesystem runtime WAL durability, WSC retained evidence recovery, WSC topology recovery, topology WAL recovery, typed missing-material obstruction, stale-claim guards, doctrine checks, and generated man-page freshness. This is a release-gate slice, not the fastest local edit loop. - `cargo xtask pr-preflight` runs the default changed-scope pre-PR gate against `origin/main`. - `cargo xtask pr-preflight --full` runs the broader explicit full pre-PR gate. - `cargo xtask dind` runs the DIND (Deterministic Ironclad Nightmare Drills) harness locally. diff --git a/scripts/check-wal-wsc-doctrine.sh b/scripts/check-wal-wsc-doctrine.sh index 9180464f..6fcf141b 100755 --- a/scripts/check-wal-wsc-doctrine.sh +++ b/scripts/check-wal-wsc-doctrine.sh @@ -48,11 +48,29 @@ reject_literal() { fi } +reject_literal_anywhere() { + local label="$1" + local literal="$2" + shift 2 + + local file + for file in "$@"; do + if [[ ! -f "$file" ]]; then + fail "${label}: missing file ${file}" + continue + fi + if grep -Fq -- "$literal" "$file"; then + fail "${label}: rejected stale literal still present in ${file}: ${literal}" + fi + done +} + bearing="${repo_root}/docs/BEARING.md" workitems="${repo_root}/docs/WorkItems.md" sequencing="${repo_root}/docs/design/work-item-sequencing-and-prioritization.md" wal_design="${repo_root}/docs/design/causal-wal-end-to-end.md" wal_doctrine="${repo_root}/docs/design/wal-wsc-durability-roadmap.md" +wal_topic="${repo_root}/docs/topics/WAL.md" release_contract="${repo_root}/docs/releases/echo-1.0-contract.md" require_file "BEARING signpost" "$bearing" @@ -60,6 +78,7 @@ require_file "Work tracking boundary" "$workitems" require_file "GitHub-native sequencing doctrine" "$sequencing" require_file "causal WAL design" "$wal_design" require_file "WAL/WSC durability doctrine" "$wal_doctrine" +require_file "WAL topic" "$wal_topic" require_file "Echo 1.0 release contract" "$release_contract" project_url="https://github.com/users/flyingrobots/projects/15" @@ -273,6 +292,29 @@ reject_literal "WAL doctrine removes update date" "$wal_doctrine" "Last updated: reject_literal "WAL doctrine removes goalpost sections" "$wal_doctrine" "## Goalpost " reject_literal "WAL doctrine removes current PR tracking" "$wal_doctrine" "https://github.com/flyingrobots/echo/pull/582" +durability_claim_docs=( + "$bearing" + "$workitems" + "$sequencing" + "$wal_design" + "$wal_doctrine" + "$wal_topic" + "$release_contract" +) + +reject_literal_anywhere \ + "durability docs reject missing filesystem runtime WAL witness claim" \ + "filesystem runtime WAL witness is missing" \ + "${durability_claim_docs[@]}" +reject_literal_anywhere \ + "durability docs reject premature WSC import authority" \ + "WSC import recovery is authoritative without WAL-backed validation" \ + "${durability_claim_docs[@]}" +reject_literal_anywhere \ + "durability docs reject posture-only retained payload recovery" \ + "retained payload recovery can rely on posture-only refs" \ + "${durability_claim_docs[@]}" + if [[ "$failures" -ne 0 ]]; then exit 1 fi diff --git a/scripts/tests/check_wal_wsc_doctrine_test.sh b/scripts/tests/check_wal_wsc_doctrine_test.sh index abe60b9e..b33cf37a 100755 --- a/scripts/tests/check_wal_wsc_doctrine_test.sh +++ b/scripts/tests/check_wal_wsc_doctrine_test.sh @@ -25,7 +25,7 @@ fail() { copy_fixture() { local tmp="$1" - mkdir -p "${tmp}/docs/design" "${tmp}/docs/releases" + mkdir -p "${tmp}/docs/design" "${tmp}/docs/releases" "${tmp}/docs/topics" cp "${repo_root}/docs/BEARING.md" "${tmp}/docs/BEARING.md" cp "${repo_root}/docs/WorkItems.md" "${tmp}/docs/WorkItems.md" cp \ @@ -37,6 +37,7 @@ copy_fixture() { cp \ "${repo_root}/docs/design/wal-wsc-durability-roadmap.md" \ "${tmp}/docs/design/wal-wsc-durability-roadmap.md" + cp "${repo_root}/docs/topics/WAL.md" "${tmp}/docs/topics/WAL.md" cp \ "${repo_root}/docs/releases/echo-1.0-contract.md" \ "${tmp}/docs/releases/echo-1.0-contract.md" @@ -147,6 +148,32 @@ EOF } } +test_stale_durability_claims_fail() { + local tmp out + make_fixture tmp + + cat >>"${tmp}/docs/topics/WAL.md" <<'EOF' + +filesystem runtime WAL witness is missing +WSC import recovery is authoritative without WAL-backed validation +retained payload recovery can rely on posture-only refs +EOF + + out="$({ ECHO_REPO_ROOT="$tmp" "$checker"; } 2>&1 || true)" + echo "$out" | grep -q "durability docs reject missing filesystem runtime WAL witness claim" || { + echo "$out" >&2 + fail "checker did not report stale filesystem runtime WAL witness claim" + } + echo "$out" | grep -q "durability docs reject premature WSC import authority" || { + echo "$out" >&2 + fail "checker did not report stale WSC import authority claim" + } + echo "$out" | grep -q "durability docs reject posture-only retained payload recovery" || { + echo "$out" >&2 + fail "checker did not report stale retained payload recovery claim" + } +} + main() { [[ -x "$checker" ]] || fail "checker script missing or not executable: $checker" @@ -157,6 +184,7 @@ main() { test_missing_release_project_link_fails test_live_roadmap_issue_map_fails test_live_workitems_audit_fails + test_stale_durability_claims_fail } main "$@" diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 0e983f8d..948193df 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -112,6 +112,8 @@ enum TestSlice { RuntimeWalAck, /// Release-grade filesystem runtime WAL durability witness. DurableRuntimeWal, + /// Joined WAL/WSC durability release witness. + DurabilityRelease, } #[derive(Args)] @@ -722,6 +724,71 @@ fn build_test_slice_commands(slice: TestSlice) -> Vec { cargo_command(["test", "-p", "xtask", "runtime_wal_ack_stale_claims"]), cargo_command(["xtask", "man-pages", "--check"]), ], + TestSlice::DurabilityRelease => vec![ + cargo_command([ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "filesystem_runtime_wal_ack", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "filesystem_runtime_wal_failure", + ]), + cargo_command([ + "test", + "-p", + "warp-cli", + "--test", + "cli_integration", + "wal_submission_posture", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--test", + "wsc_store_tests", + "retention_records", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--test", + "wsc_store_tests", + "topology_records", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--test", + "causal_wal_tests", + "topology_", + ]), + cargo_command([ + "test", + "-p", + "warp-core", + "--test", + "causal_wal_tests", + "missing_retained_material_returns_typed_obstruction", + ]), + cargo_command(["test", "-p", "xtask", "durability_stale_claims"]), + script_command("scripts/check-wal-wsc-doctrine.sh"), + cargo_command(["xtask", "man-pages", "--check"]), + ], } } @@ -731,6 +798,10 @@ fn cargo_command(args: [&str; N]) -> Command { command } +fn script_command(path: &str) -> Command { + Command::new(path) +} + fn display_command(command: &Command) -> String { let mut parts = Vec::new(); parts.push(command.get_program().to_string_lossy().into_owned()); @@ -6622,7 +6693,110 @@ mod tests { } #[test] - fn runtime_wal_ack_stale_claims_stay_current() -> Result<(), Box> { + fn test_slice_durability_release_stays_explicit() { + let commands = build_test_slice_commands(TestSlice::DurabilityRelease); + assert_eq!(commands.len(), 10); + + let expected = [ + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "filesystem_runtime_wal_ack", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--features", + "native_rule_bootstrap trusted_runtime host_test", + "--test", + "trusted_runtime_host_loop_tests", + "filesystem_runtime_wal_failure", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-cli", + "--test", + "cli_integration", + "wal_submission_posture", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--test", + "wsc_store_tests", + "retention_records", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--test", + "wsc_store_tests", + "topology_records", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--test", + "causal_wal_tests", + "topology_", + ], + ), + ( + "cargo", + vec![ + "test", + "-p", + "warp-core", + "--test", + "causal_wal_tests", + "missing_retained_material_returns_typed_obstruction", + ], + ), + ( + "cargo", + vec!["test", "-p", "xtask", "durability_stale_claims"], + ), + ("scripts/check-wal-wsc-doctrine.sh", Vec::<&str>::new()), + ("cargo", vec!["xtask", "man-pages", "--check"]), + ]; + + for (command, (expected_program, expected_args)) in commands.iter().zip(expected) { + let (program, args) = command_program_and_args(command); + assert_eq!(program, expected_program); + assert_eq!(args, expected_args); + } + } + + #[test] + fn runtime_wal_ack_stale_claims_and_durability_stale_claims_stay_current( + ) -> Result<(), Box> { let repo_root = Path::new(env!("CARGO_MANIFEST_DIR")) .parent() .ok_or_else(|| { @@ -6632,6 +6806,8 @@ mod tests { "docs/BEARING.md", "docs/design/v0.1.0-jedit-release-gate.md", "docs/design/causal-wal-hardening-matrix.md", + "docs/design/wal-wsc-durability-roadmap.md", + "docs/topics/WAL.md", "docs/workflows.md", ]; let stale_claims = [ @@ -6640,6 +6816,9 @@ mod tests { "tick receipts are visible before WAL commit", "accepted submission evidence can be returned before WAL commit", "jedit recovery fixture contract is still missing", + "filesystem runtime WAL witness is missing", + "WSC import recovery is authoritative without WAL-backed validation", + "retained payload recovery can rely on posture-only refs", ]; for relative_path in checked_docs { @@ -6666,6 +6845,7 @@ mod tests { let workflows = fs::read_to_string(repo_root.join("docs/workflows.md"))?; assert!(workflows.contains("cargo xtask test-slice runtime-wal-ack")); assert!(workflows.contains("cargo xtask test-slice durable-runtime-wal")); + assert!(workflows.contains("cargo xtask test-slice durability-release")); Ok(()) } From fa8848d4020890f7488d006f498840a28bc1d231 Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 04:09:46 -0700 Subject: [PATCH 6/7] Align docs map with README --- AGENTS.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 5cd2451e..6e7bd1f9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -58,8 +58,7 @@ Do not audit the repository by recursively walking the filesystem. Follow the au - **`README.md`**: Public front door, core value prop, and quick tour. - **`GUIDE.md`**: Orientation and productive-fast path. -- **`docs/README.md`**: VitePress documentation map. -- **`docs/index.md`**: Documentation map. +- **`docs/README.md`**: Documentation map. ### 2. The Bedrock From 881797ce2beb2d21811b0e5b406a36688a439e5d Mon Sep 17 00:00:00 2001 From: James Ross Date: Fri, 26 Jun 2026 04:25:26 -0700 Subject: [PATCH 7/7] Address topology WAL WSC review findings --- crates/warp-core/src/causal_wal.rs | 77 ++++++- crates/warp-core/src/wsc/store.rs | 222 ++++++++++++++++----- crates/warp-core/tests/causal_wal_tests.rs | 95 +++++++++ crates/warp-core/tests/wsc_store_tests.rs | 150 +++++++++++++- docs/topics/WAL.md | 16 +- 5 files changed, 496 insertions(+), 64 deletions(-) diff --git a/crates/warp-core/src/causal_wal.rs b/crates/warp-core/src/causal_wal.rs index d93b2653..fc670cce 100644 --- a/crates/warp-core/src/causal_wal.rs +++ b/crates/warp-core/src/causal_wal.rs @@ -38,6 +38,7 @@ const WAL_RECOVERED_INDEX_ROOT_DOMAIN: &[u8] = b"echo:causal_wal:recovered_index const WAL_HEADER_CHECKSUM_DOMAIN: &[u8] = b"echo:causal_wal:header_checksum:v1\0"; const WAL_FRAME_CHECKSUM_DOMAIN: &[u8] = b"echo:causal_wal:frame_checksum:v1\0"; const WAL_DISK_RECORD_DOMAIN: &[u8] = b"echo:causal_wal:disk_record:v1\0"; +const WRITER_HEAD_KEY_PAYLOAD_LEN: usize = 64; const CHECKPOINT_FILE_MAGIC: &[u8; 8] = b"ECWALCP1"; const WAL_SEGMENT_RECORD_MAGIC: &[u8; 8] = b"ECWALR1!"; const WAL_SEGMENTS_DIR: &str = "segments"; @@ -3523,9 +3524,18 @@ pub struct StrandForkRecord { } impl StrandForkRecord { + /// Returns a copy with writer heads in canonical `(worldline_id, head_id)` order. + #[must_use] + pub fn canonicalized(&self) -> Self { + let mut record = self.clone(); + record.writer_heads = canonical_writer_heads(&record.writer_heads); + record + } + /// Encodes the record as deterministic WAL payload bytes. #[must_use] pub fn to_payload_bytes(&self) -> Vec { + let writer_heads = canonical_writer_heads(&self.writer_heads); let mut out = Vec::new(); push_hash(&mut out, &self.topology_intent_id); push_strand_id(&mut out, self.strand_id); @@ -3534,8 +3544,8 @@ impl StrandForkRecord { push_hash(&mut out, &self.source_commit_hash); push_hash(&mut out, &self.source_boundary_hash); push_worldline_id(&mut out, self.child_worldline_id); - out.extend_from_slice(&len_u64(self.writer_heads.len()).to_le_bytes()); - for head in &self.writer_heads { + out.extend_from_slice(&len_u64(writer_heads.len()).to_le_bytes()); + for head in &writer_heads { push_writer_head_key(&mut out, *head); } push_hash(&mut out, &self.retention_posture_digest); @@ -3556,10 +3566,17 @@ impl StrandForkRecord { let child_worldline_id = cursor.read_worldline_id()?; let writer_count = usize::try_from(cursor.read_u64()?).map_err(|_| WalDecodeError::UnexpectedEof)?; + let writer_bytes = writer_count + .checked_mul(WRITER_HEAD_KEY_PAYLOAD_LEN) + .ok_or(WalDecodeError::UnexpectedEof)?; + if writer_bytes > cursor.remaining_len() { + return Err(WalDecodeError::UnexpectedEof); + } let mut writer_heads = Vec::with_capacity(writer_count); for _ in 0..writer_count { writer_heads.push(cursor.read_writer_head_key()?); } + let writer_heads = canonical_writer_heads(&writer_heads); let retention_posture_digest = cursor.read_hash()?; let issuer_evidence_digest = cursor.read_hash()?; let idempotency_key_digest = cursor.read_optional_hash()?; @@ -3921,6 +3938,14 @@ impl RecoveredTopologyIndex { for record in records { match record { TopologyIntentRecord::StrandFork(record) => { + let record = record.canonicalized(); + if let Some(drop) = index.strand_drops.get(&record.strand_id) { + if drop.child_worldline_id != record.child_worldline_id { + return Err(WalRecoveryIndexError::ConflictingStrandFork { + strand_id: record.strand_id, + }); + } + } insert_unique( &mut index.strand_forks, record.strand_id, @@ -3939,6 +3964,13 @@ impl RecoveredTopologyIndex { )?; } TopologyIntentRecord::StrandDrop(record) => { + if let Some(fork) = index.strand_forks.get(&record.strand_id) { + if fork.child_worldline_id != record.child_worldline_id { + return Err(WalRecoveryIndexError::ConflictingStrandDrop { + strand_id: record.strand_id, + }); + } + } insert_unique( &mut index.strand_drops, record.strand_id, @@ -3949,6 +3981,7 @@ impl RecoveredTopologyIndex { )?; } TopologyIntentRecord::BraidEvent(record) => { + validate_topology_braid_event(&record)?; let per_braid = braid_event_maps.entry(record.braid_id).or_default(); insert_unique( per_braid, @@ -6082,6 +6115,42 @@ fn braid_status_from_code(code: u8) -> Result { } } +fn canonical_writer_heads(heads: &[WriterHeadKey]) -> Vec { + let mut sorted = heads.to_vec(); + sorted.sort_by(|left, right| { + left.worldline_id + .as_bytes() + .cmp(right.worldline_id.as_bytes()) + .then_with(|| left.head_id.as_bytes().cmp(right.head_id.as_bytes())) + }); + sorted +} + +fn validate_topology_braid_event( + record: &TopologyBraidEventRecord, +) -> Result<(), WalRecoveryIndexError> { + if let BraidEvent::BraidCreated { braid_id, .. } = &record.event { + if braid_id != &record.braid_id { + return Err(WalRecoveryIndexError::ConflictingBraidEvent { + braid_id: record.braid_id, + event_index: record.event_index, + }); + } + } + let expected_status = match &record.event { + BraidEvent::BraidCreated { .. } | BraidEvent::MemberWoven { .. } => BraidStatus::Active, + BraidEvent::SettlementFinalized { .. } => BraidStatus::Finalized, + BraidEvent::BraidCollapsed { .. } => BraidStatus::Collapsed, + }; + if record.status_after != expected_status { + return Err(WalRecoveryIndexError::ConflictingBraidEvent { + braid_id: record.braid_id, + event_index: record.event_index, + }); + } + Ok(()) +} + fn insert_unique( map: &mut BTreeMap, key: K, @@ -6275,6 +6344,10 @@ impl<'a> WalPayloadCursor<'a> { Self { bytes, offset: 0 } } + fn remaining_len(&self) -> usize { + self.bytes.len().saturating_sub(self.offset) + } + fn read_u8(&mut self) -> Result { let Some(value) = self.bytes.get(self.offset).copied() else { return Err(WalDecodeError::UnexpectedEof); diff --git a/crates/warp-core/src/wsc/store.rs b/crates/warp-core/src/wsc/store.rs index 700cca1a..0bffa1fb 100644 --- a/crates/warp-core/src/wsc/store.rs +++ b/crates/warp-core/src/wsc/store.rs @@ -1159,48 +1159,7 @@ pub fn topology_records_from_wsc_envelope( let view = file .warp_view(0) .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?; - let mut records = Vec::new(); - for node_index in 0..view.nodes().len() { - for attachment in view.node_attachments(node_index) { - let payload = atom_payload_bytes(&view, attachment, wsc_digest)?; - if attachment.type_or_warp == make_type_id(WSC_TOPOLOGY_STRAND_FORK_ATTACHMENT_TYPE).0 { - records.push(TopologyIntentRecord::StrandFork( - StrandForkRecord::from_payload_bytes(payload) - .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, - )); - } else if attachment.type_or_warp - == make_type_id(WSC_TOPOLOGY_STRAND_DROP_ATTACHMENT_TYPE).0 - { - records.push(TopologyIntentRecord::StrandDrop( - StrandDropRecord::from_payload_bytes(payload) - .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, - )); - } else if attachment.type_or_warp - == make_type_id(WSC_TOPOLOGY_BRAID_EVENT_ATTACHMENT_TYPE).0 - { - records.push(TopologyIntentRecord::BraidEvent( - TopologyBraidEventRecord::from_payload_bytes(payload) - .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, - )); - } else if attachment.type_or_warp - == make_type_id(WSC_TOPOLOGY_BRAID_SHELL_ATTACHMENT_TYPE).0 - { - records.push(TopologyIntentRecord::BraidShell( - BraidShellRetentionRecord::from_payload_bytes(payload) - .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, - )); - } else if attachment.type_or_warp - == make_type_id(WSC_TOPOLOGY_SUFFIX_IMPORT_ATTACHMENT_TYPE).0 - { - records.push(TopologyIntentRecord::SuffixImport( - SuffixImportRecord::from_payload_bytes(payload) - .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, - )); - } else { - return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); - } - } - } + let records = topology_records_from_wsc_view(&view, wsc_digest)?; let records = canonical_topology_records(&records)?; let basis_digest = topology_basis_digest(&records); if envelope.basis_digest() != &basis_digest { @@ -1212,6 +1171,119 @@ pub fn topology_records_from_wsc_envelope( Ok(split_topology_records(records)) } +fn topology_records_from_wsc_view( + view: &super::view::WarpView<'_>, + wsc_digest: Hash, +) -> Result, WscStoreObstruction> { + let expected_warp = make_warp_id(WSC_TOPOLOGY_WARP).0; + let root = make_node_id(WSC_TOPOLOGY_ROOT); + let node_type = make_type_id(WSC_TOPOLOGY_NODE_TYPE).0; + let edge_type = make_type_id(WSC_TOPOLOGY_EDGE_TYPE).0; + + if view.warp_id() != &expected_warp || view.root_node_id() != &root.0 { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + let Some(root_ix) = view.node_ix(&root.0) else { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + }; + if view.nodes()[root_ix].node_type != node_type || !view.node_attachments(root_ix).is_empty() { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + + let mut records = Vec::new(); + let mut record_node_ids = BTreeSet::new(); + for (node_ix, node) in view.nodes().iter().enumerate() { + if node.node_type != node_type { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + if node.node_id == root.0 { + continue; + } + let attachments = view.node_attachments(node_ix); + if attachments.len() != 1 { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + let attachment = &attachments[0]; + let payload = atom_payload_bytes(view, attachment, wsc_digest)?; + let (role, record) = + topology_record_from_attachment(attachment.type_or_warp, payload, wsc_digest)?; + if node.node_id != topology_node_id(role, payload).0 + || !record_node_ids.insert(node.node_id) + { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + records.push(record); + } + + let mut edge_targets = BTreeSet::new(); + for (edge_ix, edge) in view.edges().iter().enumerate() { + if !view.edge_attachments(edge_ix).is_empty() + || edge.edge_type != edge_type + || edge.from_node_id != root.0 + || edge.edge_id != topology_edge_id(&edge.to_node_id).0 + || !record_node_ids.contains(&edge.to_node_id) + || !edge_targets.insert(edge.to_node_id) + { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + } + if edge_targets != record_node_ids { + return Err(WscStoreObstruction::invalid_wsc(wsc_digest)); + } + + Ok(records) +} + +fn topology_record_from_attachment( + attachment_type: Hash, + payload: &[u8], + wsc_digest: Hash, +) -> Result<(&'static [u8], TopologyIntentRecord), WscStoreObstruction> { + if attachment_type == make_type_id(WSC_TOPOLOGY_STRAND_FORK_ATTACHMENT_TYPE).0 { + Ok(( + b"strand-fork", + TopologyIntentRecord::StrandFork( + StrandForkRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + ), + )) + } else if attachment_type == make_type_id(WSC_TOPOLOGY_STRAND_DROP_ATTACHMENT_TYPE).0 { + Ok(( + b"strand-drop", + TopologyIntentRecord::StrandDrop( + StrandDropRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + ), + )) + } else if attachment_type == make_type_id(WSC_TOPOLOGY_BRAID_EVENT_ATTACHMENT_TYPE).0 { + Ok(( + b"braid-event", + TopologyIntentRecord::BraidEvent( + TopologyBraidEventRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + ), + )) + } else if attachment_type == make_type_id(WSC_TOPOLOGY_BRAID_SHELL_ATTACHMENT_TYPE).0 { + Ok(( + b"braid-shell", + TopologyIntentRecord::BraidShell( + BraidShellRetentionRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + ), + )) + } else if attachment_type == make_type_id(WSC_TOPOLOGY_SUFFIX_IMPORT_ATTACHMENT_TYPE).0 { + Ok(( + b"suffix-import", + TopologyIntentRecord::SuffixImport( + SuffixImportRecord::from_payload_bytes(payload) + .map_err(|_| WscStoreObstruction::invalid_wsc(wsc_digest))?, + ), + )) + } else { + Err(WscStoreObstruction::invalid_wsc(wsc_digest)) + } +} + /// Recovers topology records from committed WSC store envelopes. pub fn topology_records_from_wsc_store

( store: &P, @@ -1239,73 +1311,121 @@ fn canonical_topology_records( ) -> Result, WscStoreObstruction> { let mut by_payload = BTreeMap::new(); let mut strand_forks = BTreeMap::new(); + let mut strand_forks_by_idempotency = BTreeMap::new(); let mut strand_drops = BTreeMap::new(); + let mut strand_drops_by_idempotency = BTreeMap::new(); let mut braid_events = BTreeMap::new(); + let mut braid_events_by_idempotency = BTreeMap::new(); let mut braid_shells = BTreeMap::new(); + let mut braid_shells_by_idempotency = BTreeMap::new(); let mut suffix_imports = BTreeMap::new(); let mut suffix_imports_by_idempotency = BTreeMap::new(); let mut suffix_imports_by_bundle = BTreeMap::new(); for record in records { - match record { + let record = canonical_topology_record(record); + match &record { TopologyIntentRecord::StrandFork(record) => { insert_wsc_unique( &mut strand_forks, *record.strand_id.as_bytes(), - record, + record.clone(), record.strand_id.as_bytes(), )?; + if let Some(idempotency_key_digest) = record.idempotency_key_digest { + insert_wsc_unique( + &mut strand_forks_by_idempotency, + idempotency_key_digest, + record.clone(), + &idempotency_key_digest, + )?; + } } TopologyIntentRecord::StrandDrop(record) => { insert_wsc_unique( &mut strand_drops, *record.strand_id.as_bytes(), - record, + record.clone(), record.strand_id.as_bytes(), )?; + if let Some(idempotency_key_digest) = record.idempotency_key_digest { + insert_wsc_unique( + &mut strand_drops_by_idempotency, + idempotency_key_digest, + record.clone(), + &idempotency_key_digest, + )?; + } } TopologyIntentRecord::BraidEvent(record) => { insert_wsc_unique( &mut braid_events, (record.braid_id, record.event_index), - record, + record.clone(), &record.braid_id, )?; + if let Some(idempotency_key_digest) = record.idempotency_key_digest { + insert_wsc_unique( + &mut braid_events_by_idempotency, + idempotency_key_digest, + record.clone(), + &idempotency_key_digest, + )?; + } } TopologyIntentRecord::BraidShell(record) => { insert_wsc_unique( &mut braid_shells, record.shell_digest, - record, + record.clone(), &record.shell_digest, )?; + if let Some(idempotency_key_digest) = record.idempotency_key_digest { + insert_wsc_unique( + &mut braid_shells_by_idempotency, + idempotency_key_digest, + record.clone(), + &idempotency_key_digest, + )?; + } } TopologyIntentRecord::SuffixImport(record) => { insert_wsc_unique( &mut suffix_imports, record.import_id, - record, + record.clone(), &record.import_id, )?; insert_wsc_unique( &mut suffix_imports_by_idempotency, record.idempotency_key_digest, - &record.import_id, + record.import_id, &record.idempotency_key_digest, )?; insert_wsc_unique( &mut suffix_imports_by_bundle, record.bundle_digest, - &record.import_id, + record.import_id, &record.bundle_digest, )?; } } - by_payload.insert(topology_record_sort_key(record), record.clone()); + by_payload.insert(topology_record_sort_key(&record), record); } Ok(by_payload.into_values().collect()) } +fn canonical_topology_record(record: &TopologyIntentRecord) -> TopologyIntentRecord { + match record { + TopologyIntentRecord::StrandFork(record) => { + TopologyIntentRecord::StrandFork(record.canonicalized()) + } + _ => record.clone(), + } +} + +/// Returns [`WscStoreObstructionKind::DuplicateEnvelopeMismatch`] for both +/// committed-store duplicate mismatches and canonical topology payload conflicts. fn insert_wsc_unique( map: &mut BTreeMap, key: K, diff --git a/crates/warp-core/tests/causal_wal_tests.rs b/crates/warp-core/tests/causal_wal_tests.rs index 49334b2e..14ba70ae 100644 --- a/crates/warp-core/tests/causal_wal_tests.rs +++ b/crates/warp-core/tests/causal_wal_tests.rs @@ -1071,6 +1071,101 @@ fn topology_duplicate_idempotent_records_replay_once_and_divergent_records_obstr )); } +#[test] +fn topology_strand_fork_writer_heads_are_canonical_for_payload_and_recovery() { + let mut fork = match &topology_records()[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + fork.writer_heads = vec![head(9, worldline(9)), head(1, worldline(1))]; + let mut reversed = fork.clone(); + reversed.writer_heads.reverse(); + + assert_eq!(fork.to_payload_bytes(), reversed.to_payload_bytes()); + let decoded = must_ok(StrandForkRecord::from_payload_bytes( + &fork.to_payload_bytes(), + )); + assert_eq!(decoded.writer_heads, reversed.writer_heads); + + let index = must_ok(RecoveredTopologyIndex::from_topology_records([ + TopologyIntentRecord::StrandFork(fork), + TopologyIntentRecord::StrandFork(reversed), + ])); + let recovered = must_some(index.strand_forks.values().next()); + assert_eq!(recovered.writer_heads, decoded.writer_heads); +} + +#[test] +fn topology_strand_fork_decode_bounds_writer_head_count_before_allocating() { + let fork = match &topology_records()[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + let mut payload = fork.to_payload_bytes(); + let writer_count_offset = 200; + payload[writer_count_offset..writer_count_offset + 8].copy_from_slice(&u64::MAX.to_le_bytes()); + + assert!(StrandForkRecord::from_payload_bytes(&payload).is_err()); +} + +#[test] +fn topology_strand_fork_and_drop_must_name_same_child_worldline() { + let records = topology_records(); + let fork = match &records[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + let mut drop = match &records[1] { + TopologyIntentRecord::StrandDrop(record) => record.clone(), + _ => panic!("expected strand drop fixture"), + }; + drop.child_worldline_id = worldline(99); + + let obstruction = RecoveredTopologyIndex::from_topology_records([ + TopologyIntentRecord::StrandFork(fork), + TopologyIntentRecord::StrandDrop(drop), + ]) + .expect_err("fork/drop child mismatch obstructs"); + + assert!(matches!( + obstruction, + WalRecoveryIndexError::ConflictingStrandDrop { .. } + )); +} + +#[test] +fn topology_braid_event_records_must_be_self_consistent() { + let braid_event = match &topology_records()[2] { + TopologyIntentRecord::BraidEvent(record) => record.clone(), + _ => panic!("expected braid event fixture"), + }; + let mut mismatched_braid = braid_event.clone(); + mismatched_braid.event = BraidEvent::BraidCreated { + braid_id: digest("topology:other-braid"), + creator_domain: authority(9), + }; + let mismatch = + RecoveredTopologyIndex::from_topology_records([TopologyIntentRecord::BraidEvent( + mismatched_braid, + )]) + .expect_err("mismatched embedded braid id obstructs"); + assert!(matches!( + mismatch, + WalRecoveryIndexError::ConflictingBraidEvent { .. } + )); + + let mut impossible_status = braid_event; + impossible_status.status_after = BraidStatus::Collapsed; + let status = RecoveredTopologyIndex::from_topology_records([TopologyIntentRecord::BraidEvent( + impossible_status, + )]) + .expect_err("impossible braid status obstructs"); + assert!(matches!( + status, + WalRecoveryIndexError::ConflictingBraidEvent { .. } + )); +} + #[test] fn security_and_redaction_postures_decode_without_becoming_missing() { for posture in [ diff --git a/crates/warp-core/tests/wsc_store_tests.rs b/crates/warp-core/tests/wsc_store_tests.rs index 24b19f37..a1f512e3 100644 --- a/crates/warp-core/tests/wsc_store_tests.rs +++ b/crates/warp-core/tests/wsc_store_tests.rs @@ -15,6 +15,7 @@ use warp_core::causal_wal::{ TickReceiptRecord, TopologyBraidEventRecord, TopologyImportOutcomeKind, TopologyIntentRecord, WalReceiptCorrelationRecord, WalTickDecision, }; +use warp_core::wsc::types::{AttRow, NodeRow, Range}; use warp_core::wsc::{ accepted_submission_records_from_wsc_envelope, accepted_submission_records_from_wsc_store, accepted_submission_records_to_wsc_envelope, receipt_correlation_records_from_wsc_envelope, @@ -26,8 +27,9 @@ use warp_core::wsc::{ WscStoreEnvelope, WscStoreObstructionKind, WscStorePort, WscStoreRecordKind, WscStoreSubject, }; use warp_core::{ - make_strand_id, AuthorityDomainId, AuthorityDomainRef, BraidEvent, BraidStatus, HeadId, - OriginId, WorldlineId, WorldlineTick, WriterHeadKey, + make_node_id, make_strand_id, make_type_id, make_warp_id, AuthorityDomainId, + AuthorityDomainRef, BraidEvent, BraidStatus, HeadId, OriginId, WorldlineId, WorldlineTick, + WriterHeadKey, }; #[test] @@ -640,12 +642,100 @@ fn topology_records_reject_conflicting_duplicate_strand_fork() { let obstruction = topology_records_to_wsc_envelope(&records) .expect_err("conflicting topology duplicate obstructs"); + // This kind covers canonical topology payload conflicts as well as + // committed-store duplicate-envelope collisions. assert_eq!( obstruction.kind, WscStoreObstructionKind::DuplicateEnvelopeMismatch ); } +#[test] +fn topology_records_reject_idempotency_conflicts_for_each_record_family() { + let records = topology_records(); + + let fork = match &records[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + let mut conflicting_fork = fork.clone(); + conflicting_fork.strand_id = make_strand_id("wsc-topology-other-strand"); + conflicting_fork.child_worldline_id = worldline(52); + let obstruction = topology_records_to_wsc_envelope(&[ + TopologyIntentRecord::StrandFork(fork), + TopologyIntentRecord::StrandFork(conflicting_fork), + ]) + .expect_err("conflicting fork idempotency key obstructs"); + assert_eq!( + obstruction.kind, + WscStoreObstructionKind::DuplicateEnvelopeMismatch + ); + + let drop = match &records[1] { + TopologyIntentRecord::StrandDrop(record) => record.clone(), + _ => panic!("expected strand drop fixture"), + }; + let mut conflicting_drop = drop.clone(); + conflicting_drop.strand_id = make_strand_id("wsc-topology-other-drop"); + conflicting_drop.child_worldline_id = worldline(53); + let obstruction = topology_records_to_wsc_envelope(&[ + TopologyIntentRecord::StrandDrop(drop), + TopologyIntentRecord::StrandDrop(conflicting_drop), + ]) + .expect_err("conflicting drop idempotency key obstructs"); + assert_eq!( + obstruction.kind, + WscStoreObstructionKind::DuplicateEnvelopeMismatch + ); + + let braid_event = match &records[2] { + TopologyIntentRecord::BraidEvent(record) => record.clone(), + _ => panic!("expected braid event fixture"), + }; + let mut conflicting_braid_event = braid_event.clone(); + conflicting_braid_event.event_index = 1; + let obstruction = topology_records_to_wsc_envelope(&[ + TopologyIntentRecord::BraidEvent(braid_event), + TopologyIntentRecord::BraidEvent(conflicting_braid_event), + ]) + .expect_err("conflicting braid-event idempotency key obstructs"); + assert_eq!( + obstruction.kind, + WscStoreObstructionKind::DuplicateEnvelopeMismatch + ); + + let braid_shell = match &records[3] { + TopologyIntentRecord::BraidShell(record) => record.clone(), + _ => panic!("expected braid shell fixture"), + }; + let mut conflicting_braid_shell = braid_shell.clone(); + conflicting_braid_shell.shell_digest = [126; 32]; + conflicting_braid_shell.material_digest = [127; 32]; + let obstruction = topology_records_to_wsc_envelope(&[ + TopologyIntentRecord::BraidShell(braid_shell), + TopologyIntentRecord::BraidShell(conflicting_braid_shell), + ]) + .expect_err("conflicting braid-shell idempotency key obstructs"); + assert_eq!( + obstruction.kind, + WscStoreObstructionKind::DuplicateEnvelopeMismatch + ); +} + +#[test] +fn topology_records_reject_root_level_topology_attachment() { + let fork = match &topology_records()[0] { + TopologyIntentRecord::StrandFork(record) => record.clone(), + _ => panic!("expected strand fork fixture"), + }; + let envelope = topology_envelope_with_root_attachment(TopologyIntentRecord::StrandFork(fork)); + + let obstruction = topology_records_from_wsc_envelope(&envelope) + .expect_err("root-level topology attachment obstructs"); + + assert_eq!(obstruction.kind, WscStoreObstructionKind::InvalidWsc); +} + #[test] fn retention_records_from_committed_wsc_store_rejects_conflicting_material_digest() { let material = retained_material( @@ -798,14 +888,14 @@ fn fixture_wsc_bytes(tick: u64) -> Vec { let input = OneWarpInput { warp_id: [1; 32], root_node_id: [2; 32], - nodes: vec![warp_core::wsc::types::NodeRow { + nodes: vec![NodeRow { node_id: [2; 32], node_type: [3; 32], }], edges: vec![], - out_index: vec![warp_core::wsc::types::Range::default()], + out_index: vec![Range::default()], out_edges: vec![], - node_atts_index: vec![warp_core::wsc::types::Range::default()], + node_atts_index: vec![Range::default()], node_atts: vec![], edge_atts_index: vec![], edge_atts: vec![], @@ -814,6 +904,56 @@ fn fixture_wsc_bytes(tick: u64) -> Vec { write_wsc_one_warp(&input, [8; 32], tick).expect("fixture WSC bytes") } +fn topology_envelope_with_root_attachment(record: TopologyIntentRecord) -> WscStoreEnvelope { + let canonical = topology_records_to_wsc_envelope(&[record.clone()]) + .expect("canonical topology WSC envelope"); + let payload = record.to_payload_bytes(); + let root = make_node_id("echo/wsc-store/topology/root"); + let input = OneWarpInput { + warp_id: make_warp_id("echo/wsc-store/topology").0, + root_node_id: root.0, + nodes: vec![NodeRow { + node_id: root.0, + node_type: make_type_id("echo/wsc-store/topology/node/v1").0, + }], + edges: vec![], + out_index: vec![Range::default()], + out_edges: vec![], + node_atts_index: vec![Range { + start_le: 0u64.to_le(), + len_le: 1u64.to_le(), + }], + node_atts: vec![AttRow { + tag: AttRow::TAG_ATOM, + reserved0: [0; 7], + type_or_warp: make_type_id(topology_attachment_type(&record)).0, + blob_off_le: 0u64.to_le(), + blob_len_le: (payload.len() as u64).to_le(), + }], + edge_atts_index: vec![], + edge_atts: vec![], + blobs: payload, + }; + let wsc_bytes = write_wsc_one_warp(&input, make_type_id("echo/wsc-store/topology/v1").0, 0) + .expect("root-attachment topology WSC bytes"); + WscStoreEnvelope::validated( + WscStoreRecordKind::CausalHistory, + *canonical.basis_digest(), + wsc_bytes, + ) + .expect("root-attachment topology WSC envelope") +} + +fn topology_attachment_type(record: &TopologyIntentRecord) -> &'static str { + match record { + TopologyIntentRecord::StrandFork(_) => "echo/wsc-store/topology/strand-fork/v1", + TopologyIntentRecord::StrandDrop(_) => "echo/wsc-store/topology/strand-drop/v1", + TopologyIntentRecord::BraidEvent(_) => "echo/wsc-store/topology/braid-event/v1", + TopologyIntentRecord::BraidShell(_) => "echo/wsc-store/topology/braid-shell/v1", + TopologyIntentRecord::SuffixImport(_) => "echo/wsc-store/topology/suffix-import/v1", + } +} + fn submission_acceptance(submission_byte: u8, envelope_byte: u8) -> SubmissionAcceptanceRecord { SubmissionAcceptanceRecord { submission_id: [submission_byte; 32], diff --git a/docs/topics/WAL.md b/docs/topics/WAL.md index 21b77679..799ea7a5 100644 --- a/docs/topics/WAL.md +++ b/docs/topics/WAL.md @@ -60,9 +60,10 @@ The useful postures are: | Posture | Meaning | | --- | --- | | `not_accepted` | The intent never reached WAL-backed accepted submission posture. | -| `accepted_pending` | The intent was accepted and can be recovered, but no decided receipt was recovered. | +| `accepted_pending` | Accepted-submission evidence was recovered, but no decided receipt was recovered. | | `decided_applied` | A recovered receipt says the work applied under named law. | -| `decided_rejected` | A recovered receipt says the work was rejected, conflicted, or obstructed. | +| `decided_rejected` | A recovered receipt says the work was rejected or conflicted. | +| `obstructed` | Recovery found accepted or decided evidence, but required material or consistency checks obstruct restoring the work. | | `recovery_faulted` | Required committed WAL evidence or retained material is missing or corrupt. | An app such as `jedit` maps these generic postures into product language @@ -73,10 +74,13 @@ order to explain them. For Jim/jedit, "dirty" should not mean "at risk of being lost." It should mean "not currently materialized to the host file projection." If an edit intent has -received WAL-backed accepted-submission evidence, the edit belongs to -recoverable causal history even before the host file is written. If that edit is -later decided, the receipt and recovered reading are the source of truth for -restoring the editor view. +received WAL-backed accepted-submission evidence, the edit's identity and +posture belong to recoverable causal history even before the host file is +written. Restoring the editor text still requires retained material, a decided +receipt, or a recovered reading that carries the content needed for the view. If +that content is unavailable, the editor must keep the materialization warning and +surface the recovered obstruction instead of claiming the buffer can be rebuilt +from an ACK alone. The correct safety gate is therefore not a classic dirty-buffer warning on quit or file switch. The correct gate is whether the edit crossed the Echo