fix: gate chunked-receive checkpoint on stream completion

The receive path advanced check_dbversion/check_seq per applied chunk to
the chunk's last row, which can fall mid-db_version. Since the server's
cloudsync_payload_chunks resumes on db_version > since with no seq cursor,
a stop between chunks of a split db_version silently skipped the un-applied
rows on the next /check (data loss).

Mirror the send path: advance the receive cursor only after the whole chunk
stream is applied, to the stream watermark, never per chunk. cloudsync_payload_apply
gains a C-level checkpoint argument (watermark / none / last-applied); the
/check response signals watermark + final chunk, and falls back to legacy
monolithic behavior when absent. The public single-arg SQL function and the
send path are unchanged; re-delivered rows stay idempotent.

Adds do_test_payload_chunks_split_dbversion reproducing a single db_version
split across >=2 v2 chunks with partial apply.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: andinux

API.md

@@ -759,6 +759,8 @@ If a package of new changes is already available for the local site, the server
 This function is designed to be called periodically to keep the local database in sync.
 To force an update and wait for changes (with a timeout), use [`cloudsync_network_sync(wait_ms, max_retries)`].
 
+When the server delivers a download as a stream of chunks, the local receive checkpoint is advanced only after the **whole stream** has been applied — never in the middle of a source `db_version`. Mirroring the send path, the server tags the stream with its watermark and marks the final chunk; the checkpoint jumps straight to that watermark on completion. A stop between chunks therefore re-delivers the stream from the unchanged checkpoint on the next call (apply is idempotent, so re-delivered rows are harmless), so no changes can be skipped. A non-chunked (monolithic) download advances the checkpoint to the artifact's last applied position, as before.
+
 If the network is misconfigured or the remote server is unreachable, the function raises a SQL error. If the received payload cannot be applied locally (for example because of an unknown schema hash), the error is returned as a `receive.error` field in the JSON response. If the server reports an unresolved failed check job (e.g. an `encode_changes` failure), that failure is forwarded as a `receive.lastFailure` object.
 
 **Parameters:** None.

CHANGELOG.md

@@ -19,6 +19,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 - `cloudsync_payload_apply()` now accepts legacy payloads, monolithic payloads, and v3 fragment payloads without enforcing the local `payload_max_chunk_size`, preserving compatibility between peers with different settings.
 - `cloudsync_network_send_changes()` now streams outgoing changes through `cloudsync_payload_chunks()` instead of first building one monolithic payload. This bounds transport payload size for the built-in network path and lets large rowsets or oversized BLOB/TEXT values flow through the same `/apply` endpoint as regular payloads.
+- The chunked-download receive path advances the local receive checkpoint (`check_dbversion` / `check_seq`) **only after a chunk stream has been fully applied**, jumping straight to the stream watermark — never into the middle of a source `db_version`. This mirrors the send path and ensures a stop between chunks cannot skip the un-applied rows of a `db_version` split across chunks on the next `/check` (the server resumes on `db_version > since`, with no intra-version cursor). `cloudsync_payload_apply()` no longer advances the receive checkpoint per applied chunk; the built-in network `/check` path drives it from the server's watermark and final-chunk signal, and falls back to the previous monolithic behavior when the server sends no watermark. Re-delivered rows remain idempotent.
 
 ## [1.0.20] - 2026-05-26
 

src/cloudsync.c

@@ -186,6 +186,12 @@ struct cloudsync_context {
 
     // deferred column-batch merge (active during payload_apply)
     merge_pending_batch *pending_batch;
+
+    // last (db_version, seq) successfully applied during the current
+    // cloudsync_payload_apply call; used to resolve the
+    // CLOUDSYNC_CHECKPOINT_LAST_APPLIED receive-checkpoint mode (-1 = none yet).
+    int64_t    apply_last_db_version;
+    int64_t    apply_last_seq;
 };
 
 struct cloudsync_table_context {
@@ -3770,16 +3776,17 @@ static int cloudsync_payload_apply_single_decoded_row (cloudsync_context *data,
         if (rc != DBRES_OK) goto cleanup;
     }
 
-    int dbversion = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION);
-    int seq_setting = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
-    if (db_version >= dbversion) {
-        char buf[256];
-        snprintf(buf, sizeof(buf), "%" PRId64, db_version);
-        dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION, buf);
-        if (seq != seq_setting) {
-            snprintf(buf, sizeof(buf), "%" PRId64, seq);
-            dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_SEQ, buf);
-        }
+    // Do NOT advance the receive cursor here: a v3 value carries a single
+    // (db_version, seq) that can be in the middle of its source db_version, and a
+    // db_version's chunks can span multiple /check artifacts. Advancing per value
+    // would leave the cursor mid-db_version. The durable cursor is advanced once,
+    // after the whole payload/stream is applied, via cloudsync_payload_apply's
+    // checkpoint argument. Record the last applied position for the
+    // CLOUDSYNC_CHECKPOINT_LAST_APPLIED mode.
+    if (db_version > data->apply_last_db_version ||
+        (db_version == data->apply_last_db_version && seq > data->apply_last_seq)) {
+        data->apply_last_db_version = db_version;
+        data->apply_last_seq = seq;
     }
 
     if (pnrows) *pnrows += 1;
