From 1e16e06f8ae5bac216849f30362c375ac83f8de6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Sat, 15 Aug 2026 10:19:28 +0800 Subject: [PATCH] test(site-replication): pin the background retry-drain discipline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Red-light evidence for backlog#1675 P1-3: the retry queue has no background consumer — every failed peer delivery waits for a manual repair. The new tests specify the drain rules before the drain exists: - classification: IAM / bucket-meta entries collapse per (peer, path) with no persisted body, so only a bootstrap-plan snapshot resend is a faithful replay; make-with-versioning / configure-replication are re-derivable per bucket; peer edits are re-sent under a fresh generation; destructive bucket ops and internal: marker records (pending-endpoint-refresh backup store) are never background-replayed - exponential backoff (600s * 2^(n-1), 24h ceiling) gates every attempt, otherwise a dead peer's entries hit the failed threshold within 30 minutes of reconcile ticks - the actionable subset respects classification, current peer membership and backoff All fail against the placeholder implementations (no consumer). --- rustfs/src/admin/handlers/site_replication.rs | 166 ++++++++++++++++++ 1 file changed, 166 insertions(+) diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 471a624d4..6ee4024bc 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -6171,6 +6171,48 @@ fn retry_event_replayed_by_bootstrap(event: &SiteReplicationRetryEvent) -> bool ) } +/// Exponential backoff base for the background retry drain, aligned with the +/// reconcile cadence (`site_replication_reconcile::RECONCILE_INTERVAL`). +const SITE_REPLICATION_RETRY_DRAIN_BASE_BACKOFF_SECS: i64 = 600; +/// Backoff ceiling: a permanently failed peer is still probed daily. +const SITE_REPLICATION_RETRY_DRAIN_MAX_BACKOFF_SECS: i64 = 86_400; + +/// What the background drain may do for one retry event. Everything not +/// representable here is operator territory (manual repair). +#[derive(Debug, Clone, PartialEq, Eq)] +enum RetryDrainAction { + /// Constant-path IAM item deliveries collapse into one queue entry per + /// peer and their bodies are not persisted; the only faithful replay is + /// the current IAM snapshot from the bootstrap plan. + IamSnapshot, + /// Same collapse for bucket-meta deliveries: replay the bucket metadata + /// snapshot from the bootstrap plan. + BucketMetadataSnapshot, + /// A self-contained bucket op the bootstrap plan can re-derive for its + /// bucket (`make-with-versioning` / `configure-replication`). + BucketOpReplay { operation: String, bucket: String }, + /// Re-send the current peer records under a fresh edit generation. + PeerEdit, +} + +fn classify_site_replication_retry_event(_event: &SiteReplicationRetryEvent) -> Option { + // No background consumer exists yet: every event waits for a manual + // repair. + None +} + +/// Whether the drain may attempt this event now. +fn site_replication_retry_backoff_elapsed(event: &SiteReplicationRetryEvent, now: OffsetDateTime) -> bool { + let _ = (event, now); + true +} + +/// The subset of the retry queue the background drain is allowed to touch. +fn actionable_site_replication_retry_events(state: &SiteReplicationState, now: OffsetDateTime) -> Vec { + let _ = now; + state.retry_queue.clone() +} + /// Remove a retry event for (peer, path) from the queue on successful delivery. /// This is a no-op (load + no-op persist skipped) when no matching entry exists, /// avoiding unnecessary I/O on the common path. @@ -11427,6 +11469,130 @@ mod tests { assert!(target_state.peers["remote"].skip_tls_verify); } + fn drain_event(peer: &str, path: &str, retry_count: u32, updated_at: Option) -> SiteReplicationRetryEvent { + SiteReplicationRetryEvent { + id: format!("evt-{peer}"), + peer_deployment_id: peer.to_string(), + peer_endpoint: format!("https://{peer}.example.com"), + path: path.to_string(), + retry_count, + failed: retry_count >= SITE_REPLICATION_RETRY_FAILED_AFTER, + last_error: "remote-operation-failed".to_string(), + updated_at, + edit_generation: None, + } + } + + /// P1-3 red-light: the drain must only ever act on deliveries it can + /// replay faithfully. IAM / bucket-meta entries collapse per (peer, path) + /// with no body persisted — only a snapshot resend is truthful; bucket + /// makes/replication configs are re-derivable; destructive bucket ops and + /// `internal:` marker records (the pending-endpoint-refresh backup store) + /// are never background-replayed. + #[test] + fn test_classify_site_replication_retry_event_actions() { + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + let classify = |path: &str| classify_site_replication_retry_event(&drain_event("remote", path, 1, Some(now))); + + assert_eq!( + classify("/rustfs/admin/v3/site-replication/peer/iam-item"), + Some(RetryDrainAction::IamSnapshot) + ); + assert_eq!( + classify("/rustfs/admin/v3/site-replication/peer/bucket-meta"), + Some(RetryDrainAction::BucketMetadataSnapshot) + ); + assert_eq!(classify(SITE_REPLICATION_PEER_EDIT_PATH), Some(RetryDrainAction::PeerEdit)); + assert_eq!( + classify( + "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=make-with-versioning&createdAt=1" + ), + Some(RetryDrainAction::BucketOpReplay { + operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(), + bucket: "photos".to_string(), + }) + ); + assert_eq!( + classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=configure-replication"), + Some(RetryDrainAction::BucketOpReplay { + operation: SITE_REPLICATION_BUCKET_OP_CONFIGURE_REPLICATION.to_string(), + bucket: "photos".to_string(), + }) + ); + // Destructive ops are operator territory: replaying a bucket delete + // against a peer whose bucket was since recreated is irreversible. + assert_eq!( + classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket"), + None + ); + assert_eq!( + classify("/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=force-delete-bucket"), + None + ); + // `internal:` records store payloads in `last_error`, not failures. + assert_eq!(classify(SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH), None); + assert_eq!(classify("internal:some-future-marker"), None); + assert_eq!(classify("/rustfs/admin/v3/site-replication/peer/unknown"), None); + } + + /// Exponential backoff gates every attempt: without it a dead peer's + /// entries hit `failed` (retry_count >= 3) within 30 minutes of reconcile + /// ticks and the retry stats lose their signal. + #[test] + fn test_site_replication_retry_backoff_schedule() { + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + let at = |secs_ago: i64| Some(now - time::Duration::seconds(secs_ago)); + let elapsed = |retry_count: u32, secs_ago: i64| { + site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", retry_count, at(secs_ago)), now) + }; + + // No record of when it failed: attempt now. + assert!(site_replication_retry_backoff_elapsed(&drain_event("remote", "/p", 1, None), now)); + // First failure: one reconcile interval. + assert!(!elapsed(1, 599)); + assert!(elapsed(1, 601)); + // Third failure: 600 * 2^2 = 2400s. + assert!(!elapsed(3, 1200)); + assert!(elapsed(3, 2401)); + // Ceiling: a long-dead peer is still probed daily, never less often. + assert!(!elapsed(30, 86_000)); + assert!(elapsed(30, 86_401)); + } + + /// The actionable subset respects classification, peer membership and + /// backoff; everything else stays untouched in the queue. + #[test] + fn test_actionable_site_replication_retry_events_filters() { + let now = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("timestamp"); + let old = Some(now - time::Duration::seconds(700)); + let mut state = SiteReplicationState::default(); + state + .peers + .insert("remote".to_string(), peer("remote", "https://remote.example.com")); + + state.retry_queue = vec![ + // Eligible: known peer, replayable, past backoff. + drain_event("remote", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old), + // Not yet due. + drain_event("remote", "/rustfs/admin/v3/site-replication/peer/bucket-meta", 2, Some(now)), + // Unknown peer (removed since the failure was recorded). + drain_event("gone", "/rustfs/admin/v3/site-replication/peer/iam-item", 1, old), + // Marker record, not a delivery failure. + drain_event("remote", SITE_REPLICATION_ENDPOINT_REFRESH_RETRY_PATH, 0, old), + // Destructive op: operator-only. + drain_event( + "remote", + "/rustfs/admin/v3/site-replication/peer/bucket-ops?bucket=photos&operation=delete-bucket", + 1, + old, + ), + ]; + + let actionable = actionable_site_replication_retry_events(&state, now); + assert_eq!(actionable.len(), 1, "only the due, replayable, known-peer event is actionable"); + assert_eq!(actionable[0].path, "/rustfs/admin/v3/site-replication/peer/iam-item"); + } + #[test] fn test_pending_endpoint_refresh_retry_summary_redacts_pem() { let pem = "-----BEGIN CERTIFICATE-----\nsecret-marker\n-----END CERTIFICATE-----";