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

Add WAL persistence to VSR log

Browse Source 
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>
This commit is contained in:
cozis
2026-02-22 16:28:39 +01:00
co-authored by Claude Opus 4.6
parent 487f3bf010
commit 5688860719
10 changed files with 503 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -1,5 +1,5 @@
toastyfs
toastyfs_client
toastyfs_random_client
toastyfs_simulation
*.gcno
*.gcda
@@ -13,5 +13,6 @@ coverage_html/
example
example_large
examples/example
.claude/
.cluster/
vsr_boot_marker
Makefile
+3 -3
View File
@@ -15,20 +15,20 @@ SHARED_LIB = libtoastyfs.so
# ---- Server binary ----
SERVER_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
src/tcp.c src/server.c src/main.c src/log.c src/client_table.c \
src/tcp.c src/server.c src/main.c src/wal.c src/client_table.c \
src/chunk_store.c src/metadata.c
# ---- Client binary (random test client) ----
CLIENT_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
src/tcp.c src/server.c src/client.c src/random_client.c src/main.c \
src/log.c src/client_table.c src/chunk_store.c src/metadata.c
src/wal.c src/client_table.c src/chunk_store.c src/metadata.c
# ---- Simulation binary ----
SIM_SRCS = src/basic.c src/file_system.c src/byte_queue.c src/message.c \
src/tcp.c src/server.c src/client.c src/random_client.c src/main.c \
src/log.c src/client_table.c src/invariant_checker.c src/chunk_store.c \
src/wal.c src/client_table.c src/invariant_checker.c src/chunk_store.c \
src/metadata.c quakey/src/mockfs.c quakey/src/quakey.c
# ---- Default target ----
build.sh
+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/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/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/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/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/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
src/invariant_checker.c
+17 -17
View File
@@ -164,9 +164,9 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
// 1. commit_index <= log.count
// A node cannot have committed more entries than it has in its log.
if (s->commit_index > s->log.count) {
if (s->commit_index > s->wal.count) {
fprintf(stderr, "INVARIANT VIOLATED: node %d: commit_index (%d) > log.count (%d)\n",
i, s->commit_index, s->log.count);
i, s->commit_index, s->wal.count);
__builtin_trap();
}
@@ -206,11 +206,11 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
// 9. Log entry view numbers must not exceed the node's current view.
// Entries are created in the view they were proposed. No entry
// should carry a view number from the future.
for (int k = 0; k < s->log.count; k++) {
if ((uint64_t)s->log.entries[k].view_number > s->view_number) {
for (int k = 0; k < s->wal.count; k++) {
if ((uint64_t)s->wal.entries[k].view_number > s->view_number) {
fprintf(stderr, "INVARIANT VIOLATED: node %d: log[%d].view_number (%d) "
"> view_number (%lu)\n",
i, k, s->log.entries[k].view_number,
i, k, s->wal.entries[k].view_number,
(unsigned long)s->view_number);
__builtin_trap();
}
@@ -221,8 +221,8 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
// votes for its own entries.
if (s->status == STATUS_NORMAL && is_leader(s)) {
int idx = self_idx(s);
for (int k = s->commit_index; k < s->log.count; k++) {
if (!(s->log.entries[k].votes & (1 << idx))) {
for (int k = s->commit_index; k < s->wal.count; k++) {
if (!(s->wal.entries[k].votes & (1 << idx))) {
fprintf(stderr, "INVARIANT VIOLATED: node %d (leader): "
"uncommitted log[%d] missing leader's own vote bit\n",
i, k);
@@ -266,7 +266,7 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
mc = nodes[j]->commit_index;
for (int k = 0; k < mc; k++) {
if (memcmp(&nodes[i]->log.entries[k].oper, &nodes[j]->log.entries[k].oper, sizeof(MetaOper)) != 0) {
if (memcmp(&nodes[i]->wal.entries[k].oper, &nodes[j]->wal.entries[k].oper, sizeof(MetaOper)) != 0) {
fprintf(stderr, "INVARIANT VIOLATED: committed log operation mismatch at index %d "
"between node %d and node %d\n", k, i, j);
__builtin_trap();
@@ -298,11 +298,11 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
if (source_node_idx >= 0 && observed_max_commit > ic->shadow_count) {
ServerState *source = nodes[source_node_idx];
assert(source->log.count >= observed_max_commit);
assert(source->wal.count >= observed_max_commit);
for (int k = ic->shadow_count; k < observed_max_commit; k++) {
MetaOper *source_oper = &source->log.entries[k].oper;
MetaOper *source_oper = &source->wal.entries[k].oper;
// Cross-validate against other live non-recovering nodes
// that have also committed this entry.
@@ -315,10 +315,10 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
continue;
if (nodes[j]->commit_index <= k)
continue;
if (nodes[j]->log.count <= k)
if (nodes[j]->wal.count <= k)
continue;
if (memcmp(&nodes[j]->log.entries[k].oper, source_oper, sizeof(MetaOper)) != 0) {
if (memcmp(&nodes[j]->wal.entries[k].oper, source_oper, sizeof(MetaOper)) != 0) {
fprintf(stderr, "INVARIANT VIOLATED: committed entry mismatch at index %d "
"between source node %d and node %d during shadow log append\n",
k, source_node_idx, j);
@@ -340,15 +340,15 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
for (int i = 0; i < num_nodes; i++) {
if (nodes[i] == NULL)
continue;
if (nodes[i]->log.count <= k)
if (nodes[i]->wal.count <= k)
continue;
if (nodes[i]->commit_index <= k)
continue;
if (memcmp(&nodes[i]->log.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) != 0) {
if (memcmp(&nodes[i]->wal.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) != 0) {
char shadow_buf[128], node_buf[128];
meta_snprint_oper(shadow_buf, sizeof(shadow_buf), &ic->shadow_log[k]);
meta_snprint_oper(node_buf, sizeof(node_buf), &nodes[i]->log.entries[k].oper);
meta_snprint_oper(node_buf, sizeof(node_buf), &nodes[i]->wal.entries[k].oper);
fprintf(stderr, "INVARIANT VIOLATED: shadow log mismatch at index %d on node %d\n"
" shadow: %s\n"
" node: %s\n",
@@ -377,9 +377,9 @@ void invariant_checker_run(InvariantChecker *ic, ServerState **nodes, int num_no
num_dead++;
continue;
}
if (nodes[i]->log.count <= k)
if (nodes[i]->wal.count <= k)
continue;
if (memcmp(&nodes[i]->log.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) == 0)
if (memcmp(&nodes[i]->wal.entries[k].oper, &ic->shadow_log[k], sizeof(MetaOper)) == 0)
holders++;
}
src/log.c
+4 -1
View File
@@ -1,3 +1,4 @@
#if 0
#if defined(MAIN_SIMULATION) || defined(MAIN_TEST)
#define QUAKEY_ENABLE_MOCKS
#endif
@@ -52,4 +53,6 @@ int log_append(Log *log, LogEntry entry)
log->entries[log->count++] = entry;
return 0;
}
}
#endif
src/log.h
+3 -1
View File
@@ -1,3 +1,4 @@
#if 0
#ifndef LOG_INCLUDED
#define LOG_INCLUDED
@@ -27,4 +28,5 @@ void log_free(Log *log);
void log_move(Log *dst, Log *src);
int log_append(Log *log, LogEntry entry);
#endif // LOG_INCLUDED
#endif // LOG_INCLUDED
#endif
src/server.c
+95 -102
View File
@@ -11,6 +11,7 @@
typedef enum {
HR_OK,
HR_OUT_OF_MEMORY,
HR_IO_FAILURE,
HR_INVALID_MESSAGE,
} HandlerResult;
@@ -58,7 +59,7 @@ static void node_log_impl(ServerState *state, const char *event, const char *det
status_name(state->status),
state->view_number,
state->commit_index,
state->log.count,
state->wal.count,
event,
detail ? detail : "");
}
@@ -145,10 +146,10 @@ static void begin_state_transfer(ServerState *state, int sender_idx)
.length = sizeof(GetStateMessage),
},
.view_number = state->view_number,
.op_number = state->log.count,
.op_number = state->wal.count,
.sender_idx = self_idx(state),
};
node_log(state, "SEND GET_STATE", "to=%d op=%d", sender_idx, state->log.count);
node_log(state, "SEND GET_STATE", "to=%d op=%d", sender_idx, state->wal.count);
send_to_peer(state, sender_idx, &message.base);
state->state_transfer_time = state->now;
@@ -266,14 +267,14 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
}
}
LogEntry log_entry = {
WALEntry wal_entry = {
.oper = request_message.oper,
.votes = 1 << self_idx(state),
.view_number = state->view_number,
.client_id = request_message.client_id,
.request_id = request_message.request_id,
};
if (log_append(&state->log, log_entry) < 0)
if (wal_append(&state->wal, wal_entry) < 0)
return HR_OUT_OF_MEMORY;
// We forwarded the message to all peers. As soon as
@@ -287,7 +288,7 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
},
.oper = request_message.oper,
.sender_idx = self_idx(state),
.log_index = state->log.count-1,
.log_index = state->wal.count-1,
.commit_index = state->commit_index,
.view_number = state->view_number,
.client_id = request_message.client_id,
@@ -296,7 +297,7 @@ process_request(ServerState *state, int conn_idx, ByteView msg)
{
char oper_buf[128];
meta_snprint_oper(oper_buf, sizeof(oper_buf), &request_message.oper);
node_log(state, "SEND PREPARE", "to=* idx=%d %s", state->log.count-1, oper_buf);
node_log(state, "SEND PREPARE", "to=* idx=%d %s", state->wal.count-1, oper_buf);
}
broadcast_to_peers(state, &prepare_message.base);
return HR_OK;
@@ -335,11 +336,11 @@ static void reply_to_client(ServerState *state, ClientTableEntry *table_entry,
static void advance_commit_index(ServerState *state, int target_index, bool send_replies)
{
target_index = MIN(target_index, state->log.count);
target_index = MIN(target_index, state->wal.count);
while (state->commit_index < target_index) {
LogEntry *entry = &state->log.entries[state->commit_index++];
WALEntry *entry = &state->wal.entries[state->commit_index++];
MetaResult result = meta_store_update(&state->metastore, &entry->oper);
{
@@ -431,12 +432,12 @@ process_prepare_ok(ServerState *state, int conn_idx, ByteView msg)
}
assert(message.log_index > -1);
assert(message.log_index < state->log.count);
assert(message.log_index < state->wal.count);
if (message.log_index < state->commit_index)
return HR_OK; // Already processed
LogEntry *entry = &state->log.entries[message.log_index];
WALEntry *entry = &state->wal.entries[message.log_index];
add_vote(&entry->votes, message.sender_idx);
if (reached_quorum(state, entry->votes)) {
node_log(state, "QUORUM", "idx=%d %s/%s", message.log_index, entry->oper.bucket, entry->oper.key);
@@ -467,8 +468,8 @@ clear_view_change_fields(ServerState *state)
state->view_change_apply_votes = 0;
state->view_change_old_view = 0;
state->view_change_commit = 0;
log_free(&state->view_change_log);
log_init(&state->view_change_log);
wal_free(&state->view_change_log);
wal_init(&state->view_change_log);
}
static HandlerResult
@@ -476,7 +477,10 @@ complete_view_change_and_become_primary(ServerState *state)
{
assert(state->commit_index <= state->view_change_commit);
log_move(&state->log, &state->view_change_log);
if (wal_replace(&state->wal, state->view_change_log.entries, state->view_change_log.count) < 0)
return HR_IO_FAILURE;
wal_free(&state->view_change_log);
wal_init(&state->view_change_log);
state->status = STATUS_NORMAL;
state->last_normal_view = state->view_number;
@@ -494,9 +498,9 @@ complete_view_change_and_become_primary(ServerState *state)
// 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++) {
for (int i = state->commit_index; i < state->wal.count; i++) {
LogEntry *entry = &state->log.entries[i];
WALEntry *entry = &state->wal.entries[i];
entry->votes = 0;
add_vote(&entry->votes, self_idx(state));
@@ -506,15 +510,15 @@ complete_view_change_and_become_primary(ServerState *state)
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_BEGIN_VIEW,
.length = sizeof(BeginViewMessage) + state->log.count * sizeof(LogEntry),
.length = sizeof(BeginViewMessage) + state->wal.count * sizeof(WALEntry),
},
.view_number = state->view_number,
.commit_index = state->commit_index,
.op_number = state->log.count,
.op_number = state->wal.count,
};
node_log(state, "SEND BEGIN_VIEW", "to=* view=%lu log=%d commit=%d",
state->view_number, state->log.count, state->commit_index);
broadcast_to_peers_ex(state, &begin_view_message.base, state->log.entries, state->log.count * sizeof(LogEntry));
state->view_number, state->wal.count, state->commit_index);
broadcast_to_peers_ex(state, &begin_view_message.base, state->wal.entries, state->wal.count * sizeof(WALEntry));
clear_view_change_fields(state);
return HR_OK;
@@ -552,15 +556,15 @@ process_do_view_change(ServerState *state, int conn_idx, ByteView msg)
state->view_change_old_view = message.old_view_number;
LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(DoViewChangeMessage));
WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(DoViewChangeMessage));
// Parse the variable-sized log from the message
int num_entries = (msg.len - sizeof(DoViewChangeMessage)) / sizeof(LogEntry);
int num_entries = (msg.len - sizeof(DoViewChangeMessage)) / sizeof(WALEntry);
if (num_entries != message.op_number)
return HR_INVALID_MESSAGE; // Message size mismatch
log_free(&state->view_change_log);
if (log_init_from_network(&state->view_change_log, entries, num_entries) < 0)
wal_free(&state->view_change_log);
if (wal_init_from_network(&state->view_change_log, entries, num_entries) < 0)
return HR_OUT_OF_MEMORY;
}
@@ -599,15 +603,15 @@ process_recovery(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(RecoveryResponseMessage),
},
.view_number = state->view_number,
.op_number = state->log.count-1, // TODO: What if the log is empty?
.op_number = state->wal.count-1, // TODO: What if the log is empty?
.nonce = recovery_message.nonce,
.commit_index = state->commit_index,
.sender_idx = self_idx(state),
};
if (is_primary(state)) {
recovery_response_message.base.length += state->log.count * sizeof(LogEntry);
recovery_response_message.base.length += state->wal.count * sizeof(WALEntry);
send_to_peer_ex(state, recovery_message.sender_idx, &recovery_response_message.base,
state->log.entries, state->log.count * sizeof(LogEntry));
state->wal.entries, state->wal.count * sizeof(WALEntry));
} else {
send_to_peer(state, recovery_message.sender_idx, &recovery_response_message.base);
}
@@ -627,7 +631,7 @@ perform_log_transfer_for_view_change(ServerState *state)
state->view_change_commit = state->commit_index;
// TODO: This should use copy-on-write
if (log_init_from_network(&state->view_change_log, state->log.entries, state->log.count) < 0)
if (wal_init_from_network(&state->view_change_log, state->wal.entries, state->wal.count) < 0)
return HR_OUT_OF_MEMORY;
} else {
@@ -635,18 +639,18 @@ perform_log_transfer_for_view_change(ServerState *state)
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_DO_VIEW_CHANGE,
.length = sizeof(DoViewChangeMessage) + state->log.count * sizeof(LogEntry),
.length = sizeof(DoViewChangeMessage) + state->wal.count * sizeof(WALEntry),
},
.view_number = state->view_number,
.old_view_number = state->last_normal_view,
.op_number = state->log.count,
.op_number = state->wal.count,
.commit_index = state->commit_index,
.sender_idx = self_idx(state),
};
send_to_peer_ex(state, primary_idx(state), &do_view_change_message.base, state->log.entries, state->log.count * sizeof(LogEntry));
send_to_peer_ex(state, primary_idx(state), &do_view_change_message.base, state->wal.entries, state->wal.count * sizeof(WALEntry));
node_log(state, "SEND DO_VIEW_CHANGE", "to=%d view=%lu old_view=%lu log=%d commit=%d",
primary_idx(state), state->view_number, state->last_normal_view,
state->log.count, state->commit_index);
state->wal.count, state->commit_index);
}
// Clear the future array since we're changing views
@@ -740,20 +744,20 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
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);
int num_entries = (msg.len - sizeof(BeginViewMessage)) / sizeof(WALEntry);
assert(num_entries >= state->commit_index);
// Replace the local log with the authoritative log from the primary
log_free(&state->log);
if (log_init_from_network(&state->log, msg.ptr + sizeof(BeginViewMessage), num_entries) < 0)
return HR_OUT_OF_MEMORY;
// 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;
// If there are non-committed operations in the log,
// send a PREPAREOK to the new primary
if (state->log.count > message.commit_index) {
if (state->wal.count > message.commit_index) {
PrepareOKMessage ok_msg = {
.base = {
.version = MESSAGE_VERSION,
@@ -761,12 +765,12 @@ process_begin_view(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(PrepareOKMessage),
},
.sender_idx = self_idx(state),
.log_index = state->log.count - 1,
.log_index = state->wal.count - 1,
.view_number = state->view_number,
};
send_to_peer(state, primary_idx(state), &ok_msg.base);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1,
state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->wal.count - 1,
state->wal.entries[state->wal.count - 1].oper.bucket, state->wal.entries[state->wal.count - 1].oper.key);
}
advance_commit_index(state, message.commit_index, false);
@@ -798,16 +802,16 @@ process_get_state(ServerState *state, int conn_idx, ByteView msg)
// Compute the suffix of log entries the requester is missing
int start = get_state_message.op_number;
if (start < 0 || start >= state->log.count)
if (start < 0 || start >= state->wal.count)
return HR_OK; // Nothing to send
int num_entries = state->log.count - start;
int num_entries = state->wal.count - start;
NewStateMessage new_state_message = {
.base = {
.version = MESSAGE_VERSION,
.type = MESSAGE_TYPE_NEW_STATE,
.length = sizeof(NewStateMessage) + num_entries * sizeof(LogEntry),
.length = sizeof(NewStateMessage) + num_entries * sizeof(WALEntry),
},
.view_number = state->view_number,
.op_number = num_entries,
@@ -817,7 +821,7 @@ process_get_state(ServerState *state, int conn_idx, ByteView msg)
node_log(state, "SEND NEW_STATE", "to=%d entries=%d commit=%d",
get_state_message.sender_idx, num_entries, state->commit_index);
send_to_peer_ex(state, get_state_message.sender_idx, &new_state_message.base,
state->log.entries + start, num_entries * sizeof(LogEntry));
state->wal.entries + start, num_entries * sizeof(WALEntry));
return HR_OK;
}
@@ -1040,7 +1044,10 @@ complete_recovery(ServerState *state)
assert(state->commit_index <= state->recovery_commit);
state->view_number = state->recovery_view;
log_move(&state->log, &state->recovery_log);
if (wal_replace(&state->wal, state->recovery_log.entries, state->recovery_log.count) < 0)
return HR_IO_FAILURE;
wal_free(&state->recovery_log);
wal_init(&state->recovery_log);
advance_commit_index(state, state->recovery_commit, false);
state->status = STATUS_NORMAL;
@@ -1050,8 +1057,8 @@ complete_recovery(ServerState *state)
// Reset stale votes
if (is_primary(state)) {
for (int i = state->commit_index; i < state->log.count; i++) {
LogEntry *entry = &state->log.entries[i];
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));
}
@@ -1124,13 +1131,13 @@ process_recovery_response(ServerState *state, ByteView msg)
if (should_store_recovery_log(state, message)) {
LogEntry *entries = (LogEntry*) (msg.ptr + sizeof(RecoveryResponseMessage));
WALEntry *entries = (WALEntry*) (msg.ptr + sizeof(RecoveryResponseMessage));
int num_entries = message.op_number + 1;
assert(num_entries == (int) ((msg.len - sizeof(RecoveryResponseMessage)) / sizeof(LogEntry)));
assert(num_entries == (int) ((msg.len - sizeof(RecoveryResponseMessage)) / sizeof(WALEntry)));
log_free(&state->recovery_log);
if (log_init_from_network(&state->recovery_log, entries, num_entries) < 0)
wal_free(&state->recovery_log);
if (wal_init_from_network(&state->recovery_log, entries, num_entries) < 0)
return HR_OUT_OF_MEMORY;
state->recovery_log_view = message.view_number;
@@ -1151,20 +1158,20 @@ process_single_future_list_entry(ServerState *state)
{
// Look for an entry with the current log index
int i = 0;
while (i < state->num_future && state->future[i].log_index != state->log.count)
while (i < state->num_future && state->future[i].log_index != state->wal.count)
i++;
if (i == state->num_future)
return 0; // No entry
LogEntry entry = {
WALEntry entry = {
.oper = state->future[i].oper,
.votes = 0,
.view_number = state->view_number,
.client_id = state->future[i].client_id,
.request_id = state->future[i].request_id,
};
if (log_append(&state->log, entry) < 0)
if (wal_append(&state->wal, entry) < 0)
return -1;
PrepareOKMessage message = {
@@ -1174,7 +1181,7 @@ process_single_future_list_entry(ServerState *state)
.length = sizeof(PrepareOKMessage),
},
.sender_idx = self_idx(state),
.log_index = state->log.count-1,
.log_index = state->wal.count-1,
.view_number = state->view_number,
};
send_to_peer(state, state->future[i].sender_idx, &message.base);
@@ -1185,7 +1192,7 @@ static void
remove_old_future_list_entries(ServerState *state)
{
for (int i = 0; i < state->num_future; i++) {
if (state->future[i].log_index < state->log.count) {
if (state->future[i].log_index < state->wal.count) {
state->future[i--] = state->future[--state->num_future];
}
}
@@ -1234,24 +1241,24 @@ process_prepare(ServerState *state, ByteView msg)
return HR_OK;
}
if (message.log_index < state->log.count)
if (message.log_index < state->wal.count)
return HR_OK; // Message refers to an old entry. Ignore.
if (message.log_index > state->log.count) {
if (message.log_index > state->wal.count) {
if (state->num_future < FUTURE_LIMIT)
state->future[state->num_future++] = message;
begin_state_transfer(state, message.sender_idx);
return HR_OK;
}
LogEntry log_entry = {
WALEntry wal_entry = {
.oper = message.oper,
.votes = 0,
.view_number = state->view_number,
.client_id = message.client_id,
.request_id = message.request_id,
};
if (log_append(&state->log, log_entry) < 0)
if (wal_append(&state->wal, wal_entry) < 0)
return HR_OUT_OF_MEMORY;
PrepareOKMessage ok_message = {
@@ -1261,12 +1268,12 @@ process_prepare(ServerState *state, ByteView msg)
.length = sizeof(PrepareOKMessage),
},
.sender_idx = self_idx(state),
.log_index = state->log.count-1,
.log_index = state->wal.count-1,
.view_number = state->view_number,
};
send_to_peer(state, message.sender_idx, &ok_message.base);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s",
message.sender_idx, state->log.count-1, message.oper.bucket, message.oper.key);
message.sender_idx, state->wal.count-1, message.oper.bucket, message.oper.key);
process_future_list(state);
advance_commit_index(state, message.commit_index, false);
@@ -1318,7 +1325,7 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
if (new_state_message.view_number != state->view_number)
return HR_OK;
int num_entries = (msg.len - sizeof(NewStateMessage)) / sizeof(LogEntry);
int num_entries = (msg.len - sizeof(NewStateMessage)) / sizeof(WALEntry);
if (num_entries != new_state_message.op_number)
return HR_INVALID_MESSAGE;
@@ -1328,22 +1335,22 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
// Append received entries to our log.
// The entries array is a suffix of the sender's log starting at
// global position start_index. We skip entries we already have.
LogEntry *entries = (LogEntry *)((uint8_t *)msg.ptr + sizeof(NewStateMessage));
WALEntry *entries = (WALEntry *)((uint8_t *)msg.ptr + sizeof(NewStateMessage));
int start_index = new_state_message.start_index;
for (int i = 0; i < num_entries; i++) {
int global_idx = start_index + i;
if (global_idx < state->log.count)
if (global_idx < state->wal.count)
continue; // Already have this entry
LogEntry entry = {
WALEntry entry = {
.oper = entries[i].oper,
.votes = 0,
.view_number = state->view_number,
.client_id = entries[i].client_id,
.request_id = entries[i].request_id,
};
if (log_append(&state->log, entry) < 0)
if (wal_append(&state->wal, entry) < 0)
return HR_OUT_OF_MEMORY;
// Send PREPARE_OK for each appended entry
@@ -1354,12 +1361,12 @@ process_new_state(ServerState *state, int conn_idx, ByteView msg)
.length = sizeof(PrepareOKMessage),
},
.sender_idx = self_idx(state),
.log_index = state->log.count - 1,
.log_index = state->wal.count - 1,
.view_number = state->view_number,
};
send_to_peer(state, primary_idx(state), &prepare_ok_message.base);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->log.count - 1,
state->log.entries[state->log.count - 1].oper.bucket, state->log.entries[state->log.count - 1].oper.key);
node_log(state, "SEND PREPARE_OK", "to=%d idx=%d %s/%s", primary_idx(state), state->wal.count - 1,
state->wal.entries[state->wal.count - 1].oper.bucket, state->wal.entries[state->wal.count - 1].oper.key);
}
process_future_list(state);
@@ -1665,27 +1672,23 @@ int server_init(void *state_, int argc, char **argv,
state->view_change_apply_votes = 0;
state->view_change_old_view = 0;
state->view_change_commit = 0;
log_init(&state->view_change_log);
wal_init(&state->view_change_log);
// Recovery state
state->recovery_votes = 0;
state->recovery_commit = 0;
state->recovery_view = 0;
state->recovery_log_view = 0;
log_init(&state->recovery_log);
wal_init(&state->recovery_log);
// Detect whether this is a restart after a crash by checking for a
// boot marker file on disk. The disk persists across crashes, so if
// the marker exists, this node previously ran and crashed. In that
// case, enter recovery mode to learn the current view from peers
// before participating in the protocol.
//
// We use open() directly (without O_CREAT) instead of file_exists()
// because access() is not available in the simulation environment.
int marker_fd = open("vsr_boot_marker", O_RDONLY, 0);
bool previously_crashed = (marker_fd >= 0);
if (previously_crashed)
close(marker_fd);
// Open the WAL file. If it contains entries from a previous run,
// this node crashed and must enter recovery mode to learn the
// current view from peers before participating in the protocol.
if (wal_init_from_file(&state->wal, S("vsr.log")) < 0) {
fprintf(stderr, "Node :: Couldn't open WAL file\n");
return -1;
}
bool previously_crashed = (state->wal.count > 0);
if (previously_crashed) {
state->status = STATUS_RECOVERY;
state->recovery_nonce = now;
@@ -1693,7 +1696,6 @@ int server_init(void *state_, int argc, char **argv,
} else {
state->status = STATUS_NORMAL;
}
log_init(&state->log); // Initialize early so node_log can read log.count
node_log(state, "INIT", "nodes=%d%s", state->num_nodes, previously_crashed ? " (recovering)" : "");
client_table_init(&state->client_table);
@@ -1717,15 +1719,6 @@ int server_init(void *state_, int argc, char **argv,
return -1;
}
// Write the boot marker to disk so that future restarts can detect
// 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);
@@ -1793,7 +1786,7 @@ int server_tick(void *state_, void **ctxs,
tcp_close(&state->tcp, conn_idx);
break;
}
if (hret == HR_OUT_OF_MEMORY)
if (hret == HR_OUT_OF_MEMORY || hret == HR_IO_FAILURE)
return -1;
assert(hret == HR_OK);
@@ -1932,7 +1925,7 @@ int server_tick(void *state_, void **ctxs,
Time st_deadline = state->state_transfer_time + STATE_TRANSFER_TIMEOUT_SEC * 1000000000ULL;
if (st_deadline <= state->now) {
node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->log.count);
node_log(state, "TIMEOUT STATE_TRANSFER", "op=%d", state->wal.count);
GetStateMessage get_state_message = {
.base = {
@@ -1941,11 +1934,11 @@ int server_tick(void *state_, void **ctxs,
.length = sizeof(GetStateMessage),
},
.view_number = state->view_number,
.op_number = state->log.count,
.op_number = state->wal.count,
.sender_idx = self_idx(state),
};
send_to_peer(state, primary_idx(state), &get_state_message.base);
node_log(state, "SEND GET_STATE", "to=%d op=%d", primary_idx(state), state->log.count);
node_log(state, "SEND GET_STATE", "to=%d op=%d", primary_idx(state), state->wal.count);
state->state_transfer_time = state->now;
@@ -1967,9 +1960,9 @@ int server_free(void *state_)
node_log_simple(state, "CRASHED");
log_free(&state->log);
log_free(&state->recovery_log);
log_free(&state->view_change_log);
wal_free(&state->wal);
wal_free(&state->recovery_log);
wal_free(&state->view_change_log);
tcp_context_free(&state->tcp);
client_table_free(&state->client_table);
meta_store_free(&state->metastore);
src/server.h
+8 -8
View File
@@ -4,7 +4,7 @@
#include "tcp.h"
#include "basic.h"
#include "message.h"
#include "log.h"
#include "wal.h"
#include "config.h"
#include "metadata.h"
#include "chunk_store.h"
@@ -99,15 +99,15 @@ typedef struct {
int op_number; // Number of entries in the log
int commit_index;
int sender_idx;
// Followed by: LogEntry log[op_number]
// Followed by: WALEntry log[op_number]
} DoViewChangeMessage;
typedef struct {
MessageHeader base;
uint64_t view_number;
int commit_index;
int op_number; // Number of log entries that follow
// Followed by: LogEntry log[op_number]
int op_number; // Number of WAL entries that follow
// Followed by: WALEntry log[op_number]
} BeginViewMessage;
typedef struct {
@@ -138,7 +138,7 @@ typedef struct {
int op_number; // Number of log entries that follow
int commit_index;
int start_index; // Global log index of the first entry in the suffix
// Followed by: LogEntry log[op_number]
// Followed by: WALEntry log[op_number]
} NewStateMessage;
typedef struct {
@@ -228,7 +228,7 @@ typedef struct {
uint32_t recovery_votes;
uint64_t recovery_nonce;
uint64_t recovery_view;
Log recovery_log;
WAL recovery_log;
uint64_t recovery_log_view;
Time recovery_time;
int recovery_commit;
@@ -238,7 +238,7 @@ typedef struct {
uint32_t view_change_begin_votes;
uint32_t view_change_apply_votes;
Log view_change_log; // Best log seen
WAL view_change_log; // Best log seen
uint64_t view_change_old_view; // Best old_view_number seen in DoViewChange
int view_change_commit; // Best commit_index seen
@@ -255,7 +255,7 @@ typedef struct {
bool state_transfer_pending;
Time state_transfer_time;
Log log;
WAL wal;
Time heartbeat;
src/wal.c
+303
View File
@@ -0,0 +1,303 @@
#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 WALEntryDisk fields except the checksum itself.
static uint32_t wal_entry_checksum(WALEntryDisk *entry)
{
uint32_t h = 2166136261u;
const unsigned char *p = (const unsigned char *)entry;
size_t len = offsetof(WALEntryDisk, checksum);
for (size_t i = 0; i < len; i++) {
h ^= p[i];
h *= 16777619u;
}
return h;
}
static WALEntryDisk wal_entry_to_disk(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,
};
}
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)
{
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(WALEntryDisk);
size_t valid_size = raw_count * sizeof(WALEntryDisk);
if (valid_size < size)
file_truncate(handle, valid_size);
WALEntryDisk *disk_entries = malloc(raw_count * sizeof(WALEntryDisk));
if (disk_entries == NULL && raw_count > 0) {
file_close(handle);
return -1;
}
if (file_set_offset(handle, 0) < 0) {
file_close(handle);
free(disk_entries);
return -1;
}
if (raw_count > 0 && file_read_exact(handle, (char *)disk_entries, raw_count * sizeof(WALEntryDisk)) < 0) {
file_close(handle);
free(disk_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])) {
count = i;
file_truncate(handle, count * sizeof(WALEntryDisk));
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) {
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 = 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)) {
WALEntryDisk disk = wal_entry_to_disk(&entry);
if (file_write_exact(wal->handle, (char *)&disk, sizeof(disk)) < 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));
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));
return -1;
}
}
wal->entries[wal->count++] = entry;
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(WALEntryDisk)) < 0)
return -1;
if (file_set_offset(wal->handle, new_count * sizeof(WALEntryDisk)) < 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++) {
WALEntryDisk disk = wal_entry_to_disk(&entries[i]);
if (file_write_exact(tmp, (char *)&disk, sizeof(disk)) < 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;
// 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(WALEntryDisk)) < 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];
}
src/wal.h
+65
View File
@@ -0,0 +1,65 @@
#ifndef WAL_INCLUDED
#define WAL_INCLUDED
#include "metadata.h"
#include "file_system.h"
#include "config.h"
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;
int view_number;
uint64_t client_id;
uint64_t request_id;
uint32_t checksum; // FNV-1a over all preceding fields
} WALEntryDisk;
_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.
int wal_init_from_file(WAL *wal, string file);
// 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);
// 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);
#endif // WAL_INCLUDED