Fix stale block content after SQLite INSERT OR REPLACE

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Fixed
 
+- **A row rewritten in one statement no longer keeps the old blocks of a shorter value.** Writing a whole row rewrites its block column from the first position, so any block past the end of the new value stayed stored. On SQLite `INSERT OR REPLACE` skips the old row's delete trigger unless `recursive_triggers` is on, so those leftovers kept their metadata and were delivered as content: replacing `AAA\nBBB\nCCC` with `ZZZ` left `ZZZ\nBBB\nCCC` on the peers and on a later local read. On PostgreSQL they carried no metadata, so peers were unaffected, but they stayed in the blocks table for the life of the row. Blocks the new value does not cover are now retired with it.
 - **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.

docs/internal/block-rewrite-leftovers.md

@@ -0,0 +1,17 @@
+# Leftover blocks when a whole row is written
+
+Deleting a row keeps the values of its block columns in the blocks table; only their metadata goes away. Writing the row again rewrites the column from its first position, so the new blocks overwrite the old ones position by position and any block beyond the new value's length stays behind.
+
+On SQLite that leaks into the data. `INSERT OR REPLACE` skips the old row's delete trigger while `recursive_triggers` is disabled (the default), so the leftover blocks keep their live metadata: replacing `AAA\nBBB\nCCC` with `ZZZ` left `ZZZ\nBBB\nCCC` on the replicas and on a later local materialization. On PostgreSQL the leftovers carry no metadata, so they never reach a replica, but they stay in the blocks table for as long as the row exists.
+
+`local_block_update` now reads the column's stored blocks on that path too. They stay out of the diff — a whole-row write is not an edit of the previous value — and whatever the new value does not rewrite is tombstoned and its value removed. A failure there is reported like any other block write, so the enclosing statement rolls back instead of committing half a column. The parity-preserving metadata upsert from #46 is required, because the rewritten positions land on rows a previous write tombstoned.
+
+Ordinary INSERT behavior is unchanged, including the convention that NULL block text is stored as one empty block and materializes as empty text.
+
+## Regression coverage
+
+`test/unit.c` runs 120 replacement cycles with recursive triggers OFF and ON, covering shorter, longer, empty, NULL, duplicate-line and trailing-delimiter values. An intervening UPDATE creates fractional block positions. Each cycle checks duplicate payload delivery, materialization on both source and replica, a second block column and an untouched row. An injected block-write failure checks that the replacement rolls back.
+
+`test/postgresql/64_block_rewrite_leftovers.sql` covers the same shape on PostgreSQL: a row recreated with fewer blocks, and an upsert of the whole row, must leave one stored block and deliver the shorter value to a replica.
+
+Both fail against the previous implementation: the SQLite test on content, the PostgreSQL one on the leftover rows.

src/cloudsync.c

@@ -4719,35 +4719,39 @@ int local_block_update(cloudsync_context *data, cloudsync_table_context *table,
     const char *col = table_colname(table, column);
     block_list_t *old = block_list_create_empty();
     block_list_t *next = (text || initial) ? block_split(text ? text : "", table_col_delimiter(table, column)) : block_list_create_empty();
+    // Blocks already stored for this column. An update diffs against them; a write of a
+    // whole row keeps them out of the diff — it rewrites the column from its first
+    // position — and retires below whatever the new value does not cover. A row can hold
+    // them at that point because deleting a row keeps its block values, and SQLite's
+    // INSERT OR REPLACE skips the delete trigger altogether.
+    block_list_t *stored = block_list_create_empty();
     block_diff_t *diff = NULL;
     const char **parts = NULL;
     dbvm_t *vm = NULL;
     char *sql = NULL;
-    if (!old || !next) goto done;
-    if (!initial) {
+    if (!old || !next || !stored) goto done;
 #ifdef CLOUDSYNC_POSTGRESQL_BUILD
-        sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=$1 ORDER BY col_name COLLATE \"C\"", table_blocks_ref(table));
+    sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=$1 ORDER BY col_name COLLATE \"C\"", table_blocks_ref(table));
 #else
