Author SHA1 Message Date
cozis 107a9ba5b8 Add TODO 2026-02-22 22:02:45 +01:00
cozis aef44ce5f3 Fix formatting 2026-02-22 21:44:37 +01:00
Claude 34af6063b6 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
2026-02-22 17:19:00 +00:00
Claude b687de5b1c Fix: enter recovery on clean restart to prevent log divergence
A node that crashed and restarted with a clean (non-truncated) WAL was
entering STATUS_NORMAL directly, skipping the recovery protocol. This
violated VR-Revisited Section 4.3: a crashed node must recover state at
least as recent as what it had before the crash.

The bug allows committed prefix agreement to break: while the node was
down, a view change may have replaced uncommitted log entries at positions
the node already holds. State transfer (GET_STATE/NEW_STATE) only appends
the suffix the node is missing, so the divergent entries persist and can
later be committed, causing replicas to disagree on committed operations.

Fix: any restart with a non-empty WAL (clean or corrupt) now enters
STATUS_RECOVERY, which atomically replaces the local log with the
primary's authoritative copy before resuming normal operation.

Also fix wal_init_from_network to handle num=0 (malloc(0) may return
NULL on some platforms, causing a spurious failure).

Verified with deterministic simulation (seeds 1-25, 30-60s each):
all WAL invariants pass (committed prefix agreement, shadow log,
min_commit monotonicity).

https://claude.ai/code/session_01X2b5VRntA7LM32kdN9tPAM
2026-02-22 16:20:38 +00:00
cozisandClaude Opus 4.6 9bec97715a 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>
2026-02-22 17:04:07 +01:00
Claude 5e7ab662d1 Add toastyfs_client to .gitignore
https://claude.ai/code/session_01YJnNDYw8JVb4HPSMf9rpuS
2026-02-22 15:33:12 +00:00
Claude 60dfe2068d Implement mock_access() for simulation file existence checks
mock_access() uses mockfs_open(RDONLY) to probe whether a path exists
in the mock filesystem, then immediately closes the handle. Directories
are detected via the MOCKFS_ERRNO_ISDIR return from mockfs_open.
This prevents the simulation from falling back to file_open(O_CREAT),
which would create empty files as a side effect and corrupt chunk data.

