diff --git a/database/src/table.c b/database/src/table.c index a098263..e698017 100644 --- a/database/src/table.c +++ b/database/src/table.c @@ -1227,7 +1227,18 @@ static int row_apply_field_keys(wo_db *db, uint32_t class_id, uint64_t id, } for (uint32_t i = 0; i < b->len; i++) { if (b->ids[i] == id) continue; - uint64_t cand_off1 = wo_row_offset1(db, class_id, b->ids[i]); + /* Task 4 follow-up (review finding): a candidate updated + earlier in this SAME, not-yet-committed drain has its + re-point only PENDING — the durable wo_row_offset1 would + still fold its PRE-update value, letting a real unique + clash through uncaught. Unlike back_off (a pure number), + keys_fold_into DOES need to read this record's bytes to + compare values — which is why wo_wal_fold_row_at now reads + the staging buffer for an offset in the not-yet-durable + range (see scan_record_staged in wal.c); a plain + wo_row_offset1 substitution here is not enough on its own. */ + uint64_t cand_off1 = wo_wal_repoint_offset1(w, class_id, b->ids[i]); + if (!cand_off1) cand_off1 = wo_row_offset1(db, class_id, b->ids[i]); if (!cand_off1) continue; /* stale bucket entry: no row, no clash */ const char *obmsg = ""; db_row *other = diff --git a/database/src/wal.c b/database/src/wal.c index 59bca47..fc5e352 100644 --- a/database/src/wal.c +++ b/database/src/wal.c @@ -298,6 +298,41 @@ static int scan_record(int fd, uint64_t off, uint32_t *len_out, uint8_t **payloa return 0; } +/* Task 4 follow-up (review finding): scan_record's exact framing, but read + * from the STAGED bytes in [w->off, w->off + w->len) instead of the file. + * A record staged behind this batch's own not-yet-run barrier is real, + * fully-framed data sitting in `w->buf` — preading its file offset would + * see whatever was on disk BEFORE this batch (the pre-allocated zero + * tail, ordinarily), which scan_record correctly, but unhelpfully, reads + * as "no record here yet". Used only by wo_wal_fold_row_at, and only for + * an offset a caller got from wo_wal_repoint_offset1 (a PENDING re-point, + * not yet flushed) — never for a durable offset, which stays on the + * scan_record/pread path unchanged. 0 = intact, 1 = short/bad, same as + * scan_record. */ +static int scan_record_staged(const wo_wal *w, uint64_t off, uint32_t *len_out, + uint8_t **payload_out) { + if (off < w->off) return 1; + uint64_t rel = off - w->off; + if (rel + 8 > w->len) return 1; + uint32_t len, crc; + memcpy(&len, w->buf + rel, 4); + memcpy(&crc, w->buf + rel + 4, 4); + if (len == 0 || len > (64u << 20)) return 1; /* garbage: not a real frame */ + if (rel + 8 + (uint64_t)len + 4 > w->len) return 1; /* short: not fully staged */ + const uint8_t *payload = w->buf + rel + 8; + uint32_t mark; + memcpy(&mark, payload + len, 4); + if (mark != WO_WAL_MARK || crc32(payload, len) != crc) return 1; + if (payload_out) { + uint8_t *cp = malloc((size_t)len + 4); + if (!cp) return 1; + memcpy(cp, payload, (size_t)len + 4); + *payload_out = cp; + } + *len_out = len; + return 0; +} + /* ---- public API ---------------------------------------------------------- */ int wo_wal_open(wo_wal *w, const char *path, uint64_t prealloc) { @@ -926,7 +961,14 @@ int wo_wal_fold_row_at(wo_wal *w, wo_db *db, uint64_t off, uint32_t *class_out, for (;;) { uint32_t len; uint8_t *payload; - if (scan_record(w->fd, cur, &len, &payload) != 0) { + /* Task 4 follow-up: a hop into the currently-staged (not yet + durable) region reads from `w->buf`, not the file — see + scan_record_staged. Every other hop (the durable majority of + any real chain) is the original file read, unchanged. */ + int src_rc = (cur >= w->off && cur < w->off + w->len) + ? scan_record_staged(w, cur, &len, &payload) + : scan_record(w->fd, cur, &len, &payload); + if (src_rc != 0) { *msg = "no intact record at that offset"; rc = -1; break; diff --git a/runtime/test/test_wal.c b/runtime/test/test_wal.c index a23441b..e620ebe 100644 --- a/runtime/test/test_wal.c +++ b/runtime/test/test_wal.c @@ -1149,6 +1149,79 @@ static void test_keys_resident_two_updates_one_drain(void) { wo_rt_destroy(&rt); } +/* Task 4 follow-up (review finding): the unique shadow-check's candidate + * lookup must ALSO prefer a pending re-point over the durable map, for the + * same reason back_off does. Before this task, two keys-resident updates + * in one drain could not happen at all (the second crashed). Task 4 makes + * it possible, which makes THIS reachable: request 1 updates row A's + * unique-indexed column to a NEW value inside a drain (staged, not + * committed, its index bucket already moved — that part is unconditional + * RAM apply); request 2, same drain, updates a DIFFERENT row B to that + * SAME new value. The shadow check finds A sitting in the target bucket + * (correct — the bucket move is immediate) but, without the fix, verifies + * A by folding it from its still-DURABLE (pre-update) offset — reading + * A's OLD value, which does not match, so the real clash is missed and a + * duplicate would be committed. Request 2 must be REFUSED. */ +static void test_keys_resident_unique_clash_pending_repoint(void) { + char path[128]; + snprintf(path, sizeof path, "%s/keysuniqpend.wal", g_dir); + wo_rt rt; + T_EQ(wo_rt_init(&rt, 1 << 20, KEYS_UNIQUE_CLASSES, 1), 0); + wo_db db; + T_EQ(wo_db_init(&db, KEYS_UNIQUE_CLASSES, 1, 0, 1), 0); + wo_wal w; + T_EQ(wo_wal_open(&w, path, 1 << 16), 0); + db.rt = &rt; rt.wal = &w; rt.db = &db; + const char *msg = ""; + + wo_str *sa = wo_str_new(&rt, "a", 1); + wo_str *sb = wo_str_new(&rt, "b", 1); + uint64_t va[2] = {100, (uint64_t)(uintptr_t)sa}; + uint64_t vb[2] = {200, (uint64_t)(uintptr_t)sb}; + uint64_t a = wo_row_insert(&db, 0, va, &msg, NULL); + uint64_t b = wo_row_insert(&db, 0, vb, &msg, NULL); + T_CHECK(a != 0 && b != 0); + uint64_t off_a = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_insert(&w, &db, 0, a), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_drop_payload(&db, 0, a, off_a), 0); + uint64_t off_b = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_insert(&w, &db, 0, b), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_drop_payload(&db, 0, b, off_b), 0); + + /* "request" 1: a's n 100 -> 300 — staged, NOT committed, map NOT moved + (only pending), exactly db.c's request arm */ + int ek = 0; + uint64_t roff_a = wo_wal_next_offset(&w); + T_EQ(wo_row_update_field_slot(&db, 0, a, 0, 300, &msg, &ek), 0); + T_EQ(ek, DB_ERR_NONE); + T_EQ(wo_wal_pend_repoint(&w, 0, a, roff_a), 0); + + /* "request" 2, SAME drain: b's n 200 -> 300 collides with a's NEW + (still only staged) value — must be refused */ + T_EQ(wo_row_update_field_slot(&db, 0, b, 0, 300, &msg, &ek), -1); + T_EQ(ek, DB_ERR_UNIQUE); + + /* the drain's barrier: commit a's delta, flush its pending re-point */ + T_EQ(wo_wal_commit(&w), 0); + wo_db_flush_drops(&db, &w); + + /* b untouched: still 200, still the only hit for 200; a alone at 300 */ + db_row *r = wo_row_borrow(&db, 0, b, &msg); + T_CHECK(r != NULL && r->slots[0] == 200); + wo_row_release(&db, 0, r); + uint64_t *ids; + uint32_t cnt; + T_EQ(wo_idx_probe(&db, 0, 0, 300, NULL, 0, &ids, &cnt), 1); + T_CHECK(cnt == 1 && ids[0] == a); + free(ids); + + wo_wal_close(&w); + wo_db_destroy(&db); + wo_rt_destroy(&rt); +} + /* databasev2 2 (5c): boot. A keys-resident store must come back from replay * with its rows readable FROM THE LOG — the map rebuilt to offsets, not slabs. * This is the half the round-trip test cannot cover: it runs in a fresh db, @@ -1801,6 +1874,7 @@ int main(void) { test_keys_resident_update_indexed(); test_keys_resident_update_unique_violation_refused(); test_keys_resident_two_updates_one_drain(); + test_keys_resident_unique_clash_pending_repoint(); test_keys_resident_replay(); test_keys_resident_survives_compaction(); test_keys_resident_delete();