From 4f3f71e0034f760472987c5a419e9a166772110e Mon Sep 17 00:00:00 2001 From: "shoney.arickathil" Date: Sun, 30 Aug 2026 07:36:27 +0200 Subject: [PATCH] feat(db2-delta): fold a delta chain, route reads through it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - wal.h/wal.c: wo_wal_fold_row_at — THE fold. Walks BACKWARD from an offset through WO_WAL_DELTA records, remembering the first value seen per field index (newest wins, since newest is seen first), stops at the first INSERT/UPDATE, decodes it, overlays resolved fields. Returns ENGINE-owned values so reads, replay, and compaction (Tasks 3/5) can all build on the same output. - Cycle guard: caps the walk at what the log up to the starting offset could possibly hold (13 = scan_record's own record-size floor), so a corrupt or malicious back-pointer fails loudly instead of spinning. - table.c: wo_row_borrow's keys arm now calls the fold instead of wo_wal_read_row_at directly, then VM-decodes the result — same two-stage pattern wo_wal_read_row_at used internally. Per-table scratch, scratch_busy nested-borrow refusal, and the cid/id identity check all preserved unchanged. - resident: all path (wo_row_ptr) untouched. - test_wal.c: two new tests — deltas on two different fields (changed fields take the new value, the untouched field keeps its original) and two deltas on the SAME field (the newer wins, pinning direction — a reversed fold would pass with the older value instead). Verified failing pre-implementation (wo_row_borrow returned NULL since a delta record isn't INSERT/UPDATE) and passing after. Co-Authored-By: Claude Opus 5 (1M context) (cherry picked from commit a60231cde1d49d74743cedbd134d2da11158b70b) --- database/src/table.c | 27 ++++++- database/src/wal.c | 158 ++++++++++++++++++++++++++++++++++++++++ database/src/wal.h | 35 +++++++++ runtime/test/test_wal.c | 103 ++++++++++++++++++++++++++ 4 files changed, 321 insertions(+), 2 deletions(-) diff --git a/database/src/table.c b/database/src/table.c index 813982e..d1df97d 100644 --- a/database/src/table.c +++ b/database/src/table.c @@ -786,18 +786,41 @@ db_row *wo_row_borrow(wo_db *db, uint32_t class_id, uint64_t id, const char **ms t->scratch_cap = t->row_size; } db_row *r = (db_row *)t->scratch; + const wo_classdesc *c = &db->classes[class_id]; uint32_t got_cid = 0; uint64_t got_id = 0; - if (wo_wal_read_row_at((wo_wal *)db->rt->wal, db, db->rt, o1 - 1, &got_cid, &got_id, - r->slots, msg) != 0) + /* keys-resident delta updates, Task 2: the fold, not a single-record + * read — a row's current offset may point at a delta, not a base row. */ + uint64_t *eng = c->field_cnt ? calloc(c->field_cnt, sizeof *eng) : NULL; + if (c->field_cnt && !eng) { + 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) != + 0) { + free(eng); + return NULL; + } if (got_cid != class_id || got_id != id) { /* the offset pointed at someone else's record — a compaction that * moved records without rebuilding this map would land here, which is * exactly the obligation recorded at wo_wal_compact */ + for (uint32_t i = 0; i < c->field_cnt; i++) wo_db_val_free(db, c->kinds[i], eng[i]); + free(eng); if (msg) *msg = "log offset does not hold the expected row"; return NULL; } + /* two decode stages, same reason as wo_wal_read_row_at: the fold hands + back ENGINE-owned values, and the VM never sees those, so each one is + copied into a fresh VM value here before the engine originals free. */ + int ok = 1; + for (uint32_t i = 0; i < c->field_cnt; i++) { + r->slots[i] = wo_val_decode_vm(db, db->rt, c->kinds[i], eng[i], &ok, msg); + if (!ok) break; + } + for (uint32_t i = 0; i < c->field_cnt; i++) wo_db_val_free(db, c->kinds[i], eng[i]); + free(eng); + if (!ok) return NULL; r->id = id; r->class_id = class_id; r->flags = 0; diff --git a/database/src/wal.c b/database/src/wal.c index a6d2589..e4c07b2 100644 --- a/database/src/wal.c +++ b/database/src/wal.c @@ -873,6 +873,164 @@ int wo_wal_read_row_at(wo_wal *w, wo_db *db, wo_rt *rt, uint64_t off, return rc; } +/* keys-resident delta updates, Task 2: THE fold. See wal.h. Reads, replay, + * and compaction all call this one function — never a second copy. */ +int wo_wal_fold_row_at(wo_wal *w, wo_db *db, uint64_t off, uint32_t *class_out, + uint64_t *id_out, uint64_t *out_vals, const char **msg) { + /* Cycle guard: every legitimate back-pointer names a record already on + * disk before [off], so the chain cannot be longer than the number of + * minimal-sized records the bytes up to [off] could hold. 13 = the + * smallest an on-disk record can ever be (scan_record refuses len == 0, + * so 8-byte header + 1-byte payload + 4-byte mark). A malicious or + * corrupt back-pointer — even a record pointing at itself — still + * terminates here, loudly, instead of spinning forever. */ + uint64_t max_steps = off / 13u + 1u; + + uint32_t cid = 0; + uint64_t id = 0; + uint32_t field_cnt = 0; + uint8_t *resolved = NULL; /* field idx -> a delta already claimed it */ + uint64_t *resolved_val = NULL; /* field idx -> its remembered engine value */ + uint64_t cur = off; + int rc = 0; + + for (uint64_t step = 0;; step++) { + if (step >= max_steps) { + *msg = "delta chain exceeds what the log could hold (cycle?)"; + rc = -1; + break; + } + uint32_t len; + uint8_t *payload; + if (scan_record(w->fd, cur, &len, &payload) != 0) { + *msg = "no intact record at that offset"; + rc = -1; + break; + } + rbuf r = {payload, payload + len, 0}; + uint8_t kind = rd_u8(&r); + uint32_t rec_cid = rd_u32(&r); + uint64_t rec_id = rd_u64(&r); + if (r.bad || rec_cid >= db->class_cnt) { + free(payload); + *msg = "record header is malformed"; + rc = -1; + break; + } + if (step == 0) { + cid = rec_cid; + id = rec_id; + field_cnt = db->classes[cid].field_cnt; + resolved = field_cnt ? calloc(field_cnt, 1) : NULL; + resolved_val = field_cnt ? calloc(field_cnt, sizeof *resolved_val) : NULL; + if (field_cnt && (!resolved || !resolved_val)) { + free(payload); + *msg = "out of memory folding a row"; + rc = -2; + break; + } + } else if (rec_cid != cid || rec_id != id) { + free(payload); + *msg = "delta chain does not agree on row identity"; + rc = -1; + break; + } + + if (kind == WO_WAL_DELTA) { + uint32_t field_idx = rd_u32(&r); + uint64_t back_off = rd_u64(&r); + if (r.bad || field_idx >= field_cnt) { + free(payload); + *msg = "delta record is malformed"; + rc = -1; + break; + } + const wo_classdesc *c = &db->classes[cid]; + uint64_t v; + if (dec_val(&r, db, c->kinds[field_idx], &v) != 0 || + (size_t)(r.end - r.p) != 0) { + free(payload); + *msg = "delta record does not decode"; + rc = -1; + break; + } + if (!resolved[field_idx]) { + resolved[field_idx] = 1; + resolved_val[field_idx] = v; + } else { + /* shadowed by a newer delta already seen: still validated + above, but its value loses, exactly like the base row's */ + wo_db_val_free(db, c->kinds[field_idx], v); + } + free(payload); + cur = back_off; + continue; + } + + if (kind != WO_WAL_INSERT && kind != WO_WAL_UPDATE) { + free(payload); + *msg = "record in a delta chain is a tombstone, not a row"; + rc = -1; + break; + } + + /* base row: decode every field positionally (a delta-shadowed + field's bytes are still there — dec_val must still walk past + them to keep the fields after it aligned), then overlay whatever + the walk resolved. */ + const wo_classdesc *c = &db->classes[cid]; + uint64_t *slots = field_cnt ? calloc(field_cnt, sizeof *slots) : NULL; + if (field_cnt && !slots) { + free(payload); + *msg = "out of memory folding a row"; + rc = -2; + break; + } + uint32_t done = 0; + for (; done < field_cnt; done++) { + if (dec_val(&r, db, c->kinds[done], &slots[done]) != 0) { + *msg = "record at that offset does not decode"; + rc = -1; + break; + } + } + if (rc == 0 && done == field_cnt && (size_t)(r.end - r.p) != 0) { + *msg = "record at that offset has trailing bytes"; + rc = -1; + } + if (rc == 0 && done == field_cnt) { + for (uint32_t i = 0; i < field_cnt; i++) { + if (resolved[i]) { + wo_db_val_free(db, c->kinds[i], slots[i]); + slots[i] = resolved_val[i]; + } + } + memcpy(out_vals, slots, (size_t)field_cnt * sizeof *out_vals); + } else { + for (uint32_t i = 0; i < done; i++) wo_db_val_free(db, c->kinds[i], slots[i]); + } + free(slots); + free(payload); + break; + } + + if (rc != 0 && resolved && resolved_val) { + /* the walk resolved some fields but never reached a base row to + install them into: free what it was holding */ + const wo_classdesc *c = &db->classes[cid]; + for (uint32_t i = 0; i < field_cnt; i++) + if (resolved[i]) wo_db_val_free(db, c->kinds[i], resolved_val[i]); + } + free(resolved); + free(resolved_val); + + if (rc == 0) { + if (class_out) *class_out = cid; + if (id_out) *id_out = id; + } + return rc; +} + int64_t wo_wal_replay_ex(const char *path, wo_db *db, uint32_t *volatile_cid) { int fd = open(path, O_RDONLY); if (fd < 0) return errno == ENOENT ? 0 : -1; /* no WAL yet = fresh boot */ diff --git a/database/src/wal.h b/database/src/wal.h index 1b385e7..523abd3 100644 --- a/database/src/wal.h +++ b/database/src/wal.h @@ -268,6 +268,41 @@ int wo_wal_read_row_at(wo_wal *w, wo_db *db, wo_rt *rt, uint64_t off, uint32_t *class_out, uint64_t *id_out, uint64_t *out_vals, const char **msg); +/* keys-resident delta updates, Task 2: fold a delta chain into a row's + * CURRENT field values, walking BACKWARD from [off] until a full row + * (INSERT/UPDATE) is reached. + * + * [off] is the row's most recent record, exactly what wo_wal_read_row_at + * takes. Each delta names its predecessor's offset (the append-time + * back-pointer); the walk keeps hopping backward, remembering the FIRST + * value seen for each field index — the newest delta touching it, since + * newest is seen first — and skipping a delta whose field is already + * resolved. Reaching the base row decodes every field, then overlays + * whatever the walk resolved. + * + * out_vals[0..field_cnt) receive ENGINE-owned values (dec_val's + * representation, exactly what a slab row's own slots hold) — NOT VM + * values — so this one function serves every caller: a read decodes the + * result onward through wo_val_decode_vm, replay installs it straight into + * a freshly created row's slots, and compaction re-encodes it with enc_val + * into a fresh full-row record. The caller frees every slot with + * wo_db_val_free once done, on every path. This is the fold: written once, + * called by all three — a fold that disagreed between them would be a + * database that changes its mind at boot. + * + * [class_out] / [id_out] (optional) receive the row's identity, checked + * against EVERY record touched — a chain that disagrees about whose row it + * is is corruption, not a new row. + * + * A cycle in the back-pointers is corruption, possibly malicious: the walk + * refuses to look at more records than the log at [off] could possibly + * hold, and fails loudly instead of spinning. + * + * 0 ok, -1 no intact/malformed/corrupt record anywhere in the chain (or a + * REMOVE tombstone reached mid-chain), -2 out of memory (*msg set). */ +int wo_wal_fold_row_at(wo_wal *w, wo_db *db, uint64_t off, uint32_t *class_out, + uint64_t *id_out, uint64_t *out_vals, const char **msg); + /* Offline verification (no engine): scan [path], count intact records. * *intact_bytes (optional) = where the intact prefix ends. -1 = open * failure. */ diff --git a/runtime/test/test_wal.c b/runtime/test/test_wal.c index ad3e380..5a1f0c8 100644 --- a/runtime/test/test_wal.c +++ b/runtime/test/test_wal.c @@ -660,6 +660,107 @@ static void test_delta_record(void) { wo_rt_destroy(&rt); } +/* keys-resident delta updates, Task 2: the fold. A row's current record may + * be a chain of deltas, not a base row — wo_row_borrow must walk back + * through them, remembering one value per touched field, and overlay them + * onto the base row it eventually reaches. Reuses DELTA_CLASSES (3 scalar + * fields) so field 1 can stay untouched by any delta and prove the fold + * does not clobber fields nobody changed. */ +static void test_delta_fold_two_fields(void) { + char path[128]; + snprintf(path, sizeof path, "%s/foldtwo.wal", g_dir); + wo_rt rt; + T_EQ(wo_rt_init(&rt, 1 << 20, DELTA_CLASSES, 1), 0); + wo_db db; + T_EQ(wo_db_init(&db, DELTA_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 = ""; + + uint64_t vals[3] = {10, 20, 30}; + uint64_t id = wo_row_insert(&db, 0, vals, &msg, NULL); + T_CHECK(id != 0); + uint64_t base_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, base_off), 0); + + /* field 0: 10 -> 111 */ + uint64_t d1_off = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_delta(&w, &db, 0, id, 0, base_off, 111), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_set_offset(&db, 0, id, d1_off), 0); + + /* field 2: 30 -> 333, chained off the first delta */ + uint64_t d2_off = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_delta(&w, &db, 0, id, 2, d1_off, 333), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_set_offset(&db, 0, id, d2_off), 0); + + db_row *r = wo_row_borrow(&db, 0, id, &msg); + T_CHECK(r != NULL); + T_CHECK(r->slots[0] == 111); /* changed */ + T_CHECK(r->slots[1] == 20); /* untouched: original survives */ + T_CHECK(r->slots[2] == 333); /* changed */ + wo_row_release(&db, 0, r); + + wo_wal_close(&w); + wo_db_destroy(&db); + wo_rt_destroy(&rt); +} + +/* The ordering rule: two deltas to the SAME field. A fold walking the chain + * in the wrong direction sees the OLDER delta first and stops there — a + * plausible-looking but stale value, invisible unless a test pins the + * direction explicitly. */ +static void test_delta_fold_same_field_newest_wins(void) { + char path[128]; + snprintf(path, sizeof path, "%s/foldsame.wal", g_dir); + wo_rt rt; + T_EQ(wo_rt_init(&rt, 1 << 20, DELTA_CLASSES, 1), 0); + wo_db db; + T_EQ(wo_db_init(&db, DELTA_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 = ""; + + uint64_t vals[3] = {1, 2, 3}; + uint64_t id = wo_row_insert(&db, 0, vals, &msg, NULL); + T_CHECK(id != 0); + uint64_t base_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, base_off), 0); + + /* field 1: 2 -> 20 (older) -> 200 (newer) */ + uint64_t d1_off = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_delta(&w, &db, 0, id, 1, base_off, 20), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_set_offset(&db, 0, id, d1_off), 0); + + uint64_t d2_off = wo_wal_next_offset(&w); + T_EQ(wo_wal_append_delta(&w, &db, 0, id, 1, d1_off, 200), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_row_set_offset(&db, 0, id, d2_off), 0); + + db_row *r = wo_row_borrow(&db, 0, id, &msg); + T_CHECK(r != NULL); + T_CHECK(r->slots[0] == 1); + T_CHECK(r->slots[1] == 200); /* the NEWER delta wins, not the older */ + T_CHECK(r->slots[2] == 3); + wo_row_release(&db, 0, r); + + 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, @@ -1304,6 +1405,8 @@ int main(void) { test_compact_shortens_and_replays_equal(); test_keys_resident_round_trip(); test_delta_record(); + test_delta_fold_two_fields(); + test_delta_fold_same_field_newest_wins(); test_keys_resident_replay(); test_keys_resident_survives_compaction(); test_keys_resident_delete();