From 1994297d8862ca0e8ed2d1b314a809eb84e8cb39 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Thu, 23 Apr 2026 13:32:31 -0600 Subject: [PATCH 1/6] Fix block LWW tombstone parity handling Make SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION preserve the caller's requested live/tombstone parity on conflict instead of blindly incrementing col_version. This fixes the block-LWW case where re-marking a live block could flip it to an even tombstone version and disappear from materialization, and keeps repeated tombstone writes even instead of resurrecting to an odd version. Qualify the PostgreSQL ON CONFLICT target-side col_version references to avoid ambiguous-column errors, and update the shared formatter call so both backends receive the right table-name substitutions. Add unit regressions for site-id resolution, double tombstone writes, and the block-LWW upsert tombstone scenario. --- src/cloudsync.c | 4 +- src/postgresql/sql_postgresql.c | 5 +- src/sqlite/sql_sqlite.c | 5 +- test/unit.c | 387 ++++++++++++++++++++++++++++++++ 4 files changed, 398 insertions(+), 3 deletions(-) diff --git a/src/cloudsync.c b/src/cloudsync.c index 0500709d..b965fd45 100644 --- a/src/cloudsync.c +++ b/src/cloudsync.c @@ -930,7 +930,9 @@ int table_add_stmts (cloudsync_table_context *table, int ncols) { if (rc != DBRES_OK) goto cleanup; // precompile the insert/update local row statement - sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION, table->meta_ref, table->meta_ref); + sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION, + table->meta_ref, table->meta_ref, + table->meta_ref, table->meta_ref); if (!sql) {rc = DBRES_NOMEM; goto cleanup;} DEBUG_SQL("meta_row_insert_update_stmt: %s", sql); diff --git a/src/postgresql/sql_postgresql.c b/src/postgresql/sql_postgresql.c index dac1c0da..c6a630be 100644 --- a/src/postgresql/sql_postgresql.c +++ b/src/postgresql/sql_postgresql.c @@ -339,7 +339,10 @@ const char * const SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION = "INSERT INTO %s (pk, col_name, col_version, db_version, seq, site_id) " "VALUES ($1, $2, $3, $4, $5, 0) " "ON CONFLICT (pk, col_name) DO UPDATE SET " - "col_version = %s.col_version + 1, db_version = $6, seq = $7, site_id = 0;"; + "col_version = CASE " + "WHEN (%s.col_version %% 2) = (excluded.col_version %% 2) THEN %s.col_version + 2 " + "ELSE %s.col_version + 1 END, " + "db_version = $6, seq = $7, site_id = 0;"; const char * const SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL = "DELETE FROM %s WHERE pk = $1 AND col_name != '%s';"; // TODO: match SQLite delete semantics diff --git a/src/sqlite/sql_sqlite.c b/src/sqlite/sql_sqlite.c index 229fe2b9..fb048bfe 100644 --- a/src/sqlite/sql_sqlite.c +++ b/src/sqlite/sql_sqlite.c @@ -188,7 +188,10 @@ const char * const SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION = "SELECT ?, ?, ?, ?, ?, 0 " "WHERE 1 " "ON CONFLICT DO UPDATE SET " - "col_version = \"%w\".col_version + 1, db_version = ?, seq = ?, site_id = 0;"; + "col_version = CASE " + "WHEN (col_version %% 2) = (excluded.col_version %% 2) THEN col_version + 2 " + "ELSE col_version + 1 END, " + "db_version = ?, seq = ?, site_id = 0;"; const char * const SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL = "DELETE FROM \"%w\" WHERE pk=? AND col_name!='%s';"; diff --git a/test/unit.c b/test/unit.c index 83fe06e7..52127532 100644 --- a/test/unit.c +++ b/test/unit.c @@ -9402,6 +9402,390 @@ static char *do_select_text(sqlite3 *db, const char *sql) { return val; } +// Regression: metadata may store the local site ordinal as 0 internally, but +// cloudsync_changes must still resolve local changes to the real local site-id +// blob and preserve remote site-ids on imported rows. +bool do_test_site_id_resolution(bool print_result, bool cleanup_databases) { + sqlite3 *db[2] = {NULL, NULL}; + time_t timestamp = time(NULL); + int rc; + char *site0 = NULL; + char *site1 = NULL; + char *sql = NULL; + bool ok = false; + + for (int i = 0; i < 2; i++) { + db[i] = do_create_database_file(i, timestamp, test_counter++); + if (!db[i]) goto fail; + + rc = sqlite3_exec(db[i], "CREATE TABLE docs (id TEXT NOT NULL PRIMARY KEY, title TEXT);", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + rc = sqlite3_exec(db[i], "SELECT cloudsync_init('docs');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + } + + site0 = do_select_text(db[0], "SELECT hex(cloudsync_siteid());"); + site1 = do_select_text(db[1], "SELECT hex(cloudsync_siteid());"); + if (!site0 || !site1) { + printf("site_id_resolution: unable to read local site ids\n"); + goto fail; + } + if (strcmp(site0, site1) == 0) { + printf("site_id_resolution: expected distinct site ids for the two databases\n"); + goto fail; + } + + rc = sqlite3_exec(db[0], "INSERT INTO docs (id, title) VALUES ('doc1', 'alpha');", NULL, NULL, NULL); + if (rc != SQLITE_OK) { + printf("site_id_resolution: local insert failed: %s\n", sqlite3_errmsg(db[0])); + goto fail; + } + + sql = sqlite3_mprintf( + "SELECT count(*) FROM cloudsync_changes " + "WHERE tbl='docs' AND pk=cloudsync_pk_encode('doc1') AND col_name='title' AND hex(site_id)='%q';", + site0); + if (!sql) goto fail; + int64_t local_rows = do_select_int(db[0], sql); + sqlite3_free(sql); + sql = NULL; + if (local_rows != 1) { + printf("site_id_resolution: expected local title change on db[0] to export with site_id %s, got %" PRId64 "\n", site0, local_rows); + goto fail; + } + + if (!do_merge_using_payload(db[0], db[1], false, true)) { + printf("site_id_resolution: merge 0->1 failed\n"); + goto fail; + } + + sql = sqlite3_mprintf( + "SELECT count(*) FROM cloudsync_changes " + "WHERE tbl='docs' AND pk=cloudsync_pk_encode('doc1') AND col_name='title' AND hex(site_id)='%q';", + site0); + if (!sql) goto fail; + int64_t imported_rows = do_select_int(db[1], sql); + sqlite3_free(sql); + sql = NULL; + if (imported_rows != 1) { + printf("site_id_resolution: expected imported title change on db[1] to preserve sender site_id %s, got %" PRId64 "\n", site0, imported_rows); + goto fail; + } + + sql = sqlite3_mprintf( + "SELECT count(*) FROM cloudsync_changes " + "WHERE tbl='docs' AND pk=cloudsync_pk_encode('doc1') AND col_name='title' AND hex(site_id)='%q';", + site1); + if (!sql) goto fail; + int64_t wrong_local_rows = do_select_int(db[1], sql); + sqlite3_free(sql); + sql = NULL; + if (wrong_local_rows != 0) { + printf("site_id_resolution: imported row on db[1] was rewritten to local site_id %s\n", site1); + goto fail; + } + + rc = sqlite3_exec(db[1], "UPDATE docs SET title='beta' WHERE id='doc1';", NULL, NULL, NULL); + if (rc != SQLITE_OK) { + printf("site_id_resolution: local update on db[1] failed: %s\n", sqlite3_errmsg(db[1])); + goto fail; + } + + sql = sqlite3_mprintf( + "SELECT count(*) FROM cloudsync_changes " + "WHERE tbl='docs' AND pk=cloudsync_pk_encode('doc1') AND col_name='title' AND hex(site_id)='%q';", + site1); + if (!sql) goto fail; + int64_t rewritten_local_rows = do_select_int(db[1], sql); + sqlite3_free(sql); + sql = NULL; + if (rewritten_local_rows != 1) { + printf("site_id_resolution: expected local update on db[1] to export with local site_id %s, got %" PRId64 "\n", site1, rewritten_local_rows); + goto fail; + } + + ok = true; + +fail: + if (sql) sqlite3_free(sql); + if (site0) sqlite3_free(site0); + if (site1) sqlite3_free(site1); + for (int i = 0; i < 2; i++) { + if (db[i]) close_db(db[i]); + } + return ok; +} + +// Regression test for bugs/BUG_block_lww_upsert_tombstone.md. +// Re-running the raw UPSERT against an already-live block must preserve +// liveness parity; if the bug is present, the block flips to a tombstone and +// disappears from materialization. +bool do_test_block_lww_upsert_tombstone_bug(int nclients, bool print_result, bool cleanup_databases) { + sqlite3 *db[1] = {NULL}; + cloudsync_context *ctx = NULL; + cloudsync_table_context *table = NULL; + time_t timestamp = time(NULL); + int rc; + char *pk = NULL; + int64_t pklen = 0; + char *block_colname = NULL; + + db[0] = do_create_database_file(0, timestamp, test_counter++); + if (!db[0]) return false; + + rc = sqlite3_exec(db[0], "CREATE TABLE docs (id TEXT NOT NULL PRIMARY KEY, body TEXT);", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + rc = sqlite3_exec(db[0], "SELECT cloudsync_init('docs');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + rc = sqlite3_exec(db[0], "SELECT cloudsync_set_column('docs', 'body', 'algo', 'block');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + + rc = sqlite3_exec(db[0], "INSERT INTO docs (id, body) VALUES ('doc1', 'AAA\nBBB\nCCC');", NULL, NULL, NULL); + if (rc != SQLITE_OK) { + printf("block_upsert_tombstone: INSERT failed: %s\n", sqlite3_errmsg(db[0])); + goto fail; + } + + int64_t alive_before = do_select_int(db[0], + "SELECT count(*) " + "FROM docs_cloudsync_blocks b " + "JOIN docs_cloudsync m ON b.pk = m.pk AND b.col_name = m.col_name " + "WHERE m.col_name LIKE 'body' || x'1f' || '%' AND m.col_version % 2 = 1;"); + if (alive_before != 3) { + printf("block_upsert_tombstone: expected 3 alive blocks before reproducer, got %" PRId64 "\n", alive_before); + goto fail; + } + + char *target_block = do_select_text(db[0], + "SELECT CAST(b.col_value AS TEXT) " + "FROM docs_cloudsync_blocks b " + "JOIN docs_cloudsync m ON b.pk = m.pk AND b.col_name = m.col_name " + "WHERE m.col_name LIKE 'body' || x'1f' || '%' " + "ORDER BY b.col_name " + "LIMIT 1;"); + if (!target_block) { + printf("block_upsert_tombstone: could not read target block content\n"); + goto fail; + } + + ctx = cloudsync_context_create(db[0]); + if (!ctx) goto fail; + if (cloudsync_init_table(ctx, "docs", "cls", CLOUDSYNC_INIT_FLAG_NONE) != DBRES_OK) { + printf("block_upsert_tombstone: secondary context init failed\n"); + sqlite3_free(target_block); + goto fail; + } + table = table_lookup(ctx, "docs"); + if (!table) { + printf("block_upsert_tombstone: secondary context could not resolve docs table\n"); + sqlite3_free(target_block); + goto fail; + } + + if (database_select_blob(ctx, "SELECT cloudsync_pk_encode('doc1');", &pk, &pklen) != DBRES_OK || !pk) { + printf("block_upsert_tombstone: could not encode primary key for helper call\n"); + sqlite3_free(target_block); + goto fail; + } + if (database_select_text(ctx, + "SELECT col_name FROM docs_cloudsync " + "WHERE col_name LIKE 'body' || x'1f' || '%' " + "ORDER BY col_name LIMIT 1;", + &block_colname) != DBRES_OK || !block_colname) { + printf("block_upsert_tombstone: could not resolve block col_name for helper call\n"); + sqlite3_free(target_block); + goto fail; + } + + rc = local_mark_insert_or_update_meta(table, pk, (size_t)pklen, block_colname, + cloudsync_dbversion(ctx), cloudsync_bumpseq(ctx)); + if (rc != SQLITE_OK) { + printf("block_upsert_tombstone: helper call failed\n"); + sqlite3_free(target_block); + goto fail; + } + + int64_t alive_after = do_select_int(db[0], + "SELECT count(*) " + "FROM docs_cloudsync_blocks b " + "JOIN docs_cloudsync m ON b.pk = m.pk AND b.col_name = m.col_name " + "WHERE m.col_name LIKE 'body' || x'1f' || '%' AND m.col_version % 2 = 1;"); + if (alive_after != 3) { + printf("block_upsert_tombstone: bug reproduced, alive blocks changed from 3 to %" PRId64 "\n", alive_after); + sqlite3_free(target_block); + goto fail; + } + + int64_t tombstones = do_select_int(db[0], + "SELECT count(*) " + "FROM docs_cloudsync " + "WHERE col_name LIKE 'body' || x'1f' || '%' AND col_version % 2 = 0;"); + if (tombstones != 0) { + printf("block_upsert_tombstone: bug reproduced, found %" PRId64 " tombstoned block rows\n", tombstones); + sqlite3_free(target_block); + goto fail; + } + + rc = sqlite3_exec(db[0], "SELECT cloudsync_text_materialize('docs', 'body', 'doc1');", NULL, NULL, NULL); + if (rc != SQLITE_OK) { + printf("block_upsert_tombstone: materialize failed: %s\n", sqlite3_errmsg(db[0])); + sqlite3_free(target_block); + goto fail; + } + + char *body = do_select_text(db[0], "SELECT body FROM docs WHERE id = 'doc1';"); + if (!body) { + printf("block_upsert_tombstone: materialized body is NULL\n"); + sqlite3_free(target_block); + goto fail; + } + if (strstr(body, target_block) == NULL) { + printf("block_upsert_tombstone: bug reproduced, materialized body lost block [%s] -> [%s]\n", target_block, body); + sqlite3_free(target_block); + sqlite3_free(body); + goto fail; + } + + sqlite3_free(target_block); + sqlite3_free(body); + if (block_colname) cloudsync_memory_free(block_colname); + if (pk) cloudsync_memory_free(pk); + if (ctx) cloudsync_context_free(ctx); + close_db(db[0]); + return true; + +fail: + if (block_colname) cloudsync_memory_free(block_colname); + if (pk) cloudsync_memory_free(pk); + if (ctx) cloudsync_context_free(ctx); + if (db[0]) close_db(db[0]); + return false; +} + +// Regression: writing the same block tombstone twice must preserve tombstone +// parity. Before the parity fix, the second write could resurrect the block +// metadata by producing col_version = 3. +bool do_test_block_lww_double_tombstone(bool print_result, bool cleanup_databases) { + sqlite3 *db[1] = {NULL}; + cloudsync_context *ctx = NULL; + cloudsync_table_context *table = NULL; + time_t timestamp = time(NULL); + int rc; + char *pk = NULL; + int64_t pklen = 0; + char *block_colname = NULL; + char *sql = NULL; + bool ok = false; + + db[0] = do_create_database_file(0, timestamp, test_counter++); + if (!db[0]) return false; + + rc = sqlite3_exec(db[0], "CREATE TABLE docs (id TEXT NOT NULL PRIMARY KEY, body TEXT);", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + rc = sqlite3_exec(db[0], "SELECT cloudsync_init('docs');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + rc = sqlite3_exec(db[0], "SELECT cloudsync_set_column('docs', 'body', 'algo', 'block');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto fail; + + rc = sqlite3_exec(db[0], "INSERT INTO docs (id, body) VALUES ('doc1', 'AAA\nBBB\nCCC');", NULL, NULL, NULL); + if (rc != SQLITE_OK) { + printf("block_double_tombstone: INSERT failed: %s\n", sqlite3_errmsg(db[0])); + goto fail; + } + + ctx = cloudsync_context_create(db[0]); + if (!ctx) goto fail; + if (cloudsync_init_table(ctx, "docs", "cls", CLOUDSYNC_INIT_FLAG_NONE) != DBRES_OK) { + printf("block_double_tombstone: secondary context init failed\n"); + goto fail; + } + table = table_lookup(ctx, "docs"); + if (!table) { + printf("block_double_tombstone: secondary context could not resolve docs table\n"); + goto fail; + } + + if (database_select_blob(ctx, "SELECT cloudsync_pk_encode('doc1');", &pk, &pklen) != DBRES_OK || !pk) { + printf("block_double_tombstone: could not encode primary key\n"); + goto fail; + } + if (database_select_text(ctx, + "SELECT col_name FROM docs_cloudsync " + "WHERE col_name LIKE 'body' || x'1f' || '%' " + "ORDER BY col_name LIMIT 1;", + &block_colname) != DBRES_OK || !block_colname) { + printf("block_double_tombstone: could not resolve block col_name\n"); + goto fail; + } + + rc = local_mark_delete_block_meta(table, pk, (size_t)pklen, block_colname, + cloudsync_dbversion(ctx), cloudsync_bumpseq(ctx)); + if (rc != SQLITE_OK) { + printf("block_double_tombstone: first tombstone helper call failed\n"); + goto fail; + } + rc = block_delete_value_external(ctx, table, pk, (size_t)pklen, block_colname); + if (rc != SQLITE_OK) { + printf("block_double_tombstone: first block delete failed\n"); + goto fail; + } + + rc = local_mark_delete_block_meta(table, pk, (size_t)pklen, block_colname, + cloudsync_dbversion(ctx), cloudsync_bumpseq(ctx)); + if (rc != SQLITE_OK) { + printf("block_double_tombstone: second tombstone helper call failed\n"); + goto fail; + } + rc = block_delete_value_external(ctx, table, pk, (size_t)pklen, block_colname); + if (rc != SQLITE_OK) { + printf("block_double_tombstone: second block delete failed\n"); + goto fail; + } + + sql = sqlite3_mprintf("SELECT col_version FROM docs_cloudsync WHERE pk = cloudsync_pk_encode('doc1') AND col_name = '%q';", + block_colname); + if (!sql) goto fail; + int64_t col_version = do_select_int(db[0], sql); + sqlite3_free(sql); + sql = NULL; + if (col_version != 4) { + printf("block_double_tombstone: expected col_version 4 after double tombstone, got %" PRId64 "\n", col_version); + goto fail; + } + + sql = sqlite3_mprintf("SELECT count(*) FROM docs_cloudsync WHERE pk = cloudsync_pk_encode('doc1') AND col_name = '%q' AND col_version %% 2 = 0;", + block_colname); + if (!sql) goto fail; + int64_t even_rows = do_select_int(db[0], sql); + sqlite3_free(sql); + sql = NULL; + if (even_rows != 1) { + printf("block_double_tombstone: expected 1 even tombstone row after double tombstone, got %" PRId64 "\n", even_rows); + goto fail; + } + + sql = sqlite3_mprintf("SELECT count(*) FROM docs_cloudsync_blocks WHERE pk = cloudsync_pk_encode('doc1') AND col_name = '%q';", + block_colname); + if (!sql) goto fail; + int64_t remaining_blocks = do_select_int(db[0], sql); + sqlite3_free(sql); + sql = NULL; + if (remaining_blocks != 0) { + printf("block_double_tombstone: expected deleted block value to stay absent, got %" PRId64 " rows\n", remaining_blocks); + goto fail; + } + + ok = true; + +fail: + if (sql) sqlite3_free(sql); + if (block_colname) cloudsync_memory_free(block_colname); + if (pk) cloudsync_memory_free(pk); + if (ctx) cloudsync_context_free(ctx); + if (db[0]) close_db(db[0]); + return ok; +} + // Test: enabling block-level LWW on a table that already contains rows migrates // existing data into the blocks table so it is not silently ignored. // Verifies: @@ -13641,8 +14025,11 @@ int main (int argc, const char * argv[]) { result += test_report("Stale Table Settings Dropped Meta:", do_test_stale_table_settings_dropped_meta(cleanup_databases)); result += test_report("DBVersion Rebuild Error:", do_test_dbversion_rebuild_error()); result += test_report("Uninit Error Messages:", do_test_uninit_error_messages()); + result += test_report("Site ID Resolution:", do_test_site_id_resolution(print_result, cleanup_databases)); result += test_report("Block LWW Existing Data:", do_test_block_lww_existing_data(cleanup_databases)); result += test_report("Block Column Reload:", do_test_block_column_reload(cleanup_databases)); + result += test_report("Block LWW Double Tombstone:", do_test_block_lww_double_tombstone(print_result, cleanup_databases)); + result += test_report("Block LWW UPSERT Tombstone Bug:", do_test_block_lww_upsert_tombstone_bug(2, print_result, cleanup_databases)); result += test_report("CB Error Cleanup:", do_test_context_cb_error_cleanup()); finalize: From 1e8baa11b24312940122baeb4d9a6351d93c8319 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Thu, 23 Apr 2026 13:35:36 -0600 Subject: [PATCH 2/6] fix(docker): bind debug postgres to all interfaces MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The debug image builds postgres from source, which uses the upstream postgresql.conf.sample where listen_addresses defaults to localhost. The Debian-packaged postgres in the release image patches the sample to '*', so this only bites the debug compose target — host psql can't reach the container after a fresh volume init. Pass `-c listen_addresses=*` via the compose command override so the fix is applied on every fresh initdb without rebuilding the image. --- docker/postgresql/docker-compose.debug.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/docker/postgresql/docker-compose.debug.yml b/docker/postgresql/docker-compose.debug.yml index c5f38f2e..f0f4b355 100644 --- a/docker/postgresql/docker-compose.debug.yml +++ b/docker/postgresql/docker-compose.debug.yml @@ -4,6 +4,7 @@ services: context: ../.. dockerfile: docker/postgresql/Dockerfile.debug-no-optimization container_name: cloudsync-postgres + command: postgres -c listen_addresses=* environment: POSTGRES_USER: postgres POSTGRES_PASSWORD: postgres From 333bb2e95cff31f1a7d176508f58fbc90c46822e Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Thu, 23 Apr 2026 14:51:29 -0600 Subject: [PATCH 3/6] Add multi-cycle delete/resurrect regression test --- test/unit.c | 76 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 76 insertions(+) diff --git a/test/unit.c b/test/unit.c index 52127532..f168f9c2 100644 --- a/test/unit.c +++ b/test/unit.c @@ -13689,6 +13689,81 @@ bool do_test_delete_resurrect_ordering (int nclients, bool print_result, bool cl return result; } +bool do_test_delete_resurrect_multi_cycle (int nclients, bool print_result, bool cleanup_databases) { + sqlite3 *db[3] = {NULL, NULL, NULL}; + bool result = false; + + time_t timestamp = time(NULL); + int saved_counter = test_counter; + + for (int i = 0; i < 3; i++) { + db[i] = do_create_database_file(i, timestamp, test_counter++); + if (!db[i]) return false; + + int rc = sqlite3_exec(db[i], "CREATE TABLE test_tbl (id TEXT PRIMARY KEY, val TEXT);", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto finalize; + rc = sqlite3_exec(db[i], "SELECT cloudsync_init('test_tbl');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto finalize; + } + + // Initial insert and full sync + if (sqlite3_exec(db[0], "INSERT INTO test_tbl VALUES ('row1', 'original');", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[0], db[1], false, true)) goto finalize; + if (!do_merge_using_payload(db[0], db[2], false, true)) goto finalize; + + // Cycle 1: A deletes, B resurrects, C receives resurrection before stale delete. + if (sqlite3_exec(db[0], "DELETE FROM test_tbl WHERE id = 'row1';", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[0], db[1], true, true)) goto finalize; + + if (sqlite3_exec(db[1], "INSERT INTO test_tbl VALUES ('row1', 'resurrected_by_b');", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[1], db[2], true, true)) goto finalize; + if (!do_merge_using_payload(db[0], db[2], true, true)) goto finalize; + if (!do_merge_using_payload(db[1], db[0], true, true)) goto finalize; + + // Cycle 2: B deletes, A resurrects, C again receives resurrection before stale delete. + if (sqlite3_exec(db[1], "DELETE FROM test_tbl WHERE id = 'row1';", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[1], db[0], true, true)) goto finalize; + + if (sqlite3_exec(db[0], "INSERT INTO test_tbl VALUES ('row1', 'resurrected_by_a_again');", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[0], db[2], true, true)) goto finalize; + if (!do_merge_using_payload(db[1], db[2], true, true)) goto finalize; + if (!do_merge_using_payload(db[0], db[1], true, true)) goto finalize; + + const char *query = "SELECT * FROM test_tbl ORDER BY id;"; + result = do_compare_queries(db[0], query, db[1], query, -1, -1, print_result); + if (result) result = do_compare_queries(db[0], query, db[2], query, -1, -1, print_result); + + if (result) { + for (int i = 0; i < 3; i++) { + int64_t count = do_select_int(db[i], "SELECT COUNT(*) FROM test_tbl WHERE id = 'row1';"); + int64_t live_sentinel = do_select_int(db[i], "SELECT COUNT(*) FROM test_tbl_cloudsync WHERE col_name = '__[RIP]__' AND col_version % 2 = 1;"); + + if (count != 1) { + printf("delete_resurrect_multi_cycle: expected row1 to exist on db[%d], count=%" PRId64 "\n", i, count); + result = false; + break; + } + + if (live_sentinel != 1) { + printf("delete_resurrect_multi_cycle: expected 1 live sentinel on db[%d], got=%" PRId64 "\n", i, live_sentinel); + result = false; + break; + } + } + } + +finalize: + for (int i = 0; i < 3; i++) { + if (db[i]) close_db(db[i]); + if (cleanup_databases) { + char buf[256]; + do_build_database_path(buf, i, timestamp, saved_counter++); + file_delete_internal(buf); + } + } + return result; +} + bool do_test_large_composite_pk (int nclients, bool print_result, bool cleanup_databases) { sqlite3 *db[2] = {NULL, NULL}; bool result = false; @@ -14019,6 +14094,7 @@ int main (int argc, const char * argv[]) { result += test_report("Payload Idempotency Test:", do_test_payload_idempotency(2, print_result, cleanup_databases)); result += test_report("CL Tiebreak Test:", do_test_causal_length_tiebreak(3, print_result, cleanup_databases)); result += test_report("Delete/Resurrect Order:", do_test_delete_resurrect_ordering(3, print_result, cleanup_databases)); + result += test_report("Delete/Resurrect Multi-Cycle:", do_test_delete_resurrect_multi_cycle(3, print_result, cleanup_databases)); result += test_report("Large Composite PK Test:", do_test_large_composite_pk(2, print_result, cleanup_databases)); result += test_report("Schema Hash Mismatch:", do_test_schema_hash_mismatch(2, print_result, cleanup_databases)); result += test_report("Stale Table Settings:", do_test_stale_table_settings(cleanup_databases)); From b86b6f13cea6051088a8dff8a978e0795a0deb1e Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 22 Sep 2026 12:48:12 -0600 Subject: [PATCH 4/6] test: cover the block metadata parity bug through SQL only INSERT OR REPLACE on a row with a block column rewrites every block while its metadata still exists. The unconditional version bump marked each one deleted, so the peer applied them as deletions and dropped the row, while the source still read correctly. The existing tests call the metadata helpers directly; this one uses only SQL and checks what the peer ends up with. Co-Authored-By: Claude Opus 5 (1M context) --- test/unit.c | 62 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/test/unit.c b/test/unit.c index f168f9c2..9f89dedc 100644 --- a/test/unit.c +++ b/test/unit.c @@ -13689,6 +13689,67 @@ bool do_test_delete_resurrect_ordering (int nclients, bool print_result, bool cl return result; } +bool do_test_block_replace_sync (int nclients, bool print_result, bool cleanup_databases) { + sqlite3 *db[2] = {NULL, NULL}; + bool result = false; + + time_t timestamp = time(NULL); + int saved_counter = test_counter; + + for (int i = 0; i < 2; i++) { + db[i] = do_create_database_file(i, timestamp, test_counter++); + if (!db[i]) return false; + + int rc = sqlite3_exec(db[i], "CREATE TABLE docs (id TEXT NOT NULL PRIMARY KEY, body TEXT);", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto finalize; + rc = sqlite3_exec(db[i], "SELECT cloudsync_init('docs');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto finalize; + rc = sqlite3_exec(db[i], "SELECT cloudsync_set_column('docs', 'body', 'algo', 'block');", NULL, NULL, NULL); + if (rc != SQLITE_OK) goto finalize; + } + + if (sqlite3_exec(db[0], "INSERT INTO docs VALUES ('d1', 'L1' || char(10) || 'L2');", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + if (!do_merge_using_payload(db[0], db[1], false, true)) goto finalize; + + // INSERT OR REPLACE rewrites every block of the row while its metadata still + // exists. Each rewrite must keep the block live (odd col_version): an + // unconditional bump tombstoned them, and the peer then dropped the row. + if (sqlite3_exec(db[0], "INSERT OR REPLACE INTO docs VALUES ('d1', 'N1' || char(10) || 'N2');", NULL, NULL, NULL) != SQLITE_OK) goto finalize; + + int64_t tombstoned = do_select_int(db[0], + "SELECT COUNT(*) FROM docs_cloudsync_blocks b " + "JOIN docs_cloudsync m ON b.pk = m.pk AND b.col_name = m.col_name " + "WHERE m.col_version % 2 = 0;"); + if (tombstoned != 0) { + printf("block_replace_sync: %" PRId64 " stored block(s) marked as deleted after INSERT OR REPLACE\n", tombstoned); + goto finalize; + } + + if (!do_merge_using_payload(db[0], db[1], false, true)) goto finalize; + + const char *query = "SELECT id, body FROM docs ORDER BY id;"; + result = do_compare_queries(db[0], query, db[1], query, -1, -1, print_result); + + if (result) { + int64_t count = do_select_int(db[1], "SELECT COUNT(*) FROM docs WHERE id = 'd1' AND body = 'N1' || char(10) || 'N2';"); + if (count != 1) { + printf("block_replace_sync: expected the replaced body on the peer, count=%" PRId64 "\n", count); + result = false; + } + } + +finalize: + for (int i = 0; i < 2; i++) { + if (db[i]) close_db(db[i]); + if (cleanup_databases) { + char buf[256]; + do_build_database_path(buf, i, timestamp, saved_counter++); + file_delete_internal(buf); + } + } + return result; +} + bool do_test_delete_resurrect_multi_cycle (int nclients, bool print_result, bool cleanup_databases) { sqlite3 *db[3] = {NULL, NULL, NULL}; bool result = false; @@ -14095,6 +14156,7 @@ int main (int argc, const char * argv[]) { result += test_report("CL Tiebreak Test:", do_test_causal_length_tiebreak(3, print_result, cleanup_databases)); result += test_report("Delete/Resurrect Order:", do_test_delete_resurrect_ordering(3, print_result, cleanup_databases)); result += test_report("Delete/Resurrect Multi-Cycle:", do_test_delete_resurrect_multi_cycle(3, print_result, cleanup_databases)); + result += test_report("Block Replace Sync:", do_test_block_replace_sync(2, print_result, cleanup_databases)); result += test_report("Large Composite PK Test:", do_test_large_composite_pk(2, print_result, cleanup_databases)); result += test_report("Schema Hash Mismatch:", do_test_schema_hash_mismatch(2, print_result, cleanup_databases)); result += test_report("Stale Table Settings:", do_test_stale_table_settings(cleanup_databases)); From 54764fc7fd352831893fb3d259fb0818af078f18 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 22 Sep 2026 12:48:12 -0600 Subject: [PATCH 5/6] docs: changelog entry for the block metadata parity fix Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d7b1063f..dcd45940 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). - **SQLite: a payload whose commit fails no longer leaves its transaction open.** When `cloudsync_payload_apply` started the transaction itself and the commit then failed — a deferred foreign key violated at commit, or `SQLITE_BUSY` because a reader held the database — the transaction stayed open: the uncommitted rows remained visible on the connection and the next `BEGIN` failed. The failed transaction is now rolled back and the original error is returned. Changes from earlier source versions that were already committed are kept, the receive checkpoint does not move, and the rolled-back rows are no longer counted as applied, so delivering the payload again applies it. A transaction or savepoint opened by the caller is still left to the caller. - **PostgreSQL: applying a payload read from a table works at any savepoint depth.** Inside 126 or more savepoints, `SELECT cloudsync_payload_apply(payload) FROM some_table` still failed with `buffer pin ... is not owned by resource owner SubTransaction` (and a caught error at that depth could abort an assertion-enabled server): cloudsync recorded the caller's resource owner and memory context for at most 128 nesting levels, counting its own internal savepoints, and silently stopped restoring them beyond that. The fixed limit is gone; only PostgreSQL's own resource limits apply. - **A received row with a block column is applied all or nothing.** When a row's block value failed to write (a trigger that raises, a constraint, a policy), its ordinary columns and their sync metadata stayed applied, leaving the row with a missing or stale block value. The row's ordinary columns, metadata and blocks now roll back together, the receive checkpoint does not move, and delivering the payload again applies the whole row once the cause is fixed. Rows rolled back this way are no longer counted as applied. +- **A block column rewritten in place is no longer lost by the peers it syncs to.** Rewriting a row that holds a block column without changing its primary key — `INSERT OR REPLACE`, or any write that rewrites the same block positions — marked every rewritten block as deleted in the sync metadata while keeping its content locally. The local row still read correctly, so nothing looked wrong, but the peers received the blocks as deleted and dropped the row. The metadata write now keeps a block live when it is written as live, and deleted when it is written as deleted, instead of flipping the two on every write. The same flip affected a row's deletion marker. ## [1.1.4] - 2026-09-21 From 12af68902d7c2586f49ceff80d97f490fbe1f1f5 Mon Sep 17 00:00:00 2001 From: Andrea Donetti Date: Tue, 22 Sep 2026 13:02:11 -0600 Subject: [PATCH 6/6] refactor(sqlite): qualify the raw upsert's column so both backends take the same arguments The shared caller passes the table name four times, which PostgreSQL uses and SQLite left unused after the parity change. Co-Authored-By: Claude Opus 5 (1M context) --- src/sqlite/sql_sqlite.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/sqlite/sql_sqlite.c b/src/sqlite/sql_sqlite.c index fb048bfe..19e0c733 100644 --- a/src/sqlite/sql_sqlite.c +++ b/src/sqlite/sql_sqlite.c @@ -189,8 +189,8 @@ const char * const SQL_CLOUDSYNC_UPSERT_RAW_COLVERSION = "WHERE 1 " "ON CONFLICT DO UPDATE SET " "col_version = CASE " - "WHEN (col_version %% 2) = (excluded.col_version %% 2) THEN col_version + 2 " - "ELSE col_version + 1 END, " + "WHEN (\"%w\".col_version %% 2) = (excluded.col_version %% 2) THEN \"%w\".col_version + 2 " + "ELSE \"%w\".col_version + 1 END, " "db_version = ?, seq = ?, site_id = 0;"; const char * const SQL_CLOUDSYNC_DELETE_PK_EXCEPT_COL =