Test hot standby recovery from failures and interruptions

.github/workflows/server-zd-simulation-nightly.yml

@@ -45,6 +45,16 @@ on:
         required: true
         default: 300
         type: number
+      hot_standby_seed_count:
+        description: Number of consecutive engine simulation seeds in the hot standby profile
+        required: true
+        default: 48
+        type: number
+      hot_standby_active_seconds:
+        description: Simulated seconds of activity per hot standby profile seed
+        required: true
+        default: 300
+        type: number
 
 permissions:
   contents: read
@@ -577,3 +587,110 @@ jobs:
             ${{ runner.temp }}/dexzd-step-failure-simulation*.log
           if-no-files-found: warn
           retention-days: 30
+
+  engine-simulation-hot-standby-profile:
+    name: Engine simulation hot standby profile (${{ matrix.target.label }})
+    needs: simulation-refs
+    strategy:
+      fail-fast: false
+      matrix:
+        target: ${{ fromJSON(needs.simulation-refs.outputs.targets) }}
+    runs-on: ubuntu-latest
+    timeout-minutes: 90
+    defaults:
+      run:
+        working-directory: server-zd
+    env:
+      DEXZD_SIMULATION_SEEDS: ${{ inputs.hot_standby_seed_count || 48 }}
+      DEXZD_SIMULATION_SECONDS: ${{ inputs.hot_standby_active_seconds || 300 }}
+    steps:
+      - uses: actions/checkout@v7
+        with:
+          ref: ${{ matrix.target.ref }}
+      - name: Check out the nightly harness
+        uses: actions/checkout@v7
+        with:
+          ref: ${{ github.sha }}
+          path: .nightly-harness
+          sparse-checkout: script/server-zd-simulation-run.py
+          sparse-checkout-cone-mode: false
+      - name: Install Rust
+        run: rustup toolchain install 1.97.1 --profile minimal
+      - name: Install libclang for the RocksDB bindings
+        run: sudo apt-get update && sudo apt-get install -y libclang-dev
+      - name: Validate hot standby profile budget
+        shell: bash
+        run: |
+          if ! [[ "$DEXZD_SIMULATION_SEEDS" =~ ^[1-9][0-9]*$ ]] ||
+             ! [[ "$DEXZD_SIMULATION_SECONDS" =~ ^[1-9][0-9]*$ ]] ||
+             (( ${#DEXZD_SIMULATION_SEEDS} > 4 || ${#DEXZD_SIMULATION_SECONDS} > 4 )) ||
+             (( DEXZD_SIMULATION_SEEDS < 16 || DEXZD_SIMULATION_SEEDS > 1000 )) ||
+             (( DEXZD_SIMULATION_SECONDS < 60 || DEXZD_SIMULATION_SECONDS > 600 )) ||
+             (( DEXZD_SIMULATION_SEEDS * DEXZD_SIMULATION_SECONDS > 20000 )); then
+            echo 'Use 16–1000 seeds, 60–600 simulated seconds per seed, and at most 20000 simulated seconds.' >&2
+            exit 1
+          fi
+      - name: Choose random base seed
+        run: |
+          base_seed=$(od -An -N4 -tu4 /dev/urandom | tr -d '[:space:]')
+          echo "DEXZD_SIMULATION_BASE_SEED=$base_seed" >> "$GITHUB_ENV"
+          echo "Base seed: $base_seed"
+      - name: Check for the hot standby profile
+        id: profile
+        shell: bash
+        run: |
+          if grep -q 'fn hot_standby_takeovers_keep_every_invariant' crates/dexzd-simulation/tests/engine_simulation.rs; then
+            echo "is_present=true" >> "$GITHUB_OUTPUT"
+          else
+            echo "${{ matrix.target.ref }} predates the hot standby profile; skipping it."
+            echo "is_present=false" >> "$GITHUB_OUTPUT"
+          fi
+      - name: Run hot standby profile
+        id: hot_standby
+        if: steps.profile.outputs.is_present == 'true'
+        shell: bash
+        env:
+          DEXZD_SIMULATION_FAILURE_DIR: ${{ runner.temp }}/dexzd-hot-standby-simulation-failures
+        run: |
+          python3 "$GITHUB_WORKSPACE/.nightly-harness/script/server-zd-simulation-run.py" --prepare-progress
+          python3 "$GITHUB_WORKSPACE/.nightly-harness/script/server-zd-simulation-run.py" --build --timeout 1500 --log "$RUNNER_TEMP/dexzd-hot-standby-simulation-build.log" -- \
+            cargo test --locked -p dexzd-simulation --test engine_simulation --no-run
+          python3 "$GITHUB_WORKSPACE/.nightly-harness/script/server-zd-simulation-run.py" --timeout 3300 --log "$RUNNER_TEMP/dexzd-hot-standby-simulation.log" -- \
+            cargo test --locked -p dexzd-simulation --test engine_simulation hot_standby_takeovers_keep_every_invariant -- --exact --nocapture
+      - name: Summarize hot standby profile
+        if: always()
+        shell: bash
+        run: |
+          {
+            echo "## Engine simulation hot standby profile (${{ matrix.target.label }}: ${{ matrix.target.ref }})"
+            echo "Base seed: ${DEXZD_SIMULATION_BASE_SEED:-not selected}"
+            echo "Seed count: $DEXZD_SIMULATION_SEEDS; simulated seconds per seed: $DEXZD_SIMULATION_SECONDS"
+            echo "Build result: ${{ steps.hot_standby.outputs.build_result || 'not completed' }}; elapsed seconds: ${{ steps.hot_standby.outputs.build_elapsed_seconds }}"
+            echo "Result: ${{ steps.hot_standby.outputs.result || steps.hot_standby.outcome }}"
+            echo "Elapsed seconds: ${{ steps.hot_standby.outputs.elapsed_seconds }}; last completed seed: ${{ steps.hot_standby.outputs.last_completed_seed || 'none' }}"
+            echo
+            if compgen -G "$RUNNER_TEMP/dexzd-hot-standby-simulation-failures/*.txt" > /dev/null; then
+              echo "### Failing seeds"
+              echo "Run reproduction commands from \`server-zd/\`."
+              for report in "$RUNNER_TEMP"/dexzd-hot-standby-simulation-failures/*.txt; do
+                seed=${report##*/}
+                seed=${seed%.txt}
+                reproduction=$(grep -m1 '^reproduce with:' "$report")
+                echo "- Seed $seed: \`${reproduction#reproduce with: }\`"
+              done
+              echo
+              echo "Download the failure artifact for each violation and its last trace events."
+            else
+              echo "No invariant failures were recorded."
+            fi
+          } >> "$GITHUB_STEP_SUMMARY"
+      - name: Upload hot standby profile failures
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: dexzd-hot-standby-simulation-failures-${{ matrix.target.label }}
+          path: |
+            ${{ runner.temp }}/dexzd-hot-standby-simulation-failures/
+            ${{ runner.temp }}/dexzd-hot-standby-simulation*.log
+          if-no-files-found: warn
+          retention-days: 30

server-zd/crates/dexzd-engine/README.md

@@ -73,10 +73,11 @@ deduplication.
 - Its committed-command sink does nothing, and it runs no rebuild, timers, dispatch, Attribute Sync, or search projection. A serving start rebuilds those from the local store.
 - It advances idle partitions too. Each position records the epoch in effect after its batch, which recovery accepts after the position.
 - It rereads the manifest periodically. A closed segment bounds the stream it reads, and a new incarnation becomes the next stream. Log cleanup that passes its next seq or its stored positions makes it restore the newest snapshot. While it waits at one seq, it also checks whether cleanup deleted that seq.
-- `HotStandby::status` reports the state and the lag of the follow group furthest behind; a single node follows one group. Lag is zero after a read finds no next batch, so an idle log does not look behind. While the standby waits at a hole that later batches follow, lag is the time it has waited there, so a hole that stays open turns it behind. Otherwise lag is the time since the later of the last applied batch's seal time and the last read that found no next batch.
-- Only a configuration error fails the standby for good: a node identity mismatch, a regressed incarnation, a behavior version this binary does not support, or an invalid configuration. A store directory that its first bootstrap cannot open also fails it, since another process, such as the serving process on the same host, may hold the directory. After any other error, and after log cleanup outran it, it closes its store, reports `Restarting`, waits from 1 second doubling to 60 seconds with jitter, and bootstraps again. The delay resets, and its status drops the last error, only once following reads past where earlier periods stopped, or reaches the end of the log there rather than a hole. While the cluster object or the manifest is missing, it rereads them every second.
+- `HotStandby::status` reports the state and the lag of the follow group furthest behind; a single node follows one group. Lag is zero after a read finds no next batch, so an idle log does not look behind. While the standby waits at a hole that later batches follow, lag is the time it has waited there, so a hole that stays open turns it behind. Otherwise lag is the time since the later of the last applied batch's seal time and the last read that found no next batch. When three or more log reads in a row fail, `read_error` shows why, so an unreachable object storage is not mistaken for an idle node.
+- Only a configuration error fails the standby for good: a node identity mismatch, a regressed incarnation, a behavior version this binary does not support, or an invalid configuration. Before it opens a RocksDB directory, it probes the directory's lock without opening the store, so a held lock never touches the holder's files: a lock that another process holds, such as the serving node's, fails it as a configuration error, and one that a lingering handle of this process holds is retried, with a `last_error` that says so. After any other error, and after log cleanup outran it, it closes its store, reports `Restarting`, waits from 1 second doubling to 60 seconds with jitter, and bootstraps again. The delay resets, and its status drops the last error, only once following reads past where earlier periods stopped, or reaches the end of the log there rather than a hole. While the cluster object or the manifest is missing, it rereads them every second.
 - The gap check of every start and bootstrap also rejects a store in which a partition applied a batch past the recorded end of a closed stream. Such a store followed a batch that the log does not hold, so it is set aside and the newest snapshot is restored.
 - A `LogFollower` takes its write fence and applier pool from its owner. It advances every partition of the pool, so the pool must serve only the followed partitions.
+- `EngineNode::local_store_close_waiter` and `HotStandby::local_store_close_waiter` return a `LocalStoreCloseWaiter`. After a step down or a standby shutdown, await its `wait_until_closed` before another process reopens the RocksDB directory.
 - Drop the standby or call `HotStandby::shutdown` to stop following. Dropping it stops its store writes at once; `shutdown` also waits for its store to close and returns whether it closed.
 - `HotStandby::status_reader` returns a reader of the status that stays valid while a takeover owns the standby, such as while the takeover waits for the lease. Once a handover, a shutdown, or a drop stops the standby, the status shows `Stopped`.
 
@@ -85,6 +86,8 @@ deduplication.
 1. The standby keeps following while another process holds the lease. Cancellation before the lease returns the standby, still following. An error before the lease, such as a standby that follows another node or cluster, returns `HotStandbyTakeoverError` with the standby.
 2. Once this process holds the lease and has reread the manifest under it, the standby stops reading, applies its queued frames, and advances its applied positions.
 3. It stops its write fence, so no standby lane writes the store again, and its tasks are joined. The start owns this handover: when it fails or is cancelled, it stops the write fence at once and stops the standby beside its own tasks, then releases the lease, all within `EngineNode::TAKEOVER_RELEASE_BUDGET`.
-4. The start checks that the standby used the configured store and that its write fence stopped only for the handover, not after a failed store write. It then checks that every applied position names this node, and reruns the gap check against the manifest read under the lease.
+4. The start fails with `InvalidConfig` when the standby, which may have bootstrapped while the start waited for the lease, follows another cluster. It checks that the standby used the configured store and that its write fence stopped only for the handover, not after a failed store write. It then checks that every applied position names this node, and reruns the gap check against the manifest read under the lease.
 5. When a check fails, or the standby had no store, the start opens or restores the configured store, as a start without a standby does. It first waits up to 30 seconds for a lane that outlived the handover to close a RocksDB store.
 6. The rest of the start is unchanged: a new applier pool with the real hooks, log replay from the applied positions, the EpochStart, and serving.
+
+`EngineNode::is_serving_hot_standby_store` reports whether a takeover served the standby's store or opened the configured store, for example to label takeover metrics.

server-zd/crates/dexzd-engine/src/engine_node.rs

@@ -8,8 +8,11 @@
 
 //! The engine node: startup and recovery, serving, and shutdown.
 
+mod local_store_close_waiter;
 mod node_start;
 
+pub use local_store_close_waiter::LocalStoreCloseWaiter;
+
 use std::future::Future;
 use std::sync::Arc;
 use std::time::Duration;
@@ -108,7 +111,6 @@ pub enum HotStandbyTakeover {
 #[non_exhaustive]
 pub struct HotStandbyTakeoverError {
     /// Why the start failed.
-    #[source]
     pub error: EngineStartError,
     /// The standby, still following, when the start failed before the lease; `None` after it.
     pub standby: Option<HotStandby>,
@@ -136,8 +138,8 @@ pub struct EngineNode {
     tasks: RunningTasks,
     cluster_id: [u8; 16],
     /// Whether the node serves from the store a hot standby handed over.
-    #[cfg(feature = "test-util")]
     is_serving_hot_standby_store: bool,
+    local_store_close_waiter: LocalStoreCloseWaiter,
 }
 
 impl EngineNode {
@@ -213,7 +215,8 @@ impl EngineNode {
     /// [`EngineNode::start_until_cancelled`] documents. It carries the standby, still following,
     /// when the start failed before the lease, such as for a standby of another node. A failure
     /// after the lease stops the standby and releases the lease within
-    /// [`EngineNode::TAKEOVER_RELEASE_BUDGET`].
+    /// [`EngineNode::TAKEOVER_RELEASE_BUDGET`]. Take the standby's [`LocalStoreCloseWaiter`]
+    /// beforehand to reopen its directory afterwards.
     pub async fn take_over_from_hot_standby_until_cancelled<F>(
         config: &EngineNodeConfig,
         dependencies: EngineNodeDependencies,
@@ -264,12 +267,31 @@ impl EngineNode {
         self.cluster_id
     }
 
-    /// Returns whether the node serves from the store that a hot standby handed over.
-    #[cfg(feature = "test-util")]
-    pub fn is_serving_hot_standby_store_for_test(&self) -> bool {
+    /// Returns whether this node serves from the local store that its hot standby handed over.
+    ///
+    /// A takeover from a [`HotStandby`] serves the standby's warm store when the standby followed
+    /// this node into the configured store and the retained log can update it; it then replays
+    /// only the few batches the standby had not applied. Otherwise the takeover opens or restores
+    /// the configured store as an ordinary start does, which replays more of the log. Use it to
+    /// label takeover metrics and logs with how the store was obtained:
+    ///
+    /// ```ignore
+    /// let outcome = if node.is_serving_hot_standby_store() { "warm" } else { "cold" };
+    /// metrics.takeovers_total.with_label_values(&[outcome]).inc();
+    /// ```
+    ///
+    /// Returns `true` only for a node that [`EngineNode::take_over_from_hot_standby_until_cancelled`]
+    /// started from its standby's store, and `false` for every other start.
+    pub fn is_serving_hot_standby_store(&self) -> bool {
         self.is_serving_hot_standby_store
     }
 
+    /// Returns a waiter of the node's local store, which stays valid after a shutdown; see
+    /// [`LocalStoreCloseWaiter`].
+    pub fn local_store_close_waiter(&self) -> LocalStoreCloseWaiter {
+        self.local_store_close_waiter.clone()
+    }
+
     /// Returns the node epoch this process raised when it took the lease.
     pub fn node_epoch(&self) -> u64 {
         self.lease.epoch()

server-zd/crates/dexzd-engine/src/engine_node/local_store_close_waiter.rs

@@ -0,0 +1,83 @@
+// Copyright (c) 2026 Super Durable, Inc.
+//
+// Licensed under the Sustainable Use License 1.0.
+// You may not use this file except in compliance with the License.
+// See the LICENSE file in the repository root.
+//
+// SPDX-License-Identifier: LicenseRef-Sustainable-Use-1.0
+
+//! Waits until no handle keeps a local store open, so another process may reopen its directory.
+
+use std::path::PathBuf;
+use std::sync::{Arc, Mutex, Weak};
+use std::time::Duration;
+
+use dexzd_storage::is_directory_open_in_this_process;
+use tokio::time::Instant;
+
+use crate::commit::EngineLocalStore;
+use crate::recovery::LocalStoreConfig;
+
+/// How often a wait for a closed store checks it again.
+const STORE_CLOSE_POLL: Duration = Duration::from_millis(10);
+
+/// Waits until a node's or a hot standby's local store has closed, so its directory reopens.
+///
+/// Take it before a shutdown, then await [`Self::wait_until_closed`] after it.
+#[derive(Clone)]
+pub struct LocalStoreCloseWaiter {
+    /// The newest store; a hot standby's restore replaces it.
+    store: Arc<Mutex<Option<Weak<EngineLocalStore>>>>,
+    /// A RocksDB store's directory, whose lock outlives the last handle until the store closes.
+    directory: Option<PathBuf>,
+}
+
+impl LocalStoreCloseWaiter {
+    /// Returns a waiter that watches whatever store `store` names when it waits, and the
+    /// directory of the store that `config` names.
+    pub(crate) fn new(
+        store: Arc<Mutex<Option<Weak<EngineLocalStore>>>>,
+        config: &LocalStoreConfig,
+    ) -> Self {
+        let directory = match config {
+            LocalStoreConfig::RocksDb { directory, .. } => Some(directory.clone()),
+            _ => None,
+        };
+        Self { store, directory }
+    }
+
+    /// Returns a waiter of `store`, which `config` names.
+    pub(crate) fn for_store(store: &Arc<EngineLocalStore>, config: &LocalStoreConfig) -> Self {
+        Self::new(Arc::new(Mutex::new(Some(Arc::downgrade(store)))), config)
+    }
+
+    /// Waits until no handle keeps the store open, for at most `budget`.
+    ///
+    /// Returns whether the store closed, and `true` when no store was ever opened. Returns
+    /// `false` when a handle still lives after `budget`; reopening the directory then fails with
+    /// [`crate::EngineStartError::LocalStoreUnavailable`] until it closes.
+    pub async fn wait_until_closed(&self, budget: Duration) -> bool {
+        self.wait_until_closed_by(Instant::now() + budget).await
+    }
+
+    /// Returns whether the store closed by `deadline`; see [`Self::wait_until_closed`].
+    pub(crate) async fn wait_until_closed_by(&self, deadline: Instant) -> bool {
+        let store = self
+            .store
+            .lock()
+            .expect("store handle lock is not poisoned")
+            .clone();
+        // The last handle's drop closes the database after the count reached zero.
+        let is_open = || {
+            store.as_ref().is_some_and(|store| store.strong_count() > 0)
+                || self
+                    .directory
+                    .as_deref()
+                    .is_some_and(is_directory_open_in_this_process)
+        };
+        while is_open() && Instant::now() < deadline {
+            tokio::time::sleep(STORE_CLOSE_POLL).await;
+        }
+        !is_open()
+    }
+}