Preserve payload group atomicity across block materialization

CHANGELOG.md

@@ -10,6 +10,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.
 
 ## [1.1.4] - 2026-09-21
 

docs/internal/block-group-atomicity.md

@@ -0,0 +1,11 @@
+# Atomic application of mixed ordinary and block columns
+
+Before materializing a block column, payload apply flushes the row's ordinary columns so the base row exists. Previously this flush also released the PK group's savepoint. A later block-table write error therefore left the ordinary columns and their metadata committed while the block column was missing or stale.
+
+The pre-block flush now writes pending ordinary columns without releasing the group's savepoint or advancing the applied-row count. All blocks and ordinary columns remain inside that boundary until the PK, table or source database version changes. A rejected block rolls back that entire group, including metadata, blocks and resurrection changes. Previously completed groups retain their existing semantics; a failed apply does not advance the receive checkpoint. Fixing the underlying error and replaying the payload restores the complete row.
+
+## Validation
+
+`test_block_group_atomicity` in `test/review_regressions.c` runs 120 failure-and-retry cases across inserts, updates and resurrected rows, 3–10 blocks, rejection of first/middle/last blocks, and both autocommit and caller-owned transactions. It compares both directions of SQL EXCEPT for the base table, metadata and block table, checks the unchanged checkpoint and caller transaction, and verifies successful retry and duplicate delivery.
+
+The core suite and audit regressions pass under AddressSanitizer and UndefinedBehaviorSanitizer, with SQLite itself instrumented and zero outstanding SQLite memory. Restoring the original pre-block flush fails 360 assertions in the new tests. The independent branch also passes all 521 PostgreSQL 18.6 checks for shared-code compatibility. These tests use local databases; they do not validate a deployed cloud server.

src/cloudsync.c

