Raft Metadata Object-Store Archive and Restore From

.github/workflows/e2e.yml

@@ -54,6 +54,28 @@ jobs:
           java -jar cli/target/streamstack.jar ds bench --endpoint http://127.0.0.1:4437 \
             -b 1024 -n 64 -w 8 -d 8
 
+      - name: Archive metadata snapshot
+        run: |
+          curl -sf -X PUT -H 'Content-Type: application/json' -d '' http://127.0.0.1:4437/e2e/restore
+          curl -sf -X POST -H 'Content-Type: application/json' -d '{"payload":"restore-me"}' http://127.0.0.1:4437/e2e/restore
+          curl -sf -X POST http://127.0.0.1:9090/admin/snapshot | jq -e '.appliedIndex > 0'
+          timeout 60 bash -c "until curl -sf http://127.0.0.1:9090/admin/snapshots | jq -e '.snapshots | length > 0' > /dev/null; do sleep 2; done"
+
+      - name: Restore from storage after data dir loss
+        run: |
+          docker compose --env-file harness/local/.env \
+            -f harness/local/docker-compose.minio.yml \
+            -f harness/local/docker-compose.ds.yml \
+            rm -sf node1
+          docker volume rm local_node1-data
+          RESTORE_FROM_STORAGE=true docker compose --env-file harness/local/.env \
+            -f harness/local/docker-compose.minio.yml \
+            -f harness/local/docker-compose.ds.yml \
+            up -d node1
+          timeout 180 bash -c 'until curl -sf http://127.0.0.1:9090/ready; do sleep 2; done'
+          curl -sf http://127.0.0.1:9090/admin/streams/e2e/restore | jq -e '.ownerLocal == true and .streamId != null'
+          curl -sf -m 30 http://127.0.0.1:4437/e2e/restore | grep -q '"payload":"restore-me"'
+
       - name: Collect logs
         if: always()
         run: |
@@ -169,6 +191,27 @@ jobs:
           curl -sf -X PUT -H 'Content-Type: application/json' -d '' "http://127.0.0.1:${http_port}/e2e/failover"
           curl -sf -X POST -H 'Content-Type: application/json' -d '{"after":"failover"}' "http://127.0.0.1:${http_port}/e2e/failover"
           curl -sf -m 30 "http://127.0.0.1:${http_port}/e2e/failover" | grep -q '"after":"failover"'
+          echo "$leader" > killed-leader.txt
+
+      - name: Replace node with empty data dir
+        run: |
+          node=$(cat killed-leader.txt)
+          admin_port=$((9090 + node))
+          echo "replacing node ${node} with a fresh data volume"
+          docker compose --env-file harness/local/.env \
+            -f harness/local/docker-compose.minio.yml \
+            -f harness/local/docker-compose.cluster.ds.yml \
+            rm -sf "node${node}"
+          docker volume rm "local_node${node}-data"
+          docker compose --env-file harness/local/.env \
+            -f harness/local/docker-compose.minio.yml \
+            -f harness/local/docker-compose.cluster.ds.yml \
+            up -d "node${node}"
+          timeout 180 bash -c "until curl -sf http://127.0.0.1:${admin_port}/ready; do sleep 2; done"
+          curl -sf "http://127.0.0.1:${admin_port}/admin/cluster" | jq -e '.registered == true and .raft.appliedIndex > 0'
+          curl -sf "http://127.0.0.1:${admin_port}/admin/nodes" | jq -e '.nodes | length == 3'
+          http_port=$((4436 + node))
+          curl -sfL -m 30 "http://127.0.0.1:${http_port}/e2e/failover" | grep -q '"after":"failover"'
 
       - name: Collect logs
         if: always()

dashboard/src/api.ts

@@ -31,6 +31,20 @@ export interface Readiness {
   registered: boolean
 }
 
