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

.claude/skills/run-hfs-server/SKILL.md

@@ -81,6 +81,22 @@ HFS_SERVER_PORT=3000 HFS_LOG_LEVEL=debug cargo run --bin hfs
 
 Use `HFS_COMPOSITE_SYNC_MODE=synchronous` **and** `HFS_ELASTICSEARCH_WRITE_REFRESH=wait_for` when callers need read-your-write search semantics, such as integration tests or bulk loads that immediately search. Either alone still leaves a window: synchronous mode only guarantees the document reached Elasticsearch, and it is not searchable until the next index refresh. See `crates/persistence/README.md` (Search visibility on Elasticsearch-backed composites). Transaction conditional references (`Organization?identifier=…` in a resource body) and `If-None-Exist` creates are exempt: they make acknowledged writes visible before resolving, on any setting (#1047).
 
+### When the search index misses a write (#1334)
+
+On an ES-backed composite the primary is the system of record: a write succeeds once the primary has committed it, in **every** `HFS_COMPOSITE_SYNC_MODE` (synchronous included), even if Elasticsearch then refuses the document after the sync retries. The client sees the same `201`/`200`/`204` either way; the resource is readable by id but missing from (or stale in) search until it is re-synced. That failure is never silent:
+
+- **Metric** (`/metrics`): `composite_secondary_sync_failures_total{backend,operation}` counts final failures (once per failed write, not per retry; `backend` is the secondary's id, `es`; `operation` is `create`/`update`/`delete`). `composite_secondary_sync_needs_reindex` is the number of resources currently recorded as owed. Alert on the gauge staying above zero. No tenant, type or id labels — `/metrics` is public.
+- **Log event**: one `ERROR` "Secondary sync failed; …" per final failure with fields `tenant`, `resource_type`, `id`, `version`, `backend_id`, `operation`, `attempts`, `recorded`, `error`. No resource content.
+- **Durable record**: one row per (tenant, type, id, backend) in the primary's `secondary_sync_failures` table (SQLite, PostgreSQL) or collection (MongoDB): `operation`, `first_failed_at`, `last_failed_at`, `last_error`, `attempts`. It survives restarts. A later successful write of the same resource clears it. **S3 primaries keep no ledger**: metric and log only (`recorded=false`), repair with `$reindex`.
+- **Repair**: a background task re-syncs recorded resources from the primary's *current* state (a delete if the primary no longer has it) and clears the record; a secondary that is still down leaves the record for the next pass. It is idempotent and safe alongside writes. `$reindex` also rebuilds the index; the next pass then finds the records in sync and clears them.
+
+| Variable | Default | Description |
+|---|---|---|
+| `HFS_COMPOSITE_SYNC_REPAIR_INTERVAL` | `60` | Seconds between repair passes; `0` disables the task (the records are still written) |
+| `HFS_COMPOSITE_SYNC_REPAIR_BATCH` | `100` | Records examined per pass, least recently failed first |
+
+To list what is owed: `SELECT * FROM secondary_sync_failures ORDER BY last_failed_at;` on the primary.
+
 ## Storage Backends
 
 | Mode | Value |

crates/hfs/src/main.rs

@@ -148,6 +148,90 @@ fn composite_sync_mode_from_env() -> helios_persistence::composite::SyncMode {
     }
 }
 
+/// Forwards composite secondary sync failures to the Prometheus exporter
+/// (#1334): `composite_secondary_sync_failures_total{backend,operation}` and
+/// the `composite_secondary_sync_needs_reindex` gauge.
+#[cfg(feature = "elasticsearch")]
+struct CompositeSyncMetrics;
+
+#[cfg(feature = "elasticsearch")]
+impl helios_persistence::composite::SecondarySyncObserver for CompositeSyncMetrics {
+    fn sync_failed(
+        &self,
+        backend_id: &str,
+        operation: helios_persistence::composite::SyncOperation,
+    ) {
+        helios_observability::composite_metrics::record_secondary_sync_failure(
+            backend_id,
+            operation.as_str(),
+        );
+    }
+
+    fn needs_reindex(&self, outstanding: u64) {
+        helios_observability::composite_metrics::set_secondary_sync_needs_reindex(outstanding);
+    }
+}
+
+/// Reads a non-negative integer setting, falling back on anything else.
+#[cfg(feature = "elasticsearch")]
+fn env_u64(var: &str, default_value: u64) -> u64 {
+    match std::env::var(var) {
+        Ok(v) => v.trim().parse().unwrap_or_else(|_| {
+            tracing::warn!(variable = var, value = %v, default_value, "Not a number; using the default");
+            default_value
+        }),
+        Err(_) => default_value,
+    }
+}
+
+/// Periodically drains the composite's "needs reindex" records (#1334): each
+/// pass re-syncs the primary's current state of up to
+/// `HFS_COMPOSITE_SYNC_REPAIR_BATCH` (default 100) recorded resources to the
+/// secondary that missed them. Runs every
+/// `HFS_COMPOSITE_SYNC_REPAIR_INTERVAL` seconds (default 60; `0` disables),
+/// starting right away so records left by an earlier run are picked up.
+///
+/// A primary without a ledger (S3) has nothing to drain, so no task is
+/// spawned for it.
+#[cfg(feature = "elasticsearch")]
+fn spawn_secondary_sync_repair(
+    composite: Arc<helios_persistence::composite::CompositeStorage>,
+    has_ledger: bool,
+) {
+    if !has_ledger {
+        info!(
+            "This primary keeps no needs-reindex ledger: failed secondary syncs are counted and logged only; repair them with $reindex"
+        );
+        return;
+    }
+    let interval = env_u64("HFS_COMPOSITE_SYNC_REPAIR_INTERVAL", 60);
+    if interval == 0 {
+        info!(
+            "HFS_COMPOSITE_SYNC_REPAIR_INTERVAL=0: periodic repair of failed secondary syncs is off"
+        );
+        return;
+    }
+    let batch = env_u64("HFS_COMPOSITE_SYNC_REPAIR_BATCH", 100).max(1) as usize;
+    let interval = std::time::Duration::from_secs(interval);
+    tokio::spawn(async move {
+        loop {
+            match composite.repair_secondary_sync_failures(batch).await {
+                Ok(report) if report.examined > 0 => info!(
+                    examined = report.examined,
+                    repaired = report.repaired,
+                    still_failing = report.still_failing,
+                    dropped = report.dropped,
+                    remaining = report.remaining,
+                    "Secondary sync repair pass finished"
+                ),
+                Ok(_) => {}
+                Err(e) => tracing::warn!("Repair of failed secondary syncs could not run: {e}"),
+            }
+            tokio::time::sleep(interval).await;
+        }
+    });
+}
+
 #[cfg(feature = "elasticsearch")]
 fn es_write_refresh_from_config(
     config: &ServerConfig,
@@ -2428,6 +2512,10 @@ async fn start_sqlite_elasticsearch(
     let composite = CompositeStorage::new(composite_config, backends)?
         .with_search_providers(search_providers)
         .with_full_primary(sqlite.clone())
+        // A failed Elasticsearch sync is counted, logged, and recorded in the
+        // primary as "needs reindex" (#1334).
+        .with_sync_observer(Arc::new(CompositeSyncMetrics))
+        .with_sync_failure_ledger(sqlite.clone())
         // `$purge` must reach the Elasticsearch secondary too — purging only
         // the SQLite primary would leave the resource in the search index,
         // still searchable and still holding its content.
@@ -2441,6 +2529,7 @@ async fn start_sqlite_elasticsearch(
 
     let serve_audit_state = audit_state.clone();
     let composite = Arc::new(composite);
+    spawn_secondary_sync_repair(composite.clone(), true);
 
     // Seed through the composite: the primary's own indexing is offloaded, so
     // seeding it directly would leave the conformance resources unsearchable
@@ -2731,6 +2820,10 @@ async fn start_postgres_elasticsearch(
     let composite = CompositeStorage::new(composite_config, backends)?
         .with_search_providers(search_providers)
         .with_full_primary(pg.clone())
+        // A failed Elasticsearch sync is counted, logged, and recorded in the
+        // primary as "needs reindex" (#1334).
+        .with_sync_observer(Arc::new(CompositeSyncMetrics))
+        .with_sync_failure_ledger(pg.clone())
         // See `start_sqlite_elasticsearch`: `$purge` must reach the search
         // secondary, not just the primary.
         .with_purgable_backends(
@@ -2743,6 +2836,7 @@ async fn start_postgres_elasticsearch(
 
     let serve_audit_state = audit_state.clone();
     let composite = Arc::new(composite);
+    spawn_secondary_sync_repair(composite.clone(), true);
 
     // Seed through the composite: the primary's own indexing is offloaded, so
     // seeding it directly would leave the conformance resources unsearchable.
@@ -2950,6 +3044,10 @@ async fn start_mongodb_elasticsearch(
     let composite = CompositeStorage::new(composite_config, backends)?
         .with_search_providers(search_providers)
         .with_full_primary(mongo.clone())
+        // A failed Elasticsearch sync is counted, logged, and recorded in the
+        // primary as "needs reindex" (#1334).
+        .with_sync_observer(Arc::new(CompositeSyncMetrics))
+        .with_sync_failure_ledger(mongo.clone())
         // See `start_sqlite_elasticsearch`: `$purge` must reach the search
         // secondary, not just the primary.
         .with_purgable_backends(
@@ -2962,6 +3060,7 @@ async fn start_mongodb_elasticsearch(
 
     let serve_audit_state = audit_state.clone();
     let composite = Arc::new(composite);
+    spawn_secondary_sync_repair(composite.clone(), true);
 
     // Seed through the composite: the primary's own indexing is offloaded, so
     // seeding it directly would leave the conformance resources unsearchable.
@@ -3390,6 +3489,9 @@ async fn start_s3_elasticsearch(
     let composite = CompositeStorage::new(composite_config, backends)?
         .with_search_providers(search_providers)
         .with_full_primary(s3.clone())
+        // S3 keeps no needs-reindex ledger: a failed Elasticsearch sync is
+        // counted and logged, and repaired by `$reindex` (#1334).
+        .with_sync_observer(Arc::new(CompositeSyncMetrics))
         // See `start_sqlite_elasticsearch`: `$purge` must reach the search
         // secondary, not just the primary.
         .with_purgable_backends(
@@ -3402,6 +3504,7 @@ async fn start_s3_elasticsearch(
 
     let serve_audit_state = audit_state.clone();
     let composite = Arc::new(composite);
+    spawn_secondary_sync_repair(composite.clone(), false);
 
     // Seed through the composite so the conformance resources land in the S3
     // primary and get indexed into Elasticsearch — the only search index here.

crates/observability/src/composite_metrics.rs

@@ -0,0 +1,101 @@
+//! Process-level Prometheus metrics for composite storage's secondary sync
+//! (#1334).
+//!
+//! A composite write succeeds once the primary has committed it; a secondary
+//! (the Elasticsearch search index) that then refuses the change after retries
+//! leaves the resource missing from search. These metrics are what an operator
+//! alerts on: a counter of such failures, and a gauge of resources still
+//! recorded as needing a reindex.
+//!
+//! ## Tenant privacy and cardinality
+//!
+//! `/metrics` is public (see [`crate::metrics`]). The only labels are
+//! `backend` — the secondary's configured id, a handful of values fixed at
+//! startup — and `operation`, over a fixed set. No tenant, resource type or
+//! resource id: those are unbounded, and they identify data. Which resources
+//! are affected is in the structured log event and the durable record, both
+//! behind the operator's own access control.
+//!
+//! Every function is safe to call without an installed recorder: the
+//! [`metrics`] facade then records into its no-op recorder.
+
+/// Counter: secondary syncs that failed for good (after retries), labelled by
+/// `backend` and `operation`.
+pub(crate) const SECONDARY_SYNC_FAILURES: &str = "composite_secondary_sync_failures_total";
+/// Gauge: resources currently recorded as needing a reindex on a secondary.
+pub(crate) const SECONDARY_SYNC_NEEDS_REINDEX: &str = "composite_secondary_sync_needs_reindex";
+
+/// Count one final secondary sync failure. `operation` is one of `create`,
+/// `update`, `delete`.
+pub fn record_secondary_sync_failure(backend: &str, operation: &'static str) {
+    metrics::counter!(
+        SECONDARY_SYNC_FAILURES,
+        "backend" => backend.to_string(),
+        "operation" => operation
+    )
+    .increment(1);
+}
+
+/// Publish how many resources are recorded as needing a reindex.
+pub fn set_secondary_sync_needs_reindex(outstanding: u64) {
+    metrics::gauge!(SECONDARY_SYNC_NEEDS_REINDEX).set(outstanding as f64);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn record_everything() {
+        record_secondary_sync_failure("es", "create");
+        record_secondary_sync_failure("es", "create");
+        record_secondary_sync_failure("es", "delete");
+        set_secondary_sync_needs_reindex(2);
+    }
+
+    #[test]
+    fn recording_without_a_recorder_does_not_panic() {
+        record_everything();
+    }
+
+    /// Renders through a recorder built exactly as [`crate::metrics::init`]
+    /// builds the global one, installed only for this thread.
+    #[test]
+    fn metrics_render_with_backend_and_operation_labels_only() {
+        let recorder = crate::metrics::builder("test").build_recorder();
+        let handle = recorder.handle();
+        metrics::with_local_recorder(&recorder, record_everything);
+        let text = handle.render();
+
+        let series = |operation: &str, value: u64| {
+            text.lines().any(|line| {
+                line.starts_with(SECONDARY_SYNC_FAILURES)
+                    && line.contains("backend=\"es\"")
+                    && line.contains(&format!("operation=\"{operation}\""))
+                    && line.ends_with(&format!(" {value}"))
+            })
+        };
+        assert!(series("create", 2), "create counted twice:\n{text}");
+        assert!(series("delete", 1), "delete counted once:\n{text}");
+        assert!(
+            text.lines().any(|line| {
+                line.starts_with(SECONDARY_SYNC_NEEDS_REINDEX) && line.ends_with(" 2")
+            }),
+            "gauge published:\n{text}"
+        );
+
+        // Every label key is `service`, `backend` or `operation`.
+        for line in text.lines().filter(|line| !line.starts_with('#')) {
+            let Some((_, labels)) = line.split_once('{') else {
+                continue;
+            };
+            let labels = labels.split_once('}').map(|(l, _)| l).unwrap_or(labels);
+            for label in labels.split(',') {
+                let key = label.split_once('=').map(|(k, _)| k).unwrap_or(label);
+                assert!(
+                    ["service", "backend", "operation"].contains(&key),
+                    "unexpected label {key} in: {line}"
+                );
+            }
+        }
+    }
+}

crates/observability/src/lib.rs

@@ -20,6 +20,9 @@
 //!   dashboard's background counter reconcile (pass timing, storage query
 //!   latency and errors, seed queue depth, corrections). No tenant or
 //!   resource-type labels.
+//! - [`composite_metrics`] — process-level Prometheus metrics for composite
+//!   storage's secondary sync: final failures by backend and operation, and
+//!   the number of resources recorded as needing a reindex (#1334).
 //!
 //! ## Typical wiring
 //!
@@ -44,6 +47,7 @@
 //! - OTLP *metrics* are expected to be produced by an OpenTelemetry Collector
 //!   scraping `/metrics`; the app itself only pushes OTLP *traces*.
 
+pub mod composite_metrics;
 pub mod dashboard;
 pub mod dashboard_counters;
 pub mod dashboard_metrics;

crates/persistence/src/backends/mongodb/mod.rs

@@ -28,6 +28,7 @@ mod search_impl;
 pub(crate) mod search_index_builder;
 pub(crate) mod search_index_catalog;
 mod storage;
+mod sync_failures;
 mod user_settings;
 
 pub use backend::{MongoBackend, MongoBackendConfig};