Fix failed SQLite payload commit cleanup; document remaining OOM recovery

.github/workflows/main.yml

@@ -26,6 +26,12 @@ jobs:
     container: ${{ matrix.container && matrix.container || '' }}
     name: ${{ matrix.name }}${{ matrix.arch && format('-{0}', matrix.arch) || '' }} build${{ matrix.arch != 'arm64-v8a' && matrix.arch != 'armeabi-v7a' && matrix.name != 'ios-sim' && matrix.name != 'ios' && matrix.name != 'mac-catalyst' && matrix.name != 'apple-xcframework' && matrix.name != 'android-aar' && ( matrix.name != 'macos' || matrix.arch != 'x86_64' ) && ' + test' || ''}}
     timeout-minutes: 20
+    # Only this leg receives the shared chunked tenant. Lock it across branches,
+    # and queue competing runs instead of canceling a pending PR's validation.
+    concurrency:
+      group: ${{ (matrix.name == 'linux' && matrix.arch == 'x86_64') && 'cloudsync-chunked-test-tenant' || format('build-{0}-{1}-{2}', github.run_id, matrix.name, matrix.arch) }}
+      cancel-in-progress: false
+      queue: max
     strategy:
       fail-fast: false
       matrix:

Makefile

@@ -270,6 +270,8 @@ $(TEST_TARGET): $(TEST_OBJ)
 # Object files
 $(BUILD_RELEASE)/%.o: %.c
 	$(CC) $(CFLAGS) -O3 -fPIC -c $< -o $@
+$(BUILD_TEST)/integration_bootstrap.o: $(TEST_DIR)/integration.c
+
 $(BUILD_TEST)/sqlite3.o: $(SQLITE_DIR)/sqlite3.c
 	$(CC) $(CFLAGS) -DSQLITE_DQS=0 -DSQLITE_CORE -c $< -o $@
 $(BUILD_TEST)/%.o: %.c
@@ -297,9 +299,10 @@ ifneq ($(COVERAGE),false)
 endif
 
 # Run only unit tests
-unittest: $(TARGET) $(DIST_DIR)/unit$(EXE) $(DIST_DIR)/review_regressions$(EXE)
+unittest: $(TARGET) $(DIST_DIR)/unit$(EXE) $(DIST_DIR)/review_regressions$(EXE) $(DIST_DIR)/integration_bootstrap$(EXE)
 	@./$(DIST_DIR)/unit$(EXE)
 	@./$(DIST_DIR)/review_regressions$(EXE)
+	@./$(DIST_DIR)/integration_bootstrap$(EXE)
 
 # Run the SQLite unit and regression suites on a real big-endian host (s390x) under QEMU
 # emulation. The payload and primary-key encodings are byte-order sensitive; this is the

docs/internal/apply-transaction-cleanup.md

