shoney/writeonce
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

feat(db2-migrate): WO_WAL_SCHEMA — the log states the shape that wrote it

Browse source 
- 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) <noreply@anthropic.com>
(cherry picked from commit ba8519fa5e39c6ed6a3504d39e5a372f8decd045)
This commit is contained in:
shoney.arickathil 2026-08-31 21:21:23 +02:00
parent 1b6d633ed1
commit f07295b3c0
3 changed files with 357 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

146
database/src/wal.c
View file

@ -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);

72
database/src/wal.h
View file

@ -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

141
runtime/test/test_wal.c
View file

@ -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();