fix(db2-delta): unique shadow-check gets its own buffer, not r's

- row_apply_field_keys's shadow-check borrowed candidates via
  wo_row_borrow, which shares ONE per-table scratch with the row already
  borrowed for the update — every candidate borrow returned NULL, clash
  was always false, `@unique` silently accepted duplicates on update
- idx_add_row's own internal check has the identical defect at the same
  call site; discarding its result is now actually safe, since the fixed
  shadow-check clears uniqueness before it ever runs
- fix: extracted keys_fold_into (fold+decode) out of wo_row_borrow so it
  can target a throwaway per-call buffer instead of t->scratch; the
  shadow-check probes candidates into that buffer — r is never
  released-and-reborrowed (r IS t->scratch; that would overwrite it)
- wo_row_borrow itself is behavior-preserving: same checks, same order,
  same messages, just factored
- test_wal.c: new test — genuine @unique index, update collides with an
  existing row, asserts refusal (DB_ERR_UNIQUE) and both rows untouched;
  verified failing (update wrongly succeeded) pre-fix, passing after

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
(cherry picked from commit 409186da51fec0042d8d1e3dcf9f69f704937823)

database/src/table.c

@@ -751,42 +751,20 @@ db_row *wo_row_ptr(wo_db *db, uint32_t class_id, uint64_t id) {
     return slot_row(t, (uint32_t)(s1 - 1));
 }
 
-db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **msg) {
+/* keys-resident fold+decode: reads the row at [off] (the row's current
-    /* Fully-resident tables: exactly today's lookup, and releasing is a no-op.
+ * record) and decodes it into VM values inside [buf] (t->row_size bytes,
-     * The hot path pays one predicate. */
+ * caller-owned) — the piece wo_row_borrow and a unique shadow-check's
-    if (!wo_table_is_keys_resident(db, class_id)) return wo_row_ptr(db, class_id, id);
+ * candidate probe both need, factored out because they cannot share a
-
+ * buffer: wo_row_borrow writes into t->scratch and holds it busy for the
-    /* Keys-resident: the id map holds the record's LOG OFFSET (off + 1), not a
+ * whole life of the borrow, so a shadow-check that needs to look at OTHER
-     * slot, so the row is materialised into the table's scratch. */
+ * rows of the SAME table while the row under test is still borrowed must
-    db_table *t = &db->tables[class_id];
+ * use a buffer of its own, never t->scratch. [id] is checked against what
-    if (!t->row_size) return NULL;
+ * the fold actually names, same as wo_row_borrow always did. NULL on any
-    uint64_t o1 = hget(t, id);
+ * failure, *msg set. */
-    if (!o1) return NULL;
+static db_row *keys_fold_into(wo_db *db, uint32_t class_id, uint64_t id,
-    if (!db->rt || !db->rt->wal) {
+                              uint64_t off, uint8_t *buf, const char **msg) {
-        /* a keys-resident table cannot exist without a log to read from; the
-         * loader refuses the annotation outright, so this is a defensive arm */
-        if (msg) *msg = "resident: keys table without a write-ahead log";
-        return NULL;
-    }
-    if (t->scratch_busy) {
-        /* One scratch per TABLE, so two live borrows on the same table would
-         * hand back the same buffer. The unique shadow borrows one candidate
-         * at a time, which is why per-table is enough — but say so rather than
-         * corrupting the first borrow silently. */
-        if (msg) *msg = "nested borrow on one table";
-        return NULL;
-    }
-    if (t->scratch_cap < t->row_size) {
-        uint8_t *nb = realloc(t->scratch, t->row_size);
-        if (!nb) {
-            if (msg) *msg = "out of memory";
-            return NULL;
-        }
-        t->scratch = nb;
-        t->scratch_cap = t->row_size;
-    }
-    db_row *r = (db_row *)t->scratch;
     const wo_classdesc *c = &db->classes[class_id];
+    db_row *r = (db_row *)buf;
     uint32_t got_cid = 0;
     uint64_t got_id = 0;
     /* keys-resident delta updates, Task 2: the fold, not a single-record
@@ -796,8 +774,7 @@ db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **ms
         if (msg) *msg = "out of memory";
         return NULL;
     }
-    if (wo_wal_fold_row_at((wo_wal *)db->rt->wal, db, o1 - 1, &got_cid, &got_id, eng, msg) !=
+    if (wo_wal_fold_row_at((wo_wal *)db->rt->wal, db, off, &got_cid, &got_id, eng, msg) != 0) {
-        0) {
         free(eng);
         return NULL;
     }
@@ -824,6 +801,46 @@ db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **ms
     r->id = id;
     r->class_id = class_id;
     r->flags = 0;
+    return r;
+}
+
+db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **msg) {
+    /* Fully-resident tables: exactly today's lookup, and releasing is a no-op.
+     * The hot path pays one predicate. */
+    if (!wo_table_is_keys_resident(db, class_id)) return wo_row_ptr(db, class_id, id);
+
+    /* Keys-resident: the id map holds the record's LOG OFFSET (off + 1), not a
+     * slot, so the row is materialised into the table's scratch. */
+    db_table *t = &db->tables[class_id];
+    if (!t->row_size) return NULL;
+    uint64_t o1 = hget(t, id);
+    if (!o1) return NULL;
+    if (!db->rt || !db->rt->wal) {
+        /* a keys-resident table cannot exist without a log to read from; the
+         * loader refuses the annotation outright, so this is a defensive arm */
+        if (msg) *msg = "resident: keys table without a write-ahead log";
+        return NULL;
+    }
+    if (t->scratch_busy) {
+        /* One scratch per TABLE, so two live borrows on the same table would
+         * hand back the same buffer. A unique shadow-check that needs OTHER
+         * rows of this table while one is already borrowed uses its OWN
+         * throwaway buffer (row_apply_field_keys), never this one — say so
+         * rather than corrupting the first borrow silently. */
+        if (msg) *msg = "nested borrow on one table";
+        return NULL;
+    }
+    if (t->scratch_cap < t->row_size) {
+        uint8_t *nb = realloc(t->scratch, t->row_size);
+        if (!nb) {
+            if (msg) *msg = "out of memory";
+            return NULL;
+        }
+        t->scratch = nb;
+        t->scratch_cap = t->row_size;
+    }
+    db_row *r = keys_fold_into(db, class_id, id, o1 - 1, t->scratch, msg);
+    if (!r) return NULL;
     t->scratch_busy = 1;
     return r;
 }