https://claude.ai/code/session_01YJnNDYw8JVb4HPSMf9rpuS
2026-02-22 15:32:33 +00:00
cozisandClaude Opus 4.6 5688860719 Add WAL persistence to VSR log
Port the Write-Ahead Log implementation from raft/ into the VSR
server. Log entries are now written to disk with FNV-1a checksums
and fsynced before updating in-memory state. On startup, the WAL
file is loaded and validated, replacing the boot marker for crash
detection. View changes and recovery use atomic rename (tmp.log ->
vsr.log) to replace the WAL safely.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-22 16:28:39 +01:00
13 changed files with 792 additions and 174 deletions
+2
View File
@@ -1,5 +1,6 @@
toastyfs toastyfs
toastyfs_client toastyfs_client
toastyfs_random_client
toastyfs_simulation toastyfs_simulation
*.gcno *.gcno
*.gcda *.gcda
@@ -13,5 +14,6 @@ coverage_html/
example example
example_large example_large
examples/example examples/example
.claude/
.cluster/ .cluster/
vsr_boot_marker vsr_boot_marker
+3 -3
View File
@@ -15,20 +15,20 @@ SHARED_LIB = libtoastyfs.so
# ---- Server binary ---- # ---- Server binary ----
SERVER_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \ 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 src/chunk_store.c src/metadata.c
# ---- Client binary (random test client) ---- # ---- Client binary (random test client) ----
CLIENT_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \ 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/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 ---- # ---- Simulation binary ----
SIM_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \ 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/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 src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c
# ---- Default target ---- # ---- Default target ----
+1
View File
@@ -1,3 +1,4 @@
[ ] Add log compaction [ ] Add log compaction
[ ] Add log persistence [ ] Add log persistence
[ ] Add chunk locations to metadata [ ] Add chunk locations to metadata
[ ] fsync parent directory when renaming log
+3 -3
View File
@@ -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
+23 -1
View File
@@ -3249,7 +3249,29 @@ int mock_close(int fd)
// data for chunks they don't actually have). // data for chunks they don't actually have).
int mock_access(const char *path, int mode) 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, static int convert_addr(void *addr, size_t addr_len,
+8 -5
View File
@@ -219,7 +219,8 @@ static void send_message_to_server(ToastyFS *tfs, int server_idx, MessageHeader
byte_queue_write(output, msg, msg->length); 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) MessageHeader *msg, void *extra, int extra_len)
{ {
ByteQueue *output; 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; (void) conn_idx;
@@ -673,7 +674,8 @@ static int process_message(ToastyFS *tfs,
return 0; 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]; Event events[TCP_EVENT_CAPACITY];
int num_events = tcp_translate_events(&tfs->tcp, events, ctxs, pdata, pnum); 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 // TODO: The toastyfs client needs to determine a timeout based on the
// pending operation status, not just use PRIMARY_DEATH_TIMEOUT_SEC // pending operation status, not just use PRIMARY_DEATH_TIMEOUT_SEC
// for everything. // 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 now = get_current_time(); // TODO: Handle INVALID_TIME error
Time deadline = INVALID_TIME; Time deadline = INVALID_TIME;
+26 -19
View File
@@ -58,8 +58,10 @@ void invariant_checker_init(InvariantChecker *ic)
{ {
ic->last_min_commit = -1; ic->last_min_commit = -1;
ic->last_max_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_status[i] = STATUS_NORMAL;
ic->prev_alive[i] = false;
}
ic->shadow_log = NULL; ic->shadow_log = NULL;
ic->shadow_count = 0; ic->shadow_count = 0;
ic->shadow_capacity = 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) { if (min_commit < 0 || min_commit > n->commit_index) {
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) { 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_min_commit = min_commit;
ic->last_max_commit = max_commit; ic->last_max_commit = max_commit;
for (int i = 0; i < num_nodes; i++) { for (int i = 0; i < num_nodes; i++) {
ic->prev_alive[i] = (nodes[i] != NULL);
if (nodes[i]) if (nodes[i])
ic->prev_status[i] = nodes[i]->status; 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 // 1. commit_index <= log.count
// A node cannot have committed more entries than it has in its log. // 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", 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(); __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. // 9. Log entry view numbers must not exceed the node's current view.
// Entries are created in the view they were proposed. No entry // Entries are created in the view they were proposed. No entry
// should carry a view number from the future. // 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) " fprintf(stderr, "INVARIANT VIOLATED: node %d: log[%d].view_number (%d) "
"> view_number (%lu)\n", "> 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); (unsigned long)s->view_number);
__builtin_trap(); __builtin_trap();
} }
@@ -221,8 +228,8 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
// votes for its own entries. // votes for its own entries.
if (s->status == STATUS_NORMAL && is_leader(s)) { if (s->status == STATUS_NORMAL && is_leader(s)) {
int idx = self_idx(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): " fprintf(stderr, "INVARIANT VIOLATED: node %d (leader): "
"uncommitted log[%d] missing leader's own vote bit\n", "uncommitted log[%d] missing leader's own vote bit\n",
i, k); i, k);
@@ -266,7 +273,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
mc = nodes[j]->commit_index; mc = nodes[j]->commit_index;
for (int k = 0; k < mc; k++) { 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 " fprintf(stderr, "INVARIANT VIOLATED: committed log operation mismatch at index %d "
"between node %d and node %d\n", k, i, j); "between node %d and node %d\n", k, i, j);
__builtin_trap(); __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) { if (source_node_idx >= 0 && observed_max_commit > ic->shadow_count) {
ServerState *source = nodes[source_node_idx]; 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++) { 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 // Cross-validate against other live non-recovering nodes
// that have also committed this entry. // that have also committed this entry.
@@ -315,10 +322,10 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
continue; continue;
if (nodes[j]->commit_index <= k) if (nodes[j]->commit_index <= k)
continue; continue;
if (nodes[j]->log.count <= k) if (nodes[j]->wal.count <= k)
continue; 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 " fprintf(stderr, "INVARIANT VIOLATED: committed entry mismatch at index %d "
"between source node %d and node %d during shadow log append\n", "between source node %d and node %d during shadow log append\n",
k, source_node_idx, j); 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++) { for (int i = 0; i < num_nodes; i++) {
if (nodes[i] == NULL) if (nodes[i] == NULL)
continue; continue;
if (nodes[i]->log.count <= k) if (nodes[i]->wal.count <= k)
continue; continue;
if (nodes[i]->commit_index <= k) if (nodes[i]->commit_index <= k)
continue; 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]; char shadow_buf[128], node_buf[128];
meta_snprint_oper(shadow_buf, sizeof(shadow_buf), &ic->shadow_log[k]); 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" fprintf(stderr, "INVARIANT VIOLATED: shadow log mismatch at index %d on node %d\n"
" shadow: %s\n" " shadow: %s\n"
" node: %s\n", " node: %s\n",
@@ -377,9 +384,9 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
num_dead++; num_dead++;
continue; continue;
} }
if (nodes[i]->log.count <= k) if (nodes[i]->wal.count <= k)
continue; 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++; holders++;
} }
+3
View File
@@ -1,3 +1,4 @@
#if 0
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST) #if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS #define QUAKEY_ENABLE_MOCKS
#endif #endif
@@ -53,3 +54,5 @@ int log_append(Log *log, LogEntry entry)
log->entries[log->count++] = entry; log->entries[log->count++] = entry;
return 0; return 0;
} }
#endif
+2
View File
@@ -1,3 +1,4 @@
#if 0
#ifndef LOG_INCLUDED #ifndef LOG_INCLUDED
#define LOG_INCLUDED #define LOG_INCLUDED
@@ -28,3 +29,4 @@ void log_move(Log *dst, Log *src);
int log_append(Log *log, LogEntry entry); int log_append(Log *log, LogEntry entry);
#endif // LOG_INCLUDED #endif // LOG_INCLUDED
#endif
+209 -132
View File
@@ -11,6 +11,7 @@
typedef enum { typedef enum {
HR_OK, HR_OK,
HR_OUT_OF_MEMORY, HR_OUT_OF_MEMORY,
HR_IO_FAILURE,
HR_INVALID_MESSAGE, HR_INVALID_MESSAGE,
} HandlerResult; } HandlerResult;
@@ -58,7 +59,7 @@ static void node_log_impl(ServerState *state, const char *event, const char *det
status_name(state->status), status_name(state->status),
state->view_number, state->view_number,
state->commit_index, state->commit_index,
state->log.count, state->wal.count,
event, event,
detail ? detail : ""); detail ? detail : "");
} }
@@ -132,6 +133,23 @@ static void broadcast_to_peers(ServerState *state, MessageHeader *msg)
broadcast_to_peers_ex(state, msg, NULL, 0); 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) static void begin_state_transfer(ServerState *state, int sender_idx)
{ {
if (state->state_transfer_pending) if (state->state_transfer_pending)
@@ -145,10 +163,10 @@ static void begin_state_transfer(ServerState *state, int sender_idx)
.length = sizeof(GetStateMessage), .length = sizeof(GetStateMessage),
}, },
.view_number = state->view_number, .view_number = state->view_number,
.op_number = state->log.count, .op_number = state->wal.count,
.sender_idx = self_idx(state), .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); send_to_peer(state, sender_idx, &message.base);
state->state_transfer_time = state->now; 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, .oper = request_message.oper,
.votes = 1 << self_idx(state), .votes = 1 << self_idx(state),
.view_number = state->view_number, .view_number = state->view_number,
.client_id = request_message.client_id, .client_id = request_message.client_id,
.request_id = request_message.request_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; return HR_OUT_OF_MEMORY;
// We forwarded the message to all peers. As soon as // 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, .oper = request_message.oper,
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
.log_index = state->log.count-1, .log_index = state->wal.count-1,
.commit_index = state->commit_index, .commit_index = state->commit_index,
.view_number = state->view_number, .view_number = state->view_number,
.client_id = request_message.client_id, .client_id = request_message.client_id,
@@ -296,7 +314,7 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
{ {
char oper_buf[128]; char oper_buf[128];
meta_snprint_oper(oper_buf, sizeof(oper_buf), &request_message.oper); 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); broadcast_to_peers(state, &prepare_message.base);
return HR_OK; return HR_OK;
@@ -333,13 +351,20 @@ 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->log.count); target_index = MIN(target_index, state->wal.count);
while (state->commit_index < target_index) { 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); 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) 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
@@ -426,18 +453,21 @@ 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;
adopt_view(state);
persist_state(state);
begin_state_transfer(state, message.sender_idx); begin_state_transfer(state, message.sender_idx);
return HR_OK; return HR_OK;
} }
assert(message.log_index > -1); 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) if (message.log_index < state->commit_index)
return HR_OK; // Already processed 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); add_vote(&entry->votes, message.sender_idx);
wal_update_entry(&state->wal, message.log_index);
if (reached_quorum(state, entry->votes)) { if (reached_quorum(state, entry->votes)) {
node_log(state, "QUORUM", "idx=%d %s/%s", message.log_index, entry->oper.bucket, entry->oper.key); 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); 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_apply_votes = 0;
state->view_change_old_view = 0; state->view_change_old_view = 0;
state->view_change_commit = 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 static HandlerResult
@@ -476,10 +506,24 @@ complete_view_change_and_become_primary(ServerState *state)
{ {
assert(state->commit_index <= state->view_change_commit); 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->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.
@@ -490,31 +534,19 @@ complete_view_change_and_become_primary(ServerState *state)
// requests, otherwise the state machine will be stale. // requests, otherwise the state machine will be stale.
advance_commit_index(state, state->view_change_commit, false); 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 = { BeginViewMessage begin_view_message = {
.base = { .base = {
.version = MESSAGE_VERSION, .version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_BEGIN_VIEW, .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, .view_number = state->view_number,
.commit_index = state->commit_index, .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", 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); clear_view_change_fields(state);
return HR_OK; 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; 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 // 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) if (num_entries != message.op_number)
return HR_INVALID_MESSAGE; // Message size mismatch 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; return HR_OUT_OF_MEMORY;
} }
@@ -599,15 +631,15 @@ process_recovery(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(RecoveryResponseMessage), .length = sizeof(RecoveryResponseMessage),
}, },
.view_number = state->view_number, .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, .nonce = recovery_message.nonce,
.commit_index = state->commit_index, .commit_index = state->commit_index,
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
}; };
if (is_primary(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, 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 { } else {
send_to_peer(state, recovery_message.sender_idx, &recovery_response_message.base); 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; state->view_change_commit = state->commit_index;
// TODO: This should use copy-on-write // 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; return HR_OUT_OF_MEMORY;
} else { } else {
@@ -635,18 +667,18 @@ perform_log_transfer_for_view_change(ServerState *state)
.base = { .base = {
.version = MESSAGE_VERSION, .version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_DO_VIEW_CHANGE, .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, .view_number = state->view_number,
.old_view_number = state->last_normal_view, .old_view_number = state->last_normal_view,
.op_number = state->log.count, .op_number = state->wal.count,
.commit_index = state->commit_index, .commit_index = state->commit_index,
.sender_idx = self_idx(state), .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", 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, 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 // 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); 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);
@@ -734,26 +767,31 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
if (message.view_number < state->view_number) if (message.view_number < state->view_number)
return HR_OK; 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); 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->num_future = 0;
state->state_transfer_pending = false; state->state_transfer_pending = false;
// If there are non-committed operations in the log, // If there are non-committed operations in the log,
// send a PREPAREOK to the new primary // send a PREPAREOK to the new primary
if (state->log.count > message.commit_index) { if (state->wal.count > message.commit_index) {
PrepareOKMessage ok_msg = { PrepareOKMessage ok_msg = {
.base = { .base = {
.version = MESSAGE_VERSION, .version = MESSAGE_VERSION,
@@ -761,12 +799,12 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(PrepareOKMessage), .length = sizeof(PrepareOKMessage),
}, },
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
.log_index = state->log.count - 1, .log_index = state->wal.count - 1,
.view_number = state->view_number, .view_number = state->view_number,
}; };
send_to_peer(state, primary_idx(state), &ok_msg.base); 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); 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 // Compute the suffix of log entries the requester is missing
int start = get_state_message.op_number; 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 return HR_OK; // Nothing to send
int num_entries = state->log.count - start; int num_entries = state->wal.count - start;
NewStateMessage new_state_message = { NewStateMessage new_state_message = {
.base = { .base = {
.version = MESSAGE_VERSION, .version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_NEW_STATE, .type = MESSAGE_TYPE_NEW_STATE,
.length = sizeof(NewStateMessage) + num_entries * sizeof(LogEntry), .length = sizeof(NewStateMessage) + num_entries * sizeof(WALEntry),
}, },
.view_number = state->view_number, .view_number = state->view_number,
.op_number = num_entries, .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", node_log(state, "SEND NEW_STATE", "to=%d entries=%d commit=%d",
get_state_message.sender_idx, num_entries, state->commit_index); get_state_message.sender_idx, num_entries, state->commit_index);
send_to_peer_ex(state, get_state_message.sender_idx, &new_state_message.base, 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; return HR_OK;
} }
@@ -1040,23 +1078,29 @@ complete_recovery(ServerState *state)
assert(state->commit_index <= state->recovery_commit); assert(state->commit_index <= state->recovery_commit);
state->view_number = state->recovery_view; 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); advance_commit_index(state, state->recovery_commit, false);
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);
// 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 // Update heartbeat to avoid immediate view change timeout
state->heartbeat = state->now; state->heartbeat = state->now;
return HR_OK; return HR_OK;
@@ -1124,13 +1168,13 @@ process_recovery_response(ServerState *state, ByteView msg)
if (should_store_recovery_log(state, message)) { 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; 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; return HR_OUT_OF_MEMORY;
state->recovery_log_view = message.view_number; 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 // Look for an entry with the current log index
int i = 0; 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++; i++;
if (i == state->num_future) if (i == state->num_future)
return 0; // No entry return 0; // No entry
LogEntry entry = { WALEntry entry = {
.oper = state->future[i].oper, .oper = state->future[i].oper,
.votes = 0, .votes = 0,
.view_number = state->view_number, .view_number = state->view_number,
.client_id = state->future[i].client_id, .client_id = state->future[i].client_id,
.request_id = state->future[i].request_id, .request_id = state->future[i].request_id,
}; };
if (log_append(&state->log, entry) < 0) if (wal_append(&state->wal, entry) < 0)
return -1; return -1;
PrepareOKMessage message = { PrepareOKMessage message = {
@@ -1174,7 +1218,7 @@ process_single_future_list_entry(ServerState *state)
.length = sizeof(PrepareOKMessage), .length = sizeof(PrepareOKMessage),
}, },
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
.log_index = state->log.count-1, .log_index = state->wal.count-1,
.view_number = state->view_number, .view_number = state->view_number,
}; };
send_to_peer(state, state->future[i].sender_idx, &message.base); send_to_peer(state, state->future[i].sender_idx, &message.base);
@@ -1185,7 +1229,7 @@ static void
remove_old_future_list_entries(ServerState *state) remove_old_future_list_entries(ServerState *state)
{ {
for (int i = 0; i < state->num_future; i++) { 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]; 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 // 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;
adopt_view(state);
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);
return HR_OK; 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. 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) 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);
return HR_OK; return HR_OK;
} }
LogEntry log_entry = { WALEntry wal_entry = {
.oper = message.oper, .oper = message.oper,
.votes = 0, .votes = 0,
.view_number = state->view_number, .view_number = state->view_number,
.client_id = message.client_id, .client_id = message.client_id,
.request_id = message.request_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; return HR_OUT_OF_MEMORY;
PrepareOKMessage ok_message = { PrepareOKMessage ok_message = {
@@ -1261,12 +1307,12 @@ process_prepare(ServerState *state, ByteView msg)
.length = sizeof(PrepareOKMessage), .length = sizeof(PrepareOKMessage),
}, },
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
.log_index = state->log.count-1, .log_index = state->wal.count-1,
.view_number = state->view_number, .view_number = state->view_number,
}; };
send_to_peer(state, message.sender_idx, &ok_message.base); send_to_peer(state, message.sender_idx, &ok_message.base);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", 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); process_future_list(state);
advance_commit_index(state, message.commit_index, false); 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)); memcpy(&new_state_message, msg.ptr, sizeof(new_state_message));
node_log(state, "RECV NEW_STATE", "entries=%d commit=%d view=%lu", 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 // Ignore if we're in a different view
if (new_state_message.view_number != state->view_number) if (new_state_message.view_number != state->view_number)
return HR_OK; 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) if (num_entries != new_state_message.op_number)
return HR_INVALID_MESSAGE; return HR_INVALID_MESSAGE;
@@ -1328,22 +1375,22 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
// Append received entries to our log. // Append received entries to our log.
// The entries array is a suffix of the sender's log starting at // The entries array is a suffix of the sender's log starting at
// global position start_index. We skip entries we already have. // 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; int start_index = new_state_message.start_index;
for (int i = 0; i < num_entries; i++) { for (int i = 0; i < num_entries; i++) {
int global_idx = start_index + i; int global_idx = start_index + i;
if (global_idx < state->log.count) if (global_idx < state->wal.count)
continue; // Already have this entry continue; // Already have this entry
LogEntry entry = { WALEntry entry = {
.oper = entries[i].oper, .oper = entries[i].oper,
.votes = 0, .votes = 0,
.view_number = state->view_number, .view_number = state->view_number,
.client_id = entries[i].client_id, .client_id = entries[i].client_id,
.request_id = entries[i].request_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; return HR_OUT_OF_MEMORY;
// Send PREPARE_OK for each appended entry // Send PREPARE_OK for each appended entry
@@ -1354,12 +1401,12 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(PrepareOKMessage), .length = sizeof(PrepareOKMessage),
}, },
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
.log_index = state->log.count - 1, .log_index = state->wal.count - 1,
.view_number = state->view_number, .view_number = state->view_number,
}; };
send_to_peer(state, primary_idx(state), &prepare_ok_message.base); 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); process_future_list(state);
@@ -1652,10 +1699,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;
@@ -1665,41 +1709,80 @@ int server_init(void *state_, int argc, char **argv,
state->view_change_apply_votes = 0; state->view_change_apply_votes = 0;
state->view_change_old_view = 0; state->view_change_old_view = 0;
state->view_change_commit = 0; state->view_change_commit = 0;
log_init(&state->view_change_log); wal_init(&state->view_change_log);
// Recovery state // Recovery state
state->recovery_votes = 0; state->recovery_votes = 0;
state->recovery_commit = 0; state->recovery_commit = 0;
state->recovery_view = 0; state->recovery_view = 0;
state->recovery_log_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->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;
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); 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);
// 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) { 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;
@@ -1717,16 +1800,7 @@ int server_init(void *state_, int argc, char **argv,
return -1; 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); 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
@@ -1793,7 +1867,7 @@ int server_tick(void *state_, void **ctxs,
tcp_close(&state->tcp, conn_idx); tcp_close(&state->tcp, conn_idx);
break; break;
} }
if (hret == HR_OUT_OF_MEMORY) if (hret == HR_OUT_OF_MEMORY || hret == HR_IO_FAILURE)
return -1; return -1;
assert(hret == HR_OK); 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)); 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 = {
@@ -1903,6 +1978,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;
@@ -1932,7 +2008,7 @@ int server_tick(void *state_, void **ctxs,
Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL; Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL;
if (st_deadline <= state->now) { 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 = { GetStateMessage get_state_message = {
.base = { .base = {
@@ -1941,11 +2017,11 @@ int server_tick(void *state_, void **ctxs,
.length = sizeof(GetStateMessage), .length = sizeof(GetStateMessage),
}, },
.view_number = state->view_number, .view_number = state->view_number,
.op_number = state->log.count, .op_number = state->wal.count,
.sender_idx = self_idx(state), .sender_idx = self_idx(state),
}; };
send_to_peer(state, primary_idx(state), &get_state_message.base); 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; state->state_transfer_time = state->now;
@@ -1967,9 +2043,10 @@ int server_free(void *state_)
node_log_simple(state, "CRASHED"); 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); tcp_context_free(&state->tcp);
client_table_free(&state->client_table); client_table_free(&state->client_table);
meta_store_free(&state->metastore); meta_store_free(&state->metastore);
+10 -8
View File
@@ -4,7 +4,7 @@
#include "tcp.h" #include "tcp.h"
#include "basic.h" #include "basic.h"
#include "message.h" #include "message.h"
#include "log.h" #include "wal.h"
#include "config.h" #include "config.h"
#include "metadata.h" #include "metadata.h"
#include "chunk_store.h" #include "chunk_store.h"
@@ -99,15 +99,15 @@ typedef struct {
int op_number; // Number of entries in the log int op_number; // Number of entries in the log
int commit_index; int commit_index;
int sender_idx; int sender_idx;
// Followed by: LogEntry log[op_number] // Followed by: WALEntry log[op_number]
} DoViewChangeMessage; } DoViewChangeMessage;
typedef struct { typedef struct {
MessageHeader base; MessageHeader base;
uint64_t view_number; uint64_t view_number;
int commit_index; 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; } BeginViewMessage;
typedef struct { typedef struct {
@@ -138,7 +138,7 @@ typedef struct {
int op_number; // Number of log entries that follow int op_number; // Number of log entries that follow
int commit_index; int commit_index;
int start_index; // Global log index of the first entry in the suffix 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; } NewStateMessage;
typedef struct { typedef struct {
@@ -228,7 +228,7 @@ typedef struct {
uint32_t recovery_votes; uint32_t recovery_votes;
uint64_t recovery_nonce; uint64_t recovery_nonce;
uint64_t recovery_view; uint64_t recovery_view;
Log recovery_log; WAL recovery_log;
uint64_t recovery_log_view; uint64_t recovery_log_view;
Time recovery_time; Time recovery_time;
int recovery_commit; int recovery_commit;
@@ -238,7 +238,7 @@ typedef struct {
uint32_t view_change_begin_votes; uint32_t view_change_begin_votes;
uint32_t view_change_apply_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 uint64_t view_change_old_view; // Best old_view_number seen in DoViewChange
int view_change_commit; // Best commit_index seen int view_change_commit; // Best commit_index seen
@@ -255,7 +255,8 @@ typedef struct {
bool state_transfer_pending; bool state_transfer_pending;
Time state_transfer_time; Time state_transfer_time;
Log log; WAL wal;
ViewAndCommit vc;
Time heartbeat; Time heartbeat;
@@ -282,6 +283,7 @@ typedef struct {
int last_min_commit; int last_min_commit;
int last_max_commit; int last_max_commit;
Status prev_status[NODE_LIMIT]; Status prev_status[NODE_LIMIT];
bool prev_alive[NODE_LIMIT];
// External shadow log of committed operations (unbounded, dynamically allocated) // External shadow log of committed operations (unbounded, dynamically allocated)
MetaOper *shadow_log; MetaOper *shadow_log;
+422
View File
@@ -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;
}
+77
View File
@@ -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