-        sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=?1 ORDER BY col_name COLLATE BINARY", table_blocks_ref(table));
+    sql = cloudsync_memory_mprintf("SELECT col_name, col_value FROM %s WHERE pk=?1 ORDER BY col_name COLLATE BINARY", table_blocks_ref(table));
 #endif
-        if (!sql) goto done;
-        rc = databasevm_prepare(data, sql, &vm, 0);
-        if (rc != DBRES_OK) goto done;
-        rc = databasevm_bind_blob(vm, 1, pk, (int)pklen);
-        if (rc != DBRES_OK) goto done;
-        while ((rc = databasevm_step(vm)) == DBRES_ROW) {
-            const char *name = database_column_text(vm, 0);
-            const char *value = database_column_text(vm, 1);
-            const char *pos = block_extract_position_id(name);
-            /* Literal prefix comparison: SQL LIKE would mix columns containing % or _. */
-            if (pos && (size_t)(pos - name - 1) == strlen(col) && memcmp(name, col, strlen(col)) == 0) {
-                if (!block_list_add(old, value ? value : "", pos)) { rc = DBRES_NOMEM; goto done; }
-            }
+    if (!sql) goto done;
+    rc = databasevm_prepare(data, sql, &vm, 0);
+    if (rc != DBRES_OK) goto done;
+    rc = databasevm_bind_blob(vm, 1, pk, (int)pklen);
+    if (rc != DBRES_OK) goto done;
+    while ((rc = databasevm_step(vm)) == DBRES_ROW) {
+        const char *name = database_column_text(vm, 0);
+        const char *value = database_column_text(vm, 1);
+        const char *pos = block_extract_position_id(name);
+        /* Literal prefix comparison: SQL LIKE would mix columns containing % or _. */
+        if (pos && (size_t)(pos - name - 1) == strlen(col) && memcmp(name, col, strlen(col)) == 0) {
+            if (!block_list_add(initial ? stored : old, value ? value : "", pos)) { rc = DBRES_NOMEM; goto done; }
         }
-        if (rc != DBRES_DONE) goto done;
-        databasevm_finalize(vm);
-        vm = NULL;
     }
+    if (rc != DBRES_DONE) goto done;
+    databasevm_finalize(vm);
+    vm = NULL;
     rc = DBRES_NOMEM;
     if (next->count) {
         parts = cloudsync_memory_alloc((uint64_t)next->count * sizeof(*parts));
@@ -4778,13 +4782,30 @@ int local_block_update(cloudsync_context *data, cloudsync_table_context *table,
         cloudsync_memory_free(name);
         if (rc != DBRES_OK) break;
     }
+    // Retire the stored blocks the new value left untouched: their positions were not
+    // rewritten above, so they would stay live here and reach the peers as content.
+    for (int i = 0; rc == DBRES_OK && i < stored->count; i++) {
+        const char *pos = stored->entries[i].position_id;
+        bool rewritten = false;
+        for (int j = 0; j < diff->count && !rewritten; j++) {
+            rewritten = (diff->entries[j].type != BLOCK_DIFF_REMOVED &&
+                         strcmp(diff->entries[j].position_id, pos) == 0);
+        }
+        if (rewritten) continue;
+        char *name = block_build_colname(col, pos);
+        if (!name) { rc = DBRES_NOMEM; break; }
+        rc = local_mark_delete_block_meta(table, pk, pklen, name, version, cloudsync_bumpseq(data));
+        if (rc == DBRES_OK) rc = block_delete_value_external(data, table, pk, pklen, name);
+        cloudsync_memory_free(name);
+    }
 done:
     if (vm) databasevm_finalize(vm);
     cloudsync_memory_free(sql);
     cloudsync_memory_free((void *)parts);
     block_diff_free(diff);
     block_list_free(old);
     block_list_free(next);
+    block_list_free(stored);
     if (rc != DBRES_OK) {
         char message[512];
         snprintf(message, sizeof(message), "Unable to write the blocks of column \"%s\" of table \"%s\"", col, table->name);

test/postgresql/64_block_rewrite_leftovers.sql

@@ -0,0 +1,93 @@
+-- Rewriting a whole row must not leave a shorter block column's old blocks behind.
+-- Deleting a row keeps its block values, so recreating it with fewer blocks used to
+-- leave the extra ones stored, and a peer could still receive them as content.
+
+\set testid '64-block-rewrite'
+\ir helper_test_init.sql
+
+\connect postgres
+\ir helper_psql_conn_setup.sql
+DROP DATABASE IF EXISTS cloudsync_test_64_source;
+DROP DATABASE IF EXISTS cloudsync_test_64_target;
+CREATE DATABASE cloudsync_test_64_source;
+CREATE DATABASE cloudsync_test_64_target;
+
+\connect cloudsync_test_64_source
+\ir helper_psql_conn_setup.sql
+CREATE EXTENSION IF NOT EXISTS cloudsync;
+CREATE TABLE docs (id TEXT PRIMARY KEY NOT NULL, body TEXT);
+SELECT cloudsync_init('docs', 'CLS', 1) AS _init \gset
+SELECT cloudsync_set_column('docs', 'body', 'algo', 'block') AS _setcol \gset
+
+\connect cloudsync_test_64_target
+\ir helper_psql_conn_setup.sql
+CREATE EXTENSION IF NOT EXISTS cloudsync;
+CREATE TABLE docs (id TEXT PRIMARY KEY NOT NULL, body TEXT);
+SELECT cloudsync_init('docs', 'CLS', 1) AS _init \gset
+SELECT cloudsync_set_column('docs', 'body', 'algo', 'block') AS _setcol \gset
+
+-- Three blocks, delivered to the target.
+\connect cloudsync_test_64_source
+INSERT INTO docs (id, body) VALUES ('d1', 'AAA' || chr(10) || 'BBB' || chr(10) || 'CCC');
+SELECT encode(cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq), 'hex') AS payload1
+FROM cloudsync_changes WHERE site_id = cloudsync_siteid() \gset
+\connect cloudsync_test_64_target
+SELECT cloudsync_payload_apply(decode(:'payload1', 'hex')) AS _apply1 \gset
+SELECT cloudsync_text_materialize('docs', 'body', cloudsync_pk_encode('d1')) AS _mat1 \gset
+SELECT (SELECT body FROM docs WHERE id = 'd1') = 'AAA' || chr(10) || 'BBB' || chr(10) || 'CCC' AS first_ok \gset
+\if :first_ok
+\echo [PASS] (:testid) three blocks delivered
+\else
+\echo [FAIL] (:testid) three blocks not delivered
+SELECT (:fail::int + 1) AS fail \gset
+\endif
+
+-- Recreate the row with a single block: the other two must not survive.
+\connect cloudsync_test_64_source
+SELECT coalesce(max(db_version), 0) AS before_short FROM cloudsync_changes \gset
+DELETE FROM docs WHERE id = 'd1';
+INSERT INTO docs (id, body) VALUES ('d1', 'ZZZ');
+SELECT count(*) AS stored_blocks FROM docs_cloudsync_blocks WHERE pk = cloudsync_pk_encode('d1') \gset
+SELECT count(*) AS live_meta FROM docs_cloudsync
+ WHERE pk = cloudsync_pk_encode('d1') AND col_name LIKE 'body' || chr(31) || '%' AND col_version % 2 = 1 \gset
+SELECT (:stored_blocks::int = 1 AND :live_meta::int = 1) AS source_ok \gset
+\if :source_ok
+\echo [PASS] (:testid) shorter rewrite leaves one stored block and one live block
+\else
+\echo [FAIL] (:testid) shorter rewrite left :stored_blocks stored block(s), :live_meta live
+SELECT (:fail::int + 1) AS fail \gset
+\endif
+
+SELECT encode(cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq), 'hex') AS payload2
+FROM cloudsync_changes WHERE site_id = cloudsync_siteid() AND db_version > :before_short::bigint \gset
+\connect cloudsync_test_64_target
+SELECT cloudsync_payload_apply(decode(:'payload2', 'hex')) AS _apply2 \gset
+SELECT cloudsync_text_materialize('docs', 'body', cloudsync_pk_encode('d1')) AS _mat2 \gset
+SELECT (SELECT body FROM docs WHERE id = 'd1') = 'ZZZ' AS target_ok \gset
+SELECT count(*) AS target_blocks FROM docs_cloudsync_blocks WHERE pk = cloudsync_pk_encode('d1') \gset
+\if :target_ok
+\echo [PASS] (:testid) peer holds the shorter value
+\else
+\echo [FAIL] (:testid) peer body is [:target_ok] with :target_blocks stored block(s)
+SELECT (:fail::int + 1) AS fail \gset
+\endif
+
+-- An upsert of the whole row goes through the same path.
+\connect cloudsync_test_64_source
+INSERT INTO docs (id, body) VALUES ('d1', 'K1' || chr(10) || 'K2' || chr(10) || 'K3')
+  ON CONFLICT (id) DO UPDATE SET body = excluded.body;
+INSERT INTO docs (id, body) VALUES ('d1', 'K1')
+  ON CONFLICT (id) DO UPDATE SET body = excluded.body;
+SELECT count(*) AS upsert_blocks FROM docs_cloudsync_blocks WHERE pk = cloudsync_pk_encode('d1') \gset
+SELECT (:upsert_blocks::int = 1) AS upsert_ok \gset
+\if :upsert_ok
+\echo [PASS] (:testid) upsert to a shorter value keeps one stored block
+\else
+\echo [FAIL] (:testid) upsert to a shorter value left :upsert_blocks stored block(s)
+SELECT (:fail::int + 1) AS fail \gset
+\endif
+
+\connect postgres
+\ir helper_psql_conn_setup.sql
+DROP DATABASE IF EXISTS cloudsync_test_64_source;
+DROP DATABASE IF EXISTS cloudsync_test_64_target;

test/postgresql/full_test.sql

@@ -71,6 +71,7 @@
 \ir 61_fragment_cleanup_backlog.sql
 \ir 62_deferred_fk_caller_commit.sql
 \ir 63_deep_savepoints.sql
+\ir 64_block_rewrite_leftovers.sql
 
 -- 'Test summary'
 \echo '\nTest summary:'

test/unit.c

@@ -11482,6 +11482,60 @@ bool do_test_block_lww_text_to_null(int nclients, bool print_result, bool cleanu
 }
 
 // Test: Payload-based sync for block columns (vs row-by-row do_merge_values)
+// REPLACE skips delete triggers by default: removed blocks must not survive.
+bool do_test_block_lww_replace(void) {
+    const char *values[] = {"ZZZ", "", NULL, "one\ntwo\nthree\nfour", "same\nsame", "last\n"};
+    for (int recursive = 0; recursive <= 1; recursive++) {
+        sqlite3 *db[2] = {do_create_database(), do_create_database()};
+        bool ok = false;
+        for (int i = 0; i < 2; i++) {
+            if (!db[i]) goto cleanup;
+            const char *ddl = "CREATE TABLE docs(id TEXT PRIMARY KEY NOT NULL, body TEXT, other TEXT);"
+                "SELECT cloudsync_init('docs');"
+                "SELECT cloudsync_set_column('docs','body','algo','block');"
+                "SELECT cloudsync_set_column('docs','other','algo','block');"
+                "INSERT INTO docs VALUES('untouched','keep','also keep');";
+            if (sqlite3_exec(db[i], ddl, NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+            if (recursive && sqlite3_exec(db[i], "PRAGMA recursive_triggers=ON", NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+        }
+        for (int round = 0; round < 60; round++) {
+            // An UPDATE introduces fractional positions not present in initial inserts.
+            const char *seed = "INSERT OR REPLACE INTO docs VALUES('a','AAA\nBBB\nCCC','side\ncolumn');"
+                "UPDATE docs SET body='AAA\ninserted\nBBB\nCCC' WHERE id='a';";
+            if (sqlite3_exec(db[0], seed, NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+            if (!do_merge_using_payload(db[0], db[1], false, true)) goto cleanup;
+            const char *value = values[round % 6];
+            char *query = sqlite3_mprintf("INSERT OR REPLACE INTO docs VALUES('a',%Q,'side\ncolumn')", value);
+            int rc = sqlite3_exec(db[0], query, NULL, NULL, NULL);
+            sqlite3_free(query);
+            if (rc != SQLITE_OK) goto cleanup;
+            for (int delivery = 0; delivery < 2; delivery++) {
+                if (!do_merge_using_payload(db[0], db[1], false, true)) goto cleanup;
+            }
+            for (int i = 0; i < 2; i++) {
+                if (sqlite3_exec(db[i], "SELECT cloudsync_text_materialize('docs','body','a');", NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+                query = sqlite3_mprintf("SELECT body IS %Q AND other='side\ncolumn' FROM docs WHERE id='a'", value ? value : "");
+                int64_t match = do_select_int(db[i], query);
+                sqlite3_free(query);
+                if (match != 1 || do_select_int(db[i], "SELECT body='keep' AND other='also keep' FROM docs WHERE id='untouched'") != 1) {
+                    printf("replace: recursive=%d round=%d replica=%d mismatch\n", recursive, round, i);
+                    goto cleanup;
+                }
+            }
+        }
+        // A failed block write must roll back the base row and retired blocks.
+        if (sqlite3_exec(db[0], "CREATE TRIGGER reject_block BEFORE INSERT ON docs_cloudsync_blocks BEGIN SELECT RAISE(ABORT,'injected block failure'); END;", NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+        if (sqlite3_exec(db[0], "INSERT OR REPLACE INTO docs VALUES('a','rejected','other')", NULL, NULL, NULL) == SQLITE_OK) goto cleanup;
+        if (sqlite3_exec(db[0], "DROP TRIGGER reject_block; SELECT cloudsync_text_materialize('docs','body','a');", NULL, NULL, NULL) != SQLITE_OK) goto cleanup;
+        if (do_select_int(db[0], "SELECT body='last\n' AND other='side\ncolumn' FROM docs WHERE id='a'") != 1) goto cleanup;
+        ok = true;
+cleanup:
+        for (int i = 0; i < 2; i++) if (db[i]) close_db(db[i]);
+        if (!ok) return false;
+    }
+    return true;
+}
+
 bool do_test_block_lww_payload_sync(int nclients, bool print_result, bool cleanup_databases) {
     sqlite3 *db[2] = {NULL, NULL};
     time_t timestamp = time(NULL);
@@ -14137,6 +14191,7 @@ int main (int argc, const char * argv[]) {
     result += test_report("Test Block LWW Del vs Edit:", do_test_block_lww_delete_vs_edit(2, print_result, cleanup_databases));
     result += test_report("Test Block LWW TwoBlockCols:", do_test_block_lww_two_block_cols(2, print_result, cleanup_databases));
     result += test_report("Test Block LWW Text->NULL:", do_test_block_lww_text_to_null(2, print_result, cleanup_databases));
+    result += test_report("Test Block LWW Replace:", do_test_block_lww_replace());
     result += test_report("Test Block LWW PayloadSync:", do_test_block_lww_payload_sync(2, print_result, cleanup_databases));
     result += test_report("Test Block LWW Idempotent:", do_test_block_lww_idempotent(2, print_result, cleanup_databases));
     result += test_report("Test Block LWW Ordering:", do_test_block_lww_ordering(2, print_result, cleanup_databases));