Cluster Raft: Persist state and WAL in nodes.conf

design-docs/cluster-raft.md

@@ -145,7 +145,9 @@ FAILOVER <replica-id> <primary-id>
 
 NODE_INFO <node-id> <address-string> <flags>
     Update node address and self-set flags. The address-string uses
-    the nodes.conf format. Flags is "nofailover" or "noflags".
+    the nodes.conf format but excludes shard-id, which is not a
+    property of the node but of the shard and is managed by NODE_JOIN
+    and SET_REPLICA_OF. Flags is "nofailover" or "noflags".
 
 NODE_FAIL <node-id>
     Mark a node as failed. Proposed by the leader when a peer exceeds
@@ -575,9 +577,73 @@ every 10 seconds.
 
 ## Persistence
 
-TODO: The Raft log and state (currentTerm, votedFor) will be persisted
-in nodes.conf using the vars section for Raft state and additional
-lines for uncommitted log entries.
+Raft state is persisted in `nodes.conf`. The file has three sections:
+
+1. **Node lines** — the cluster state snapshot (nodes, slots, replication
+   topology) as of `lastApplied`.
+2. **Vars line** — `currentTerm`, `lastApplied`, `votedFor`, `raftLeader`.
+3. **Log lines** — entries with index > `lastApplied` appended at the end.
+
+Only `currentTerm`, `votedFor`, and the log require persistence for
+safety; `commitIndex` is rediscovered from the leader after restart
+(Raft paper §5.2, Figure 2).
+
+Log line format:
+
+    log <index> <term> <type> <data>
+
+### Full rewrite
+
+A full rewrite (atomic write to temp file + rename + fsync) is triggered
+when:
+
+- `currentTerm` changes (step-down on higher term).
+- `votedFor` changes (granting a vote or starting an election).
+- An applied entry affects `myself` (SLOT_CHANGE, SET_REPLICA_OF,
+  FAILOVER, NODE_JOIN promotion from learner).
+
+A full rewrite updates the snapshot (node lines reflect the current
+applied state), updates `lastApplied` in vars, and writes only
+unapplied log entries in the tail.
+
+### Append-only
+
+When new log entries arrive (from AE or local proposals), they are
+appended to the end of the file with a single batched write + fsync
+(deferred to `beforeSleep`). No full rewrite is needed.
+
+**Safety invariant:** The fsync must complete before the AE_ACK reaches
+the leader. This is enforced by deferring the success AE_ACK: the AE
+handler sets `todo_send_ae_ack` instead of sending immediately, and
+`beforeSleep` sends the ACK only after persistence completes. If fsync
+is ever moved to a background thread, the ACK must be deferred until
+the background fsync completes.
+
+### Startup
+
+On load:
+
+1. Node lines are parsed → cluster state restored (snapshot).
+2. Vars line is parsed → `currentTerm`, `votedFor`, `lastApplied` restored.
+   `commit_index` is set to `lastApplied`.
+3. Log lines are parsed → entries added to the in-memory log.
+4. If the node is a replica, replication is started.
+
+The leader will send AE after reconnection, updating `commit_index`.
+Entries between `lastApplied + 1` and the new `commit_index` are then
+applied.
+
+### Crash detection
+
+Not yet implemented. A future improvement will add per-line checksums
+to detect partial writes and corruption (see Future Work).
+
+### File format
+
+The node lines share the same format as the gossip protocol (code
+reuse), but the vars and log lines are raft-specific. The file is not
+compatible between protocols — switching from gossip to raft (or vice
+versa) requires removing nodes.conf.
 
 ## Shard Epoch (not yet implemented)
 
@@ -631,7 +697,6 @@ targets.
   entries, especially don't trigger primary/replica failovers in a
   minority partition).
 - Log compaction / snapshotting for lagging followers.
-- Persistence of Raft log to disk.
 - Learners (non-voting members): reduces the risk for split-vote for
   leader election in large clusters and reduces commit overhead.
 - Leader transfer on CLUSTER FORGET where the target is the leader.
