Add TODO

Several bugs found via deterministic simulation testing (500 seeds, 30s each):

WAL format: Store full WALEntry on disk (including votes and checksum)
instead of a separate WALEntryDisk. This eliminates the need to
reconstruct transient vote bits after restart, and adds wal_update_entry()
for persisting in-place vote modifications.

Crash recovery: Restore 3-case startup logic (empty=normal, corrupt=recovery,
clean=replay). Fix last_normal_view initialization on clean restart. Call
adopt_view() after replay to reinitialize leader vote bits that may be stale
if a crash interrupted wal_replace before the atomic rename.

View change ordering: Move wal_replace before persist_state in
process_begin_view so the on-disk log is always at least as recent as the
persisted metadata. Reset votes on view_change_log entries before wal_replace
in complete_view_change_and_become_primary and complete_recovery.

View adoption: Add adopt_view() helper called from process_prepare and
process_prepare_ok when a NORMAL node learns of a higher view. Updates
last_normal_view and initializes leader vote bits.

Invariant checker: Track node liveness across ticks (prev_alive[]) so
min_commit monotonicity is correctly relaxed for nodes that just restarted.

https://claude.ai/code/session_01X2b5VRntA7LM32kdN9tPAM

.gitignore

@@ -1,5 +1,6 @@
 toastyfs
 toastyfs_client
+toastyfs_random_client
 toastyfs_simulation
 *.gcno
 *.gcda
@@ -13,5 +14,6 @@ coverage_html/
 example
 example_large
 examples/example
+.claude/
 .cluster/
 vsr_boot_marker

Makefile

@@ -15,20 +15,20 @@ SHARED_LIB = libtoastyfs.so
 # ---- Server binary ----
 
 SERVER_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
-              src/tcp.c src/server.c src/main.c src/log.c src/client_table.c \
+              src/tcp.c src/server.c src/main.c src/wal.c src/client_table.c \
               src/chunk_store.c src/metadata.c
 
 # ---- Client binary (random test client) ----
 
 CLIENT_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
               src/tcp.c src/server.c src/client.c src/random_client.c src/main.c \
-              src/log.c src/client_table.c src/chunk_store.c src/metadata.c
+              src/wal.c src/client_table.c src/chunk_store.c src/metadata.c
 
 # ---- Simulation binary ----
 
 SIM_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
            src/tcp.c src/server.c src/client.c src/random_client.c src/main.c \
-           src/log.c src/client_table.c src/invariant_checker.c src/chunk_store.c \
+           src/wal.c src/client_table.c src/invariant_checker.c src/chunk_store.c \
            src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c
 
 # ---- Default target ----

TODO.txt

@@ -1,3 +1,4 @@
 [ ] Add log compaction
 [ ] Add log persistence
-[ ] Add chunk locations to metadata
+[ ] Add chunk locations to metadata
+[ ] fsync parent directory when renaming log

build.sh

@@ -1,3 +1,3 @@
-gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/random_client.c src/main.c src/log.c src/client_table.c src/invariant_checker.c src/chunk_store.c src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c -o toastyfs_simulation -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION
+gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/random_client.c src/main.c src/wal.c src/client_table.c src/invariant_checker.c src/chunk_store.c src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c -o toastyfs_simulation -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_SIMULATION -DFAULT_INJECTION
-gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs        -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_SERVER
+gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/main.c src/wal.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs        -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_SERVER
-gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/random_client.c src/main.c src/log.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs_random_client -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_CLIENT
+gcc src/basic.c src/file_system.c src/byte_queue.c src/message.c src/tcp.c src/server.c src/client.c src/random_client.c src/main.c src/wal.c src/client_table.c src/chunk_store.c src/metadata.c -o toastyfs_random_client -Iquakey/include -Iinclude -I. -Wall -Wextra -ggdb -O0 -DMAIN_CLIENT

quakey/src/quakey.c

