From 93a1684f068603b354ba3c05957a25459c73de05 Mon Sep 17 00:00:00 2001 From: DanConwayDev Date: Fri, 9 Jan 2026 14:12:24 +0000 Subject: feat(sync): add ConnectedDegraded status for failed historic sync - Add ConnectionStatus::ConnectedDegraded (status=4 in metrics) - Track batch failures via PendingBatch.failed field - Track relay-level failures via RelayState.historic_sync_had_failures - Transition to ConnectedDegraded when any batch fails during historic sync - Add is_live_sync_active() helper for cleaner match patterns - Update state machine diagram with ConnectedDegraded transitions - Update metrics docs with status=4 and example queries Fixes issue where relays with failed negentropy retries would incorrectly transition to Connected status despite missing data. Now operators can distinguish 'fully synced' vs 'degraded (partial data)'. --- docs/explanation/grasp-02-proactive-sync.md | 29 ++++++-- docs/explanation/monitoring.md | 9 ++- src/sync/algorithms.rs | 2 + src/sync/metrics.rs | 4 +- src/sync/mod.rs | 104 ++++++++++++++++++++++++---- 5 files changed, 124 insertions(+), 24 deletions(-) diff --git a/docs/explanation/grasp-02-proactive-sync.md b/docs/explanation/grasp-02-proactive-sync.md index e1fb367..b17b8bf 100644 --- a/docs/explanation/grasp-02-proactive-sync.md +++ b/docs/explanation/grasp-02-proactive-sync.md @@ -79,6 +79,8 @@ pub enum ConnectionStatus { Syncing, /// Successfully connected, historic sync completed Connected, + /// Successfully connected, historic sync failed but live sync active + ConnectedDegraded, } /// Complete state for a single relay - combines sync needs with connection lifecycle @@ -207,15 +209,19 @@ stateDiagram-v2 Disconnected --> Connecting: retry_disconnected_relays → try_connect_relay Connecting --> Syncing: success → handle_connect_or_reconnect Connecting --> Disconnected: failure + record in health tracker - Syncing --> Connected: all historic batches complete → check_and_complete_historic_sync + Syncing --> Connected: all batches succeed → check_and_complete_historic_sync + Syncing --> ConnectedDegraded: any batch failed → check_and_complete_historic_sync Syncing --> Disconnected: connection lost → handle_disconnect Connected --> Disconnected: connection lost → handle_disconnect + ConnectedDegraded --> Disconnected: connection lost → handle_disconnect Connected --> [*]: intentional disconnect via check_disconnects + ConnectedDegraded --> [*]: intentional disconnect via check_disconnects note right of Disconnected: disconnected_at set for 15min rule
RelayConnection kept in HashMap note right of Connecting: connection attempt with timeout note right of Syncing: historic sync in progress
event loop spawned here note right of Connected: historic sync complete
last_connected tracked for since filter + note right of ConnectedDegraded: historic sync failed (missing events)
live sync active, partial data ``` ### Connection Flow Methods @@ -240,17 +246,28 @@ When a relay first connects, it enters the **Syncing** state and begins historic Each layer creates one or more `PendingBatch` entries tracked in `PendingSyncIndex`. As EOSE messages arrive: - `handle_eose()` confirms each batch via `confirm_batch()` -- `confirm_batch()` moves items to confirmed state and calls `check_and_complete_historic_sync()` -- `check_and_complete_historic_sync()` checks if `PendingSyncIndex` is empty for this relay -- When empty: transitions `Syncing` → `Connected`, sets `historic_sync_completed = true` +- `confirm_batch()` moves items to confirmed state, tracks if batch failed, and calls `check_and_complete_historic_sync()` +- `check_and_complete_historic_sync()` uses a **double-check pattern** to avoid race conditions: + 1. First check: Are there pending batches? If yes, return early + 2. Wait 6 seconds (batch window + buffer) for self-subscriber to process in-flight events + 3. Second check: Are there still no pending batches? If yes, return early + 4. If no pending batches after wait: + - If any batch failed: transition `Syncing` → `ConnectedDegraded` + - If all batches succeeded: transition `Syncing` → `Connected` + - Set `historic_sync_completed = true` + +**Why the double-check?** There's an async gap between receiving EOSE and the self-subscriber processing events to create Layer 2/3 filters. The 6-second wait (5s batch window + 1s buffer) ensures we don't prematurely mark sync complete while Layer 2/3 batches are being created. + +**Batch Failure Tracking**: When negentropy retry protection triggers (relay returns zero requested events on retry), the batch is marked as `failed = true`. This causes the relay to transition to `ConnectedDegraded` instead of `Connected`, signaling that live sync is active but historic sync is incomplete. **Metrics tracking**: The `ngit_sync_relay_connected` metric shows: - `0` = Disconnected -- `1` = Connecting +- `1` = Connecting - `2` = Syncing (historic sync in progress) - `3` = Connected (historic sync complete, live sync active) +- `4` = ConnectedDegraded (historic sync failed, live sync active, partial data) -This allows operators to monitor sync progress and distinguish between "connected but still catching up" vs "fully synced and live". +This allows operators to monitor sync progress and distinguish between "connected but still catching up" vs "fully synced and live" vs "degraded (missing historic data)". ### Event Loop Lifecycle diff --git a/docs/explanation/monitoring.md b/docs/explanation/monitoring.md index d2d20c0..cc164ab 100644 --- a/docs/explanation/monitoring.md +++ b/docs/explanation/monitoring.md @@ -98,7 +98,7 @@ When GRASP-02 proactive sync is implemented, the following metrics will be added | Metric | Type | Labels | Description | |--------|------|--------|-------------| -| `ngit_sync_relay_connected` | Gauge | relay | Connection status (0=disconnected, 1=connecting, 2=syncing, 3=connected) | +| `ngit_sync_relay_connected` | Gauge | relay | Connection status (0=disconnected, 1=connecting, 2=syncing, 3=connected, 4=connected_degraded) | | `ngit_sync_connection_attempts_total` | Counter | relay, result | Connection attempt outcomes | | `ngit_sync_relay_status` | Gauge | relay | Health status (1=healthy, 2=disconnected, 3=degraded, 4=dead, 5=rate_limited) | | `ngit_sync_relay_failures` | Gauge | relay | Current consecutive failure count | @@ -115,8 +115,9 @@ The `ngit_sync_relay_connected` metric tracks the connection lifecycle: - `1` = **Connecting** - Connection attempt in progress - `2` = **Syncing** - Connected, historic sync in progress - `3` = **Connected** - Connected, historic sync complete, live sync active +- `4` = **ConnectedDegraded** - Connected, historic sync failed, live sync active, partial data -This allows operators to distinguish between "connected but still catching up" (Syncing) vs "fully synced and live" (Connected). +This allows operators to distinguish between "connected but still catching up" (Syncing) vs "fully synced and live" (Connected) vs "degraded - missing historic data" (ConnectedDegraded). ### Relay Health States @@ -136,10 +137,14 @@ sum by (relay) (ngit_sync_relay_connected == 0) # Disconnected sum by (relay) (ngit_sync_relay_connected == 1) # Connecting sum by (relay) (ngit_sync_relay_connected == 2) # Syncing sum by (relay) (ngit_sync_relay_connected == 3) # Connected +sum by (relay) (ngit_sync_relay_connected == 4) # ConnectedDegraded # Relays still syncing (not yet fully caught up) count(ngit_sync_relay_connected == 2) +# Relays with degraded sync (missing historic data) +count(ngit_sync_relay_connected == 4) + # Connection success rate over last hour sum(rate(ngit_sync_connection_attempts_total{result="success"}[1h])) / sum(rate(ngit_sync_connection_attempts_total[1h])) diff --git a/src/sync/algorithms.rs b/src/sync/algorithms.rs index e083dc8..7536f41 100644 --- a/src/sync/algorithms.rs +++ b/src/sync/algorithms.rs @@ -405,6 +405,7 @@ mod tests { requested_event_ids: None, received_event_ids: None, retry_count: 0, + failed: false, }], ); @@ -520,6 +521,7 @@ mod tests { requested_event_ids: None, received_event_ids: None, retry_count: 0, + failed: false, }], ); diff --git a/src/sync/metrics.rs b/src/sync/metrics.rs index db7dd20..0f56911 100644 --- a/src/sync/metrics.rs +++ b/src/sync/metrics.rs @@ -53,7 +53,7 @@ impl SyncMetrics { let relay_connected = IntGaugeVec::new( Opts::new( "ngit_sync_relay_connected", - "Relay connection status (0=disconnected, 1=connecting, 2=syncing, 3=connected)", + "Relay connection status (0=disconnected, 1=connecting, 2=syncing, 3=connected, 4=connected_degraded)", ), &["relay"], )?; @@ -208,6 +208,7 @@ impl SyncMetrics { /// - Connecting = 1 (connection attempt in progress) /// - Syncing = 2 (connected, historic sync in progress) /// - Connected = 3 (connected, historic sync complete) + /// - ConnectedDegraded = 4 (connected, historic sync failed but live sync active) /// /// This is separate from health state and provides more granular connection lifecycle tracking. /// @@ -222,6 +223,7 @@ impl SyncMetrics { ConnectionStatus::Connecting => 1, ConnectionStatus::Syncing => 2, ConnectionStatus::Connected => 3, + ConnectionStatus::ConnectedDegraded => 4, }; self.relay_connected .with_label_values(&[relay]) diff --git a/src/sync/mod.rs b/src/sync/mod.rs index e5b724d..2031ef4 100644 --- a/src/sync/mod.rs +++ b/src/sync/mod.rs @@ -94,6 +94,18 @@ pub enum ConnectionStatus { Syncing, /// Successfully connected, historic sync completed Connected, + /// Successfully connected, historic sync failed but live sync active + ConnectedDegraded, +} + +impl ConnectionStatus { + /// Returns true if live sync is active (can accept new filters) + pub fn is_live_sync_active(&self) -> bool { + matches!( + self, + ConnectionStatus::Syncing | ConnectionStatus::Connected | ConnectionStatus::ConnectedDegraded + ) + } } /// Complete state for a single relay - combines sync needs with connection lifecycle @@ -119,6 +131,10 @@ pub struct RelayState { pub historic_sync_completed: bool, /// When historic sync completed (None if never completed or cleared on fresh_start) pub historic_sync_completed_at: Option, + /// Whether any batch failed during historic sync + /// Set to true when retry protection triggers or other failures occur + /// Used to transition to ConnectedDegraded instead of Connected + pub historic_sync_had_failures: bool, } impl Default for RelayState { @@ -133,6 +149,7 @@ impl Default for RelayState { announcements_synced: false, historic_sync_completed: false, historic_sync_completed_at: None, + historic_sync_had_failures: false, } } } @@ -156,6 +173,7 @@ impl RelayState { self.announcements_synced = false; self.historic_sync_completed = false; self.historic_sync_completed_at = None; + self.historic_sync_had_failures = false; } } @@ -216,6 +234,9 @@ pub struct PendingBatch { /// Number of retry attempts for missing events (Negentropy only) /// Used to prevent infinite retry loops when relay consistently fails pub retry_count: usize, + /// Whether this batch failed (completed with missing data) + /// Set to true when retry protection triggers or other failures occur + pub failed: bool, } /// Items included in a pending batch @@ -682,9 +703,10 @@ impl SyncManager { // TODO: Track this failure in Prometheus metrics (sync_failed_batches_total) ); - // Extract and complete batch with partial results + // Extract and complete batch with partial results, marking as failed let batch_idx_for_completion = batch_idx; - let completed_batch = batches.remove(batch_idx_for_completion); + let mut completed_batch = batches.remove(batch_idx_for_completion); + completed_batch.failed = true; // Mark as failed for ConnectedDegraded transition if batches.is_empty() { pending.remove(relay_url); } @@ -849,6 +871,16 @@ impl SyncManager { ); } + // Track if this batch failed (for ConnectedDegraded transition) + if batch.failed { + state.historic_sync_had_failures = true; + tracing::warn!( + relay = %relay_url, + batch_id = batch_id, + "Batch failed - will transition to ConnectedDegraded instead of Connected" + ); + } + // DEBUG TRACING: Log the root events being confirmed tracing::info!( relay = %relay_url, @@ -862,6 +894,7 @@ impl SyncManager { all_root_events = ?state.root_events.iter().map(|id| id.to_hex()).collect::>(), is_generic_filter = is_generic_filter, announcements_synced = state.announcements_synced, + had_failures = state.historic_sync_had_failures, "Batch confirmed - items moved from pending to confirmed" ); } else { @@ -881,13 +914,24 @@ impl SyncManager { /// Check if historic sync is complete and transition to Connected status /// - /// This method checks if there are any pending batches for the relay. - /// If no pending batches exist and the relay is in Syncing status, - /// it transitions to Connected and updates metrics. + /// This method uses a double-check pattern to avoid race conditions with + /// the self-subscriber's batching window. The sequence is: + /// + /// 1. First check: Are there pending batches? + /// 2. Wait for batch window + buffer (6 seconds) + /// 3. Second check: Are there still no pending batches? + /// 4. If still no pending batches, transition to Connected + /// + /// This ensures that events received just before the first check have time + /// to be batched and create Layer 2/3 filters before we mark sync complete. + /// + /// The 6-second delay is based on: + /// - Self-subscriber batch window: 5 seconds (configurable via NGIT_SYNC_BATCH_WINDOW_MS) + /// - Buffer for processing: 1 second /// /// Called after each batch is confirmed to detect completion. async fn check_and_complete_historic_sync(&self, relay_url: &str) { - // Check if there are any pending batches + // First check: Are there any pending batches? let has_pending = { let pending = self.pending_sync_index.read().await; pending.get(relay_url).map_or(false, |batches| !batches.is_empty()) @@ -898,12 +942,33 @@ impl SyncManager { return; } - // No pending batches - check if we should transition to Connected + // Wait for self-subscriber batch window + buffer to catch any in-flight events + // that might create new Layer 2/3 filters + tokio::time::sleep(Duration::from_millis(6000)).await; + + // Second check: Are there still no pending batches? + let has_pending = { + let pending = self.pending_sync_index.read().await; + pending.get(relay_url).map_or(false, |batches| !batches.is_empty()) + }; + + if has_pending { + // New batches appeared during the wait - still syncing + return; + } + + // No pending batches after waiting - safe to transition to Connected or ConnectedDegraded let mut relay_index = self.relay_sync_index.write().await; if let Some(state) = relay_index.get_mut(relay_url) { if state.connection_status == ConnectionStatus::Syncing { - // Transition to Connected - state.connection_status = ConnectionStatus::Connected; + // Check if any batches failed during historic sync + let new_status = if state.historic_sync_had_failures { + ConnectionStatus::ConnectedDegraded + } else { + ConnectionStatus::Connected + }; + + state.connection_status = new_status; state.historic_sync_completed = true; state.historic_sync_completed_at = Some(Timestamp::now()); @@ -911,12 +976,15 @@ impl SyncManager { relay = %relay_url, repos_synced = state.repos.len(), root_events_synced = state.root_events.len(), - "Historic sync complete - transitioned to Connected status" + had_failures = state.historic_sync_had_failures, + status = ?new_status, + "Historic sync complete - transitioned to {} status", + if state.historic_sync_had_failures { "ConnectedDegraded" } else { "Connected" } ); // Update metrics if let Some(ref metrics) = self.metrics { - metrics.record_connection_status(relay_url, ConnectionStatus::Connected); + metrics.record_connection_status(relay_url, new_status); } } } @@ -1147,8 +1215,8 @@ impl SyncManager { ); return; } - Some(ConnectionStatus::Syncing) | Some(ConnectionStatus::Connected) => { - // Continue to subscribe - both Syncing and Connected can accept new filters + Some(status) if status.is_live_sync_active() => { + // Continue to subscribe - live sync is active, can accept new filters } } @@ -1514,8 +1582,8 @@ impl SyncManager { "Cleared sync state in fresh_start" ); } - // Only sync if we're connected (either Syncing or fully Connected) - if matches!(state.connection_status, ConnectionStatus::Syncing | ConnectionStatus::Connected) { + // Only sync if we're connected (live sync active) + if state.connection_status.is_live_sync_active() { drop(index); self.sync_generic_filters(relay_url, None).await; // Step 5: compute_actions for L2+L3 (will be triggered by EOSE) @@ -2474,6 +2542,7 @@ impl SyncManager { requested_event_ids: None, // Will be set after negentropy diff received_event_ids: None, // Will be set after negentropy diff retry_count: 0, + failed: false, }; // Add to pending_sync_index @@ -2712,6 +2781,7 @@ impl SyncManager { requested_event_ids: None, // Not used for REQ+EOSE received_event_ids: None, // Not used for REQ+EOSE retry_count: 0, // Not used for REQ+EOSE + failed: false, }; // Add to pending_sync_index @@ -2925,12 +2995,14 @@ mod tests { requested_event_ids: Some(HashSet::new()), received_event_ids: Some(HashSet::new()), retry_count: 0, + failed: false, }; assert!(batch.requested_event_ids.is_some()); assert!(batch.received_event_ids.is_some()); assert_eq!(batch.sync_method, SyncMethod::Negentropy); assert_eq!(batch.retry_count, 0); + assert!(!batch.failed); } #[test] @@ -2945,10 +3017,12 @@ mod tests { requested_event_ids: None, received_event_ids: None, retry_count: 0, + failed: false, }; assert!(batch.requested_event_ids.is_none()); assert!(batch.received_event_ids.is_none()); assert_eq!(batch.sync_method, SyncMethod::ReqEose); + assert!(!batch.failed); } } -- cgit v1.2.3