From 775729d1d907c09fca96d6d2276029313282bead Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Fri, 3 Jul 2026 08:50:28 -0600 Subject: [PATCH] feat(network): announce send chunks as an all-or-nothing batch One send_changes call = one batch: every chunk announces the same global window [send_checkpoint+1 .. watermark] plus batchId (UUIDv7 per call), chunkIndex and isFinal, replacing the per-chunk announce-window tiling. The server can then advance optimistic only when the final chunk is received and confirm the whole window only when every chunk of the batch applied - a failed batch is never merged, so a partially applied db_version can no longer be masked as covered; the client re-sends the window under a new batchId. Co-Authored-By: Claude Fable 5 --- src/network/network.c | 59 ++++++++++++++++++++--------------- src/network/network_private.h | 9 +----- test/network_unit.c | 55 +++++++++++++++++++------------- 3 files changed, 68 insertions(+), 55 deletions(-) diff --git a/src/network/network.c b/src/network/network.c index 4ed61c4..038adff 100644 --- a/src/network/network.c +++ b/src/network/network.c @@ -1499,14 +1499,15 @@ static int network_apply_check_chunk(sqlite3_context *context, const char *chunk return rc; } -static char *network_apply_json_payload(const char *transport_key, const char *transport_value, - int64_t db_version_min, int64_t db_version_max) { - if (!transport_key || !transport_value) return NULL; +char *network_apply_json_payload(const char *transport_key, const char *transport_value, + int64_t db_version_min, int64_t db_version_max, + const char *batch_id, int chunk_index, bool is_final) { + if (!transport_key || !transport_value || !batch_id) return NULL; char *escaped_value = json_escape_string(transport_value); if (!escaped_value) return NULL; - size_t requested = strlen(transport_key) + strlen(escaped_value) + 128; + size_t requested = strlen(transport_key) + strlen(escaped_value) + strlen(batch_id) + 192; char *json_payload = cloudsync_memory_alloc((uint64_t)requested); if (!json_payload) { cloudsync_memory_free(escaped_value); @@ -1514,8 +1515,10 @@ static char *network_apply_json_payload(const char *transport_key, const char *t } snprintf(json_payload, requested, - "{\"%s\":\"%s\", \"dbVersionMin\":%" PRId64 ", \"dbVersionMax\":%" PRId64 "}", - transport_key, escaped_value, db_version_min, db_version_max); + "{\"%s\":\"%s\", \"dbVersionMin\":%" PRId64 ", \"dbVersionMax\":%" PRId64 + ", \"batchId\":\"%s\", \"chunkIndex\":%d, \"isFinal\":%s}", + transport_key, escaped_value, db_version_min, db_version_max, + batch_id, chunk_index, is_final ? "true" : "false"); cloudsync_memory_free(escaped_value); return json_payload; @@ -1524,6 +1527,7 @@ static char *network_apply_json_payload(const char *transport_key, const char *t static int network_send_payload_to_apply(sqlite3_context *context, network_data *netdata, const void *blob, int blob_size, int64_t db_version_min, int64_t db_version_max, + const char *batch_id, int chunk_index, bool is_final, NETWORK_RESULT *res_out) { memset(res_out, 0, sizeof(*res_out)); if (!blob || blob_size <= 0) { @@ -1533,11 +1537,15 @@ static int network_send_payload_to_apply(sqlite3_context *context, network_data #ifdef CLOUDSYNC_NETWORK_TRACE fprintf(stderr, - "[cloudsync-network] send_changes chunk_size=%d fast-lane:%s db_version_min=%" PRId64 " db_version_max=%" PRId64 "\n", + "[cloudsync-network] send_changes chunk_size=%d fast-lane:%s db_version_min=%" PRId64 " db_version_max=%" PRId64 + " batch_id=%s chunk_index=%d is_final=%d\n", blob_size, blob_size <= CLOUDSYNC_NETWORK_FAST_LANE_MAX_BLOB_SIZE ? "true" : "false", db_version_min, - db_version_max); + db_version_max, + batch_id, + chunk_index, + is_final ? 1 : 0); #endif if (blob_size <= CLOUDSYNC_NETWORK_FAST_LANE_MAX_BLOB_SIZE) { @@ -1548,7 +1556,8 @@ static int network_send_payload_to_apply(sqlite3_context *context, network_data return SQLITE_NOMEM; } - char *json_payload = network_apply_json_payload("blob", blob_base64, db_version_min, db_version_max); + char *json_payload = network_apply_json_payload("blob", blob_base64, db_version_min, db_version_max, + batch_id, chunk_index, is_final); cloudsync_memory_free(blob_base64); if (!json_payload) { sqlite3_result_error(context, "cloudsync_network_send_changes: unable to allocate apply request payload.", -1); @@ -1585,7 +1594,8 @@ static int network_send_payload_to_apply(sqlite3_context *context, network_data return SQLITE_ERROR; } - char *json_payload = network_apply_json_payload("url", s3_url, db_version_min, db_version_max); + char *json_payload = network_apply_json_payload("url", s3_url, db_version_min, db_version_max, + batch_id, chunk_index, is_final); cloudsync_memory_free(s3_url); if (!json_payload) { sqlite3_result_error(context, "cloudsync_network_send_changes: unable to allocate apply request payload.", -1); @@ -1706,7 +1716,7 @@ int cloudsync_network_send_changes_internal (sqlite3_context *context, int argc, sqlite3 *db = sqlite3_context_db_handle(context); sqlite3_stmt *stmt = NULL; const char *chunk_sql = - "SELECT payload, payload_size, db_version_min, db_version_max, watermark_db_version " + "SELECT payload, payload_size, watermark_db_version, is_final " "FROM cloudsync_payload_chunks WHERE since_db_version = ?"; int rc = sqlite3_prepare_v2(db, chunk_sql, -1, &stmt, NULL); if (rc != SQLITE_OK) { @@ -1725,33 +1735,31 @@ int cloudsync_network_send_changes_internal (sqlite3_context *context, int argc, bool sent_any = false; int sent_chunks = 0; // payload chunks sent this call int64_t sent_bytes = 0; // serialized payload bytes sent this call - // Lower bound of the coverage window announced to /apply. The server tracks - // per-site applied ranges from version 1, so the announced ranges must tile - // [send_checkpoint+1 .. watermark] contiguously; otherwise db_versions consumed - // without a local-site change (e.g. merged remote changes) read back as gaps. - int64_t announce_lo = db_version + 1; + // One send call = one all-or-nothing batch: every chunk announces the same + // global window [send_checkpoint+1 .. watermark] (holes from non-local-change + // db_versions included) plus batchId/chunkIndex/isFinal. The server advances + // optimistic on the final chunk and confirms the whole window only when every + // chunk of the batch applied; a failed batch is re-sent whole under a new id. + char batch_id[UUID_STR_MAXLEN]; + cloudsync_uuid_v7_string(batch_id, true); while ((rc = sqlite3_step(stmt)) == SQLITE_ROW) { const void *blob = sqlite3_column_blob(stmt, 0); int blob_size = sqlite3_column_bytes(stmt, 0); int64_t payload_size = sqlite3_column_int64(stmt, 1); - int64_t db_version_min = sqlite3_column_int64(stmt, 2); - int64_t db_version_max = sqlite3_column_int64(stmt, 3); - int64_t watermark = sqlite3_column_int64(stmt, 4); + int64_t watermark = sqlite3_column_int64(stmt, 2); + bool is_final = sqlite3_column_int(stmt, 3) != 0; if (!blob || blob_size <= 0 || payload_size != blob_size || payload_size > INT_MAX || - db_version_min <= 0 || db_version_max <= 0 || db_version_min > db_version_max) { + watermark <= db_version) { sqlite3_result_error(context, "cloudsync_network_send_changes: invalid payload chunk generated.", -1); rc = SQLITE_ERROR; goto cleanup; } - // Announce the covered window, not just where the data sits, so db_versions - // skipped by non-local-change transactions don't read back as server-side gaps. - int64_t announce_min = network_announce_min(announce_lo, db_version_min); - NETWORK_RESULT res = {0}; - rc = network_send_payload_to_apply(context, netdata, blob, blob_size, announce_min, db_version_max, &res); + rc = network_send_payload_to_apply(context, netdata, blob, blob_size, db_version + 1, watermark, + batch_id, sent_chunks, is_final, &res); if (rc != SQLITE_OK) goto cleanup; if (res.code == CLOUDSYNC_NETWORK_BUFFER && res.buffer) { @@ -1769,7 +1777,6 @@ int cloudsync_network_send_changes_internal (sqlite3_context *context, int argc, sent_chunks++; sent_bytes += payload_size; if (watermark > new_db_version) new_db_version = watermark; - announce_lo = db_version_max + 1; // next chunk continues the window here } if (rc != SQLITE_DONE) { sqlite3_result_error(context, sqlite3_errmsg(db), -1); diff --git a/src/network/network_private.h b/src/network/network_private.h index d541a8c..21a46fd 100644 --- a/src/network/network_private.h +++ b/src/network/network_private.h @@ -52,14 +52,7 @@ NETWORK_RESULT network_receive_buffer (network_data *data, const char *endpoint, // Exposed (non-static) for the network unit test; otherwise internal to network.c. void network_sync_state_update_from_response(NETWORK_RESULT *res, int64_t *last_optimistic_version, int64_t *last_confirmed_version, int *gaps_size, char **apply_failure_json, char **check_failure_json); const char *network_compute_status(int64_t last_optimistic, int64_t last_confirmed, int gaps_size, int64_t local_version); - -// Lower bound of the coverage window a send chunk announces to /apply: the running -// window start, but never above the chunk's own min (so consecutive chunks sharing a -// db_version — a value fragmented across chunks — keep min<=max). The caller advances -// window_lo to chunk_db_version_max+1 after each chunk so the ranges tile contiguously. -static inline int64_t network_announce_min(int64_t window_lo, int64_t chunk_db_version_min) { - return window_lo < chunk_db_version_min ? window_lo : chunk_db_version_min; -} +char *network_apply_json_payload(const char *transport_key, const char *transport_value, int64_t db_version_min, int64_t db_version_max, const char *batch_id, int chunk_index, bool is_final); #ifdef CLOUDSYNC_NETWORK_TRACE const char *network_trace_endpoint_name(network_data *data, const char *endpoint); diff --git a/test/network_unit.c b/test/network_unit.c index d9f4299..f4ccb27 100644 --- a/test/network_unit.c +++ b/test/network_unit.c @@ -13,6 +13,7 @@ #include #include #include +#include "utils.h" #include "network_private.h" static int failures = 0; @@ -78,28 +79,40 @@ static bool test_non_buffer_is_noop(void) { return optimistic == 7 && confirmed == 3 && gaps == 0; } -// The send announces a contiguous coverage window, not the raw chunk ranges, so -// skipped db_versions don't look like gaps. Walk the tiling exactly as the send loop -// does: announce [network_announce_min(lo, chunk_min) .. chunk_max], then lo = max+1. -static bool test_announce_window_tiling(void) { +// One send call = one all-or-nothing batch: every chunk of it announces the same +// global window [checkpoint+1 .. watermark] plus batchId/chunkIndex/isFinal, so the +// server confirms the whole window only when every chunk of the batch applied and a +// failed batch is re-sent whole under a new id. +static bool test_apply_json_payload_batch(void) { bool ok = true; - int64_t lo = 0 + 1; // send checkpoint (since) = 0 - // leading hole: a single change at db_version 7 must announce from 1, not 7. - ok = ok && network_announce_min(lo, 7) == 1; - lo = 7 + 1; - - // inter-chunk hole: next change at 20 (8..19 empty) announces from 8 → no gap. - ok = ok && network_announce_min(lo, 20) == 8; - lo = 20 + 1; - - // same-db_version fragments (a value split across two chunks): first fragment - // announces from the running lo, second must clamp to the shared version (30), - // never lo (31) — that would make min > max. - ok = ok && network_announce_min(lo, 30) == 21; // first fragment of v30 - lo = 30 + 1; - ok = ok && network_announce_min(lo, 30) == 30; // second fragment of v30: 30, not 31 - lo = 30 + 1; + char *j0 = network_apply_json_payload("blob", "QUJD", 1, 10, "batch-uuid-1", 0, false); + ok = ok && j0 != NULL; + if (j0) { + ok = ok && strstr(j0, "\"blob\":\"QUJD\"") != NULL; + ok = ok && strstr(j0, "\"dbVersionMin\":1") != NULL; + ok = ok && strstr(j0, "\"dbVersionMax\":10") != NULL; + ok = ok && strstr(j0, "\"batchId\":\"batch-uuid-1\"") != NULL; + ok = ok && strstr(j0, "\"chunkIndex\":0") != NULL; + ok = ok && strstr(j0, "\"isFinal\":false") != NULL; + cloudsync_memory_free(j0); + } + + // last chunk, url transport: same window, higher index, isFinal true + char *j1 = network_apply_json_payload("url", "https://s3/part", 1, 10, "batch-uuid-1", 3, true); + ok = ok && j1 != NULL; + if (j1) { + ok = ok && strstr(j1, "\"url\":\"https://s3/part\"") != NULL; + ok = ok && strstr(j1, "\"dbVersionMin\":1") != NULL; + ok = ok && strstr(j1, "\"dbVersionMax\":10") != NULL; + ok = ok && strstr(j1, "\"batchId\":\"batch-uuid-1\"") != NULL; + ok = ok && strstr(j1, "\"chunkIndex\":3") != NULL; + ok = ok && strstr(j1, "\"isFinal\":true") != NULL; + cloudsync_memory_free(j1); + } + + // a batch id is mandatory for the chunked send path + ok = ok && network_apply_json_payload("blob", "QUJD", 1, 10, NULL, 0, false) == NULL; return ok; } @@ -118,7 +131,7 @@ int main(void) { printf("\nNetwork unit tests\n"); check("optimistic/confirmed version folds latest-valid (allows rollback):", test_optimistic_version_rollback()); check("non-buffer response is a no-op:", test_non_buffer_is_noop()); - check("send announce-window tiling (hole / inter-chunk / fragment):", test_announce_window_tiling()); + check("send batch /apply payload (window / batchId / chunkIndex / isFinal):", test_apply_json_payload_batch()); check("network_compute_status:", test_compute_status()); if (failures) { printf("\n%d test(s) FAILED\n", failures); return 1; } printf("\nAll network unit tests passed\n");