From 8c799753db3ef368feb0181bcffa52d6329d40a7 Mon Sep 17 00:00:00 2001 From: smunini Date: Mon, 21 Sep 2026 15:01:22 -0400 Subject: [PATCH 1/3] feat(composite): count, report and durably record failed secondary syncs A composite write succeeds once the primary has committed it, in every HFS_COMPOSITE_SYNC_MODE. When a secondary then refused the change after SyncManager's retries, nothing recorded it: - synchronous / hybrid: `sync_to_secondaries` dropped the `Vec` that carried the failure, so its "Failed to sync ... to secondaries" warning only ever fired for a closed queue. The final failure was never logged at all, only the per-attempt "Sync attempt failed, retrying". - asynchronous: the worker logged "Async sync failed" with a backend id and an error, and no resource type, id or tenant. - batch (`create_many`, transaction bundles): statuses discarded likewise. The resource stayed missing from, or stale in, every search the secondary serves, with nothing to alert on and nothing naming it afterwards. Every final outcome now goes through a `SyncFailureRecorder` owned by the SyncManager (the one place the synchronous paths, the asynchronous worker and the batch path all pass through). Per decision on #1334 the write still succeeds; a final failure is 1. counted, through a `SecondarySyncObserver` the server wires to metrics (labels: backend, operation only), 2. emitted as one structured `error!` event with tenant, resource_type, id, version, backend_id, operation, attempts and error, never content, 3. recorded in a `SecondarySyncFailureLedger`: one row per (tenant, type, id, backend) holding operation, first/last_failed_at, last_error and attempts. Implemented on the SQLite (schema v34), PostgreSQL (schema v41) and MongoDB primaries; S3 has no ledger and degrades to metric + event. A later successful sync of the same resource clears its record (free while nothing is outstanding), and `CompositeStorage::repair_secondary_sync_failures` drains records in bounded batches by pushing the primary's current state (a delete when the primary no longer has it), re-reading the primary afterwards so it cannot leave an older version behind a concurrent write. `BackendError::Unavailable`'s Display omits its message, and that is what an exhausted Elasticsearch write reports since #1382, so the recorder keeps the detail explicitly. Also: SyncManager treated a secondary's NotFound on a Delete event as a failure and spent every retry on it. Not having the resource is the state a delete asks for; it is now success. Refs #1334 --- .../persistence/src/backends/mongodb/mod.rs | 1 + .../src/backends/mongodb/sync_failures.rs | 146 ++++ .../persistence/src/backends/postgres/mod.rs | 1 + .../src/backends/postgres/schema.rs | 33 +- .../src/backends/postgres/sync_failures.rs | 125 +++ crates/persistence/src/backends/sqlite/mod.rs | 1 + .../persistence/src/backends/sqlite/schema.rs | 32 +- .../src/backends/sqlite/sync_failures.rs | 147 ++++ crates/persistence/src/composite/mod.rs | 7 + crates/persistence/src/composite/storage.rs | 50 +- crates/persistence/src/composite/sync.rs | 91 ++- .../src/composite/sync_failures.rs | 685 ++++++++++++++++ .../composite_secondary_sync_failures.rs | 729 ++++++++++++++++++ 13 files changed, 2031 insertions(+), 17 deletions(-) create mode 100644 crates/persistence/src/backends/mongodb/sync_failures.rs create mode 100644 crates/persistence/src/backends/postgres/sync_failures.rs create mode 100644 crates/persistence/src/backends/sqlite/sync_failures.rs create mode 100644 crates/persistence/src/composite/sync_failures.rs create mode 100644 crates/persistence/tests/composite_secondary_sync_failures.rs diff --git a/crates/persistence/src/backends/mongodb/mod.rs b/crates/persistence/src/backends/mongodb/mod.rs index 7886fefa9d..643319de65 100644 --- a/crates/persistence/src/backends/mongodb/mod.rs +++ b/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}; diff --git a/crates/persistence/src/backends/mongodb/sync_failures.rs b/crates/persistence/src/backends/mongodb/sync_failures.rs new file mode 100644 index 0000000000..4d611b9280 --- /dev/null +++ b/crates/persistence/src/backends/mongodb/sync_failures.rs @@ -0,0 +1,146 @@ +//! MongoDB-backed "needs reindex" ledger for failed secondary syncs (#1334). +//! +//! One document per (tenant, resource type, id, secondary) in the +//! `secondary_sync_failures` collection, keyed by a compound `_id` so the +//! uniqueness a record needs comes from the collection itself: no extra index +//! and no schema-version step. Listing sorts by `last_failed_at` under a +//! `limit`, which MongoDB answers with a bounded top-k sort. See +//! [`crate::composite::sync_failures`]. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use futures::TryStreamExt; +use mongodb::bson::{Document, doc}; + +use crate::composite::sync_failures::{ + SecondarySyncFailure, SecondarySyncFailureLedger, SyncFailureKey, SyncFailureReport, + SyncOperation, +}; +use crate::error::{BackendError, StorageError, StorageResult}; + +use super::MongoBackend; +use super::retry::retry_transient; + +/// Name of the collection holding the records. +pub(crate) const SYNC_FAILURES_COLLECTION: &str = "secondary_sync_failures"; + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "mongodb".to_string(), + message, + source: None, + }) +} + +/// The compound `_id`. Field order is part of a document's identity, so it is +/// built in exactly one place. +fn record_id(key: &SyncFailureKey) -> Document { + doc! { + "tenant_id": &key.tenant_id, + "resource_type": &key.resource_type, + "resource_id": &key.resource_id, + "backend_id": &key.backend_id, + } +} + +/// Fixed-width RFC 3339 (UTC, microseconds): sorts as text. +fn format_time(time: DateTime) -> String { + time.to_rfc3339_opts(chrono::SecondsFormat::Micros, true) +} + +fn parse_time(value: Option<&str>) -> DateTime { + value + .and_then(|v| DateTime::parse_from_rfc3339(v).ok()) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(Utc::now) +} + +#[async_trait] +impl SecondarySyncFailureLedger for MongoBackend { + async fn record_sync_failure(&self, report: &SyncFailureReport) -> StorageResult { + let db = self.get_database().await?; + let collection = db.collection::(SYNC_FAILURES_COLLECTION); + let failed_at = format_time(report.failed_at); + let result = retry_transient(|| async { + collection + .update_one( + doc! { "_id": record_id(&report.key) }, + doc! { + "$setOnInsert": { "first_failed_at": &failed_at }, + "$set": { + "operation": report.operation.as_str(), + "last_failed_at": &failed_at, + "last_error": &report.error, + }, + "$inc": { "attempts": i64::from(report.attempts) }, + }, + ) + .upsert(true) + .await + }) + .await + .map_err(|e| backend_err(format!("record secondary sync failure: {e}")))?; + Ok(result.upserted_id.is_some()) + } + + async fn clear_sync_failure(&self, key: &SyncFailureKey) -> StorageResult { + let db = self.get_database().await?; + let collection = db.collection::(SYNC_FAILURES_COLLECTION); + let result = retry_transient(|| async { + collection.delete_one(doc! { "_id": record_id(key) }).await + }) + .await + .map_err(|e| backend_err(format!("clear secondary sync failure: {e}")))?; + Ok(result.deleted_count > 0) + } + + async fn list_sync_failures(&self, limit: usize) -> StorageResult> { + let db = self.get_database().await?; + let collection = db.collection::(SYNC_FAILURES_COLLECTION); + let limit = i64::try_from(limit).unwrap_or(i64::MAX); + let documents: Vec = retry_transient(|| async { + collection + .find(doc! {}) + .sort(doc! { "last_failed_at": 1_i32, "_id": 1_i32 }) + .limit(limit) + .await? + .try_collect() + .await + }) + .await + .map_err(|e| backend_err(format!("list secondary sync failures: {e}")))?; + + Ok(documents + .iter() + .filter_map(|document| { + let id = document.get_document("_id").ok()?; + Some(SecondarySyncFailure { + key: SyncFailureKey { + tenant_id: id.get_str("tenant_id").ok()?.to_string(), + resource_type: id.get_str("resource_type").ok()?.to_string(), + resource_id: id.get_str("resource_id").ok()?.to_string(), + backend_id: id.get_str("backend_id").ok()?.to_string(), + }, + operation: SyncOperation::from_stored( + document.get_str("operation").unwrap_or_default(), + ), + first_failed_at: parse_time(document.get_str("first_failed_at").ok()), + last_failed_at: parse_time(document.get_str("last_failed_at").ok()), + last_error: document + .get_str("last_error") + .unwrap_or_default() + .to_string(), + attempts: document.get_i64("attempts").unwrap_or_default().max(0) as u64, + }) + }) + .collect()) + } + + async fn count_sync_failures(&self) -> StorageResult { + let db = self.get_database().await?; + let collection = db.collection::(SYNC_FAILURES_COLLECTION); + retry_transient(|| async { collection.count_documents(doc! {}).await }) + .await + .map_err(|e| backend_err(format!("count secondary sync failures: {e}"))) + } +} diff --git a/crates/persistence/src/backends/postgres/mod.rs b/crates/persistence/src/backends/postgres/mod.rs index 35ef1e634f..a553761d3f 100644 --- a/crates/persistence/src/backends/postgres/mod.rs +++ b/crates/persistence/src/backends/postgres/mod.rs @@ -81,6 +81,7 @@ pub(crate) mod schema; pub mod search; mod search_impl; mod storage; +mod sync_failures; mod transaction; mod user_settings; diff --git a/crates/persistence/src/backends/postgres/schema.rs b/crates/persistence/src/backends/postgres/schema.rs index 0f0180719b..7d3ec2ccb3 100644 --- a/crates/persistence/src/backends/postgres/schema.rs +++ b/crates/persistence/src/backends/postgres/schema.rs @@ -9,7 +9,7 @@ use crate::core::bulk_submit_legacy::{ use crate::error::{BackendError, StorageResult}; /// Current schema version. -pub const SCHEMA_VERSION: i32 = 40; +pub const SCHEMA_VERSION: i32 = 41; /// Advisory-lock key serializing schema migration across HFS instances sharing /// one database. Arbitrary but must stay stable across releases. @@ -375,6 +375,7 @@ async fn migrate_schema( 37 => migrate_v37_to_v38(client).await?, 38 => migrate_v38_to_v39(client).await?, 39 => migrate_v39_to_v40(client).await?, + 40 => migrate_v40_to_v41(client).await?, _ => { return Err(pg_error(format!("Unknown schema version: {}", version))); } @@ -3710,6 +3711,36 @@ async fn migrate_v39_to_v40(client: &deadpool_postgres::Client) -> StorageResult Ok(()) } +/// v40 -> v41: `secondary_sync_failures` (#1334). +/// +/// The durable "needs reindex" ledger for a composite whose secondary refused +/// a change this primary had already committed. One row per (tenant, resource, +/// secondary), so a repeat failure folds into the existing row instead of +/// growing the table; `last_failed_at` orders the repair queue. +async fn migrate_v40_to_v41(client: &deadpool_postgres::Client) -> StorageResult<()> { + client + .batch_execute( + "CREATE TABLE IF NOT EXISTS secondary_sync_failures ( + tenant_id TEXT NOT NULL, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + backend_id TEXT NOT NULL, + operation TEXT NOT NULL, + first_failed_at TIMESTAMPTZ NOT NULL, + last_failed_at TIMESTAMPTZ NOT NULL, + last_error TEXT NOT NULL, + attempts BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (tenant_id, resource_type, resource_id, backend_id) + ); + CREATE INDEX IF NOT EXISTS idx_secondary_sync_failures_queue + ON secondary_sync_failures (last_failed_at);", + ) + .await + .map_err(|e| pg_error(format!("Migration v40->v41 failed: {}", e)))?; + + Ok(()) +} + /// v23 -> v24: drop `fk_search_resource`. /// /// `search_index` carried a composite FK to `resources` with `ON DELETE diff --git a/crates/persistence/src/backends/postgres/sync_failures.rs b/crates/persistence/src/backends/postgres/sync_failures.rs new file mode 100644 index 0000000000..8357d05212 --- /dev/null +++ b/crates/persistence/src/backends/postgres/sync_failures.rs @@ -0,0 +1,125 @@ +//! PostgreSQL-backed "needs reindex" ledger for failed secondary syncs (#1334). +//! +//! The PostgreSQL counterpart of the SQLite ledger: rows in +//! `secondary_sync_failures` (schema v41). See +//! [`crate::composite::sync_failures`]. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; + +use crate::composite::sync_failures::{ + SecondarySyncFailure, SecondarySyncFailureLedger, SyncFailureKey, SyncFailureReport, + SyncOperation, +}; +use crate::error::{BackendError, StorageError, StorageResult}; + +use super::PostgresBackend; + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "postgres".to_string(), + message, + source: None, + }) +} + +#[async_trait] +impl SecondarySyncFailureLedger for PostgresBackend { + async fn record_sync_failure(&self, report: &SyncFailureReport) -> StorageResult { + let client = self.get_client().await?; + let attempts = i64::from(report.attempts); + let row = client + .query_one( + "INSERT INTO secondary_sync_failures + (tenant_id, resource_type, resource_id, backend_id, operation, + first_failed_at, last_failed_at, last_error, attempts) + VALUES ($1, $2, $3, $4, $5, $6, $6, $7, $8) + ON CONFLICT (tenant_id, resource_type, resource_id, backend_id) + DO UPDATE SET operation = EXCLUDED.operation, + last_failed_at = EXCLUDED.last_failed_at, + last_error = EXCLUDED.last_error, + attempts = secondary_sync_failures.attempts + EXCLUDED.attempts + RETURNING (xmax = 0) AS inserted", + &[ + &report.key.tenant_id, + &report.key.resource_type, + &report.key.resource_id, + &report.key.backend_id, + &report.operation.as_str(), + &report.failed_at, + &report.error, + &attempts, + ], + ) + .await + .map_err(|e| backend_err(format!("record secondary sync failure: {e}")))?; + Ok(row.get(0)) + } + + async fn clear_sync_failure(&self, key: &SyncFailureKey) -> StorageResult { + let client = self.get_client().await?; + let removed = client + .execute( + "DELETE FROM secondary_sync_failures + WHERE tenant_id = $1 AND resource_type = $2 + AND resource_id = $3 AND backend_id = $4", + &[ + &key.tenant_id, + &key.resource_type, + &key.resource_id, + &key.backend_id, + ], + ) + .await + .map_err(|e| backend_err(format!("clear secondary sync failure: {e}")))?; + Ok(removed > 0) + } + + async fn list_sync_failures(&self, limit: usize) -> StorageResult> { + let client = self.get_client().await?; + let limit = i64::try_from(limit).unwrap_or(i64::MAX); + let rows = client + .query( + "SELECT tenant_id, resource_type, resource_id, backend_id, operation, + first_failed_at, last_failed_at, last_error, attempts + FROM secondary_sync_failures + ORDER BY last_failed_at, tenant_id, resource_type, resource_id, backend_id + LIMIT $1", + &[&limit], + ) + .await + .map_err(|e| backend_err(format!("list secondary sync failures: {e}")))?; + Ok(rows + .into_iter() + .map(|row| { + let operation: String = row.get(4); + let first_failed_at: DateTime = row.get(5); + let last_failed_at: DateTime = row.get(6); + let attempts: i64 = row.get(8); + SecondarySyncFailure { + key: SyncFailureKey { + tenant_id: row.get(0), + resource_type: row.get(1), + resource_id: row.get(2), + backend_id: row.get(3), + }, + operation: SyncOperation::from_stored(&operation), + first_failed_at, + last_failed_at, + last_error: row.get(7), + attempts: attempts.max(0) as u64, + } + }) + .collect()) + } + + async fn count_sync_failures(&self) -> StorageResult { + let client = self.get_client().await?; + let row = client + .query_one("SELECT COUNT(*) FROM secondary_sync_failures", &[]) + .await + .map_err(|e| backend_err(format!("count secondary sync failures: {e}")))?; + let count: i64 = row.get(0); + Ok(count.max(0) as u64) + } +} diff --git a/crates/persistence/src/backends/sqlite/mod.rs b/crates/persistence/src/backends/sqlite/mod.rs index bc51c90d55..a724741870 100644 --- a/crates/persistence/src/backends/sqlite/mod.rs +++ b/crates/persistence/src/backends/sqlite/mod.rs @@ -75,6 +75,7 @@ mod schema; pub mod search; mod search_impl; mod storage; +mod sync_failures; mod transaction; mod user_settings; diff --git a/crates/persistence/src/backends/sqlite/schema.rs b/crates/persistence/src/backends/sqlite/schema.rs index e33c6020ca..e621cf548f 100644 --- a/crates/persistence/src/backends/sqlite/schema.rs +++ b/crates/persistence/src/backends/sqlite/schema.rs @@ -10,7 +10,7 @@ use crate::core::bulk_submit_legacy::{ use crate::error::StorageResult; /// Current schema version. -pub const SCHEMA_VERSION: i32 = 33; +pub const SCHEMA_VERSION: i32 = 34; /// The `search_index` value indexes. Excludes `idx_search_composite`, which the /// delete-by-resource path needs at all times, and `idx_search_token_display`, @@ -444,6 +444,7 @@ fn migrate_schema(conn: &Connection, from_version: i32) -> StorageResult<()> { 30 => migrate_v30_to_v31(conn)?, 31 => migrate_v31_to_v32(conn)?, 32 => migrate_v32_to_v33(conn)?, + 33 => migrate_v33_to_v34(conn)?, _ => { return Err(crate::error::StorageError::Backend( crate::error::BackendError::Internal { @@ -1567,6 +1568,35 @@ fn migrate_v32_to_v33(conn: &Connection) -> StorageResult<()> { Ok(()) } +/// Migrate from schema version 33 to version 34. +/// +/// Adds `secondary_sync_failures`: the durable "needs reindex" ledger for a +/// composite whose secondary refused a change the primary had already +/// committed (#1334). One row per (tenant, resource, secondary), so a repeat +/// failure folds into the existing row instead of growing the table. +/// `last_failed_at` orders the repair queue; both timestamps are fixed-width +/// RFC 3339 text, which sorts chronologically. +fn migrate_v33_to_v34(conn: &Connection) -> StorageResult<()> { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS secondary_sync_failures ( + tenant_id TEXT NOT NULL, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + backend_id TEXT NOT NULL, + operation TEXT NOT NULL, + first_failed_at TEXT NOT NULL, + last_failed_at TEXT NOT NULL, + last_error TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (tenant_id, resource_type, resource_id, backend_id) + ); + CREATE INDEX IF NOT EXISTS idx_secondary_sync_failures_queue + ON secondary_sync_failures (last_failed_at);", + ) + .map_err(|e| migration_err(format!("v34 create secondary_sync_failures: {e}")))?; + Ok(()) +} + /// Migrate from schema version 10 to version 11. /// /// Adds columns supporting `_contained` search: index rows extracted from a diff --git a/crates/persistence/src/backends/sqlite/sync_failures.rs b/crates/persistence/src/backends/sqlite/sync_failures.rs new file mode 100644 index 0000000000..6263e66f0e --- /dev/null +++ b/crates/persistence/src/backends/sqlite/sync_failures.rs @@ -0,0 +1,147 @@ +//! SQLite-backed "needs reindex" ledger for failed secondary syncs (#1334). +//! +//! When SQLite is a composite's primary, a secondary that refuses a change +//! after retries is recorded in `secondary_sync_failures` (schema v34), next +//! to the resources it concerns, so the record survives a restart. See +//! [`crate::composite::sync_failures`]. + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use rusqlite::params; + +use crate::composite::sync_failures::{ + SecondarySyncFailure, SecondarySyncFailureLedger, SyncFailureKey, SyncFailureReport, + SyncOperation, +}; +use crate::error::{BackendError, StorageError, StorageResult}; + +use super::SqliteBackend; + +fn backend_err(message: String) -> StorageError { + StorageError::Backend(BackendError::Internal { + backend_name: "sqlite".to_string(), + message, + source: None, + }) +} + +fn parse_time(value: &str) -> DateTime { + DateTime::parse_from_rfc3339(value) + .map(|dt| dt.with_timezone(&Utc)) + .unwrap_or_else(|_| Utc::now()) +} + +#[async_trait] +impl SecondarySyncFailureLedger for SqliteBackend { + async fn record_sync_failure(&self, report: &SyncFailureReport) -> StorageResult { + let report = report.clone(); + self.run_blocking(move |conn| { + // Timestamps are fixed-width RFC 3339 (UTC, microseconds), so + // they sort as text. + let failed_at = report + .failed_at + .to_rfc3339_opts(chrono::SecondsFormat::Micros, true); + let total: i64 = conn + .query_row( + "INSERT INTO secondary_sync_failures + (tenant_id, resource_type, resource_id, backend_id, operation, + first_failed_at, last_failed_at, last_error, attempts) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?6, ?7, ?8) + ON CONFLICT (tenant_id, resource_type, resource_id, backend_id) + DO UPDATE SET operation = excluded.operation, + last_failed_at = excluded.last_failed_at, + last_error = excluded.last_error, + attempts = attempts + excluded.attempts + RETURNING attempts", + params![ + report.key.tenant_id, + report.key.resource_type, + report.key.resource_id, + report.key.backend_id, + report.operation.as_str(), + failed_at, + report.error, + i64::from(report.attempts), + ], + |row| row.get(0), + ) + .map_err(|e| backend_err(format!("record secondary sync failure: {e}")))?; + // A fresh row holds exactly this report's attempts; a folded one + // holds more. + Ok(total == i64::from(report.attempts)) + }) + .await + } + + async fn clear_sync_failure(&self, key: &SyncFailureKey) -> StorageResult { + let key = key.clone(); + self.run_blocking(move |conn| { + let removed = conn + .execute( + "DELETE FROM secondary_sync_failures + WHERE tenant_id = ?1 AND resource_type = ?2 + AND resource_id = ?3 AND backend_id = ?4", + params![ + key.tenant_id, + key.resource_type, + key.resource_id, + key.backend_id + ], + ) + .map_err(|e| backend_err(format!("clear secondary sync failure: {e}")))?; + Ok(removed > 0) + }) + .await + } + + async fn list_sync_failures(&self, limit: usize) -> StorageResult> { + let limit = i64::try_from(limit).unwrap_or(i64::MAX); + self.run_blocking(move |conn| { + let mut stmt = conn + .prepare( + "SELECT tenant_id, resource_type, resource_id, backend_id, operation, + first_failed_at, last_failed_at, last_error, attempts + FROM secondary_sync_failures + ORDER BY last_failed_at, tenant_id, resource_type, resource_id, backend_id + LIMIT ?1", + ) + .map_err(|e| backend_err(format!("list secondary sync failures: {e}")))?; + let rows = stmt + .query_map([limit], |row| { + let operation: String = row.get(4)?; + let first_failed_at: String = row.get(5)?; + let last_failed_at: String = row.get(6)?; + let attempts: i64 = row.get(8)?; + Ok(SecondarySyncFailure { + key: SyncFailureKey { + tenant_id: row.get(0)?, + resource_type: row.get(1)?, + resource_id: row.get(2)?, + backend_id: row.get(3)?, + }, + operation: SyncOperation::from_stored(&operation), + first_failed_at: parse_time(&first_failed_at), + last_failed_at: parse_time(&last_failed_at), + last_error: row.get(7)?, + attempts: attempts.max(0) as u64, + }) + }) + .map_err(|e| backend_err(format!("list secondary sync failures: {e}")))?; + rows.collect::, _>>() + .map_err(|e| backend_err(format!("read secondary sync failure: {e}"))) + }) + .await + } + + async fn count_sync_failures(&self) -> StorageResult { + self.run_blocking(|conn| { + let count: i64 = conn + .query_row("SELECT COUNT(*) FROM secondary_sync_failures", [], |row| { + row.get(0) + }) + .map_err(|e| backend_err(format!("count secondary sync failures: {e}")))?; + Ok(count.max(0) as u64) + }) + .await + } +} diff --git a/crates/persistence/src/composite/mod.rs b/crates/persistence/src/composite/mod.rs index 2a58bc354b..d29e0af271 100644 --- a/crates/persistence/src/composite/mod.rs +++ b/crates/persistence/src/composite/mod.rs @@ -86,6 +86,7 @@ //! - [`storage`] - CompositeStorage implementation (Phase 2) //! - [`merger`] - Result merging strategies (Phase 2) //! - [`sync`] - Secondary synchronization (Phase 2) +//! - [`sync_failures`] - Metric, event and durable record for failed syncs //! - [`cost`] - Cost-based optimization (Phase 3) //! - [`health`] - Health monitoring (Phase 3) @@ -100,6 +101,7 @@ pub mod merger; pub mod router; pub mod storage; pub mod sync; +pub mod sync_failures; // Re-export main types pub use analyzer::{ @@ -122,6 +124,11 @@ pub use sync::{ BackendSyncStatus, ReconciliationResult, SyncEvent, SyncManager, SyncReconciler, SyncStatus, }; +pub use sync_failures::{ + SecondarySyncFailure, SecondarySyncFailureLedger, SecondarySyncObserver, SyncFailureKey, + SyncFailureRecorder, SyncFailureReport, SyncOperation, SyncRepairReport, +}; + // Phase 3: Cost estimation and health monitoring pub use cost::{ BenchmarkMeasurement, BenchmarkOperation, BenchmarkResults, CostBreakdown, CostComparison, diff --git a/crates/persistence/src/composite/storage.rs b/crates/persistence/src/composite/storage.rs index 6f2ecc2d8f..ac60208452 100644 --- a/crates/persistence/src/composite/storage.rs +++ b/crates/persistence/src/composite/storage.rs @@ -64,6 +64,9 @@ use super::config::{CompositeConfig, SyncMode}; use super::merger::{MergeOptions, ResultMerger}; use super::router::{QueryRouter, RoutingDecision, RoutingError}; use super::sync::{SyncEvent, SyncManager, SyncStatus}; +use super::sync_failures::{ + SecondarySyncFailureLedger, SecondarySyncObserver, SyncFailureRecorder, +}; /// A dynamically typed storage backend. pub type DynStorage = Arc; @@ -472,13 +475,58 @@ impl CompositeStorage { } /// Synchronizes a resource change to secondary backends. + /// + /// `Err` means the event could not even be handed over (the asynchronous + /// queue is gone). A secondary that *took* the event and then refused it + /// after retries is not an error here — the primary has committed and the + /// write stands (#1334) — but it is never silent either: the + /// [`SyncManager`] reports every final outcome, from the synchronous path + /// and from the asynchronous worker alike, to its + /// [`SyncFailureRecorder`], which counts it, emits the structured event + /// and records the resource as needing a reindex. pub(crate) async fn sync_to_secondaries(&self, event: SyncEvent) -> StorageResult<()> { if let Some(ref sync_manager) = self.sync_manager { - sync_manager.sync(&event, &self.secondaries).await?; + let statuses = sync_manager.sync(&event, &self.secondaries).await?; + for status in statuses.iter().filter(|status| !status.success) { + // Already counted, logged and recorded by the recorder; this + // only ties the failure to the request's own trace span. + debug!( + backend_id = %status.backend_id, + retries = status.retry_count, + "Write committed on the primary; secondary sync failed and was recorded" + ); + } } Ok(()) } + /// Keeps "needs reindex" records for failed secondary syncs in `ledger` + /// — normally the primary backend — so they survive a restart and + /// [`repair_secondary_sync_failures`](Self::repair_secondary_sync_failures) + /// can work through them (#1334). Without one, failures are still counted + /// and logged, just not listed. + pub fn with_sync_failure_ledger(self, ledger: Arc) -> Self { + if let Some(recorder) = self.sync_failure_recorder() { + recorder.set_ledger(ledger); + } + self + } + + /// Forwards secondary sync failures to a metrics exporter (#1334). + pub fn with_sync_observer(self, observer: Arc) -> Self { + if let Some(recorder) = self.sync_failure_recorder() { + recorder.set_observer(observer); + } + self + } + + /// Where secondary sync outcomes are reported; `None` without secondaries. + pub fn sync_failure_recorder(&self) -> Option<&Arc> { + self.sync_manager + .as_ref() + .map(|manager| manager.failure_recorder()) + } + /// Routes and executes a search query. /// /// With a dedicated Search backend the whole query goes there; otherwise diff --git a/crates/persistence/src/composite/sync.rs b/crates/persistence/src/composite/sync.rs index efc3721e7c..175521bbd6 100644 --- a/crates/persistence/src/composite/sync.rs +++ b/crates/persistence/src/composite/sync.rs @@ -36,14 +36,15 @@ use parking_lot::RwLock; use serde_json::Value; use tokio::sync::{mpsc, oneshot}; use tokio::time::sleep; -use tracing::{debug, error, warn}; +use tracing::{debug, warn}; use crate::core::ResourceStorage; -use crate::error::{BackendError, StorageError, StorageResult}; +use crate::error::{BackendError, ResourceError, StorageError, StorageResult}; use crate::tenant::{TenantContext, TenantId, TenantPermissions}; use crate::types::StoredResource; use super::config::{RetryConfig, SyncConfig, SyncMode}; +use super::sync_failures::{SyncFailureKey, SyncFailureRecorder, SyncOperation}; /// A synchronization event to propagate to secondary backends. #[derive(Debug, Clone)] @@ -166,6 +167,10 @@ pub struct SyncManager { /// Sync status per backend. status: Arc>>, + + /// Where every final outcome is reported (#1334): the failure metric, + /// the structured event, and the durable "needs reindex" record. + recorder: Arc, } /// Status tracking for a backend. @@ -220,9 +225,15 @@ impl SyncManager { config, event_sender: None, status: Arc::new(RwLock::new(HashMap::new())), + recorder: Arc::new(SyncFailureRecorder::default()), } } + /// The recorder every final sync outcome is reported to (#1334). + pub fn failure_recorder(&self) -> &Arc { + &self.recorder + } + /// Starts the async sync worker. pub fn start_async_worker( &mut self, @@ -233,9 +244,10 @@ impl SyncManager { let config = self.config.clone(); let status = self.status.clone(); + let recorder = self.recorder.clone(); tokio::spawn(async move { - Self::async_worker(receiver, backends, config, status).await; + Self::async_worker(receiver, backends, config, status, recorder).await; }) } @@ -245,6 +257,7 @@ impl SyncManager { backends: HashMap>, config: SyncConfig, status: Arc>>, + recorder: Arc, ) { let mut batch = Vec::new(); let batch_timeout = Duration::from_millis(100); @@ -302,6 +315,22 @@ impl SyncManager { ) .await; + // Nobody is waiting on this event any more: the + // recorder is the only witness of how it ended. + match &result { + Ok(()) => recorder.sync_succeeded(&queued.event, backend_id).await, + Err(e) => { + recorder + .sync_failed( + &queued.event, + backend_id, + e, + config.retry.max_retries + 1, + ) + .await + } + } + // Update status let mut status_map = status.write(); let backend_status = status_map.entry(backend_id.clone()).or_default(); @@ -312,14 +341,7 @@ impl SyncManager { backend_status.total_synced += 1; backend_status.healthy = true; } - Err(e) => { - backend_status.total_errors += 1; - error!( - backend = %backend_id, - error = %e, - "Async sync failed" - ); - } + Err(_) => backend_status.total_errors += 1, } if backend_status.pending_events > 0 { @@ -416,9 +438,16 @@ impl SyncManager { let backend = backend.clone(); let backend_id = backend_id.clone(); let retry_config = self.config.retry.clone(); + let recorder = self.recorder.clone(); tasks.spawn(async move { let start = std::time::Instant::now(); + let key = |resource_id: &str| SyncFailureKey { + tenant_id: tenant.tenant_id().as_str().to_string(), + resource_type: resource_type.clone(), + resource_id: resource_id.to_string(), + backend_id: backend_id.clone(), + }; let contents = resources .iter() .map(|(_, content)| content.clone()) @@ -434,6 +463,7 @@ impl SyncManager { for ((resource_id, content), result) in resources.iter().zip(results) { let Err(batch_error) = result else { synced += 1; + recorder.resource_sync_succeeded(key(resource_id)).await; continue; }; warn!( @@ -452,9 +482,22 @@ impl SyncManager { }; match Self::sync_event_to_backend(&event, backend.as_ref(), &retry_config).await { - Ok(()) => synced += 1, + Ok(()) => { + synced += 1; + recorder.resource_sync_succeeded(key(resource_id)).await; + } Err(e) => { errors += 1; + recorder + .resource_sync_failed( + key(resource_id), + SyncOperation::Create, + content.pointer("/meta/versionId").and_then(|v| v.as_str()), + &e, + // The batch attempt, then the retried single sync. + retry_config.max_retries + 2, + ) + .await; last_error = Some(e.to_string()); failed_resource_ids.push(resource_id.clone()); } @@ -522,11 +565,22 @@ impl SyncManager { let backend = backend.clone(); let backend_id = backend_id.clone(); let retry_config = self.config.retry.clone(); + let recorder = self.recorder.clone(); tasks.spawn(async move { let start = std::time::Instant::now(); - match Self::sync_event_to_backend(&event, backend.as_ref(), &retry_config).await { + let result = + Self::sync_event_to_backend(&event, backend.as_ref(), &retry_config).await; + match &result { + Ok(()) => recorder.sync_succeeded(&event, &backend_id).await, + Err(e) => { + recorder + .sync_failed(&event, &backend_id, e, retry_config.max_retries + 1) + .await + } + } + match result { Ok(_) => SyncStatus { backend_id, success: true, @@ -682,7 +736,16 @@ impl SyncManager { } => { let tenant = TenantContext::new(tenant_id.clone(), TenantPermissions::full_access()); - backend.delete(&tenant, resource_type, resource_id).await + match backend.delete(&tenant, resource_type, resource_id).await { + // The secondary does not have it, which is the state + // the delete asks for: a create that never reached it, + // or a delete whose answer was lost. Retrying cannot + // change that, and it is not a failure (#1334). + Err(StorageError::Resource( + ResourceError::NotFound { .. } | ResourceError::Gone { .. }, + )) => Ok(()), + other => other, + } } SyncEvent::BulkSync { resources, diff --git a/crates/persistence/src/composite/sync_failures.rs b/crates/persistence/src/composite/sync_failures.rs new file mode 100644 index 0000000000..2b3a039166 --- /dev/null +++ b/crates/persistence/src/composite/sync_failures.rs @@ -0,0 +1,685 @@ +//! Bookkeeping for secondary syncs that failed for good (#1334). +//! +//! A composite write succeeds once the primary has committed it, in every +//! [`SyncMode`](super::config::SyncMode): the primary is the system of record. +//! A secondary that then rejects the change, after [`SyncManager`]'s retries, +//! leaves the resource missing from (or stale in) every search that secondary +//! serves. This module is what makes that visible and repairable instead of a +//! free-text log line: +//! +//! 1. a **counter** and a **gauge**, through [`SecondarySyncObserver`] (the +//! persistence crate has no metrics dependency; the server plugs its +//! exporter in), +//! 2. one **structured event** per final failure, carrying the tenant, +//! resource type, id, version, backend and operation — never resource +//! content, +//! 3. a **durable "needs reindex" record** in a [`SecondarySyncFailureLedger`] +//! kept by the primary backend, so the affected resources are still known +//! after a restart and +//! [`CompositeStorage::repair_secondary_sync_failures`] can re-sync them. +//! +//! A record is a *hint*, not a queued write: the repair always pushes the +//! primary's **current** state, so a stale or duplicate record costs one +//! redundant idempotent write and nothing else. +//! +//! [`SyncManager`]: super::sync::SyncManager +//! [`CompositeStorage::repair_secondary_sync_failures`]: +//! super::storage::CompositeStorage::repair_secondary_sync_failures + +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use parking_lot::{Mutex, RwLock}; +use tracing::{error, warn}; + +use crate::error::{BackendError, ResourceError, StorageError, StorageResult}; +use crate::tenant::{TenantContext, TenantId, TenantPermissions}; +use crate::types::StoredResource; + +use super::storage::{CompositeStorage, DynStorage}; +use super::sync::SyncEvent; + +/// Longest error text kept on a record. Backend errors can be long (an +/// Elasticsearch rejection nests its causes); the record needs enough to +/// triage, not the whole response. +pub const MAX_RECORDED_ERROR_CHARS: usize = 1024; + +/// Most outstanding keys a process keeps in memory to decide whether a +/// successful sync has a record to clear. Past it, every successful sync +/// issues the (indexed, usually no-op) delete instead. +const MAX_TRACKED_KEYS: usize = 50_000; + +/// The kind of change a secondary failed to take. The fixed values of the +/// `operation` metric label. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum SyncOperation { + /// A new resource. + Create, + /// A new version of an existing resource. + Update, + /// A delete. + Delete, +} + +impl SyncOperation { + /// The label / stored value. + pub fn as_str(self) -> &'static str { + match self { + SyncOperation::Create => "create", + SyncOperation::Update => "update", + SyncOperation::Delete => "delete", + } + } + + /// Reads a stored value back. Anything unrecognised is an `Update`: the + /// repair does not branch on the operation, so the safe reading is the + /// generic one rather than an error that would wedge the ledger. + pub fn from_stored(value: &str) -> Self { + match value { + "create" => SyncOperation::Create, + "delete" => SyncOperation::Delete, + _ => SyncOperation::Update, + } + } +} + +/// What a record is unique on: one resource on one secondary. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct SyncFailureKey { + /// Tenant the resource belongs to. + pub tenant_id: String, + /// Resource type. + pub resource_type: String, + /// Resource id. + pub resource_id: String, + /// The secondary that missed the change. + pub backend_id: String, +} + +/// One final sync failure, as handed to the ledger. +#[derive(Debug, Clone)] +pub struct SyncFailureReport { + /// Which resource on which secondary. + pub key: SyncFailureKey, + /// The change that failed. + pub operation: SyncOperation, + /// Attempts this failure took (the first try plus its retries). + pub attempts: u32, + /// The last error, truncated to [`MAX_RECORDED_ERROR_CHARS`]. + pub error: String, + /// When it failed. + pub failed_at: DateTime, +} + +/// A durable "needs reindex" record. +#[derive(Debug, Clone)] +pub struct SecondarySyncFailure { + /// Which resource on which secondary. + pub key: SyncFailureKey, + /// The most recent change that failed. + pub operation: SyncOperation, + /// When the resource first fell out of sync. + pub first_failed_at: DateTime, + /// When it most recently failed (a write, or a repair attempt). + pub last_failed_at: DateTime, + /// The most recent error. + pub last_error: String, + /// Attempts made in total, across every failure folded into this record. + pub attempts: u64, +} + +/// Where "needs reindex" records live so they survive a restart. +/// +/// Implemented by the primary backends that can keep a small table of their +/// own (SQLite, PostgreSQL, MongoDB). A composite whose primary has no ledger +/// (S3) still counts and logs every failure; it just cannot list them later. +#[async_trait] +pub trait SecondarySyncFailureLedger: Send + Sync { + /// Records a failure. One record per [`SyncFailureKey`]: a repeat keeps + /// `first_failed_at`, adds to `attempts`, and replaces the operation, the + /// error and `last_failed_at`. Returns whether a **new** record was made. + async fn record_sync_failure(&self, report: &SyncFailureReport) -> StorageResult; + + /// Drops the record, if any. Returns whether there was one. + async fn clear_sync_failure(&self, key: &SyncFailureKey) -> StorageResult; + + /// Up to `limit` records across all tenants, least recently failed first, + /// so a record that keeps failing moves to the back instead of starving + /// the rest of a bounded batch. + async fn list_sync_failures(&self, limit: usize) -> StorageResult>; + + /// How many records are outstanding. + async fn count_sync_failures(&self) -> StorageResult; +} + +/// Receives the countable side of a failure; the server forwards it to its +/// metrics exporter. No tenant, resource type or id reaches this trait: the +/// metrics endpoint is public and those would be unbounded label values. +pub trait SecondarySyncObserver: Send + Sync { + /// One final failure (after retries) of `operation` on `backend_id`. + fn sync_failed(&self, backend_id: &str, operation: SyncOperation); + + /// The number of outstanding "needs reindex" records changed. + fn needs_reindex(&self, outstanding: u64); +} + +/// The error as text, including the detail `BackendError::Unavailable`'s +/// display leaves out — which is exactly what an exhausted Elasticsearch write +/// reports (#1382), so without this every such record would read +/// "backend unavailable: elasticsearch" and nothing else. +pub(crate) fn error_detail(error: &StorageError) -> String { + match error { + StorageError::Backend(BackendError::Unavailable { message, .. }) => { + format!("{error}: {message}") + } + other => other.to_string(), + } +} + +/// Truncates an error for the ledger, on a character boundary. +pub(crate) fn truncate_error(error: &str) -> String { + match error.char_indices().nth(MAX_RECORDED_ERROR_CHARS) { + Some((cut, _)) => format!("{}…", &error[..cut]), + None => error.to_string(), + } +} + +/// The resources an event names, with the operation and (when the event +/// knows it) the version each one was at. +fn event_subjects(event: &SyncEvent) -> Vec<(String, String, SyncOperation, Option)> { + match event { + SyncEvent::Create { + resource_type, + resource_id, + content, + .. + } => vec![( + resource_type.clone(), + resource_id.clone(), + SyncOperation::Create, + content + .pointer("/meta/versionId") + .and_then(|v| v.as_str()) + .map(str::to_string), + )], + SyncEvent::Update { + resource_type, + resource_id, + version, + .. + } => vec![( + resource_type.clone(), + resource_id.clone(), + SyncOperation::Update, + Some(version.clone()), + )], + SyncEvent::Delete { + resource_type, + resource_id, + .. + } => vec![( + resource_type.clone(), + resource_id.clone(), + SyncOperation::Delete, + None, + )], + SyncEvent::BulkSync { resources, .. } => resources + .iter() + .map(|resource| { + ( + resource.resource_type().to_string(), + resource.id().to_string(), + SyncOperation::Update, + Some(resource.version_id().to_string()), + ) + }) + .collect(), + } +} + +/// Turns sync outcomes into the metric, the event and the ledger record. +/// +/// Owned by the [`SyncManager`](super::sync::SyncManager), because that is +/// the only place every outcome passes through: the asynchronous worker +/// finishes an event long after the write that queued it has returned. +#[derive(Default)] +pub struct SyncFailureRecorder { + ledger: RwLock>>, + observer: RwLock>>, + /// Outstanding records, as far as this process knows. + outstanding: AtomicU64, + /// Keys this process knows to be outstanding; see [`MAX_TRACKED_KEYS`]. + tracked: Mutex>, + /// `tracked` does not cover the ledger (too many records, or the ledger + /// could not be read): clear on every success rather than trust it. + untracked: AtomicBool, + hydrated: tokio::sync::OnceCell<()>, +} + +impl SyncFailureRecorder { + /// Sets the durable ledger. + pub(crate) fn set_ledger(&self, ledger: Arc) { + *self.ledger.write() = Some(ledger); + } + + /// Sets the metrics observer. + pub(crate) fn set_observer(&self, observer: Arc) { + *self.observer.write() = Some(observer); + } + + pub(crate) fn ledger(&self) -> Option> { + self.ledger.read().clone() + } + + /// The record for `key` is gone from the ledger; stop tracking it. + pub(crate) fn forget(&self, key: &SyncFailureKey) { + self.tracked.lock().remove(key); + } + + /// Outstanding records, as far as this process knows. + pub fn outstanding(&self) -> u64 { + self.outstanding.load(Ordering::Relaxed) + } + + fn publish_outstanding(&self) { + if let Some(observer) = self.observer.read().clone() { + observer.needs_reindex(self.outstanding()); + } + } + + /// Replaces the outstanding count with the ledger's own (another process + /// may have added or repaired records). + pub(crate) fn set_outstanding(&self, outstanding: u64) { + self.outstanding.store(outstanding, Ordering::Relaxed); + self.publish_outstanding(); + } + + /// Loads what an earlier run left behind, once. Runs on the first outcome + /// rather than at construction so building a composite stays synchronous. + async fn hydrate(&self) { + self.hydrated + .get_or_init(|| async { + let Some(ledger) = self.ledger() else { + return; + }; + match ledger.list_sync_failures(MAX_TRACKED_KEYS + 1).await { + Ok(records) => { + if records.len() > MAX_TRACKED_KEYS { + self.untracked.store(true, Ordering::Relaxed); + } + let mut tracked = self.tracked.lock(); + tracked.extend( + records + .into_iter() + .take(MAX_TRACKED_KEYS) + .map(|record| record.key), + ); + } + Err(e) => { + self.untracked.store(true, Ordering::Relaxed); + warn!( + error = %e, + "Could not read the secondary sync failure ledger; \ + every successful sync will try to clear its record" + ); + } + } + match ledger.count_sync_failures().await { + Ok(outstanding) => self.set_outstanding(outstanding), + Err(e) => warn!( + error = %e, + "Could not count the secondary sync failure ledger" + ), + } + }) + .await; + } + + /// A sync of `event` to `backend_id` failed for good. + pub(crate) async fn sync_failed( + &self, + event: &SyncEvent, + backend_id: &str, + error: &StorageError, + attempts: u32, + ) { + let tenant_id = event.tenant_id().as_str().to_string(); + for (resource_type, resource_id, operation, version) in event_subjects(event) { + self.resource_sync_failed( + SyncFailureKey { + tenant_id: tenant_id.clone(), + resource_type, + resource_id, + backend_id: backend_id.to_string(), + }, + operation, + version.as_deref(), + error, + attempts, + ) + .await; + } + } + + /// One resource's sync failed for good: count it, say so, record it. + pub(crate) async fn resource_sync_failed( + &self, + key: SyncFailureKey, + operation: SyncOperation, + version: Option<&str>, + error: &StorageError, + attempts: u32, + ) { + self.hydrate().await; + let error = error_detail(error); + let ledger = self.ledger(); + if let Some(observer) = self.observer.read().clone() { + observer.sync_failed(&key.backend_id, operation); + } + error!( + tenant = %key.tenant_id, + resource_type = %key.resource_type, + id = %key.resource_id, + version = version.unwrap_or(""), + backend_id = %key.backend_id, + operation = operation.as_str(), + attempts, + recorded = ledger.is_some(), + error = %error, + "Secondary sync failed; the primary holds the write and the secondary needs a reindex of this resource" + ); + + let Some(ledger) = ledger else { + return; + }; + let report = SyncFailureReport { + key: key.clone(), + operation, + attempts, + error: truncate_error(&error), + failed_at: Utc::now(), + }; + match ledger.record_sync_failure(&report).await { + Ok(inserted) => { + { + let mut tracked = self.tracked.lock(); + if tracked.len() < MAX_TRACKED_KEYS { + tracked.insert(key); + } else if !tracked.contains(&key) { + self.untracked.store(true, Ordering::Relaxed); + } + } + if inserted { + self.outstanding.fetch_add(1, Ordering::Relaxed); + self.publish_outstanding(); + } + } + Err(e) => error!( + tenant = %key.tenant_id, + resource_type = %key.resource_type, + id = %key.resource_id, + backend_id = %key.backend_id, + error = %e, + "Could not record the secondary sync failure; only this log line and the metric say the resource needs a reindex" + ), + } + } + + /// A sync of `event` to `backend_id` succeeded: whatever was owed for + /// these resources on that backend no longer is. + pub(crate) async fn sync_succeeded(&self, event: &SyncEvent, backend_id: &str) { + let tenant_id = event.tenant_id().as_str(); + for (resource_type, resource_id, _, _) in event_subjects(event) { + self.resource_sync_succeeded(SyncFailureKey { + tenant_id: tenant_id.to_string(), + resource_type, + resource_id, + backend_id: backend_id.to_string(), + }) + .await; + } + } + + /// One resource's sync succeeded. Free unless something is outstanding: + /// the healthy path never touches the ledger. + pub(crate) async fn resource_sync_succeeded(&self, key: SyncFailureKey) { + self.hydrate().await; + let Some(ledger) = self.ledger() else { + return; + }; + let known = self.tracked.lock().remove(&key); + if !known && !self.untracked.load(Ordering::Relaxed) { + return; + } + match ledger.clear_sync_failure(&key).await { + Ok(true) => { + // Saturating: another process may have counted this record. + let _ = self + .outstanding + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |n| { + Some(n.saturating_sub(1)) + }); + self.publish_outstanding(); + } + Ok(false) => {} + Err(e) => { + // Still owed as far as the ledger says; the repair pass will + // find it in sync and clear it. + self.tracked.lock().insert(key.clone()); + warn!( + tenant = %key.tenant_id, + resource_type = %key.resource_type, + id = %key.resource_id, + backend_id = %key.backend_id, + error = %e, + "Resource synced, but its needs-reindex record could not be cleared" + ); + } + } + } +} + +/// What one [`repair_secondary_sync_failures`] pass did. +/// +/// [`repair_secondary_sync_failures`]: +/// super::storage::CompositeStorage::repair_secondary_sync_failures +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct SyncRepairReport { + /// Records looked at (at most the requested batch). + pub examined: usize, + /// Records whose resource was re-synced and whose record was cleared. + pub repaired: usize, + /// Records whose secondary still refused; they stay for the next pass. + pub still_failing: usize, + /// Records dropped because they name a backend this composite does not + /// have (a secondary removed from the configuration). + pub dropped: usize, + /// Records outstanding after the pass. + pub remaining: u64, +} + +/// How often one repair re-reads the primary after pushing, when a write +/// slipped in between. Past it the record stays for the next pass. +const REPAIR_MAX_ROUNDS: usize = 3; + +impl CompositeStorage { + /// Works through up to `limit` "needs reindex" records: pushes the + /// primary's **current** state of each recorded resource to the secondary + /// that missed it — a delete when the primary no longer has it — and + /// clears the record (#1334). + /// + /// * **Idempotent.** The push is `create_or_update` / `delete`, and a + /// secondary that already lacks a deleted resource counts as done. + /// * **Bounded.** One attempt per record per pass, no retry loop: a + /// secondary that is still down costs `limit` failed calls, and the + /// records, re-stamped, go to the back of the queue. + /// * **Safe alongside writes.** After pushing, the primary is read again; + /// if a write changed the resource meanwhile, the newer state is pushed + /// too, so the repair cannot leave behind an older version than the one + /// the write's own sync delivered. + /// + /// Without a ledger (no secondaries, or a primary that keeps none) there + /// is nothing to drain and the report is empty. + pub async fn repair_secondary_sync_failures( + &self, + limit: usize, + ) -> StorageResult { + let mut report = SyncRepairReport::default(); + let Some(recorder) = self.sync_failure_recorder() else { + return Ok(report); + }; + let Some(ledger) = recorder.ledger() else { + return Ok(report); + }; + + for record in ledger.list_sync_failures(limit).await? { + report.examined += 1; + let key = &record.key; + let Some(backend) = self.secondary(&key.backend_id) else { + warn!( + tenant = %key.tenant_id, + resource_type = %key.resource_type, + id = %key.resource_id, + backend_id = %key.backend_id, + "Dropping a needs-reindex record for a backend this composite no longer has" + ); + ledger.clear_sync_failure(key).await?; + recorder.forget(key); + report.dropped += 1; + continue; + }; + + let tenant = TenantContext::new( + TenantId::new(&key.tenant_id), + TenantPermissions::full_access(), + ); + match self.resync_current_state(&tenant, key, backend).await { + Ok(()) => { + ledger.clear_sync_failure(key).await?; + recorder.forget(key); + report.repaired += 1; + } + Err(e) => { + report.still_failing += 1; + let error = error_detail(&e); + warn!( + tenant = %key.tenant_id, + resource_type = %key.resource_type, + id = %key.resource_id, + backend_id = %key.backend_id, + operation = record.operation.as_str(), + error = %error, + "Repair of a failed secondary sync failed again; the record stays" + ); + ledger + .record_sync_failure(&SyncFailureReport { + key: key.clone(), + operation: record.operation, + attempts: 1, + error: truncate_error(&error), + failed_at: Utc::now(), + }) + .await?; + } + } + } + + report.remaining = ledger.count_sync_failures().await?; + recorder.set_outstanding(report.remaining); + Ok(report) + } + + /// Makes `backend` hold what the primary holds for `key`, right now. + async fn resync_current_state( + &self, + tenant: &TenantContext, + key: &SyncFailureKey, + backend: &DynStorage, + ) -> StorageResult<()> { + let read = || async { + match self + .primary() + .read(tenant, &key.resource_type, &key.resource_id) + .await + { + // Deleted on the primary, however the backend words it. + Err(StorageError::Resource( + ResourceError::Gone { .. } | ResourceError::NotFound { .. }, + )) => Ok(None), + other => other, + } + }; + + let mut current = read().await?; + for _ in 0..REPAIR_MAX_ROUNDS { + match ¤t { + Some(resource) => { + backend + .create_or_update( + tenant, + &key.resource_type, + &key.resource_id, + resource.content().clone(), + resource.fhir_version(), + ) + .await?; + } + None => match backend + .delete(tenant, &key.resource_type, &key.resource_id) + .await + { + Ok(()) + | Err(StorageError::Resource( + ResourceError::NotFound { .. } | ResourceError::Gone { .. }, + )) => {} + Err(e) => return Err(e), + }, + } + + let after = read().await?; + let version = |resource: &Option| { + resource.as_ref().map(|r| r.version_id().to_string()) + }; + if version(&after) == version(¤t) { + return Ok(()); + } + current = after; + } + Err(StorageError::Backend(BackendError::Unavailable { + backend_name: key.backend_id.clone(), + message: format!( + "{}/{} kept changing on the primary during repair; left for the next pass", + key.resource_type, key.resource_id + ), + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn operation_round_trips_and_unknown_is_update() { + for op in [ + SyncOperation::Create, + SyncOperation::Update, + SyncOperation::Delete, + ] { + assert_eq!(SyncOperation::from_stored(op.as_str()), op); + } + assert_eq!(SyncOperation::from_stored("?"), SyncOperation::Update); + } + + #[test] + fn errors_are_truncated_on_a_character_boundary() { + assert_eq!(truncate_error("short"), "short"); + let long = "é".repeat(MAX_RECORDED_ERROR_CHARS + 10); + let cut = truncate_error(&long); + assert_eq!(cut.chars().count(), MAX_RECORDED_ERROR_CHARS + 1); + assert!(cut.ends_with('…')); + } +} diff --git a/crates/persistence/tests/composite_secondary_sync_failures.rs b/crates/persistence/tests/composite_secondary_sync_failures.rs new file mode 100644 index 0000000000..694b552fff --- /dev/null +++ b/crates/persistence/tests/composite_secondary_sync_failures.rs @@ -0,0 +1,729 @@ +//! #1334: a composite write succeeds once the primary has committed it, in +//! every sync mode, even when a secondary refuses the change after retries. +//! That refusal must not be silent: it is counted, reported, and recorded +//! durably as "needs reindex" so the resource can be repaired. +//! +//! The secondary here is a SQLite backend behind a switch that makes every +//! indexing write fail, so "the secondary is down" is a fact of the test +//! rather than a race. Asynchronous outcomes are awaited through the sync +//! queue's barrier (`ensure_writes_visible`), never by sleeping. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; +use std::time::Duration; + +use async_trait::async_trait; +use helios_fhir::FhirVersion; +use helios_persistence::backends::sqlite::{SqliteBackend, SqliteBackendConfig}; +use helios_persistence::composite::{ + CompositeConfig, CompositeStorage, DynSearchProvider, DynStorage, RetryConfig, + SecondarySyncFailure, SecondarySyncFailureLedger, SecondarySyncObserver, SyncConfig, SyncMode, + SyncOperation, +}; +use helios_persistence::core::search::{SearchProvider, SearchResult}; +use helios_persistence::core::{BackendKind, ResourceStorage}; +use helios_persistence::error::{BackendError, StorageError, StorageResult}; +use helios_persistence::search::SearchParameterRegistry; +use helios_persistence::tenant::{TenantContext, TenantId, TenantPermissions}; +use helios_persistence::types::{ + SearchParamType, SearchParameter, SearchQuery, SearchValue, StoredResource, +}; +use serde_json::{Value, json}; + +/// Retries the sync manager makes after the first attempt. +const MAX_RETRIES: u32 = 2; +/// Attempts one final failure therefore stands for. +const ATTEMPTS: u64 = MAX_RETRIES as u64 + 1; + +const MODES: [SyncMode; 4] = [ + SyncMode::Synchronous, + SyncMode::Asynchronous, + SyncMode::Hybrid { + sync_for_search: true, + }, + SyncMode::Hybrid { + sync_for_search: false, + }, +]; + +fn tenant() -> TenantContext { + TenantContext::new(TenantId::new("default"), TenantPermissions::full_access()) +} + +fn sqlite_at(path: &str) -> SqliteBackend { + let data_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(|p| p.parent()) + .map(|p| p.join("data")) + .expect("workspace data dir"); + let backend = SqliteBackend::with_config( + path, + SqliteBackendConfig { + data_dir: Some(data_dir), + ..Default::default() + }, + ) + .expect("sqlite"); + backend.init_schema().expect("schema"); + backend +} + +/// A search secondary that can be switched off: while `failing`, every +/// indexing write is refused the way an unreachable cluster refuses it. +struct FlakyIndex { + inner: SqliteBackend, + failing: AtomicBool, + delete_calls: AtomicUsize, +} + +impl FlakyIndex { + fn new() -> Arc { + Arc::new(Self { + inner: sqlite_at(":memory:"), + failing: AtomicBool::new(false), + delete_calls: AtomicUsize::new(0), + }) + } + + fn set_failing(&self, failing: bool) { + self.failing.store(failing, Ordering::SeqCst); + } + + fn refuse(&self) -> StorageResult<()> { + if self.failing.load(Ordering::SeqCst) { + return Err(StorageError::Backend(BackendError::Unavailable { + backend_name: "flaky-index".to_string(), + message: "index is down".to_string(), + })); + } + Ok(()) + } + + async fn holds(&self, resource_type: &str, id: &str) -> Option { + self.inner + .read(&tenant(), resource_type, id) + .await + .unwrap_or(None) + } +} + +#[async_trait] +impl ResourceStorage for FlakyIndex { + fn backend_name(&self) -> &'static str { + "flaky-index" + } + + async fn create( + &self, + tenant: &TenantContext, + resource_type: &str, + resource: Value, + fhir_version: FhirVersion, + ) -> StorageResult { + self.refuse()?; + self.inner + .create(tenant, resource_type, resource, fhir_version) + .await + } + + async fn create_or_update( + &self, + tenant: &TenantContext, + resource_type: &str, + id: &str, + resource: Value, + fhir_version: FhirVersion, + ) -> StorageResult<(StoredResource, bool)> { + self.refuse()?; + self.inner + .create_or_update(tenant, resource_type, id, resource, fhir_version) + .await + } + + async fn read( + &self, + tenant: &TenantContext, + resource_type: &str, + id: &str, + ) -> StorageResult> { + self.inner.read(tenant, resource_type, id).await + } + + async fn update( + &self, + tenant: &TenantContext, + current: &StoredResource, + resource: Value, + ) -> StorageResult { + self.refuse()?; + self.inner.update(tenant, current, resource).await + } + + async fn delete( + &self, + tenant: &TenantContext, + resource_type: &str, + id: &str, + ) -> StorageResult<()> { + self.delete_calls.fetch_add(1, Ordering::SeqCst); + self.refuse()?; + self.inner.delete(tenant, resource_type, id).await + } + + async fn count( + &self, + tenant: &TenantContext, + resource_type: Option<&str>, + ) -> StorageResult { + self.inner.count(tenant, resource_type).await + } +} + +#[async_trait] +impl SearchProvider for FlakyIndex { + async fn search( + &self, + tenant: &TenantContext, + query: &SearchQuery, + ) -> StorageResult { + self.inner.search(tenant, query).await + } + + async fn search_count( + &self, + tenant: &TenantContext, + query: &SearchQuery, + ) -> StorageResult { + self.inner.search_count(tenant, query).await + } + + fn search_param_registry( + &self, + tenant: &TenantContext, + ) -> Arc> { + self.inner.search_param_registry(tenant) + } +} + +/// What the server's metrics exporter would have been told. +#[derive(Default)] +struct Metrics { + failures: Mutex>, + needs_reindex: AtomicU64, +} + +impl Metrics { + fn failures(&self) -> Vec<(String, SyncOperation)> { + self.failures.lock().unwrap().clone() + } + + fn needs_reindex(&self) -> u64 { + self.needs_reindex.load(Ordering::SeqCst) + } +} + +impl SecondarySyncObserver for Metrics { + fn sync_failed(&self, backend_id: &str, operation: SyncOperation) { + self.failures + .lock() + .unwrap() + .push((backend_id.to_string(), operation)); + } + + fn needs_reindex(&self, outstanding: u64) { + self.needs_reindex.store(outstanding, Ordering::SeqCst); + } +} + +struct Rig { + composite: CompositeStorage, + primary: Arc, + index: Arc, + metrics: Arc, +} + +impl Rig { + /// Everything queued before this call has reached the index (or failed + /// for good). A no-op wait under synchronous sync. + async fn settle(&self) { + self.composite + .ensure_writes_visible(&tenant(), &["Organization"]) + .await + .expect("sync barrier"); + } + + async fn records(&self) -> Vec { + self.primary.list_sync_failures(100).await.expect("ledger") + } + + async fn record_for(&self, id: &str) -> Option { + self.records() + .await + .into_iter() + .find(|record| record.key.resource_id == id) + } + + async fn found_by_search(&self, identifier: &str) -> usize { + self.composite + .search(&tenant(), &by_identifier(identifier)) + .await + .expect("search through composite") + .resources + .items + .len() + } +} + +/// A production-shaped composite: primary with its own index offloaded, a +/// dedicated search secondary, the primary as the failure ledger. +fn rig_on(mode: SyncMode, primary_path: &str, with_ledger: bool) -> Rig { + let mut primary = sqlite_at(primary_path); + primary.set_search_offloaded(true); + let primary = Arc::new(primary); + let index = FlakyIndex::new(); + let metrics = Arc::new(Metrics::default()); + + let config = CompositeConfig::builder() + .primary("sqlite", BackendKind::Sqlite) + .search_backend("search", BackendKind::Sqlite) + .with_sync_config(SyncConfig { + mode, + retry: RetryConfig { + max_retries: MAX_RETRIES, + initial_delay: Duration::from_millis(1), + max_delay: Duration::from_millis(2), + backoff_multiplier: 1.0, + }, + ..Default::default() + }) + .build() + .expect("composite config"); + + let mut backends: HashMap = HashMap::new(); + backends.insert("sqlite".to_string(), primary.clone() as DynStorage); + backends.insert("search".to_string(), index.clone() as DynStorage); + let mut providers: HashMap = HashMap::new(); + providers.insert("sqlite".to_string(), primary.clone() as DynSearchProvider); + providers.insert("search".to_string(), index.clone() as DynSearchProvider); + + let mut composite = CompositeStorage::new(config, backends) + .expect("composite") + .with_search_providers(providers) + .with_full_primary(primary.clone()) + .with_sync_observer(metrics.clone()); + if with_ledger { + composite = composite.with_sync_failure_ledger(primary.clone()); + } + + Rig { + composite: composite.start_sync_workers(), + primary, + index, + metrics, + } +} + +fn rig(mode: SyncMode) -> Rig { + rig_on(mode, ":memory:", true) +} + +fn organization(identifier: &str, name: &str) -> Value { + json!({ + "resourceType": "Organization", + "identifier": [{"system": "urn:zzz:probe", "value": identifier}], + "name": name + }) +} + +fn by_identifier(identifier: &str) -> SearchQuery { + SearchQuery::new("Organization").with_parameter(SearchParameter { + name: "identifier".to_string(), + param_type: SearchParamType::Token, + modifier: None, + values: vec![SearchValue::token(Some("urn:zzz:probe"), identifier)], + chain: vec![], + components: vec![], + }) +} + +/// The whole contract, in every sync mode: create, update and delete against +/// a secondary that is down all succeed, each final failure is counted exactly +/// once, one record per resource says what is owed, a later successful write +/// clears its record, and the repair drains the rest. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn failed_syncs_are_counted_recorded_cleared_and_repaired_in_every_mode() { + for mode in MODES { + let rig = rig(mode); + let t = tenant(); + + // Positive control: a healthy secondary indexes the write, and + // nothing is counted or recorded. + let kept = rig + .composite + .create( + &t, + "Organization", + organization("KEPT", "v1"), + FhirVersion::R4, + ) + .await + .expect("healthy create"); + rig.settle().await; + assert_eq!(rig.found_by_search("KEPT").await, 1, "{mode:?}: control"); + assert!(rig.metrics.failures().is_empty(), "{mode:?}"); + assert!(rig.records().await.is_empty(), "{mode:?}"); + + // The secondary goes down. Every write still succeeds. + rig.index.set_failing(true); + let lost = rig + .composite + .create( + &t, + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .unwrap_or_else(|e| panic!("{mode:?}: create must succeed: {e}")); + rig.settle().await; + assert_eq!( + rig.metrics.failures(), + vec![("search".to_string(), SyncOperation::Create)], + "{mode:?}: one final failure, counted once (not once per retry)" + ); + let record = rig.record_for(lost.id()).await.expect("create recorded"); + assert_eq!(record.key.tenant_id, "default"); + assert_eq!(record.key.resource_type, "Organization"); + assert_eq!(record.key.backend_id, "search"); + assert_eq!(record.operation, SyncOperation::Create); + assert_eq!(record.attempts, ATTEMPTS, "{mode:?}"); + assert!(record.last_error.contains("index is down"), "{mode:?}"); + assert!( + !record.last_error.contains("urn:zzz:probe"), + "no resource content on the record" + ); + assert_eq!(rig.found_by_search("LOST").await, 0, "{mode:?}: it is lost"); + + let updated = rig + .composite + .update(&t, &kept, organization("KEPT", "v2")) + .await + .unwrap_or_else(|e| panic!("{mode:?}: update must succeed: {e}")); + rig.settle().await; + assert_eq!(updated.version_id(), "2"); + assert_eq!(rig.metrics.failures().len(), 2, "{mode:?}"); + assert_eq!( + rig.metrics.failures()[1], + ("search".to_string(), SyncOperation::Update) + ); + let record = rig.record_for(kept.id()).await.expect("update recorded"); + assert_eq!(record.operation, SyncOperation::Update); + + rig.composite + .delete(&t, "Organization", kept.id()) + .await + .unwrap_or_else(|e| panic!("{mode:?}: delete must succeed: {e}")); + rig.settle().await; + assert_eq!(rig.metrics.failures().len(), 3, "{mode:?}"); + assert_eq!( + rig.metrics.failures()[2], + ("search".to_string(), SyncOperation::Delete) + ); + // Same resource, same secondary: folded into the record it had. + assert_eq!(rig.records().await.len(), 2, "{mode:?}: one per resource"); + let folded = rig.record_for(kept.id()).await.expect("still recorded"); + assert_eq!(folded.operation, SyncOperation::Delete); + assert_eq!(folded.attempts, 2 * ATTEMPTS, "{mode:?}"); + assert_eq!(folded.first_failed_at, record.first_failed_at, "{mode:?}"); + assert_eq!(rig.metrics.needs_reindex(), 2, "{mode:?}: gauge"); + + // The secondary comes back. A successful write of a recorded + // resource delivers its current state, so its record is cleared. + rig.index.set_failing(false); + rig.composite + .update(&t, &lost, organization("LOST", "v2")) + .await + .expect("update after recovery"); + rig.settle().await; + assert!(rig.record_for(lost.id()).await.is_none(), "{mode:?}"); + assert_eq!(rig.found_by_search("LOST").await, 1, "{mode:?}"); + assert_eq!(rig.metrics.needs_reindex(), 1, "{mode:?}: gauge"); + + // The repair takes care of the one nobody wrote again: deleted on + // the primary, still indexed on the secondary. + assert!(rig.index.holds("Organization", kept.id()).await.is_some()); + let report = rig + .composite + .repair_secondary_sync_failures(10) + .await + .expect("repair"); + assert_eq!((report.examined, report.repaired), (1, 1), "{mode:?}"); + assert_eq!(report.remaining, 0, "{mode:?}"); + assert!(rig.index.holds("Organization", kept.id()).await.is_none()); + assert_eq!(rig.found_by_search("KEPT").await, 0, "{mode:?}"); + assert!(rig.records().await.is_empty(), "{mode:?}"); + assert_eq!(rig.metrics.needs_reindex(), 0, "{mode:?}: gauge"); + assert_eq!( + rig.metrics.failures().len(), + 3, + "{mode:?}: recovery and repair count nothing" + ); + + // Idempotent: nothing left, nothing done. + let again = rig.composite.repair_secondary_sync_failures(10).await; + assert_eq!(again.expect("repair").examined, 0, "{mode:?}"); + } +} + +/// The repair pushes the primary's current state of a resource the secondary +/// never received, and a secondary that is still down leaves the record in +/// place without counting a new write failure. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn repair_resyncs_a_create_the_secondary_never_received() { + for mode in MODES { + let rig = rig(mode); + let t = tenant(); + + rig.index.set_failing(true); + let lost = rig + .composite + .create( + &t, + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .expect("create"); + rig.settle().await; + assert_eq!(rig.metrics.failures().len(), 1, "{mode:?}"); + + // Still down: bounded, one attempt, the record stays and ages. + let report = rig + .composite + .repair_secondary_sync_failures(10) + .await + .expect("repair against a down secondary is not an error"); + assert_eq!((report.repaired, report.still_failing), (0, 1), "{mode:?}"); + assert_eq!(report.remaining, 1, "{mode:?}"); + let record = rig.record_for(lost.id()).await.expect("record stays"); + assert_eq!(record.attempts, ATTEMPTS + 1, "{mode:?}"); + assert_eq!(rig.metrics.failures().len(), 1, "{mode:?}: writes only"); + + rig.index.set_failing(false); + let report = rig + .composite + .repair_secondary_sync_failures(10) + .await + .expect("repair"); + assert_eq!((report.repaired, report.remaining), (1, 0), "{mode:?}"); + assert_eq!(rig.found_by_search("LOST").await, 1, "{mode:?}: searchable"); + assert!(rig.records().await.is_empty(), "{mode:?}"); + } +} + +/// The record lives in the primary, so it outlives the process: re-open the +/// primary's database and the resource is still owed, and still repairable. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn records_survive_a_restart_of_the_primary() { + let dir = tempfile::tempdir().expect("tempdir"); + let path: &Path = &dir.path().join("primary.db"); + let path = path.to_str().expect("utf-8 temp path"); + + let lost_id = { + let rig = rig_on(SyncMode::Asynchronous, path, true); + rig.index.set_failing(true); + let lost = rig + .composite + .create( + &tenant(), + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .expect("create"); + rig.settle().await; + assert!(rig.record_for(lost.id()).await.is_some()); + lost.id().to_string() + }; + + // A new process: fresh composite, fresh (empty) index, same database. + let rig = rig_on(SyncMode::Asynchronous, path, true); + assert_eq!(rig.found_by_search("LOST").await, 0, "control: not indexed"); + let record = rig.record_for(&lost_id).await.expect("record survived"); + assert_eq!(record.operation, SyncOperation::Create); + + let report = rig + .composite + .repair_secondary_sync_failures(10) + .await + .expect("repair"); + assert_eq!((report.repaired, report.remaining), (1, 0)); + assert_eq!(rig.found_by_search("LOST").await, 1); +} + +/// A record left by an earlier run is cleared by the first successful write +/// of that resource in this one, not only by the repair. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_successful_write_clears_a_record_left_by_an_earlier_run() { + let dir = tempfile::tempdir().expect("tempdir"); + let path = dir.path().join("primary.db"); + let path = path.to_str().expect("utf-8 temp path"); + + let lost = { + let rig = rig_on(SyncMode::Synchronous, path, true); + rig.index.set_failing(true); + rig.composite + .create( + &tenant(), + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .expect("create") + }; + + let rig = rig_on(SyncMode::Synchronous, path, true); + rig.composite + .update(&tenant(), &lost, organization("LOST", "v2")) + .await + .expect("update"); + assert!(rig.records().await.is_empty()); + assert_eq!(rig.found_by_search("LOST").await, 1); + assert_eq!(rig.metrics.needs_reindex(), 0); +} + +/// Deleting what the secondary does not have is the delete having worked, +/// not a failure to retry: one call, nothing counted, and the record the +/// missed create left behind is settled by it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn delete_of_a_resource_the_secondary_lacks_is_success() { + for mode in MODES { + let rig = rig(mode); + let t = tenant(); + + rig.index.set_failing(true); + let lost = rig + .composite + .create( + &t, + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .expect("create"); + rig.settle().await; + rig.index.set_failing(false); + + rig.composite + .delete(&t, "Organization", lost.id()) + .await + .expect("delete"); + rig.settle().await; + + assert_eq!( + rig.index.delete_calls.load(Ordering::SeqCst), + 1, + "{mode:?}: NotFound on a delete must not be retried" + ); + assert_eq!(rig.metrics.failures().len(), 1, "{mode:?}: only the create"); + assert!(rig.records().await.is_empty(), "{mode:?}: in sync again"); + } +} + +/// The batch path (`create_many`: conformance seeding, bulk loads) reports +/// each resource the secondary refused, not the batch as one anonymous unit. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_refused_batch_is_recorded_per_resource() { + for mode in MODES { + let rig = rig(mode); + rig.index.set_failing(true); + let results = rig + .composite + .create_many( + &tenant(), + "Organization", + vec![organization("A", "a"), organization("B", "b")], + FhirVersion::R4, + ) + .await; + assert!(results.iter().all(Result::is_ok), "{mode:?}"); + rig.settle().await; + + assert_eq!(rig.metrics.failures().len(), 2, "{mode:?}"); + assert_eq!(rig.records().await.len(), 2, "{mode:?}"); + + rig.index.set_failing(false); + let report = rig.composite.repair_secondary_sync_failures(10).await; + assert_eq!(report.expect("repair").repaired, 2, "{mode:?}"); + assert_eq!(rig.found_by_search("A").await, 1, "{mode:?}"); + assert_eq!(rig.found_by_search("B").await, 1, "{mode:?}"); + } +} + +/// A primary with no ledger (S3): the write still succeeds and the failure +/// is still counted and logged; there is simply nothing to list or drain. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn without_a_ledger_failures_are_still_counted() { + for mode in MODES { + let rig = rig_on(mode, ":memory:", false); + rig.index.set_failing(true); + rig.composite + .create( + &tenant(), + "Organization", + organization("LOST", "v1"), + FhirVersion::R4, + ) + .await + .expect("create"); + rig.settle().await; + + assert_eq!(rig.metrics.failures().len(), 1, "{mode:?}"); + assert!(rig.records().await.is_empty(), "{mode:?}"); + let report = rig.composite.repair_secondary_sync_failures(10).await; + assert_eq!(report.expect("repair").examined, 0, "{mode:?}"); + } +} + +/// The bound is honoured, and a record that keeps failing goes to the back +/// of the queue rather than starving the ones behind it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn repair_is_bounded_and_rotates_through_the_queue() { + let rig = rig(SyncMode::Synchronous); + rig.index.set_failing(true); + for identifier in ["A", "B", "C"] { + rig.composite + .create( + &tenant(), + "Organization", + organization(identifier, "v1"), + FhirVersion::R4, + ) + .await + .expect("create"); + } + let first = rig.records().await[0].key.clone(); + + let report = rig.composite.repair_secondary_sync_failures(1).await; + let report = report.expect("repair"); + assert_eq!((report.examined, report.still_failing), (1, 1)); + assert_eq!(report.remaining, 3); + assert_ne!( + rig.records().await[0].key, + first, + "the failed one moved back" + ); + + rig.index.set_failing(false); + let report = rig.composite.repair_secondary_sync_failures(2).await; + let report = report.expect("repair"); + assert_eq!((report.examined, report.repaired), (2, 2)); + assert_eq!(report.remaining, 1); +} From 1ac19d1452fcbdf3c90bcd8dee3cb7a4673f06a5 Mon Sep 17 00:00:00 2001 From: smunini Date: Mon, 21 Sep 2026 15:08:53 -0400 Subject: [PATCH 2/3] feat(hfs): export secondary sync failure metrics and repair them periodically Wires #1334's recorder into the server's four Elasticsearch composites: - `helios_observability::composite_metrics`: `composite_secondary_sync_failures_total{backend,operation}` and the `composite_secondary_sync_needs_reindex` gauge. Named like the crate's other metrics (no `hfs_` prefix; the exporter's global `service` label tells servers apart). Labels are the configured backend id and a fixed operation set only: `/metrics` is public, and tenant / type / id are unbounded. - SQLite, PostgreSQL and MongoDB primaries are their composite's ledger. S3 has none and says so at startup: metric + event only. - A periodic task, spawned like the search-parameter refresh tasks, drains the ledger through `repair_secondary_sync_failures` (HFS_COMPOSITE_SYNC_REPAIR_INTERVAL, default 60 s, 0 = off; HFS_COMPOSITE_SYNC_REPAIR_BATCH, default 100). It runs once at startup so records left by an earlier run are picked up. Adds the backend-agnostic ledger contract suite and runs it on SQLite, PostgreSQL and MongoDB, and documents the operator-facing behaviour in the run-hfs-server skill. Fixes #1334 --- .claude/skills/run-hfs-server/SKILL.md | 16 +++ crates/hfs/src/main.rs | 103 ++++++++++++++++ crates/observability/src/composite_metrics.rs | 101 ++++++++++++++++ crates/observability/src/lib.rs | 4 + .../tests/common/sync_failure_ledger_suite.rs | 114 ++++++++++++++++++ .../composite_secondary_sync_failures.rs | 10 ++ crates/persistence/tests/mongodb_tests.rs | 19 +++ crates/persistence/tests/postgres_tests.rs | 16 +++ 8 files changed, 383 insertions(+) create mode 100644 crates/observability/src/composite_metrics.rs create mode 100644 crates/persistence/tests/common/sync_failure_ledger_suite.rs diff --git a/.claude/skills/run-hfs-server/SKILL.md b/.claude/skills/run-hfs-server/SKILL.md index 5303f2b637..afb219747d 100644 --- a/.claude/skills/run-hfs-server/SKILL.md +++ b/.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 | diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index a65fb7615e..12566c16f6 100644 --- a/crates/hfs/src/main.rs +++ b/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, + 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, + "Repaired failed secondary syncs" + ), + 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. diff --git a/crates/observability/src/composite_metrics.rs b/crates/observability/src/composite_metrics.rs new file mode 100644 index 0000000000..b145852a92 --- /dev/null +++ b/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}" + ); + } + } + } +} diff --git a/crates/observability/src/lib.rs b/crates/observability/src/lib.rs index 27b4c77ede..2b86f01798 100644 --- a/crates/observability/src/lib.rs +++ b/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; diff --git a/crates/persistence/tests/common/sync_failure_ledger_suite.rs b/crates/persistence/tests/common/sync_failure_ledger_suite.rs new file mode 100644 index 0000000000..bf12a55fee --- /dev/null +++ b/crates/persistence/tests/common/sync_failure_ledger_suite.rs @@ -0,0 +1,114 @@ +//! Backend-agnostic contract for [`SecondarySyncFailureLedger`] (#1334): the +//! durable "needs reindex" records a composite keeps in its primary. +//! +//! The ledger is shared by every test that runs against the same database, +//! so everything here is scoped to a tenant id unique to the caller and the +//! global count is only ever compared with itself. + +use chrono::{Duration, Utc}; +use helios_persistence::composite::{ + SecondarySyncFailure, SecondarySyncFailureLedger, SyncFailureKey, SyncFailureReport, + SyncOperation, +}; + +fn key(tenant: &str, id: &str, backend: &str) -> SyncFailureKey { + SyncFailureKey { + tenant_id: tenant.to_string(), + resource_type: "Patient".to_string(), + resource_id: id.to_string(), + backend_id: backend.to_string(), + } +} + +async fn mine(ledger: &dyn SecondarySyncFailureLedger, tenant: &str) -> Vec { + ledger + .list_sync_failures(100_000) + .await + .expect("list") + .into_iter() + .filter(|record| record.key.tenant_id == tenant) + .collect() +} + +/// Records fold per (tenant, type, id, backend), list least-recently-failed +/// first, and clear exactly once. +pub async fn ledger_folds_orders_clears_and_counts( + ledger: &dyn SecondarySyncFailureLedger, + tenant: &str, +) { + assert!( + mine(ledger, tenant).await.is_empty(), + "control: clean slate" + ); + let before = ledger.count_sync_failures().await.expect("count"); + + let t0 = Utc::now() - Duration::seconds(30); + let report = |id: &str, backend: &str, operation, seconds, error: &str| SyncFailureReport { + key: key(tenant, id, backend), + operation, + attempts: 4, + error: error.to_string(), + failed_at: t0 + Duration::seconds(seconds), + }; + + let first = report("a", "es", SyncOperation::Create, 0, "first"); + assert!(ledger.record_sync_failure(&first).await.expect("record")); + assert!( + ledger + .record_sync_failure(&report("b", "es", SyncOperation::Update, 1, "b")) + .await + .expect("record") + ); + // Same resource, another secondary: its own record. + assert!( + ledger + .record_sync_failure(&report("a", "graph", SyncOperation::Create, 2, "g")) + .await + .expect("record") + ); + // Same resource, same secondary: folded, not inserted. + assert!( + !ledger + .record_sync_failure(&report("a", "es", SyncOperation::Delete, 3, "latest")) + .await + .expect("record") + ); + + let records = mine(ledger, tenant).await; + let order: Vec<_> = records + .iter() + .map(|r| (r.key.resource_id.as_str(), r.key.backend_id.as_str())) + .collect(); + assert_eq!( + order, + [("b", "es"), ("a", "graph"), ("a", "es")], + "least recently failed first; the re-failed record moved to the back" + ); + let folded = &records[2]; + assert_eq!(folded.operation, SyncOperation::Delete); + assert_eq!(folded.attempts, 8); + assert_eq!(folded.last_error, "latest"); + // Stores keep at least millisecond precision. + let drift = + |a: chrono::DateTime, b: chrono::DateTime| (a - b).num_milliseconds().abs(); + assert!(drift(folded.first_failed_at, first.failed_at) <= 1); + assert!(drift(folded.last_failed_at, t0 + Duration::seconds(3)) <= 1); + assert_eq!( + ledger.count_sync_failures().await.expect("count"), + before + 3 + ); + + // The limit is honoured. + assert_eq!(ledger.list_sync_failures(1).await.expect("list").len(), 1); + + assert!(ledger.clear_sync_failure(&first.key).await.expect("clear")); + assert!( + !ledger.clear_sync_failure(&first.key).await.expect("clear"), + "already gone" + ); + for record in mine(ledger, tenant).await { + assert!(ledger.clear_sync_failure(&record.key).await.expect("clear")); + } + assert!(mine(ledger, tenant).await.is_empty()); + assert_eq!(ledger.count_sync_failures().await.expect("count"), before); +} diff --git a/crates/persistence/tests/composite_secondary_sync_failures.rs b/crates/persistence/tests/composite_secondary_sync_failures.rs index 694b552fff..42f3c234a0 100644 --- a/crates/persistence/tests/composite_secondary_sync_failures.rs +++ b/crates/persistence/tests/composite_secondary_sync_failures.rs @@ -349,6 +349,16 @@ fn by_identifier(identifier: &str) -> SearchQuery { }) } +/// The ledger contract PostgreSQL and MongoDB are held to as well. +#[path = "common/sync_failure_ledger_suite.rs"] +mod sync_failure_ledger_suite; + +#[tokio::test] +async fn sqlite_sync_failure_ledger_contract() { + let backend = sqlite_at(":memory:"); + sync_failure_ledger_suite::ledger_folds_orders_clears_and_counts(&backend, "ledger-1334").await; +} + /// The whole contract, in every sync mode: create, update and delete against /// a secondary that is down all succeed, each final failure is counted exactly /// once, one record per resource says what is owed, a later successful write diff --git a/crates/persistence/tests/mongodb_tests.rs b/crates/persistence/tests/mongodb_tests.rs index aa73e7d832..e0cc0c903d 100644 --- a/crates/persistence/tests/mongodb_tests.rs +++ b/crates/persistence/tests/mongodb_tests.rs @@ -15514,3 +15514,22 @@ async fn mongodb_integration_composite_multi_batch_driver_paging() { .expect("search_count must agree with search on the same multi-batch query"); assert_eq!(count as usize, MATCHING); } + +/// The backend-agnostic contract of the secondary sync failure ledger +/// (#1334). Same `#[path]` arrangement as the search suites. +#[path = "common/sync_failure_ledger_suite.rs"] +mod sync_failure_ledger_suite; + +/// #1334: the "needs reindex" ledger a composite keeps in this primary. +#[tokio::test] +async fn mongodb_sync_failure_ledger_contract() { + let Some(backend) = create_backend("sync_failure_ledger_1334").await else { + eprintln!("skipping: no MongoDB container available"); + return; + }; + sync_failure_ledger_suite::ledger_folds_orders_clears_and_counts( + &backend, + &format!("ledger-1334-{}", uuid::Uuid::new_v4()), + ) + .await; +} diff --git a/crates/persistence/tests/postgres_tests.rs b/crates/persistence/tests/postgres_tests.rs index 9c856f40d9..9f51afcee0 100644 --- a/crates/persistence/tests/postgres_tests.rs +++ b/crates/persistence/tests/postgres_tests.rs @@ -123,6 +123,11 @@ mod versioned_write_race_suite; #[path = "search/conditional_patch_suite.rs"] mod conditional_patch_suite; +/// The backend-agnostic contract of the secondary sync failure ledger +/// (#1334). Same `#[path]` arrangement. +#[path = "common/sync_failure_ledger_suite.rs"] +mod sync_failure_ledger_suite; + #[path = "common/container_cleanup.rs"] mod container_cleanup; @@ -19313,4 +19318,15 @@ mod postgres_integration { ) .await; } + + /// #1334: the "needs reindex" ledger a composite keeps in this primary. + #[tokio::test] + async fn postgres_integration_sync_failure_ledger_contract() { + let backend = create_backend().await; + super::sync_failure_ledger_suite::ledger_folds_orders_clears_and_counts( + &backend, + &format!("ledger-1334-{}", uuid::Uuid::new_v4()), + ) + .await; + } } From 8a95d81726f5adeb941179b8a554a3d1574f0c2e Mon Sep 17 00:00:00 2001 From: smunini Date: Mon, 21 Sep 2026 15:51:50 -0400 Subject: [PATCH 3/3] fix(composite): report version 1 for a failed create sync; name the repair pass log Found running hfs against a stopped Elasticsearch: a stored body need not carry meta.versionId, so the structured event said version="" for every failed create. A create is version 1 unless the content says otherwise. The periodic task's log line said "Repaired" even for a pass that repaired nothing; it now names the pass and lets the counts speak. Refs #1334 --- crates/hfs/src/main.rs | 2 +- crates/persistence/src/composite/sync_failures.rs | 13 +++++++++---- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/crates/hfs/src/main.rs b/crates/hfs/src/main.rs index 12566c16f6..e1f90b4c9c 100644 --- a/crates/hfs/src/main.rs +++ b/crates/hfs/src/main.rs @@ -222,7 +222,7 @@ fn spawn_secondary_sync_repair( still_failing = report.still_failing, dropped = report.dropped, remaining = report.remaining, - "Repaired failed secondary syncs" + "Secondary sync repair pass finished" ), Ok(_) => {} Err(e) => tracing::warn!("Repair of failed secondary syncs could not run: {e}"), diff --git a/crates/persistence/src/composite/sync_failures.rs b/crates/persistence/src/composite/sync_failures.rs index 2b3a039166..bc4ff7350e 100644 --- a/crates/persistence/src/composite/sync_failures.rs +++ b/crates/persistence/src/composite/sync_failures.rs @@ -200,10 +200,15 @@ fn event_subjects(event: &SyncEvent) -> Vec<(String, String, SyncOperation, Opti resource_type.clone(), resource_id.clone(), SyncOperation::Create, - content - .pointer("/meta/versionId") - .and_then(|v| v.as_str()) - .map(str::to_string), + // A stored body need not carry `meta`; a create is version 1 + // unless the content says otherwise. + Some( + content + .pointer("/meta/versionId") + .and_then(|v| v.as_str()) + .unwrap_or("1") + .to_string(), + ), )], SyncEvent::Update { resource_type,