🐛 fix(ha): fence cross-datacenter transfers (#1859)

* 🐛 fix(ha): fence cross-DC transfers

Pending placements survived a restart without a durable way to distinguish an
abandoned transfer from a live copy. That blocked recovery or risked two owners
publishing the same placement.

Bind each transfer to the consensus term and a monotonic attempt token. A newer
term can reclaim pending work, while stale checkpoints and terminal writes fail
their ownership check.

* 🧪 test(ha): set placement attempt

Author: gaborbernat

crates/peryx-driver/tests/unit/availability/tests.rs

@@ -121,6 +121,7 @@ fn blob_placement_view_projects_and_orders_states() {
             },
             state: state_value,
             fence: 1,
+            transfer_attempt: 1,
             generation: 1,
             updated_at_unix: updated_at,
         };

crates/peryx-ha-distributed/src/copy_planning.rs

@@ -25,7 +25,11 @@ pub struct CrossDcCopy {
     pub fence: NonZeroU64,
 }
 
-pub fn copy_backlog_entry(records: &[BlobPlacementRecord], local_dc: &DataCenterId) -> Option<CopyBacklogEntry> {
+pub fn copy_backlog_entry(
+    records: &[BlobPlacementRecord],
+    local_dc: &DataCenterId,
+    fence: NonZeroU64,
+) -> Option<CopyBacklogEntry> {
     let mut local_settled = false;
     let mut sources = Vec::new();
     for record in records {
@@ -36,11 +40,10 @@ pub fn copy_backlog_entry(records: &[BlobPlacementRecord], local_dc: &DataCenter
                 generation: record.generation,
                 size,
             }),
-            BlobPlacementState::Verified { .. } | BlobPlacementState::Pending | BlobPlacementState::Revoked
-                if is_local =>
-            {
+            BlobPlacementState::Verified { .. } | BlobPlacementState::Revoked if is_local => {
                 local_settled = true;
             }
+            BlobPlacementState::Pending if is_local && record.fence >= fence.get() => local_settled = true,
             _ => {}
         }
     }

crates/peryx-ha-distributed/src/copy_runtime.rs

@@ -13,7 +13,7 @@ use crate::{BlobTransport, CopyError, HttpBlobTransport, TransferLimits, Transpo
 use peryx_core::Clock;
 use peryx_ha::{
     AvailabilityTaskError, AvailabilityTaskReport, BackendId, BlobPlacementFailure, BlobPlacementKey,
-    BlobPlacementTransition, DataCenterId,
+    BlobPlacementOutcome, BlobPlacementTransition, DataCenterId,
 };
 use peryx_storage::blob::{BlobStore, Digest};
 use peryx_storage::meta::MetaStore;
@@ -95,7 +95,7 @@ impl CrossDcBlobCopier {
                 .scan_blob_placement_groups(cursor.as_deref(), batch)
                 .map_err(|error| task_error("copy_backlog_scan", error))?;
             planned.extend(page.groups.iter().filter_map(|records| {
-                copy_backlog_entry(records, &self.local_dc)
+                copy_backlog_entry(records, &self.local_dc, fence)
                     .map(|entry| plan_cross_dc_copy(&entry, &self.local_dc, &self.backend, fence))
             }));
             match page.next_cursor {
@@ -112,27 +112,29 @@ impl CrossDcBlobCopier {
             tracing::warn!(source_dc, "cross-datacenter copy has no reachable source peer");
             return false;
         };
-        if !record(
+        let Some(BlobPlacementOutcome::Applied(staged)) = record(
             meta,
             &copy.target,
             &BlobPlacementTransition::Stage,
             copy.fence.get(),
             clock,
-        ) {
+        ) else {
             return false;
-        }
+        };
         let digest = Digest::from_hex(copy.target.digest.sha256()).expect("artifact digests are validated SHA-256");
         let outcome = copy_blob_to_target(transport, &self.store, &digest).await;
         let transition = match &outcome {
             Ok(()) => BlobPlacementTransition::Verify {
+                attempt: staged.transfer_attempt,
                 observed: copy.target.digest.clone(),
                 size: copy.size,
             },
             Err(error) => BlobPlacementTransition::Fail {
+                attempt: staged.transfer_attempt,
                 class: failure_class(error),
             },
         };
-        let recorded = record(meta, &copy.target, &transition, copy.fence.get(), clock);
+        let recorded = record(meta, &copy.target, &transition, copy.fence.get(), clock).is_some();
         outcome.is_ok() && recorded
     }
 }
@@ -173,12 +175,12 @@ fn record(
     transition: &BlobPlacementTransition,
     fence: u64,
     clock: &Clock,
-) -> bool {
+) -> Option<BlobPlacementOutcome> {
     match apply_blob_placement(meta, key, transition, fence, (clock)()) {
-        Ok(_) => true,
+        Ok(outcome) => Some(outcome),
         Err(error) => {
             tracing::warn!(%error, ?transition, "cross-datacenter copy could not record a placement");
-            false
+            None
         }
     }
 }

crates/peryx-ha-distributed/src/placement_policy.rs

@@ -96,11 +96,12 @@ pub fn record_local_placement(
     {
         return Ok(BlobPlacementOutcome::Unchanged(record));
     }
-    apply_blob_placement(meta, &key, &BlobPlacementTransition::Stage, fence, now)?;
+    let staged = apply_blob_placement(meta, &key, &BlobPlacementTransition::Stage, fence, now)?;
     apply_blob_placement(
         meta,
         &key,
         &BlobPlacementTransition::Verify {
+            attempt: staged.record().transfer_attempt,
             observed: digest.clone(),
             size,
         },

crates/peryx-ha-distributed/src/placement_runtime.rs

@@ -6,8 +6,8 @@ use std::num::NonZeroUsize;
 
 use peryx_core::Clock;
 use peryx_ha::{
-    AvailabilityTaskError, AvailabilityTaskReport, BlobPlacementFailure, BlobPlacementKey, BlobPlacementState,
-    BlobPlacementTransition, DataCenterId, ReclamationState, ReclamationStore,
+    AvailabilityTaskError, AvailabilityTaskReport, BlobPlacementKey, BlobPlacementState, BlobPlacementTransition,
+    DataCenterId, ReclamationState, ReclamationStore,
 };
 use peryx_identity::ArtifactDigest;
 use peryx_storage::blob::{BlobErrorKind, BlobStore, Digest};
@@ -97,15 +97,7 @@ impl FilesystemPlacementReconciler {
             return false;
         }
         // Stop routing to corrupt bytes before clearing the path for a replacement.
-        if !record_transition(
-            meta,
-            key,
-            &BlobPlacementTransition::Fail {
-                class: BlobPlacementFailure::DigestMismatch,
-            },
-            fence,
-            clock,
-        ) {
+        if !record_transition(meta, key, &BlobPlacementTransition::Invalidate, fence, clock) {
             return false;
         }
         if let Err(error) = self.store.remove(&digest) {

crates/peryx-ha-distributed/tests/unit/blob_placements_http_tests.rs

@@ -52,6 +52,7 @@ async fn build_app(corrupt: bool) -> (tempfile::TempDir, Arc<AppState>) {
     drop(authorization);
     drop(users);
     let verify = BlobPlacementTransition::Verify {
+        attempt: 1,
         observed: digest(),
         size: 4096,
     };
@@ -63,6 +64,7 @@ async fn build_app(corrupt: bool) -> (tempfile::TempDir, Arc<AppState>) {
         &[
             BlobPlacementTransition::Stage,
             BlobPlacementTransition::Fail {
+                attempt: 1,
                 class: BlobPlacementFailure::SourceUnavailable,
             },
         ],

crates/peryx-ha-distributed/tests/unit/blob_plane_tests.rs

@@ -78,6 +78,7 @@ fn seed_verified_placement_on(meta: &MetaStore, digest: &Digest, dc: &str, backe
         meta,
         &key,
         &BlobPlacementTransition::Verify {
+            attempt: 1,
             observed: artifact,
             size,
         },

crates/peryx-ha-distributed/tests/unit/copy_planning_tests.rs

@@ -30,6 +30,7 @@ fn record(
         key: key(suffix, data_center, location),
         state,
         fence: 1,
+        transfer_attempt: 1,
         generation,
         updated_at_unix: 0,
     }
@@ -43,6 +44,7 @@ fn test_backlog_selects_verified_remote_sources() {
             record(1, "south", "south/01", 4, BlobPlacementState::Verified { size: 20 }),
         ],
         &DataCenterId::new("west").unwrap(),
+        NonZeroU64::new(1).unwrap(),
     )
     .unwrap();
 
@@ -66,6 +68,7 @@ fn test_backlog_breaks_generation_ties_by_ledger_order() {
             record(1, "south", "south/01", 4, BlobPlacementState::Verified { size: 20 }),
         ],
         &DataCenterId::new("west").unwrap(),
+        NonZeroU64::new(1).unwrap(),
     )
     .unwrap();
 
@@ -94,6 +97,7 @@ fn test_backlog_skips_settled_local_placements() {
                     record(1, "west", "west/01", 1, state),
                 ],
                 &local,
+                NonZeroU64::new(1).unwrap(),
             )
             .is_none()
         );
@@ -116,6 +120,21 @@ fn test_backlog_retries_failed_local_placements() {
             ),
         ],
         &DataCenterId::new("west").unwrap(),
+        NonZeroU64::new(1).unwrap(),
+    );
+
+    assert!(entry.is_some());
+}
+
+#[test]
+fn test_backlog_retries_pending_after_an_ownership_transition() {
+    let entry = copy_backlog_entry(
+        &[
+            record(1, "east", "east/01", 1, BlobPlacementState::Verified { size: 10 }),
+            record(1, "west", "west/01", 1, BlobPlacementState::Pending),
+        ],
+        &DataCenterId::new("west").unwrap(),
+        NonZeroU64::new(2).unwrap(),
     );
 
     assert!(entry.is_some());
@@ -125,6 +144,13 @@ fn test_backlog_retries_failed_local_placements() {
 fn test_backlog_requires_records_and_a_verified_source() {
     let local = DataCenterId::new("west").unwrap();
 
-    assert!(copy_backlog_entry(&[], &local).is_none());
-    assert!(copy_backlog_entry(&[record(1, "east", "east/01", 1, BlobPlacementState::Pending)], &local).is_none());
+    assert!(copy_backlog_entry(&[], &local, NonZeroU64::new(1).unwrap()).is_none());
+    assert!(
+        copy_backlog_entry(
+            &[record(1, "east", "east/01", 1, BlobPlacementState::Pending)],
+            &local,
+            NonZeroU64::new(1).unwrap(),
+        )
+        .is_none()
+    );
 }