From 49c29a1255e1ad3f7fde6c64910c7d2c38edad78 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Fri, 25 Sep 2026 15:37:53 +0100 Subject: Commit several drafts of the tree store atomically pure_draft_commit_all publishes drafts of one store with a single head: every key must still be at its draft's base revision, all new extents and catalog records are written and synced once, and either every document becomes visible or none does. A single draft commit is the one-draft case. pure_state_open stores runtime-owned values (app states) under '.' plus a key. Document keys never start with '.', listings skip these values, and document operations refuse their names, so an app's state cannot collide with a document a client names. Tests kill a process at every commit stage of a three-draft batch and see all or none, inject each fault, stop a batch on one stale draft, and commit fifty keys in no order at once. --- prototypes/connections/state.h | 8 ++ prototypes/connections/store-tree.c | 132 +++++++++++++++++++++---------- prototypes/connections/test-tree-store.c | 132 ++++++++++++++++++++++++++++++- 3 files changed, 230 insertions(+), 42 deletions(-) diff --git a/prototypes/connections/state.h b/prototypes/connections/state.h index 461f71a..55ad257 100644 --- a/prototypes/connections/state.h +++ b/prototypes/connections/state.h @@ -17,6 +17,10 @@ typedef struct PureDraft PureDraft; PureDraft *pure_draft_new(void); /* A missing key opens as an empty draft at revision 0; commit creates it. */ PureResult pure_draft_open(const char *path, const char *key, PureDraft **out); +/* A runtime-owned value of an app, such as a list or text state, stored + * under a name that is never a document key (hosts listing documents do not + * see it). name is a valid key of at most 63 bytes. */ +PureResult pure_state_open(const char *path, const char *name, PureDraft **out); /* Memory only. Fails without change on a bad range or when memory runs out. */ PureResult pure_draft_splice(PureDraft *, size_t offset, size_t remove, PureSpan insert); size_t pure_draft_size(const PureDraft *); @@ -40,5 +44,9 @@ void pure_draft_restore(PureDraft *, PureVersion); /* Publish if the key is still at the draft's base revision. On success the * draft continues from the new revision; on any failure it is unchanged. */ PureResult pure_draft_commit(PureDraft *, int64_t *revision); +/* Publish drafts of one store in one commit: each key must still be at its + * draft's base revision (a conflict names the stored revision in revisions), + * and either every draft is published or none is. Keys must differ. */ +PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t *revisions); void pure_draft_close(PureDraft *); #endif diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c index 414c8ce..96bd65f 100644 --- a/prototypes/connections/store-tree.c +++ b/prototypes/connections/store-tree.c @@ -86,6 +86,12 @@ static bool valid_key(const char *key) { return true; } +/* Runtime-owned values (app states) live under '.' followed by a valid key. + * Document keys never start with '.', so the two never meet. */ +static bool runtime_key(const char *key) { + return key && key[0] == '.' && strlen(key) <= KEY && valid_key(key + 1); +} + static uint64_t decode64(const unsigned char *bytes) { uint64_t value = 0; for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i); @@ -839,28 +845,39 @@ static PureResult open_session(const char *path, Session *session) { return PURE_OK; } -/* The document's new nodes are already appended at writer. Record it in the - * catalog, then make the commit durable before its head becomes visible. */ -static PureResult publish(Session *session, Writer *writer, const char *key, uint64_t index, - bool found, Entry document, int64_t *revision) { - unsigned char record[RECORD] = {0}; - Entry catalog = {session->head.catalog, session->head.catalog_bytes}, updated; - memcpy(record, key, strlen(key)); - encode64(record + KEY, document.ref); - encode64(record + KEY + 8, document.bytes); - encode64(record + KEY + 16, (uint64_t)*revision + 1); - bool written = rebuild(&session->tree, catalog, index * RECORD, found ? RECORD : 0, - (PureSpan){record, RECORD}, &updated) && - write_root(&session->tree, writer, &updated, false) && flush(writer); +/* The documents' new nodes are already appended at writer. Record each in + * the catalog (revisions come in as the base and leave as the new one), + * then make the commit durable before its head becomes visible: every + * document becomes visible at once, or none does. */ +static PureResult publish(Session *session, Writer *writer, size_t count, const char *const *keys, + const Entry *documents, int64_t *revisions) { + Entry catalog = {session->head.catalog, session->head.catalog_bytes}; + bool written = true; + for (size_t i = 0; written && i < count; ++i) { + unsigned char record[RECORD] = {0}; + uint64_t index; + bool found; + Entry current, updated; + int64_t stored; + memcpy(record, keys[i], strlen(keys[i])); + encode64(record + KEY, documents[i].ref); + encode64(record + KEY + 8, documents[i].bytes); + encode64(record + KEY + 16, (uint64_t)revisions[i] + 1); + written = find(&session->tree, catalog, keys[i], &index, &found, ¤t, &stored) && + rebuild(&session->tree, catalog, index * RECORD, found ? RECORD : 0, + (PureSpan){record, RECORD}, &updated); + catalog = updated; + } + written = written && write_root(&session->tree, writer, &catalog, false) && flush(writer); if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || !sync_fd(session->fd) || fault(PURE_BEFORE_RENAME)) { if (ftruncate(session->fd, (off_t)session->head.end) == 0) sync_fd(session->fd); return PURE_STORE_ERROR; } - Head published = {session->head.seq + 1, writer->at, updated.ref, updated.bytes, session->head.live}; + Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live}; unsigned char slot[SLOT]; encode_head(slot, &published); - ++*revision; /* Past this point the head may already be visible. */ + for (size_t i = 0; i < count; ++i) ++revisions[i]; /* Past this point the head may already be visible. */ if (!write_at(session->fd, slot, SLOT, (published.seq % 2) * SLOT) || fault(PURE_BEFORE_DIR_SYNC) || !sync_fd(session->fd)) return PURE_UNCERTAIN; if (published.end - DATA > 4 * published.live + COMPACT_SLACK) @@ -898,14 +915,15 @@ static PureResult mutate(const char *path, const char *key, int64_t expected, if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); goto finish; } - result = publish(&session, writer, key, index, found, next, revision); + result = publish(&session, writer, 1, &key, &next, revision); finish: free(writer); close_session(&session); return result; } -/* Keys come from the catalog in ascending byte order. */ +/* Keys come from the catalog in ascending byte order. Runtime values are not + * documents and are left out. */ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *key), void *context) { int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); @@ -925,6 +943,7 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k for (uint64_t i = 0; i < catalog.bytes / RECORD; ++i) { if (!read_record(&tree, catalog, i, record, &document, &revision)) { result = PURE_STORE_ERROR; break; } memcpy(key, record, KEY); + if (runtime_key(key)) continue; if (!valid_key(key)) { result = PURE_STORE_ERROR; break; } if (!visit(context, key)) break; } @@ -977,7 +996,7 @@ static PureResult load_draft(PureDraft *draft) { PureResult pure_draft_open(const char *path, const char *key, PureDraft **out) { *out = NULL; - if (!path || !valid_key(key)) return PURE_INVALID; + if (!path || !(valid_key(key) || runtime_key(key))) return PURE_INVALID; PureDraft *draft = calloc(1, sizeof(*draft)); if (!draft || !(draft->path = strdup(path))) { free(draft); return PURE_ALLOCATION; } memcpy(draft->key, key, strlen(key) + 1); @@ -987,6 +1006,14 @@ PureResult pure_draft_open(const char *path, const char *key, PureDraft **out) { return PURE_OK; } +PureResult pure_state_open(const char *path, const char *name, PureDraft **out) { + char key[KEY + 2]; + *out = NULL; + if (!valid_key(name) || strlen(name) >= KEY) return PURE_INVALID; + snprintf(key, sizeof(key), ".%s", name); + return pure_draft_open(path, key, out); +} + PureDraft *pure_draft_new(void) { return calloc(1, sizeof(PureDraft)); /* no path: nothing mapped, nothing to commit */ } @@ -1079,41 +1106,66 @@ void pure_draft_hold(PureDraft *draft, bool held) { if (!held) collect(draft); } -PureResult pure_draft_commit(PureDraft *draft, int64_t *revision) { - *revision = draft->revision; - if (!draft->path) return PURE_INVALID; +PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t *revisions) { + for (size_t i = 0; i < count; ++i) { + revisions[i] = drafts[i]->revision; + if (!drafts[i]->path || strcmp(drafts[i]->path, drafts[0]->path)) return PURE_INVALID; + for (size_t k = 0; k < i; ++k) if (!strcmp(drafts[k]->key, drafts[i]->key)) return PURE_INVALID; + } + if (!count) return PURE_OK; Session session; Writer *writer = NULL; - Entry current, root = draft->root; - uint64_t index; - int64_t stored; - bool found; - PureResult result = open_session(draft->path, &session); - if (result != PURE_OK) goto finish; + Entry *roots = malloc(count * sizeof(*roots)); + const char **keys = malloc(count * sizeof(*keys)); + PureResult result = roots && keys ? open_session(drafts[0]->path, &session) : PURE_ALLOCATION; + if (result != PURE_OK) { + if (roots && keys) close_session(&session); + free(roots); + free((void *)keys); + return result; + } result = PURE_STORE_ERROR; - if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, draft->key, - &index, &found, ¤t, &stored)) goto finish; - if (stored != draft->revision) { *revision = stored; result = PURE_CONFLICT; goto finish; } - if (stored == INT64_MAX) { result = PURE_LIMIT; goto finish; } - /* After compaction the draft's stored references name the old file. */ - bool relocate = session.file.st_dev != draft->device || session.file.st_ino != draft->inode; + /* Every key must still be at its draft's base revision. */ + for (size_t i = 0; i < count; ++i) { + Entry current; + uint64_t index; + int64_t stored; + bool found; + if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, drafts[i]->key, + &index, &found, ¤t, &stored)) goto finish; + if (stored != drafts[i]->revision) { revisions[i] = stored; result = PURE_CONFLICT; goto finish; } + if (stored == INT64_MAX) { result = PURE_LIMIT; goto finish; } + roots[i] = drafts[i]->root; + keys[i] = drafts[i]->key; + } writer = malloc(sizeof(*writer)); if (!writer) goto finish; *writer = (Writer){.fd = session.fd, .at = session.head.end}; - if (!write_root(&draft->tree, writer, &root, relocate)) { - if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); - goto finish; + for (size_t i = 0; i < count; ++i) { + /* After compaction a draft's stored references name the old file. */ + bool relocate = session.file.st_dev != drafts[i]->device || session.file.st_ino != drafts[i]->inode; + if (!write_root(&drafts[i]->tree, writer, &roots[i], relocate)) { + if (ftruncate(session.fd, (off_t)session.head.end) == 0) sync_fd(session.fd); + goto finish; + } } - result = publish(&session, writer, draft->key, index, found, root, revision); - /* Rebase while still holding the writer lock; the arena is no longer needed. - * If remapping fails the draft keeps equal content at the new revision. */ - if (result == PURE_OK && load_draft(draft) != PURE_OK) draft->revision = *revision; + result = publish(&session, writer, count, keys, roots, revisions); + /* Rebase while still holding the writer lock; the arenas are no longer + * needed. If remapping fails a draft keeps equal content at its new revision. */ + for (size_t i = 0; result == PURE_OK && i < count; ++i) + if (load_draft(drafts[i]) != PURE_OK) drafts[i]->revision = revisions[i]; finish: free(writer); + free(roots); + free((void *)keys); close_session(&session); return result; } +PureResult pure_draft_commit(PureDraft *draft, int64_t *revision) { + return pure_draft_commit_all(&draft, 1, revision); +} + PureResult pure_store_put(const char *path, const char *key, int64_t expected, PureSpan body, int64_t *revision) { return mutate(path, key, expected, body, false, 0, 0, PURE_OBJECT_LIMIT, revision); diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c index 971abba..f6a4b1b 100644 --- a/prototypes/connections/test-tree-store.c +++ b/prototypes/connections/test-tree-store.c @@ -638,6 +638,133 @@ static void draft_faults(const char *path) { CHECK(pure_draft_open(path, "bad/key", &draft) == PURE_INVALID && !draft); } +static bool list_key(void *context, const char *key) { + char *seen = context; + strcat(seen, key); + strcat(seen, ","); + return true; +} + +/* Several drafts publish in one commit: all or none, under process crashes, + * faults and a stale draft. Runtime values stay out of the document keys. */ +static void batch(const char *path) { + unsigned char scratch[64]; + int64_t revision, revisions[3]; + CHECK(pure_store_put(path, "alpha", 0, (PureSpan){(const unsigned char *)"a0", 2}, &revision) == PURE_OK); + CHECK(pure_store_put(path, "beta", 0, (PureSpan){(const unsigned char *)"b0", 2}, &revision) == PURE_OK); + PureDraft *drafts[3]; + CHECK(pure_draft_open(path, "alpha", &drafts[0]) == PURE_OK && pure_draft_open(path, "beta", &drafts[1]) == PURE_OK); + CHECK(pure_state_open(path, "values", &drafts[2]) == PURE_OK && pure_draft_revision(drafts[2]) == 0); + PureDraft *bad; + CHECK(pure_state_open(path, "bad.name", &bad) == PURE_INVALID && !bad); + CHECK(pure_draft_splice(drafts[0], 2, 0, (PureSpan){(const unsigned char *)"+1", 2}) == PURE_OK); + CHECK(pure_draft_splice(drafts[1], 0, 2, (PureSpan){(const unsigned char *)"B", 1}) == PURE_OK); + CHECK(pure_draft_splice(drafts[2], 0, 0, (PureSpan){(const unsigned char *)"state", 5}) == PURE_OK); + Oracle alpha = {(unsigned char *)"a0", 2, 1}, beta = {(unsigned char *)"b0", 2, 1}; + Oracle alpha_new = {(unsigned char *)"a0+1", 4, 1}, beta_new = {(unsigned char *)"B", 1, 1}; + Oracle state_new = {(unsigned char *)"state", 5, 0}; + /* A process that dies anywhere in the commit leaves all three as before, + * or all three published once the head is written. */ + for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) { + pid_t child = fork(); + CHECK(child >= 0); + if (!child) { + crash = 1; + fail_stage = stage; + pure_draft_commit_all(drafts, 3, revisions); + _exit(0); + } + int status; + CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77); + PureDraft *state; + CHECK(pure_state_open(path, "values", &state) == PURE_OK); + if (stage < PURE_BEFORE_DIR_SYNC) { + expect(path, "alpha", &alpha); + expect(path, "beta", &beta); + CHECK(pure_draft_revision(state) == 0 && pure_draft_size(state) == 0); + pure_draft_close(state); + continue; + } + Oracle a = alpha_new, b = beta_new, v = state_new; + a.revision = b.revision = 2; + v.revision = 1; + expect(path, "alpha", &a); + expect(path, "beta", &b); + expect_draft(state, &v, scratch); + pure_draft_close(state); + /* Undo the crashed commit's effect for the next checks. */ + CHECK(pure_store_put(path, "alpha", 2, (PureSpan){(const unsigned char *)"a0", 2}, &revision) == PURE_OK); + } + /* The same batch in this process is now stale on every key. */ + CHECK(pure_draft_commit_all(drafts, 3, revisions) == PURE_CONFLICT && revisions[0] == 3); + for (unsigned i = 0; i < 3; ++i) pure_draft_close(drafts[i]); + CHECK(pure_draft_open(path, "alpha", &drafts[0]) == PURE_OK && pure_draft_open(path, "beta", &drafts[1]) == PURE_OK); + CHECK(pure_state_open(path, "values", &drafts[2]) == PURE_OK); + CHECK(pure_draft_splice(drafts[0], 0, 2, (PureSpan){(const unsigned char *)"A", 1}) == PURE_OK); + CHECK(pure_draft_splice(drafts[1], 1, 0, (PureSpan){(const unsigned char *)"!", 1}) == PURE_OK); + CHECK(pure_draft_splice(drafts[2], 5, 0, (PureSpan){(const unsigned char *)"+", 1}) == PURE_OK); + alpha = (Oracle){(unsigned char *)"a0", 2, 3}; + beta = (Oracle){(unsigned char *)"B", 1, 2}; + uint64_t size = file_size(path); + for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; ++fail_stage) { + CHECK(pure_draft_commit_all(drafts, 3, revisions) == PURE_STORE_ERROR); + expect(path, "alpha", &alpha); + expect(path, "beta", &beta); + CHECK(file_size(path) == size); + } + fail_stage = -1; + /* One stale draft stops the whole batch. */ + PureDraft *other; + CHECK(pure_draft_open(path, "beta", &other) == PURE_OK); + CHECK(pure_draft_splice(other, 0, 0, (PureSpan){(const unsigned char *)"x", 1}) == PURE_OK); + CHECK(pure_draft_commit(other, &revision) == PURE_OK && revision == 3); + pure_draft_close(other); + CHECK(pure_draft_commit_all(drafts, 3, revisions) == PURE_CONFLICT && revisions[1] == 3); + expect(path, "alpha", &alpha); + pure_draft_close(drafts[1]); + CHECK(pure_draft_open(path, "beta", &drafts[1]) == PURE_OK); + CHECK(pure_draft_splice(drafts[1], 0, 0, (PureSpan){(const unsigned char *)"y", 1}) == PURE_OK); + CHECK(pure_draft_commit_all(drafts, 3, revisions) == PURE_OK); + CHECK(revisions[0] == 4 && revisions[1] == 4 && revisions[2] == 2); + expect(path, "alpha", &(Oracle){(unsigned char *)"A", 1, 4}); + expect(path, "beta", &(Oracle){(unsigned char *)"yxB", 3, 4}); + expect_draft(drafts[2], &(Oracle){(unsigned char *)"state+", 6, 2}, scratch); + /* Drafts continue from the new revisions. */ + CHECK(pure_draft_splice(drafts[0], 1, 0, (PureSpan){(const unsigned char *)"z", 1}) == PURE_OK); + CHECK(pure_draft_commit_all(drafts, 1, revisions) == PURE_OK && revisions[0] == 5); + /* Invalid batches change nothing. */ + PureDraft *twice[2] = {drafts[0], drafts[0]}, *anonymous = pure_draft_new(); + CHECK(pure_draft_commit_all(twice, 2, revisions) == PURE_INVALID); + PureDraft *mixed[2] = {drafts[1], anonymous}; + CHECK(pure_draft_commit_all(mixed, 2, revisions) == PURE_INVALID); + pure_draft_close(anonymous); + for (unsigned i = 0; i < 3; ++i) pure_draft_close(drafts[i]); + /* Runtime values are not documents: listings skip them and document + * operations refuse their names. */ + char seen[256] = ""; + CHECK(pure_store_keys(path, list_key, seen) == PURE_OK && !strcmp(seen, "alpha,beta,")); + CHECK(pure_store_put(path, ".values", 0, (PureSpan){(const unsigned char *)"x", 1}, &revision) != PURE_OK); + /* Many keys, in no particular order, in one commit. */ + PureDraft *many[50]; + int64_t many_revisions[50]; + char key[16]; + for (unsigned i = 0; i < 50; ++i) { + snprintf(key, sizeof(key), "k%02u", (i * 37) % 50); + CHECK((i % 2 ? pure_state_open(path, key, &many[i]) : pure_draft_open(path, key, &many[i])) == PURE_OK); + CHECK(pure_draft_splice(many[i], 0, 0, (PureSpan){(const unsigned char *)key, strlen(key)}) == PURE_OK); + } + CHECK(pure_draft_commit_all(many, 50, many_revisions) == PURE_OK); + for (unsigned i = 0; i < 50; ++i) { + CHECK(many_revisions[i] == 1); + snprintf(key, sizeof(key), "k%02u", (i * 37) % 50); + PureDraft *check; + CHECK((i % 2 ? pure_state_open(path, key, &check) : pure_draft_open(path, key, &check)) == PURE_OK); + expect_draft(check, &(Oracle){(unsigned char *)key, strlen(key), 1}, scratch); + pure_draft_close(check); + pure_draft_close(many[i]); + } +} + static void clean(const char *path) { DIR *entries = opendir(path); CHECK(entries); @@ -654,7 +781,7 @@ int main(void) { void (*checks[])(const char *) = {oracle, amplification, faults, crashes, torn_head, isolation, corruption, many_keys, revision_limit, keys, drafts, typing, draft_conflict, draft_relocation, draft_faults, - draft_versions, anonymous_draft}; + draft_versions, anonymous_draft, batch}; for (size_t i = 0; i < sizeof(checks) / sizeof(*checks); ++i) { char path[] = "/tmp/pure-tree-check-XXXXXX"; CHECK(mkdtemp(path) && pure_store_init(path) == PURE_OK); @@ -664,6 +791,7 @@ int main(void) { } puts("{\"tree_store\":\"random-oracle-balance-held-views-one-byte-amplification-faults-" "process-crash-torn-head-lockless-isolation-compaction-corruption-1024-keys-revision-limit-key-listing-" - "draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous\"}"); + "draft-oracle-typing-arena-conflict-relocation-commit-faults-held-versions-anonymous-" + "atomic-batch-runtime-keys\"}"); return 0; } -- cgit v1.2.3