cozis/ToastyFS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix WAL correctness: persist votes, fix crash recovery and view changes

Browse Source 
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
This commit is contained in:
Claude
2026-02-22 17:19:00 +00:00
parent b687de5b1c
commit 34af6063b6
5 changed files with 138 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/invariant_checker.c
+9 -2
View File
@@ -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;
}
src/server.c
+72 -47
View File
@@ -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);
}
src/server.h
+1
View File
@@ -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;
src/wal.c
+51 -55
View File
@@ -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;
}
src/wal.h
+5 -10
View File
@@ -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);