@@ -641,3 +706,5 @@ targets.
   atomically or not.
 - `valkey-cli --cluster` tooling compatibility (uses MULTI internally in
   some cases, which doesn't work with blocking admin commands).
+- Checksums for appended log lines to detect partial writes and
+  bit-rot, instead of relying solely on trailing newline detection.

src/cluster.c

@@ -158,12 +158,12 @@ void clusterCron(void) {
     clusterCurrentBus->cron();
     clusterCheckReplicaMigration();
 }
-void clusterBeforeSleep(void) {
-    if (server.cluster->before_sleep_handle_slot_migration) {
+void clusterBeforeSleep(bool blocked) {
+    if (!blocked && server.cluster->before_sleep_handle_slot_migration) {
         server.cluster->before_sleep_handle_slot_migration = 0;
         clusterSlotMigrationCron();
     }
-    clusterCurrentBus->beforeSleep();
+    clusterCurrentBus->beforeSleep(blocked);
 }
 static void clusterAutoFailoverOnShutdown(void) {
     if (!nodeIsPrimary(myself)) return;

src/cluster.h

@@ -54,7 +54,7 @@ struct clusterState;
 void clusterInit(void);
 void clusterInitLast(void);
 void clusterCron(void);
-void clusterBeforeSleep(void);
+void clusterBeforeSleep(bool blocked);
 int verifyClusterConfigWithData(void);
 void clusterPrepareShutdown(void);
 void clusterHandleServerShutdown(bool auto_failover);

src/cluster_bus.h

@@ -20,7 +20,7 @@ typedef struct clusterBusType {
     void (*init)(void);
     void (*initLast)(void);
     void (*cron)(void);
-    void (*beforeSleep)(void);
+    void (*beforeSleep)(bool blocked);
     void (*prepareShutdown)(void); /* Called early in shutdown to allow graceful handoff. */
     void (*handleServerShutdown)(void);
 
@@ -83,6 +83,13 @@ typedef struct clusterBusType {
      * Returns 1 if the variable was handled, 0 otherwise. */
     int (*parseVarsLine)(const char *name, const char *value);
 
+    /* Parse a "log" line from nodes.conf. argv/argc are the split line
+     * (argv[0] is "log"). */
+    void (*parseLogLine)(sds *argv, int argc);
+
+    /* Append "log" lines during a full config rewrite. */
+    sds (*appendLogLines)(sds config);
+
     /* Called after nodes.conf is fully loaded, for post-load fixups
      * (e.g. ensuring currentEpoch >= max configEpoch). If NULL, skipped. */
     void (*postLoad)(void);

src/cluster_legacy.c

@@ -4653,7 +4653,8 @@ static void clusterLegacyCron(void) {
  * reaction to events fired but that are not safe to perform inside event
  * handlers, or to perform potentially expansive tasks that we need to do
  * a single time before replying to clients. */
-static void clusterLegacyBeforeSleep(void) {
+static void clusterLegacyBeforeSleep(bool blocked) {
+    if (blocked) return;
     int flags = LEGACY_STATE()->todo_before_sleep;
 
     /* Reset our flags (not strictly needed since every single function

src/cluster_nodes.c

@@ -95,9 +95,13 @@ static int auxShardIdSetter(clusterNode *n, void *value, size_t length) {
     if (verifyClusterNodeId(value, length) == C_ERR) {
         return C_ERR;
     }
+    if (memcmp(n->shard_id, value, CLUSTER_NAMELEN) == 0) {
+        return C_OK;
+    }
+    clusterRemoveNodeFromShard(n);
     memcpy(n->shard_id, value, CLUSTER_NAMELEN);
-    /* if n already has replicas, make sure they all agree
-     * on the shard id. If not, update them. */
+    clusterAddNodeToShard(n->shard_id, n);
+    /* If n has replicas, make sure they follow the new shard id. */
     for (int i = 0; i < n->num_replicas; i++) {
         if (memcmp(n->replicas[i]->shard_id, n->shard_id, CLUSTER_NAMELEN) != 0) {
             serverLog(LL_NOTICE,
@@ -109,7 +113,6 @@ static int auxShardIdSetter(clusterNode *n, void *value, size_t length) {
             clusterAddNodeToShard(n->shard_id, n->replicas[i]);
         }
     }
-    clusterAddNodeToShard(value, n);
     return C_OK;
 }
 
@@ -328,8 +331,9 @@ static sds representSlotInfo(sds ci, uint16_t *slot_info_pairs, int slot_info_pa
 }
 
 /* Append the address+aux string for a node to an sds: ip:port@cport[,hostname][,aux=val]*
- * This is the same format used in the second column of nodes.conf. */
-sds clusterNodeAppendAddressString(sds s, clusterNode *node, int tls_primary) {
+ * This is the same format used in the second column of nodes.conf.
+ * skip_aux is a bitmask of auxFieldIndex values to omit. */
+static sds clusterNodeAppendAddressStringSkip(sds s, clusterNode *node, int tls_primary, unsigned int skip_aux) {
     int port = tls_primary ? node->tls_port : node->tcp_port;
     s = sdscatfmt(s, "%s:%i@%i", node->ip, port, node->cport);
     if (sdslen(node->hostname) != 0) {
@@ -338,6 +342,7 @@ sds clusterNodeAppendAddressString(sds s, clusterNode *node, int tls_primary) {
         s = sdscatlen(s, ",", 1);
     }
     for (int i = af_count - 1; i >= 0; i--) {
+        if (skip_aux & (1u << i)) continue;
         if ((tls_primary && i == af_tls_port) || (!tls_primary && i == af_tcp_port)) continue;
         if (auxFieldHandlers[i].isPresent(node)) {
             s = sdscatfmt(s, ",%s=", auxFieldHandlers[i].field);
@@ -347,6 +352,16 @@ sds clusterNodeAppendAddressString(sds s, clusterNode *node, int tls_primary) {
     return s;
 }
 
+sds clusterNodeAppendAddressString(sds s, clusterNode *node, int tls_primary) {
+    return clusterNodeAppendAddressStringSkip(s, node, tls_primary, 0);
+}
+
+/* Like clusterNodeAppendAddressString but omits shard-id, which is managed
+ * by dedicated raft log entries (NODE_JOIN, SET_REPLICA_OF). */
+sds clusterNodeAppendAddressStringNoShardId(sds s, clusterNode *node, int tls_primary) {
+    return clusterNodeAppendAddressStringSkip(s, node, tls_primary, 1u << af_shard_id);
+}
+
 /* Parse an address+aux string onto a node. The string format is:
  * ip:port@cport[,hostname][,aux=val]*
  * Returns C_OK on success, C_ERR on parse error. The input string is modified. */
@@ -670,6 +685,14 @@ int clusterLoadConfig(char *filename) {
             continue;
         }
 
+        /* Handle "log" lines (protocol-specific WAL entries). */
+        if (strcasecmp(argv[0], "log") == 0) {
+            if (clusterCurrentBus->parseLogLine)
+                clusterCurrentBus->parseLogLine(argv, argc);
+            sdsfreesplitres(argv, argc);
+            continue;
+        }
+
         /* Regular config lines have at least eight fields */
         if (argc < 8) {
             sdsfreesplitres(argv, argc);
@@ -894,6 +917,8 @@ int clusterSaveConfig(int do_fsync) {
     ci = clusterGenNodesDescription(NULL, CLUSTER_NODE_HANDSHAKE, 0);
     if (clusterCurrentBus->appendVarsLine)
         ci = clusterCurrentBus->appendVarsLine(ci);
+    if (clusterCurrentBus->appendLogLines)
+        ci = clusterCurrentBus->appendLogLines(ci);
     content_size = sdslen(ci);
 
     /* Create a temp file with the new content. */

src/cluster_nodes.h

@@ -5,6 +5,7 @@
 
 /* Node address string: ip:port@cport[,hostname][,aux=val]* */
 sds clusterNodeAppendAddressString(sds s, clusterNode *node, int tls_primary);
+sds clusterNodeAppendAddressStringNoShardId(sds s, clusterNode *node, int tls_primary);
 int clusterNodeParseAddressString(clusterNode *n, char *str);
 
 /* Node description / serialization. */