feat(composite): count, report, durably record and repair failed secondary syncs (#1334)

[smunini]

Summary

Implements the decision recorded on #1334: a composite write keeps succeeding in every HFS_COMPOSITE_SYNC_MODE (the primary is the system of record and has already committed), and a secondary sync that fails for good is now

1. counted — composite_secondary_sync_failures_total{backend,operation} plus a composite_secondary_sync_needs_reindex gauge,
2. emitted as a structured event — one ERROR with tenant, resource_type, id, version, backend_id, operation, attempts, recorded, error; never resource content,
3. recorded durably as "needs reindex" in the primary, de-duplicated per (tenant, type, id, backend), cleared by a later successful sync of the same resource, and drained by a repair function that a periodic server task calls.

It also fixes SyncManager burning every retry on a secondary's NotFound for a Delete event.

Fixes #1334

Observed pre-fix behaviour (reproduced first)

A failing-secondary stub (a SQLite index behind an on/off switch), using only APIs that exist on the base branch, run against the base's composite/{storage,sync}.rs. Output (the stub's max_retries is 2, so 3 = every attempt spent):

REPRO Synchronous: create=true update=true delete=true
REPRO Synchronous: LOST searchable=0 ; deleted KEPT still indexed=true
REPRO Synchronous: delete calls for one Delete+NotFound event = 3
REPRO Asynchronous: create=true update=true delete=true
REPRO Asynchronous: LOST searchable=0 ; deleted KEPT still indexed=true
REPRO Asynchronous: delete calls for one Delete+NotFound event = 3
REPRO Hybrid { sync_for_search: true }: create=true update=true delete=true
REPRO Hybrid { sync_for_search: true }: LOST searchable=0 ; deleted KEPT still indexed=true
REPRO Hybrid { sync_for_search: true }: delete calls for one Delete+NotFound event = 3
REPRO Hybrid { sync_for_search: false }: create=true update=true delete=true
REPRO Hybrid { sync_for_search: false }: LOST searchable=0 ; deleted KEPT still indexed=true
REPRO Hybrid { sync_for_search: false }: delete calls for one Delete+NotFound event = 3
test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 1.44s

So, in every mode: the create is unsearchable, the update is stale, the deleted resource is still served by search, all three writes returned Ok, and a Delete the secondary answers NotFound to is retried to exhaustion. The issue's description holds, with one correction: it says "the only trace is a warn! log line". In the synchronous modes there was not even that for the final failure — see the table.

Where a write's secondary sync goes (pre-fix)

All of create, update, delete, delete_versioned, create_or_update, conditional_create/update/delete/patch, update_with_match, delete_with_match end in CompositeStorage::sync_to_secondaries(event) → SyncManager::sync.