@@ -1159,9 +1176,18 @@ static int row_apply_field_keys(wo_db *db, uint32_t class_id, uint64_t id,
     }
 
     /* unique shadow-check: run with the NEW value before anything durable or
-       indexed moves, exactly row_apply_field_slot's promise */
+       indexed moves, exactly row_apply_field_slot's promise.
+       CRITICAL: candidates are probed into a THROWAWAY buffer, never
+       t->scratch. r (the row under update) already lives in t->scratch and
+       wo_row_borrow refuses ANY nested borrow on the same table's scratch —
+       reusing it here would make every candidate probe return NULL, so a
+       clash could never be detected (a silent hole: keys-resident @unique
+       would accept duplicates). Candidates are always in this same,
+       keys-resident table, so keys_fold_into (bypassing wo_row_borrow and
+       its scratch_busy gate) is safe to call directly. */
     uint64_t old_vm = r->slots[field];
     r->slots[field] = nv_vm;
+    uint8_t *cand_buf = NULL;
     for (uint32_t x = 0; x < t->index_cnt; x++) {
         db_index *ix = &t->indexes[x];
         if (!(ix->flags & 1u)) continue;
@@ -1171,14 +1197,34 @@ static int row_apply_field_keys(wo_db *db, uint32_t class_id, uint64_t id,
         if (!touches) continue;
         db_ibucket *b = idx_bucket(ix, idx_hash(c, ix, r), 0);
         if (!b) continue;
+        if (!cand_buf) {
+            cand_buf = malloc(t->row_size);
+            if (!cand_buf) {
+                r->slots[field] = old_vm;
+                wo_row_release(db, class_id, r);
+                wo_drop_kind(db->rt, c->kinds[field], nv_vm);
+                db_val_free(c->kinds[field], nv);
+                if (err_kind) *err_kind = DB_ERR_OOM;
+                *msg = "out of memory";
+                return -1;
+            }
+        }
         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]);
+            if (!cand_off1) continue; /* stale bucket entry: no row, no clash */
             const char *obmsg = "";