@@ -0,0 +1,34 @@
+# SQLite payload apply: failed commit cleanup
+
+A deferred foreign-key violation can make the final `RELEASE` fail while leaving the transaction open. Apply returned an error but retained uncommitted rows and metadata, and the next `BEGIN` failed. A reader blocking commit in rollback-journal mode produces a similar failure with `SQLITE_BUSY`.
+
+Error exits now converge on cleanup, preserve the original error, release pending merge allocations and unwind the failed group. If apply started a transaction while the connection was in autocommit mode and its commit fails, cleanup rolls back that owned transaction. `ROLLBACK TO` followed by `RELEASE` is insufficient for a busy commit because the release can fail again. Caller-owned transactions are never rolled back wholesale. Groups committed in earlier source database versions remain applied; rolled-back rows are excluded from the applied count, and the checkpoint does not advance on failure.
+
+The group savepoint must also open successfully before rows are merged. A separate allocation leak found during the stress sweep is fixed: `database_pk_names` now frees the partially built names array if the second schema scan fails.
+
+## Validation
+
+The audit regression suite runs 100 deferred-constraint failure/retry cycles, covering both final commit and a commit at an intermediate source database-version boundary. It verifies data and metadata rollback, retained committed prefixes, unchanged checkpoints, a subsequent unrelated transaction and successful replay. It also tests preservation of a caller-owned transaction and 30 repeated busy-commit failures followed by successful retry.
+
+The core and audit suites pass with AddressSanitizer and UndefinedBehaviorSanitizer, including instrumentation of SQLite itself and zero outstanding SQLite memory. The initial negative control with the old apply implementation failed 400 assertions in the deferred-constraint tests.
+
+## Remaining engine-level OOM limitation
+
+The diagnostic `test/stress/payload_oom.c` fails each allocation of a SQL-function apply in turn, checks transaction state, performs explicit recovery where necessary and verifies retry and memory usage. It deliberately returns nonzero if any transaction remains open; it is not included in the passing regression suite.
+
+The ordinary build sweep covers allocation indices 0–347: zero memory leaks, zero failed retries after recovery, but 126 attempts still leave a transaction open. During these engine-level failures, SQLite sets its connection's malloc-failed/interrupted state and rejects reentrant cleanup SQL while the outer user-defined function is executing. The extension cannot clear that state through the public SQLite API. This PR does **not** claim to fix those 126 cases. With the fully instrumented ASan/UBSan build, the sweep reaches index 351 and reports 128 open transactions, again with zero leaks and zero failed retries after recovery; no sanitizer diagnostic was emitted. Allocation positions depend on build configuration. A safe complete solution requires a host-side recovery boundary or a change to the SQL apply execution model; replacing application trace callbacks or accessing private SQLite state would introduce compatibility risks.
+
+After a `SQLITE_NOMEM` error, the host should reset/finalize the failed statement and inspect `sqlite3_get_autocommit()`. If the host began the operation in autocommit mode and the connection is still in a transaction, it should explicitly roll back before reuse. If the host owns a transaction, recovery must follow its transaction policy rather than blindly rolling back unrelated work.
+
+To reproduce the diagnostic on macOS after building `dist/review_regressions`:
+
+```sh
+cc -g -O1 -Isrc -Isrc/sqlite -Isrc/network -Isqlite -Imodules/fractional-indexing \
+  -DSQLITE_CORE -DCLOUDSYNC_UNITTEST -DCLOUDSYNC_OMIT_NETWORK \
+  test/stress/payload_oom.c \
+  $(find build/test -name '*.o' ! -name 'unit.o' ! -name '*bench.o' ! -name 'integration.o' ! -name 'integration_bootstrap.o' ! -name 'review_regressions.o') \
+  -framework Security -o /tmp/payload-oom
+/tmp/payload-oom
+```
+
+The independent branch also passes all 521 PostgreSQL 15.19 checks, validating the shared apply cleanup. These are local database tests; no deployed cloud server was modified or exercised.

docs/internal/cloud-e2e.md

@@ -18,9 +18,20 @@ gh run watch RUN_ID --repo sqliteai/sqlite-sync --exit-status
 
 A push already starts that workflow, so do not dispatch a second run for the same
 commit unnecessarily. The workflow cancels older runs of the same branch. Avoid