@@ -3985,7 +3992,42 @@ static int cloudsync_payload_apply_fragment_row (cloudsync_context *data, clouds
 
 // #ifndef CLOUDSYNC_OMIT_RLS_VALIDATION
 
-int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *pnrows) {
+// Advance the durable receive cursor (check_dbversion/check_seq) after a payload
+// (or a fully-applied chunk stream) has been applied. See the checkpoint-mode
+// documentation on cloudsync_payload_apply in cloudsync.h. The advance is
+// strictly monotonic so re-delivered rows never regress the cursor.
+static void cloudsync_payload_apply_checkpoint (cloudsync_context *data, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
+    int64_t target_db_version;
+    int64_t target_seq;
+
+    if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_NONE) return;
+    if (checkpoint_db_version == CLOUDSYNC_CHECKPOINT_LAST_APPLIED) {
+        // Nothing applied -> nothing to checkpoint.
+        if (data->apply_last_db_version < 0) return;
+        target_db_version = data->apply_last_db_version;
+        target_seq = data->apply_last_seq;
+    } else {
+        target_db_version = checkpoint_db_version;
+        target_seq = checkpoint_seq;
+    }
+
+    int64_t cur_db_version = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION);
+    int64_t cur_seq = dbutils_settings_get_int64_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
+
+    // monotonic guard: never move the cursor backwards
+    if (target_db_version < cur_db_version) return;
+    if (target_db_version == cur_db_version && target_seq <= cur_seq) return;
+
+    char buf[256];
+    snprintf(buf, sizeof(buf), "%" PRId64, target_db_version);
+    dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION, buf);
+    if (target_seq != cur_seq) {
+        snprintf(buf, sizeof(buf), "%" PRId64, target_seq);
+        dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_SEQ, buf);
+    }
+}
+
+int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *pnrows, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
     // Guard against calling payload_apply before cloudsync_init: without this,
     // the settings lookups at the top of this function would each emit a
     // "no such table: cloudsync_settings" debug line, control would fall
@@ -4001,7 +4043,12 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
 
     // sanity check
     if (blen < (int)sizeof(cloudsync_payload_header)) return cloudsync_set_error(data, "Error on cloudsync_payload_apply: invalid payload length", DBRES_MISUSE);
-    
+
+    // track the last (db_version, seq) applied by this call so the receive
+    // checkpoint can be computed once, after the whole payload is applied
+    data->apply_last_db_version = -1;
+    data->apply_last_seq = -1;
+
     // decode header
     cloudsync_payload_header header;
     memcpy(&header, payload, sizeof(cloudsync_payload_header));
@@ -4084,9 +4131,13 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
         }
         if (clone) cloudsync_memory_free(clone);
         if (pnrows) *pnrows = applied_rows;
+        // Advance the receive cursor only after the whole payload is applied,
+        // gated on the caller-supplied checkpoint (a non-final chunk passes
+        // CLOUDSYNC_CHECKPOINT_NONE and leaves the cursor untouched).
+        if (rc == DBRES_OK) cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
         return rc;
     }
-    
+
     // precompile the insert statement
     dbvm_t *vm = NULL;
     int rc = databasevm_prepare(data, SQL_CHANGES_INSERT_ROW, &vm, 0);
@@ -4099,8 +4150,6 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
     uint16_t ncols = header.ncols;
     uint32_t nrows = header.nrows;
     int64_t last_payload_db_version = -1;
-    int dbversion = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION);
-    int seq = dbutils_settings_get_int_value(data, CLOUDSYNC_KEY_CHECK_SEQ);
     cloudsync_pk_decode_bind_context decoded_context = {.vm = vm};
 
     // Initialize deferred column-batch merge
