From 34af6063b6223fb8dac16a0914fcca1bf511427c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 22 Feb 2026 17:19:00 +0000 Subject: [PATCH] Fix WAL correctness: persist votes, fix crash recovery and view changes 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 --- src/invariant_checker.c | 11 +++- src/server.c | 119 ++++++++++++++++++++++++---------------- src/server.h | 1 + src/wal.c | 106 +++++++++++++++++------------------ src/wal.h | 15 ++--- 5 files changed, 138 insertions(+), 114 deletions(-) diff --git a/src/invariant_checker.c b/src/invariant_checker.c index 4f63f41..8e69d3b 100644 --- a/src/invariant_checker.c +++ b/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; } diff --git a/src/server.c b/src/server.c index 88f89dc..b9b9b30 100644 --- a/src/server.c +++ b/src/server.c @@ -133,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) @@ -436,6 +453,7 @@ 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; @@ -449,6 +467,7 @@ process_prepare_ok(ServerState *state, int conn_idx, ByteView msg) 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); @@ -487,6 +506,16 @@ complete_view_change_and_become_primary(ServerState *state) { assert(state->commit_index <= state->view_change_commit); + // 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); @@ -505,18 +534,6 @@ 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->wal.count; i++) { - - WALEntry *entry = &state->wal.entries[i]; - - entry->votes = 0; - add_vote(&entry->votes, self_idx(state)); - } - BeginViewMessage begin_view_message = { .base = { .version = MESSAGE_VERSION, @@ -750,21 +767,25 @@ 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); + assert(num_entries >= state->commit_index); + // Replace the local WAL with the authoritative log from the primary. + // This MUST complete before persist_state so that the on-disk log is + // always at least as recent as the persisted view/commit metadata. + // 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); - int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(WALEntry); - assert(num_entries >= state->commit_index); - - // Replace the local WAL with the authoritative log from the primary - WALEntry *new_entries = (WALEntry *)(msg.ptr + sizeof(BeginViewMessage)); - if (wal_replace(&state->wal, new_entries, num_entries) < 0) - return HR_IO_FAILURE; - state->num_future = 0; state->state_transfer_pending = false; @@ -1057,6 +1078,17 @@ complete_recovery(ServerState *state) assert(state->commit_index <= state->recovery_commit); state->view_number = state->recovery_view; + + // 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); @@ -1069,15 +1101,6 @@ complete_recovery(ServerState *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->wal.count; i++) { - WALEntry *entry = &state->wal.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; @@ -1249,6 +1272,7 @@ 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; @@ -1702,19 +1726,10 @@ int server_init(void *state_, int argc, char **argv, state->last_normal_view = state->vc.last_normal_view; state->commit_index = 0; // Will be advanced below after replaying the log - // Load the WAL from disk. Two cases: - // 1. Empty log, no truncation (first start) -> enter normal mode - // 2. Any restart (clean or corrupt log) -> enter recovery mode - // - // VR-Revisited Section 4.3: "When a node recovers after a crash it - // cannot participate in request processing and view changes until it - // has a state at least as recent as what it had before it crashed." - // - // A clean restart without recovery is unsafe: while the node was - // down, a view change may have replaced uncommitted log entries. - // Entering NORMAL with the stale log would cause the node to hold - // divergent entries that could later be committed via state transfer, - // violating committed prefix agreement. + // Load the WAL from disk. Three cases: + // 1. Empty log (first start) -> enter normal mode + // 2. Corrupt log (truncated) -> enter recovery mode + // 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"); @@ -1726,11 +1741,14 @@ int server_init(void *state_, int argc, char **argv, if (state->wal.count == 0 && !was_truncated) { // First start: no previous log on disk enter_recovery = false; - } else { - // Restart (clean or corrupt log). The node must enter recovery - // to learn the current view and obtain an authoritative log - // from the primary. + } 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) { @@ -1739,6 +1757,7 @@ int server_init(void *state_, int argc, char **argv, state->recovery_time = now; } else { state->status = STATUS_NORMAL; + state->last_normal_view = state->view_number; } node_log(state, "INIT", "nodes=%d%s", state->num_nodes, enter_recovery ? " (recovering)" : ""); @@ -1754,6 +1773,12 @@ int server_init(void *state_, int argc, char **argv, 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); } diff --git a/src/server.h b/src/server.h index 380d392..eee5a31 100644 --- a/src/server.h +++ b/src/server.h @@ -283,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; diff --git a/src/wal.c b/src/wal.c index 4542f50..91fa173 100644 --- a/src/wal.c +++ b/src/wal.c @@ -9,12 +9,12 @@ #include "wal.h" -// FNV-1a checksum over all WALEntryDisk fields except the checksum itself. -static uint32_t wal_entry_checksum(WALEntryDisk *entry) +// 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(WALEntryDisk, checksum); + size_t len = offsetof(WALEntry, checksum); for (size_t i = 0; i < len; i++) { h ^= p[i]; h *= 16777619u; @@ -22,27 +22,9 @@ static uint32_t wal_entry_checksum(WALEntryDisk *entry) return h; } -static WALEntryDisk wal_entry_to_disk(WALEntry *entry) +static void wal_entry_set_checksum(WALEntry *entry) { - WALEntryDisk disk = { - .oper = entry->oper, - .view_number = entry->view_number, - .client_id = entry->client_id, - .request_id = entry->request_id, - }; - disk.checksum = wal_entry_checksum(&disk); - return disk; -} - -static WALEntry wal_entry_from_disk(WALEntryDisk *disk) -{ - return (WALEntry) { - .oper = disk->oper, - .votes = 0, - .view_number = disk->view_number, - .client_id = disk->client_id, - .request_id = disk->request_id, - }; + entry->checksum = wal_entry_checksum(entry); } static bool wal_is_file_backed(WAL *wal) @@ -86,54 +68,43 @@ int wal_init_from_file(WAL *wal, string file, bool *was_truncated) } // Discard any partial trailing entry (crash during write). - int raw_count = size / sizeof(WALEntryDisk); - size_t valid_size = raw_count * sizeof(WALEntryDisk); + int raw_count = size / sizeof(WALEntry); + size_t valid_size = raw_count * sizeof(WALEntry); if (valid_size < size) file_truncate(handle, valid_size); - WALEntryDisk *disk_entries = malloc(raw_count * sizeof(WALEntryDisk)); - if (disk_entries == NULL && raw_count > 0) { + 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(disk_entries); + free(entries); return -1; } - if (raw_count > 0 && file_read_exact(handle, (char *)disk_entries, raw_count * sizeof(WALEntryDisk)) < 0) { + if (raw_count > 0 && file_read_exact(handle, (char *)entries, raw_count * sizeof(WALEntry)) < 0) { file_close(handle); - free(disk_entries); + 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 (disk_entries[i].checksum != wal_entry_checksum(&disk_entries[i])) { + if (entries[i].checksum != wal_entry_checksum(&entries[i])) { count = i; - file_truncate(handle, count * sizeof(WALEntryDisk)); + file_truncate(handle, count * sizeof(WALEntry)); if (was_truncated) *was_truncated = true; break; } } - // Convert disk entries to in-memory entries. - WALEntry *entries = malloc(count * sizeof(WALEntry)); - if (entries == NULL && count > 0) { - file_close(handle); - free(disk_entries); - return -1; - } - for (int i = 0; i < count; i++) - entries[i] = wal_entry_from_disk(&disk_entries[i]); - free(disk_entries); - // Position file offset at end for future appends. - if (file_set_offset(handle, count * sizeof(WALEntryDisk)) < 0) { + if (file_set_offset(handle, count * sizeof(WALEntry)) < 0) { file_close(handle); free(entries); return -1; @@ -186,19 +157,19 @@ int wal_append(WAL *wal, WALEntry entry) } if (wal_is_file_backed(wal)) { - WALEntryDisk disk = wal_entry_to_disk(&entry); + wal_entry_set_checksum(&entry); - if (file_write_exact(wal->handle, (char *)&disk, sizeof(disk)) < 0) { + 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(WALEntryDisk)); - file_set_offset(wal->handle, wal->count * sizeof(WALEntryDisk)); + 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(WALEntryDisk)); - file_set_offset(wal->handle, wal->count * sizeof(WALEntryDisk)); + file_truncate(wal->handle, wal->count * sizeof(WALEntry)); + file_set_offset(wal->handle, wal->count * sizeof(WALEntry)); return -1; } } @@ -207,6 +178,31 @@ int wal_append(WAL *wal, WALEntry 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); @@ -214,9 +210,9 @@ int wal_truncate(WAL *wal, int new_count) return 0; if (wal_is_file_backed(wal)) { - if (file_truncate(wal->handle, new_count * sizeof(WALEntryDisk)) < 0) + if (file_truncate(wal->handle, new_count * sizeof(WALEntry)) < 0) return -1; - if (file_set_offset(wal->handle, new_count * sizeof(WALEntryDisk)) < 0) + if (file_set_offset(wal->handle, new_count * sizeof(WALEntry)) < 0) return -1; } @@ -248,8 +244,8 @@ int wal_replace(WAL *wal, WALEntry *entries, int count) // Write all entries to tmp.log. for (int i = 0; i < count; i++) { - WALEntryDisk disk = wal_entry_to_disk(&entries[i]); - if (file_write_exact(tmp, (char *)&disk, sizeof(disk)) < 0) { + wal_entry_set_checksum(&entries[i]); + if (file_write_exact(tmp, (char *)&entries[i], sizeof(entries[i])) < 0) { file_close(tmp); return -1; } @@ -274,7 +270,7 @@ int wal_replace(WAL *wal, WALEntry *entries, int count) Handle new_handle; if (file_open(wal_path, &new_handle) < 0) return -1; - if (file_set_offset(new_handle, count * sizeof(WALEntryDisk)) < 0) { + if (file_set_offset(new_handle, count * sizeof(WALEntry)) < 0) { file_close(new_handle); return -1; } diff --git a/src/wal.h b/src/wal.h index cee02da..42318ce 100644 --- a/src/wal.h +++ b/src/wal.h @@ -7,20 +7,12 @@ typedef struct { MetaOper oper; - uint32_t votes; // transient, not persisted to disk - int view_number; - uint64_t client_id; - uint64_t request_id; -} WALEntry; - -// On-disk representation of a WAL entry (excludes transient 'votes' field). -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 -} WALEntryDisk; +} WALEntry; _Static_assert(NODE_LIMIT <= 32, ""); @@ -53,6 +45,9 @@ void wal_move(WAL *dst, WAL *src); // 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);