diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 4c0ffaf..556abfd 100644 --- a/.github/workflows/main.yml +++ b/.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: diff --git a/Makefile b/Makefile index e94242d..aeec422 100644 --- a/Makefile +++ b/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 diff --git a/docs/internal/cloud-e2e.md b/docs/internal/cloud-e2e.md index cfbd202..0a844a9 100644 --- a/docs/internal/cloud-e2e.md +++ b/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 diff --git a/docs/internal/deep-savepoints.md b/docs/internal/deep-savepoints.md new file mode 100644 index 0000000..701a8e9 --- /dev/null +++ b/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. diff --git a/src/postgresql/database_postgresql.c b/src/postgresql/database_postgresql.c index 0701f67..da4e55d 100644 --- a/src/postgresql/database_postgresql.c +++ b/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(); diff --git a/test/integration.c b/test/integration.c index 27fadea..9e0b99e 100644 --- a/test/integration.c +++ b/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 diff --git a/test/integration_bootstrap.c b/test/integration_bootstrap.c new file mode 100644 index 0000000..f5be8bf --- /dev/null +++ b/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; +} diff --git a/test/postgresql/62_deep_savepoints.sql b/test/postgresql/62_deep_savepoints.sql new file mode 100644 index 0000000..144d810 --- /dev/null +++ b/test/postgresql/62_deep_savepoints.sql @@ -0,0 +1,294 @@ +-- Exercise caller-owned buffers across internal subtransactions at deep nesting. +\set testid '62-deep-savepoints' +\ir helper_test_init.sql +\connect postgres +\ir helper_psql_conn_setup.sql +DROP DATABASE IF EXISTS cloudsync_test_62_source; +DROP DATABASE IF EXISTS cloudsync_test_62_target; +CREATE DATABASE cloudsync_test_62_source; +CREATE DATABASE cloudsync_test_62_target; +\connect cloudsync_test_62_source +\ir helper_psql_conn_setup.sql +CREATE EXTENSION cloudsync; +CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); +SELECT cloudsync_init('t'); +INSERT INTO t SELECT i::text, repeat('value', 50) FROM generate_series(1, 20) i; +SELECT encode(cloudsync_payload_encode(tbl,pk,col_name,col_value,col_version,db_version,site_id,cl,seq),'hex') AS payload FROM cloudsync_changes \gset +\connect cloudsync_test_62_target +\ir helper_psql_conn_setup.sql +CREATE EXTENSION cloudsync; +CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); +SELECT cloudsync_init('t'); +CREATE TABLE transport(payload BYTEA); +INSERT INTO transport SELECT decode(:'payload','hex') FROM generate_series(1, 10); + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 1) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 1 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 1 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 1 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 1 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 1 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 1 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 125) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 125 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 125 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 125 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 125 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 125 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 125 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 126) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 126 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 126 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 126 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 126 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 126 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 126 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 127) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 127 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 127 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 127 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 127 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 127 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 127 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 128) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 128 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 128 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 128 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 128 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 128 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 128 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 256) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 256 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 256 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 256 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 256 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 256 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 256 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 1024) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 1024 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 1024 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 1024 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 1024 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 1024 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 1024 +\endif +TRUNCATE t, t_cloudsync; + +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 2048) \gexec +-- A heap scan owns the input buffer; an aggregate consumes every apply result. +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) apply at depth 2048 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) apply at depth 2048 +\endif +ROLLBACK TO user_sp; +SELECT count(*) = 0 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) rollback at depth 2048 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) rollback at depth 2048 +\endif +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) reuse and commit at depth 2048 +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) reuse and commit at depth 2048 +\endif +TRUNCATE t, t_cloudsync; + +-- Catch errors outside the internal wrappers, then reuse the same backend. +CREATE FUNCTION reject_row() RETURNS trigger LANGUAGE plpgsql AS $$ +BEGIN RAISE EXCEPTION 'deep apply rejected'; END $$; +CREATE TRIGGER reject_row BEFORE INSERT ON t FOR EACH ROW EXECUTE FUNCTION reject_row(); +BEGIN; +SELECT 'SAVEPOINT user_sp' FROM generate_series(1, 256) \gexec +DO $$ +BEGIN + FOR i IN 1..100 LOOP + BEGIN + PERFORM cloudsync_payload_apply(payload) FROM transport; + RAISE EXCEPTION 'expected trigger rejection'; + EXCEPTION WHEN OTHERS THEN + IF SQLERRM NOT LIKE '%deep apply rejected%' THEN RAISE; END IF; + END; + END LOOP; +END $$; +DROP TRIGGER reject_row ON t; +SELECT sum(cloudsync_payload_apply(payload)) FROM transport; +COMMIT; +SELECT count(*) = 20 AS ok FROM t \gset +\if :ok +\echo [PASS] (:testid) 100 caught errors followed by successful apply +\else +SELECT (:fail::int + 1) AS fail \gset +\echo [FAIL] (:testid) error recovery +\endif +\connect postgres +DROP DATABASE cloudsync_test_62_source; +DROP DATABASE cloudsync_test_62_target; diff --git a/test/postgresql/full_test.sql b/test/postgresql/full_test.sql index 81e6cc9..b921c6d 100644 --- a/test/postgresql/full_test.sql +++ b/test/postgresql/full_test.sql @@ -69,6 +69,7 @@ \ir 59_rls_denial_retry.sql \ir 60_fragment_concurrency.sql \ir 61_fragment_cleanup_backlog.sql +\ir 62_deep_savepoints.sql -- 'Test summary' \echo '\nTest summary:'