@@ -4213,16 +4262,15 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
 
     if (rc == DBRES_DONE) rc = DBRES_OK;
     if (rc == DBRES_OK) {
-        char buf[256];
-        if (decoded_context.db_version >= dbversion) {
-            snprintf(buf, sizeof(buf), "%" PRId64, decoded_context.db_version);
-            dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_DBVERSION, buf);
-            
-            if (decoded_context.seq != seq) {
-                snprintf(buf, sizeof(buf), "%" PRId64, decoded_context.seq);
-                dbutils_settings_set_key_value(data, CLOUDSYNC_KEY_CHECK_SEQ, buf);
-            }
+        // Record the last applied (db_version, seq) and advance the receive cursor
+        // once, gated on the caller-supplied checkpoint. A non-final chunk passes
+        // CLOUDSYNC_CHECKPOINT_NONE so the cursor never lands mid-db_version.
+        if (decoded_context.db_version > data->apply_last_db_version ||
+            (decoded_context.db_version == data->apply_last_db_version && decoded_context.seq > data->apply_last_seq)) {
+            data->apply_last_db_version = decoded_context.db_version;
+            data->apply_last_seq = decoded_context.seq;
         }
+        cloudsync_payload_apply_checkpoint(data, checkpoint_db_version, checkpoint_seq);
     }
 
 cleanup:

src/cloudsync.h

@@ -94,7 +94,23 @@ const char *cloudsync_schema (cloudsync_context *data);
 const char *cloudsync_table_schema (cloudsync_context *data, const char *table_name);
 
 // Payload
-int    cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *nrows);
+// Receive-checkpoint modes for cloudsync_payload_apply's checkpoint_db_version
+// argument. The receive cursor (check_dbversion/check_seq) must only ever land
+// on a complete db_version boundary, otherwise a stop between chunks of a single
+// source db_version silently skips the unapplied rows on the next /check (the
+// server's cloudsync_payload_chunks uses db_version > since with no seq cursor).
+//   >= 0                              advance the cursor to exactly this
+//                                     (watermark_db_version), with checkpoint_seq.
+//                                     Used once a chunk stream is fully applied.
+//   CLOUDSYNC_CHECKPOINT_NONE         do not advance the cursor. Used for a
+//                                     non-final chunk of a multi-chunk stream.
+//   CLOUDSYNC_CHECKPOINT_LAST_APPLIED advance to this artifact's last applied
+//                                     (db_version, seq). Legacy/monolithic
+//                                     behavior: safe only for a complete payload
+//                                     that ends on a db_version boundary.
+#define CLOUDSYNC_CHECKPOINT_NONE          (-1)
+#define CLOUDSYNC_CHECKPOINT_LAST_APPLIED  (-2)
+int    cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *nrows, int64_t checkpoint_db_version, int64_t checkpoint_seq);
 int    cloudsync_payload_encode_step (cloudsync_payload_context *payload, cloudsync_context *data, int argc, dbvalue_t **argv);
 int    cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data);
 char  *cloudsync_payload_blob (cloudsync_payload_context *payload, int64_t *blob_size, int64_t *nrows);

src/network/network.c

@@ -726,7 +726,7 @@ int network_set_sqlite_result (sqlite3_context *context, NETWORK_RESULT *result)
 // on the sqlite3_context. This lets composite callers (cloudsync_network_sync)
 // surface apply errors as structured JSON. Endpoint/network errors always raise
 // a SQL error regardless of err_out.
