From f07295b3c07b6ad86e477575b4837df9c7a443aa Mon Sep 17 00:00:00 2001 From: "shoney.arickathil" Date: Mon, 31 Aug 2026 21:21:23 +0200 Subject: [PATCH] =?UTF-8?q?feat(db2-migrate):=20WO=5FWAL=5FSCHEMA=20?= =?UTF-8?q?=E2=80=94=20the=20log=20states=20the=20shape=20that=20wrote=20i?= =?UTF-8?q?t?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - new record kind 5: class and field NAMES, kinds and the two encoding-relevant metadata words (field_class, field_elem), CRC-framed like every record. Index layout deliberately absent: indexes rebuild from rows at boot and never touch record bytes - names are byte pointers, not constant-pool indices — the database layer never sees the module's consts, so the runtime resolves them once; a decoded schema owns a private copy of its bytes - wo_wal_set_schema adopts the compiled schema; wo_wal_ensure_schema writes it as a fresh log's first record; compaction writes it at the head of every replacement, which is how a legacy log becomes self-describing without a migration step of its own - apply_record skips it BEFORE reading cid/id (its class count would be misread as a cid and bounds-refused); replay does not count it - schema unset = byte-for-byte today's behaviour: all 5700 prior assertions pass untouched; four new tests cover roundtrip, fresh-log head, legacy adoption via compaction, and absent/empty files - test_wal 5743 pass, 0 fail Co-Authored-By: Claude Opus 5 (1M context) (cherry picked from commit ba8519fa5e39c6ed6a3504d39e5a372f8decd045) --- database/src/wal.c | 146 +++++++++++++++++++++++++++++++++++++++- database/src/wal.h | 72 +++++++++++++++++++- runtime/test/test_wal.c | 141 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 357 insertions(+), 2 deletions(-) diff --git a/database/src/wal.c b/database/src/wal.c index 3cfa0c0..55cd3d4 100644 --- a/database/src/wal.c +++ b/database/src/wal.c @@ -420,6 +420,9 @@ void wo_db_flush_drops(wo_db *db, wo_wal *w) { } void wo_wal_close(wo_wal *w) { + free(w->schema); + w->schema = NULL; + w->schema_len = 0; free(w->pend); free(w->repoint); if (w->fd >= 0) close(w->fd); @@ -458,6 +461,131 @@ static int stage(wo_wal *w, const wbuf *payload) { return 0; } +/* ---- databasev2 12: the schema record -------------------------------- */ + +int wo_schema_encode(const wo_schema *sc, uint8_t **payload_out, uint32_t *len_out) { + wbuf p = {0}; + wput_u8(&p, WO_WAL_SCHEMA); + wput_u32(&p, sc->class_cnt); + for (uint32_t c = 0; c < sc->class_cnt; c++) { + const wo_schema_class *k = &sc->classes[c]; + wput_u32(&p, k->name_len); + wput(&p, k->name, k->name_len); + wput_u32(&p, k->flags); + wput_u32(&p, k->field_cnt); + for (uint32_t f = 0; f < k->field_cnt; f++) { + const wo_schema_field *fl = &k->fields[f]; + wput_u32(&p, fl->name_len); + wput(&p, fl->name, fl->name_len); + wput_u8(&p, fl->kind); + wput_u32(&p, fl->fclass); + wput_u32(&p, fl->felem); + } + } + if (p.oom) { + free(p.b); + return -1; + } + *payload_out = p.b; + *len_out = (uint32_t)p.len; + return 0; +} + +wo_schema *wo_schema_decode(const uint8_t *payload, uint32_t len) { + if (len < 5 || payload[0] != WO_WAL_SCHEMA) return NULL; + /* names point into one private copy of the bytes, so the schema outlives + * whatever buffer the caller hands in */ + wo_schema *sc = calloc(1, sizeof *sc); + if (!sc) return NULL; + sc->owned = malloc(len); + if (!sc->owned) { + free(sc); + return NULL; + } + memcpy(sc->owned, payload, len); + rbuf r = {sc->owned + 1, sc->owned + len, 0}; + sc->class_cnt = rd_u32(&r); + if (r.bad || sc->class_cnt > 65536) goto bad; + sc->classes = calloc(sc->class_cnt ? sc->class_cnt : 1, sizeof *sc->classes); + if (!sc->classes) goto bad; + for (uint32_t c = 0; c < sc->class_cnt; c++) { + wo_schema_class *k = &sc->classes[c]; + k->name_len = rd_u32(&r); + if (r.bad || (size_t)(r.end - r.p) < k->name_len) goto bad; + k->name = r.p; + r.p += k->name_len; + k->flags = rd_u32(&r); + k->field_cnt = rd_u32(&r); + if (r.bad || k->field_cnt > 65536) goto bad; + k->fields = calloc(k->field_cnt ? k->field_cnt : 1, sizeof *k->fields); + if (!k->fields) goto bad; + for (uint32_t f = 0; f < k->field_cnt; f++) { + wo_schema_field *fl = &k->fields[f]; + fl->name_len = rd_u32(&r); + if (r.bad || (size_t)(r.end - r.p) < fl->name_len) goto bad; + fl->name = r.p; + r.p += fl->name_len; + fl->kind = rd_u8(&r); + fl->fclass = rd_u32(&r); + fl->felem = rd_u32(&r); + if (r.bad) goto bad; + } + } + if (r.p != r.end) goto bad; /* trailing bytes = malformed */ + return sc; +bad: + wo_schema_free(sc); + return NULL; +} + +void wo_schema_free(wo_schema *sc) { + if (!sc) return; + if (sc->classes) + for (uint32_t c = 0; c < sc->class_cnt; c++) free(sc->classes[c].fields); + free(sc->classes); + free(sc->owned); + free(sc); +} + +int wo_wal_set_schema(wo_wal *w, const wo_schema *sc) { + uint8_t *p; + uint32_t len; + if (wo_schema_encode(sc, &p, &len) != 0) return -1; + free(w->schema); + w->schema = p; + w->schema_len = len; + return 0; +} + +int wo_wal_ensure_schema(wo_wal *w) { + if (!w->schema) return 0; /* never set: legacy behaviour */ + if (w->off != 0 || w->len != 0) return 0; /* records exist or staged */ + wbuf p = {0}; + wput(&p, w->schema, w->schema_len); + int rc = stage(w, &p); + free(p.b); + if (rc != 0) return -1; + return wo_wal_commit(w); +} + +int wo_wal_read_schema(const char *path, uint8_t **payload_out, uint32_t *len_out) { + int fd = open(path, O_RDONLY); + if (fd < 0) return errno == ENOENT ? 1 : -1; + uint32_t len; + uint8_t *payload; + int rc = scan_record(fd, 0, &len, &payload); + close(fd); + if (rc != 0) return 1; /* empty or torn head: a legacy log */ + if (len < 1 || payload[0] != WO_WAL_SCHEMA) { + free(payload); + return 1; + } + if (payload_out) *payload_out = payload; + else free(payload); + if (len_out) *len_out = len; + return 0; +} + 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 — @@ -830,6 +958,18 @@ int wo_wal_compact(wo_wal *w, wo_db *db) { * missing, with the log otherwise intact and self-consistent. */ if (wo_wal_open(&nw, tmp, w->prealloc) != 0) return -1; + /* databasev2 12: the replacement log's first record is the schema, so a + * compacted log is always self-describing — including the first + * compaction of a legacy log, which is how existing WO_DATA dirs become + * diffable without any migration step of their own. */ + if (w->schema) { + wbuf sp = {0}; + wput(&sp, w->schema, w->schema_len); + int src = stage(&nw, &sp); + free(sp.b); + if (src != 0) goto fail; + } + /* one INSERT per live row, in the existing grammar, through the existing * append path — so replay needs no second decoder and ids are preserved * exactly (wo_wal_append_insert takes the id and reads the row) */ @@ -1016,6 +1156,10 @@ static int apply_delta(wo_db *db, uint32_t cid, uint64_t id, rbuf *r) { static int apply_record(wo_db *db, const uint8_t *payload, uint32_t len) { rbuf r = {payload, payload + len, 0}; uint8_t kind = rd_u8(&r); + /* databasev2 12: a schema record is descriptive, not a row — and it has + * no cid/id fields, so it must be skipped BEFORE those are read (its + * class count would be misread as a cid and bounds-refused). */ + if (kind == WO_WAL_SCHEMA) return 0; uint32_t cid = rd_u32(&r); uint64_t id = rd_u64(&r); if (r.bad || cid >= db->class_cnt) return -1; @@ -1382,7 +1526,7 @@ int64_t wo_wal_replay_ex(const char *path, wo_db *db, uint32_t *volatile_cid) { rec_id && wo_table_is_keys_resident(db, rec_cid)) (void)wo_row_drop_payload(db, rec_cid, rec_id, off); off += 8u + len + 4u; - applied++; + if (rec_kind != WO_WAL_SCHEMA) applied++; } close(fd); REPLAY_RETURN(applied); diff --git a/database/src/wal.h b/database/src/wal.h index a61d075..0675234 100644 --- a/database/src/wal.h +++ b/database/src/wal.h @@ -46,7 +46,19 @@ #define WO_WAL_MARK 0x574F4C31u /* "WOL1" LE */ -enum { WO_WAL_INSERT = 1, WO_WAL_REMOVE = 2, WO_WAL_UPDATE = 3, WO_WAL_DELTA = 4 }; +enum { + WO_WAL_INSERT = 1, + WO_WAL_REMOVE = 2, + WO_WAL_UPDATE = 3, + WO_WAL_DELTA = 4, + /* databasev2 12: the log's own statement of the shape that wrote it — + * class and field NAMES, kinds and encoding-relevant metadata. Written as + * the FIRST record of a fresh log and of every compacted log, so the head + * of a log always describes everything after it. Replay skips it; boot + * diffs it against the compiled classes to migrate or refuse. A log + * without one is a legacy log: nothing recorded, nothing diffable. */ + WO_WAL_SCHEMA = 5, +}; typedef struct wo_wal { int fd; @@ -101,8 +113,66 @@ typedef struct wo_wal { uint64_t stat_compactions; uint64_t stat_compact_us_max; uint64_t stat_compact_us_total; + /* databasev2 12: the encoded WO_WAL_SCHEMA payload for the COMPILED + * classes, set once at boot by wo_wal_set_schema. Owned here, freed by + * wo_wal_close. When set, a fresh log gets it as its first record + * (wo_wal_ensure_schema) and compaction writes it at the head of every + * replacement log. When unset (every existing test, and legacy boots) + * nothing changes anywhere. */ + uint8_t *schema; + uint32_t schema_len; } wo_wal; +/* databasev2 12: the schema a log carries, and the diff against the compiled + * one. Names are byte pointers, NOT constant-table indices — the database + * layer never sees the module's constant pool, so the runtime resolves names + * once when it builds the compiled-side schema, and a decoded schema's names + * point into the record's own bytes. `fclass`/`felem` mirror the classdesc's + * field_class/field_elem because they change how a value is ENCODED; index + * layout is deliberately absent — indexes are rebuilt from rows at boot and + * never touch record bytes. */ +typedef struct wo_schema_field { + const uint8_t *name; + uint32_t name_len; + uint8_t kind; + uint32_t fclass; /* referenced class id, or WO_SCHEMA_NONE */ + uint32_t felem; /* container element kinds, or WO_SCHEMA_NONE */ +} wo_schema_field; +typedef struct wo_schema_class { + const uint8_t *name; + uint32_t name_len; + uint32_t flags; + uint32_t field_cnt; + wo_schema_field *fields; +} wo_schema_class; +typedef struct wo_schema { + uint32_t class_cnt; + wo_schema_class *classes; + uint8_t *owned; /* decode backing buffer; NULL on a caller-built schema */ +} wo_schema; +#define WO_SCHEMA_NONE 0xFFFFFFFFu + +/* Encode a schema as a WO_WAL_SCHEMA record payload (kind byte included). + * Returns 0 and a malloc'd buffer the caller frees. */ +int wo_schema_encode(const wo_schema *sc, uint8_t **payload_out, uint32_t *len_out); +/* Decode a WO_WAL_SCHEMA payload. NULL = malformed. Free the result with + * wo_schema_free; its name pointers live in the returned struct's own copy + * of the bytes, not in the caller's buffer. */ +wo_schema *wo_schema_decode(const uint8_t *payload, uint32_t len); +void wo_schema_free(wo_schema *sc); + +/* Adopt `sc` as this log's compiled schema (encoded and owned by the wal). */ +int wo_wal_set_schema(wo_wal *w, const wo_schema *sc); +/* A fresh, empty log gets the schema as its first record — durable before + * any row record can be staged behind it. No-op when a schema was never set + * or when records already exist (a legacy log stays legacy until its next + * compaction writes the record at the head of the replacement). */ +int wo_wal_ensure_schema(wo_wal *w); +/* Peek the log's head record. 0 = schema record found (*payload_out is + * malloc'd, caller frees); 1 = no log, empty log, or a legacy head record; + * -1 = I/O error. */ +int wo_wal_read_schema(const char *path, uint8_t **payload_out, uint32_t *len_out); + /* databasev2 2: the file offset the NEXT staged record will occupy. * * Exact, and knowable at append time — no deferral to flush is needed, which diff --git a/runtime/test/test_wal.c b/runtime/test/test_wal.c index 074cb47..a3e6b0f 100644 --- a/runtime/test/test_wal.c +++ b/runtime/test/test_wal.c @@ -1008,6 +1008,143 @@ static void test_should_compact_absolute_and_ceiling(void) { T_EQ(wo_wal_should_compact(2048, 1024, floor_b, 2), 0); /* not yet */ } +/* ---- databasev2 12: the schema record ---------------------------------- */ + +/* a hand-built two-class schema exercising every payload field */ +static wo_schema mig_schema_sample(void) { + static wo_schema_field f0[] = { + {(const uint8_t *)"n", 1, WO_K_SCALAR, WO_SCHEMA_NONE, WO_SCHEMA_NONE}, + {(const uint8_t *)"label", 5, WO_K_TEXT, WO_SCHEMA_NONE, WO_SCHEMA_NONE}, + }; + static wo_schema_field f1[] = { + {(const uint8_t *)"part", 4, WO_K_OWNED, 0, WO_SCHEMA_NONE}, + {(const uint8_t *)"tags", 4, WO_K_MULTI, WO_SCHEMA_NONE, WO_K_TEXT}, + }; + static wo_schema_class cls[] = { + {(const uint8_t *)"row", 3, 0, 2, f0}, + {(const uint8_t *)"box", 3, WO_CLASSF_RESIDENT_KEYS, 2, f1}, + }; + wo_schema sc = {2, cls, NULL}; + return sc; +} + +static void test_schema_roundtrip(void) { + wo_schema sc = mig_schema_sample(); + uint8_t *p; + uint32_t len; + T_EQ(wo_schema_encode(&sc, &p, &len), 0); + T_CHECK(len > 5 && p[0] == WO_WAL_SCHEMA); + wo_schema *back = wo_schema_decode(p, len); + T_CHECK(back != NULL); + T_EQ(back->class_cnt, 2u); + T_CHECK(back->classes[0].name_len == 3 && memcmp(back->classes[0].name, "row", 3) == 0); + T_EQ(back->classes[0].field_cnt, 2u); + T_CHECK(back->classes[0].fields[1].kind == WO_K_TEXT && + back->classes[0].fields[1].name_len == 5 && + memcmp(back->classes[0].fields[1].name, "label", 5) == 0); + T_EQ(back->classes[1].flags, (uint32_t)WO_CLASSF_RESIDENT_KEYS); + T_CHECK(back->classes[1].fields[0].fclass == 0 && + back->classes[1].fields[1].felem == WO_K_TEXT); + /* the decode owns its bytes: the encode buffer can die first */ + free(p); + T_CHECK(memcmp(back->classes[1].name, "box", 3) == 0); + /* a truncated payload is malformed, not a crash */ + uint8_t *p2; + uint32_t len2; + T_EQ(wo_schema_encode(&sc, &p2, &len2), 0); + T_CHECK(wo_schema_decode(p2, len2 - 3) == NULL); + free(p2); + wo_schema_free(back); +} + +/* a fresh log opened with a schema carries it as its FIRST record; replay + * skips it without counting it, and rows behind it land intact */ +static void test_schema_fresh_log(void) { + char path[128]; + snprintf(path, sizeof path, "%s/schemafresh.wal", g_dir); + wo_db db; + T_EQ(wo_db_init(&db, CLASSES, 1, 0, 1), 0); + wo_wal w; + T_EQ(wo_wal_open(&w, path, 1 << 16), 0); + wo_schema sc = mig_schema_sample(); + T_EQ(wo_wal_set_schema(&w, &sc), 0); + T_EQ(wo_wal_ensure_schema(&w), 0); + /* head record is the schema */ + uint8_t *p; + uint32_t len; + T_EQ(wo_wal_read_schema(path, &p, &len), 0); + wo_schema *back = wo_schema_decode(p, len); + T_CHECK(back != NULL && back->class_cnt == 2); + wo_schema_free(back); + free(p); + /* a second ensure is a no-op: records exist now */ + uint64_t before = wo_wal_next_offset(&w); + T_EQ(wo_wal_ensure_schema(&w), 0); + T_EQ(wo_wal_next_offset(&w), before); + /* a row behind it replays; the schema record is not counted */ + const char *msg = ""; + uint64_t vals[2] = {7, 0}; + uint64_t id = wo_row_insert(&db, 0, vals, &msg, NULL); + T_CHECK(id != 0); + T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0); + T_EQ(wo_wal_commit(&w), 0); + wo_wal_close(&w); + wo_db_destroy(&db); + wo_db db2; + T_EQ(wo_db_init(&db2, CLASSES, 1, 0, 1), 0); + T_EQ(wo_wal_replay(path, &db2), 1); /* one row, not two records */ + db_row *r = wo_row_ptr(&db2, 0, id); + T_CHECK(r != NULL && r->slots[0] == 7); + wo_db_destroy(&db2); +} + +/* a LEGACY log (rows, no schema record) reports 1 from read_schema, and its + * first compaction with a schema set writes the record at the head */ +static void test_schema_compaction_adopts_legacy(void) { + char path[128]; + snprintf(path, sizeof path, "%s/schemalegacy.wal", g_dir); + wo_db db; + T_EQ(wo_db_init(&db, CLASSES, 1, 0, 1), 0); + wo_wal w; + T_EQ(wo_wal_open(&w, path, 1 << 16), 0); + const char *msg = ""; + uint64_t vals[2] = {1, 0}; + uint64_t id = wo_row_insert(&db, 0, vals, &msg, NULL); + T_CHECK(id != 0); + T_EQ(wo_wal_append_insert(&w, &db, 0, id), 0); + T_EQ(wo_wal_commit(&w), 0); + T_EQ(wo_wal_read_schema(path, NULL, NULL), 1); /* legacy: head is a row */ + wo_schema sc = mig_schema_sample(); + T_EQ(wo_wal_set_schema(&w, &sc), 0); + T_EQ(wo_wal_ensure_schema(&w), 0); /* no-op: not empty */ + T_EQ(wo_wal_read_schema(path, NULL, NULL), 1); + T_EQ(wo_wal_compact(&w, &db), 0); + uint8_t *p; + uint32_t len; + T_EQ(wo_wal_read_schema(path, &p, &len), 0); /* adopted at the head */ + free(p); + /* and the compacted log still replays its row */ + wo_wal_close(&w); + wo_db_destroy(&db); + wo_db db2; + T_EQ(wo_db_init(&db2, CLASSES, 1, 0, 1), 0); + T_EQ(wo_wal_replay(path, &db2), 1); + db_row *r = wo_row_ptr(&db2, 0, id); + T_CHECK(r != NULL && r->slots[0] == 1); + wo_db_destroy(&db2); +} + +/* missing and empty files are legacy, not errors */ +static void test_schema_read_absent(void) { + char path[128]; + snprintf(path, sizeof path, "%s/schemanone.wal", g_dir); + T_EQ(wo_wal_read_schema(path, NULL, NULL), 1); /* no file */ + FILE *f = fopen(path, "w"); + T_CHECK(f != NULL); + fclose(f); + T_EQ(wo_wal_read_schema(path, NULL, NULL), 1); /* empty file */ +} + static void test_delta_chain_flattens_at_k(void) { char path[128]; snprintf(path, sizeof path, "%s/chainflat.wal", g_dir); @@ -2673,6 +2810,10 @@ int main(void) { test_keys_resident_update_field(); test_keys_resident_update_indexed(); test_keys_resident_indexed_across_flatten(); + test_schema_roundtrip(); + test_schema_fresh_log(); + test_schema_compaction_adopts_legacy(); + test_schema_read_absent(); test_should_compact_absolute_and_ceiling(); test_delta_chain_flattens_at_k(); test_delta_chain_flatten_replays();