+export interface ArchivedSnapshot {
+  key: string
+  appliedIndex: number
+  timestampMs: number
+  size: number
+}
+
+export interface SnapshotArchiveInfo {
+  archiveSuccessCount: number
+  archiveFailureCount: number
+  lastArchivedIndex: number
+  snapshots: ArchivedSnapshot[]
+}
+
 async function get<T>(path: string): Promise<T> {
   const res = await fetch(path, { headers: { Accept: 'application/json' } })
   const body = (await res.json()) as T
@@ -45,3 +59,17 @@ async function get<T>(path: string): Promise<T> {
 export const fetchCluster = () => get<ClusterInfo>('/admin/cluster')
 export const fetchNodes = () => get<{ nodes: NodeInfo[] }>('/admin/nodes')
 export const fetchReady = () => get<Readiness>('/ready')
+
+export async function fetchSnapshots(): Promise<SnapshotArchiveInfo | null> {
+  const res = await fetch('/admin/snapshots', { headers: { Accept: 'application/json' } })
+
+  if (res.status === 404) {
+    return null
+  }
+
+  if (!res.ok) {
+    throw new Error(`GET /admin/snapshots failed: ${res.status}`)
+  }
+
+  return (await res.json()) as SnapshotArchiveInfo
+}

dashboard/src/app.tsx

@@ -3,9 +3,11 @@ import {
   fetchCluster,
   fetchNodes,
   fetchReady,
+  fetchSnapshots,
   type ClusterInfo,
   type NodeInfo,
   type Readiness,
+  type SnapshotArchiveInfo,
 } from './api'
 
 const POLL_INTERVAL_MS = 2000
@@ -44,6 +46,7 @@ export function App() {
   const [cluster, setCluster] = useState<ClusterInfo | null>(null)
   const [nodes, setNodes] = useState<NodeInfo[]>([])
   const [ready, setReady] = useState<Readiness | null>(null)
+  const [snapshots, setSnapshots] = useState<SnapshotArchiveInfo | null>(null)
   const [error, setError] = useState<string | null>(null)
   const [updatedAt, setUpdatedAt] = useState<Date | null>(null)
 
@@ -52,7 +55,12 @@ export function App() {
 
     async function poll() {
       try {
-        const [c, n, r] = await Promise.all([fetchCluster(), fetchNodes(), fetchReady()])
+        const [c, n, r, s] = await Promise.all([
+          fetchCluster(),
+          fetchNodes(),
+          fetchReady(),
+          fetchSnapshots(),
+        ])
 
         if (!alive) {
           return
@@ -61,6 +69,7 @@ export function App() {
         setCluster(c)
         setNodes(n.nodes)
         setReady(r)
+        setSnapshots(s)
         setError(null)
         setUpdatedAt(new Date())
       } catch (e) {
@@ -185,6 +194,37 @@ export function App() {
         )}
       </section>
 
+      <section class="ss-section">
+        <h2>Metadata snapshots</h2>
+        {!snapshots ? (
+          <div class="ss-empty">Snapshot archive disabled</div>
+        ) : snapshots.snapshots.length === 0 ? (
+          <div class="ss-empty">No archived snapshots yet</div>
+        ) : (
+          <table>
+            <thead>
+              <tr>
+                <th>Applied index</th>
+                <th>Archived at</th>
+                <th>Size</th>
+              </tr>
+            </thead>
+            <tbody>
+              {[...snapshots.snapshots].reverse().map((s) => (
+                <tr key={s.key}>
+                  <td class="mono">{s.appliedIndex}</td>
+                  <td class="mono">{new Date(s.timestampMs).toLocaleString()}</td>
+                  <td class="mono">{s.size}</td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        )}
+        {snapshots && snapshots.archiveFailureCount > 0 ? (
+          <div class="ss-empty">Archive failures: {snapshots.archiveFailureCount}</div>
+        ) : null}
+      </section>
+
       <footer class="ss-footer">
         {error
           ? `Last error: ${error}`

docs/pages/deployment.md

@@ -51,9 +51,34 @@ Every node serves an admin plane on a separate port, independent of the protocol
 | `POST /admin/peers` | Add a raft peer: `{"peer": "host:port"}` (leader only) |
 | `DELETE /admin/peers/{peer}` | Remove a raft peer (leader only) |
 | `POST /admin/transfer-leader` | Transfer raft leadership: `{"peer": "host:port"}` (leader only) |
+| `POST /admin/snapshot` | Trigger a metadata raft snapshot immediately |
+| `GET /admin/snapshots` | Archived metadata snapshots in object storage, with archive health |
 
 The admin port also serves a read-only dashboard at `/` with cluster, node, and raft status.
 
 Point load balancer target-group health checks at `/ready`. On shutdown the node flips `/ready` to `503` first and waits `shutdownDrainSec` (topo config, default `0`) before closing, so the load balancer drains in-flight traffic.
 
-At startup the node verifies the storage and WAL buckets are reachable, retrying with backoff for up to ten attempts before failing; `/health` responds while it retries.
+At startup the node verifies the storage and WAL buckets are reachable, retrying with backoff for up to ten attempts before failing; `/health` responds while it retries.
+
+## Metadata durability
+
+Stream data lives in object storage, but metadata (streams, object registry, KV) lives in each node's raft `dataDir`: a periodic local snapshot plus a bounded log. Raft replication covers single-node loss; the snapshot archive covers losing the data directories of a quorum.
+
+The raft leader archives every metadata snapshot to the storage bucket under `_streamstack/metadata/{clusterId}/snapshots/`, keeping the last five. Archival is asynchronous and never blocks the snapshot itself; failures are logged and reported on `GET /admin/snapshots` and the dashboard. Disable with `metadataArchive: false` in the topo config, `--metadata-archive false` on the CLI, or `METADATA_ARCHIVE=false` in the Docker image. Keep the `_streamstack/` key prefix reserved for StreamStack internals.
+
+Run production nodes on durable disks (EBS, not instance storage) for `dataDir`, with a stable DNS name per node.
+
+### Replacing a node
+
+- Disk survived: attach the volume to a replacement machine with the same `nodeId` and DNS name and start it. The node replays its local snapshot and log and rejoins.
+- Disk lost: start a replacement with the same `nodeId` and DNS name and an empty `dataDir`. The raft leader streams its latest snapshot to the new peer and replays the log tail. If the address changed, fix membership with `POST /admin/peers` and `DELETE /admin/peers/{peer}`.
+
+### Restoring from the archive
+
+If a quorum of data directories is lost, bootstrap from the archive:
+
+1. Start node 1 with an empty `dataDir`, a single-peer topo, and `--restore-from-storage true` (or `RESTORE_FROM_STORAGE=true`). The node loads the latest archived snapshot before forming raft and persists it into a local raft snapshot right after election.
+2. Verify with `GET /admin/cluster` and `GET /admin/streams/{name}`.
+3. Add nodes 2 and 3 with empty data dirs via `POST /admin/peers`; they receive the snapshot from the leader.
+
+Restore only runs on a fresh `dataDir`; if local raft state exists it is skipped, and a log without a snapshot fails with instructions to wipe first. Data appended after the last archived snapshot keeps its bytes in the bucket but has no metadata entries; the snapshot interval (30s) bounds that window.

harness/README.md

@@ -17,6 +17,8 @@ docker compose --env-file harness/local/.env \
 
 Swap `docker-compose.ds.yml` for `docker-compose.s2.yml` to run S2 instead. Node listens on `127.0.0.1:4437`. Admin plane and dashboard: http://127.0.0.1:9090 (`/health`, `/ready`, `/admin/*`).
 
+The raft leader archives metadata snapshots to the storage bucket under `_streamstack/metadata/{clusterId}/snapshots/` (disable with `METADATA_ARCHIVE=false`). Recover a wiped node from the archive with `RESTORE_FROM_STORAGE=true` on a fresh data dir — see the deployment docs for the runbooks.
+
 [BENCH.md](BENCH.md) has DS and S2 smoke/load tests.
 MinIO console: http://127.0.0.1:9001 (`minioadmin` / `minioadmin`).
 

harness/docker/entrypoint.sh

@@ -20,6 +20,12 @@ fi
 if [ -n "${ROUTING:-}" ]; then
   set -- "$@" --routing "$ROUTING"
 fi
+if [ -n "${METADATA_ARCHIVE:-}" ]; then
+  set -- "$@" --metadata-archive "$METADATA_ARCHIVE"
+fi
+if [ -n "${RESTORE_FROM_STORAGE:-}" ]; then
+  set -- "$@" --restore-from-storage "$RESTORE_FROM_STORAGE"
+fi
 
 bucket="${DATA_BUCKET:-${BUCKET_NAME:-}}"
 if [ -n "$bucket" ]; then

harness/local/docker-compose.ds.yml

@@ -16,6 +16,8 @@ services:
       AWS_REGION: ${AWS_REGION:-us-east-1}
       TOPO: /opt/streamstack/topo/topo.yaml
       NODE_ID: "1"
+      METADATA_ARCHIVE: ${METADATA_ARCHIVE:-}
+      RESTORE_FROM_STORAGE: ${RESTORE_FROM_STORAGE:-}
     volumes:
       - .:/opt/streamstack/topo:ro
       - node1-data:/tmp/streamstack

metadata/src/main/java/io/streamstack/metadata/raft/MetadataNode.java

@@ -49,6 +49,8 @@ public final class MetadataNode implements AutoCloseable {
     private final MetadataLifecycle lifecycle;
     private final MetadataHealth health;
     private final ScheduledExecutorService registrar;
+    private final SnapshotArchive snapshotArchive;
+    private final boolean restoredFromArchive;
 
     private final AtomicBoolean registered = new AtomicBoolean(false);
 
@@ -98,6 +100,21 @@ public MetadataNode(
         ObjectStorage objectStorage,
         Options options,
         String httpAddress) throws IOException {
+        this(nodeId, host, port, dataDir, peers, nodeEpoch, objectStorage, options, httpAddress, null, false);
+    }
+
+    public MetadataNode(
+        int nodeId,
+        String host,
+        int port,
+        File dataDir,
+        List<String> peers,
+        long nodeEpoch,
+        ObjectStorage objectStorage,
+        Options options,
+        String httpAddress,
+        SnapshotArchive snapshotArchive,
+        boolean restoreFromArchive) throws IOException {
         this.nodeId = nodeId;
         this.nodeEpoch = nodeEpoch;
         this.httpAddress = Objects.isNull(httpAddress) ? "" : httpAddress;
@@ -111,6 +128,9 @@ public MetadataNode(
         Files.createDirectories(logDir.toPath());
         Files.createDirectories(metaDir.toPath());
         Files.createDirectories(snapshotDir.toPath());
+        this.snapshotArchive = snapshotArchive;
+        this.restoredFromArchive = restoreFromArchive && restore(logDir, snapshotDir);
+        this.stateMachine.setSnapshotArchive(snapshotArchive);
         NodeOptions nodeOptions = new NodeOptions();
 
         nodeOptions.setElectionTimeoutMs(options.electionTimeoutMs());
@@ -163,6 +183,37 @@ public MetadataNode(
         this.registrar.scheduleWithFixedDelay(this::tryRegister, 100, 500, TimeUnit.MILLISECONDS);
     }
 
+    private boolean restore(File logDir, File snapshotDir) throws IOException {
+        if (Objects.isNull(snapshotArchive)) {
+            throw new IllegalStateException("restore from storage requested but no snapshot archive configured");
+        }
+
+        if (!isEmptyDir(snapshotDir)) {
+            LOGGER.info("local raft snapshot exists, skipping restore from storage nodeId={}", nodeId);
+            return false;
+        }
+
+        if (!isEmptyDir(logDir)) {
+            throw new IllegalStateException(
+                "raft log exists without a local snapshot; wipe the metadata data dir before restoring from storage");
+        }
+
+        SnapshotArchive.ArchivedSnapshot latest = snapshotArchive.latest().orElseThrow(() ->
+            new IllegalStateException("restore from storage requested but no archived metadata snapshot found"));
+
+        stateMachine.restore(snapshotArchive.read(latest));
+        LOGGER.info("restored metadata from archived snapshot nodeId={} key={} appliedIndex={}",
+            nodeId, latest.key(), latest.appliedIndex());
+
+        return true;
+    }
+
+    private static boolean isEmptyDir(File dir) throws IOException {
+        try (var entries = Files.list(dir.toPath())) {
+            return entries.findAny().isEmpty();
+        }
+    }
+
     private void tryRegister() {
         if (registered.get()) {
             registrar.shutdown();
@@ -211,6 +262,14 @@ public MetadataStateMachine stateMachine() {
         return stateMachine;
     }
 
+    public SnapshotArchive snapshotArchive() {
+        return snapshotArchive;
+    }
+
+    public boolean restoredFromArchive() {
+        return restoredFromArchive;
+    }
+
     public MetadataClient client() {
         return client;
     }