fix(db2-delta): borrow the pending re-point, not the stale durable offset

- wo_row_borrow's keys arm folded at hget()'s DURABLE offset even
  when an earlier update in the same drain had only a PENDING
  re-point
- idx_remove_row then hashed the pre-first-update value, found no
  matching bucket entry (already moved by the earlier update), and
  idx_add_row added a second one — N same-drain updates leaked N-1
  entries, unbounded, nothing reclaims them but a restart
- now prefers wo_wal_repoint_offset1() over the durable offset, same
  as back_off already does, closing it for every borrow
- new test: 5 updates to one row in one drain, assert exactly one
  index entry — fails (5) before the fix, passes (1) after

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

database/src/table.c

@@ -808,14 +808,24 @@ db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **ms
      * 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;
+    uint64_t durable1 = hget(t, id);
+    if (!durable1) 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;
     }
+    /* CRITICAL 2 (review finding): a row already updated once behind this
+     * not-yet-committed barrier has its re-point only PENDING — hget still
+     * names the pre-drain durable offset. Folding there hands back the
+     * row's value from BEFORE the earlier update, which made every caller
+     * (row_apply_field_keys's idx_remove_row included) hash stale column
+     * values and leak an index entry per repeat update in one drain.
+     * Preferring the pending re-point, same as back_off already does below,
+     * closes it for every borrow, not just the update path. */
+    uint64_t pending1 = wo_wal_repoint_offset1((wo_wal *)db->rt->wal, class_id, id);
+    uint64_t o1 = pending1 ? pending1 : durable1;
     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

runtime/test/test_wal.c

@@ -1249,6 +1249,81 @@ static void test_keys_resident_two_updates_one_drain(void) {
     wo_rt_destroy(&rt);
 }
 
+/* CRITICAL 2 (review finding): wo_row_borrow itself must prefer a pending
+ * re-point over the durable map, the same reason back_off and the unique
+ * shadow-check's candidate lookup already do. Without it, the SECOND (and
+ * every later) update to one row in one drain borrows the row via `hget` —
+ * the still-DURABLE, pre-drain offset — so row_apply_field_keys's
+ * idx_remove_row hashes the row's ORIGINAL column value. That value was
+ * already removed from the index by the FIRST update in this drain, so the
+ * remove finds nothing, and idx_add_row adds a SECOND entry. N updates to
+ * one row in one drain used to leave N entries for it; this asserts
+ * exactly one, however many updates ran. */
+static void test_keys_resident_repeat_updates_one_drain_index(void) {
+    char path[128];
+    snprintf(path, sizeof path, "%s/keysrepeatidx.wal", g_dir);
+    wo_rt rt;
+    T_EQ(wo_rt_init(&rt, 1 << 20, KEYS_IDX_CLASSES, 1), 0);
+    wo_db db;
+    T_EQ(wo_db_init(&db, KEYS_IDX_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 *s = wo_str_new(&rt, "x", 1);
+    uint64_t vals[2] = {100, (uint64_t)(uintptr_t)s};
+    uint64_t id = wo_row_insert(&db, 0, vals, &msg, NULL);
+    T_CHECK(id != 0);
+    uint64_t off = wo_wal_next_offset(&w);
+    T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0);
+    T_EQ(wo_wal_commit(&w), 0);
+    T_EQ(wo_row_drop_payload(&db, 0, id, off), 0);
+
+    /* 5 updates to the SAME row, ALL staged behind the SAME barrier —
+       db.c's request-path shape: stage each, remember its pending
+       re-point, only commit + flush once at the end of the drain. */
+    int ek = 0;
+    uint64_t next_v = 200;
+    for (int i = 0; i < 5; i++) {
+        uint64_t roff = wo_wal_next_offset(&w);
+        T_EQ(wo_row_update_field_slot(&db, 0, id, 0, next_v, &msg, &ek), 0);
+        T_EQ(ek, DB_ERR_NONE);
+        T_EQ(wo_wal_pend_repoint(&w, 0, id, roff), 0);
+        next_v += 100;
+    }
+    T_EQ(wo_wal_commit(&w), 0);
+    wo_db_flush_drops(&db, &w);
+
+    /* exactly one entry for this row, across EVERY bucket — a leaked entry
+       would sit in a STALE bucket (the pre-first-update value's), not the
+       current one, so this must scan the whole index, not just probe. */
+    db_table *t = &db.tables[0];
+    db_index *ix = &t->indexes[0];
+    uint32_t hits = 0;
+    for (size_t bi = 0; bi < ix->bcap; bi++) {
+        db_ibucket *b = &ix->buckets[bi];
+        for (uint32_t k = 0; k < b->len; k++)
+            if (b->ids[k] == id) hits++;
+    }
+    T_EQ(hits, 1u);
+
+    /* and it is reachable by its final value, 600 */
+    uint64_t *ids;
+    uint32_t cnt;
+    T_EQ(wo_idx_probe(&db, 0, 0, 600, NULL, 0, &ids, &cnt), 1);
+    T_CHECK(cnt == 1 && ids[0] == id);
+    free(ids);
+
+    db_row *r = wo_row_borrow(&db, 0, id, &msg);
+    T_CHECK(r != NULL && r->slots[0] == 600);
+    wo_row_release(&db, 0, r);
+
+    wo_wal_close(&w);
+    wo_db_destroy(&db);
+    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
@@ -2320,6 +2395,7 @@ int main(void) {
     test_keys_resident_update_indexed_text();
     test_keys_resident_update_unique_violation_refused();
     test_keys_resident_two_updates_one_drain();
+    test_keys_resident_repeat_updates_one_drain_index();
     test_keys_resident_unique_clash_pending_repoint();
     test_keys_resident_replay();
     test_keys_resident_survives_compaction();