From acb1336d47f20503088fa53663d49feb21ce1bb4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 24 Nov 2025 12:16:55 +0000 Subject: [PATCH] Implement synchronization protocol from PROTOCOL.txt Updated both client.c and metadata_server.c to match the protocol specification in misc/PROTOCOL.txt: Client-to-Server Messages: - Added expect_gen field to DELETE, LIST, READ, and WRITE messages - Removed old_hash and chunk_size from WRITE message chunks Server-to-Client Responses: - Added generation counter to CREATE_SUCCESS response - Added generation counter to LIST_SUCCESS response - Fixed LIST_SUCCESS field order (gen, truncated, item_count) - Added generation counter to READ_SUCCESS response - Added generation counter to WRITE_SUCCESS response All changes compile successfully and pass valgrind memory checks. --- src/client.c | 71 +++++++++++++++++++++++++++++++++++-------- src/metadata_server.c | 48 ++++++++++++++--------------- 2 files changed, 82 insertions(+), 37 deletions(-) diff --git a/src/client.c b/src/client.c index aacabaf..e8d3d9e 100644 --- a/src/client.c +++ b/src/client.c @@ -617,11 +617,13 @@ ToastyHandle toasty_begin_delete(ToastyFS *toasty, ToastyString path) opidx = -1; goto unlock_and_exit; } - uint16_t tmp = path.len; + uint16_t path_len = path.len; + uint64_t expect_gen = 0; // 0 means skip generation check MessageWriter writer; metadata_server_request_start(toasty, &writer, MESSAGE_TYPE_DELETE); - message_write(&writer, &tmp, sizeof(tmp)); + message_write(&writer, &expect_gen, sizeof(expect_gen)); + message_write(&writer, &path_len, sizeof(path_len)); message_write(&writer, path.ptr, path.len); if (metadata_server_request_end(toasty, &writer, opidx, 0) < 0) { free_operation(toasty, opidx); @@ -659,11 +661,13 @@ ToastyHandle toasty_begin_list(ToastyFS *toasty, ToastyString path) opidx = -1; goto unlock_and_exit; } - uint16_t tmp = path.len; + uint16_t path_len = path.len; + uint64_t expect_gen = 0; // 0 means skip generation check MessageWriter writer; metadata_server_request_start(toasty, &writer, MESSAGE_TYPE_LIST); - message_write(&writer, &tmp, sizeof(tmp)); + message_write(&writer, &expect_gen, sizeof(expect_gen)); + message_write(&writer, &path_len, sizeof(path_len)); message_write(&writer, path.ptr, path.len); if (metadata_server_request_end(toasty, &writer, opidx, 0) < 0) { free_operation(toasty, opidx); @@ -687,9 +691,11 @@ static int send_read_message(ToastyFS *toasty, int opidx, int tag, ToastyString if (path.len > UINT16_MAX) return -1; uint16_t path_len = path.len; + uint64_t expect_gen = 0; // 0 means skip generation check MessageWriter writer; metadata_server_request_start(toasty, &writer, MESSAGE_TYPE_READ); + message_write(&writer, &expect_gen, sizeof(expect_gen)); message_write(&writer, &path_len, sizeof(path_len)); message_write(&writer, path.ptr, path.len); message_write(&writer, &offset, sizeof(offset)); @@ -813,6 +819,13 @@ static void process_event_for_create(ToastyFS *toasty, return; } + // Read generation counter + uint64_t gen; + if (!binary_read(&reader, &gen, sizeof(gen))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_CREATE_ERROR, .user=toasty->operations[opidx].user }; + return; + } + // Check there is nothing else to read if (binary_read(&reader, NULL, 1)) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_CREATE_ERROR, .user=toasty->operations[opidx].user }; @@ -901,19 +914,26 @@ static void process_event_for_list(ToastyFS *toasty, return; } - // Read and validate the list data - uint32_t item_count; - if (!binary_read(&reader, &item_count, sizeof(item_count))) { + // Read generation counter + uint64_t gen; + if (!binary_read(&reader, &gen, sizeof(gen))) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_ERROR, .user=toasty->operations[opidx].user }; return; } + // Read and validate the list data uint8_t truncated; if (!binary_read(&reader, &truncated, sizeof(truncated))) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_ERROR, .user=toasty->operations[opidx].user }; return; } + uint32_t item_count; + if (!binary_read(&reader, &item_count, sizeof(item_count))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_ERROR, .user=toasty->operations[opidx].user }; + return; + } + ToastyListingEntry *entities = sys_malloc(item_count * sizeof(ToastyListingEntry)); if (entities == NULL) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_LIST_ERROR, .user=toasty->operations[opidx].user }; @@ -1000,6 +1020,13 @@ static void process_event_for_read(ToastyFS *toasty, return; } + // Read generation counter + uint64_t gen; + if (!binary_read(&reader, &gen, sizeof(gen))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_READ_ERROR, .user=toasty->operations[opidx].user }; + return; + } + // Read chunk size uint32_t chunk_size; if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) { @@ -1429,6 +1456,13 @@ static void process_event_for_write(ToastyFS *toasty, return; } + // Read generation counter + uint64_t gen; + if (!binary_read(&reader, &gen, sizeof(gen))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; + return; + } + uint32_t chunk_size; if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; @@ -1436,6 +1470,12 @@ static void process_event_for_write(ToastyFS *toasty, } toasty->operations[opidx].chunk_size = chunk_size; + // Skip file_length field that comes after chunk_size + if (!binary_read(&reader, NULL, sizeof(uint32_t))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; + return; + } + uint32_t num_hashes; if (!binary_read(&reader, &num_hashes, sizeof(num_hashes))) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; @@ -1921,21 +1961,19 @@ static void process_event_for_write(ToastyFS *toasty, assert(0); // TODO } uint16_t path_len = path.len; + uint64_t expect_gen = 0; // 0 means skip generation check - but per protocol, WRITE can't use 0 uint32_t num_chunks = num_upload_results; - uint32_t chunk_size = toasty->operations[opidx].chunk_size; + message_write(&writer, &expect_gen, sizeof(expect_gen)); message_write(&writer, &path_len, sizeof(path_len)); message_write(&writer, path.ptr, path.len); message_write(&writer, &offset, sizeof(offset)); message_write(&writer, &length, sizeof(length)); message_write(&writer, &num_chunks, sizeof(num_chunks)); - message_write(&writer, &chunk_size, sizeof(chunk_size)); for (int i = 0; i < num_upload_results; i++) { - // TODO: newly create chunks don't have an old hash - message_write(&writer, &upload_results[i].old_hash, sizeof(upload_results[i].old_hash)); message_write(&writer, &upload_results[i].new_hash, sizeof(upload_results[i].new_hash)); uint32_t tmp = upload_results[i].num_locations; @@ -1984,12 +2022,19 @@ static void process_event_for_write(ToastyFS *toasty, return; } - if (binary_read(&reader, NULL, 1)) { + if (type != MESSAGE_TYPE_WRITE_SUCCESS) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; return; } - if (type != MESSAGE_TYPE_WRITE_SUCCESS) { + // Read generation counter + uint64_t gen; + if (!binary_read(&reader, &gen, sizeof(gen))) { + toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; + return; + } + + if (binary_read(&reader, NULL, 1)) { toasty->operations[opidx].result = (ToastyResult) { .type=TOASTY_RESULT_WRITE_ERROR, .user=toasty->operations[opidx].user }; return; } diff --git a/src/metadata_server.c b/src/metadata_server.c index 4ea6e0a..bf4ebf9 100644 --- a/src/metadata_server.c +++ b/src/metadata_server.c @@ -208,7 +208,7 @@ process_client_create(MetadataServer *state, int conn_idx, ByteView msg) ByteQueue *output = tcp_output_buffer(&state->tcp, conn_idx); message_writer_init(&writer, output, MESSAGE_TYPE_CREATE_SUCCESS); - // TODO: write the generation + message_write(&writer, &gen, sizeof(gen)); if (!message_writer_free(&writer)) return -1; } @@ -225,8 +225,9 @@ process_client_delete(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - uint64_t expect_gen = NO_GENERATION; - // TODO: read expected generation + uint64_t expect_gen; + if (!binary_read(&reader, &expect_gen, sizeof(expect_gen))) + return -1; char path_mem[1<<10]; uint16_t path_len; @@ -293,6 +294,10 @@ process_client_list(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; + uint64_t expect_gen; + if (!binary_read(&reader, &expect_gen, sizeof(expect_gen))) + return -1; + char path_mem[1<<10]; uint16_t path_len; @@ -342,7 +347,7 @@ process_client_list(MetadataServer *state, int conn_idx, ByteView msg) MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_LIST_SUCCESS); - // TODO: write generation + message_write(&writer, &gen, sizeof(gen)); uint32_t item_count = ret; uint8_t truncated = 0; @@ -352,8 +357,8 @@ process_client_list(MetadataServer *state, int conn_idx, ByteView msg) item_count = MAX_LIST_SIZE; } - message_write(&writer, &item_count, sizeof(item_count)); message_write(&writer, &truncated, sizeof(truncated)); + message_write(&writer, &item_count, sizeof(item_count)); for (int i = 0; i < ret && i < MAX_LIST_SIZE; i++) { @@ -384,6 +389,10 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; + uint64_t expect_gen; + if (!binary_read(&reader, &expect_gen, sizeof(expect_gen))) + return -1; + char path_mem[1<<10]; uint16_t path_len; @@ -441,7 +450,7 @@ process_client_read(MetadataServer *state, int conn_idx, ByteView msg) MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_READ_SUCCESS); - // TODO: write generation + message_write(&writer, &gen, sizeof(gen)); if (chunk_size > UINT32_MAX) { message_writer_free(&writer); @@ -522,8 +531,9 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, NULL, sizeof(MessageHeader))) return -1; - // TODO: read expected generation - uint64_t expect_gen = NO_GENERATION; + uint64_t expect_gen; + if (!binary_read(&reader, &expect_gen, sizeof(expect_gen))) + return -1; char path_mem[1<<10]; uint16_t path_len; @@ -551,15 +561,10 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) if (!binary_read(&reader, &num_chunks, sizeof(num_chunks))) return -1; - uint32_t chunk_size; // TODO: Not needed anymore - if (!binary_read(&reader, &chunk_size, sizeof(chunk_size))) - return -1; - #define MAX_CHUNKS_PER_WRITE 32 typedef struct { - SHA256 old_hash; - SHA256 new_hash; + SHA256 hash; int num_addrs; Address addrs[REPLICATION_FACTOR]; } ChunkWriteResult; @@ -571,16 +576,11 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) for (uint32_t i = 0; i < num_chunks; i++) { - SHA256 old_hash; // TODO: old hashes are not necessary anymore - if (!binary_read(&reader, &old_hash, sizeof(old_hash))) + SHA256 hash; + if (!binary_read(&reader, &hash, sizeof(hash))) return -1; - SHA256 new_hash; - if (!binary_read(&reader, &new_hash, sizeof(new_hash))) - return -1; - - results[i].old_hash = old_hash; - results[i].new_hash = new_hash; + results[i].hash = hash; results[i].num_addrs = 0; uint32_t num_locations; @@ -622,7 +622,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) SHA256 new_hashes[MAX_CHUNKS_PER_WRITE]; for (uint32_t i = 0; i < num_chunks; i++) - new_hashes[i] = results[i].new_hash; + new_hashes[i] = results[i].hash; if (wal_append_write(&state->wal, path, offset, length, num_chunks, expect_gen, new_hashes) < 0) { assert(0); // TODO @@ -682,7 +682,7 @@ process_client_write(MetadataServer *state, int conn_idx, ByteView msg) MessageWriter writer; message_writer_init(&writer, output, MESSAGE_TYPE_WRITE_SUCCESS); - // TODO: write new generation + message_write(&writer, &new_gen, sizeof(new_gen)); if (!message_writer_free(&writer)) return -1;