aboutsummaryrefslogtreecommitdiffstats
path: root/prototypes/connections
diff options
context:
space:
mode:
Diffstat (limited to 'prototypes/connections')
-rw-r--r--prototypes/connections/state.h8
-rw-r--r--prototypes/connections/store-tree.c132
-rw-r--r--prototypes/connections/test-tree-store.c132
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, &current, &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, &current, &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, &current, &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;
}
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 13:55:37 +0000