Mode	Where the secondary call happens	Retried	Where the final result went	Client
synchronous	inline in the request; all secondaries in parallel (JoinSet)	sync_event_to_backend: 1 + 3 attempts, 100 ms ×2 back-off (on top of the ES backend's own 3 since #1397)	SyncStatus { success: false, error } + BackendSyncStatus::total_errors; the Vec<SyncStatus> was dropped by sync_to_secondaries, whose warn!("Failed to sync … to secondaries") only fires for a closed queue. Only the per-attempt warn!("Sync attempt failed, retrying") (no resource identity) was logged; the final failure never	success, after waiting out the retries
hybrid	HFS maps hybrid to Hybrid { sync_for_search: true }, and every Create/Update/Delete counts as search-related, so single writes behave exactly as synchronous	same	same	same
asynchronous (default)	enqueued on an mpsc(1000); one worker drains it sequentially	same, in the worker	error!("Async sync failed", backend, error) — no tenant, type, id or version — + total_errors	success immediately (unless the queue is full; see "Found, not fixed")

Batch paths go through SyncManager::sync_creates instead: create_many (conformance seeding, bulk loads), transaction/batch Bundles (sync_bundle_results) and bulk-submit's sync_ingested. In the synchronous modes each secondary gets one create_many, then every rejected item is retried singly (1 + 3); in asynchronous mode one event per resource is queued. create_many and the Bundle path discarded the statuses; bulk-submit turns rejected ids into processing-error entry results (#1007) and still does.

Root cause

The outcome of a secondary sync was produced (SyncStatus) but had no consumer: the synchronous path threw it away, the asynchronous worker had nowhere to put it but an anonymous log line, and nothing persisted which resource was affected.

Fix

• composite/sync_failures.rs (new): SyncFailureRecorder, owned by the SyncManager because that is the one place every outcome passes through — the synchronous paths, the asynchronous worker (which finishes long after the write returned) and the batch path. On a final failure it notifies a SecondarySyncObserver (metrics), emits the structured event and upserts a record in a SecondarySyncFailureLedger. On a success it clears the resource's record — free on the healthy path: a process-local set of outstanding keys (hydrated from the ledger on first use, capped at 50 000, beyond which it degrades to always issuing the indexed delete) means the ledger is only touched while something is owed.
• sync_to_secondaries now reads the statuses (a debug! tying the failure to the request's span; the recorder has already counted/logged/recorded it, so it is not logged twice).
• Durable store — decided with evidence. sqlite-es: the automatic rebuild after an import fails on oversized _bulk requests, leaves all 11,704 Provenance resources unsearchable (1 % cut), spends 1,551 s on its first attempt alone before retrying all 24 types, and writes a SQLite index nothing reads #1125's ledger (DeferredReindexLedger) is not a general persistence abstraction: it is a single rebuild_finished callback whose storage is a marker column on the bulk-submit manifest row (clear_manifest_index_pending). There is nothing per-resource to extend. What I reused is its pattern: a small trait in persistence, implemented by the primary backends, handed to the component as Arc<dyn …> from main.rs. Implemented for SQLite (table secondary_sync_failures, schema v33→v34), PostgreSQL (same table, v40→v41) and MongoDB (collection keyed by a compound _id, so no index and no schema-version step). S3 has no ledger and degrades to metric + event (recorded=false on the event, and an info! at startup says so).
• Repair: CompositeStorage::repair_secondary_sync_failures(limit) — pushes the primary's current state of each recorded resource (create_or_update; delete if the primary no longer has it, NotFound = done), one attempt per record per pass, least-recently-failed first (a record that fails again is re-stamped and goes to the back, so it cannot starve a bounded batch). Safe alongside writes: after pushing it re-reads the primary and, if the version moved, pushes again (≤ 3 rounds) so it does not leave an older version behind a concurrent write's own sync; a resource that keeps changing through all three rounds keeps its record for the next pass. Records naming a backend the composite no longer has are dropped with a warn!.
• Wiring (crates/hfs/src/main.rs): the four *-es composites get the metrics observer; sqlite/pg/mongo also get the ledger. A periodic task — same tokio::spawn + sleep shape as the existing search-parameter refresh tasks — calls the repair (HFS_COMPOSITE_SYNC_REPAIR_INTERVAL, default 60 s, 0 = off; HFS_COMPOSITE_SYNC_REPAIR_BATCH, default 100), first pass at startup so records from an earlier run are picked up. $reindex is not hooked directly (search/reindex.rs is outside this PR's ownership): it rebuilds the index, and the next repair pass then finds the records in sync and clears them.
• Metrics (crates/observability/src/composite_metrics.rs): named like the crate's other metrics — no hfs_ prefix (nothing in the repo has one; the exporter's global service="hfs" label does that job), so composite_secondary_sync_failures_total rather than the issue's suggested hfs_composite_…. Labels: backend (configured id) and operation only.
• Unavailable detail: BackendError::Unavailable's Display omits its message, and since fix(persistence): retry and stop discarding Elasticsearch write failures (#1382) #1397 that is exactly what an exhausted ES write returns — every record would have read "backend unavailable: elasticsearch". The recorder keeps the detail (truncated to 1024 chars).
• Delete + NotFound/Gone on a secondary is success in sync_event_to_backend.

Behaviour changes clients / operators can observe

• Clients: none. Status codes, bodies and (in asynchronous mode) latency are unchanged. In synchronous mode a failed sync additionally costs one upsert on the primary; a successful one costs nothing unless records are outstanding.
• Operators: two new metrics; the final-failure log line is now ERROR "Secondary sync failed; …" with structured fields (the asynchronous worker's anonymous "Async sync failed" is replaced by it; synchronous modes gain it); a new table/collection secondary_sync_failures in the primary (SQLite schema 34, PostgreSQL schema 41); a background repair task with two new env vars; deletes of resources the index never had no longer spend 3 retries.

Verification

Ran and passed (branch head, base unchanged since branching; CARGO_BUILD_JOBS=4):

• cargo test -p helios-persistence --features sqlite,postgres,mongodb,elasticsearch,s3 --no-fail-fast --lib --test composite_secondary_sync_failures --test composite_read_your_writes --test composite_polyglot_tests --test composite_conditional_capabilities --test composite_routing_tests --test composite_conformance_sync --test composite_index_during_ingest --test sqlite_tests → lib 1911, new suite 9, 5 / 14 / 3 / 25 / 4 / 2, sqlite_tests 98; 0 failed.
• cargo test -p helios-persistence --features sqlite,postgres,mongodb --no-fail-fast --test postgres_tests --test mongodb_tests --test composite_secondary_sync_failures → 274 / 212 / 9, 0 failed (includes the new ledger contract on real PostgreSQL and MongoDB containers).
• cargo test -p helios-observability --lib composite_metrics → 2 passed.
• cargo test -p helios-rest --no-fail-fast → 1720 passed, 0 failed. (A first attempt died with No space left on device; the disk had been filled by other work on the machine. Re-run clean.)
• cargo +1.98.1 clippy with the CI allow-list, all rc=0: -p helios-persistence --features sqlite,postgres,mongodb,elasticsearch,s3 --all-targets; -p helios-observability --all-targets; -p helios-hfs --features sqlite,postgres,mongodb,elasticsearch,s3 --all-targets; and the CI guard sets -p helios-hfs --no-default-features --features R4,sqlite,postgres / R4,postgres / R4,mongodb / R4,s3 / R4,s3,elasticsearch.
• cargo fmt -p helios-persistence -p helios-observability -p helios-hfs -- --check → clean.
• cargo +1.98.1 test --workspace --all-features --no-run --exclude pysof → did not complete: twice killed by No space left on device (the link step's target/ reached 432 GB on a disk shared with other builds; the only errors in the log are disk errors). In its place, cargo +1.98.1 check --workspace --all-features --all-targets --exclude pysof (same feature set and targets, type-checked but not linked): rc=0. No cfg(all(feature ...))-gated code was added by this PR.
• Manual run (hfs sqlite-es, HFS_COMPOSITE_SYNC_MODE=synchronous, repair interval 5 s, real Elasticsearch 8.15 container): control create + search OK → docker stop ES → POST /Patient = 201 → /metrics showed composite_secondary_sync_failures_total{service="hfs",backend="es",operation="create"} 1 and composite_secondary_sync_needs_reindex 1; the ERROR event carried tenant=default resource_type=Patient id=… backend_id=es operation="create" attempts=4 recorded=true error=backend unavailable: elasticsearch: Failed to check index existence …; repair passes logged still_failing=1 while ES was down → docker start ES → next pass repaired=1 remaining=0, gauge back to 0, counter still 1, and GET /Patient?family=Lost1334 found the resource.

What the new tests cover (tests/composite_secondary_sync_failures.rs, failing-secondary stub × Synchronous / Asynchronous / Hybrid{true} / Hybrid{false}, asynchronous outcomes awaited through the sync queue barrier, no sleeps or timing assertions): positive control (healthy secondary indexes, nothing counted); create/update/delete against a down secondary all succeed; the observer is called exactly once per final failure with the right (backend, operation); one record per resource, a repeat failure folds (attempts adds up, first_failed_at kept, operation replaced); gauge tracks; a later successful write clears the record; the repair deletes what the primary deleted and creates what the secondary never received; a still-down secondary leaves the record and does not count a write failure; bounded batch + rotation; records survive re-opening the primary's database file; a record from an earlier run is cleared by the first successful write in the next; Delete + NotFound is one call and success; create_many records per resource; no ledger (S3 shape) still counts.

Not run: elasticsearch_tests, s3_tests/minio_s3_tests (no code they exercise changed beyond SyncManager, which the composite suites and --lib cover), the Playwright UI ring, and nothing on Windows.

Found, not fixed

• Transaction/batch Bundle DELETE entries are never synced to secondaries (code reading, not reproduced): sync_bundle_results only looks at entries that carry a resource body, and a delete has none, so the deleted resource stays in the search index. It also sends Bundle updates through sync_creates → create_many; harmless on Elasticsearch (index = upsert), wrong for a secondary whose create rejects an existing id.
• Index-during-ingest (ingest_index_sink.rs) bypasses SyncManager, so its rejections are reported as processing-error entry results but are not in the new ledger or counter.
• Asynchronous mode back-pressure: one worker drains the queue sequentially and a down secondary costs ≥ 0.7 s per event (more with fix(persistence): retry and stop discarding Elasticsearch write failures (#1382) #1397's inner retries), so the 1000-slot queue fills and sender.send().await then blocks client writes. Pre-existing; this PR adds one primary upsert per failed event to the worker and nothing to the client path.
• A Create event answered AlreadyExists by a non-upsert secondary (a lost response on the first attempt) is still retried to exhaustion and then recorded; the repair fixes it (create_or_update), but it should not fail in the first place.
• HealthMonitor is still not instantiated by the server, and CompositeStorage::update_health is still never told about secondary outcomes.
• No REST/admin surface lists the records (REST routing is owned by System-level search is not routed: GET /?_type=… returns 405, POST /_search returns 404 #1338); operators query the table.
• Tenant purge does not delete the tenant's ledger rows; the next repair pass finds the primary empty, deletes from the secondary (a no-op) and clears them.

Stacking

Stacked on #1422 (feat/1406-shared-conditional-patch, which contains #1421 and #1419) because those PRs edit the same composite write paths; base is that branch, not main. Retarget to main once the chain merges. This PR does not touch conditional_patch / delete_versioned / conditional code, the ES backend, or REST routing.