Add persistent view_number/commit_index and WAL replay on startup

On startup, the server now loads view_number, last_normal_view and
commit_index from a durable state file (vsr.state) and replays the
WAL up to the saved commit_index when the log is intact. If the log
is corrupt (checksum failure), the server truncates it and enters
recovery mode to fetch a valid log from peers.

ViewAndCommit follows the same persistence pattern as raft's
TermAndVote: fixed-size record with FNV-1a checksum, written with
fsync on every mutation.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-02-22 17:04:07 +01:00
co-authored by Claude Opus 4.6
parent 5e7ab662d1
commit 9bec97715a
4 changed files with 195 additions and 15 deletions
+64 -12
View File
@@ -334,6 +334,13 @@ static void reply_to_client(ServerState *state, ClientTableEntry *table_entry,
byte_queue_write(output, &message, sizeof(message)); 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) static void advance_commit_index(ServerState *state, int target_index, bool send_replies)
{ {
target_index = MIN(target_index, state->wal.count); target_index = MIN(target_index, state->wal.count);
@@ -376,6 +383,8 @@ static void advance_commit_index(ServerState *state, int target_index, bool send
if (send_replies) if (send_replies)
reply_to_client(state, table_entry, entry->request_id, &entry->oper, result); 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 // When the primary appends an entry to its log, it sends a
@@ -427,6 +436,7 @@ process_prepare_ok(ServerState *state, int conn_idx, ByteView msg)
if (message.view_number > state->view_number) { if (message.view_number > state->view_number) {
state->view_number = message.view_number; state->view_number = message.view_number;
persist_state(state);
begin_state_transfer(state, message.sender_idx); begin_state_transfer(state, message.sender_idx);
return HR_OK; return HR_OK;
} }
@@ -484,6 +494,7 @@ complete_view_change_and_become_primary(ServerState *state)
state->status = STATUS_NORMAL; state->status = STATUS_NORMAL;
state->last_normal_view = state->view_number; state->last_normal_view = state->view_number;
persist_state(state);
node_log(state, "STATUS NORMAL", "became primary (view change complete)"); node_log(state, "STATUS NORMAL", "became primary (view change complete)");
// Apply committed entries that haven't been executed yet. // Apply committed entries that haven't been executed yet.
@@ -703,6 +714,7 @@ process_begin_view_change(ServerState *state, int conn_idx, ByteView msg)
clear_view_change_fields(state); clear_view_change_fields(state);
state->view_number = message.view_number; state->view_number = message.view_number;
persist_state(state);
state->heartbeat = state->now; state->heartbeat = state->now;
state->status = STATUS_CHANGE_VIEW; state->status = STATUS_CHANGE_VIEW;
node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number); node_log(state, "STATUS CHANGE_VIEW", "view=%lu", state->view_number);
@@ -742,6 +754,7 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
state->status = STATUS_NORMAL; state->status = STATUS_NORMAL;
state->last_normal_view = state->view_number; state->last_normal_view = state->view_number;
persist_state(state);
node_log(state, "STATUS NORMAL", "new view=%lu (follower)", state->view_number); node_log(state, "STATUS NORMAL", "new view=%lu (follower)", state->view_number);
int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(WALEntry); int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(WALEntry);
@@ -1052,6 +1065,7 @@ complete_recovery(ServerState *state)
state->status = STATUS_NORMAL; state->status = STATUS_NORMAL;
state->last_normal_view = state->view_number; state->last_normal_view = state->view_number;
persist_state(state);
node_log(state, "STATUS NORMAL", "recovery complete view=%lu commit=%d", node_log(state, "STATUS NORMAL", "recovery complete view=%lu commit=%d",
state->view_number, state->commit_index); state->view_number, state->commit_index);
@@ -1235,6 +1249,7 @@ process_prepare(ServerState *state, ByteView msg)
// itself up to date before processing the message // itself up to date before processing the message
if (message.view_number > state->view_number) { if (message.view_number > state->view_number) {
state->view_number = message.view_number; state->view_number = message.view_number;
persist_state(state);
if (state->num_future < FUTURE_LIMIT) if (state->num_future < FUTURE_LIMIT)
state->future[state->num_future++] = message; state->future[state->num_future++] = message;
begin_state_transfer(state, message.sender_idx); begin_state_transfer(state, message.sender_idx);
@@ -1659,10 +1674,7 @@ int server_init(void *state_, int argc, char **argv,
Time deadline = INVALID_TIME; Time deadline = INVALID_TIME;
state->view_number = 0;
state->last_normal_view = 0;
state->heartbeat = now; state->heartbeat = now;
state->commit_index = 0;
state->num_future = 0; state->num_future = 0;
state->state_transfer_pending = false; state->state_transfer_pending = false;
state->state_transfer_time = 0; state->state_transfer_time = 0;
@@ -1681,27 +1693,64 @@ int server_init(void *state_, int argc, char **argv,
state->recovery_log_view = 0; state->recovery_log_view = 0;
wal_init(&state->recovery_log); wal_init(&state->recovery_log);
// Open the WAL file. If it contains entries from a previous run, // Load persistent view_number, last_normal_view and commit_index.
// this node crashed and must enter recovery mode to learn the if (view_and_commit_init(&state->vc, S("vsr.state")) < 0) {
// current view from peers before participating in the protocol. fprintf(stderr, "Node :: Couldn't open state file\n");
if (wal_init_from_file(&state->wal, S("vsr.log")) < 0) {
fprintf(stderr, "Node :: Couldn't open WAL file\n");
return -1; return -1;
} }
bool previously_crashed = (state->wal.count > 0); state->view_number = state->vc.view_number;
if (previously_crashed) { 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. 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");
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->status = STATUS_RECOVERY;
state->recovery_nonce = now; state->recovery_nonce = now;
state->recovery_time = now; state->recovery_time = now;
} else { } else {
state->status = STATUS_NORMAL; state->status = STATUS_NORMAL;
} }
node_log(state, "INIT", "nodes=%d%s", state->num_nodes, previously_crashed ? " (recovering)" : ""); node_log(state, "INIT", "nodes=%d%s", state->num_nodes, enter_recovery ? " (recovering)" : "");
client_table_init(&state->client_table); client_table_init(&state->client_table);
state->next_client_tag = NODE_LIMIT; // Make sure they don't overlap with node indices state->next_client_tag = NODE_LIMIT; // Make sure they don't overlap with node indices
meta_store_init(&state->metastore); 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);
node_log(state, "REPLAY", "applied %d entries", state->commit_index);
}
if (chunk_store_init(&state->chunk_store, chunks_path) < 0) { if (chunk_store_init(&state->chunk_store, chunks_path) < 0) {
fprintf(stderr, "Node :: Couldn't initialize chunk store at '%s'\n", chunks_path); fprintf(stderr, "Node :: Couldn't initialize chunk store at '%s'\n", chunks_path);
return -1; return -1;
@@ -1719,7 +1768,7 @@ int server_init(void *state_, int argc, char **argv,
return -1; return -1;
} }
if (previously_crashed) { if (enter_recovery) {
node_log(state, "STATUS RECOVERY", "nonce=%lu (crash detected)", state->recovery_nonce); node_log(state, "STATUS RECOVERY", "nonce=%lu (crash detected)", state->recovery_nonce);
// Broadcast RECOVERY to all peers to learn the current view // Broadcast RECOVERY to all peers to learn the current view
@@ -1843,6 +1892,7 @@ int server_tick(void *state_, void **ctxs,
add_vote(&state->view_change_begin_votes, self_idx(state)); add_vote(&state->view_change_begin_votes, self_idx(state));
state->view_number++; state->view_number++;
persist_state(state);
state->heartbeat = state->now; state->heartbeat = state->now;
BeginViewChangeMessage begin_view_change_message = { BeginViewChangeMessage begin_view_change_message = {
@@ -1896,6 +1946,7 @@ int server_tick(void *state_, void **ctxs,
add_vote(&state->view_change_begin_votes, self_idx(state)); add_vote(&state->view_change_begin_votes, self_idx(state));
state->view_number++; state->view_number++;
persist_state(state);
state->status = STATUS_CHANGE_VIEW; state->status = STATUS_CHANGE_VIEW;
state->heartbeat = state->now; state->heartbeat = state->now;
@@ -1961,6 +2012,7 @@ int server_free(void *state_)
node_log_simple(state, "CRASHED"); node_log_simple(state, "CRASHED");
wal_free(&state->wal); wal_free(&state->wal);
view_and_commit_free(&state->vc);
wal_free(&state->recovery_log); wal_free(&state->recovery_log);
wal_free(&state->view_change_log); wal_free(&state->view_change_log);
tcp_context_free(&state->tcp); tcp_context_free(&state->tcp);
+1
View File
@@ -256,6 +256,7 @@ typedef struct {
Time state_transfer_time; Time state_transfer_time;
WAL wal; WAL wal;
ViewAndCommit vc;
Time heartbeat; Time heartbeat;
+111 -1
View File
@@ -70,8 +70,11 @@ void wal_init(WAL *wal)
wal->handle = (Handle) { 0 }; wal->handle = (Handle) { 0 };
} }
int wal_init_from_file(WAL *wal, string file) int wal_init_from_file(WAL *wal, string file, bool *was_truncated)
{ {
if (was_truncated)
*was_truncated = false;
Handle handle; Handle handle;
if (file_open(file, &handle) < 0) if (file_open(file, &handle) < 0)
return -1; return -1;
@@ -112,6 +115,8 @@ int wal_init_from_file(WAL *wal, string file)
if (disk_entries[i].checksum != wal_entry_checksum(&disk_entries[i])) { if (disk_entries[i].checksum != wal_entry_checksum(&disk_entries[i])) {
count = i; count = i;
file_truncate(handle, count * sizeof(WALEntryDisk)); file_truncate(handle, count * sizeof(WALEntryDisk));
if (was_truncated)
*was_truncated = true;
break; break;
} }
} }
@@ -301,3 +306,108 @@ WALEntry *wal_peek_entry(WAL *wal, int idx)
assert(idx < wal->count); assert(idx < wal->count);
return &wal->entries[idx]; 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)) {
ViewAndCommitDisk disk;
if (file_set_offset(handle, 0) < 0) {
file_close(handle);
return -1;
}
if (file_read_exact(handle, (char *)&disk, sizeof(disk)) < 0) {
file_close(handle);
return -1;
}
if (disk.checksum == vc_checksum(disk.view_number, disk.last_normal_view, disk.commit_index)) {
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;
}
+19 -2
View File
@@ -35,8 +35,10 @@ typedef struct {
void wal_init(WAL *wal); void wal_init(WAL *wal);
// Initialize a WAL from an on-disk file. Recovers valid entries, // Initialize a WAL from an on-disk file. Recovers valid entries,
// discards partial/corrupted trailing entries. // discards partial/corrupted trailing entries. If was_truncated is
int wal_init_from_file(WAL *wal, string file); // 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). // Initialize a WAL from network data (memory-only, no file).
int wal_init_from_network(WAL *wal, void *src, int num); int wal_init_from_network(WAL *wal, void *src, int num);
@@ -62,4 +64,19 @@ int wal_replace(WAL *wal, WALEntry *entries, int count);
int wal_entry_count(WAL *wal); int wal_entry_count(WAL *wal);
WALEntry *wal_peek_entry(WAL *wal, int idx); 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 #endif // WAL_INCLUDED