diff --git a/docs/block-replace-fix.md b/docs/block-replace-fix.md new file mode 100644 index 0000000..b9d1858 --- /dev/null +++ b/docs/block-replace-fix.md @@ -0,0 +1,13 @@ +# SQLite block replacement fix + +`INSERT OR REPLACE` can skip SQLite delete triggers when `recursive_triggers` is disabled (the default). Previously, replacing a block-mode value `AAA\nBBB\nCCC` with `ZZZ` left the old trailing blocks alive. Replicas and later local materialization could produce `ZZZ\nBBB\nCCC`. + +The insert callback now snapshots the existing block names for the exact primary key and column, tombstones those blocks and removes their stored values, then writes the replacement blocks. Names are selected with an exact prefix comparison rather than LIKE wildcards. The snapshot avoids modifying a table while iterating its rows. Errors are propagated so SQLite rolls back the enclosing replacement rather than committing incomplete block state. + +The branch builds on PR #46: its parity-preserving metadata upsert is needed when replacement blocks reuse retired position IDs. Ordinary INSERT behavior is preserved, including the existing convention that inserted NULL block text is represented by one empty block and materializes as empty text. This change does not redefine NULL handling. + +## Regression coverage + +The unit test performs 120 replacement cycles across recursive triggers OFF and ON, covering shorter, longer, empty, NULL, duplicate-line and trailing-delimiter values. An intervening UPDATE creates fractional block positions. It checks duplicate payload application, both source and replica materialization, another block column, and an untouched row. An injected block-write failure verifies rollback of the replacement and retired blocks. + +The new regression fails against the previous implementation and passes with this fix. The complete SQLite ASan/UBSan unit build passes its functional tests and memory check; two existing signed-shift UBSan diagnostics remain in the unrelated random row-ID test generator. diff --git a/src/sqlite/cloudsync_sqlite.c b/src/sqlite/cloudsync_sqlite.c index c416b01..ad8ba64 100644 --- a/src/sqlite/cloudsync_sqlite.c +++ b/src/sqlite/cloudsync_sqlite.c @@ -411,6 +411,84 @@ void dbsync_pk_decode (sqlite3_context *context, int argc, sqlite3_value **argv) // MARK: - +// REPLACE may skip the delete trigger. Snapshot the old block names before +// changing the blocks table, then retire them before installing the new value. +static int dbsync_replace_blocks(cloudsync_context *data, cloudsync_table_context *table, + const char *pk, size_t pklen, int col_idx, + const char *text, int64_t db_version) { + const char *col = table_colname(table, col_idx); + char **old_names = NULL; + int old_count = 0, old_capacity = 0; + block_list_t *blocks = block_split(text ? text : "", table_col_delimiter(table, col_idx)); + char **positions = NULL; + char *prefix = block_build_colname(col, ""); + char *sql = cloudsync_memory_mprintf( + "SELECT col_name FROM %s WHERE pk=?1 AND substr(col_name,1,length(?2))=?2", + table_blocks_ref(table)); + dbvm_t *vm = NULL; + int rc = SQLITE_NOMEM; + if (!blocks || !prefix || !sql) goto cleanup; + rc = databasevm_prepare(data, sql, &vm, 0); + if (rc != SQLITE_OK) goto cleanup; + rc = databasevm_bind_blob(vm, 1, pk, (int)pklen); + if (rc != SQLITE_OK) goto cleanup; + rc = databasevm_bind_text(vm, 2, prefix, -1); + if (rc != SQLITE_OK) goto cleanup; + while ((rc = databasevm_step(vm)) == SQLITE_ROW) { + if (old_count == old_capacity) { + int capacity = old_capacity ? old_capacity * 2 : 16; + char **names = cloudsync_memory_realloc(old_names, (uint64_t)capacity * sizeof(char *)); + if (!names) { rc = SQLITE_NOMEM; goto cleanup; } + old_names = names; + old_capacity = capacity; + } + char *name = cloudsync_string_dup(database_column_text(vm, 0)); + if (!name) { rc = SQLITE_NOMEM; goto cleanup; } + old_names[old_count++] = name; + } + if (rc != SQLITE_DONE) goto cleanup; + databasevm_finalize(vm); + vm = NULL; + if (blocks->count) { + positions = block_initial_positions(blocks->count); + if (!positions) { rc = SQLITE_NOMEM; goto cleanup; } + } + rc = SQLITE_OK; + for (int b = 0; b < old_count && rc == SQLITE_OK; b++) { + const char *name = old_names[b]; + rc = local_mark_delete_block_meta(table, pk, pklen, name, db_version, cloudsync_bumpseq(data)); + if (rc == SQLITE_OK) rc = block_delete_value_external(data, table, pk, pklen, name); + } + for (int b = 0; b < blocks->count && rc == SQLITE_OK; b++) { + char *name = block_build_colname(col, positions[b]); + if (!name) { rc = SQLITE_NOMEM; break; } + rc = local_mark_insert_or_update_meta(table, pk, pklen, name, db_version, cloudsync_bumpseq(data)); + dbvm_t *write = table_block_value_write_stmt(table); + if (rc == SQLITE_OK && !write) rc = SQLITE_MISUSE; + if (rc == SQLITE_OK) rc = databasevm_bind_blob(write, 1, pk, (int)pklen); + if (rc == SQLITE_OK) rc = databasevm_bind_text(write, 2, name, -1); + if (rc == SQLITE_OK) rc = databasevm_bind_text(write, 3, blocks->entries[b].content, -1); + if (rc == SQLITE_OK) { + rc = databasevm_step(write); + if (rc == SQLITE_DONE) rc = SQLITE_OK; + } + if (write) databasevm_reset(write); + cloudsync_memory_free(name); + } +cleanup: + if (vm) databasevm_finalize(vm); + if (positions) { + for (int b = 0; b < blocks->count; b++) cloudsync_memory_free(positions[b]); + cloudsync_memory_free(positions); + } + for (int b = 0; b < old_count; b++) cloudsync_memory_free(old_names[b]); + cloudsync_memory_free(old_names); + block_list_free(blocks); + cloudsync_memory_free(prefix); + cloudsync_memory_free(sql); + return rc; +} + void dbsync_insert (sqlite3_context *context, int argc, sqlite3_value **argv) { DEBUG_FUNCTION("cloudsync_insert %s", database_value_text(argv[0])); // debug_values(argc-1, &argv[1]); @@ -481,37 +559,7 @@ void dbsync_insert (sqlite3_context *context, int argc, sqlite3_value **argv) { rc = databasevm_step((dbvm_t *)val_vm); if (rc == DBRES_ROW) { const char *text = database_column_text((dbvm_t *)val_vm, 0); - const char *delim = table_col_delimiter(table, i); - const char *col = table_colname(table, i); - - block_list_t *blocks = block_split(text ? text : "", delim); - if (blocks) { - char **positions = block_initial_positions(blocks->count); - if (positions) { - for (int b = 0; b < blocks->count; b++) { - char *block_cn = block_build_colname(col, positions[b]); - if (block_cn) { - rc = local_mark_insert_or_update_meta(table, pk, pklen, block_cn, db_version, cloudsync_bumpseq(data)); - - // Store block value in blocks table - dbvm_t *wvm = table_block_value_write_stmt(table); - if (wvm && rc == SQLITE_OK) { - databasevm_bind_blob(wvm, 1, pk, (int)pklen); - databasevm_bind_text(wvm, 2, block_cn, -1); - databasevm_bind_text(wvm, 3, blocks->entries[b].content, -1); - databasevm_step(wvm); - databasevm_reset(wvm); - } - - cloudsync_memory_free(block_cn); - } - cloudsync_memory_free(positions[b]); - if (rc != SQLITE_OK) break; - } - cloudsync_memory_free(positions); - } - block_list_free(blocks); - } + rc = dbsync_replace_blocks(data, table, pk, pklen, i, text, db_version); } databasevm_reset((dbvm_t *)val_vm); if (rc == DBRES_ROW || rc == DBRES_DONE) rc = SQLITE_OK; diff --git a/test/unit.c b/test/unit.c index b058d5b..287f20d 100644 --- a/test/unit.c +++ b/test/unit.c @@ -11418,6 +11418,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); @@ -14008,6 +14062,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));