-running another branch or a local process against the shared chunked tenant at the
-same time: the negative-cache test requires an idle, exclusive tenant, and the
-workflow's concurrency group is per branch, not per tenant.
+running a local process or an older workflow against the shared chunked tenant at
+the same time: the negative-cache test requires an idle, exclusive tenant. The
+Linux x86_64 job now takes a shared job-level concurrency lock across branches.
+`queue: max` retains competing validations instead of replacing a pending run;
+other matrix jobs remain parallel. See [GitHub's concurrency documentation](https://docs.github.com/en/actions/how-tos/write-workflows/choose-when-workflows-run/control-workflow-concurrency).
+
+Fresh receivers use bounded polling: HTTP 202 while a download is prepared is not
+proof of failure or successful synchronization. Bootstrap checks require received
+rows and the expected fixture data, and reject SQL/protocol errors immediately.
+The negative-cache idle assertion remains strict: any unexpected rows still fail.
+`test/integration_bootstrap.c`, included in `make unittest`, exercises delayed and
+partial delivery, exhaustion of the retry budget, absent data, absent received
+rows, protocol failures, malformed JSON and SQL errors without a cloud connection.
+It also asserts one sync call per attempt and zero outstanding SQLite memory.
 
 Inspect the **linux-x86_64 build + test** job. Only that matrix leg receives
 `INTEGRATION_TEST_CHUNKED_DATABASE_ID`. A green job alone is insufficient: optional

src/cloudsync.c

@@ -4324,12 +4324,13 @@ static bool cloudsync_payload_row_is_block (const cloudsync_pk_decode_bind_conte
            memchr(row->col_name, BLOCK_SEPARATOR, (size_t)row->col_name_len) != NULL;
 }
 
-// Opens the savepoint around a PK group of payload rows (see merge_pending_batch). If it
-// cannot be opened the group runs without one and the flush falls back to its own.
-static void cloudsync_payload_group_open (cloudsync_context *data, merge_pending_batch *batch) {
-    if (batch->group_savepoint) return;
-    batch->group_savepoint = (database_begin_savepoint(data, "cloudsync_merge_group") == DBRES_OK);
-    if (!batch->group_savepoint) cloudsync_reset_error(data);
+// Do not write group metadata unless its rollback boundary was established.
+static int cloudsync_payload_group_open (cloudsync_context *data, merge_pending_batch *batch) {
+    if (batch->group_savepoint) return DBRES_OK;
+    int rc = database_begin_savepoint(data, "cloudsync_merge_group");
+    batch->group_savepoint = (rc == DBRES_OK);
+    if (rc != DBRES_OK) cloudsync_set_error(data, "Unable to start a payload group", rc);
+    return rc;
 }
 
 // Flushes the pending PK group and closes its savepoint: released when the flush
@@ -4537,13 +4538,7 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
         size_t seek = 0;
         int res = pk_decode((char *)buffer, buf_len, ncols, &seek, data->skip_decode_idx, cloudsync_payload_decode_callback, &decoded_context);
         if (res == -1) {
-            cloudsync_payload_group_abandon(data, &batch);
-            data->pending_batch = NULL;
-            if (batch.cached_vm) { databasevm_finalize(batch.cached_vm); batch.cached_vm = NULL; }
-            if (batch.cached_col_names) { cloudsync_memory_free(batch.cached_col_names); batch.cached_col_names = NULL; }
-            if (batch.entries) { cloudsync_memory_free(batch.entries); batch.entries = NULL; }
-            if (in_savepoint) database_rollback_savepoint(data, "cloudsync_payload_apply");
-            rc = DBRES_ERROR;
+            rc = cloudsync_set_error(data, "Unable to decode a payload row", DBRES_ERROR);
             goto cleanup;
         }
 
@@ -4572,8 +4567,6 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
         if (in_savepoint && db_version_changed) {
             rc = database_commit_savepoint(data, "cloudsync_payload_apply");
             if (rc != DBRES_OK) {
-                merge_pending_free_entries(&batch);
-                data->pending_batch = NULL;
                 cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to release a savepoint", rc);
                 goto cleanup;
             }
@@ -4583,8 +4576,6 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
         if (!in_savepoint && db_version_changed && !database_in_transaction(data)) {
             rc = database_begin_savepoint(data, "cloudsync_payload_apply");
             if (rc != DBRES_OK) {
-                merge_pending_free_entries(&batch);
-                data->pending_batch = NULL;
                 cloudsync_set_error(data, "Error on cloudsync_payload_apply: unable to start a transaction", rc);
                 goto cleanup;
             }
@@ -4609,7 +4600,8 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
             applied = (int)i;
         }
 
-        cloudsync_payload_group_open(data, &batch);
+        fail_rc = cloudsync_payload_group_open(data, &batch);
+        if (fail_rc != DBRES_OK) break;
         int step_rc = cloudsync_payload_apply_row(data, vm);
         buffer += seek;
         buf_len -= seek;
@@ -4648,9 +4640,10 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
                 snprintf(fail_message, sizeof(fail_message), "%s", cloudsync_errmsg(data));
                 fail_sqlstate = cloudsync_sqlstate(data);
             }
+        } else {
+            in_savepoint = false;
         }
     }
-    cloudsync_apply_stats_add(data, applied);
 
     rc = fail_rc;
     if (rc != DBRES_OK) {
@@ -4672,6 +4665,29 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
     }
 
 cleanup:
+    // A failed RELEASE may leave our transaction open. in_savepoint is only set
+    // when apply began the transaction from autocommit mode, so a full rollback
+    // here cannot discard a caller-owned transaction. ROLLBACK TO plus RELEASE
+    // is insufficient: RELEASE can still hit SQLITE_BUSY while ending the write
+    // transaction. Preserve the original error across cleanup.
+    if (rc != DBRES_OK) {
+        char message[1024];
+        snprintf(message, sizeof(message), "%s", cloudsync_errmsg(data));
+        int sqlstate = cloudsync_sqlstate(data);
+        cloudsync_payload_group_abandon(data, &batch);
+        if (in_savepoint) {
+            applied = applied_at_savepoint;
+            if (database_in_transaction(data))
+                database_exec(data, "ROLLBACK");
+        }
+        cloudsync_reset_error(data);
+        cloudsync_set_error(data, message[0] ? message : "Unable to apply payload changes", rc);
+        cloudsync_set_sqlstate(data, sqlstate);
+    }
+    data->pending_batch = NULL;
+    merge_pending_free_entries(&batch);
+    cloudsync_apply_stats_add(data, applied);
+
     // cleanup merge_pending_batch
     if (batch.cached_vm) { databasevm_finalize(batch.cached_vm); batch.cached_vm = NULL; }
     if (batch.cached_col_names) { cloudsync_memory_free(batch.cached_col_names); batch.cached_col_names = NULL; }

src/sqlite/database_sqlite.c

