feat(db2-keys): rewire remaining readers, survive compaction

- wo_row_read and the @unique shadow probe go through borrow/release;
  release runs before every exit, including wo_row_read's early return
- updates on a keys table refused explicitly in wo_row_update_field and
  the slot variant: no slab slot to mutate, and writing the borrow's
  scratch would discard the write silently. Needs read-modify-append
- compaction walked the bitmap, which a keys row has no bit in — every
  such row would have been dropped from the new log. Now walks
  wo_row_next_id and re-points each row to where it lands
- moves records byte-for-byte (copy_record) rather than decoding: a
  borrowed row holds VM values, enc_val expects engine values, and ASan
  caught that mismatch as a 4294967292-byte memcpy
- wo_row_set_offset updates a value in place and never rehashes, so a
  wo_row_next_id cursor stays valid while compaction re-points
- a compaction that fails after moving rows is fatal: the map would name
  an unlinked temp file, and the intact log replays correctly
- test_keys_resident_survives_compaction pins both failure modes; rows
  rewrite in hash order so offsets really move
- loader still refuses resident: keys — updates are not implemented

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

database/src/table.c

@@ -365,7 +365,12 @@ int wo_idx_probe(wo_db *db, uint32_t class_id, uint32_t index, uint64_t key_scal
     if (!ids) return -1;
     uint32_t n = 0;
     for (uint32_t i = 0; i < b->len; i++) {
-        db_row *r = wo_row_ptr(db, class_id, b->ids[i]);
+        /* databasev2 2 (5d): THE unique shadow — the site the plan called the
+         * real coupling, because it needs a row it cannot get from a slab. For
+         * a keys-resident table each candidate costs a pread and a
+         * materialisation: the disclosed price of `@unique` there, bounded by
+         * the bucket rather than the table. */
+        db_row *r = wo_row_borrow(db, class_id, b->ids[i], NULL);
         if (!r) continue;
         int eq;
         if (kind == WO_K_TEXT) {
@@ -378,6 +383,7 @@ int wo_idx_probe(wo_db *db, uint32_t class_id, uint32_t index, uint64_t key_scal
              * exact comparison, so probe results never differ from scan
              * results (the hash canonicalized only to FIND the bucket) */
             eq = r->slots[col] == key_scalar;
+        wo_row_release(db, class_id, r); /* before any use of the result */
         if (eq) ids[n++] = b->ids[i];
     }
     if (!n) {
@@ -801,14 +807,20 @@ void wo_row_release(wo_db *db, uint32_t class_id, db_row *r) {
 
 int wo_row_read(wo_db *db, wo_rt *rt, uint32_t class_id, uint64_t id,
                 uint64_t *out_vals, const char **msg) {
-    db_row *r = wo_row_ptr(db, class_id, id);
+    db_row *r = wo_row_borrow(db, class_id, id, msg);
     if (!r) return -1;
     const wo_classdesc *c = &db->classes[class_id];
     int ok = 1;
     for (uint32_t i = 0; i < c->field_cnt; i++) {
+        /* decode out of the row BEFORE releasing: a keys-resident row's slots
+         * point into the scratch that release frees */
         out_vals[i] = db_val_decode(rt, c->kinds[i], r->slots[i], &ok, msg);
-        if (!ok) return -2;
+        if (!ok) {
+            wo_row_release(db, class_id, r);
+            return -2;
         }
+    }
+    wo_row_release(db, class_id, r);
     return 0;
 }
 
@@ -942,6 +954,17 @@ static int row_apply_field_slot(wo_db *db, db_table *t, const wo_classdesc *c,
 int wo_row_update_field(wo_db *db, uint32_t class_id, uint64_t id, uint32_t field,
                         uint64_t vm_val, const char **msg, int *err_kind) {
     if (err_kind) *err_kind = DB_ERR_MISC;
+    /* databasev2 2 (5d): a keys-resident row lives in the LOG, so there is no
+     * slab slot to mutate — writing into the borrow's scratch would discard
+     * the update silently, which is the one failure mode this iteration must
+     * not ship. Updating such a row means read-modify-APPEND (a new record,
+     * then re-point the offset), and that is not built yet. Refuse loudly.
+     * The loader refuses `resident: keys` outright, so this is defence in
+     * depth and a marker for the next implementer. */
+    if (wo_table_is_keys_resident(db, class_id)) {
+        *msg = "update on a `resident: keys` table is not implemented";
+        return -1;
+    }
     db_row *r = wo_row_ptr(db, class_id, id);
     if (!r) {
         *msg = "no such row";
@@ -1042,6 +1065,14 @@ static int row_apply_field_slot(wo_db *db, db_table *t, const wo_classdesc *c,
 int wo_row_update_field_slot(wo_db *db, uint32_t class_id, uint64_t id, uint32_t field,
                              uint64_t slot, const char **msg, int *err_kind) {
     if (err_kind) *err_kind = DB_ERR_MISC;
+    /* databasev2 2 (5d): same reason as wo_row_update_field — a keys-resident
+     * row has no slab slot to mutate, and writing into the borrow's scratch
+     * would discard the update silently. Read-modify-APPEND is the shape that
+     * works, and it is not built yet. */
+    if (wo_table_is_keys_resident(db, class_id)) {
+        *msg = "update on a `resident: keys` table is not implemented";
+        return -1;
+    }
     /* bounds first: the RPC requester validated cid/field to encode at all,
        so these are defensive; the slot's kind is unknowable on a class
        violation and the value leaks rather than dies by the wrong kind */
@@ -1174,3 +1205,36 @@ int wo_row_remove(wo_db *db, uint32_t class_id, uint64_t id) {
     t->free_slots[t->free_cnt++] = g;
     return 0;
 }
+
+/* databasev2 2 (5d): re-point a keys-resident row at a NEW log offset.
+ *
+ * Deliberately not hput(): hput runs the load-factor check and can rehash,
+ * which would reorder hkeys/hvals underneath a wo_row_next_id cursor. This
+ * only ever overwrites the value of a key that already exists, so the table's
+ * shape cannot change and a walk in progress stays valid. That property is
+ * what lets compaction re-point rows as it writes them instead of buffering
+ * one (cid, id, offset) triple per live row. Returns -1 if the id is absent. */
+int wo_row_set_offset(wo_db *db, uint32_t class_id, uint64_t id, uint64_t wal_off) {
+    if (class_id >= db->class_cnt) return -1;
+    db_table *t = &db->tables[class_id];
+    if (!t->hcap) return -1;
+    size_t j = hmix(id) & (t->hcap - 1);
+    while (t->hkeys[j]) {
+        if (t->hkeys[j] == id) {
+            t->hvals[j] = wal_off + 1;
+            return 0;
+        }
+        j = (j + 1) & (t->hcap - 1);
+    }
+    return -1;
+}
+
+/* databasev2 2 (5d): the log offset a keys-resident row currently reads from,
+ * as stored (off + 1), so 0 means "no such row". Compaction needs the raw
+ * offset to copy the record without materialising it. */
+uint64_t wo_row_offset1(const wo_db *db, uint32_t class_id, uint64_t id) {
+    if (class_id >= db->class_cnt) return 0;
+    const db_table *t = &db->tables[class_id];
+    if (!t->hcap) return 0;
+    return hget(t, id);
+}

database/src/table.h

@@ -193,6 +193,8 @@ int wo_row_remove(wo_db *db, uint32_t class_id, uint64_t id);
  *
  * 0 ok, -1 unknown class/row. */
 int wo_row_drop_payload(wo_db *db, uint32_t class_id, uint64_t id, uint64_t wal_off);
+int wo_row_set_offset(wo_db *db, uint32_t class_id, uint64_t id, uint64_t wal_off);
+uint64_t wo_row_offset1(const wo_db *db, uint32_t class_id, uint64_t id);
 
 /* databasev2 2 (5d): iterate the live row IDS of a table, whichever backing it
  * has. [*cursor] starts at 0 and is opaque; returns 1 with *id_out set, or 0

database/src/wal.c

@@ -390,6 +390,12 @@ static int stage(wo_wal *w, const wbuf *payload) {
 }
 
 int wo_wal_append_insert(wo_wal *w, wo_db *db, uint32_t class_id, uint64_t id) {
+    /* wo_row_ptr, NOT wo_row_borrow: at append time a keys-resident row is
+     * still in its slab and the id map still holds a SLOT, not an offset —
+     * the drop happens after the commit. Borrowing here would read the log at
+     * a byte position that is really a slot number. (Compaction, which does
+     * face rows that live only in the log, moves their bytes instead — see
+     * copy_record.) */
     db_row *r = wo_row_ptr(db, class_id, id);
     if (!r) return -1; /* commit order: RAM apply comes FIRST */
     wbuf p = {0};
@@ -404,6 +410,9 @@ int wo_wal_append_insert(wo_wal *w, wo_db *db, uint32_t class_id, uint64_t id) {
 }
 
 int wo_wal_append_update(wo_wal *w, wo_db *db, uint32_t class_id, uint64_t id) {
+    /* wo_row_ptr is right here for the same reason as append_insert, and the
+     * keys case cannot arrive at all: wo_row_update_field{,_slot} refuse a
+     * keys-resident table before any log record is staged. */
     db_row *r = wo_row_ptr(db, class_id, id);
     if (!r) return -1;
     wbuf p = {0};
@@ -563,10 +572,58 @@ static uint64_t mono_us(void) {
  * Nothing fails today because that storage half does not exist yet. It will
  * fail later, and it will look like data corruption rather than a design gap.
  * ==========================================================================*/
+/* databasev2 2 (5d): move one already-durable record from the old log into
+ * the new one, VERBATIM.
+ *
+ * Compaction cannot re-encode a keys-resident row the way it re-encodes a
+ * resident one. enc_val expects the ENGINE representation a slab row holds
+ * (db_text: len + bytes), while a row read back out of the log arrives in the
+ * VM representation (wo_str: len + data). The two are not the same struct, so
+ * feeding a borrowed row to enc_val reads the length out of the wrong field —
+ * ASan caught exactly that as a 4294967292-byte memcpy.
+ *
+ * Copying the bytes sidesteps the whole question, and is strictly better than
+ * decode-then-encode anyway: no allocation per row, no arena pressure, and the
+ * record that lands is bit-identical to the one that was acked. scan_record
+ * verifies the CRC, so a torn record is refused rather than propagated.
+ *
+ * The kind byte is normalised to INSERT: a live row's newest image may have
+ * been logged as an UPDATE, and the compacted log is supposed to read as one
+ * INSERT per live row. */
+static int copy_record(wo_wal *nw, wo_wal *ow, uint64_t off, uint32_t want_cid,
+                       uint64_t want_id, const char **why) {
+    uint32_t len = 0;
+    uint8_t *payload = NULL;
+    if (scan_record(ow->fd, off, &len, &payload) != 0) {
+        *why = "no intact record at a row's recorded offset";
+        return -1;
+    }
+    rbuf r = {payload, payload + len, 0};
+    uint8_t kind = rd_u8(&r);
+    uint32_t cid = rd_u32(&r);
+    uint64_t id = rd_u64(&r);
+    if (r.bad || cid != want_cid || id != want_id ||
+        (kind != WO_WAL_INSERT && kind != WO_WAL_UPDATE)) {
+        free(payload);
+        *why = "a row's recorded offset does not hold that row";
+        return -1;
+    }
+    payload[0] = WO_WAL_INSERT;
+    wbuf p = {0};
+    wput(&p, payload, len);
+    int rc = stage(nw, &p);
+    free(p.b);
+    free(payload);
+    if (rc != 0) *why = "staging a moved record failed";
+    return rc;
+}
+
 int wo_wal_compact(wo_wal *w, wo_db *db) {
     /* staged records would be written into a file about to be replaced */
     if (!w->path || w->len != 0) return -1;
     uint64_t t0 = mono_us();
+    int repointed = 0; /* has any keys-resident row been moved to the new log? */
+    const char *why = NULL; /* the reason a fail arm was taken, when it is known */
 
     char tmp[4096];
     if ((size_t)snprintf(tmp, sizeof tmp, "%s%s", w->path, WO_WAL_TMP_SUFFIX) >= sizeof tmp)
@@ -587,15 +644,40 @@ int wo_wal_compact(wo_wal *w, wo_db *db) {
      * append path — so replay needs no second decoder and ids are preserved
      * exactly (wo_wal_append_insert takes the id and reads the row) */
     uint32_t pending = 0;
+    /* databasev2 2 (5d): the walk is wo_row_next_id, not the bitmap. A
+     * keys-resident row has NO bitmap bit — its slot was returned to the free
+     * list when the payload was dropped — so a bitmap walk would omit every
+     * such row from the new log and call it compaction. That is silent data
+     * loss, and it is the failure the obligation note above was written for. */
     for (uint32_t cid = 0; cid < db->class_cnt; cid++) {
         db_table *t = &db->tables[cid];
-        if (!t->slabs) continue; /* tables are created lazily */
+        if (!t->row_size) continue; /* tables are created lazily */
-        uint32_t total = t->slab_cnt * DB_SLAB_ROWS;
+        int keys = wo_table_is_keys_resident(db, cid);
-        for (uint32_t g = 0; g < total; g++) {
+        size_t cur = 0;
-            if (!(t->bitmap[g >> 6] & (1ull << (g & 63)))) continue;
+        uint64_t id;
-            db_row *r = (db_row *)(t->slabs[g / DB_SLAB_ROWS] +
+        while (wo_row_next_id(db, cid, &cur, &id)) {
-                                   (size_t)(g % DB_SLAB_ROWS) * t->row_size);
+            if (!keys) {
-            if (wo_wal_append_insert(&nw, db, cid, r->id) != 0) goto fail;
+                if (wo_wal_append_insert(&nw, db, cid, id) != 0) goto fail;
+            } else {
+                /* read from the OLD log (still open, still the live file at
+                 * this point), write into the new one, and re-point the map
+                 * to where it landed. The offset is captured BEFORE staging:
+                 * off is what has reached the file, len what is staged behind
+                 * it, so their sum is the position of the next record. */
+                uint64_t o1 = wo_row_offset1(db, cid, id);
+                if (!o1) {
+                    why = "a live keys-resident row has no recorded offset";
+                    goto fail;
+                }
+                uint64_t at = nw.off + (uint64_t)nw.len;
+                if (copy_record(&nw, w, o1 - 1, cid, id, &why) != 0) goto fail;
+                /* value-only update: cannot rehash, so `cur` stays valid */
+                if (wo_row_set_offset(db, cid, id, at) != 0) {
+                    why = "row vanished from the id map mid-compaction";
+                    goto fail;
+                }
+                repointed = 1;
+            }
             if (++pending >= WO_WAL_COMPACT_FLUSH) {
                 if (wal_write_nosync(&nw) != 0) goto fail;
                 pending = 0;
@@ -635,6 +717,21 @@ int wo_wal_compact(wo_wal *w, wo_db *db) {
 fail:
     wo_wal_close(&nw);
     (void)unlink(tmp);
+    if (repointed) {
+        /* databasev2 2 (5d): rows already re-pointed name offsets inside the
+         * temp file just unlinked, so the id map now describes a file that no
+         * longer exists — reads would return another row's bytes or nothing.
+         * The log ON DISK is still the intact original, so replay rebuilds the
+         * map correctly; carrying on in this process cannot. Same doctrine as
+         * a failed commit barrier: stop rather than serve wrong rows. */
+        fprintf(stderr,
+                "writeonce: DURABILITY FAILURE — compaction of %s failed after "
+                "moving rows: %s\n"
+                "  No data was lost: the original log is intact on disk.\n"
+                "  The process is stopping: replay rebuilds the row offsets.\n",
+                w->path, why ? why : "write or sync error");
+        exit(WO_EXIT_DURABILITY);
+    }
     return -1; /* the live log is untouched and still usable */
 }
 

docs/stories/databasev2/02-table-storage-modes.md

@@ -61,8 +61,8 @@ declared per-table policy. Durability is untouched and unconditional.
 | 4 | `durable: false` skips the WAL append and replay | ✅ `dd67e31` |
 | 5a | `wo_wal_next_offset` — exact record offsets | ✅ `ac7d8af` |
 | 5b | `wo_wal_read_row_at` — a row from a log offset | ✅ `d0c370c` |
-| 5c | shared borrow/release accessor, then id→offset storage | 🔄 step 1 ✅ `2e347de` (pure refactor, `db-bench --quick` 85/0); offset storage next |
+| 5c | shared borrow/release accessor, then id→offset storage | ✅ `2e347de` (accessor, pure refactor, `db-bench --quick` 85/0), `18ce4d5` (offset storage), `f9c36ef` (insert + boot wiring) |
-| 5d | rewire the readers: remaining `wo_row_ptr` sites (6 table.c, 2 db.c, 2 wal.c), slab scans, FK restrict, `@unique` across the boundary | ⬜ scope recorded |
+| 5d | rewire the readers: remaining `wo_row_ptr` sites, slab scans, FK restrict, `@unique` across the boundary | ✅ `11a92df` (db.c), + this commit (table.c, wal.c, compaction). Updates **refused**, not rewired — see below |
 | 6 | the two runtime refusals (no-`WO_DATA`, the byte budget) | ⬜ |
 | 7 | measure, gate, document, close out | ⬜ |
 
@@ -102,15 +102,36 @@ Met:
 - **Given** a v6 image, **when** loaded, **then** refused on version rather
   than misread. ✅
 
+- **Given** a `resident: keys` table, **when** rows are read by id and scanned,
+  **then** every row is byte-identical including heap-valued columns. ✅ 5d.
+  Every read path goes through `wo_row_borrow`/`wo_row_release`, and the scans
+  go through `wo_row_next_id` — deliberately the id map for a keys table and
+  the bitmap for a resident one, since hash order would reorder every
+  unordered query.
+- **Given** `@unique` on a `resident: keys` table, **when** a duplicate arrives
+  whose conflicting row is not resident, **then** it is refused. ✅ 5d. The
+  shadow probe borrows each bucket candidate, so the check costs one `pread`
+  per candidate — bounded by the bucket, not the table — and never silently
+  narrows to the resident subset.
+- **Given** a `resident: keys` table and a WAL checkpoint, **when** the log is
+  compacted, **then** every such row survives and still reads correctly. ✅ 5d,
+  and this is the obligation databasev2 3 left behind. Two independent ways to
+  fail it, both pinned by `test_keys_resident_survives_compaction`: compaction
+  walked the *bitmap*, which a keys row has no bit in, so every one of them
+  would have been dropped from the new log; and the id map would still have
+  named offsets into the replaced file. Rows are rewritten in hash order, so
+  offsets genuinely move and a missing re-point cannot pass by luck.
+
 Outstanding:
 
-- **Given** a `resident: keys` table larger than any plausible resident budget,
+- **Given** an `update` to a row on a `resident: keys` table, **when** it runs,
-  **when** rows are read by id and scanned, **then** every row is byte-identical
+  **then** it is applied. ❌ **refused explicitly** by
-  including heap-valued columns. *(needs 5c/5d)*
+  `wo_row_update_field{,_slot}`. A keys row lives in the log with no slab slot
-- **Given** `@unique` on a `resident: keys` table, **when** a duplicate arrives
+  to mutate; writing into the borrow's scratch would discard the write
-  whose conflicting row is not resident, **then** it is refused. *(5d — the
+  *silently*, which is the one failure this iteration must not ship. Doing it
-  correctness core; a constraint that silently checks only resident rows must
+  properly is read-modify-**append** — a new record, then re-point the offset —
-  never ship)*
+  and that is its own piece of work. **The loader's refusal of `resident: keys`
+  stays until it lands**, so no program can reach the half-feature.
 - **Given** `durable: true` and no `WO_DATA`, **when** the program starts,
   **then** it refuses. *(task 6 — today this combination silently discards
   every write)*

runtime/test/test_wal.c

@@ -578,6 +578,77 @@ static void test_keys_resident_round_trip(void) {
  * 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,
  * exactly as a restart would. */
+static void test_keys_resident_survives_compaction(void) {
+    /* databasev2 2 (5d): the obligation recorded at wo_wal_compact. Two ways
+     * to fail it, both checked here:
+     *   1. compaction walks the bitmap, so keys-resident rows — which hold no
+     *      bitmap bit — are never written to the new log and vanish;
+     *   2. compaction writes them but leaves the id map naming OLD offsets.
+     * Rows are written back in HASH order, not insertion order, so almost
+     * every offset really does move: a map left un-repointed cannot pass by
+     * coincidence, it lands on another row and fails the id check. */
+    char path[128];
+    snprintf(path, sizeof path, "%s/keyscompact.wal", g_dir);
+    wo_rt rt;
+    T_EQ(wo_rt_init(&rt, 1 << 20, KEYS_CLASSES, 1), 0);
+    wo_db db;
+    T_EQ(wo_db_init(&db, KEYS_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 = "";
+
+    enum { N = 5 };
+    uint64_t ids[N];
+    char texts[N][8];
+    for (int i = 0; i < N; i++) {
+        /* varying lengths, so a record's position depends on what precedes it */
+        int tl = 1 + i;
+        memset(texts[i], 'a' + i, (size_t)tl);
+        texts[i][tl] = 0;
+        wo_str *sv = wo_str_new(&rt, texts[i], (size_t)tl);
+        uint64_t vals[2] = {(uint64_t)(i * 101 + 7), (uint64_t)(uintptr_t)sv};
+        ids[i] = wo_row_insert(&db, 0, vals, &msg, NULL);
+        T_CHECK(ids[i] != 0);
+        uint64_t off = wo_wal_next_offset(&w);
+        T_EQ(wo_wal_append_insert(&w, &db, 0, ids[i]), 0);
+        T_EQ(wo_wal_commit(&w), 0);
+        T_EQ(wo_row_drop_payload(&db, 0, ids[i], off), 0);
+    }
+
+    T_EQ(wo_wal_compact(&w, &db), 0);
+
+    /* every row still readable, with its own values, through the new log */
+    for (int i = 0; i < N; i++) {
+        db_row *r = wo_row_borrow(&db, 0, ids[i], &msg);
+        T_CHECK(r != NULL);
+        T_CHECK(r->slots[0] == (uint64_t)(i * 101 + 7));
+        wo_str *back = (wo_str *)(uintptr_t)r->slots[1];
+        T_CHECK(back != NULL && back->len == (size_t)(1 + i));
+        T_CHECK(memcmp(back->data, texts[i], (size_t)(1 + i)) == 0);
+        wo_row_release(&db, 0, r);
+    }
+    wo_wal_close(&w);
+    wo_db_destroy(&db);
+
+    /* and the compacted log replays to the same set in a fresh process */
+    wo_db db2;
+    T_EQ(wo_db_init(&db2, KEYS_CLASSES, 1, 0, 1), 0);
+    T_EQ(wo_wal_replay(path, &db2), N);
+    wo_wal w2;
+    T_EQ(wo_wal_open(&w2, path, 1 << 16), 0);
+    db2.rt = &rt; rt.wal = &w2; rt.db = &db2;
+    for (int i = 0; i < N; i++) {
+        db_row *r = wo_row_borrow(&db2, 0, ids[i], &msg);
+        T_CHECK(r != NULL);
+        T_CHECK(r->slots[0] == (uint64_t)(i * 101 + 7));
+        wo_row_release(&db2, 0, r);
+    }
+    wo_wal_close(&w2);
+    wo_db_destroy(&db2);
+    wo_rt_destroy(&rt);
+}
+
 static void test_keys_resident_replay(void) {
     char path[128];
     snprintf(path, sizeof path, "%s/keysboot.wal", g_dir);
@@ -1047,6 +1118,7 @@ int main(void) {
     test_compact_shortens_and_replays_equal();
     test_keys_resident_round_trip();
     test_keys_resident_replay();
+    test_keys_resident_survives_compaction();
     test_stale_compact_temp_is_removed();
     test_should_compact_policy();
     test_compact_refuses_with_staged_records();