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/apply-transaction-cleanup.md b/docs/internal/apply-transaction-cleanup.md new file mode 100644 index 0000000..1625490 --- /dev/null +++ b/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. 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/src/cloudsync.c b/src/cloudsync.c index 8597ef4..56a84e5 100644 --- a/src/cloudsync.c +++ b/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; } diff --git a/src/sqlite/database_sqlite.c b/src/sqlite/database_sqlite.c index 63ac9f8..cc1e0e7 100644 --- a/src/sqlite/database_sqlite.c +++ b/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; 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/review_regressions.c b/test/review_regressions.c index 5618a5d..d3db170 100644 --- a/test/review_regressions.c +++ b/test/review_regressions.c @@ -503,6 +503,73 @@ static void test_block_materialize_errors(void) { CHECK(scalar(db, "SELECT body = 'a' || char(10) || 'b' FROM docs WHERE id='1'") == 1); CHECK(close_db(db) == SQLITE_OK); } +static void test_failed_apply_commit(void) { + const char *schema = "PRAGMA foreign_keys=ON; CREATE TABLE parent(id TEXT PRIMARY KEY);" + "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED);" + "SELECT cloudsync_init('t'); CREATE TABLE caller_work(value TEXT);"; + for (int iteration = 0; iteration < 100; iteration++) { + sqlite3 *source = open_db(), *target = open_db(); + CHECK(sql(source, schema) == SQLITE_OK); + CHECK(sql(target, schema) == SQLITE_OK); + CHECK(sql(source, "INSERT INTO parent VALUES('valid'),('missing'); INSERT INTO t VALUES('a','valid');" + "INSERT INTO t VALUES('b','missing');") == SQLITE_OK); + if (iteration % 2) CHECK(sql(source, "INSERT INTO t VALUES('c','valid');") == SQLITE_OK); + CHECK(sql(target, "INSERT INTO parent VALUES('valid');") == SQLITE_OK); + CHECK(apply_payload(source, target) == SQLITE_CONSTRAINT); + CHECK(sqlite3_get_autocommit(target)); + CHECK(scalar(target, "SELECT count(*) FROM t WHERE id='b'") == 0); + CHECK(scalar(target, "SELECT count(*) FROM t_cloudsync WHERE pk=cloudsync_pk_encode('b')") == 0); + CHECK(scalar(target, "SELECT count(*) FROM t WHERE id='a'") == 1); + CHECK(scalar(target, "SELECT coalesce((SELECT value FROM cloudsync_settings WHERE key='check_dbversion'),0)") == 0); + CHECK(sql(target, "BEGIN; INSERT INTO caller_work VALUES('kept'); COMMIT;") == SQLITE_OK); + CHECK(sql(target, "INSERT INTO parent VALUES('missing');") == SQLITE_OK); + CHECK(apply_payload(source, target) == SQLITE_ROW); + CHECK(scalar(target, "SELECT count(*) FROM t") == 2 + iteration % 2); + CHECK(close_db(source) == SQLITE_OK); + CHECK(close_db(target) == SQLITE_OK); + } + // A caller-owned transaction must survive a rejected group unchanged. + sqlite3 *source = open_db(), *target = open_db(); + CHECK(sql(source, schema) == SQLITE_OK && sql(target, schema) == SQLITE_OK); + CHECK(sql(source, "INSERT INTO parent VALUES('valid'); INSERT INTO t VALUES('a','valid');") == SQLITE_OK); + CHECK(sql(target, "INSERT INTO parent VALUES('valid'); CREATE TRIGGER deny BEFORE INSERT ON t BEGIN SELECT RAISE(ABORT,'denied'); END;" + "BEGIN; INSERT INTO caller_work VALUES('kept'); SAVEPOINT caller_sp;") == SQLITE_OK); + CHECK(apply_payload(source, target) == SQLITE_CONSTRAINT); + CHECK(!sqlite3_get_autocommit(target)); + CHECK(scalar(target, "SELECT count(*) FROM caller_work") == 1); + CHECK(sql(target, "ROLLBACK TO caller_sp; RELEASE caller_sp; DROP TRIGGER deny;") == SQLITE_OK); + CHECK(apply_payload(source, target) == SQLITE_ROW); + CHECK(sql(target, "COMMIT") == SQLITE_OK); + CHECK(scalar(target, "SELECT count(*) FROM caller_work") == 1); + CHECK(close_db(source) == SQLITE_OK && close_db(target) == SQLITE_OK); + // A reader permits writes but prevents the outer RELEASE from committing. + CHECK(scratch_create()); + char path[512]; + snprintf(path, sizeof(path), "%s/commit-busy.db", scratch_dir); + source = open_db(); target = NULL; + sqlite3 *reader = NULL; + CHECK(sqlite3_open(path, &target) == SQLITE_OK); + CHECK(sqlite3_cloudsync_init(target, NULL, NULL) == SQLITE_OK); + CHECK(sql(source, schema) == SQLITE_OK && sql(target, schema) == SQLITE_OK); + CHECK(sql(source, "INSERT INTO parent VALUES('valid'); INSERT INTO t VALUES('a','valid');") == SQLITE_OK); + CHECK(sql(target, "INSERT INTO parent VALUES('valid');") == SQLITE_OK); + CHECK(sqlite3_open(path, &reader) == SQLITE_OK); + for (int i = 0; i < 30; i++) { + CHECK(sql(reader, "BEGIN; SELECT * FROM t;") == SQLITE_OK); + CHECK(apply_payload(source, target) == SQLITE_BUSY); + CHECK(sqlite3_get_autocommit(target)); + CHECK(scalar(target, "SELECT count(*) FROM t") == 0); + CHECK(sql(reader, "ROLLBACK") == SQLITE_OK); + } + CHECK(apply_payload(source, target) == SQLITE_ROW); + CHECK(scalar(target, "SELECT count(*) FROM t") == 1); + CHECK(sqlite3_close(reader) == SQLITE_OK); + CHECK(close_db(source) == SQLITE_OK && close_db(target) == SQLITE_OK); + const char *files[] = {"commit-busy.db", "commit-busy.db-journal"}; + scratch_remove(files, 2); + +} + int main(void) { CHECK(sqlite3_config(SQLITE_CONFIG_GETMALLOC, &memory) == SQLITE_OK); sqlite3_mem_methods faults = memory; @@ -513,6 +580,7 @@ int main(void) { test_clocks_and_double(); test_best_index(); test_payload_errors(); + test_failed_apply_commit(); test_payload_high_compression(); test_resurrected_group_rollback(); test_batched_update_missing_row(); diff --git a/test/stress/payload_oom.c b/test/stress/payload_oom.c new file mode 100644 index 0000000..c609edd --- /dev/null +++ b/test/stress/payload_oom.c @@ -0,0 +1,68 @@ +// Diagnostic stress test. A nonzero exit reports ANY remaining cleanup failure. +// Explicit recovery lets the sweep continue without hiding open transactions. +#define main regression_main +#include "../review_regressions.c" +#undef main + +int main(int argc, char **argv) { + CHECK(sqlite3_config(SQLITE_CONFIG_GETMALLOC, &memory) == SQLITE_OK); + sqlite3_mem_methods faults = memory; + faults.xMalloc = fault_malloc; + faults.xRealloc = fault_realloc; + CHECK(sqlite3_config(SQLITE_CONFIG_MALLOC, &faults) == SQLITE_OK); + CHECK(sqlite3_initialize() == SQLITE_OK); + sqlite3 *source = open_db(); + const char *schema = "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, value TEXT); SELECT cloudsync_init('t');"; + CHECK(sql(source, schema) == SQLITE_OK); + CHECK(sql(source, "INSERT INTO t VALUES('a','one'),('b','two'),('c','three')") == SQLITE_OK); + sqlite3_stmt *read = NULL; + CHECK(sqlite3_prepare_v2(source, "SELECT cloudsync_payload_encode(tbl,pk,col_name,col_value,col_version,db_version,site_id,cl,seq) FROM cloudsync_changes", -1, &read, NULL) == SQLITE_OK); + CHECK(sqlite3_step(read) == SQLITE_ROW); + int start = argc > 1 ? atoi(argv[1]) : 0; + int end = argc > 2 ? atoi(argv[2]) : 1500; + int open_transactions = 0, leaks = 0, retries_failed = 0; + bool exhausted = false; + for (int n = start; n < end; n++) { + sqlite3_int64 before = sqlite3_memory_used(); + sqlite3 *target = open_db(); + CHECK(sql(target, schema) == SQLITE_OK); + sqlite3_stmt *write = NULL; + CHECK(sqlite3_prepare_v2(target, "SELECT cloudsync_payload_decode(?1)", -1, &write, NULL) == SQLITE_OK); + CHECK(sqlite3_bind_value(write, 1, sqlite3_column_value(read, 0)) == SQLITE_OK); + fail_once = true; + fail_after = n; + int rc = sqlite3_step(write); + int left = fail_after; + fail_after = -1; + fail_once = false; + sqlite3_finalize(write); + if (!sqlite3_get_autocommit(target)) { + open_transactions++; + fprintf(stderr, "n=%d rc=%d TRANSACTION LEFT OPEN\n", n, rc); + CHECK(sql(target, "ROLLBACK") == SQLITE_OK); + } + if (rc != SQLITE_ROW) { + int retry_rc = apply_payload(source, target); + if (retry_rc != SQLITE_ROW || scalar(target, "SELECT count(*) FROM t") != 3) { + retries_failed++; + fprintf(stderr, "n=%d rc=%d RETRY FAILED rc=%d %s\n", n, rc, retry_rc, sqlite3_errmsg(target)); + } + } + CHECK(close_db(target) == SQLITE_OK); + sqlite3_int64 leak = sqlite3_memory_used() - before; + if (leak) { + leaks++; + fprintf(stderr, "n=%d rc=%d LEAK=%lld\n", n, rc, leak); + } + if (left > 0) { + exhausted = true; + printf("All allocations exhausted at n=%d\n", n); + break; + } + } + sqlite3_finalize(read); + CHECK(close_db(source) == SQLITE_OK); + printf("open_transactions=%d leaks=%d retries_failed=%d final_memory=%lld\n", + open_transactions, leaks, retries_failed, sqlite3_memory_used()); + return failures || !exhausted || open_transactions || leaks || retries_failed || sqlite3_memory_used() ? 1 : 0; +}