@@ -1227,7 +1227,8 @@ int database_pk_names (cloudsync_context *data, const char *table_name, char ***
         if (!r[i]) { rc = SQLITE_NOMEM; goto cleanup_r;}
         i++;
     }
-    if (rc == SQLITE_DONE) rc = SQLITE_OK;
+    if (rc != SQLITE_DONE) goto cleanup_r;
+    rc = SQLITE_OK;
 
     *names = r;
     *count = rows;

test/integration.c

@@ -232,6 +232,37 @@ int db_select_receive (sqlite3 *db, const char *sql, int *chunks, int *complete,
     return sqlite3_finalize(stmt);
 }
 
+// A fresh site's first check may return HTTP 202 while its download is prepared.
+// Require actual received rows AND the expected data, allowing bounded empty polls.
+// Materialize the scalar result so JSON projections cannot invoke sync repeatedly.
+int db_sync_await(sqlite3 *db, const char *expected_sql, int max_attempts, int delay_ms) {
+    bool received = false;
+    for (int attempt = 0; attempt < max_attempts; attempt++) {
+        int rows = 0, valid = 0;
+        char error[512];
+        int rc = db_select_receive(db,
+            "WITH result AS MATERIALIZED (SELECT cloudsync_network_sync(250,10) AS j) "
+            "SELECT j ->> '$.receive.rows', "
+            "coalesce((j ->> '$.send.status') <> 'error' AND "
+            "json_type(j,'$.receive.rows') = 'integer', 0), "
+            "coalesce(j ->> '$.receive.error', j ->> '$.send.lastFailure', j ->> '$.receive.lastFailure') "
+            "FROM result;", &rows, &valid, error, sizeof(error));
+        if (rc != SQLITE_OK) return rc;
+        if (!valid || rows < 0 || error[0]) {
+            printf("Error: bootstrap sync failed: %s\n", error[0] ? error : "invalid sync status");
+            return SQLITE_ERROR;
+        }
+        received = received || rows > 0;
+        int ready = 0;
+        rc = db_select_int(db, expected_sql, &ready);
+        if (rc != SQLITE_OK) return rc;
+        if (received && ready) return SQLITE_OK;
+        if (attempt + 1 < max_attempts) sqlite3_sleep(delay_ms);
+    }
+    printf("Error: bootstrap sync did not deliver the expected data after %d attempts\n", max_attempts);
+    return SQLITE_ERROR;
+}
+
 int db_expect_min (sqlite3 *db, const char *sql, int expect_min) {
     int value = 0;
     int rc = db_select_int(db, sql, &value);
@@ -587,7 +618,7 @@ int test_init (const char *db_path, int init) {
     snprintf(sql, sizeof(sql), "INSERT INTO users (id, name) VALUES ('%s', '%s');", value, value);
     rc = db_exec(db, sql); RCHECK
     rc = db_expect_int(db, "SELECT COUNT(*) as count FROM users;", 1); RCHECK
-    rc = db_expect_gt0(db, "SELECT cloudsync_network_sync(250,10) ->> '$.receive.rows';"); RCHECK
+    rc = db_sync_await(db, "SELECT count(*) > 0 FROM activities;", 120, 250); RCHECK
     rc = db_expect_gt0(db, "SELECT COUNT(*) as count FROM users;"); RCHECK
     rc = db_expect_gt0(db, "SELECT COUNT(*) as count FROM activities;"); RCHECK
     rc = db_expect_int(db, "SELECT COUNT(*) as count FROM workouts;", 0); RCHECK
@@ -689,7 +720,8 @@ int test_enable_disable(const char *db_path) {
         rc = db_exec(db2, set_apikey2); RCHECK
     }
 
-    rc = db_expect_gt0(db2, "SELECT cloudsync_network_sync(250,10) ->> '$.receive.rows';"); RCHECK
+    snprintf(sql, sizeof(sql), "SELECT COUNT(*) = 1 FROM users WHERE name='%s-should-sync';", value);
+    rc = db_sync_await(db2, sql, 120, 250); RCHECK
 
     snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM users WHERE name='%s';", value);
     rc = db_expect_int(db2, sql, 0); RCHECK
@@ -799,7 +831,7 @@ int test_token_auth (void) {
     char sql[256];
     snprintf(sql, sizeof(sql), "INSERT INTO users (id, name) VALUES ('%s', '%s');", value, value);
     rc = db_exec(db, sql); RCHECK
-    rc = db_expect_gt0(db, "SELECT cloudsync_network_sync(250,10) ->> '$.receive.rows';"); RCHECK
+    rc = db_sync_await(db, "SELECT count(*) > 0 FROM activities;", 120, 250); RCHECK
     rc = db_exec(db, "SELECT cloudsync_terminate();");
 
 ABORT_TEST