@@ -3249,7 +3249,29 @@ int mock_close(int fd)
 // data for chunks they don't actually have).
 int mock_access(const char *path, int mode)
 {
-    assert(0); // TODO
+    (void) mode; // No permission bits in mockfs; all checks reduce to existence
+
+    Host *host = host___;
+    if (host == NULL)
+        abort_("Call to mock_access() with no node scheduled\n");
+
+    if (!host_is_linux(host))
+        abort_("Call to mock_access() not from Linux\n");
+
+    MockFS_OpenFile open_file;
+    int ret = mockfs_open(host->mfs, (char *) path, strlen(path), MOCKFS_O_RDONLY, &open_file);
+    if (ret == MOCKFS_ERRNO_ISDIR) {
+        // Path exists and is a directory
+        return 0;
+    }
+    if (ret < 0) {
+        *host_errno_ptr(host) = ENOENT;
+        return -1;
+    }
+
+    // File exists; close it immediately
+    mockfs_close_file(&open_file);
+    return 0;
 }
 
 static int convert_addr(void *addr, size_t addr_len,

src/client.c

@@ -219,7 +219,8 @@ static void send_message_to_server(ToastyFS *tfs, int server_idx, MessageHeader
     byte_queue_write(output, msg, msg->length);
 }
 
-static void send_message_to_server_ex(ToastyFS *tfs, int server_idx,
+static void
+send_message_to_server_ex(ToastyFS *tfs, int server_idx,
     MessageHeader *msg, void *extra, int extra_len)
 {
     ByteQueue *output;
@@ -381,8 +382,8 @@ static void replay_request(ToastyFS *tfs)
     }
 }
 
-static int process_message(ToastyFS *tfs,
+static int process_message(ToastyFS *tfs, int conn_idx,
-    int conn_idx, uint8_t type, ByteView msg)
+    uint8_t type, ByteView msg)
 {
     (void) conn_idx;
 
@@ -673,7 +674,8 @@ static int process_message(ToastyFS *tfs,
     return 0;
 }
 
-void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pnum)
+void toastyfs_process_events(ToastyFS *tfs, void **ctxs,
+    struct pollfd *pdata, int pnum)
 {
     Event events[TCP_EVENT_CAPACITY];
     int num_events = tcp_translate_events(&tfs->tcp, events, ctxs, pdata, pnum);
@@ -748,7 +750,8 @@ void toastyfs_process_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, i
 // TODO: The toastyfs client needs to determine a timeout based on the
 //       pending operation status, not just use PRIMARY_DEATH_TIMEOUT_SEC
 //       for everything.
-int toastyfs_register_events(ToastyFS *tfs, void **ctxs, struct pollfd *pdata, int pcap, int *timeout)
+int toastyfs_register_events(ToastyFS *tfs, void **ctxs,
+    struct pollfd *pdata, int pcap, int *timeout)
 {
     Time now = get_current_time(); // TODO: Handle INVALID_TIME error
     Time deadline = INVALID_TIME;

src/invariant_checker.c

@@ -58,8 +58,10 @@ void invariant_checker_init(InvariantChecker *ic)
 {
     ic->last_min_commit = -1;
     ic->last_max_commit = -1;
-    for (int i = 0; i < NODE_LIMIT; i++)
+    for (int i = 0; i < NODE_LIMIT; i++) {
         ic->prev_status[i] = STATUS_NORMAL;
+        ic->prev_alive[i] = false;
+    }
     ic->shadow_log = NULL;
     ic->shadow_count = 0;
     ic->shadow_capacity = 0;
@@ -96,7 +98,11 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
 
         if (min_commit < 0 || min_commit > n->commit_index) {
             min_commit = n->commit_index;
-            min_commit_just_recovered = (ic->prev_status[i] == STATUS_RECOVERY);
+            // A node that just recovered or just restarted after a crash
+            // may have a stale commit_index (persisted state can lag behind
+            // the in-memory value if the crash interrupted file_sync).
+            min_commit_just_recovered = (ic->prev_status[i] == STATUS_RECOVERY)
+                                     || !ic->prev_alive[i];
         }
 
         if (max_commit < 0 || max_commit < n->commit_index) {
@@ -153,6 +159,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
     ic->last_min_commit = min_commit;
     ic->last_max_commit = max_commit;
     for (int i = 0; i < num_nodes; i++) {
+        ic->prev_alive[i] = (nodes[i] != NULL);
         if (nodes[i])
             ic->prev_status[i] = nodes[i]->status;
     }
@@ -164,9 +171,9 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
 
         // 1. commit_index <= log.count
         //    A node cannot have committed more entries than it has in its log.
-        if (s->commit_index > s->log.count) {
+        if (s->commit_index > s->wal.count) {
             fprintf(stderr, "INVARIANT VIOLATED: node %d: commit_index (%d) > log.count (%d)\n",
-                i, s->commit_index, s->log.count);
+                i, s->commit_index, s->wal.count);
             __builtin_trap();
         }
 
@@ -206,11 +213,11 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
         // 9. Log entry view numbers must not exceed the node's current view.
         //    Entries are created in the view they were proposed. No entry
         //    should carry a view number from the future.
-        for (int k = 0; k < s->log.count; k++) {
+        for (int k = 0; k < s->wal.count; k++) {
-            if ((uint64_t)s->log.entries[k].view_number > s->view_number) {
+            if ((uint64_t)s->wal.entries[k].view_number > s->view_number) {
                 fprintf(stderr, "INVARIANT VIOLATED: node %d: log[%d].view_number (%d) "
                     "> view_number (%lu)\n",
-                    i, k, s->log.entries[k].view_number,
+                    i, k, s->wal.entries[k].view_number,
                     (unsigned long)s->view_number);
                 __builtin_trap();
             }
@@ -221,8 +228,8 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
         //     votes for its own entries.
         if (s->status == STATUS_NORMAL && is_leader(s)) {
             int idx = self_idx(s);
-            for (int k = s->commit_index; k < s->log.count; k++) {
+            for (int k = s->commit_index; k < s->wal.count; k++) {
-                if (!(s->log.entries[k].votes & (1 << idx))) {
+                if (!(s->wal.entries[k].votes & (1 << idx))) {
                     fprintf(stderr, "INVARIANT VIOLATED: node %d (leader): "
                         "uncommitted log[%d] missing leader's own vote bit\n",
                         i, k);
@@ -266,7 +273,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
                 mc = nodes[j]->commit_index;
 
             for (int k = 0; k < mc; k++) {
-                if (memcmp(&nodes[i]->log.entries[k].oper, &nodes[j]->log.entries[k].oper, sizeof(MetaOper)) != 0) {
+                if (memcmp(&nodes[i]->wal.entries[k].oper, &nodes[j]->wal.entries[k].oper, sizeof(MetaOper)) != 0) {
                     fprintf(stderr, "INVARIANT VIOLATED: committed log operation mismatch at index %d "
                         "between node %d and node %d\n", k, i, j);
                     __builtin_trap();
@@ -298,11 +305,11 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
     if (source_node_idx >= 0 && observed_max_commit > ic->shadow_count) {
 
         ServerState *source = nodes[source_node_idx];
-        assert(source->log.count >= observed_max_commit);
+        assert(source->wal.count >= observed_max_commit);
 
         for (int k = ic->shadow_count; k < observed_max_commit; k++) {
 
-            MetaOper *source_oper = &source->log.entries[k].oper;
+            MetaOper *source_oper = &source->wal.entries[k].oper;
 
             // Cross-validate against other live non-recovering nodes
             // that have also committed this entry.
@@ -315,10 +322,10 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
                     continue;
                 if (nodes[j]->commit_index <= k)
                     continue;
-                if (nodes[j]->log.count <= k)
+                if (nodes[j]->wal.count <= k)
                     continue;
 
-                if (memcmp(&nodes[j]->log.entries[k].oper, source_oper, sizeof(MetaOper)) != 0) {
+                if (memcmp(&nodes[j]->wal.entries[k].oper, source_oper, sizeof(MetaOper)) != 0) {
                     fprintf(stderr, "INVARIANT VIOLATED: committed entry mismatch at index %d "
                         "between source node %d and node %d during shadow log append\n",
                         k, source_node_idx, j);
@@ -340,15 +347,15 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
         for (int i = 0; i < num_nodes; i++) {
             if (nodes[i] == NULL)
                 continue;
-            if (nodes[i]->log.count <= k)
+            if (nodes[i]->wal.count <= k)
                 continue;
             if (nodes[i]->commit_index <= k)
                 continue;
 
-            if (memcmp(&nodes[i]->log.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) != 0) {
+            if (memcmp(&nodes[i]->wal.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) != 0) {
                 char shadow_buf[128], node_buf[128];
                 meta_snprint_oper(shadow_buf, sizeof(shadow_buf), &ic->shadow_log[k]);
-                meta_snprint_oper(node_buf, sizeof(node_buf), &nodes[i]->log.entries[k].oper);
+                meta_snprint_oper(node_buf, sizeof(node_buf), &nodes[i]->wal.entries[k].oper);
                 fprintf(stderr, "INVARIANT VIOLATED: shadow log mismatch at index %d on node %d\n"
                     "  shadow: %s\n"
                     "  node:   %s\n",
@@ -377,9 +384,9 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
                     num_dead++;
                     continue;
                 }
-                if (nodes[i]->log.count <= k)
+                if (nodes[i]->wal.count <= k)
                     continue;
-                if (memcmp(&nodes[i]->log.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) == 0)
+                if (memcmp(&nodes[i]->wal.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) == 0)
                     holders++;
             }
 

src/log.c

@@ -1,3 +1,4 @@
+#if 0
 #if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
 #define QUAKEY_ENABLE_MOCKS
 #endif
@@ -52,4 +53,6 @@ int log_append(Log *log, LogEntry entry)
 
     log->entries[log->count++] = entry;
     return 0;
-}
+}
+
+#endif

src/log.h

@@ -1,3 +1,4 @@
+#if 0
 #ifndef LOG_INCLUDED
 #define LOG_INCLUDED
 
@@ -27,4 +28,5 @@ void log_free(Log *log);
 void log_move(Log *dst, Log *src);
 int  log_append(Log *log, LogEntry entry);
 
-#endif // LOG_INCLUDED
+#endif // LOG_INCLUDED
+#endif

src/server.c

@@ -11,6 +11,7 @@
 typedef enum {
     HR_OK,
     HR_OUT_OF_MEMORY,
+    HR_IO_FAILURE,
     HR_INVALID_MESSAGE,
 } HandlerResult;
 
@@ -58,7 +59,7 @@ static void node_log_impl(ServerState *state, const char *event, const char *det
         status_name(state->status),
         state->view_number,
         state->commit_index,
-        state->log.count,
+        state->wal.count,
         event,
         detail ? detail : "");
 }
@@ -132,6 +133,23 @@ static void broadcast_to_peers(ServerState *state, MessageHeader *msg)
     broadcast_to_peers_ex(state, msg, NULL, 0);
 }
 
+// Called when a node in NORMAL status adopts a higher view_number
+// (e.g. from a PREPARE or PREPARE_OK sent in a newer view).
+// Updates last_normal_view to match and, if the new view makes this
+// node the primary, initialises vote bits for uncommitted entries.
+static void adopt_view(ServerState *state)
+{
+    state->last_normal_view = state->view_number;
+    if (is_primary(state)) {
+        for (int i = state->commit_index; i < state->wal.count; i++) {
+            WALEntry *entry = &state->wal.entries[i];
+            entry->votes = 0;
+            add_vote(&entry->votes, self_idx(state));
+            wal_update_entry(&state->wal, i);
+        }
+    }
+}
+
 static void begin_state_transfer(ServerState *state, int sender_idx)
 {
     if (state->state_transfer_pending)
@@ -145,10 +163,10 @@ static void begin_state_transfer(ServerState *state, int sender_idx)
             .length  = sizeof(GetStateMessage),
         },
         .view_number = state->view_number,
-        .op_number   = state->log.count,
+        .op_number   = state->wal.count,
         .sender_idx  = self_idx(state),
     };
-    node_log(state, "SEND GET_STATE", "to=%d op=%d", sender_idx, state->log.count);
+    node_log(state, "SEND GET_STATE", "to=%d op=%d", sender_idx, state->wal.count);
     send_to_peer(state, sender_idx, &message.base);
 
     state->state_transfer_time = state->now;
@@ -266,14 +284,14 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
         }
     }
 
-    LogEntry log_entry = {
+    WALEntry wal_entry = {
         .oper        = request_message.oper,
         .votes       = 1 << self_idx(state),
         .view_number = state->view_number,
         .client_id   = request_message.client_id,
         .request_id  = request_message.request_id,
     };
-    if (log_append(&state->log, log_entry) < 0)
+    if (wal_append(&state->wal, wal_entry) < 0)
         return HR_OUT_OF_MEMORY;
 
     // We forwarded the message to all peers. As soon as
@@ -287,7 +305,7 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
         },
         .oper         = request_message.oper,
         .sender_idx   = self_idx(state),
-        .log_index    = state->log.count-1,
+        .log_index    = state->wal.count-1,
         .commit_index = state->commit_index,
         .view_number  = state->view_number,
         .client_id    = request_message.client_id,
@@ -296,7 +314,7 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
     {
         char oper_buf[128];
         meta_snprint_oper(oper_buf, sizeof(oper_buf), &request_message.oper);
-        node_log(state, "SEND PREPARE", "to=* idx=%d %s", state->log.count-1, oper_buf);
+        node_log(state, "SEND PREPARE", "to=* idx=%d %s", state->wal.count-1, oper_buf);
     }
     broadcast_to_peers(state, &prepare_message.base);
     return HR_OK;
@@ -333,13 +351,20 @@ static void reply_to_client(ServerState *state, ClientTableEntry *table_entry,
     byte_queue_write(output, &message, sizeof(message));
 }
 
+// Persist view_number, last_normal_view and commit_index to disk.
+static void persist_state(ServerState *state)
+{
+    set_view_and_commit(&state->vc, state->view_number,
+                        state->last_normal_view, state->commit_index);
+}
+
 static void advance_commit_index(ServerState *state, int target_index, bool send_replies)
 {
-    target_index = MIN(target_index, state->log.count);
+    target_index = MIN(target_index, state->wal.count);
 
     while (state->commit_index < target_index) {
 
-        LogEntry *entry = &state->log.entries[state->commit_index++];
+        WALEntry *entry = &state->wal.entries[state->commit_index++];
 
         MetaResult result = meta_store_update(&state->metastore, &entry->oper);
         {
@@ -375,6 +400,8 @@ static void advance_commit_index(ServerState *state, int target_index, bool send
         if (send_replies)
             reply_to_client(state, table_entry, entry->request_id, &entry->oper, result);
     }
+
+    persist_state(state);
 }
 
 // When the primary appends an entry to its log, it sends a
@@ -426,18 +453,21 @@ process_prepare_ok(ServerState *state, int conn_idx, ByteView msg)
 
     if (message.view_number > state->view_number) {
         state->view_number = message.view_number;
+        adopt_view(state);
+        persist_state(state);
         begin_state_transfer(state, message.sender_idx);
         return HR_OK;
     }
 
     assert(message.log_index > -1);
-    assert(message.log_index < state->log.count);
+    assert(message.log_index < state->wal.count);
 
     if (message.log_index < state->commit_index)
         return HR_OK; // Already processed
 
-    LogEntry *entry = &state->log.entries[message.log_index];
+    WALEntry *entry = &state->wal.entries[message.log_index];
     add_vote(&entry->votes, message.sender_idx);
+    wal_update_entry(&state->wal, message.log_index);
     if (reached_quorum(state, entry->votes)) {
         node_log(state, "QUORUM", "idx=%d %s/%s", message.log_index, entry->oper.bucket, entry->oper.key);
         advance_commit_index(state, message.log_index+1, true);
@@ -467,8 +497,8 @@ clear_view_change_fields(ServerState *state)
     state->view_change_apply_votes = 0;
     state->view_change_old_view = 0;
     state->view_change_commit = 0;
-    log_free(&state->view_change_log);
+    wal_free(&state->view_change_log);
-    log_init(&state->view_change_log);
+    wal_init(&state->view_change_log);
 }
 
 static HandlerResult
@@ -476,10 +506,24 @@ complete_view_change_and_become_primary(ServerState *state)
 {
     assert(state->commit_index <= state->view_change_commit);
 
-    log_move(&state->log, &state->view_change_log);
+    // Reset vote tracking for uncommitted entries before writing to
+    // disk. The entries inherited from DO_VIEW_CHANGE carry stale
+    // votes from a previous view. Resetting here ensures that if a
+    // crash occurs after wal_replace, the on-disk entries already
+    // have the correct leader vote bit set.
+    for (int i = state->view_change_commit; i < state->view_change_log.count; i++) {
+        state->view_change_log.entries[i].votes = 0;
+        add_vote(&state->view_change_log.entries[i].votes, self_idx(state));
+    }
+
+    if (wal_replace(&state->wal, state->view_change_log.entries, state->view_change_log.count) < 0)
+        return HR_IO_FAILURE;
+    wal_free(&state->view_change_log);
+    wal_init(&state->view_change_log);
 
     state->status = STATUS_NORMAL;
     state->last_normal_view = state->view_number;
+    persist_state(state);
     node_log(state, "STATUS NORMAL", "became primary (view change complete)");
 
     // Apply committed entries that haven't been executed yet.
@@ -490,31 +534,19 @@ complete_view_change_and_become_primary(ServerState *state)
     // requests, otherwise the state machine will be stale.
     advance_commit_index(state, state->view_change_commit, false);
 
-    // Reset vote tracking for uncommitted entries. The log
-    // entries inherited from DO_VIEW_CHANGE have stale
-    // votes from the previous view. The new leader starts
-    // with its own vote for each entry.
-    for (int i = state->commit_index; i < state->log.count; i++) {
-
-        LogEntry *entry = &state->log.entries[i];
-
-        entry->votes = 0;
-        add_vote(&entry->votes, self_idx(state));
-    }
-
     BeginViewMessage begin_view_message = {
         .base = {
             .version = MESSAGE_VERSION,
             .type    = MESSAGE_TYPE_BEGIN_VIEW,
-            .length  = sizeof(BeginViewMessage) + state->log.count * sizeof(LogEntry),
+            .length  = sizeof(BeginViewMessage) + state->wal.count * sizeof(WALEntry),
         },
         .view_number = state->view_number,
         .commit_index = state->commit_index,
-        .op_number = state->log.count,
+        .op_number = state->wal.count,
     };
     node_log(state, "SEND BEGIN_VIEW", "to=* view=%lu log=%d commit=%d",
-        state->view_number, state->log.count, state->commit_index);
+        state->view_number, state->wal.count, state->commit_index);
-    broadcast_to_peers_ex(state, &begin_view_message.base, state->log.entries, state->log.count * sizeof(LogEntry));
+    broadcast_to_peers_ex(state, &begin_view_message.base, state->wal.entries, state->wal.count * sizeof(WALEntry));
 
     clear_view_change_fields(state);
     return HR_OK;
@@ -552,15 +584,15 @@ process_do_view_change(ServerState *state, int conn_idx, ByteView msg)
 
             state->view_change_old_view = message.old_view_number;
 
-            LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(DoViewChangeMessage));
+            WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(DoViewChangeMessage));
 
             // Parse the variable-sized log from the message
-            int num_entries = (msg.len - sizeof(DoViewChangeMessage)) / sizeof(LogEntry);
+            int num_entries = (msg.len - sizeof(DoViewChangeMessage)) / sizeof(WALEntry);
             if (num_entries != message.op_number)
                 return HR_INVALID_MESSAGE; // Message size mismatch
 
-            log_free(&state->view_change_log);
+            wal_free(&state->view_change_log);
-            if (log_init_from_network(&state->view_change_log, entries, num_entries) < 0)
+            if (wal_init_from_network(&state->view_change_log, entries, num_entries) < 0)
                 return HR_OUT_OF_MEMORY;
         }
 
@@ -599,15 +631,15 @@ process_recovery(ServerState *state, int conn_idx, ByteView msg)
             .length  = sizeof(RecoveryResponseMessage),
         },
         .view_number  = state->view_number,
-        .op_number    = state->log.count-1, // TODO: What if the log is empty?
+        .op_number    = state->wal.count-1, // TODO: What if the log is empty?
         .nonce        = recovery_message.nonce,
         .commit_index = state->commit_index,
         .sender_idx   = self_idx(state),
     };
     if (is_primary(state)) {
-        recovery_response_message.base.length += state->log.count * sizeof(LogEntry);
+        recovery_response_message.base.length += state->wal.count * sizeof(WALEntry);
         send_to_peer_ex(state, recovery_message.sender_idx, &recovery_response_message.base,
-            state->log.entries, state->log.count * sizeof(LogEntry));
+            state->wal.entries, state->wal.count * sizeof(WALEntry));
     } else {
         send_to_peer(state, recovery_message.sender_idx, &recovery_response_message.base);
     }
@@ -627,7 +659,7 @@ perform_log_transfer_for_view_change(ServerState *state)
         state->view_change_commit = state->commit_index;
 
         // TODO: This should use copy-on-write
-        if (log_init_from_network(&state->view_change_log, state->log.entries, state->log.count) < 0)
+        if (wal_init_from_network(&state->view_change_log, state->wal.entries, state->wal.count) < 0)
             return HR_OUT_OF_MEMORY;
 
     } else {
@@ -635,18 +667,18 @@ perform_log_transfer_for_view_change(ServerState *state)
             .base = {
                 .version = MESSAGE_VERSION,
                 .type    = MESSAGE_TYPE_DO_VIEW_CHANGE,
-                .length  = sizeof(DoViewChangeMessage) + state->log.count * sizeof(LogEntry),
+                .length  = sizeof(DoViewChangeMessage) + state->wal.count * sizeof(WALEntry),
             },
             .view_number = state->view_number,
             .old_view_number = state->last_normal_view,
-            .op_number = state->log.count,
+            .op_number = state->wal.count,
             .commit_index = state->commit_index,
             .sender_idx = self_idx(state),
         };
-        send_to_peer_ex(state, primary_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry));
+        send_to_peer_ex(state, primary_idx(state), &do_view_change_message.base, state->wal.entries, state->wal.count * sizeof(WALEntry));
         node_log(state, "SEND DO_VIEW_CHANGE", "to=%d view=%lu old_view=%lu log=%d commit=%d",
             primary_idx(state), state->view_number, state->last_normal_view,
-            state->log.count, state->commit_index);
+            state->wal.count, state->commit_index);
     }
 
     // Clear the future array since we're changing views
@@ -699,6 +731,7 @@ process_begin_view_change(ServerState *state, int conn_idx, ByteView msg)
 
         clear_view_change_fields(state);
         state->view_number = message.view_number;
+        persist_state(state);
         state->heartbeat = state->now;
         state->status = STATUS_CHANGE_VIEW;
         node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number);
@@ -734,26 +767,31 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
     if (message.view_number < state->view_number)
         return HR_OK;
 
-    state->view_number = message.view_number;
+    int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(WALEntry);
-
-    state->status = STATUS_NORMAL;
-    state->last_normal_view = state->view_number;
-    node_log(state, "STATUS NORMAL", "new view=%lu (follower)", state->view_number);
-
-    int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(LogEntry);
     assert(num_entries >= state->commit_index);
 
-    // Replace the local log with the authoritative log from the primary
+    // Replace the local WAL with the authoritative log from the primary.
-    log_free(&state->log);
+    // This MUST complete before persist_state so that the on-disk log is
-    if (log_init_from_network(&state->log, msg.ptr + sizeof(BeginViewMessage), num_entries) < 0)
+    // always at least as recent as the persisted view/commit metadata.
-        return HR_OUT_OF_MEMORY;
+    // Otherwise a crash between persist_state and wal_replace would leave
+    // the stale log on disk with a view_number that references the new
+    // view, causing the node to replay divergent entries on restart.
+    WALEntry *new_entries = (WALEntry *)(msg.ptr + sizeof(BeginViewMessage));
+    if (wal_replace(&state->wal, new_entries, num_entries) < 0)
+        return HR_IO_FAILURE;
+
+    state->view_number = message.view_number;
+    state->status = STATUS_NORMAL;
+    state->last_normal_view = state->view_number;
+    persist_state(state);
+    node_log(state, "STATUS NORMAL", "new view=%lu (follower)", state->view_number);
 
     state->num_future = 0;
     state->state_transfer_pending = false;
 
     // If there are non-committed operations in the log,
     // send a PREPAREOK to the new primary
-    if (state->log.count > message.commit_index) {
+    if (state->wal.count > message.commit_index) {
         PrepareOKMessage ok_msg = {
             .base = {
                 .version = MESSAGE_VERSION,
@@ -761,12 +799,12 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
                 .length  = sizeof(PrepareOKMessage),
             },
             .sender_idx = self_idx(state),
-            .log_index  = state->log.count - 1,
+            .log_index  = state->wal.count - 1,
             .view_number = state->view_number,
         };
         send_to_peer(state, primary_idx(state), &ok_msg.base);
-        node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1,
+        node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->wal.count - 1,
-            state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key);
+            state->wal.entries[state->wal.count - 1].oper.bucket, state->wal.entries[state->wal.count - 1].oper.key);
     }
 
     advance_commit_index(state, message.commit_index, false);
@@ -798,16 +836,16 @@ process_get_state(ServerState *state, int conn_idx, ByteView msg)
 
     // Compute the suffix of log entries the requester is missing
     int start = get_state_message.op_number;
-    if (start < 0 || start >= state->log.count)
+    if (start < 0 || start >= state->wal.count)
         return HR_OK; // Nothing to send
 
-    int num_entries = state->log.count - start;
+    int num_entries = state->wal.count - start;
 
     NewStateMessage new_state_message = {
         .base = {
             .version = MESSAGE_VERSION,
             .type    = MESSAGE_TYPE_NEW_STATE,
-            .length  = sizeof(NewStateMessage) + num_entries * sizeof(LogEntry),
+            .length  = sizeof(NewStateMessage) + num_entries * sizeof(WALEntry),
         },
         .view_number  = state->view_number,
         .op_number    = num_entries,
@@ -817,7 +855,7 @@ process_get_state(ServerState *state, int conn_idx, ByteView msg)
     node_log(state, "SEND NEW_STATE", "to=%d entries=%d commit=%d",
         get_state_message.sender_idx, num_entries, state->commit_index);
     send_to_peer_ex(state, get_state_message.sender_idx, &new_state_message.base,
-        state->log.entries + start, num_entries * sizeof(LogEntry));
+        state->wal.entries + start, num_entries * sizeof(WALEntry));
     return HR_OK;
 }
 
@@ -1040,23 +1078,29 @@ complete_recovery(ServerState *state)
     assert(state->commit_index <= state->recovery_commit);
 
     state->view_number = state->recovery_view;
-    log_move(&state->log, &state->recovery_log);
+
+    // Reset stale votes before writing to disk so that a crash
+    // after wal_replace leaves the correct leader vote on disk.
+    int new_primary = state->view_number % state->num_nodes;
+    if (new_primary == self_idx(state)) {
+        for (int i = state->recovery_commit; i < state->recovery_log.count; i++) {
+            state->recovery_log.entries[i].votes = 0;
+            add_vote(&state->recovery_log.entries[i].votes, self_idx(state));
+        }
+    }
+
+    if (wal_replace(&state->wal, state->recovery_log.entries, state->recovery_log.count) < 0)
+        return HR_IO_FAILURE;
+    wal_free(&state->recovery_log);
+    wal_init(&state->recovery_log);
     advance_commit_index(state, state->recovery_commit, false);
 
     state->status = STATUS_NORMAL;
     state->last_normal_view = state->view_number;
+    persist_state(state);
     node_log(state, "STATUS NORMAL", "recovery complete view=%lu commit=%d",
         state->view_number, state->commit_index);
 
-    // Reset stale votes
-    if (is_primary(state)) {
-        for (int i = state->commit_index; i < state->log.count; i++) {
-            LogEntry *entry = &state->log.entries[i];
-            entry->votes = 0;
-            add_vote(&entry->votes, self_idx(state));
-        }
-    }
-
     // Update heartbeat to avoid immediate view change timeout
     state->heartbeat = state->now;
     return HR_OK;
@@ -1124,13 +1168,13 @@ process_recovery_response(ServerState *state, ByteView msg)
 
     if (should_store_recovery_log(state, message)) {
 
-        LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(RecoveryResponseMessage));
+        WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(RecoveryResponseMessage));
         int   num_entries = message.op_number + 1;
 
-        assert(num_entries == (int) ((msg.len - sizeof(RecoveryResponseMessage)) / sizeof(LogEntry)));
+        assert(num_entries == (int) ((msg.len - sizeof(RecoveryResponseMessage)) / sizeof(WALEntry)));
 
-        log_free(&state->recovery_log);
+        wal_free(&state->recovery_log);
-        if (log_init_from_network(&state->recovery_log, entries, num_entries) < 0)
+        if (wal_init_from_network(&state->recovery_log, entries, num_entries) < 0)
             return HR_OUT_OF_MEMORY;
 
         state->recovery_log_view = message.view_number;
@@ -1151,20 +1195,20 @@ process_single_future_list_entry(ServerState *state)
 {
     // Look for an entry with the current log index
     int i = 0;
-    while (i < state->num_future && state->future[i].log_index != state->log.count)
+    while (i < state->num_future && state->future[i].log_index != state->wal.count)
         i++;
 
     if (i == state->num_future)
         return 0; // No entry
 
-    LogEntry entry = {
+    WALEntry entry = {
         .oper = state->future[i].oper,
         .votes = 0,
         .view_number = state->view_number,
         .client_id  = state->future[i].client_id,
         .request_id = state->future[i].request_id,
     };
-    if (log_append(&state->log, entry) < 0)
+    if (wal_append(&state->wal, entry) < 0)
         return -1;
 
     PrepareOKMessage message = {
@@ -1174,7 +1218,7 @@ process_single_future_list_entry(ServerState *state)
             .length  = sizeof(PrepareOKMessage),
         },
         .sender_idx  = self_idx(state),
-        .log_index   = state->log.count-1,
+        .log_index   = state->wal.count-1,
         .view_number = state->view_number,
     };
     send_to_peer(state, state->future[i].sender_idx, &message.base);
@@ -1185,7 +1229,7 @@ static void
 remove_old_future_list_entries(ServerState *state)
 {
     for (int i = 0; i < state->num_future; i++) {
-        if (state->future[i].log_index < state->log.count) {
+        if (state->future[i].log_index < state->wal.count) {
             state->future[i--] = state->future[--state->num_future];
         }
     }
@@ -1228,30 +1272,32 @@ process_prepare(ServerState *state, ByteView msg)
     //           itself up to date before processing the message
     if (message.view_number > state->view_number) {
         state->view_number = message.view_number;
+        adopt_view(state);
+        persist_state(state);
         if (state->num_future < FUTURE_LIMIT)
             state->future[state->num_future++] = message;
         begin_state_transfer(state, message.sender_idx);
         return HR_OK;
     }
 
-    if (message.log_index < state->log.count)
+    if (message.log_index < state->wal.count)
         return HR_OK; // Message refers to an old entry. Ignore.
 
-    if (message.log_index > state->log.count) {
+    if (message.log_index > state->wal.count) {
         if (state->num_future < FUTURE_LIMIT)
             state->future[state->num_future++] = message;
         begin_state_transfer(state, message.sender_idx);
         return HR_OK;
     }
 
-    LogEntry log_entry = {
+    WALEntry wal_entry = {
         .oper = message.oper,
         .votes = 0,
         .view_number = state->view_number,
         .client_id  = message.client_id,
         .request_id = message.request_id,
     };
-    if (log_append(&state->log, log_entry) < 0)
+    if (wal_append(&state->wal, wal_entry) < 0)
         return HR_OUT_OF_MEMORY;
 
     PrepareOKMessage ok_message = {
@@ -1261,12 +1307,12 @@ process_prepare(ServerState *state, ByteView msg)
             .length  = sizeof(PrepareOKMessage),
         },
         .sender_idx = self_idx(state),
-        .log_index  = state->log.count-1,
+        .log_index  = state->wal.count-1,
         .view_number = state->view_number,
     };
     send_to_peer(state, message.sender_idx, &ok_message.base);
     node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s",
-        message.sender_idx, state->log.count-1, message.oper.bucket, message.oper.key);
+        message.sender_idx, state->wal.count-1, message.oper.bucket, message.oper.key);
 
     process_future_list(state);
     advance_commit_index(state, message.commit_index, false);
@@ -1312,13 +1358,14 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
     memcpy(&new_state_message, msg.ptr, sizeof(new_state_message));
 
     node_log(state, "RECV NEW_STATE", "entries=%d commit=%d view=%lu",
-        new_state_message.op_number, new_state_message.commit_index, new_state_message.view_number);
+        new_state_message.op_number, new_state_message.commit_index,
+        new_state_message.view_number);
 
     // Ignore if we're in a different view
     if (new_state_message.view_number != state->view_number)
         return HR_OK;
 
-    int num_entries = (msg.len - sizeof(NewStateMessage)) / sizeof(LogEntry);
+    int num_entries = (msg.len - sizeof(NewStateMessage)) / sizeof(WALEntry);
     if (num_entries != new_state_message.op_number)
         return HR_INVALID_MESSAGE;
 
@@ -1328,22 +1375,22 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
     // Append received entries to our log.
     // The entries array is a suffix of the sender's log starting at
     // global position start_index. We skip entries we already have.
-    LogEntry *entries = (LogEntry *)((uint8_t *)msg.ptr + sizeof(NewStateMessage));
+    WALEntry *entries = (WALEntry *)((uint8_t *)msg.ptr + sizeof(NewStateMessage));
     int start_index = new_state_message.start_index;
     for (int i = 0; i < num_entries; i++) {
 
         int global_idx = start_index + i;
-        if (global_idx < state->log.count)
+        if (global_idx < state->wal.count)
             continue; // Already have this entry
 
-        LogEntry entry = {
+        WALEntry entry = {
             .oper = entries[i].oper,
             .votes = 0,
             .view_number = state->view_number,
             .client_id = entries[i].client_id,
             .request_id = entries[i].request_id,
         };
-        if (log_append(&state->log, entry) < 0)
+        if (wal_append(&state->wal, entry) < 0)
             return HR_OUT_OF_MEMORY;
 
         // Send PREPARE_OK for each appended entry
@@ -1354,12 +1401,12 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
                 .length  = sizeof(PrepareOKMessage),
             },
             .sender_idx  = self_idx(state),
-            .log_index   = state->log.count - 1,
+            .log_index   = state->wal.count - 1,
             .view_number = state->view_number,
         };
         send_to_peer(state, primary_idx(state), &prepare_ok_message.base);
-        node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1,
+        node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->wal.count - 1,
-            state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key);
+            state->wal.entries[state->wal.count - 1].oper.bucket, state->wal.entries[state->wal.count - 1].oper.key);
     }
 
     process_future_list(state);
@@ -1652,10 +1699,7 @@ int server_init(void *state_, int argc, char **argv,
 
     Time deadline = INVALID_TIME;
 
-    state->view_number = 0;
-    state->last_normal_view = 0;
     state->heartbeat = now;
-    state->commit_index = 0;
     state->num_future = 0;
     state->state_transfer_pending = false;
     state->state_transfer_time = 0;
@@ -1665,41 +1709,80 @@ int server_init(void *state_, int argc, char **argv,
     state->view_change_apply_votes = 0;
     state->view_change_old_view = 0;
     state->view_change_commit = 0;
-    log_init(&state->view_change_log);
+    wal_init(&state->view_change_log);
 
     // Recovery state
     state->recovery_votes = 0;
     state->recovery_commit = 0;
     state->recovery_view = 0;
     state->recovery_log_view = 0;
-    log_init(&state->recovery_log);
+    wal_init(&state->recovery_log);
 
-    // Detect whether this is a restart after a crash by checking for a
+    // Load persistent view_number, last_normal_view and commit_index.
-    // boot marker file on disk. The disk persists across crashes, so if
+    if (view_and_commit_init(&state->vc, S("vsr.state")) < 0) {
-    // the marker exists, this node previously ran and crashed. In that
+        fprintf(stderr, "Node :: Couldn't open state file\n");
-    // case, enter recovery mode to learn the current view from peers
+        return -1;
-    // before participating in the protocol.
+    }
-    //
+    state->view_number = state->vc.view_number;
-    // We use open() directly (without O_CREAT) instead of file_exists()
+    state->last_normal_view = state->vc.last_normal_view;
-    // because access() is not available in the simulation environment.
+    state->commit_index = 0; // Will be advanced below after replaying the log
-    int marker_fd = open("vsr_boot_marker", O_RDONLY, 0);
+
-    bool previously_crashed = (marker_fd >= 0);
+    // Load the WAL from disk. Three cases:
-    if (previously_crashed)
+    //   1. Empty log (first start) -> enter normal mode
-        close(marker_fd);
+    //   2. Corrupt log (truncated) -> enter recovery mode
-    if (previously_crashed) {
+    //   3. Clean log              -> replay up to commit_index, enter normal mode
+    bool was_truncated = false;
+    if (wal_init_from_file(&state->wal, S("vsr.log"), &was_truncated) < 0) {
+        fprintf(stderr, "Node :: Couldn't open WAL file\n");
+        view_and_commit_free(&state->vc);
+        return -1;
+    }
+
+    bool enter_recovery;
+    if (state->wal.count == 0 && !was_truncated) {
+        // First start: no previous log on disk
+        enter_recovery = false;
+    } else if (was_truncated) {
+        // Log was corrupt — must enter recovery to get a valid log
+        // from peers
+        enter_recovery = true;
+    } else {
+        // Clean restart with a valid log — replay committed entries
+        // and resume normal operation
+        enter_recovery = false;
+    }
+
+    if (enter_recovery) {
         state->status = STATUS_RECOVERY;
         state->recovery_nonce = now;
         state->recovery_time = now;
     } else {
         state->status = STATUS_NORMAL;
+        state->last_normal_view = state->view_number;
     }
-    log_init(&state->log); // Initialize early so node_log can read log.count
+    node_log(state, "INIT", "nodes=%d%s", state->num_nodes, enter_recovery ? " (recovering)" : "");
-    node_log(state, "INIT", "nodes=%d%s", state->num_nodes, previously_crashed ? " (recovering)" : "");
 
     client_table_init(&state->client_table);
     state->next_client_tag = NODE_LIMIT; // Make sure they don't overlap with node indices
 
     meta_store_init(&state->metastore);
+
+    // On a clean restart (non-corrupt log present), replay committed
+    // entries into the state machine before accepting new traffic.
+    if (!enter_recovery && state->wal.count > 0) {
+        int replay_target = state->vc.commit_index;
+        if (replay_target > state->wal.count)
+            replay_target = state->wal.count;
+        advance_commit_index(state, replay_target, false);
+
+        // If this node is the primary, ensure the leader's own vote
+        // bit is set for uncommitted entries. A crash during
+        // wal_replace (before the atomic rename) can leave on-disk
+        // entries with stale votes from a previous role.
+        adopt_view(state);
+        node_log(state, "REPLAY", "applied %d entries", state->commit_index);
+    }
+
     if (chunk_store_init(&state->chunk_store, chunks_path) < 0) {
         fprintf(stderr, "Node :: Couldn't initialize chunk store at '%s'\n", chunks_path);
         return -1;
@@ -1717,16 +1800,7 @@ int server_init(void *state_, int argc, char **argv,
         return -1;
     }
 
-    // Write the boot marker to disk so that future restarts can detect
+    if (enter_recovery) {
-    // a previous crash. This must happen after TCP init so that the
-    // marker is only written if the node successfully started.
-    if (!previously_crashed) {
-        int fd = open("vsr_boot_marker", O_WRONLY | O_CREAT, 0644);
-        if (fd >= 0)
-            close(fd);
-    }
-
-    if (previously_crashed) {
         node_log(state, "STATUS RECOVERY", "nonce=%lu (crash detected)", state->recovery_nonce);
 
         // Broadcast RECOVERY to all peers to learn the current view
@@ -1793,7 +1867,7 @@ int server_tick(void *state_, void **ctxs,
                 tcp_close(&state->tcp, conn_idx);
                 break;
             }
-            if (hret == HR_OUT_OF_MEMORY)
+            if (hret == HR_OUT_OF_MEMORY || hret == HR_IO_FAILURE)
                 return -1;
             assert(hret == HR_OK);
 
@@ -1850,6 +1924,7 @@ int server_tick(void *state_, void **ctxs,
             add_vote(&state->view_change_begin_votes, self_idx(state));
 
             state->view_number++;
+            persist_state(state);
             state->heartbeat = state->now;
 
             BeginViewChangeMessage begin_view_change_message = {
@@ -1903,6 +1978,7 @@ int server_tick(void *state_, void **ctxs,
                 add_vote(&state->view_change_begin_votes, self_idx(state));
 
                 state->view_number++;
+                persist_state(state);
                 state->status = STATUS_CHANGE_VIEW;
                 state->heartbeat = state->now;
 
@@ -1932,7 +2008,7 @@ int server_tick(void *state_, void **ctxs,
 
         Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL;
         if (st_deadline <= state->now) {
-            node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->log.count);
+            node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->wal.count);
 
             GetStateMessage get_state_message = {
                 .base = {
@@ -1941,11 +2017,11 @@ int server_tick(void *state_, void **ctxs,
                     .length  = sizeof(GetStateMessage),
                 },
                 .view_number = state->view_number,
-                .op_number   = state->log.count,
+                .op_number   = state->wal.count,
                 .sender_idx  = self_idx(state),
             };
             send_to_peer(state, primary_idx(state), &get_state_message.base);
-            node_log(state, "SEND GET_STATE", "to=%d op=%d", primary_idx(state), state->log.count);
+            node_log(state, "SEND GET_STATE", "to=%d op=%d", primary_idx(state), state->wal.count);
 
             state->state_transfer_time = state->now;
 
@@ -1967,9 +2043,10 @@ int server_free(void *state_)
 
     node_log_simple(state, "CRASHED");
 
-    log_free(&state->log);
+    wal_free(&state->wal);
-    log_free(&state->recovery_log);
+    view_and_commit_free(&state->vc);
-    log_free(&state->view_change_log);
+    wal_free(&state->recovery_log);
+    wal_free(&state->view_change_log);
     tcp_context_free(&state->tcp);
     client_table_free(&state->client_table);
     meta_store_free(&state->metastore);

src/server.h

@@ -4,7 +4,7 @@
 #include "tcp.h"
 #include "basic.h"
 #include "message.h"
-#include "log.h"
+#include "wal.h"
 #include "config.h"
 #include "metadata.h"
 #include "chunk_store.h"
@@ -99,15 +99,15 @@ typedef struct {
     int op_number;              // Number of entries in the log
     int commit_index;
     int sender_idx;
-    // Followed by: LogEntry log[op_number]
+    // Followed by: WALEntry log[op_number]
 } DoViewChangeMessage;
 
 typedef struct {
     MessageHeader base;
     uint64_t view_number;
     int commit_index;
-    int op_number;  // Number of log entries that follow
+    int op_number;  // Number of WAL entries that follow
-    // Followed by: LogEntry log[op_number]
+    // Followed by: WALEntry log[op_number]
 } BeginViewMessage;
 
 typedef struct {
@@ -138,7 +138,7 @@ typedef struct {
     int op_number;      // Number of log entries that follow
     int commit_index;
     int start_index;    // Global log index of the first entry in the suffix
-    // Followed by: LogEntry log[op_number]
+    // Followed by: WALEntry log[op_number]
 } NewStateMessage;
 
 typedef struct {
@@ -228,7 +228,7 @@ typedef struct {
     uint32_t recovery_votes;
     uint64_t recovery_nonce;
     uint64_t recovery_view;
-    Log      recovery_log;
+    WAL      recovery_log;
     uint64_t recovery_log_view;
     Time     recovery_time;
     int      recovery_commit;
@@ -238,7 +238,7 @@ typedef struct {
 
     uint32_t view_change_begin_votes;
     uint32_t view_change_apply_votes;
-    Log      view_change_log; // Best log seen
+    WAL      view_change_log; // Best log seen
     uint64_t view_change_old_view;  // Best old_view_number seen in DoViewChange
     int      view_change_commit;    // Best commit_index seen
 
@@ -255,7 +255,8 @@ typedef struct {
     bool state_transfer_pending;
     Time state_transfer_time;
 
-    Log log;
+    WAL wal;
+    ViewAndCommit vc;
 
     Time heartbeat;
 
@@ -282,6 +283,7 @@ typedef struct {
     int last_min_commit;
     int last_max_commit;
     Status prev_status[NODE_LIMIT];
+    bool   prev_alive[NODE_LIMIT];
 
     // External shadow log of committed operations (unbounded, dynamically allocated)
     MetaOper *shadow_log;

src/wal.c

@@ -0,0 +1,422 @@
+#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
+#define QUAKEY_ENABLE_MOCKS
+#endif
+
+#include <quakey.h>
+#include <assert.h>
+#include <stddef.h>
+#include <string.h>
+
+#include "wal.h"
+
+// FNV-1a checksum over all WALEntry fields except the checksum itself.
+static uint32_t wal_entry_checksum(WALEntry *entry)
+{
+    uint32_t h = 2166136261u;
+    const unsigned char *p = (const unsigned char *)entry;
+    size_t len = offsetof(WALEntry, checksum);
+    for (size_t i = 0; i < len; i++) {
+        h ^= p[i];
+        h *= 16777619u;
+    }
+    return h;
+}
+
+static void wal_entry_set_checksum(WALEntry *entry)
+{
+    entry->checksum = wal_entry_checksum(entry);
+}
+
+static bool wal_is_file_backed(WAL *wal)
+{
+    return wal->handle.data != 0;
+}
+
+static int wal_grow(WAL *wal)
+{
+    int n = 2 * wal->capacity;
+    if (n < 8) n = 8;
+    WALEntry *p = realloc(wal->entries, n * sizeof(WALEntry));
+    if (p == NULL)
+        return -1;
+    wal->entries = p;
+    wal->capacity = n;
+    return 0;
+}
+
+void wal_init(WAL *wal)
+{
+    wal->count = 0;
+    wal->capacity = 0;
+    wal->entries = NULL;
+    wal->handle = (Handle) { 0 };
+}
+
+int wal_init_from_file(WAL *wal, string file, bool *was_truncated)
+{
+    if (was_truncated)
+        *was_truncated = false;
+
+    Handle handle;
+    if (file_open(file, &handle) < 0)
+        return -1;
+
+    size_t size;
+    if (file_size(handle, &size) < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    // Discard any partial trailing entry (crash during write).
+    int raw_count = size / sizeof(WALEntry);
+    size_t valid_size = raw_count * sizeof(WALEntry);
+    if (valid_size < size)
+        file_truncate(handle, valid_size);
+
+    WALEntry *entries = malloc(raw_count * sizeof(WALEntry));
+    if (entries == NULL && raw_count > 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    if (file_set_offset(handle, 0) < 0) {
+        file_close(handle);
+        free(entries);
+        return -1;
+    }
+
+    if (raw_count > 0 && file_read_exact(handle, (char *)entries, raw_count * sizeof(WALEntry)) < 0) {
+        file_close(handle);
+        free(entries);
+        return -1;
+    }
+
+    // Verify checksums: truncate at the first corrupted entry.
+    int count = raw_count;
+    for (int i = 0; i < raw_count; i++) {
+        if (entries[i].checksum != wal_entry_checksum(&entries[i])) {
+            count = i;
+            file_truncate(handle, count * sizeof(WALEntry));
+            if (was_truncated)
+                *was_truncated = true;
+            break;
+        }
+    }
+
+    // Position file offset at end for future appends.
+    if (file_set_offset(handle, count * sizeof(WALEntry)) < 0) {
+        file_close(handle);
+        free(entries);
+        return -1;
+    }
+
+    wal->count = count;
+    wal->capacity = count;
+    wal->entries = entries;
+    wal->handle = handle;
+    return 0;
+}
+
+int wal_init_from_network(WAL *wal, void *src, int num)
+{
+    wal->count = num;
+    wal->capacity = num;
+    wal->entries = NULL;
+    if (num > 0) {
+        wal->entries = malloc(num * sizeof(WALEntry));
+        if (wal->entries == NULL)
+            return -1;
+        memcpy(wal->entries, src, num * sizeof(WALEntry));
+    }
+    wal->handle = (Handle) { 0 };
+    return 0;
+}
+
+void wal_free(WAL *wal)
+{
+    free(wal->entries);
+    if (wal_is_file_backed(wal))
+        file_close(wal->handle);
+}
+
+void wal_move(WAL *dst, WAL *src)
+{
+    free(dst->entries);
+    dst->count = src->count;
+    dst->capacity = src->capacity;
+    dst->entries = src->entries;
+    // Do NOT touch dst->handle — caller manages file backing.
+    wal_init(src);
+}
+
+int wal_append(WAL *wal, WALEntry entry)
+{
+    if (wal->count == wal->capacity) {
+        if (wal_grow(wal) < 0)
+            return -1;
+    }
+
+    if (wal_is_file_backed(wal)) {
+        wal_entry_set_checksum(&entry);
+
+        if (file_write_exact(wal->handle, (char *)&entry, sizeof(entry)) < 0) {
+            // Partial write may have advanced file offset. Truncate and
+            // rewind so the file stays consistent with in-memory count.
+            file_truncate(wal->handle, wal->count * sizeof(WALEntry));
+            file_set_offset(wal->handle, wal->count * sizeof(WALEntry));
+            return -1;
+        }
+
+        if (file_sync(wal->handle) < 0) {
+            file_truncate(wal->handle, wal->count * sizeof(WALEntry));
+            file_set_offset(wal->handle, wal->count * sizeof(WALEntry));
+            return -1;
+        }
+    }
+
+    wal->entries[wal->count++] = entry;
+    return 0;
+}
+
+int wal_update_entry(WAL *wal, int idx)
+{
+    assert(idx >= 0 && idx < wal->count);
+
+    if (wal_is_file_backed(wal)) {
+        WALEntry entry = wal->entries[idx];
+        wal_entry_set_checksum(&entry);
+
+        size_t offset = idx * sizeof(WALEntry);
+        size_t end_offset = wal->count * sizeof(WALEntry);
+
+        if (file_set_offset(wal->handle, offset) < 0)
+            return -1;
+        if (file_write_exact(wal->handle, (char *)&entry, sizeof(entry)) < 0)
+            return -1;
+        if (file_sync(wal->handle) < 0)
+            return -1;
+
+        // Restore file offset to end for future appends.
+        if (file_set_offset(wal->handle, end_offset) < 0)
+            return -1;
+    }
+    return 0;
+}
+
+int wal_truncate(WAL *wal, int new_count)
+{
+    assert(new_count <= wal->count);
+    if (wal->count == new_count)
+        return 0;
+
+    if (wal_is_file_backed(wal)) {
+        if (file_truncate(wal->handle, new_count * sizeof(WALEntry)) < 0)
+            return -1;
+        if (file_set_offset(wal->handle, new_count * sizeof(WALEntry)) < 0)
+            return -1;
+    }
+
+    wal->count = new_count;
+    return 0;
+}
+
+int wal_replace(WAL *wal, WALEntry *entries, int count)
+{
+    assert(wal_is_file_backed(wal));
+
+    string tmp_path = S("tmp.log");
+    string wal_path = S("vsr.log");
+
+    // Open tmp.log (file_open creates if not exists, opens if exists).
+    Handle tmp;
+    if (file_open(tmp_path, &tmp) < 0)
+        return -1;
+
+    // Truncate in case tmp.log already exists from a previous crash.
+    if (file_truncate(tmp, 0) < 0) {
+        file_close(tmp);
+        return -1;
+    }
+    if (file_set_offset(tmp, 0) < 0) {
+        file_close(tmp);
+        return -1;
+    }
+
+    // Write all entries to tmp.log.
+    for (int i = 0; i < count; i++) {
+        wal_entry_set_checksum(&entries[i]);
+        if (file_write_exact(tmp, (char *)&entries[i], sizeof(entries[i])) < 0) {
+            file_close(tmp);
+            return -1;
+        }
+    }
+
+    // Fsync tmp.log to ensure all data is on disk.
+    if (file_sync(tmp) < 0) {
+        file_close(tmp);
+        return -1;
+    }
+    file_close(tmp);
+
+    // Close the current WAL handle before rename.
+    file_close(wal->handle);
+    wal->handle = (Handle) { 0 };
+
+    // Atomically replace the WAL file.
+    if (rename_file_or_dir(tmp_path, wal_path) < 0)
+        return -1;
+
+    // TODO: fsync on parent directory
+
+    // Reopen the WAL file and seek to end for future appends.
+    Handle new_handle;
+    if (file_open(wal_path, &new_handle) < 0)
+        return -1;
+    if (file_set_offset(new_handle, count * sizeof(WALEntry)) < 0) {
+        file_close(new_handle);
+        return -1;
+    }
+    wal->handle = new_handle;
+
+    // Update in-memory state.
+    free(wal->entries);
+    wal->entries = NULL;
+    wal->count = 0;
+    wal->capacity = 0;
+
+    if (count > 0) {
+        wal->entries = malloc(count * sizeof(WALEntry));
+        if (wal->entries == NULL)
+            return -1;
+        memcpy(wal->entries, entries, count * sizeof(WALEntry));
+    }
+    wal->count = count;
+    wal->capacity = count;
+
+    return 0;
+}
+
+int wal_entry_count(WAL *wal)
+{
+    return wal->count;
+}
+
+WALEntry *wal_peek_entry(WAL *wal, int idx)
+{
+    assert(idx >= 0);
+    assert(idx < wal->count);
+    return &wal->entries[idx];
+}
+
+///////////////////////////////////////////////////////////////////////////////
+// ViewAndCommit — persistent view_number, last_normal_view and commit_index
+///////////////////////////////////////////////////////////////////////////////
+
+// On-disk layout (24 bytes):
+//   [view_number: 8] [last_normal_view: 8] [commit_index: 4] [checksum: 4]
+typedef struct {
+    uint64_t view_number;
+    uint64_t last_normal_view;
+    int      commit_index;
+    uint32_t checksum;
+} ViewAndCommitDisk;
+
+static uint32_t vc_checksum(uint64_t view_number,
+    uint64_t last_normal_view, int commit_index)
+{
+    uint32_t h = 2166136261u;
+
+    const unsigned char *p = (const unsigned char *)&view_number;
+    for (int i = 0; i < (int)sizeof(view_number); i++) {
+        h ^= p[i];
+        h *= 16777619u;
+    }
+
+    p = (const unsigned char *)&last_normal_view;
+    for (int i = 0; i < (int)sizeof(last_normal_view); i++) {
+        h ^= p[i];
+        h *= 16777619u;
+    }
+
+    p = (const unsigned char *)&commit_index;
+    for (int i = 0; i < (int)sizeof(commit_index); i++) {
+        h ^= p[i];
+        h *= 16777619u;
+    }
+
+    return h;
+}
+
+int view_and_commit_init(ViewAndCommit *vc, string file)
+{
+    Handle handle;
+    if (file_open(file, &handle) < 0)
+        return -1;
+
+    vc->handle = handle;
+    vc->view_number = 0;
+    vc->last_normal_view = 0;
+    vc->commit_index = 0;
+
+    size_t size;
+    if (file_size(handle, &size) < 0) {
+        file_close(handle);
+        return -1;
+    }
+
+    if (size >= sizeof(ViewAndCommitDisk)) {
+
+        if (file_set_offset(handle, 0) < 0) {
+            file_close(handle);
+            return -1;
+        }
+
+        ViewAndCommitDisk disk;
+        if (file_read_exact(handle, (char *)&disk, sizeof(disk)) < 0) {
+            file_close(handle);
+            return -1;
+        }
+
+        uint32_t expected_checksum = vc_checksum(disk.view_number,
+            disk.last_normal_view, disk.commit_index);
+
+        if (disk.checksum == expected_checksum) {
+            vc->view_number = disk.view_number;
+            vc->last_normal_view = disk.last_normal_view;
+            vc->commit_index = disk.commit_index;
+        }
+
+        // If checksum doesn't match, start from defaults (0, 0, 0)
+    }
+
+    return 0;
+}
+
+void view_and_commit_free(ViewAndCommit *vc)
+{
+    if (vc->handle.data != 0)
+        file_close(vc->handle);
+}
+
+int set_view_and_commit(ViewAndCommit *vc, uint64_t view_number,
+    uint64_t last_normal_view, int commit_index)
+{
+    ViewAndCommitDisk disk = {
+        .view_number = view_number,
+        .last_normal_view = last_normal_view,
+        .commit_index = commit_index,
+    };
+    disk.checksum = vc_checksum(view_number, last_normal_view, commit_index);
+
+    if (file_set_offset(vc->handle, 0) < 0)
+        return -1;
+    if (file_write_exact(vc->handle, (char *)&disk, sizeof(disk)) < 0)
+        return -1;
+    if (file_sync(vc->handle) < 0)
+        return -1;
+
+    vc->view_number = view_number;
+    vc->last_normal_view = last_normal_view;
+    vc->commit_index = commit_index;
+    return 0;
+}

src/wal.h

@@ -0,0 +1,77 @@
+#ifndef WAL_INCLUDED
+#define WAL_INCLUDED
+
+#include "metadata.h"
+#include "file_system.h"
+#include "config.h"
+
+typedef struct {
+    MetaOper oper;
+    uint32_t votes;
+    int      view_number;
+    uint64_t client_id;
+    uint64_t request_id;
+    uint32_t checksum;    // FNV-1a over all preceding fields
+} WALEntry;
+
+_Static_assert(NODE_LIMIT <= 32, "");
+
+typedef struct {
+    int       count;
+    int       capacity;
+    WALEntry *entries;
+    Handle    handle;     // file handle to the WAL (0 = memory-only)
+} WAL;
+
+// Initialize an empty, memory-only WAL (no file backing).
+void wal_init(WAL *wal);
+
+// Initialize a WAL from an on-disk file. Recovers valid entries,
+// discards partial/corrupted trailing entries. If was_truncated is
+// non-NULL, *was_truncated is set to true when entries were discarded
+// due to corruption (not just a partial trailing write).
+int  wal_init_from_file(WAL *wal, string file, bool *was_truncated);
+
+// Initialize a WAL from network data (memory-only, no file).
+int  wal_init_from_network(WAL *wal, void *src, int num);
+
+void wal_free(WAL *wal);
+
+// Move ownership from src to dst. If dst is file-backed, the file
+// is NOT affected; use wal_replace for atomic file replacement.
+void wal_move(WAL *dst, WAL *src);
+
+// Append an entry. If the WAL is file-backed, writes to disk and
+// fsyncs before updating the in-memory buffer.
+int  wal_append(WAL *wal, WALEntry entry);
+
+// Write a modified in-memory entry back to disk (recomputes checksum).
+int  wal_update_entry(WAL *wal, int idx);
+
+// Truncate the WAL to new_count entries.
+int  wal_truncate(WAL *wal, int new_count);
+
+// Atomically replace the WAL file contents with the given entries.
+// Writes to a temporary file, fsyncs, then renames over the WAL.
+// Updates the in-memory buffer and reopens the file handle.
+int  wal_replace(WAL *wal, WALEntry *entries, int count);
+
+int       wal_entry_count(WAL *wal);
+WALEntry *wal_peek_entry(WAL *wal, int idx);
+
+// Persistent view_number, last_normal_view and commit_index,
+// analogous to raft's TermAndVote. Backed by a small file
+// ("vsr.state") with a checksum for integrity.
+typedef struct {
+    uint64_t view_number;
+    uint64_t last_normal_view;
+    int      commit_index;
+    Handle   handle;
+} ViewAndCommit;
+
+int  view_and_commit_init(ViewAndCommit *vc, string file);
+void view_and_commit_free(ViewAndCommit *vc);
+int  set_view_and_commit(ViewAndCommit *vc, uint64_t view_number,
+                         uint64_t last_normal_view, int commit_index);
+
+#endif // WAL_INCLUDED