Remove the PostgreSQL savepoint resource-owner depth limit

.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/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

docs/internal/deep-savepoints.md

@@ -0,0 +1,13 @@
+# PostgreSQL savepoint depth
+
+A heap scan feeding `cloudsync_payload_apply(payload)` failed with `buffer pin ... is not owned by resource owner SubTransaction` at 126 user savepoints. The caller's resource owner and memory context were stored in a 128-entry array; Cloudsync's internal savepoints consumed additional levels and silently exceeded the array.
+
+The backend now keeps one dynamically allocated frame per Cloudsync subtransaction. Each frame records the caller's owner, memory context and subtransaction ID. Commit and rollback restore a local copy after PostgreSQL ends the subtransaction. Subtransaction callbacks discard frames on both normal completion and external abort; transaction callbacks clear the stack before its owning memory context disappears. Snapshot replacement remains restricted to the outermost Cloudsync savepoint.
+
+This removes Cloudsync's fixed depth cap; PostgreSQL's own resource limits still apply. Allocations happen before opening a new subtransaction.
+
+## Validation
+
+`test/postgresql/62_deep_savepoints.sql` is included in `full_test.sql`. It exercises depths 1, 125, 126, 127, 128, 256, 1024 and 2048, reads payloads from a heap table, checks rollback, reapplies in the same transaction and commits. It also catches 100 trigger failures at depth 256 and verifies a subsequent successful apply in the same backend.
+
+The full suite passes all 546 checks on PostgreSQL 15.19, 17.11 and 18.6. Restoring the old implementation reproduces the buffer-owner error at depth 126. These are local database tests, not tests of a deployed cloud server.

src/postgresql/database_postgresql.c

