Fix rebase conflicts, log and ATTACH durability, recovery and row diff OOM

CHANGELOG.md

@@ -5,6 +5,9 @@ The format of the files on disk (`-mwlock`, `-mwidx`, `-mw.N`, `-mw`) is the sam
 
 ## Unreleased
 
+### Fixed
+- A commit that an `fsync` already in flight covers is no longer refused when the log is failed meanwhile. The append of one commit failed with a later commit already behind it (the log is failed for good); a commit parked in the sync, whose record the running `fsync` was making durable, saw the failure first and was told that it had failed, and then the `fsync` finished and made it visible: it was refused and in the database (and in the log after a reopen). It now waits for the `fsync` that is running and is acknowledged if that one covers it; the ones that it does not cover are refused and stay invisible, as before. Found as a failure of `mw_rollbackrace` on a runner of two CPUs (2 of 150 runs on a machine of two, with and without the other changes); it is reproduced in every run by `mw_holesync`.
+
 ### Changed
 - Documentation: the wording of the isolation limits in the README and in the changelog says what is prevented (the classic write skew, by the validation of the reads) and where it is still possible, instead of "write skew is possible".
 

Makefile

@@ -63,15 +63,44 @@ $(BUILD)/t_%.o: test/%.c $(HEADERS) test/mw_test.h
 $(DIST)/mw_%$(EXE): $(BUILD)/t_mw_%.o $(LIB_OBJ)
 	@mkdir -p $(DIST)
 	$(CC) $^ -o $@ $(LDFLAGS)
+# Deterministic fault injection, confined to the translation unit under test.
+$(BUILD)/inject_log.o: $(SRC_DIR)/multiwriter_log.c $(HEADERS)
+	@mkdir -p $(BUILD)
+	$(CC) $(CFLAGS) -Dpwritev=inject_pwritev -c $< -o $@
+$(BUILD)/inject_rowdiff.o: $(SRC_DIR)/multiwriter_rowdiff.c $(HEADERS)
+	@mkdir -p $(BUILD)
+	$(CC) $(CFLAGS) -Dmalloc=inject_malloc -Dcalloc=inject_calloc -Drealloc=inject_realloc -c $< -o $@
+$(BUILD)/inject_fsync.o: $(SRC_DIR)/multiwriter_log.c $(HEADERS)
+	@mkdir -p $(BUILD)
+	$(CC) $(CFLAGS) -Dfsync=inject_fsync -c $< -o $@
+$(DIST)/mw_holesync$(EXE): $(BUILD)/t_mw_holesync.o $(filter-out $(BUILD)/multiwriter_log.o,$(LIB_OBJ)) $(BUILD)/inject_fsync.o
+	@mkdir -p $(DIST)
+	$(CC) $^ -o $@ $(LDFLAGS)
+$(DIST)/mw_stagednormal$(EXE): $(BUILD)/t_mw_stagednormal.o $(filter-out $(BUILD)/multiwriter_log.o,$(LIB_OBJ)) $(BUILD)/inject_log.o
+	@mkdir -p $(DIST)
+	$(CC) $^ -o $@ $(LDFLAGS)
+$(DIST)/mw_rowdiffoom$(EXE): $(BUILD)/t_mw_rowdiffoom.o $(filter-out $(BUILD)/multiwriter_rowdiff.o,$(LIB_OBJ)) $(BUILD)/inject_rowdiff.o
+	@mkdir -p $(DIST)
+	$(CC) $^ -o $@ $(LDFLAGS)
+
 .PRECIOUS: $(BUILD)/t_%.o
 
