From cdee4ce6906bac1ec65ddaad3957d8aa098488a3 Mon Sep 17 00:00:00 2001
From: Zhe Wu <zhe.wu@mystenlabs.com>
Date: Thu, 2 Jul 2026 13:31:55 -0700
Subject: [PATCH 1/4] feat: keep node recovery task running across epoch
 changes

When an epoch change starts while node recovery is in progress, the
recovery task was canceled and restarted from scratch, re-scanning the
full blob info table on every epoch change even though the previous
progress was still valid.

The recovery task now keeps running across epoch changes:

- The epoch-change path only advances the recovery target
  (RecoveryInProgress epoch) and starts shard syncs for gained shards;
  both happen under a new status mutex so that they are atomic with
  respect to the recovery task's completion. A new task is only started
  if the running task completed concurrently with the epoch change or
  stopped unexpectedly; a spurious start is harmless since the extra
  task exits once it observes that the node is already active.
- The task's scan bound stays frozen at the epoch it was started with:
  blobs certified later are covered by event processing, and shards
  gained at later epoch changes are covered by shard sync, which the
  task waits for.
- On completion, the task attests epoch sync done for the recovery
  target currently recorded in the node status (which epoch changes
  advance), instead of the frozen spawn epoch whose attestation would
  be dropped as stale.
- Attestation requires a clean scan pass during which no shard sync
  started, tracked by a new shard sync generation counter: a shard sync
  that starts and terminally fails during a pass leaves missing blobs
  behind that only a new scan pass finds. The blob scan also pauses
  promptly when a shard sync starts mid-pass, since per-blob recovery
  would redundantly decode slivers for shards that shard sync copies
  in bulk.

The shard-gain simtest now also asserts that epoch changes processed
while recovering do not restart the recovery task.
---
 crates/walrus-service/src/node.rs             |  46 +++--
 .../walrus-service/src/node/node_recovery.rs  | 182 ++++++++++++++----
 crates/walrus-service/src/node/shard_sync.rs  |  31 ++-
 crates/walrus-simtest/tests/simtest_core.rs   |  16 +-
 4 files changed, 214 insertions(+), 61 deletions(-)

diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs
index b982b9d3c8..152c54963f 100644
--- a/crates/walrus-service/src/node.rs
+++ b/crates/walrus-service/src/node.rs
@@ -2295,13 +2295,14 @@ impl StorageNode {
         if let NodeStatus::RecoveryInProgress(recovering_epoch) =
             self.inner.storage.node_status()?
         {
-            // If the node is already in recovery mode, we need to restart node recovery to recover
-            // to the latest epoch. This is to make sure that the node is always recovering to the
-            // latest epoch. Since the node is up-to-date with events, newly gained shards are
-            // synced from their previous owners instead of being filled by blob recovery.
+            // If the node is already in recovery mode, we advance the recovery target to the
+            // latest epoch, so that the node always recovers to the latest epoch. Since the node
+            // is up-to-date with events, newly gained shards are synced from their previous
+            // owners instead of being filled by blob recovery, and the running recovery task
+            // keeps its progress instead of being restarted.
             tracing::info!(
-                "node is currently recovering to epoch {recovering_epoch}, restarting \
-                node recovery to recover to the latest epoch {}",
+                "node is currently recovering to epoch {recovering_epoch}, advancing the \
+                recovery target to the latest epoch {}",
                 event.epoch
             );
             self.process_shard_changes_in_new_epoch_while_recovering(
@@ -2405,9 +2406,9 @@ impl StorageNode {
     /// In contrast to [`Self::process_shard_changes_in_new_epoch_and_start_node_recovery`], which
     /// handles a node that has lost track of the previous epoch's shard assignment, the node here
     /// is up-to-date with events, so newly gained shards are filled using shard sync from their
-    /// previous owners instead of per-blob recovery. The restarted node recovery task waits for
-    /// these shard syncs to finish before recovering blobs, and attests epoch sync done once both
-    /// are complete.
+    /// previous owners instead of per-blob recovery. The running node recovery task is not
+    /// restarted: it waits for these shard syncs to finish before recovering blobs, and attests
+    /// epoch sync done for the advanced recovery target once both are complete.
     ///
     /// As all functions that are passed an [`EventHandle`], this is responsible for marking the
     /// event as completed.
@@ -2417,9 +2418,23 @@ impl StorageNode {
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
     ) -> anyhow::Result<()> {
+        // Advancing the recovery target and starting the shard syncs for gained shards must be
+        // atomic with respect to the recovery task's completion, which checks both under the
+        // same mutex: a completing task either observes the advanced target together with the
+        // new shard syncs, or completes entirely before this transition (detected below via the
+        // node status, in which case a new task is started).
+        let status_guard = self.node_recovery_handler.lock_status().await;
+
+        // If the running recovery task completed concurrently (after this event handler decided
+        // to take the recovering path), it has flipped the node status away from
+        // RecoveryInProgress; its completion no longer covers this epoch change.
+        let recovery_task_completed_concurrently = !matches!(
+            self.inner.storage.node_status()?,
+            NodeStatus::RecoveryInProgress(_)
+        );
+
         // Advance the recovery target so that the recovery task attests epoch sync done for the
-        // latest epoch; a stale attestation would be dropped by the contract service. This must
-        // happen before starting the shard syncs and restarting the recovery task below.
+        // latest epoch; a stale attestation would be dropped by the contract service.
         self.inner
             .set_node_status(NodeStatus::RecoveryInProgress(event.epoch))?;
 
@@ -2440,6 +2455,8 @@ impl StorageNode {
         self.create_new_shards_and_start_sync(shard_map_lock, shards_gained, &committees, false)
             .await?;
 
+        drop(status_guard);
+
         // For shards that just moved out, we need to lock them to not store more data in them.
         for shard_id in shard_diff_calculator.shards_to_lock() {
             let Some(shard_storage) = self.inner.storage.shard_storage(*shard_id).await else {
@@ -2457,10 +2474,11 @@ impl StorageNode {
                 .context("failed to lock shard")?;
         }
 
-        // Restart node recovery to recover to the latest epoch. The recovery task waits for the
-        // shard syncs started above to finish before scanning for blobs to recover.
+        // The recovery task keeps running across epoch changes: it waits for the shard syncs
+        // started above to finish before recovering blobs, and attests epoch sync done for the
+        // advanced target on completion. A new task is only started when none is running.
         self.node_recovery_handler
-            .start_node_recovery(event.epoch)
+            .ensure_recovery_task_running(event.epoch, recovery_task_completed_concurrently)
             .await?;
 
         // The recovery task is in charge of attesting epoch sync done, so the finisher is always
diff --git a/crates/walrus-service/src/node/node_recovery.rs b/crates/walrus-service/src/node/node_recovery.rs
index fd450cca03..05a2fc8437 100644
--- a/crates/walrus-service/src/node/node_recovery.rs
+++ b/crates/walrus-service/src/node/node_recovery.rs
@@ -37,6 +37,14 @@ pub struct NodeRecoveryHandler {
     // There can be at most one background shard removal task at a time.
     task_handle: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
 
+    // Serializes recovery-related node status transitions: the epoch-change path advances the
+    // recovery target and starts shard syncs while holding this mutex, and the recovery task
+    // transitions the node to `Active` while holding it. This guarantees that a completing
+    // recovery task either observes the advanced target together with the new shard syncs, or
+    // has completed entirely before the transition (in which case the epoch-change path starts
+    // a new task).
+    status_mutex: Arc<Mutex<()>>,
+
     // Configuration for node recovery.
     config: NodeRecoveryConfig,
 }
@@ -53,14 +61,63 @@ impl NodeRecoveryHandler {
             blob_sync_handler,
             shard_sync_handler,
             task_handle: Arc::new(Mutex::new(None)),
+            status_mutex: Arc::new(Mutex::new(())),
             config,
         }
     }
 
+    /// Locks the recovery status mutex.
+    ///
+    /// The epoch-change path must hold the returned guard across advancing the recovery target
+    /// (setting the node status to a newer `RecoveryInProgress` epoch) and starting the shard
+    /// syncs for newly gained shards, so that these are atomic with respect to the recovery
+    /// task's completion.
+    pub async fn lock_status(&self) -> tokio::sync::MutexGuard<'_, ()> {
+        self.status_mutex.lock().await
+    }
+
+    /// Ensures a recovery task is running to recover to the given epoch.
+    ///
+    /// The recovery task keeps running across epoch changes and picks up the advanced recovery
+    /// target on its own, so this normally does nothing. A new task is only started if the
+    /// caller observed that the previous task completed concurrently with the epoch change
+    /// (`task_completed_concurrently`), or if no task is running (for example, because it
+    /// stopped unexpectedly).
+    pub async fn ensure_recovery_task_running(
+        &self,
+        epoch: Epoch,
+        task_completed_concurrently: bool,
+    ) -> Result<(), TypedStoreError> {
+        let task_running = self
+            .task_handle
+            .lock()
+            .await
+            .as_ref()
+            .is_some_and(|handle| !handle.is_finished());
+
+        if task_completed_concurrently || !task_running {
+            tracing::info!(
+                walrus.epoch = epoch,
+                task_completed_concurrently,
+                task_running,
+                "no running node recovery task; starting a new one"
+            );
+            self.start_node_recovery(epoch).await?;
+        }
+
+        Ok(())
+    }
+
     /// Starts the node recovery process to recover blobs that are certified before the given epoch.
     /// For blobs that are certified after `certified_before_epoch`, the event processing is in
     /// charge of making sure the blob is stored at all shards.
     ///
+    /// The task keeps running across epoch changes: blobs certified after the scan bound are
+    /// covered by event processing, and shards gained at later epoch changes are covered by shard
+    /// sync (which the task waits for), so the scan bound stays valid. On completion, the task
+    /// attests epoch sync done for the recovery target currently recorded in the node status,
+    /// which the epoch-change path advances.
+    ///
     /// Any existing recovery task will be canceled.
     // TODO(WAL-864): Refactor this function to make it readable.
     pub async fn start_node_recovery(
@@ -80,6 +137,7 @@ impl NodeRecoveryHandler {
         let node = self.node.clone();
         let blob_sync_handler = self.blob_sync_handler.clone();
         let shard_sync_handler = self.shard_sync_handler.clone();
+        let status_mutex = self.status_mutex.clone();
         let max_concurrent_blob_syncs_during_recovery =
             self.config.max_concurrent_blob_syncs_during_recovery;
         let task_handle = tokio::spawn(async move {
@@ -99,6 +157,14 @@ impl NodeRecoveryHandler {
             fail_point_async!("start_node_recovery_entry");
 
             loop {
+                // Snapshot the shard sync generation before waiting: a completed pass may only
+                // attest if no shard sync started since this point. A shard sync that starts
+                // (and possibly terminally fails) during the pass leaves blobs behind that only
+                // a new scan pass finds, so a generation change forces a re-scan. Taking the
+                // snapshot before the wait errs towards a spurious re-scan when syncs start and
+                // drain while parked, never towards a missed one.
+                let sync_generation_before_scan = shard_sync_handler.sync_generation();
+
                 // Block until the node is ready to run a recovery scan pass. Stops the recovery
                 // task if the node started catching up.
                 match wait_until_ready_to_scan(&node, &shard_sync_handler).await {
@@ -127,6 +193,19 @@ impl NodeRecoveryHandler {
                             .ok()
                     })
                 {
+                    // An epoch change may have started shard syncs for newly gained shards while
+                    // this pass is running. Stop starting new blob syncs and park: per-blob
+                    // recovery would redundantly decode slivers for the shards that shard sync is
+                    // copying in bulk. Marking the pass as having more blobs makes the loop drain
+                    // the in-flight syncs and re-scan after the park.
+                    if shard_sync_handler.has_sync_in_progress() {
+                        tracing::info!(
+                            "shard sync started during recovery scan pass; pausing blob recovery"
+                        );
+                        has_more_blobs = true;
+                        break;
+                    }
+
                     node.metrics
                         .node_recovery_recover_blob_progress
                         .set(i64::from(blob_id.first_two_bytes()));
@@ -235,54 +314,77 @@ impl NodeRecoveryHandler {
                     }
                 }
 
-                if !has_more_blobs {
-                    tracing::info!("no recovery blob found; stop recovery task");
-                    break;
-                }
-
                 // Wait for all ongoing syncs to complete
                 while (ongoing_syncs.next().await).is_some() {
                     // Each sync completion automatically releases its permit
                 }
 
-                // TODO(WAL-669): right now, we have to do one more loop to check if all the blobs
-                // are recovered. This is not efficient because checking blob existence is
-                // expensive. It's better that blob sync handler can return the blob sync status
-                // and we can avoid the extra loop of all the blob syncs finished successfully.
-            }
-
-            // A shard sync may have been started by an epoch change after the last scan
-            // pass began. The node is fully synced for the epoch only once both blob
-            // recovery and all shard syncs are complete, so drain them before attesting.
-            shard_sync_handler.wait_until_no_sync_in_progress().await;
+                if has_more_blobs {
+                    // TODO(WAL-669): right now, we have to do one more loop to check if all the
+                    // blobs are recovered. This is not efficient because checking blob existence
+                    // is expensive. It's better that blob sync handler can return the blob sync
+                    // status and we can avoid the extra loop of all the blob syncs finished
+                    // successfully.
+                    continue;
+                }
 
-            let current_node_status = node
-                .storage
-                .node_status()
-                .expect("reading node status should not fail");
-            if current_node_status == NodeStatus::RecoveryInProgress(certified_before_epoch) {
-                tracing::info!("node recovery task finished; set node status to active");
-                match node.set_node_status(NodeStatus::Active) {
-                    Ok(()) => {
-                        // While the node is recovering, this is the only place that attests
-                        // epoch sync done: shard sync skips its own attestation in
-                        // RecoveryInProgress state (see the epoch_sync_done handling in
-                        // shard_sync.rs), so that the attestation covers both the recovered
-                        // blobs and all synced shards.
-                        node.contract_service
-                            .epoch_sync_done(certified_before_epoch, node.node_capability())
-                            .await
-                    }
-                    Err(error) => {
-                        tracing::error!(?error, "failed to set node status to active");
-                    }
+                // Completion attempt. The node is fully synced for the epoch only once both
+                // blob recovery and all shard syncs are complete. An epoch change can advance
+                // the recovery target and start new shard syncs at any time without restarting
+                // this task, so the checks below run under the status mutex, which serializes
+                // completion against the epoch-change path: either this task observes the new
+                // shard syncs (and runs another scan pass), or it completes entirely before the
+                // target is advanced (and the epoch-change path starts a new task).
+                let status_guard = status_mutex.lock().await;
+
+                // A shard sync that started since this pass began invalidates the pass: the
+                // sync may have terminally failed, leaving missing blobs behind that only a new
+                // scan pass finds. The epoch-change path starts shard syncs while holding the
+                // status mutex, so no sync can start between this check and the status update
+                // below.
+                if shard_sync_handler.has_sync_in_progress()
+                    || shard_sync_handler.sync_generation() != sync_generation_before_scan
+                {
+                    tracing::info!(
+                        "shard syncs started during the recovery scan pass; running another pass"
+                    );
+                    drop(status_guard);
+                    continue;
                 }
-            } else {
-                tracing::warn!(
-                    node_status = %current_node_status,
-                    "node recovery task finished; but node status is not RecoveryInProgress; \
-                    skip setting node status to active"
+
+                let current_node_status = node
+                    .storage
+                    .node_status()
+                    .expect("reading node status should not fail");
+                let NodeStatus::RecoveryInProgress(target_epoch) = current_node_status else {
+                    tracing::warn!(
+                        node_status = %current_node_status,
+                        "node recovery task finished; but node status is not RecoveryInProgress; \
+                        skip setting node status to active"
+                    );
+                    return;
+                };
+
+                tracing::info!(
+                    walrus.epoch = target_epoch,
+                    "node recovery task finished; set node status to active"
                 );
+                if let Err(error) = node.set_node_status(NodeStatus::Active) {
+                    tracing::error!(?error, "failed to set node status to active");
+                    return;
+                }
+                drop(status_guard);
+
+                // While the node is recovering, this is the only place that attests epoch sync
+                // done: shard sync skips its own attestation in RecoveryInProgress state (see
+                // the epoch_sync_done handling in shard_sync.rs), so that the attestation covers
+                // both the recovered blobs and all synced shards. The attested epoch is the
+                // recovery target currently recorded in the node status, which may be newer than
+                // the epoch this task was started with.
+                node.contract_service
+                    .epoch_sync_done(target_epoch, node.node_capability())
+                    .await;
+                return;
             }
         });
         *locked_task_handle = Some(task_handle);
diff --git a/crates/walrus-service/src/node/shard_sync.rs b/crates/walrus-service/src/node/shard_sync.rs
index 78935ebe5c..8345f7ef09 100644
--- a/crates/walrus-service/src/node/shard_sync.rs
+++ b/crates/walrus-service/src/node/shard_sync.rs
@@ -3,7 +3,10 @@
 
 use std::{
     collections::{HashMap, hash_map::Entry},
-    sync::Arc,
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
     time::Duration,
 };
 
@@ -49,13 +52,16 @@ enum SyncShardResult {
 /// RAII guard tracking a running shard-sync-related task in the sync task count watch channel.
 ///
 /// Increments the count on creation and decrements it on drop, so that the count stays accurate
-/// even if the tracked task is aborted or panics.
+/// even if the tracked task is aborted or panics. Creation also bumps the sync generation, which
+/// only ever increases and lets node recovery detect that a shard sync started during a scan
+/// pass, even if it also finished during that pass.
 #[derive(Debug)]
 struct SyncTaskCountGuard(Arc<watch::Sender<usize>>);
 
 impl SyncTaskCountGuard {
-    fn new(counter: Arc<watch::Sender<usize>>) -> Self {
+    fn new(counter: Arc<watch::Sender<usize>>, generation: &AtomicU64) -> Self {
         counter.send_modify(|count| *count += 1);
+        generation.fetch_add(1, Ordering::SeqCst);
         sui_macros::fail_point!("fail_point_shard_sync_task_started");
         Self(counter)
     }
@@ -79,6 +85,10 @@ pub struct ShardSyncHandler {
     // the individual per-shard syncs. Used by node recovery to wait until all shard syncs have
     // finished (successfully or not) before recovering blobs.
     sync_task_count: Arc<watch::Sender<usize>>,
+    // Total number of sync tasks ever started; only ever increases. Used by node recovery to
+    // detect whether any shard sync started during a scan pass (a terminally failed sync leaves
+    // missing blobs behind that only a new scan pass finds).
+    sync_task_generation: Arc<AtomicU64>,
     config: ShardSyncConfig,
 }
 
@@ -90,6 +100,7 @@ impl ShardSyncHandler {
             task_handle: Arc::new(Mutex::new(None)),
             shard_sync_semaphore: Arc::new(Semaphore::new(config.shard_sync_concurrency)),
             sync_task_count: Arc::new(watch::channel(0).0),
+            sync_task_generation: Arc::new(AtomicU64::new(0)),
             config,
         }
     }
@@ -99,6 +110,14 @@ impl ShardSyncHandler {
         *self.sync_task_count.borrow() > 0
     }
 
+    /// Returns a generation number that increases every time a shard sync task starts.
+    ///
+    /// Comparing generations around a section of code detects whether any shard sync started
+    /// while it ran, including syncs that also finished within the section.
+    pub fn sync_generation(&self) -> u64 {
+        self.sync_task_generation.load(Ordering::SeqCst)
+    }
+
     /// Waits until no shard sync task is running.
     ///
     /// A shard sync that failed terminally (requiring a node restart to be retried) does not
@@ -133,7 +152,8 @@ impl ShardSyncHandler {
 
         // Count the task that starts the individual shard syncs as a running sync, so that
         // waiters cannot observe a zero count before the per-shard sync tasks are spawned.
-        let count_guard = SyncTaskCountGuard::new(self.sync_task_count.clone());
+        let count_guard =
+            SyncTaskCountGuard::new(self.sync_task_count.clone(), &self.sync_task_generation);
         let new_task = tokio::spawn(async move {
             let _count_guard = count_guard;
             sync_handler.sync_shards_task(shards).await
@@ -416,7 +436,8 @@ impl ShardSyncHandler {
         };
 
         let shard_sync_handler_clone = self.clone();
-        let count_guard = SyncTaskCountGuard::new(self.sync_task_count.clone());
+        let count_guard =
+            SyncTaskCountGuard::new(self.sync_task_count.clone(), &self.sync_task_generation);
         let shard_sync_task = tokio::spawn(async move {
             let _count_guard = count_guard;
             let shard_index = shard_storage.id();
diff --git a/crates/walrus-simtest/tests/simtest_core.rs b/crates/walrus-simtest/tests/simtest_core.rs
index b43ccf4d49..0b1c64b6c5 100644
--- a/crates/walrus-simtest/tests/simtest_core.rs
+++ b/crates/walrus-simtest/tests/simtest_core.rs
@@ -1083,8 +1083,9 @@ mod tests {
     }
 
     // Tests that an epoch change occurring while node recovery is in progress fills newly
-    // gained shards using shard sync, and that node recovery does not start any blob syncs
-    // while shard syncs are running.
+    // gained shards using shard sync, that node recovery does not start any blob syncs while
+    // shard syncs are running, and that the recovery task keeps running across the epoch
+    // changes instead of being restarted.
     //
     // The test crashes a node long enough for it to enter RecoveryInProgress state, holds the
     // recovery task using a fail point, stakes additional weight on the node so that it gains
@@ -1184,15 +1185,21 @@ mod tests {
 
         // Holds the recovery task of the target node so that the recovery reliably spans
         // multiple epoch changes; released once the node has gained shards while recovering.
+        // Also counts how many recovery tasks are spawned on the target node: epoch changes
+        // processed while recovering must not restart the recovery task.
         let hold_recovery = Arc::new(AtomicBool::new(true));
+        let recovery_task_spawn_count = Arc::new(AtomicU64::new(0));
         {
             let hold_recovery = hold_recovery.clone();
+            let recovery_task_spawn_count = recovery_task_spawn_count.clone();
             register_fail_point_async("start_node_recovery_entry", move || {
                 let hold_recovery = hold_recovery.clone();
+                let recovery_task_spawn_count = recovery_task_spawn_count.clone();
                 async move {
                     if sui_simulator::current_simnode_id() != target_node_id {
                         return;
                     }
+                    recovery_task_spawn_count.fetch_add(1, Ordering::SeqCst);
                     tracing::info!("holding node recovery until released by the test");
                     while hold_recovery.load(Ordering::SeqCst) {
                         tokio::time::sleep(Duration::from_secs(1)).await;
@@ -1274,6 +1281,11 @@ mod tests {
             !ordering_violation.load(Ordering::SeqCst),
             "node recovery must not start blob syncs while shard syncs are running"
         );
+        assert_eq!(
+            recovery_task_spawn_count.load(Ordering::SeqCst),
+            1,
+            "the recovery task must not be restarted by epoch changes processed while recovering"
+        );
 
         // The gained shards should be owned by the node and be ready to serve traffic.
         let node_health_info =

From a17f8e4432bafb6978c71ca0cf9a90caa2b10e78 Mon Sep 17 00:00:00 2001
From: Zhe Wu <zhe.wu@mystenlabs.com>
Date: Mon, 13 Jul 2026 11:28:38 -0700
Subject: [PATCH 2/4] fix: prevent stale recovery task from completing a
 catch-up recovery target

Addresses review feedback on the persistent recovery task:

- Serialize the catch-up restart path with recovery completion: a
  recovery task from before the node started catching up only scanned
  blobs certified before its own start epoch, and blob certified events
  were skipped while catching up, so it must not complete the recovery
  target written by the catch-up path. The catch-up path now holds the
  recovery status mutex across writing its recovery target and aborting
  the previous task; a stale task either observes the RecoveryCatchUp
  status and exits without attesting, or is aborted before it can
  complete.
- Use a dedicated scan_pass_interrupted flag instead of overloading
  has_more_blobs when a shard sync interrupts a scan pass, so that
  removing the verification re-scan (WAL-669) leaves the interrupted-
  pass signal intact.
---
 crates/walrus-service/src/node.rs             | 11 ++++++
 .../walrus-service/src/node/node_recovery.rs  | 36 ++++++++++++++-----
 2 files changed, 38 insertions(+), 9 deletions(-)

diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs
index 152c54963f..79f2bb2edf 100644
--- a/crates/walrus-service/src/node.rs
+++ b/crates/walrus-service/src/node.rs
@@ -2332,6 +2332,15 @@ impl StorageNode {
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
     ) -> anyhow::Result<()> {
+        // Serialize this restart against the completion of a possibly still-running recovery
+        // task from before the node started catching up. Such a task only scanned blobs
+        // certified before its own start epoch, and blob certified events were skipped while
+        // catching up, so it must not complete the recovery target written below. Holding the
+        // status mutex guarantees that: if the old task's completion runs first, it observes the
+        // RecoveryCatchUp status and exits without attesting; otherwise, it is aborted by
+        // `start_node_recovery` below before it can complete.
+        let status_guard = self.node_recovery_handler.lock_status().await;
+
         self.inner
             .set_node_status(NodeStatus::RecoveryInProgress(event.epoch))?;
 
@@ -2383,6 +2392,8 @@ impl StorageNode {
             .start_node_recovery(event.epoch)
             .await?;
 
+        drop(status_guard);
+
         // Last but not least, we need to remove any shards that are no longer owned by the node.
         let shards_to_remove = shard_diff_calculator.shards_to_remove();
         if !shards_to_remove.is_empty() {
diff --git a/crates/walrus-service/src/node/node_recovery.rs b/crates/walrus-service/src/node/node_recovery.rs
index 05a2fc8437..a6d86346c4 100644
--- a/crates/walrus-service/src/node/node_recovery.rs
+++ b/crates/walrus-service/src/node/node_recovery.rs
@@ -38,11 +38,14 @@ pub struct NodeRecoveryHandler {
     task_handle: Arc<Mutex<Option<tokio::task::JoinHandle<()>>>>,
 
     // Serializes recovery-related node status transitions: the epoch-change path advances the
-    // recovery target and starts shard syncs while holding this mutex, and the recovery task
-    // transitions the node to `Active` while holding it. This guarantees that a completing
-    // recovery task either observes the advanced target together with the new shard syncs, or
-    // has completed entirely before the transition (in which case the epoch-change path starts
-    // a new task).
+    // recovery target and starts shard syncs while holding this mutex, the catch-up path holds
+    // it across writing a new recovery target and aborting the previous recovery task, and the
+    // recovery task transitions the node to `Active` while holding it. This guarantees that a
+    // completing recovery task either observes the advanced target together with the new shard
+    // syncs, or has completed entirely before the transition (in which case the epoch-change
+    // path starts a new task) — and that a task from before a catch-up can never complete a
+    // recovery target written after the catch-up, which its scan does not cover (blob certified
+    // events are skipped while catching up).
     status_mutex: Arc<Mutex<()>>,
 
     // Configuration for node recovery.
@@ -71,7 +74,9 @@ impl NodeRecoveryHandler {
     /// The epoch-change path must hold the returned guard across advancing the recovery target
     /// (setting the node status to a newer `RecoveryInProgress` epoch) and starting the shard
     /// syncs for newly gained shards, so that these are atomic with respect to the recovery
-    /// task's completion.
+    /// task's completion. The catch-up path must hold it across writing its recovery target and
+    /// aborting the previous recovery task (via [`Self::start_node_recovery`]), so that a task
+    /// from before the catch-up cannot complete a target whose blobs it never scanned.
     pub async fn lock_status(&self) -> tokio::sync::MutexGuard<'_, ()> {
         self.status_mutex.lock().await
     }
@@ -178,6 +183,13 @@ impl NodeRecoveryHandler {
 
                 // Keep track of whether there are more blobs to recover.
                 let mut has_more_blobs = false;
+                // Whether this scan pass was interrupted because a shard sync started. An
+                // interrupted pass has not verified all blobs, so a new pass is required even if
+                // no blob needing recovery was found. This is deliberately kept separate from
+                // `has_more_blobs`: once blob sync outcomes are tracked directly and the
+                // verification re-scan is removed (WAL-669), an interrupted pass still requires
+                // a new round of blob recovery.
+                let mut scan_pass_interrupted = false;
                 tracing::info!(
                     "scanning blobs to recover certified blobs before epoch {}",
                     certified_before_epoch
@@ -196,13 +208,13 @@ impl NodeRecoveryHandler {
                     // An epoch change may have started shard syncs for newly gained shards while
                     // this pass is running. Stop starting new blob syncs and park: per-blob
                     // recovery would redundantly decode slivers for the shards that shard sync is
-                    // copying in bulk. Marking the pass as having more blobs makes the loop drain
-                    // the in-flight syncs and re-scan after the park.
+                    // copying in bulk. The loop drains the in-flight syncs and starts a new scan
+                    // pass after the park.
                     if shard_sync_handler.has_sync_in_progress() {
                         tracing::info!(
                             "shard sync started during recovery scan pass; pausing blob recovery"
                         );
-                        has_more_blobs = true;
+                        scan_pass_interrupted = true;
                         break;
                     }
 
@@ -319,6 +331,12 @@ impl NodeRecoveryHandler {
                     // Each sync completion automatically releases its permit
                 }
 
+                // An interrupted pass has not verified all blobs; run a new pass (which parks
+                // until the shard syncs that interrupted it have finished).
+                if scan_pass_interrupted {
+                    continue;
+                }
+
                 if has_more_blobs {
                     // TODO(WAL-669): right now, we have to do one more loop to check if all the
                     // blobs are recovered. This is not efficient because checking blob existence

From 0af33cc3703631726d1cc6156e65ee5a5f50c9b1 Mon Sep 17 00:00:00 2001
From: Zhe Wu <zhe.wu@mystenlabs.com>
Date: Fri, 17 Jul 2026 10:07:54 -0700
Subject: [PATCH 3/4] fix: lock moved-away shards before recovery completion
 can attest epoch sync done

Hold the recovery status mutex through the shards_to_lock loop in
process_shard_changes_in_new_epoch_while_recovering. Previously the guard
was dropped after starting shard syncs but before locking lost shards, so
on an epoch change with lost shards only, the running recovery task could
complete and attest epoch sync done while the node still accepted slivers
for shards it no longer owns. This matches the catch-up path, which
already locks shards under the guard.
---
 crates/walrus-service/src/node.rs               | 17 ++++++++++-------
 crates/walrus-service/src/node/node_recovery.rs | 12 +++++++-----
 2 files changed, 17 insertions(+), 12 deletions(-)

diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs
index 79f2bb2edf..2f33356d28 100644
--- a/crates/walrus-service/src/node.rs
+++ b/crates/walrus-service/src/node.rs
@@ -2429,11 +2429,14 @@ impl StorageNode {
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
     ) -> anyhow::Result<()> {
-        // Advancing the recovery target and starting the shard syncs for gained shards must be
-        // atomic with respect to the recovery task's completion, which checks both under the
-        // same mutex: a completing task either observes the advanced target together with the
-        // new shard syncs, or completes entirely before this transition (detected below via the
-        // node status, in which case a new task is started).
+        // Advancing the recovery target, starting the shard syncs for gained shards, and locking
+        // the shards that moved away must be atomic with respect to the recovery task's
+        // completion, which runs under the same mutex: a completing task either observes the
+        // advanced target together with the new shard syncs and the locked shards, or completes
+        // entirely before this transition (detected below via the node status, in which case a
+        // new task is started). In particular, completion must not attest epoch sync done before
+        // the lost shards are locked, as the node would still accept slivers for shards it no
+        // longer owns.
         let status_guard = self.node_recovery_handler.lock_status().await;
 
         // If the running recovery task completed concurrently (after this event handler decided
@@ -2466,8 +2469,6 @@ impl StorageNode {
         self.create_new_shards_and_start_sync(shard_map_lock, shards_gained, &committees, false)
             .await?;
 
-        drop(status_guard);
-
         // For shards that just moved out, we need to lock them to not store more data in them.
         for shard_id in shard_diff_calculator.shards_to_lock() {
             let Some(shard_storage) = self.inner.storage.shard_storage(*shard_id).await else {
@@ -2485,6 +2486,8 @@ impl StorageNode {
                 .context("failed to lock shard")?;
         }
 
+        drop(status_guard);
+
         // The recovery task keeps running across epoch changes: it waits for the shard syncs
         // started above to finish before recovering blobs, and attests epoch sync done for the
         // advanced target on completion. A new task is only started when none is running.
diff --git a/crates/walrus-service/src/node/node_recovery.rs b/crates/walrus-service/src/node/node_recovery.rs
index a6d86346c4..c99dcd2da4 100644
--- a/crates/walrus-service/src/node/node_recovery.rs
+++ b/crates/walrus-service/src/node/node_recovery.rs
@@ -72,11 +72,13 @@ impl NodeRecoveryHandler {
     /// Locks the recovery status mutex.
     ///
     /// The epoch-change path must hold the returned guard across advancing the recovery target
-    /// (setting the node status to a newer `RecoveryInProgress` epoch) and starting the shard
-    /// syncs for newly gained shards, so that these are atomic with respect to the recovery
-    /// task's completion. The catch-up path must hold it across writing its recovery target and
-    /// aborting the previous recovery task (via [`Self::start_node_recovery`]), so that a task
-    /// from before the catch-up cannot complete a target whose blobs it never scanned.
+    /// (setting the node status to a newer `RecoveryInProgress` epoch), starting the shard
+    /// syncs for newly gained shards, and locking the shards that moved away, so that these are
+    /// atomic with respect to the recovery task's completion; otherwise, the task could attest
+    /// epoch sync done while the node still accepts slivers for shards it no longer owns. The
+    /// catch-up path must hold it across writing its recovery target and aborting the previous
+    /// recovery task (via [`Self::start_node_recovery`]), so that a task from before the
+    /// catch-up cannot complete a target whose blobs it never scanned.
     pub async fn lock_status(&self) -> tokio::sync::MutexGuard<'_, ()> {
         self.status_mutex.lock().await
     }

From 8878d687f54f8842b3b15cb9a2f2348c0e108fcb Mon Sep 17 00:00:00 2001
From: Zhe Wu <zhe.wu@mystenlabs.com>
Date: Sun, 19 Jul 2026 22:49:24 -0700
Subject: [PATCH 4/4] fix: gate recovery completion on all owned shards being
 active

Node recovery and shard sync have separate responsibilities: the recovery
task recovers blobs certified before its scan bound, while shard sync
fills gained shards with all of their blobs. The epoch_sync_done
attestation sent on recovery completion is however a joint claim, and the
completion check previously measured shard sync activity (no sync task
running, none started during the scan pass) rather than its outcome. A
terminally failed shard sync satisfied that check, so recovery could set
the node active and attest while a gained shard was still missing the
blobs certified between the scan bound and the epoch the shard was
gained, which neither the frozen-bound scan nor event processing covers.

Gate completion directly on the outcome: every owned shard at the latest
committee epoch must be in Active status. If shards are unsynced, wait
for running syncs, or park with a periodic warning when no sync is
running (a terminally failed sync requires a node restart to be retried,
and restart_syncs resumes it from persisted progress). The sync
generation counter is now redundant and is removed.

Reading shard statuses under the status mutex reverses the lock order
against the epoch-change paths, which acquired the status mutex while
holding the shard map write lock. Acquire the status mutex before the
shard map lock in process_epoch_change_start and thread the guard into
the recovery-related paths, establishing the global order status mutex
-> shard map lock.
---
 crates/walrus-service/src/node.rs             |  68 ++++--
 .../walrus-service/src/node/node_recovery.rs  | 204 ++++++++++++------
 crates/walrus-service/src/node/shard_sync.rs  |  35 +--
 3 files changed, 191 insertions(+), 116 deletions(-)

diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs
index 2f33356d28..8ccf2e96c8 100644
--- a/crates/walrus-service/src/node.rs
+++ b/crates/walrus-service/src/node.rs
@@ -1995,13 +1995,19 @@ impl StorageNode {
 
         // During epoch change, we need to lock the read access to shard map until all the new
         // shards are created.
+        //
+        // Lock ordering: the recovery status mutex must be acquired before the shard map lock.
+        // The recovery task's completion attempt reads shard statuses while holding the status
+        // mutex, so acquiring the status mutex after the shard map lock (as the recovery-related
+        // epoch-change paths do) would deadlock with it.
+        let recovery_status_guard = self.node_recovery_handler.lock_status().await;
         let shard_map_lock = self.inner.storage.lock_shards().await;
 
         // Now the general tasks around epoch change are done. Next, entering epoch change logic
         // to bring the node state to the next epoch. `execute_epoch_change` ends by spawning
         // the finisher task (shard removal + `epoch_sync_done` + `mark_as_complete`), so the
         // finisher is guaranteed to fire only after phase 1 succeeded.
-        self.execute_epoch_change(event_handle, event, shard_map_lock)
+        self.execute_epoch_change(event_handle, event, shard_map_lock, recovery_status_guard)
             .await?;
 
         // Update the latest event epoch to the new epoch. Now, blob syncs will use this epoch to
@@ -2138,15 +2144,25 @@ impl StorageNode {
 
     /// Storage node execution of the epoch change start event, to bring the node state to the next
     /// epoch.
+    ///
+    /// `status_guard` is the recovery status mutex guard, acquired by the caller before the shard
+    /// map lock (see the lock ordering comment at the acquisition site); the recovery-related
+    /// paths below hold it across their node status transitions.
     async fn execute_epoch_change(
         &self,
         event_handle: EventHandle,
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
+        status_guard: tokio::sync::MutexGuard<'_, ()>,
     ) -> anyhow::Result<()> {
         if self.inner.storage.node_status()?.is_catching_up() {
-            self.execute_epoch_change_while_catching_up(event_handle, event, shard_map_lock)
-                .await?;
+            self.execute_epoch_change_while_catching_up(
+                event_handle,
+                event,
+                shard_map_lock,
+                status_guard,
+            )
+            .await?;
         } else {
             match self.begin_committee_change(event.epoch).await? {
                 BeginCommitteeChangeAction::ExecuteEpochChange => {
@@ -2154,6 +2170,7 @@ impl StorageNode {
                         event_handle,
                         event,
                         shard_map_lock,
+                        status_guard,
                     )
                     .await?;
                 }
@@ -2171,6 +2188,7 @@ impl StorageNode {
                         event_handle,
                         event,
                         shard_map_lock,
+                        status_guard,
                     )
                     .await?;
                 }
@@ -2189,6 +2207,7 @@ impl StorageNode {
         event_handle: EventHandle,
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
+        status_guard: tokio::sync::MutexGuard<'_, ()>,
     ) -> anyhow::Result<()> {
         self.inner
             .committee_service
@@ -2226,7 +2245,9 @@ impl StorageNode {
         {
             tracing::info!("node just became a new committee member, process shard changes");
             // This node just became a new committee member. Process shard changes as a new
-            // committee member.
+            // committee member; this path performs no recovery-related status transitions, so
+            // the status guard is not needed.
+            drop(status_guard);
             self.process_shard_changes_in_new_epoch_while_node_is_in_sync(
                 event_handle,
                 event,
@@ -2242,6 +2263,7 @@ impl StorageNode {
                 event_handle,
                 event,
                 shard_map_lock,
+                status_guard,
             )
             .await?;
         }
@@ -2256,6 +2278,7 @@ impl StorageNode {
         event_handle: EventHandle,
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
+        status_guard: tokio::sync::MutexGuard<'_, ()>,
     ) -> anyhow::Result<()> {
         // For blobs that are expired in the new epoch, sends a notification to all the tasks
         // that may be affected by the blob expiration.
@@ -2309,9 +2332,13 @@ impl StorageNode {
                 event_handle,
                 event,
                 shard_map_lock,
+                status_guard,
             )
             .await
         } else {
+            // This path performs no recovery-related status transitions, so the status guard is
+            // not needed.
+            drop(status_guard);
             self.process_shard_changes_in_new_epoch_while_node_is_in_sync(
                 event_handle,
                 event,
@@ -2331,16 +2358,16 @@ impl StorageNode {
         event_handle: EventHandle,
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
+        status_guard: tokio::sync::MutexGuard<'_, ()>,
     ) -> anyhow::Result<()> {
-        // Serialize this restart against the completion of a possibly still-running recovery
-        // task from before the node started catching up. Such a task only scanned blobs
-        // certified before its own start epoch, and blob certified events were skipped while
-        // catching up, so it must not complete the recovery target written below. Holding the
-        // status mutex guarantees that: if the old task's completion runs first, it observes the
-        // RecoveryCatchUp status and exits without attesting; otherwise, it is aborted by
-        // `start_node_recovery` below before it can complete.
-        let status_guard = self.node_recovery_handler.lock_status().await;
-
+        // The status guard (acquired by the caller, before the shard map lock) serializes this
+        // restart against the completion of a possibly still-running recovery task from before
+        // the node started catching up. Such a task only scanned blobs certified before its own
+        // start epoch, and blob certified events were skipped while catching up, so it must not
+        // complete the recovery target written below. Holding the status mutex guarantees that:
+        // if the old task's completion runs first, it observes the RecoveryCatchUp status and
+        // exits without attesting; otherwise, it is aborted by `start_node_recovery` below
+        // before it can complete.
         self.inner
             .set_node_status(NodeStatus::RecoveryInProgress(event.epoch))?;
 
@@ -2428,16 +2455,17 @@ impl StorageNode {
         event_handle: EventHandle,
         event: &EpochChangeStart,
         shard_map_lock: StorageShardLock,
+        status_guard: tokio::sync::MutexGuard<'_, ()>,
     ) -> anyhow::Result<()> {
         // Advancing the recovery target, starting the shard syncs for gained shards, and locking
         // the shards that moved away must be atomic with respect to the recovery task's
-        // completion, which runs under the same mutex: a completing task either observes the
-        // advanced target together with the new shard syncs and the locked shards, or completes
-        // entirely before this transition (detected below via the node status, in which case a
-        // new task is started). In particular, completion must not attest epoch sync done before
-        // the lost shards are locked, as the node would still accept slivers for shards it no
-        // longer owns.
-        let status_guard = self.node_recovery_handler.lock_status().await;
+        // completion, which runs under the same mutex (the guard is acquired by the caller,
+        // before the shard map lock): a completing task either observes the advanced target
+        // together with the new shard syncs and the locked shards, or completes entirely before
+        // this transition (detected below via the node status, in which case a new task is
+        // started). In particular, completion must not attest epoch sync done before the lost
+        // shards are locked, as the node would still accept slivers for shards it no longer
+        // owns.
 
         // If the running recovery task completed concurrently (after this event handler decided
         // to take the recovering path), it has flipped the node status away from
diff --git a/crates/walrus-service/src/node/node_recovery.rs b/crates/walrus-service/src/node/node_recovery.rs
index c99dcd2da4..1e61906cd0 100644
--- a/crates/walrus-service/src/node/node_recovery.rs
+++ b/crates/walrus-service/src/node/node_recovery.rs
@@ -7,7 +7,7 @@ use futures::stream::{FuturesUnordered, StreamExt};
 use sui_macros::fail_point_async;
 use tokio::sync::Mutex;
 use typed_store::TypedStoreError;
-use walrus_core::Epoch;
+use walrus_core::{Epoch, ShardIndex};
 use walrus_utils::backoff::{BackoffStrategy, ExponentialBackoff};
 
 use super::{
@@ -16,7 +16,10 @@ use super::{
     config::NodeRecoveryConfig,
     shard_sync::ShardSyncHandler,
 };
-use crate::node::{NodeStatus, storage::blob_info::CertifiedBlobInfoApi};
+use crate::node::{
+    NodeStatus,
+    storage::{ShardStatus, blob_info::CertifiedBlobInfoApi},
+};
 
 /// Exponential backoff bounds for re-checking shard creation while node recovery waits for event
 /// processing to create a shard it owns at the latest epoch. The wait starts small so the common
@@ -25,6 +28,11 @@ use crate::node::{NodeStatus, storage::blob_info::CertifiedBlobInfoApi};
 const SHARD_NOT_CREATED_BACKOFF_MIN: Duration = Duration::from_secs(1);
 const SHARD_NOT_CREATED_BACKOFF_MAX: Duration = Duration::from_secs(60);
 
+/// Interval at which a recovery task that has finished blob recovery re-checks the status of
+/// owned shards that are not `Active` while no shard sync is running (a terminally failed shard
+/// sync requires a node restart to be retried).
+const UNSYNCED_SHARD_RECHECK_INTERVAL: Duration = Duration::from_secs(60);
+
 #[derive(Debug, Clone)]
 pub struct NodeRecoveryHandler {
     node: Arc<StorageNodeInner>,
@@ -79,6 +87,11 @@ impl NodeRecoveryHandler {
     /// catch-up path must hold it across writing its recovery target and aborting the previous
     /// recovery task (via [`Self::start_node_recovery`]), so that a task from before the
     /// catch-up cannot complete a target whose blobs it never scanned.
+    ///
+    /// Lock ordering: this mutex must be acquired *before* the storage shard map lock. The
+    /// recovery task's completion attempt reads shard statuses (which takes the shard map read
+    /// lock) while holding this mutex, so acquiring this mutex while holding the shard map lock
+    /// deadlocks with it.
     pub async fn lock_status(&self) -> tokio::sync::MutexGuard<'_, ()> {
         self.status_mutex.lock().await
     }
@@ -121,9 +134,10 @@ impl NodeRecoveryHandler {
     ///
     /// The task keeps running across epoch changes: blobs certified after the scan bound are
     /// covered by event processing, and shards gained at later epoch changes are covered by shard
-    /// sync (which the task waits for), so the scan bound stays valid. On completion, the task
-    /// attests epoch sync done for the recovery target currently recorded in the node status,
-    /// which the epoch-change path advances.
+    /// sync (which the task waits for), so the scan bound stays valid. The task completes only
+    /// once blob recovery is done and every owned shard is in `Active` status, that is, shard
+    /// sync has delivered every gained shard; it then attests epoch sync done for the recovery
+    /// target currently recorded in the node status, which the epoch-change path advances.
     ///
     /// Any existing recovery task will be canceled.
     // TODO(WAL-864): Refactor this function to make it readable.
@@ -164,14 +178,6 @@ impl NodeRecoveryHandler {
             fail_point_async!("start_node_recovery_entry");
 
             loop {
-                // Snapshot the shard sync generation before waiting: a completed pass may only
-                // attest if no shard sync started since this point. A shard sync that starts
-                // (and possibly terminally fails) during the pass leaves blobs behind that only
-                // a new scan pass finds, so a generation change forces a re-scan. Taking the
-                // snapshot before the wait errs towards a spurious re-scan when syncs start and
-                // drain while parked, never towards a missed one.
-                let sync_generation_before_scan = shard_sync_handler.sync_generation();
-
                 // Block until the node is ready to run a recovery scan pass. Stops the recovery
                 // task if the node started catching up.
                 match wait_until_ready_to_scan(&node, &shard_sync_handler).await {
@@ -348,61 +354,9 @@ impl NodeRecoveryHandler {
                     continue;
                 }
 
-                // Completion attempt. The node is fully synced for the epoch only once both
-                // blob recovery and all shard syncs are complete. An epoch change can advance
-                // the recovery target and start new shard syncs at any time without restarting
-                // this task, so the checks below run under the status mutex, which serializes
-                // completion against the epoch-change path: either this task observes the new
-                // shard syncs (and runs another scan pass), or it completes entirely before the
-                // target is advanced (and the epoch-change path starts a new task).
-                let status_guard = status_mutex.lock().await;
-
-                // A shard sync that started since this pass began invalidates the pass: the
-                // sync may have terminally failed, leaving missing blobs behind that only a new
-                // scan pass finds. The epoch-change path starts shard syncs while holding the
-                // status mutex, so no sync can start between this check and the status update
-                // below.
-                if shard_sync_handler.has_sync_in_progress()
-                    || shard_sync_handler.sync_generation() != sync_generation_before_scan
-                {
-                    tracing::info!(
-                        "shard syncs started during the recovery scan pass; running another pass"
-                    );
-                    drop(status_guard);
-                    continue;
-                }
-
-                let current_node_status = node
-                    .storage
-                    .node_status()
-                    .expect("reading node status should not fail");
-                let NodeStatus::RecoveryInProgress(target_epoch) = current_node_status else {
-                    tracing::warn!(
-                        node_status = %current_node_status,
-                        "node recovery task finished; but node status is not RecoveryInProgress; \
-                        skip setting node status to active"
-                    );
-                    return;
-                };
-
-                tracing::info!(
-                    walrus.epoch = target_epoch,
-                    "node recovery task finished; set node status to active"
-                );
-                if let Err(error) = node.set_node_status(NodeStatus::Active) {
-                    tracing::error!(?error, "failed to set node status to active");
-                    return;
-                }
-                drop(status_guard);
-
-                // While the node is recovering, this is the only place that attests epoch sync
-                // done: shard sync skips its own attestation in RecoveryInProgress state (see
-                // the epoch_sync_done handling in shard_sync.rs), so that the attestation covers
-                // both the recovered blobs and all synced shards. The attested epoch is the
-                // recovery target currently recorded in the node status, which may be newer than
-                // the epoch this task was started with.
-                node.contract_service
-                    .epoch_sync_done(target_epoch, node.node_capability())
+                // Blob recovery (this task's scan) is done; wait for shard sync to deliver
+                // every owned shard and then complete the recovery.
+                complete_recovery_once_shards_synced(&node, &shard_sync_handler, &status_mutex)
                     .await;
                 return;
             }
@@ -522,3 +476,117 @@ async fn wait_until_ready_to_scan(
 
     readiness
 }
+
+/// Completes node recovery once shard sync has delivered every owned shard, then attests epoch
+/// sync done. Called after the recovery task's blob scan has finished cleanly; returns when the
+/// recovery task should exit.
+///
+/// Completion waits for shard sync even though recovering blobs and syncing shards are separate
+/// responsibilities (the recovery task only recovers blobs certified before its scan bound,
+/// while shard sync fills a gained shard with all of its blobs), because the `epoch_sync_done`
+/// attestation sent on completion is a joint claim: it states that the node holds *all* the data
+/// it should have for the epoch, gained shards included. While the node is recovering, the
+/// recovery task is the sole attester (shard sync suppresses its own attestation, see the
+/// epoch_sync_done handling in shard_sync.rs), so it must not attest before shard sync's half of
+/// the claim is true — measured directly by every owned shard being in `Active` status. The
+/// separation still holds for the work itself: a shard that reached `Active` needs no re-scan,
+/// and a shard that has not is shard sync's job to finish — never the recovery task's.
+///
+/// The checks run under the status mutex, which serializes completion against the epoch-change
+/// path: either this task observes the advanced target together with the not-yet-`Active` gained
+/// shards, or it completes entirely before the target is advanced (and the epoch-change path
+/// starts a new task).
+async fn complete_recovery_once_shards_synced(
+    node: &StorageNodeInner,
+    shard_sync_handler: &ShardSyncHandler,
+    status_mutex: &Mutex<()>,
+) {
+    loop {
+        let status_guard = status_mutex.lock().await;
+
+        let unsynced_shards = unsynced_owned_shards(node).await;
+        if unsynced_shards.is_empty() {
+            let current_node_status = node
+                .storage
+                .node_status()
+                .expect("reading node status should not fail");
+            let NodeStatus::RecoveryInProgress(target_epoch) = current_node_status else {
+                tracing::warn!(
+                    node_status = %current_node_status,
+                    "node recovery task finished; but node status is not RecoveryInProgress; \
+                    skip setting node status to active"
+                );
+                return;
+            };
+
+            tracing::info!(
+                walrus.epoch = target_epoch,
+                "node recovery task finished; set node status to active"
+            );
+            if let Err(error) = node.set_node_status(NodeStatus::Active) {
+                tracing::error!(?error, "failed to set node status to active");
+                return;
+            }
+            drop(status_guard);
+
+            // While the node is recovering, this is the only place that attests epoch sync
+            // done: shard sync skips its own attestation in RecoveryInProgress state (see the
+            // epoch_sync_done handling in shard_sync.rs), so that the attestation covers both
+            // the recovered blobs and all synced shards. The attested epoch is the recovery
+            // target currently recorded in the node status, which may be newer than the epoch
+            // this task was started with.
+            node.contract_service
+                .epoch_sync_done(target_epoch, node.node_capability())
+                .await;
+            return;
+        }
+        drop(status_guard);
+
+        if shard_sync_handler.has_sync_in_progress() {
+            tracing::info!(
+                ?unsynced_shards,
+                "waiting for ongoing shard syncs before completing node recovery"
+            );
+            shard_sync_handler.wait_until_no_sync_in_progress().await;
+        } else {
+            // The syncs for these shards stopped without reaching `Active` status (a terminally
+            // failed shard sync requires a node restart to be retried). Completing now would
+            // attest epoch sync done while the shard data is still missing, so park and
+            // re-check instead.
+            tracing::warn!(
+                ?unsynced_shards,
+                "owned shards are not active and no shard sync is running; node recovery \
+                cannot complete; restart the node to retry shard sync"
+            );
+            tokio::time::sleep(UNSYNCED_SHARD_RECHECK_INTERVAL).await;
+        }
+    }
+}
+
+/// Returns the shards owned at the latest committee epoch whose local storage is missing or whose
+/// status is not [`ShardStatus::Active`], meaning shard sync has not (yet) completed for them.
+///
+/// A shard whose status cannot be read is conservatively reported as unsynced.
+async fn unsynced_owned_shards(node: &StorageNodeInner) -> Vec<ShardIndex> {
+    let mut unsynced = Vec::new();
+    for shard in node.owned_shards_at_latest_epoch() {
+        let status = match node.storage.shard_storage(shard).await {
+            Some(shard_storage) => shard_storage
+                .status()
+                .await
+                .inspect_err(|error| {
+                    tracing::warn!(
+                        walrus.shard_index = %shard,
+                        ?error,
+                        "failed to read shard status; treating shard as unsynced"
+                    );
+                })
+                .ok(),
+            None => None,
+        };
+        if !matches!(status, Some(ShardStatus::Active)) {
+            unsynced.push(shard);
+        }
+    }
+    unsynced
+}
diff --git a/crates/walrus-service/src/node/shard_sync.rs b/crates/walrus-service/src/node/shard_sync.rs
index 8345f7ef09..ab1fd48290 100644
--- a/crates/walrus-service/src/node/shard_sync.rs
+++ b/crates/walrus-service/src/node/shard_sync.rs
@@ -3,10 +3,7 @@
 
 use std::{
     collections::{HashMap, hash_map::Entry},
-    sync::{
-        Arc,
-        atomic::{AtomicU64, Ordering},
-    },
+    sync::Arc,
     time::Duration,
 };
 
@@ -52,16 +49,13 @@ enum SyncShardResult {
 /// RAII guard tracking a running shard-sync-related task in the sync task count watch channel.
 ///
 /// Increments the count on creation and decrements it on drop, so that the count stays accurate
-/// even if the tracked task is aborted or panics. Creation also bumps the sync generation, which
-/// only ever increases and lets node recovery detect that a shard sync started during a scan
-/// pass, even if it also finished during that pass.
+/// even if the tracked task is aborted or panics.
 #[derive(Debug)]
 struct SyncTaskCountGuard(Arc<watch::Sender<usize>>);
 
 impl SyncTaskCountGuard {
-    fn new(counter: Arc<watch::Sender<usize>>, generation: &AtomicU64) -> Self {
+    fn new(counter: Arc<watch::Sender<usize>>) -> Self {
         counter.send_modify(|count| *count += 1);
-        generation.fetch_add(1, Ordering::SeqCst);
         sui_macros::fail_point!("fail_point_shard_sync_task_started");
         Self(counter)
     }
@@ -85,10 +79,6 @@ pub struct ShardSyncHandler {
     // the individual per-shard syncs. Used by node recovery to wait until all shard syncs have
     // finished (successfully or not) before recovering blobs.
     sync_task_count: Arc<watch::Sender<usize>>,
-    // Total number of sync tasks ever started; only ever increases. Used by node recovery to
-    // detect whether any shard sync started during a scan pass (a terminally failed sync leaves
-    // missing blobs behind that only a new scan pass finds).
-    sync_task_generation: Arc<AtomicU64>,
     config: ShardSyncConfig,
 }
 
@@ -100,7 +90,6 @@ impl ShardSyncHandler {
             task_handle: Arc::new(Mutex::new(None)),
             shard_sync_semaphore: Arc::new(Semaphore::new(config.shard_sync_concurrency)),
             sync_task_count: Arc::new(watch::channel(0).0),
-            sync_task_generation: Arc::new(AtomicU64::new(0)),
             config,
         }
     }
@@ -110,19 +99,11 @@ impl ShardSyncHandler {
         *self.sync_task_count.borrow() > 0
     }
 
-    /// Returns a generation number that increases every time a shard sync task starts.
-    ///
-    /// Comparing generations around a section of code detects whether any shard sync started
-    /// while it ran, including syncs that also finished within the section.
-    pub fn sync_generation(&self) -> u64 {
-        self.sync_task_generation.load(Ordering::SeqCst)
-    }
-
     /// Waits until no shard sync task is running.
     ///
     /// A shard sync that failed terminally (requiring a node restart to be retried) does not
-    /// count as running; its shard remains in `ActiveSync` status and missing blobs in it are
-    /// recovered through the regular blob recovery path.
+    /// count as running; its shard remains in `ActiveSync` or `ActiveRecover` status, which
+    /// blocks node recovery from completing until the sync is retried.
     pub async fn wait_until_no_sync_in_progress(&self) {
         let mut receiver = self.sync_task_count.subscribe();
         receiver
@@ -152,8 +133,7 @@ impl ShardSyncHandler {
 
         // Count the task that starts the individual shard syncs as a running sync, so that
         // waiters cannot observe a zero count before the per-shard sync tasks are spawned.
-        let count_guard =
-            SyncTaskCountGuard::new(self.sync_task_count.clone(), &self.sync_task_generation);
+        let count_guard = SyncTaskCountGuard::new(self.sync_task_count.clone());
         let new_task = tokio::spawn(async move {
             let _count_guard = count_guard;
             sync_handler.sync_shards_task(shards).await
@@ -436,8 +416,7 @@ impl ShardSyncHandler {
         };
 
         let shard_sync_handler_clone = self.clone();
-        let count_guard =
-            SyncTaskCountGuard::new(self.sync_task_count.clone(), &self.sync_task_generation);
+        let count_guard = SyncTaskCountGuard::new(self.sync_task_count.clone());
         let shard_sync_task = tokio::spawn(async move {
             let _count_guard = count_guard;
             let shard_index = shard_storage.id();