Skip to content

feat(network): announce send chunks as an all-or-nothing batch #53

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
andinux merged 1 commit into from
Jul 3, 2026
Merged

feat(network): announce send chunks as an all-or-nothing batch #53

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 33 additions & 26 deletions src/network/network.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1499,23 +1499,26 @@ 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);
return NULL;
}

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;
Expand All @@ -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) {
Expand All @@ -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) {
Expand All @@ -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);
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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) {
Expand All @@ -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) {
Expand All @@ -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);
Expand Down
9 changes: 1 addition & 8 deletions src/network/network_private.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
55 changes: 34 additions & 21 deletions test/network_unit.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
#include <stdbool.h>
#include <stdint.h>
#include <inttypes.h>
#include "utils.h"
#include "network_private.h"

static int failures = 0;
Expand Down Expand Up @@ -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;
}
Expand All @@ -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");
Expand Down
Loading