@@ -4595,9 +4595,11 @@ int cloudsync_payload_apply (cloudsync_context *data, const char *payload, int b
         last_tbl_len = decoded_context.tbl_len;
 
         if (batch.count > 0 && cloudsync_payload_row_is_block(&decoded_context)) {
-            fail_rc = cloudsync_payload_group_flush(data, &batch);
+            // Materialization needs the ordinary columns to exist, but they and
+            // every block still belong to the same atomic PK group. Keep its
+            // savepoint open until the actual PK/table/db_version boundary.
+            fail_rc = merge_flush_pending(data);
             if (fail_rc != DBRES_OK) break;
-            applied = (int)i;
         }
 
         fail_rc = cloudsync_payload_group_open(data, &batch);

test/review_regressions.c

@@ -568,6 +568,67 @@ static void test_failed_apply_commit(void) {
     const char *files[] = {"commit-busy.db", "commit-busy.db-journal"};
     scratch_remove(files, 2);
 }
+static void test_block_group_atomicity(void) {
+    const char *schema = "CREATE TABLE t(id TEXT PRIMARY KEY NOT NULL, owner TEXT, body TEXT);"
+        "SELECT cloudsync_init('t'); SELECT cloudsync_set_column('t','body','algo','block');"
+        "CREATE TABLE caller_work(value TEXT);";
+    // Insert, update and resurrection; every block position; caller and internal
+    // transactions. Repeat with varying payload sizes to exercise batch reuse.
+    for (int trial = 0; trial < 120; trial++) {
+        int mode = trial % 3, blocks = 3 + (trial / 3) % 8;
+        int denied = (trial * 7) % blocks;
+        bool caller = (trial / 24) % 2;
+        sqlite3 *source = open_db(), *target = open_db();
+        CHECK(sql(source, schema) == SQLITE_OK && sql(target, schema) == SQLITE_OK);
+        if (mode) {
+            CHECK(sql(source, "INSERT INTO t VALUES('a','old','old body');") == SQLITE_OK);
+            CHECK(apply_payload(source, target) == SQLITE_ROW);
+            if (mode == 2) {
+                CHECK(sql(source, "DELETE FROM t;") == SQLITE_OK);
+                CHECK(apply_payload(source, target) == SQLITE_ROW);
+            }
+        }
+        char body[1024] = {0}, stmt[2048];
+        for (int j = 0; j < blocks; j++) {
+            char part[64];
+            snprintf(part, sizeof(part), "%sblock-%d-trial-%d", j ? "\n" : "", j, trial);
+            strcat(body, part);
+        }
+        snprintf(stmt, sizeof(stmt), "INSERT INTO t VALUES('a','new','%s') ON CONFLICT(id) DO UPDATE SET owner=excluded.owner,body=excluded.body;", body);
+        CHECK(sql(source, stmt) == SQLITE_OK);
+        CHECK(sql(target, "CREATE TEMP TABLE before_meta AS SELECT * FROM t_cloudsync;"
+                          "CREATE TEMP TABLE before_blocks AS SELECT * FROM t_cloudsync_blocks;"
+                          "CREATE TEMP TABLE before_data AS SELECT * FROM t;") == SQLITE_OK);
+        int64_t checkpoint = scalar(target, "SELECT coalesce((SELECT value FROM cloudsync_settings WHERE key='check_dbversion'),0)");
+        snprintf(stmt, sizeof(stmt), "CREATE TRIGGER deny_block BEFORE INSERT ON t_cloudsync_blocks "
+                 "WHEN NEW.col_value='block-%d-trial-%d' BEGIN SELECT RAISE(ABORT,'block rejected'); END", denied, trial);
+        CHECK(sql(target, stmt) == SQLITE_OK);
+        if (caller) CHECK(sql(target, "BEGIN; INSERT INTO caller_work VALUES('kept'); SAVEPOINT caller_sp;") == SQLITE_OK);
+        int rejected_rc = apply_payload(source, target);
+        CHECK(rejected_rc != SQLITE_ROW);
+        CHECK(strstr(sqlite3_errmsg(target), "block rejected") != NULL);
+        CHECK(sqlite3_get_autocommit(target) == !caller);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM t EXCEPT SELECT * FROM before_data)") == 0);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM before_data EXCEPT SELECT * FROM t)") == 0);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM t_cloudsync EXCEPT SELECT * FROM before_meta)") == 0);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM before_meta EXCEPT SELECT * FROM t_cloudsync)") == 0);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM t_cloudsync_blocks EXCEPT SELECT * FROM before_blocks)") == 0);
+        CHECK(scalar(target, "SELECT count(*) FROM (SELECT * FROM before_blocks EXCEPT SELECT * FROM t_cloudsync_blocks)") == 0);
+        CHECK(scalar(target, "SELECT coalesce((SELECT value FROM cloudsync_settings WHERE key='check_dbversion'),0)") == checkpoint);
+        if (caller) {
+            CHECK(scalar(target, "SELECT count(*) FROM caller_work") == 1);
+            CHECK(sql(target, "RELEASE caller_sp;") == SQLITE_OK);
+        }
+        CHECK(sql(target, "DROP TRIGGER deny_block;") == SQLITE_OK);
+        CHECK(apply_payload(source, target) == SQLITE_ROW);
+        snprintf(stmt, sizeof(stmt), "SELECT count(*) FROM t WHERE owner='new' AND body='%s'", body);
+        CHECK(scalar(target, stmt) == 1);
+        CHECK(apply_payload(source, target) == SQLITE_ROW);
+        CHECK(scalar(target, stmt) == 1);
+        if (caller) CHECK(sql(target, "COMMIT") == SQLITE_OK);
+        CHECK(close_db(source) == SQLITE_OK && close_db(target) == SQLITE_OK);
+    }
+}
 
 int main(void) {
     CHECK(sqlite3_config(SQLITE_CONFIG_GETMALLOC, &memory) == SQLITE_OK);
@@ -588,6 +649,7 @@ int main(void) {
     test_block_materialize_errors();
     test_block_migration_orphan();
     test_block_not_null_payload();
+    test_block_group_atomicity();
     test_refill_error();
     test_block_oom();
     cloudsync_memory_finalize();