-int network_download_changes (sqlite3_context *context, const char *download_url, int *pnrows, char **err_out) {
+int network_download_changes (sqlite3_context *context, const char *download_url, int *pnrows, char **err_out, int64_t checkpoint_db_version, int64_t checkpoint_seq) {
     DEBUG_FUNCTION("network_download_changes");
 
     cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
@@ -740,7 +740,7 @@ int network_download_changes (sqlite3_context *context, const char *download_url
 
     int rc = SQLITE_OK;
     if (result.code == CLOUDSYNC_NETWORK_BUFFER) {
-        rc = cloudsync_payload_apply(data, result.buffer, (int)result.blen, pnrows);
+        rc = cloudsync_payload_apply(data, result.buffer, (int)result.blen, pnrows, checkpoint_db_version, checkpoint_seq);
         if (rc != DBRES_OK) {
             const char *msg = cloudsync_errmsg(data);
             if (!msg || !msg[0]) msg = "cloudsync_payload_apply failed";
@@ -861,6 +861,28 @@ static int64_t json_extract_int(const char *json, size_t json_len, const char *k
     return strtoll(json + val->start, NULL, 10);
 }
 
+static bool json_extract_bool(const char *json, size_t json_len, const char *key, bool default_value) {
+    if (!json || json_len == 0 || !key) return default_value;
+
+    jsmn_parser parser;
+    jsmntok_t tokens[JSMN_MAX_TOKENS];
+    jsmn_init(&parser);
+    int ntokens = jsmn_parse(&parser, json, json_len, tokens, JSMN_MAX_TOKENS);
+    if (ntokens < 1 || tokens[0].type != JSMN_OBJECT) return default_value;
+
+    int i = jsmn_find_key(json, tokens, ntokens, key);
+    if (i < 0 || i + 1 >= ntokens) return default_value;
+
+    jsmntok_t *val = &tokens[i + 1];
+    if (val->type != JSMN_PRIMITIVE) return default_value;
+
+    // JSON booleans (true/false) and numeric flags (1/0) are both accepted.
+    char c = json[val->start];
+    if (c == 't' || c == 'T') return true;
+    if (c == 'f' || c == 'F' || c == 'n' || c == 'N') return false;
+    return strtoll(json + val->start, NULL, 10) != 0;
+}
+
 static int json_extract_array_size(const char *json, size_t json_len, const char *key) {
     if (!json || json_len == 0 || !key) return -1;
 
@@ -1630,7 +1652,30 @@ int cloudsync_network_check_internal(sqlite3_context *context, int *pnrows, sync
         // Branch on the presence of "url" rather than HTTP status; both shapes arrive as BUFFER.
         char *download_url = json_extract_string(result.buffer, result.blen, "url");
         if (download_url) {
-            rc = network_download_changes(context, download_url, pnrows, err_out);
+            // Receive checkpoint, mirroring the send-path watermark. The /check
+            // response signals how far the durable cursor may advance:
+            //   no "watermark"          -> legacy / monolithic artifact: advance
+            //                              to the artifact's last applied
+            //                              (db_version, seq) (CHECKPOINT_LAST_APPLIED).
+            //   "watermark" + "final":  -> chunked stream. Advance the cursor to
+            //                              watermark only on the final chunk;
+            //                              non-final chunks leave it untouched
+            //                              (CHECKPOINT_NONE) so it never lands in
+            //                              the middle of a source db_version.
+            // The server serves successive chunks of the same stream across /check
+            // calls (the durable cursor stays at "since" until "final"), tracking
+            // chunk-level progress on its side; apply is idempotent, so a stop
+            // before "final" simply re-delivers the stream from "since".
+            int64_t watermark = json_extract_int(result.buffer, result.blen, "watermark", -1);
+            int64_t checkpoint_db_version;
+            int64_t checkpoint_seq = 0;
+            if (watermark < 0) {
+                checkpoint_db_version = CLOUDSYNC_CHECKPOINT_LAST_APPLIED;
+            } else {
+                bool final_chunk = json_extract_bool(result.buffer, result.blen, "final", true);
+                checkpoint_db_version = final_chunk ? watermark : CLOUDSYNC_CHECKPOINT_NONE;
+            }
+            rc = network_download_changes(context, download_url, pnrows, err_out, checkpoint_db_version, checkpoint_seq);
             cloudsync_memory_free(download_url);
         }
         // failures.check may appear in either shape; extract opportunistically.

src/postgresql/cloudsync_postgresql.c

@@ -1420,7 +1420,9 @@ Datum cloudsync_payload_decode (PG_FUNCTION_ARGS) {
 
     PG_TRY();
     {
-        rc = cloudsync_payload_apply(data, payload, blen, &nrows);
+        // PostgreSQL applies a complete monolithic payload: legacy last-applied
+        // checkpoint (ends on a db_version boundary, so it is safe).
+        rc = cloudsync_payload_apply(data, payload, blen, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
     }
     PG_CATCH();
     {

src/sqlite/cloudsync_sqlite.c

@@ -1092,11 +1092,16 @@ void dbsync_payload_decode (sqlite3_context *context, int argc, sqlite3_value **
 
     // obtain payload
     const char *payload = (const char *)database_value_blob(argv[0]);
-    
+
     // apply changes
+    // The public SQL function applies a single complete payload: advance the
+    // receive cursor to its last applied (db_version, seq) (legacy behavior, safe
+    // for a payload that ends on a db_version boundary). The chunked-download
+    // receive path gates cursor advancement on stream completion via the C-level
+    // checkpoint argument instead (see cloudsync_payload_apply in cloudsync.h).
     int nrows = 0;
     cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
-    int rc = cloudsync_payload_apply(data, payload, blen, &nrows);
+    int rc = cloudsync_payload_apply(data, payload, blen, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
     if (rc != SQLITE_OK) {
         sqlite3_result_error(context, cloudsync_errmsg(data), -1);
         sqlite3_result_error_code(context, rc);
@@ -1541,7 +1546,9 @@ void dbsync_payload_load (sqlite3_context *context, int argc, sqlite3_value **ar
 
     int nrows = 0;
     cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context);
-    int rc = cloudsync_payload_apply (data, payload, (int)payload_size, &nrows);
+    // File-based load applies a complete monolithic payload: legacy last-applied
+    // checkpoint (ends on a db_version boundary, so it is safe).
+    int rc = cloudsync_payload_apply (data, payload, (int)payload_size, &nrows, CLOUDSYNC_CHECKPOINT_LAST_APPLIED, 0);
     if (payload) cloudsync_memory_free(payload);
 
     if (rc != SQLITE_OK) {