Add a store handle that shares one mapping

Opening a draft by path opened the directory and the file, mapped the
whole file, read the head, and undid all of it on close: nine system
calls for every GET. pure_store_open keeps the store open for one owner
thread; drafts opened through it share a reference-counted mapping of the
newest version, and opening one only checks whether the file changed (its
identity and its head: two system calls). A commit of its drafts maps the
new version once and moves them all onto it. A compaction by another
writer is noticed and followed, while drafts of the old file keep reading
it and relocate their extents when they commit.

pure_store_data gives modules the same through a handle.

5 files changed, 233 insertions, 8 deletions

diff --git a/prototypes/connections/module-store.c b/prototypes/connections/module-store.c
index 02d6361..7e759fd 100644
--- a/prototypes/connections/module-store.c
+++ b/prototypes/connections/module-store.c
@@ -58,6 +58,29 @@ static PureResult save(void *context, void *const *documents, size_t document_co
     return result;
 }
 
+/* Through a store handle: documents and states share its mapping. */
+static PureResult open_handle_document(void *context, uint32_t collection, const char *key, void **document,
+                                       int64_t *revision) {
+    (void)collection;
+    PureDraft *draft;
+    PureResult result = pure_store_draft(context, key, &draft);
+    if (result != PURE_OK) return result;
+    pure_draft_hold(draft, true);
+    *document = draft;
+    *revision = pure_draft_revision(draft);
+    return PURE_OK;
+}
+
+static PureResult handle_state(void *context, const char *name, PureDraft **out) {
+    return pure_store_state(context, name, out);
+}
+
+PureData pure_store_data(PureStore *store) {
+    return (PureData){.context = store, .open = open_handle_document, .version = version, .read = read,
+                      .splice = splice, .commit = commit, .close = close_document, .state = handle_state,
+                      .save = save};
+}
+
 PureData pure_tree_data(const char *path) {
     return (PureData){.context = (void *)path, .open = open_document, .version = version,
                       .read = read, .splice = splice, .commit = commit, .close = close_document,
diff --git a/prototypes/connections/module-store.h b/prototypes/connections/module-store.h
index 0a4b494..d14ab3d 100644
--- a/prototypes/connections/module-store.h
+++ b/prototypes/connections/module-store.h
@@ -5,4 +5,7 @@
  * one event opens the message key as a held draft and commits it together
  * with the states the event changed. The path outlives the app. */
 PureData pure_tree_data(const char *path);
+/* The same through a store handle (state.h): opening a document or state
+ * checks only whether the store changed. The handle outlives the app. */
+PureData pure_store_data(PureStore *);
 #endif
diff --git a/prototypes/connections/state.h b/prototypes/connections/state.h
index 55ad257..25c0f61 100644
--- a/prototypes/connections/state.h
+++ b/prototypes/connections/state.h
@@ -21,6 +21,15 @@ PureResult pure_draft_open(const char *path, const char *key, PureDraft **out);
  * 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);
+/* A store kept open by one owner thread. Drafts opened through it share one
+ * mapping of the newest version, and opening one checks only whether the file
+ * changed; commits of its drafts map the new version once. It must outlive
+ * its drafts. */
+typedef struct PureStore PureStore;
+PureResult pure_store_open(const char *path, PureStore **out);
+void pure_store_close(PureStore *);
+PureResult pure_store_draft(PureStore *, const char *key, PureDraft **out);
+PureResult pure_store_state(PureStore *, 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 *);
diff --git a/prototypes/connections/store-tree.c b/prototypes/connections/store-tree.c
index 96bd65f..c41393b 100644
--- a/prototypes/connections/store-tree.c
+++ b/prototypes/connections/store-tree.c
@@ -48,9 +48,12 @@ typedef struct {
     size_t size;          /* leaf bytes, child count, or next free slot + 1 */
     unsigned height;      /* FREE for a reclaimed slot */
 } Node;
+/* One mapped version of the file, shared by the drafts of a store handle. */
+typedef struct { const unsigned char *map; uint64_t end; unsigned refs; } Mapping;
 typedef struct {
     const unsigned char *map;
     uint64_t end;
+    Mapping *shared; /* map belongs to this shared mapping, not to the tree */
     Node *nodes;
     size_t count, capacity, free_head, free_count; /* free_head is slot + 1 */
     size_t bytes; /* payload held by arena nodes, including unreachable ones */
@@ -182,8 +185,15 @@ static bool map_tree(int fd, const Head *head, Tree *tree) {
     return true;
 }
 
+static void release_mapping(Mapping *mapping) {
+    if (!mapping || --mapping->refs) return;
+    munmap((void *)mapping->map, (size_t)mapping->end);
+    free(mapping);
+}
+
 static void release_tree(Tree *tree) {
-    if (tree->map) munmap((void *)tree->map, (size_t)tree->end);
+    if (tree->shared) release_mapping(tree->shared);
+    else if (tree->map) munmap((void *)tree->map, (size_t)tree->end);
     for (size_t i = 0; i < tree->count; ++i) {
         free(tree->nodes[i].bytes);
         free(tree->nodes[i].children);
@@ -953,8 +963,20 @@ PureResult pure_store_keys(const char *path, bool (*visit)(void *, const char *k
     return result;
 }
 
-struct PureDraft {
+/* A store kept open by one owner thread: its current file and a mapping of
+ * its newest version, shared by the drafts opened through it. */
+struct PureStore {
     char *path;
+    int directory, fd;
+    dev_t device;
+    ino_t inode;
+    Head head;
+    Mapping *mapping;
+};
+
+struct PureDraft {
+    PureStore *store; /* the handle it was opened through, or NULL */
+    char *path;       /* owned unless opened through a store handle */
     char key[KEY + 1];
     Tree tree;        /* the base file's mapping plus this draft's node arena */
     dev_t device;     /* identity of the base file; compaction replaces it */
@@ -1021,10 +1043,110 @@ PureDraft *pure_draft_new(void) {
 void pure_draft_close(PureDraft *draft) {
     if (!draft) return;
     release_tree(&draft->tree);
-    free(draft->path);
+    if (!draft->store) free(draft->path);
     free(draft);
 }
 
+/* Bring the handle to the newest published version: the same file and head
+ * cost two system calls; a new commit maps again, a compaction reopens. */
+static PureResult refresh(PureStore *store) {
+    struct stat info;
+    if (fstatat(store->directory, tree_name, &info, AT_SYMLINK_NOFOLLOW) || !S_ISREG(info.st_mode))
+        return PURE_STORE_ERROR;
+    if (store->fd < 0 || info.st_dev != store->device || info.st_ino != store->inode) {
+        int fd = openat(store->directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+        if (fd < 0 || fstat(fd, &info)) { if (fd >= 0) close(fd); return PURE_STORE_ERROR; }
+        if (store->fd >= 0) close(store->fd);
+        store->fd = fd;
+        store->device = info.st_dev;
+        store->inode = info.st_ino;
+        store->head = (Head){0};
+    }
+    Head head;
+    if (!read_head(store->fd, &head)) return PURE_STORE_ERROR;
+    if (store->mapping && head.seq == store->head.seq && head.end == store->head.end) return PURE_OK;
+    Tree tree;
+    Mapping *mapping = malloc(sizeof(*mapping));
+    if (!mapping) return PURE_ALLOCATION;
+    if (!map_tree(store->fd, &head, &tree)) { free(mapping); return PURE_STORE_ERROR; }
+    *mapping = (Mapping){tree.map, tree.end, 1};
+    release_mapping(store->mapping);
+    store->mapping = mapping;
+    store->head = head;
+    return PURE_OK;
+}
+
+/* Base a draft on the handle's newest version of its key, sharing the mapping.
+ * On failure the draft is unchanged. */
+static PureResult rebase(PureDraft *draft) {
+    PureStore *store = draft->store;
+    Mapping *mapping = store->mapping;
+    Tree tree = {.map = mapping->map, .end = mapping->end, .shared = mapping};
+    Entry document;
+    uint64_t index;
+    int64_t revision;
+    bool found;
+    if (!find(&tree, (Entry){store->head.catalog, store->head.catalog_bytes}, draft->key, &index, &found, &document,
+              &revision)) return PURE_STORE_ERROR;
+    ++mapping->refs;
+    release_tree(&draft->tree);
+    draft->tree = tree;
+    draft->device = store->device;
+    draft->inode = store->inode;
+    draft->root = document;
+    draft->revision = revision;
+    draft->live = 0;
+    return PURE_OK;
+}
+
+PureResult pure_store_open(const char *path, PureStore **out) {
+    *out = NULL;
+    PureStore *store = calloc(1, sizeof(*store));
+    if (!store || !path || !(store->path = strdup(path))) { free(store); return PURE_ALLOCATION; }
+    store->fd = -1;
+    store->directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    PureResult result = store->directory < 0 ? PURE_STORE_ERROR : refresh(store);
+    if (result != PURE_OK) { pure_store_close(store); return result; }
+    *out = store;
+    return PURE_OK;
+}
+
+void pure_store_close(PureStore *store) {
+    if (!store) return;
+    release_mapping(store->mapping);
+    if (store->fd >= 0) close(store->fd);
+    if (store->directory >= 0) close(store->directory);
+    free(store->path);
+    free(store);
+}
+
+static PureResult open_through(PureStore *store, const char *key, PureDraft **out) {
+    *out = NULL;
+    PureResult result = refresh(store);
+    if (result != PURE_OK) return result;
+    PureDraft *draft = calloc(1, sizeof(*draft));
+    if (!draft) return PURE_ALLOCATION;
+    *draft = (PureDraft){.store = store, .path = store->path};
+    memcpy(draft->key, key, strlen(key) + 1);
+    result = rebase(draft);
+    if (result != PURE_OK) { free(draft); return result; }
+    *out = draft;
+    return PURE_OK;
+}
+
+PureResult pure_store_draft(PureStore *store, const char *key, PureDraft **out) {
+    *out = NULL;
+    return store && valid_key(key) ? open_through(store, key, out) : PURE_INVALID;
+}
+
+PureResult pure_store_state(PureStore *store, const char *name, PureDraft **out) {
+    char key[KEY + 2];
+    *out = NULL;
+    if (!store || !valid_key(name) || strlen(name) >= KEY) return PURE_INVALID;
+    snprintf(key, sizeof(key), ".%s", name);
+    return open_through(store, key, out);
+}
+
 size_t pure_draft_size(const PureDraft *draft) { return (size_t)draft->root.bytes; }
 
 int64_t pure_draft_revision(const PureDraft *draft) { return draft->revision; }
@@ -1151,9 +1273,18 @@ PureResult pure_draft_commit_all(PureDraft *const *drafts, size_t count, int64_t
     }
     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];
+     * needed. Drafts of a store handle share one new mapping. If remapping
+     * fails a draft keeps equal content at its new revision. */
+    PureStore *refreshed = NULL;
+    for (size_t i = 0; result == PURE_OK && i < count; ++i) {
+        PureStore *store = drafts[i]->store;
+        bool rebased;
+        if (store) {
+            if (store != refreshed && refresh(store) == PURE_OK) refreshed = store;
+            rebased = store == refreshed && rebase(drafts[i]) == PURE_OK;
+        } else rebased = load_draft(drafts[i]) == PURE_OK;
+        if (!rebased) drafts[i]->revision = revisions[i];
+    }
 finish:
     free(writer);
     free(roots);
diff --git a/prototypes/connections/test-tree-store.c b/prototypes/connections/test-tree-store.c
index f6a4b1b..3f6d2f0 100644
--- a/prototypes/connections/test-tree-store.c
+++ b/prototypes/connections/test-tree-store.c
@@ -765,6 +765,65 @@ static void batch(const char *path) {
     }
 }
 
+/* A store handle shares one mapping among its drafts, sees commits made by
+ * other writers when a draft opens, follows a compaction that replaced the
+ * file, and maps a batch it committed once. */
+static void store_handle(const char *path) {
+    unsigned char scratch[64];
+    int64_t revision, revisions[3];
+    CHECK(pure_store_put(path, "shared", 0, (PureSpan){(const unsigned char *)"one", 3}, &revision) == PURE_OK);
+    PureStore *store;
+    CHECK(pure_store_open(path, &store) == PURE_OK);
+    PureDraft *a, *b, *state, *bad;
+    CHECK(pure_store_draft(store, "shared", &a) == PURE_OK);
+    expect_draft(a, &(Oracle){(unsigned char *)"one", 3, 1}, scratch);
+    CHECK(pure_store_draft(store, "bad.key", &bad) == PURE_INVALID && !bad);
+    /* Another writer commits: the next draft sees it, the open one keeps its version. */
+    CHECK(pure_store_put(path, "shared", 1, (PureSpan){(const unsigned char *)"two", 3}, &revision) == PURE_OK);
+    CHECK(pure_store_draft(store, "shared", &b) == PURE_OK);
+    expect_draft(b, &(Oracle){(unsigned char *)"two", 3, 2}, scratch);
+    expect_draft(a, &(Oracle){(unsigned char *)"one", 3, 1}, scratch);
+    CHECK(pure_draft_splice(a, 3, 0, (PureSpan){(const unsigned char *)"!", 1}) == PURE_OK);
+    CHECK(pure_draft_commit(a, &revision) == PURE_CONFLICT && revision == 2);
+    pure_draft_close(a);
+    /* A batch of handle drafts, a runtime value among them. */
+    CHECK(pure_store_state(store, "values", &state) == PURE_OK && pure_draft_revision(state) == 0);
+    CHECK(pure_store_draft(store, "other", &a) == PURE_OK);
+    CHECK(pure_draft_splice(b, 3, 0, (PureSpan){(const unsigned char *)"+", 1}) == PURE_OK);
+    CHECK(pure_draft_splice(state, 0, 0, (PureSpan){(const unsigned char *)"v", 1}) == PURE_OK);
+    CHECK(pure_draft_splice(a, 0, 0, (PureSpan){(const unsigned char *)"o", 1}) == PURE_OK);
+    PureDraft *batch[3] = {b, state, a};
+    CHECK(pure_draft_commit_all(batch, 3, revisions) == PURE_OK && revisions[0] == 3 && revisions[1] == 1);
+    expect(path, "shared", &(Oracle){(unsigned char *)"two+", 4, 3});
+    expect_draft(state, &(Oracle){(unsigned char *)"v", 1, 1}, scratch);
+    /* Other writers grow the file past compaction: it is replaced under the handle. */
+    unsigned char *big = malloc(100000);
+    uint32_t seed = 5;
+    CHECK(big);
+    fill(big, 100000, &seed);
+    int64_t filler = 0;
+    char name[512];
+    struct stat before, replaced;
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    CHECK(stat(name, &before) == 0);
+    for (int i = 0; i < 8; ++i) CHECK(pure_store_put(path, "filler", filler, (PureSpan){big, 100000}, &filler) == PURE_OK);
+    CHECK(stat(name, &replaced) == 0 && replaced.st_ino != before.st_ino);
+    free(big);
+    PureDraft *after;
+    CHECK(pure_store_draft(store, "shared", &after) == PURE_OK);
+    expect_draft(after, &(Oracle){(unsigned char *)"two+", 4, 3}, scratch);
+    /* Drafts of the old file still read and commit (their extents are copied). */
+    expect_draft(b, &(Oracle){(unsigned char *)"two+", 4, 3}, scratch);
+    CHECK(pure_draft_splice(b, 0, 0, (PureSpan){(const unsigned char *)">", 1}) == PURE_OK);
+    CHECK(pure_draft_commit(b, &revision) == PURE_OK && revision == 4);
+    expect(path, "shared", &(Oracle){(unsigned char *)">two+", 5, 4});
+    pure_draft_close(after);
+    pure_draft_close(a);
+    pure_draft_close(b);
+    pure_draft_close(state);
+    pure_store_close(store);
+}
+
 static void clean(const char *path) {
     DIR *entries = opendir(path);
     CHECK(entries);
@@ -781,7 +840,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, batch};
+                                      draft_versions, anonymous_draft, batch, store_handle};
     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);
@@ -792,6 +851,6 @@ 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-"
-         "atomic-batch-runtime-keys\"}");
+         "atomic-batch-runtime-keys-store-handle\"}");
     return 0;
 }