diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs index b982b9d3c8..8ee64c9e14 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. @@ -2295,22 +2318,27 @@ 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( 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, @@ -2330,7 +2358,18 @@ impl StorageNode { event_handle: EventHandle, event: &EpochChangeStart, shard_map_lock: StorageShardLock, + status_guard: tokio::sync::MutexGuard<'_, ()>, ) -> anyhow::Result<()> { + // A recovery task from before the node started catching up may still be running. 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. The status guard (acquired by the caller, before the shard map lock) + // keeps any completion attempt of that task parked on the mutex, and aborting the task + // here — before the new target is written — guarantees that no stale task survives to + // observe it, even if a later step in this function fails and returns before reaching + // `start_node_recovery` (which would otherwise perform the abort). + self.node_recovery_handler.abort_recovery_task().await; + self.inner .set_node_status(NodeStatus::RecoveryInProgress(event.epoch))?; @@ -2382,6 +2421,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() { @@ -2405,9 +2446,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. @@ -2416,10 +2457,28 @@ 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 (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 + // 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))?; @@ -2457,10 +2516,13 @@ 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. + 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. 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..176893f9c7 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, @@ -37,6 +45,17 @@ pub struct NodeRecoveryHandler { // There can be at most one background shard removal task at a time. task_handle: Arc>>>, + // Serializes recovery-related node status transitions: the epoch-change path advances the + // 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>, + // Configuration for node recovery. config: NodeRecoveryConfig, } @@ -53,14 +72,83 @@ 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), 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. + /// + /// 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 + } + + /// Aborts the running recovery task, if any, and waits for it to exit. + /// + /// The catch-up path calls this while holding the status mutex, before writing its new + /// recovery target: the stale task must be gone before the target becomes visible, so that + /// it cannot complete the target even if the caller fails and returns before reaching + /// [`Self::start_node_recovery`] (which would otherwise perform the abort). + pub async fn abort_recovery_task(&self) { + abort_task(self.task_handle.lock().await.take()).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. 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. pub async fn start_node_recovery( @@ -69,17 +157,13 @@ impl NodeRecoveryHandler { ) -> Result<(), TypedStoreError> { let mut locked_task_handle = self.task_handle.lock().await; - // Cancel any existing recovery task - if let Some(old_task) = locked_task_handle.take() { - tracing::info!("canceling existing node recovery task"); - old_task.abort(); - // Wait for the old task to complete (it will return a JoinError due to cancellation) - let _ = old_task.await; - } + // Cancel any existing recovery task. + abort_task(locked_task_handle.take()).await; 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 { @@ -112,6 +196,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 @@ -127,6 +218,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. 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" + ); + scan_pass_interrupted = true; + break; + } + node.metrics .node_recovery_recover_blob_progress .set(i64::from(blob_id.first_two_bytes())); @@ -235,54 +339,31 @@ 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; + // 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; + } - 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"); - } + 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; } - } else { - tracing::warn!( - node_status = %current_node_status, - "node recovery task finished; but node status is not RecoveryInProgress; \ - skip setting node status to active" - ); + + // 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; } }); *locked_task_handle = Some(task_handle); @@ -400,3 +481,127 @@ 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; + } + } +} + +/// Aborts the given recovery task, if any, and waits for it to exit. +async fn abort_task(task: Option>) { + if let Some(old_task) = task { + tracing::info!("canceling existing node recovery task"); + old_task.abort(); + // Wait for the old task to finish (it will return a JoinError due to cancellation). + let _ = old_task.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 { + 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 78935ebe5c..ab1fd48290 100644 --- a/crates/walrus-service/src/node/shard_sync.rs +++ b/crates/walrus-service/src/node/shard_sync.rs @@ -102,8 +102,8 @@ impl ShardSyncHandler { /// 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 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 =