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

@@ -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
+    // Read generation counter
-    uint32_t item_count;
+    uint64_t gen;
-    if (!binary_read(&reader, &item_count, sizeof(item_count))) {
+    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;
         }

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;
+    uint64_t expect_gen;
-    // TODO: read expected generation
+    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;
-    uint64_t expect_gen = NO_GENERATION;
+    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  hash;
-        SHA256  new_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
+        SHA256 hash;
-        if (!binary_read(&reader, &old_hash, sizeof(old_hash)))
+        if (!binary_read(&reader, &hash, sizeof(hash)))
             return -1;
 
-        SHA256 new_hash;
+        results[i].hash = 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].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;