-test: $(FAST_BIN)
+# the exit status of a failing test must survive the sanitizer runner (test/sanitize.sh); not for Windows, where the runner is not used
+.PHONY: test-sanitize-runner
+test-sanitize-runner:
+ifeq ($(PLATFORM),windows)
+	@echo "sanitizer runner regression skipped (Windows)"
+else
+	@if command -v bash >/dev/null 2>&1; then bash test/sanitize_exit.sh; else echo "sanitizer runner regression skipped (bash unavailable)"; fi
+endif
+
+test: $(FAST_BIN) test-sanitize-runner
 	@set -e; for t in $(FAST_BIN); do echo "== $$t"; ./$$t; done; echo "== mw_rebase (a WITHOUT ROWID table)"; MW_TEST_WR=1 ./$(DIST)/mw_rebase$(EXE)
 
 # the suite of the CI legs that run it on shared machines: without mw_serial (about a minute and a half of stress, which has a threshold of events that a loaded machine can miss); `make test` runs everything
 CI_BIN := $(filter-out $(DIST)/mw_serial$(EXE),$(FAST_BIN))
 .PHONY: test-ci
-test-ci: $(CI_BIN)
+test-ci: $(CI_BIN) test-sanitize-runner
 	@set -e; for t in $(CI_BIN); do echo "== $$t"; ./$$t; done; echo "== mw_rebase (a WITHOUT ROWID table)"; MW_TEST_WR=1 ./$(DIST)/mw_rebase$(EXE)
 
 # errors of the file system: a fault at every n-th file call of a workload, and a real full disk (a small disk image)
@@ -94,9 +123,9 @@ clean:
 	rm -rf $(BUILD) $(DIST)
 
 # the transaction tests with the shared multi-process mode (mw_mp=1) in place of the thread mode
-MP_TESTS := mw_rebase mw_rebasefk mw_rebasefkstress mw_rebasenoop mw_rebaseskew mw_rebasewr mw_serial mw_relocprep mw_shared
+MP_TESTS := mw_strictany mw_rebasepolicy mw_rebase mw_rebasefk mw_rebasefkstress mw_rebasenoop mw_rebaseskew mw_rebasewr mw_serial mw_relocprep mw_shared
 ifeq ($(PLATFORM),windows)
-MP_TESTS := mw_rebase mw_rebasefk mw_rebasefkstress mw_rebasenoop mw_rebaseskew mw_rebasewr mw_serial     # (the others fork)
+MP_TESTS := mw_strictany mw_rebasepolicy mw_rebase mw_rebasefk mw_rebasefkstress mw_rebasenoop mw_rebaseskew mw_rebasewr mw_serial     # (the others fork)
 endif
 .PHONY: test-mp
 test-mp: $(addsuffix $(EXE),$(addprefix $(DIST)/,$(MP_TESTS)))

src/multiwriter_catalog.c

