Fix critical resolution and recovery defects; prepare v0.2.0

.github/workflows/ci.yml

@@ -4,6 +4,10 @@ on:
   push:
     branches: [main]
   pull_request:
+  workflow_call:
+
+permissions:
+  contents: read
 
 jobs:
   msrv:
@@ -21,6 +25,8 @@ jobs:
         uses: dtolnay/rust-toolchain@1.88.0
       - name: Cache Rust
         uses: Swatinem/rust-cache@v2
+      - name: Install native build dependencies
+        run: sudo apt-get update && sudo apt-get install -y libclang-dev cmake
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -46,6 +52,8 @@ jobs:
           components: rustfmt, clippy
       - name: Cache Rust
         uses: Swatinem/rust-cache@v2
+      - name: Install native build dependencies
+        run: sudo apt-get update && sudo apt-get install -y libclang-dev cmake
       - name: Install protoc
         uses: arduino/setup-protoc@v3
         with:
@@ -58,9 +66,13 @@ jobs:
         run: cargo fmt --check
       - name: Clippy
         run: cargo clippy --locked --all-targets --all-features -- -D warnings
+      - name: Verify release package
+        run: cargo package --locked
       - name: Validate deployment manifests
         run: |
-          bash -n scripts/*.sh
+          for script in scripts/*.sh; do
+            bash -n "$script"
+          done
           docker compose -f compose.yaml config --quiet
 
   container-image:

.github/workflows/release.yml

@@ -8,26 +8,35 @@ on:
 permissions:
   contents: write
 
+concurrency:
+  group: release-${{ github.ref }}
+  cancel-in-progress: false
+
 jobs:
+  validate:
+    uses: ./.github/workflows/ci.yml
+
   publish:
+    needs: validate
     runs-on: ubuntu-latest
     steps:
+      - name: Free disk space
+        run: |
+          sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc /opt/hostedtoolcache/CodeQL
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
       - uses: Swatinem/rust-cache@v2
+      - name: Install native build dependencies
+        run: sudo apt-get update && sudo apt-get install -y libclang-dev cmake
       - uses: arduino/setup-protoc@v3
         with:
           version: "24.x"
-      - name: cargo test
-        run: cargo test --locked --all-features
-      - name: cargo fmt
-        run: cargo fmt --check
-      - name: cargo clippy
-        run: cargo clippy --locked --all-targets --all-features -- -D warnings
+      - name: Verify tag matches package version
+        run: |
+          package_version="$(cargo metadata --locked --no-deps --format-version 1 | python3 -c 'import json,sys; data=json.load(sys.stdin); print(next(p["version"] for p in data["packages"] if p["name"] == "unirust-rs"))')"
+          test "$GITHUB_REF_NAME" = "v$package_version"
       - name: cargo package
         run: cargo package --locked
-      - name: Build production image
-        run: docker build --file Containerfile --tag unirust-release .
       - name: cargo publish
         run: cargo publish --locked
         env:

CHANGELOG.md

@@ -3,10 +3,38 @@
 All notable changes to this project are documented here. This project follows
 Semantic Versioning.
 
-## [Unreleased]
+## [0.2.0] - 2026-09-05
+
+### Added
+
+- Persistent distributed shards, streaming ingest, cross-shard reconciliation,
+  coordinated checkpoints, verified off-host backups, and synchronous replication.
+- Mutual TLS transport, semantic readiness checks, and fail-closed recovery for
+  partial reconciliation, interrupted ingestion, and inconsistent restores.
+- Enabled persistent regressions for the September correctness audit and query
+  equivalence across bridge merges, cache invalidation, and restart.
 
 ### Fixed
 
+- Transitive cross-shard merges now compare component-wide temporal strong IDs,
+  including observations from previous reconciliation rounds.
+- Distributed queries assemble complete canonical entities before temporal
+  conjunction and golden attribute mastering.
+- Occupied record IDs cannot overwrite durable records; rejected batches discard
+  staged records and partial resolution state before another request can commit.
+- Bounded interner caches preserve durable IDs after restart. Corrupt query
+  lookups fail explicitly instead of returning empty matches.
+- Repeated common keys and the memory-saver profile continue entity resolution.
+  Tiered index updates preserve complete buckets through eviction and recovery,
+  respect capacity settings, and propagate storage failures.
+- Persistent DSU cluster enumeration, deferred membership updates, and restored
+  query-label cache invalidation now use authoritative cluster state.
+- Unbounded temporal intervals no longer overflow duration calculations or
+  require enumerating every temporal bucket; Allen relations preserve orientation.
+- Rust 1.98 Clippy failures are resolved. Release publishing now depends on full
+  CI validation and a matching tag/package version, including package and image
+  builds. Shell validation checks every deployment script.
+
 - Ingest acknowledgement now waits for a synchronous RocksDB WAL flush. The
   external ingest WAL is removed only after records and entity-resolution state
   have reached stable storage, closing a power-loss window for acknowledged data.
@@ -51,7 +79,14 @@ Semantic Versioning.
 - Adopted Rust 2024 with a documented MSRV of Rust 1.88.
 - Updated the active gRPC, terminal UI, cache, and compression dependencies and
   removed unused direct HTTP, TOML, and protobuf-types dependencies.
-- The package version is now `0.2.0` in preparation for the next release.
+- The live distributed protocol is now 6. Upgrade routers, shards, and replicas
+  together; WAL/checkpoint version 1 and durable reservation format 5 remain
+  readable. Public router clients continue using the existing request contract.
+- Selective queries reuse resolved membership and master only candidate entities;
+  composite labels retain global prefix collision semantics. Shard requests run
+  concurrently with bounded fan-out and entity hydration batches.
+- Persistent ingestion lends staged records directly to parallel extraction,
+  avoiding record cloning and correctly resolving batches above 100,000 records.
 - The application lockfile is tracked for reproducible binary and container
   builds.
 - Distributed integration tests now use temporary persistent shard stores; the

README.md

@@ -281,6 +281,17 @@ duplicates make startup fail closed for operator repair. The internal router and
 shard protocol is versioned, and mixed versions are rejected, so upgrades must
 replace the full cluster with one coordinated Unirust version before restarting
 the router. Shard gRPC ports are internal APIs; client ingest must use the router.
+
+The current live protocol is **6**. Protocol 5 routers, shards, and replicas must
+be stopped and upgraded together: older coordinators do not implement the
+component guards or canonical entity queries required by this release. This
+handshake change does not change WAL or checkpoint format version 1, record
+snapshot encoding, or durable source-reservation format version 5. Existing
+volumes, pending WAL batches, and coordinated checkpoints remain readable;
+valid version 5 reservation markers retain their original shard-count binding
+without another backfill. Internal ingest/import callers must send the current
+live protocol, while public router clients continue leaving that field unset.
+
 The persisted reservation directory is also bound to the configured shard count.
 Router startup rejects shard-count changes because the current import API cannot
 atomically move a record, update its reservation, and delete the old copy.

benches/bench_distributed.rs

@@ -410,12 +410,6 @@ fn bench_shardnode_ingest(c: &mut Criterion) {
     group.warm_up_time(Duration::from_secs(1));
     group.measurement_time(Duration::from_secs(6));
 
-    // SAFETY: benchmark configuration runs before this function creates any worker threads.
-    unsafe { std::env::set_var("UNIRUST_PARTITIONED", "1") };
-    let partition_count = env_usize("UNIRUST_DIST_PARTITIONS", 8);
-    // SAFETY: benchmark configuration runs before this function creates any worker threads.
-    unsafe { std::env::set_var("UNIRUST_PARTITION_COUNT", partition_count.to_string()) };
-
     let count = env_u32("UNIRUST_DIST_RECORDS", 50_000);
     let overlap = env_f64("UNIRUST_DIST_OVERLAP", 0.05);
     let config = default_dist_config();
@@ -424,18 +418,28 @@ fn bench_shardnode_ingest(c: &mut Criterion) {
     group.bench_function(BenchmarkId::new("batch", count), |b| {
         b.iter_batched(
             || {
+                let data_dir = tempfile::tempdir().expect("persistent benchmark directory");
                 let rt = Runtime::new().expect("runtime");
                 let shard = rt.block_on(async {
-                    ShardNode::new(0, config.clone(), StreamingTuning::default()).expect("shard")
+                    ShardNode::new_with_data_dir(
+                        0,
+                        config.clone(),
+                        StreamingTuning::default(),
+                        Some(data_dir.path().to_path_buf()),
+                        false,
+                        None,
+                    )
+                    .expect("persistent shard")
                 });
                 let records = generate_proto_batch(1, count, overlap, 45);
-                (rt, shard, records)
+                (rt, shard, records, data_dir)
             },
-            |(rt, shard, records)| {
+            |(rt, shard, records, _data_dir)| {
                 let response = rt.block_on(async {
                     shard
                         .ingest_records(Request::new(proto::IngestRecordsRequest {
-                            internal_protocol_version: 2,
+                            internal_protocol_version:
+                                unirust_rs::distributed::DISTRIBUTED_PROTOCOL_VERSION,
                             records,
                         }))
                         .await
@@ -457,12 +461,6 @@ fn bench_shardnode_streaming_simulated(c: &mut Criterion) {
     group.warm_up_time(Duration::from_secs(1));
     group.measurement_time(Duration::from_secs(6));
 
-    // SAFETY: benchmark configuration runs before this function creates any worker threads.
-    unsafe { std::env::set_var("UNIRUST_PARTITIONED", "1") };
-    let partition_count = env_usize("UNIRUST_DIST_PARTITIONS", 8);
-    // SAFETY: benchmark configuration runs before this function creates any worker threads.
-    unsafe { std::env::set_var("UNIRUST_PARTITION_COUNT", partition_count.to_string()) };
-
     let total = env_u32("UNIRUST_DIST_STREAM_TOTAL", 20_000);
     let chunk = env_u32("UNIRUST_DIST_STREAM_CHUNK", 512).max(1);
     let overlap = env_f64("UNIRUST_DIST_OVERLAP", 0.05);
@@ -472,14 +470,23 @@ fn bench_shardnode_streaming_simulated(c: &mut Criterion) {
     group.bench_function(BenchmarkId::new("chunked", chunk), |b| {
         b.iter_batched(
             || {
+                let data_dir = tempfile::tempdir().expect("persistent benchmark directory");
                 let rt = Runtime::new().expect("runtime");
                 let shard = rt.block_on(async {
-                    ShardNode::new(0, config.clone(), StreamingTuning::default()).expect("shard")
+                    ShardNode::new_with_data_dir(
+                        0,
+                        config.clone(),
+                        StreamingTuning::default(),
+                        Some(data_dir.path().to_path_buf()),
+                        false,
+                        None,
+                    )
+                    .expect("persistent shard")
                 });
                 let records = generate_proto_batch(1, total, overlap, 46);
-                (rt, shard, records)
+                (rt, shard, records, data_dir)
             },
-            |(rt, shard, records)| {
+            |(rt, shard, records, _data_dir)| {
                 let response = rt.block_on(async {
                     let mut offset = 0usize;
                     let mut assignments = Vec::new();
@@ -488,7 +495,8 @@ fn bench_shardnode_streaming_simulated(c: &mut Criterion) {
                         let batch = records[offset..end].to_vec();
                         let resp = shard
                             .ingest_records(Request::new(proto::IngestRecordsRequest {
-                                internal_protocol_version: 2,
+                                internal_protocol_version:
+                                    unirust_rs::distributed::DISTRIBUTED_PROTOCOL_VERSION,
                                 records: batch,
                             }))
                             .await

docs/critical-audit-2026-09-05.md

@@ -0,0 +1,62 @@
+Audit and repairs based on merged commit `3d99fe0d27d7344b08b3bd5b23bb05c75479f1bd`, September 5, 2026.
+
+Three parallel reviews covered entity resolution, persistence/recovery, and distributed correctness. All eight original defects were reproduced with persistent storage and have enabled regression coverage. Every accepted record still completes entity resolution before its record data commits.
+
+| Defect and original trigger | Implemented repair | Regression suite |
+| --- | --- | --- |
+| A—B—C cross-shard identity chain merged incompatible strong IDs on A and C. | Check accumulated component observations before every union, across metadata chunks and prior reconciliation rounds; hydrate authoritative remote fragments. | `distributed_entity_regressions` |
+| Reusing occupied `RecordId(42)` overwrote another durable record. | Reject occupied IDs in ordinary, explicit, batch, and staged storage APIs while preserving valid source-identity retries. | `durable_ingest_regressions` |
+| A rejected batch left an orphan staged; a later request committed it without resolving it. | Discard aborted staging and rebuild partial derived state. Commit record data only after linking succeeds. Recover partial assignment writes from resolved durable records. | `durable_ingest_regressions` |
+| Bounded interner caches changed durable values to `unknown` after restart. | Hydrate durable descriptor IDs before preparation; query lookup uses durable reverse indexes without allocating strings in the store. | `durable_ingest_regressions` |
+| 300 identical observations permanently stopped later matching. | Avoid redundant interval-tree nodes; exhausted scan budgets use exact candidate fallback rather than permanently tainting a key. | `resolution_capacity_regressions` |
+| Memory-saver left two matching source systems unmerged. | Respect disabled adaptive limits and use a positive fallback cap. | `resolution_capacity_regressions` |
+| Cross-shard conjunction missed entities and returned incomplete golden data. | Discover candidate IDs, hydrate all fragments, then evaluate temporal conjunction and master conflicting attribute values globally. | `distributed_entity_regressions` |
+| Cached-key insertion ignored hot/warm capacity limits. | Run amortized tier maintenance, persist complete buckets before eviction, promote complete buckets before updates, and propagate I/O/decode errors. | `resolution_capacity_regressions` |
+
+Additional regression coverage includes all four ingest APIs, a 100,001-record batch, an aborted first ingest that already flushed persistent DSU nodes, partial durable assignment failure, tier eviction/update/restart, corrupt cold buckets, deferred merge membership, temporal adjacency, and local query equivalence before/after restart. Authoritative membership also fixes the previously unsupported cluster enumeration with persistent DSU.
+
+Query execution reuses the linker's authoritative membership and masters golden data only for indexed candidate entities. Single-attribute labels are candidate-local. Composite labels retain a separate global key cache because their shortest unique prefixes depend on other entities; rebuilding that metadata after ingestion still requires a global pass. An unopened streaming engine retains the existing recovery/cache path.
+
+Router queries dispatch up to 16 shard requests concurrently and hydrate at most 1,000 entity IDs per request. A barrier-based regression proves that both query phases dispatch concurrently. Strong-ID-only reconciliation hydration skips golden data and label construction. Persistent staging lends records directly to parallel extraction, avoiding an extra record clone and the old bounded-cache cutoff.
+
+Baseline release-mode selective-query measurements used `PersistentStore`, batches of 1,000 distinct emails, and one matching entity:
+
+| Stored records | First query before repair | Warm query before repair | After one unrelated insert before repair |
+| ---: | ---: | ---: | ---: |
+| 5,000 | 19.454 ms | 0.002 ms | 15.732 ms |
+| 20,000 | 66.230 ms | 0.006 ms | 61.903 ms |
+| 80,000 | 306.473 ms | 0.003 ms | 300.151 ms |
+
+After repair, the same release-mode diagnostic measured:
+
+| Stored records | First query after repair | Warm query after repair | After one unrelated insert after repair |
+| ---: | ---: | ---: | ---: |
+| 5,000 | 0.055 ms | 0.006 ms | 0.010 ms |
+| 20,000 | 0.022 ms | 0.005 ms | 0.007 ms |
+| 80,000 | 0.029 ms | 0.005 ms | 0.008 ms |
+
+These timings are local diagnostics, not a universal latency guarantee; warm-cache microsecond differences are below meaningful precision for this single-run measurement. The standard historical 410K records/second figure was not reproduced on this machine; comparative throughput uses the same five persistent shards and workload for both revisions.
+
+The comparative ingest run used five persistent shards with the high-throughput profile, 1,000,000 records, 16 streams, batches of 5,000, 10% overlap, and seed 42:
+
+| Revision | Acknowledged records | Throughput | Average RPC latency | Failed batches |
+| --- | ---: | ---: | ---: | ---: |
+| Merged baseline | 1,000,000 | 43,415 records/s | 1,585 ms | 0 |
+| Audit repairs | 1,000,000 | 43,939 records/s | 1,657 ms | 0 |
+
+Ingest throughput was effectively unchanged in this single comparison; the small throughput and latency differences do not establish a statistically significant change. This workload measures durable ingest, not a universal cross-shard query or reconciliation latency. The selective-query improvement is measured separately above.
+
+CI now passes the Rust 1.98 lints that failed on main (`chunks_exact_to_as_chunks` and `manual_slice_fill`). PR validation explicitly installs native build dependencies and verifies the release package. Release publishing depends on the reusable complete CI workflow, including MSRV, tests, lint, manifests, and production image, and rejects tags that do not match the package version. Workflow syntax is checked with actionlint.
+
+Validation commands:
+
+```sh
+cargo test --locked --all-features
+cargo +1.98.0 clippy --locked --all-targets --all-features -- -D warnings
+cargo +1.88.0 check --locked --all-targets --all-features
+cargo fmt --check
+cargo package --locked
+cargo test --release --test audit_performance -- --ignored --nocapture
+```
+
+The correctness suites run by default. Only the manual timing diagnostic is ignored; it has no machine-dependent timing assertion. Distributed benchmarks now include persistent shard ingestion rather than treating the in-memory partition path as evidence of production throughput.