@@ -3062,32 +3062,48 @@ static int database_refresh_snapshot (void) {
 // pin of the scan feeding cloudsync_payload_apply, say) is then charged to the wrong
 // owner: "buffer pin ... is not owned by resource owner TopTransaction". So, as the
 // procedural languages do, remember both at the start of each subtransaction and
-// restore them when it ends. Indexed by nesting level so an owner never leaks across
-// a subtransaction that was aborted elsewhere.
-#define CLOUDSYNC_SAVEPOINT_MAX_DEPTH 128
-static ResourceOwner savepoint_owner[CLOUDSYNC_SAVEPOINT_MAX_DEPTH];
-static MemoryContext savepoint_context[CLOUDSYNC_SAVEPOINT_MAX_DEPTH];
-
-// Only the outermost savepoint opened here may swap the active snapshot when it ends: a
-// snapshot pushed while an enclosing savepoint is still open belongs to that
-// subtransaction, and rolling the enclosing one back pops it — in place of the caller's
-// snapshot the swap popped, leaving the caller's portal without one (an assertion
-// failure in EnsurePortalSnapshotExists). Nested, advancing the command counter is
-// enough to make the changes visible: every SPI statement takes a fresh snapshot.
-// True when the subtransaction at level is nested inside another savepoint opened here.
-static bool savepoint_is_nested (int level) {
-    for (int k = 2; k < level && k < CLOUDSYNC_SAVEPOINT_MAX_DEPTH; k++) {
-        if (savepoint_owner[k]) return true;
-    }
-    return false;
-}
-
-static void savepoint_restore_caller (int level) {
-    if (level <= 0 || level >= CLOUDSYNC_SAVEPOINT_MAX_DEPTH || !savepoint_owner[level]) return;
-    MemoryContextSwitchTo(savepoint_context[level]);
-    CurrentResourceOwner = savepoint_owner[level];
-    savepoint_owner[level] = NULL;
-    savepoint_context[level] = NULL;
+// restore them when it ends. Frames live in TopTransactionContext and are removed
+// by transaction callbacks even when a caller aborts outside these wrappers.
+typedef struct cloudsync_savepoint_frame {
+    ResourceOwner owner;
+    MemoryContext context;
+    SubTransactionId subid;
+    struct cloudsync_savepoint_frame *previous;
+} cloudsync_savepoint_frame;
+
+static cloudsync_savepoint_frame *savepoint_stack;
+static bool savepoint_callbacks_registered;
+
+static void savepoint_xact_callback (XactEvent event, void *arg) {
+    if (event == XACT_EVENT_COMMIT || event == XACT_EVENT_ABORT ||
+        event == XACT_EVENT_PARALLEL_COMMIT || event == XACT_EVENT_PARALLEL_ABORT ||
+        event == XACT_EVENT_PREPARE)
+        savepoint_stack = NULL; // TopTransactionContext owns the allocations.
+}
+
+static void savepoint_subxact_callback (SubXactEvent event, SubTransactionId subid,
+                                       SubTransactionId parent, void *arg) {
+    if (event != SUBXACT_EVENT_COMMIT_SUB && event != SUBXACT_EVENT_ABORT_SUB) return;
+    if (savepoint_stack && savepoint_stack->subid == subid) {
+        cloudsync_savepoint_frame *frame = savepoint_stack;
+        savepoint_stack = frame->previous;
+        pfree(frame);
+    }
+}
+
+// Only the outermost cloudsync savepoint may replace the caller's snapshot.
+// Copy the frame before ending the subtransaction: its callback frees the frame.
+static cloudsync_savepoint_frame savepoint_caller (void) {
+    cloudsync_savepoint_frame caller = {0};
+    if (savepoint_stack && savepoint_stack->subid == GetCurrentSubTransactionId())
+        caller = *savepoint_stack;
+    return caller;
+}
+
+static void savepoint_restore_caller (cloudsync_savepoint_frame caller) {
+    if (!caller.owner) return;
+    MemoryContextSwitchTo(caller.context);
+    CurrentResourceOwner = caller.owner;
 }
 
 int database_begin_savepoint (cloudsync_context *data, const char *savepoint_name) {
@@ -3098,12 +3114,20 @@ int database_begin_savepoint (cloudsync_context *data, const char *savepoint_nam
     ResourceOwner oldowner = CurrentResourceOwner;
     PG_TRY();
     {
-        BeginInternalSubTransaction(NULL);
-        int level = GetCurrentTransactionNestLevel();
-        if (level > 0 && level < CLOUDSYNC_SAVEPOINT_MAX_DEPTH) {
-            savepoint_owner[level] = oldowner;
-            savepoint_context[level] = oldcontext;
+        if (!savepoint_callbacks_registered) {
+            RegisterXactCallback(savepoint_xact_callback, NULL);
+            RegisterSubXactCallback(savepoint_subxact_callback, NULL);
+            savepoint_callbacks_registered = true;
         }
+        // Allocate before opening a subtransaction so an allocation error cannot
+        // leave an untracked rollback boundary behind.
+        cloudsync_savepoint_frame *frame = MemoryContextAlloc(TopTransactionContext, sizeof(*frame));
+        frame->owner = oldowner;
+        frame->context = oldcontext;
+        frame->previous = savepoint_stack;
+        BeginInternalSubTransaction(NULL);
+        frame->subid = GetCurrentSubTransactionId();
+        savepoint_stack = frame;
         // Keep allocating in the caller's context; the subtransaction's resource owner
         // stays current so what the savepoint acquires is released with it.
         MemoryContextSwitchTo(oldcontext);
@@ -3128,13 +3152,12 @@ int database_commit_savepoint (cloudsync_context *data, const char *savepoint_na
     int rc = DBRES_OK;
 
     MemoryContext oldcontext = CurrentMemoryContext;
-    int level = GetCurrentTransactionNestLevel();
+    cloudsync_savepoint_frame caller = savepoint_caller();
     PG_TRY();
     {
         ReleaseCurrentSubTransaction();
-        bool nested = savepoint_is_nested(level);
-        savepoint_restore_caller(level);
-        if (nested) CommandCounterIncrement();
+        savepoint_restore_caller(caller);
+        if (caller.previous) CommandCounterIncrement();
         else database_refresh_snapshot();
     }
     PG_CATCH();
@@ -3157,13 +3180,12 @@ int database_rollback_savepoint (cloudsync_context *data, const char *savepoint_
     int rc = DBRES_OK;
 
     MemoryContext oldcontext = CurrentMemoryContext;
-    int level = GetCurrentTransactionNestLevel();
+    cloudsync_savepoint_frame caller = savepoint_caller();
     PG_TRY();
     {
         RollbackAndReleaseCurrentSubTransaction();
-        bool nested = savepoint_is_nested(level);
-        savepoint_restore_caller(level);
-        if (nested) CommandCounterIncrement();
+        savepoint_restore_caller(caller);
+        if (caller.previous) CommandCounterIncrement();
         else database_refresh_snapshot();
     }
     PG_CATCH();

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

test/integration_bootstrap.c

@@ -0,0 +1,74 @@
+// Exercise the integration wait policy without contacting the cloud.
+#define main integration_main
+#include "integration.c"
+#undef main
+
+static int failures;
+#define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); failures++; } } while (0)
+
+typedef struct {
+    const char *json;
+    int ready;
+} sync_reply;
+typedef struct {
+    const sync_reply *replies;
+    int count;
+    int calls;
+} sync_script;
+
+static void mock_sync(sqlite3_context *ctx, int argc, sqlite3_value **argv) {
+    sync_script *script = sqlite3_user_data(ctx);
+    CHECK(argc == 2 && sqlite3_value_int(argv[0]) == 250 && sqlite3_value_int(argv[1]) == 10);
+    if (script->calls >= script->count) {
+        sqlite3_result_error(ctx, "unexpected extra sync", -1);
+        return;
+    }
+    sync_reply reply = script->replies[script->calls++];
+    if (reply.ready) CHECK(sqlite3_exec(sqlite3_context_db_handle(ctx), "UPDATE fixture SET ready=1", NULL, NULL, NULL) == SQLITE_OK);
+    if (reply.json) sqlite3_result_text(ctx, reply.json, -1, SQLITE_STATIC);
+    else sqlite3_result_error(ctx, "injected SQL error", -1);
+}
+
+static void run_case(const sync_reply *replies, int count, int expected_rc, int expected_calls) {
+    sqlite3 *db = NULL;
+    CHECK(sqlite3_open(":memory:", &db) == SQLITE_OK);
+    CHECK(sqlite3_exec(db, "CREATE TABLE fixture(ready); INSERT INTO fixture VALUES(0)", NULL, NULL, NULL) == SQLITE_OK);
+    sync_script script = {replies, count, 0};
+    CHECK(sqlite3_create_function(db, "cloudsync_network_sync", 2, SQLITE_UTF8, &script, mock_sync, NULL, NULL) == SQLITE_OK);
+    CHECK(db_sync_await(db, "SELECT ready FROM fixture", count, 0) == expected_rc);
+    CHECK(script.calls == expected_calls);
+    CHECK(sqlite3_close(db) == SQLITE_OK);
+}
+
+int main(void) {
+    const char *empty = "{\"send\":{\"status\":\"ok\"},\"receive\":{\"rows\":0}}";
+    const char *rows = "{\"send\":{\"status\":\"ok\"},\"receive\":{\"rows\":3}}";
+    sync_reply delayed[] = {{empty,0},{empty,0},{rows,1}};
+    sync_reply partial[] = {{rows,0},{empty,0},{empty,1}};
+    sync_reply never[] = {{empty,0},{empty,0},{empty,0}};
+    sync_reply no_data[] = {{rows,0},{rows,0}};
+    sync_reply no_rows[] = {{empty,1},{empty,1}};
+    sync_reply immediate[] = {{rows,1}};
+    for (int i = 0; i < 100; i++) {
+        run_case(delayed, 3, SQLITE_OK, 3);
+        run_case(partial, 3, SQLITE_OK, 3);
+        run_case(immediate, 1, SQLITE_OK, 1);
+    }
+    run_case(never, 3, SQLITE_ERROR, 3);
+    run_case(no_data, 2, SQLITE_ERROR, 2);
+    run_case(no_rows, 2, SQLITE_ERROR, 2);
+    const char *errors[] = {
+        "{\"send\":{\"status\":\"error\"},\"receive\":{\"rows\":3}}",
+        "{\"send\":{\"status\":\"ok\",\"lastFailure\":{\"message\":\"denied\"}},\"receive\":{\"rows\":3}}",
+        "{\"send\":{\"status\":\"ok\"},\"receive\":{\"rows\":3,\"error\":\"denied\"}}",
+        "{\"send\":{\"status\":\"ok\"},\"receive\":{\"rows\":3,\"lastFailure\":{\"message\":\"denied\"}}}",
+        "{\"receive\":{\"rows\":3}}", "{}", "null", "{invalid", NULL
+    };
+    for (unsigned i = 0; i < sizeof(errors)/sizeof(errors[0]); i++) {
+        sync_reply error[] = {{errors[i],1},{rows,1}};
+        run_case(error, 2, SQLITE_ERROR, 1);
+    }
+    CHECK(sqlite3_memory_used() == 0);
+    printf("Integration bootstrap policy: %d failures\n", failures);
+    return failures ? 1 : 0;
+}