@@ -126,13 +126,14 @@ static void tab_parse (sqlite3 *scratch, const srow *r, mw_tab *t) {
     t->without_rowid = strcasestr(r->sql, "WITHOUT ROWID") != NULL;                                       // (the text: a default or a comment that says so fools it, and so does a second blank; below, SQLite says it)
     char *err = NULL;
     if (sqlite3_exec(scratch, r->sql, NULL, NULL, &err) != SQLITE_OK) { sqlite3_free(err); return; }
-    {   // SQLite 3.37 and later: PRAGMA table_list has a column wr
+    bool strict = false;
+    {   // SQLite 3.37 and later: PRAGMA table_list has columns wr and strict
         char *wq = sqlite3_mprintf("PRAGMA table_list(\"%w\")", r->name); sqlite3_stmt *ws = NULL;
         if (wq && sqlite3_prepare_v2(scratch, wq, -1, &ws, NULL) == SQLITE_OK) {
-            int wrcol = -1; for (int i = 0; i < sqlite3_column_count(ws); i++) if (!strcmp(sqlite3_column_name(ws, i), "wr")) wrcol = i;
+            int wrcol = -1, strictcol = -1; for (int i = 0; i < sqlite3_column_count(ws); i++) { if (!strcmp(sqlite3_column_name(ws, i), "wr")) wrcol = i; if (!strcmp(sqlite3_column_name(ws, i), "strict")) strictcol = i; }
             if (wrcol >= 0) while (sqlite3_step(ws) == SQLITE_ROW) {
                 const char *nm = (const char *)sqlite3_column_text(ws, 1);
-                if (nm && !strcasecmp(nm, r->name)) { t->without_rowid = sqlite3_column_int(ws, wrcol) != 0; break; }
+                if (nm && !strcasecmp(nm, r->name)) { t->without_rowid = sqlite3_column_int(ws, wrcol) != 0; strict = strictcol >= 0 && sqlite3_column_int(ws, strictcol) != 0; break; }
             }
         }
         sqlite3_finalize(ws); sqlite3_free(wq);
@@ -182,7 +183,7 @@ static void tab_parse (sqlite3 *scratch, const srow *r, mw_tab *t) {
         int i = order[j];
         if (cols[i].hidden == 2) continue;                                                                   // (generated, virtual: not stored)
         if (t->alias_pk && cols[i].pk == 1) t->alias_rec = pos;
-        if (t->noaff) t->noaff[pos] = type_has_no_affinity(cols[i].type);
+        if (t->noaff) t->noaff[pos] = type_has_no_affinity(cols[i].type) || (strict && !strcasecmp(cols[i].type, "ANY"));
         t->rec_name[pos++] = (cols[i].hidden == 3) ? NULL : strdup(cols[i].name);                            // (generated, stored: in the record, derived)
     }
     t->nrec = pos;

src/multiwriter_catalog.h

@@ -24,7 +24,7 @@ typedef struct {
     int nrec;                           // columns stored in a record
     int alias_rec;                      // the record column of the rowid alias (-1: none)
     char **rec_name;                    // the name of the column stored in each record column; NULL for a generated one (derived: not written, not compared)
-    bool *noaff;                        // of each record column: it has no affinity (declared with no type, or BLOB): 1 and 1.0 are different values there, and an equality of values does not tell them apart
+    bool *noaff;                        // of each record column: it has no affinity (declared with no type, BLOB, or ANY of a STRICT table): 1 and 1.0 are different values there, and an equality of values does not tell them apart
     mw_val *dflt;                       // the default of each record column (what a row written before an ALTER TABLE ADD COLUMN has in the columns it lacks); NULL: all NULL
     int nfk; char **fk_parent, **fk_to;  // foreign keys of this table: the parent table, and the parent column (NULL: its primary key)
     bool is_parent;                     // some foreign key refers to this table

src/multiwriter_lane.c

@@ -173,7 +173,17 @@ static bool multirow_insert (const char *sql) {
     }
     return false;
 }
+static bool has_word (const char *sql, const char *w);
 static bool stmt_is_point (mw_lane *lane, sqlite3_stmt *st, const char *sql) {
+    // Row images do not retain a statement's OR policy, nor an UPSERT clause. Replaying it with
+    // the schema's policy could delete or ignore a concurrent writer's row.
+    const char *head = mw_sql_skip(sql);
+    if (sqlite3_strnicmp(head, "INSERT", 6) == 0 || sqlite3_strnicmp(head, "UPDATE", 6) == 0) {
+        const char *next = mw_sql_skip(head + 6);
+        if (sqlite3_strnicmp(next, "OR", 2) == 0 && !isalnum((unsigned char)next[2]) && next[2] != '_') return false;
+    }
+    // (ON CONFLICT ... DO NOTHING / DO UPDATE: a plain INSERT would take the schema's policy for the constraint instead; a match inside a string only makes a point statement look refused)
+    if (sqlite3_strnicmp(head, "INSERT", 6) == 0 && has_word(head, "CONFLICT")) return false;
     rd_ent *c = lane->rd_cache;
     if (!c) { c = lane->rd_cache = sqlite3_malloc64(RD_CACHE * sizeof *c); if (!c) return false; memset(c, 0, RD_CACHE * sizeof *c); }
     // The answer depends on the schema (the same text of an UPDATE ... WHERE email = ? is one row while a UNIQUE index exists and a scan of the table after it is dropped): it is kept for the
@@ -687,6 +697,9 @@ static int pragma_is (const char *name, const char *want) { return name && sqlit
 static int lm_file_control (sqlite3_file *pf, int op, void *arg) {
     mw_lane *lane = FILE_LANE(pf);
     tls_lane = lane;
+    // SQLite supplies the owning connection when each b-tree opens, including
+    // ATTACH. Capture it here so PRAGMA checks do not depend on trace hooks.
+    if (op == SQLITE_FCNTL_PDB) { lane->rd_db = *(sqlite3 **)arg; return SQLITE_OK; }
     if (op == MW_FCNTL_TXINFO) { *(mw_tx_info *)arg = lane->tx; return SQLITE_OK; }
     if (op == MW_FCNTL_WRITESET) {
         mw_writeset *ws = (mw_writeset *)arg;
@@ -719,12 +732,21 @@ static int lm_file_control (sqlite3_file *pf, int op, void *arg) {
             return SQLITE_ERROR;
         }
         if (pragma_is(name, "synchronous") && val) {          // observed, not handled: SQLite still executes it
-            int lv = (int)strtol(val, NULL, 10);
-            if (!strcmp(val, "0") || !sqlite3_stricmp(val, "off")) lv = 0;
-            else if (!sqlite3_stricmp(val, "normal")) lv = 1;
+            // SQLite rejects this PRAGMA even after a BEGIN that has not
+            // acquired a snapshot yet. Do not change the engine's level.
+            if (lane->rd_db && !sqlite3_get_autocommit(lane->rd_db)) return SQLITE_NOTFOUND;
+            // Match SQLite's getSafetyLevel and its pager mask, including
+            // boolean names, unknown names, and out-of-range integers.
+            int lv = 1;
+            if (val[0] >= '0' && val[0] <= '9') {
+                int base = val[0] == '0' && (val[1] == 'x' || val[1] == 'X') && isxdigit((unsigned char)val[2]) ? 16 : 10;
+                unsigned long n = strtoul(val, NULL, base);
+                lv = n > 2147483647UL ? 0 : (int)(n & 255);
+            } else if (!sqlite3_stricmp(val, "off") || !sqlite3_stricmp(val, "no") || !sqlite3_stricmp(val, "false")) lv = 0;
             else if (!sqlite3_stricmp(val, "full")) lv = 2;
             else if (!sqlite3_stricmp(val, "extra")) lv = 3;
-            lane->sync_level = lv;
+            lv = (lv + 1) & 7;
+            lane->sync_level = (lv ? lv : 1) - 1;
             return SQLITE_NOTFOUND;
         }
         if (pragma_is(name, "mmap_size") && val) { az[0] = sqlite3_mprintf("0"); return SQLITE_OK; }   // normalised to 0

src/multiwriter_log.c

@@ -612,6 +612,28 @@ void mw_log_stage_reset (mw_db *db, uint64_t off) {
     pthread_mutex_unlock(&db->log_mu);
 }
 
+// NORMAL/OFF still require the record to leave process-private memory
+// before acknowledgement. Another writer may be flushing an earlier prefix.
+static int stage_wait_written (mw_db *db, uint64_t epoch) {
+    pthread_mutex_lock(&db->log_mu);
+    while (db->flushed_epoch < epoch) {
+        if (atomic_load(&db->failed)) { pthread_mutex_unlock(&db->log_mu); return SQLITE_IOERR_WRITE; }
+        if (!db->sync_running && !db->stage_writing &&
+            (pipeline_on() || !db->stage_fsyncing) && db->written_end > db->flushed_off) {
+            pthread_mutex_unlock(&db->log_mu);
+            int rc = stage_flush(db, true);
+            if (rc != SQLITE_OK) return rc;
+            pthread_mutex_lock(&db->log_mu);
+        } else {
+            db->sync_waiters++;
+            pthread_cond_wait(&db->sync_cv, &db->log_mu);
+            db->sync_waiters--;
+        }
+    }
+    pthread_mutex_unlock(&db->log_mu);
+    return SQLITE_OK;
+}
+
 // Marks a finished record (epoch, ending at `end`) as copied: advances the contiguous prefix. log_mu held. Returns false on NOMEM.
 static bool mark_written (mw_db *db, uint64_t epoch, uint64_t end) {
     if (epoch == db->written_upto + 1) {
@@ -702,7 +724,7 @@ static int append_staged (mw_db *db, uint64_t off, uint64_t epoch, uint32_t dbsi
     if (!mark_written(db, epoch, end)) { pthread_mutex_unlock(&db->log_mu); return SQLITE_NOMEM; }
     if (db->written_end != before && db->sync_waiters > 0 && !db->sync_running && !db->stage_writing) pthread_cond_broadcast(&db->sync_cv);   // somebody may be able to write now
     pthread_mutex_unlock(&db->log_mu);
-    if (!sync) return stage_flush(db, true);                                    // no durability wait: the record still goes to the file promptly (a killed process keeps it)
+    if (!sync) return stage_wait_written(db, epoch);                            // no fsync, but our record must be in the file before returning
     return SQLITE_OK;
 }
 
@@ -717,7 +739,13 @@ static int sync_staged (mw_db *db, uint64_t my_epoch, uint64_t my_end) {
     // (done when the file is synced up to the record, or when everything up to its epoch is durable: a compaction that finds nothing in flight cuts the log back to its header and the
     // offsets restart, while a commit whose epoch the group leader has just made visible is still on its way out of this loop)
     while (db->synced_off < my_end && db->synced_upto < my_epoch) {
-        if (atomic_load(&db->failed)) { rc = SQLITE_IOERR_FSYNC; break; }
+        if (atomic_load(&db->failed)) {
+            // The log was failed (the append of an earlier commit did not make it, and a later one already has its place behind the hole). An fsync that is in flight covers what was written
+            // before it began, and its leader makes those commits durable and visible when it is done: a commit that it covers was not refused, and is not told that it was. What it does not
+            // cover is told, and stays invisible: no other fsync begins once the log is failed.
+            if (db->stage_fsyncing) { db->sync_waiters++; pthread_cond_wait(&db->sync_cv, &db->log_mu); db->sync_waiters--; continue; }
+            rc = SQLITE_IOERR_FSYNC; break;
+        }
         // Two roles, each held by one thread at a time, and (unless MW_NO_PIPELINE) working at the same time: the *writer* pwritev's the contiguous prefix of
         // finished records that is not in the file yet; the *syncer* fsyncs everything that was written before its fsync started. While one thread is in
         // fsync for group n another writes group n+1; group n+1 then needs its own fsync after the first one ends (an fsync covers only what was written before it began).

src/multiwriter_rebase.c

@@ -99,7 +99,7 @@ static int bind_col (sqlite3_stmt *st, int idx, const rec_t *r, int col) {
     uint64_t t = r->ty[col]; const uint8_t *p = r->rec + r->off[col]; uint32_t l = r->len[col];
     if (t == 0) return sqlite3_bind_null(st, idx);
     if (t == 8 || t == 9) return sqlite3_bind_int64(st, idx, t == 9);
-    if (t >= 1 && t <= 6) { int64_t x = (int8_t)p[0]; for (uint32_t b = 1; b < l; b++) x = (x << 8) | p[b]; return sqlite3_bind_int64(st, idx, x); }
+    if (t >= 1 && t <= 6) { uint64_t x = (uint64_t)(int64_t)(int8_t)p[0]; for (uint32_t b = 1; b < l; b++) x = (x << 8) | p[b]; return sqlite3_bind_int64(st, idx, (int64_t)x); }
     if (t == 7) { uint64_t u = 0; for (int b = 0; b < 8; b++) u = (u << 8) | p[b]; double d; memcpy(&d, &u, 8); return sqlite3_bind_double(st, idx, d); }
     if (t >= 12 && (t & 1)) return sqlite3_bind_text(st, idx, l ? (const char *)p : "", (int)l, SQLITE_TRANSIENT);
     if (t >= 12) return sqlite3_bind_blob(st, idx, l ? (const void *)p : (const void *)"", (int)l, SQLITE_TRANSIENT);