From 2ae6f534d1b02d27d5e51cc65334137e6e65cc7d Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 17 Mar 2026 16:25:44 -0600 Subject: [PATCH 1/2] test: improved stress test command --- .../commands/stress-test-sync-sqlitecloud.md | 25 +++++++++++++------ 1 file changed, 18 insertions(+), 7 deletions(-) diff --git a/.claude/commands/stress-test-sync-sqlitecloud.md b/.claude/commands/stress-test-sync-sqlitecloud.md index 2540008..4bb41fa 100644 --- a/.claude/commands/stress-test-sync-sqlitecloud.md +++ b/.claude/commands/stress-test-sync-sqlitecloud.md @@ -66,9 +66,17 @@ For the network init call throughout the test, use: - Default address: `SELECT cloudsync_network_init('');` - Custom address: `SELECT cloudsync_network_init_custom('', '');` -### Step 4: Get Auth Tokens (if RLS enabled) +### Step 4: Set Up Authentication -Create tokens for the test users. Create as many users as needed for the number of concurrent databases (assign 2 databases per user, or 1 per user if NUM_DBS <= 2). +Authentication depends on the RLS mode: + +**If RLS is disabled:** Use the project `APIKEY` (already extracted from the connection string). After each `cloudsync_network_init`/`cloudsync_network_init_custom` call, authenticate with: +```sql +SELECT cloudsync_network_set_apikey(''); +``` +No tokens are needed. Skip token creation entirely. + +**If RLS is enabled:** Create tokens for the test users. Create as many users as needed for the number of concurrent databases (assign 2 databases per user, or 1 per user if NUM_DBS <= 2). For each user N: ```bash @@ -78,9 +86,12 @@ curl -s -X "POST" "https:///v2/tokens" \ -d '{"name": "claude@sqlitecloud.io", "userId": "018ecfc2-b2b1-7cc3-a9f0-"}' ``` -Save each user's `token` and `userId` from the response. +Save each user's `token` and `userId` from the response. After each `cloudsync_network_init`/`cloudsync_network_init_custom` call, authenticate with: +```sql +SELECT cloudsync_network_set_token(''); +``` -If RLS is disabled, skip this step — tokens are not required. +**IMPORTANT:** Using a token when RLS is disabled will cause the server to silently reject all sent changes (send appears to succeed but data is not persisted remotely). Always use `cloudsync_network_set_apikey` when RLS is off. ### Step 5: Run the Concurrent Stress Test @@ -95,9 +106,9 @@ Create a bash script at `/tmp/stress_test_concurrent.sh` that: 2. **Defines a worker function** that runs in a subshell for each database: - Each worker logs all output to `/tmp/sync_concurrent_.log` - Each iteration does: - a. **DELETE all rows** → `send_changes()` → `check_changes()` - b. **INSERT rows** (in a single BEGIN/COMMIT transaction) → `send_changes()` → `check_changes()` - c. **UPDATE all rows** → `send_changes()` → `check_changes()` + a. **DELETE all rows** → `cloudsync_network_sync(100, 10)` + b. **INSERT rows** (in a single BEGIN/COMMIT transaction) → `cloudsync_network_sync(100, 10)` + c. **UPDATE all rows** → `cloudsync_network_sync(100, 10)` - Each session must: `.load` the extension, call `cloudsync_network_init()`, `cloudsync_network_set_token()` (if RLS), do the work, call `cloudsync_terminate()` - Include labeled output lines like `[DB][iter ] deleted/inserted/updated, count=` for grep-ability From c7ade3a86ab6585c330446a1f0c87ba847b37c72 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 17 Mar 2026 17:03:30 -0600 Subject: [PATCH 2/2] refactor: move all savepoint management from shared layer to platform wrappers (#17) * refactor: move all savepoint management from shared layer to platform wrappers Move database_begin_savepoint, database_commit_savepoint, and database_rollback_savepoint calls out of cloudsync_begin_alter and cloudsync_commit_alter (shared CRDT logic) into the platform-specific wrappers in cloudsync_sqlite.c and cloudsync_postgresql.c. This fixes the PostgreSQL "subtransaction left non-empty SPI stack" warning by ensuring SPI_connect() is called before the savepoint boundary, and creates architectural symmetry where shared code is pure business logic and all transaction management lives in platform wrappers. --- src/cloudsync.c | 20 +-------- src/cloudsync.h | 2 +- src/postgresql/cloudsync_postgresql.c | 63 +++++++++++++++++++++------ src/sqlite/cloudsync_sqlite.c | 28 +++++++++--- 4 files changed, 74 insertions(+), 39 deletions(-) diff --git a/src/cloudsync.c b/src/cloudsync.c index b1bdbaa..9987037 100644 --- a/src/cloudsync.c +++ b/src/cloudsync.c @@ -2273,16 +2273,10 @@ int cloudsync_begin_alter (cloudsync_context *data, const char *table_name) { return cloudsync_set_error(data, buffer, DBRES_MISUSE); } - // create a savepoint to manage the alter operations as a transaction - int rc = database_begin_savepoint(data, "cloudsync_alter"); - if (rc != DBRES_OK) { - return cloudsync_set_error(data, "Unable to create cloudsync_begin_alter savepoint", DBRES_MISUSE); - } - // retrieve primary key(s) char **names = NULL; int nrows = 0; - rc = database_pk_names(data, table_name, &names, &nrows); + int rc = database_pk_names(data, table_name, &names, &nrows); if (rc != DBRES_OK) { char buffer[1024]; snprintf(buffer, sizeof(buffer), "Unable to get primary keys for table %s", table_name); @@ -2311,7 +2305,6 @@ int cloudsync_begin_alter (cloudsync_context *data, const char *table_name) { return DBRES_OK; rollback_begin_alter: - database_rollback_savepoint(data, "cloudsync_alter"); if (names) table_pknames_free(names, nrows); return rc; } @@ -2430,18 +2423,9 @@ int cloudsync_commit_alter (cloudsync_context *data, const char *table_name) { rc = cloudsync_init_table(data, table_name, cloudsync_algo_name(algo_current), true); if (rc != DBRES_OK) goto rollback_finalize_alter; - // release savepoint - rc = database_commit_savepoint(data, "cloudsync_alter"); - if (rc != DBRES_OK) { - cloudsync_set_dberror(data); - goto rollback_finalize_alter; - } - - cloudsync_update_schema_hash(data); return DBRES_OK; - + rollback_finalize_alter: - database_rollback_savepoint(data, "cloudsync_alter"); if (table) table_set_pknames(table, NULL); return rc; } diff --git a/src/cloudsync.h b/src/cloudsync.h index 8673d5f..ff388e0 100644 --- a/src/cloudsync.h +++ b/src/cloudsync.h @@ -18,7 +18,7 @@ extern "C" { #endif -#define CLOUDSYNC_VERSION "0.9.200" +#define CLOUDSYNC_VERSION "0.9.201" #define CLOUDSYNC_MAX_TABLENAME_LEN 512 #define CLOUDSYNC_VALUE_NOTSET -1 diff --git a/src/postgresql/cloudsync_postgresql.c b/src/postgresql/cloudsync_postgresql.c index 9e6cd85..5809454 100644 --- a/src/postgresql/cloudsync_postgresql.c +++ b/src/postgresql/cloudsync_postgresql.c @@ -763,26 +763,27 @@ Datum pg_cloudsync_begin_alter (PG_FUNCTION_ARGS) { const char *table_name = text_to_cstring(PG_GETARG_TEXT_PP(0)); cloudsync_context *data = get_cloudsync_context(); int rc = DBRES_OK; - bool spi_connected = false; - int spi_rc = SPI_connect(); - if (spi_rc != SPI_OK_CONNECT) { - ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed: %d", spi_rc))); + if (SPI_connect() != SPI_OK_CONNECT) { + ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed"))); } - spi_connected = true; PG_TRY(); { + database_begin_savepoint(data, "cloudsync_alter"); rc = cloudsync_begin_alter(data, table_name); + if (rc != DBRES_OK) { + database_rollback_savepoint(data, "cloudsync_alter"); + } } PG_CATCH(); { - if (spi_connected) SPI_finish(); + SPI_finish(); PG_RE_THROW(); } PG_END_TRY(); - if (spi_connected) SPI_finish(); + SPI_finish(); if (rc != DBRES_OK) { ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), @@ -792,6 +793,14 @@ Datum pg_cloudsync_begin_alter (PG_FUNCTION_ARGS) { } // cloudsync_commit_alter - Commit schema alteration +// +// This wrapper manages SPI in two phases to avoid the PostgreSQL warning +// "subtransaction left non-empty SPI stack". The subtransaction was opened +// by a prior cloudsync_begin_alter call, so SPI_connect() here creates a +// connection at the subtransaction level. We must disconnect SPI before +// cloudsync_commit_alter releases that subtransaction, then reconnect +// for post-commit work (cloudsync_update_schema_hash). +// Prepared statements survive SPI_finish via SPI_keepplan/TopMemoryContext. PG_FUNCTION_INFO_V1(pg_cloudsync_commit_alter); Datum pg_cloudsync_commit_alter (PG_FUNCTION_ARGS) { if (PG_ARGISNULL(0)) { @@ -801,13 +810,11 @@ Datum pg_cloudsync_commit_alter (PG_FUNCTION_ARGS) { const char *table_name = text_to_cstring(PG_GETARG_TEXT_PP(0)); cloudsync_context *data = get_cloudsync_context(); int rc = DBRES_OK; - bool spi_connected = false; - int spi_rc = SPI_connect(); - if (spi_rc != SPI_OK_CONNECT) { - ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed: %d", spi_rc))); + // Phase 1: SPI work before savepoint release + if (SPI_connect() != SPI_OK_CONNECT) { + ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed"))); } - spi_connected = true; PG_TRY(); { @@ -815,15 +822,43 @@ Datum pg_cloudsync_commit_alter (PG_FUNCTION_ARGS) { } PG_CATCH(); { - if (spi_connected) SPI_finish(); + SPI_finish(); PG_RE_THROW(); } PG_END_TRY(); - if (spi_connected) SPI_finish(); + // Disconnect SPI before savepoint boundary + SPI_finish(); + if (rc != DBRES_OK) { + // Rollback savepoint (SPI disconnected, no warning) + database_rollback_savepoint(data, "cloudsync_alter"); ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("%s", cloudsync_errmsg(data)))); } + + // Release savepoint (SPI disconnected, no warning) + rc = database_commit_savepoint(data, "cloudsync_alter"); + if (rc != DBRES_OK) { + ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("Unable to release cloudsync_alter savepoint: %s", database_errmsg(data)))); + } + + // Phase 2: reconnect SPI for post-commit work + if (SPI_connect() != SPI_OK_CONNECT) { + ereport(ERROR, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("SPI_connect failed after savepoint release"))); + } + + PG_TRY(); + { + cloudsync_update_schema_hash(data); + } + PG_CATCH(); + { + SPI_finish(); + PG_RE_THROW(); + } + PG_END_TRY(); + + SPI_finish(); PG_RETURN_BOOL(true); } diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index ebdd1cc..f0b4670 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -919,15 +919,20 @@ void dbsync_init1 (sqlite3_context *context, int argc, sqlite3_value **argv) { void dbsync_begin_alter (sqlite3_context *context, int argc, sqlite3_value **argv) { DEBUG_FUNCTION("dbsync_begin_alter"); - - //retrieve table argument + const char *table_name = (const char *)database_value_text(argv[0]); - - // retrieve context cloudsync_context *data = (cloudsync_context *)sqlite3_user_data(context); - - int rc = cloudsync_begin_alter(data, table_name); + + int rc = database_begin_savepoint(data, "cloudsync_alter"); + if (rc != DBRES_OK) { + sqlite3_result_error(context, "Unable to create cloudsync_alter savepoint", -1); + sqlite3_result_error_code(context, rc); + return; + } + + rc = cloudsync_begin_alter(data, table_name); if (rc != DBRES_OK) { + database_rollback_savepoint(data, "cloudsync_alter"); sqlite3_result_error(context, cloudsync_errmsg(data), -1); sqlite3_result_error_code(context, rc); } @@ -944,9 +949,20 @@ void dbsync_commit_alter (sqlite3_context *context, int argc, sqlite3_value **ar int rc = cloudsync_commit_alter(data, table_name); if (rc != DBRES_OK) { + database_rollback_savepoint(data, "cloudsync_alter"); sqlite3_result_error(context, cloudsync_errmsg(data), -1); sqlite3_result_error_code(context, rc); + return; + } + + rc = database_commit_savepoint(data, "cloudsync_alter"); + if (rc != DBRES_OK) { + sqlite3_result_error(context, database_errmsg(data), -1); + sqlite3_result_error_code(context, rc); + return; } + + cloudsync_update_schema_hash(data); } // MARK: - Payload -