-            db_row *other = wo_row_borrow(db, class_id, b->ids[i], &obmsg);
+            db_row *other =
+                keys_fold_into(db, class_id, b->ids[i], cand_off1 - 1, cand_buf, &obmsg);
             int clash = other && idx_cols_equal(c, ix, r, other);
-            wo_row_release(db, class_id, other);
+            /* keys_fold_into decoded fresh VM values for EVERY field, same
+               as a real borrow — nobody else owns them, so drop them here */
+            if (other)
+                for (uint32_t k = 0; k < c->field_cnt; k++)
+                    wo_drop_kind(db->rt, c->kinds[k], other->slots[k]);
             if (clash) {
                 r->slots[field] = old_vm; /* untouched, promised */
+                free(cand_buf);
                 wo_row_release(db, class_id, r);
                 wo_drop_kind(db->rt, c->kinds[field], nv_vm);
                 db_val_free(c->kinds[field], nv);
@@ -1188,6 +1234,7 @@ static int row_apply_field_keys(wo_db *db, uint32_t class_id, uint64_t id,
             }
         }
     }
+    free(cand_buf);
     r->slots[field] = old_vm; /* restored: still the OLD row until committed */
 
     wo_wal *w = (wo_wal *)db->rt->wal;

runtime/test/test_wal.c

@@ -981,6 +981,76 @@ static void test_keys_resident_update_indexed(void) {
     wo_rt_destroy(&rt);
 }
 
+/* class 0: Row { n: scalar @unique, label: Text } — same shape as
+ * KEYS_IDX_CLASSES, but the index is genuinely unique this time. */
+static const uint8_t keys_uniq_kinds[] = {WO_K_SCALAR, WO_K_TEXT};
+static const uint32_t keys_uniq_meta[] = {1 /*unique*/, 1, 0 /*col: n*/};
+static const wo_classdesc KEYS_UNIQUE_CLASSES[] = {
+    {.name = 0, .flags = WO_CLASSF_RESIDENT_KEYS, .field_cnt = 2, .kinds = keys_uniq_kinds,
+     .idx_cnt = 1, .idx_meta = keys_uniq_meta},
+};
+
+/* Review finding (Task 3 follow-up): the unique shadow-check borrowed its
+ * candidate through wo_row_borrow, which shares ONE scratch buffer per
+ * table with the row already borrowed for the update itself — so the
+ * candidate borrow always failed (NULL), clash was always false, and a
+ * `resident: keys` table with a `@unique` index silently accepted
+ * duplicates on update. This is the test that would have caught it: two
+ * rows, update one's unique column to collide with the other's value, the
+ * update must be REFUSED and the row left exactly as it was. */
+static void test_keys_resident_update_unique_violation_refused(void) {
+    char path[128];
+    snprintf(path, sizeof path, "%s/keysuniq.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);
+
+    /* a's n: 100 -> 200 collides with b's live value — must be refused */
+    int ek = 0;
+    T_EQ(wo_row_update_field(&db, 0, a, 0, 200, &msg, &ek), -1);
+    T_EQ(ek, DB_ERR_UNIQUE);
+
+    /* a untouched: still 100, still the only hit for 100 */
+    db_row *r = wo_row_borrow(&db, 0, a, &msg);
+    T_CHECK(r != NULL && r->slots[0] == 100);
+    wo_row_release(&db, 0, r);
+    uint64_t *ids;
+    uint32_t cnt;
+    T_EQ(wo_idx_probe(&db, 0, 0, 100, NULL, 0, &ids, &cnt), 1);
+    T_CHECK(cnt == 1 && ids[0] == a);
+    free(ids);
+
+    /* b untouched: still the only hit for 200 */
+    T_EQ(wo_idx_probe(&db, 0, 0, 200, NULL, 0, &ids, &cnt), 1);
+    T_CHECK(cnt == 1 && ids[0] == b);
+    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,
@@ -1631,6 +1701,7 @@ int main(void) {
     test_fold_refuses_forward_pointing_delta();
     test_keys_resident_update_field();
     test_keys_resident_update_indexed();
+    test_keys_resident_update_unique_violation_refused();
     test_keys_resident_replay();
     test_keys_resident_survives_compaction();
     test_keys_resident_delete();