Lay Pure out as a core with the parts selected around it

The app model, the tree store and the HTTP host lived in
prototypes/connections among the earlier document runtime, the store was
known only by its format's name, and Vega compiled six Pure sources by path.
Now, as the README's principle has it:

  core/   runtime, flow and the app model (module*.c), with what the core
          asks of storage (state.h, store.h)
  store/  the tree store (PURETRE3, PUREREC1)
  serve/  pure-serve: HTTP host, outbound delivery, JSON in place
  cli/    pure-run, pure-edit-module

A top-level make builds libpure.a (app model and store, for hosts to link)
and the hosts; make check runs the store's and the app model's tests. The
earlier runtime still builds from prototypes/connections against core/ and
store/ (check, check-tree and check-core pass). Design 009 records the rule:
parts by role behind libraries; git repositories split by coupling, so the
store stays here while its contract moves with the core, and Vega stays apart.

2 files changed, 2987 insertions, 0 deletions

diff --git a/store/store-tree.c b/store/store-tree.c
new file mode 100644
index 0000000..146e919
--- /dev/null
+++ b/store/store-tree.c
@@ -0,0 +1,1856 @@
+/* Selected storage experiment: every document lives in one append-only file as
+ * a persistent B-tree of byte extents. An edit writes only the leaves and the
+ * path it changes, so inserting or deleting in a large document costs O(log n)
+ * nodes instead of a whole-document copy. Unchanged extents are shared by old
+ * and new versions. The document catalog is the same structure holding fixed
+ * records sorted by key. A commit appends a record (its head: where its nodes
+ * are and their digest) and then its nodes, and syncs once: one contiguous
+ * range, which a flash device writes about twice as fast as two. Every few
+ * commits the head also goes into the older of two slots at the start, where
+ * a reader begins before following the records after it; readers take no
+ * lock and map one immutable prefix. */
+#define _POSIX_C_SOURCE 200809L
+#ifdef __APPLE__
+#define _DARWIN_C_SOURCE
+#endif
+#include "store.h"
+#include "state.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <inttypes.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <pthread.h>
+#include <sys/mman.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+enum { FREE = 0xffff };
+/* Each head slot has a block of its own: a torn write of one never touches
+ * the other. */
+enum { SLOT = 128, BLOCK = 4096, DATA = 2 * BLOCK, LEAF_MAX = 4096, LEAF_MIN = 1024,
+       FAN_MAX = 32, FAN_MIN = 8, RECORD = 96, KEY = 64, HEIGHT_MAX = 12,
+       BUFFER = 65536 };
+/* Written zeros kept after the published end (see prepare_tail). */
+enum { TAIL_MIN = 65536, TAIL_STEP = 262144 };
+#ifdef PURE_STORE_TEST
+enum { COMPACT_SLACK = 65536 };
+#else
+enum { COMPACT_SLACK = 1048576 };
+#endif
+#define LEAF_BIT (UINT64_C(1) << 63)
+#define MEMORY_BIT (UINT64_C(1) << 62)
+#define LOCATION (MEMORY_BIT - 1)
+/* Slots of this format, and of the one before tail records (still read). */
+static const unsigned char head_magic[8] = {'P','U','R','E','T','R','E','3'};
+static const unsigned char earlier_magic[8] = {'P','U','R','E','T','R','E','2'};
+static const unsigned char record_magic[8] = {'P','U','R','E','R','E','C','1'};
+/* Commits between two heads written to a slot. */
+enum { CHECKPOINT = 16 };
+static const unsigned char branch_magic[4] = {'P','T','B','1'};
+static const char tree_name[] = "pure.tree";
+/* Threads of one process take turns as the writer: the file lock arbitrates
+ * only between processes (it is the process's, and closing any descriptor of
+ * the lock file drops it). */
+static pthread_mutex_t writers = PTHREAD_MUTEX_INITIALIZER;
+static uint64_t temporary_sequence;
+
+/* A reference is a file offset, or an index into this edit's node arena.
+ * bytes is the size of the whole subtree. Leaves hold raw document bytes. */
+typedef struct { uint64_t ref, bytes; } Entry;
+typedef struct { Entry *items; size_t count, capacity; } List;
+typedef struct {
+    unsigned char *bytes; /* leaf only */
+    Entry *children;      /* branch only */
+    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
+ * and by whoever still sends extents from it. */
+typedef struct PureMapping { 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 */
+} Tree;
+/* A commit's bytes are start..end (none for a file synced whole before it was
+ * published); check is their digest. */
+typedef struct { uint64_t seq, end, catalog, catalog_bytes, live, start, check; } Head;
+/* The xxHash64 construction without a seed: four multiply-rotate lanes over
+ * 32-byte stripes, several GB/s. An all-zero Digest is an empty one. */
+typedef struct { uint64_t lane[4], length; unsigned char held[32]; size_t count; } Digest;
+typedef struct {
+    int fd;
+    uint64_t at; /* file offset of the next byte */
+    size_t used;
+    bool failed;
+    Digest digest; /* of every byte written */
+    unsigned char buffer[BUFFER];
+} Writer;
+typedef struct { PureSpan part[3]; size_t index, offset; } Source;
+
+static bool fault(PureStoreStage stage) {
+#ifdef PURE_STORE_TEST
+    if (pure_store_test_fault(stage)) { errno = EIO; return true; }
+#else
+    (void)stage;
+#endif
+    return false;
+}
+
+static bool valid_key(const char *key) {
+    if (!key) return false;
+    size_t length = strlen(key);
+    if (!length || length > KEY) return false;
+    for (size_t i = 0; i < length; ++i) {
+        char b = key[i];
+        if (!((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') ||
+              (b >= '0' && b <= '9') || b == '-' || b == '_')) return false;
+    }
+    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);
+    return value;
+}
+
+static void encode64(unsigned char *bytes, uint64_t value) {
+    for (unsigned i = 0; i < 8; ++i) bytes[i] = (unsigned char)(value >> (8 * i));
+}
+
+static uint64_t load64(const unsigned char *bytes) {
+    uint64_t value;
+    memcpy(&value, bytes, 8);
+#if defined(__BYTE_ORDER__) && __BYTE_ORDER__ == __ORDER_BIG_ENDIAN__
+    value = __builtin_bswap64(value);
+#endif
+    return value;
+}
+
+/* Digests find torn and missing writes; they are not a defence against a
+ * hostile file. */
+static const uint64_t P1 = UINT64_C(0x9E3779B185EBCA87), P2 = UINT64_C(0xC2B2AE3D27D4EB4F),
+                      P3 = UINT64_C(0x165667B19E3779F9), P4 = UINT64_C(0x85EBCA77C2B2AE63),
+                      P5 = UINT64_C(0x27D4EB2F165667C5);
+
+static uint64_t rotate(uint64_t value, unsigned bits) { return value << bits | value >> (64 - bits); }
+static uint64_t mix(uint64_t lane, uint64_t word) { return rotate(lane + word * P2, 31) * P1; }
+
+static void stripes(Digest *digest, const unsigned char *bytes, size_t count) {
+    uint64_t a = digest->lane[0], b = digest->lane[1], c = digest->lane[2], d = digest->lane[3];
+    for (; count; --count, bytes += 32) {
+        a = mix(a, load64(bytes));
+        b = mix(b, load64(bytes + 8));
+        c = mix(c, load64(bytes + 16));
+        d = mix(d, load64(bytes + 24));
+    }
+    digest->lane[0] = a, digest->lane[1] = b, digest->lane[2] = c, digest->lane[3] = d;
+}
+
+static void digest_add(Digest *digest, const unsigned char *bytes, size_t length) {
+    digest->length += length;
+    if (digest->count) {
+        size_t take = 32 - digest->count < length ? 32 - digest->count : length;
+        memcpy(digest->held + digest->count, bytes, take);
+        digest->count += take;
+        bytes += take;
+        length -= take;
+        if (digest->count < 32) return;
+        stripes(digest, digest->held, 1);
+        digest->count = 0;
+    }
+    stripes(digest, bytes, length / 32);
+    digest->count = length % 32;
+    if (digest->count) memcpy(digest->held, bytes + length - digest->count, digest->count);
+}
+
+static uint64_t digest_end(const Digest *digest) {
+    const uint64_t *lane = digest->lane;
+    uint64_t hash = rotate(lane[0], 1) + rotate(lane[1], 7) + rotate(lane[2], 12) + rotate(lane[3], 18);
+    for (unsigned i = 0; i < 4; ++i) hash = (hash ^ mix(0, lane[i])) * P1 + P4;
+    hash += digest->length;
+    size_t i = 0;
+    for (; i + 8 <= digest->count; i += 8) hash = rotate(hash ^ mix(0, load64(digest->held + i)), 27) * P1 + P4;
+    for (; i < digest->count; ++i) hash = rotate(hash ^ digest->held[i] * P5, 11) * P1;
+    hash ^= hash >> 33;
+    hash *= P2;
+    hash ^= hash >> 29;
+    hash *= P3;
+    return hash ^ hash >> 32;
+}
+
+static uint64_t digest_of(const unsigned char *bytes, size_t length) {
+    Digest digest = {0};
+    digest_add(&digest, bytes, length);
+    return digest_end(&digest);
+}
+
+#ifdef PURE_STORE_TEST
+static unsigned long syncs;
+unsigned long pure_tree_test_syncs(void) { return __atomic_load_n(&syncs, __ATOMIC_RELAXED); }
+#define COUNT_SYNC() __atomic_add_fetch(&syncs, 1, __ATOMIC_RELAXED)
+#else
+#define COUNT_SYNC() ((void)0)
+#endif
+
+static bool sync_fd(int fd) {
+    int rc;
+    COUNT_SYNC();
+    do { rc = fsync(fd); } while (rc && errno == EINTR);
+    return rc == 0;
+}
+
+/* Data and the size needed to read it back, without timestamps. */
+static bool sync_data(int fd) {
+#ifdef __APPLE__
+    return sync_fd(fd);
+#else
+    int rc;
+    COUNT_SYNC();
+    do { rc = fdatasync(fd); } while (rc && errno == EINTR);
+    return rc == 0;
+#endif
+}
+
+static bool write_at(int fd, const unsigned char *bytes, size_t length, uint64_t offset) {
+    while (length) {
+        ssize_t n = pwrite(fd, bytes, length, (off_t)offset);
+        if (n < 0 && errno == EINTR) continue;
+        if (n <= 0) return false;
+        bytes += (size_t)n;
+        length -= (size_t)n;
+        offset += (uint64_t)n;
+    }
+    return true;
+}
+
+static bool read_at(int fd, unsigned char *bytes, size_t length, uint64_t offset) {
+    while (length) {
+        ssize_t n = pread(fd, bytes, length, (off_t)offset);
+        if (n < 0 && errno == EINTR) continue;
+        if (n <= 0) return false;
+        bytes += (size_t)n;
+        length -= (size_t)n;
+        offset += (uint64_t)n;
+    }
+    return true;
+}
+
+static void encode_head(unsigned char out[SLOT], const unsigned char magic[8], const Head *head) {
+    memset(out, 0, SLOT);
+    memcpy(out, magic, 8);
+    encode64(out + 8, head->seq);
+    encode64(out + 16, head->end);
+    encode64(out + 24, head->catalog);
+    encode64(out + 32, head->catalog_bytes);
+    encode64(out + 40, head->live);
+    encode64(out + 48, head->start);
+    encode64(out + 56, head->check);
+    encode64(out + SLOT - 8, digest_of(out, SLOT - 8));
+}
+
+/* A slot's head (record false) or a commit record's. */
+static bool decode_head(const unsigned char in[SLOT], bool record, Head *head) {
+    bool magic = record ? !memcmp(in, record_magic, 8) : !memcmp(in, head_magic, 8) || !memcmp(in, earlier_magic, 8);
+    if (!magic || decode64(in + SLOT - 8) != digest_of(in, SLOT - 8)) return false;
+    *head = (Head){decode64(in + 8), decode64(in + 16), decode64(in + 24), decode64(in + 32),
+                   decode64(in + 40), decode64(in + 48), decode64(in + 56)};
+    return head->seq && head->end <= (uint64_t)INT64_MAX && head->end <= SIZE_MAX &&
+           head->start >= DATA && head->start <= head->end && head->catalog_bytes % RECORD == 0 &&
+           !head->catalog == !head->catalog_bytes;
+}
+
+/* ---- Trusted heads --------------------------------------------------------
+ * One sync makes a commit's bytes and its head durable together, so a crash
+ * during it can leave the head on the device without all of the bytes: a
+ * head counts only once its digest matches them. The older slot needs no
+ * such luck. A writer builds only on a durable head, so the newer head was
+ * written after the older one's commit was complete on the device.
+ * The process remembers the heads it published or checked, so it reads a
+ * commit's bytes at most once, and which of them it made durable. */
+enum { UNKNOWN, BROKEN, CHECKED, DURABLE, PUBLISHING };
+enum { KNOWN = 32 };
+typedef struct { dev_t device; ino_t inode; uint64_t seq, check; int state; } Known;
+/* The newest head that counts per file, to follow records from. */
+typedef struct { dev_t device; ino_t inode; Head head; } Newest;
+enum { NEWEST = 4 };
+static struct {
+    pthread_mutex_t lock;
+    Known heads[KNOWN];
+    unsigned next;
+    Known publishing; /* written here, not durable yet */
+    Newest newest[NEWEST];
+    unsigned next_newest;
+} known = {.lock = PTHREAD_MUTEX_INITIALIZER};
+
+static bool names(const Known *known_head, const struct stat *file, const Head *head) {
+    return known_head->state && known_head->device == file->st_dev && known_head->inode == file->st_ino &&
+           known_head->seq == head->seq && known_head->check == head->check;
+}
+
+static int recall(const struct stat *file, const Head *head) {
+    int state = UNKNOWN;
+    pthread_mutex_lock(&known.lock);
+    if (names(&known.publishing, file, head)) state = PUBLISHING;
+    for (unsigned i = 0; i < KNOWN && state == UNKNOWN; ++i)
+        if (names(&known.heads[i], file, head)) state = known.heads[i].state;
+    pthread_mutex_unlock(&known.lock);
+    return state;
+}
+
+/* PUBLISHING marks the one head being made durable; any other state ends
+ * that mark (UNKNOWN only ends it). */
+static void remember(const struct stat *file, const Head *head, int state) {
+    Known entry = {file->st_dev, file->st_ino, head->seq, head->check, state};
+    pthread_mutex_lock(&known.lock);
+    if (state == PUBLISHING) known.publishing = entry;
+    else if (names(&known.publishing, file, head)) known.publishing.state = UNKNOWN;
+    if (state != PUBLISHING && state != UNKNOWN) {
+        Known *at = NULL;
+        for (unsigned i = 0; i < KNOWN && !at; ++i)
+            if (names(&known.heads[i], file, head)) at = &known.heads[i];
+        if (!at) known.heads[known.next++ % KNOWN] = entry;
+        else if (at->state != DURABLE) *at = entry; /* a durable head stays durable */
+    }
+    if (state == CHECKED || state == DURABLE) {
+        Newest *at = NULL;
+        for (unsigned i = 0; i < NEWEST && !at; ++i)
+            if (known.newest[i].head.seq && known.newest[i].device == file->st_dev &&
+                known.newest[i].inode == file->st_ino) at = &known.newest[i];
+        if (!at) at = &known.newest[known.next_newest++ % NEWEST];
+        else if (at->head.seq >= head->seq) at = NULL;
+        if (at) *at = (Newest){file->st_dev, file->st_ino, *head};
+    }
+    pthread_mutex_unlock(&known.lock);
+}
+
+static bool newest_known(const struct stat *file, Head *out) {
+    bool found = false;
+    pthread_mutex_lock(&known.lock);
+    for (unsigned i = 0; i < NEWEST && !found; ++i)
+        if (known.newest[i].head.seq && known.newest[i].device == file->st_dev && known.newest[i].inode == file->st_ino) {
+            *out = known.newest[i].head;
+            found = true;
+        }
+    pthread_mutex_unlock(&known.lock);
+    return found;
+}
+
+#ifdef PURE_STORE_TEST
+/* Test-only: forget every head, as a new process after a crash would. */
+void pure_tree_test_forget(void) {
+    pthread_mutex_lock(&known.lock);
+    memset(known.heads, 0, sizeof(known.heads));
+    memset(known.newest, 0, sizeof(known.newest));
+    known.publishing.state = UNKNOWN;
+    pthread_mutex_unlock(&known.lock);
+}
+#endif
+
+/* Whether a head's commit has all of its bytes. */
+static bool whole(int fd, const Head *head) {
+    unsigned char buffer[16384];
+    Digest digest = {0};
+    for (uint64_t at = head->start; at < head->end;) {
+        size_t n = head->end - at < sizeof(buffer) ? (size_t)(head->end - at) : sizeof(buffer);
+        if (!read_at(fd, buffer, n, at)) return false;
+        digest_add(&digest, buffer, n);
+        at += n;
+    }
+    return digest_end(&digest) == head->check;
+}
+
+/* The newest head that counts. A reader passes over the head this process is
+ * still making durable, so the process never answers with a commit that a
+ * crash could take back; a writer never meets one (it is the only writer). */
+static bool pick(int fd, const struct stat *file, const unsigned char *slots[2], bool writer, Head *out,
+                 uint64_t sequences[2]) {
+    Head heads[2];
+    bool valid[2];
+    for (unsigned i = 0; i < 2; ++i) {
+        valid[i] = decode_head(slots[i], false, &heads[i]) && heads[i].end <= (uint64_t)file->st_size;
+        if (sequences) sequences[i] = valid[i] ? heads[i].seq : 0;
+    }
+    unsigned newest = valid[1] && (!valid[0] || heads[1].seq > heads[0].seq);
+    for (unsigned turn = 0; turn < 2; ++turn) {
+        unsigned slot = turn ? !newest : newest;
+        const Head *head = &heads[slot];
+        if (!valid[slot]) continue;
+        int state = recall(file, head);
+        if (state == BROKEN || (state == PUBLISHING && !writer)) continue;
+        if (state == UNKNOWN) {
+            bool counts = whole(fd, head);
+            remember(file, head, counts ? CHECKED : BROKEN);
+            if (!counts) continue;
+        }
+        *out = *head;
+        return true;
+    }
+    return false;
+}
+
+/* The commit whose record follows head's bytes, if its own bytes are whole. */
+static bool follow(int fd, const struct stat *file, bool writer, const Head *head, Head *out) {
+    unsigned char record[SLOT];
+    uint64_t size = (uint64_t)file->st_size;
+    Head next;
+    if (head->end > size || size - head->end < SLOT || !read_at(fd, record, SLOT, head->end) ||
+        !decode_head(record, true, &next) || next.seq != head->seq + 1 || next.start != head->end + SLOT ||
+        next.end > size) return false;
+    int state = recall(file, &next);
+    if (state == BROKEN || (state == PUBLISHING && !writer)) return false;
+    if (state == UNKNOWN) {
+        bool counts = whole(fd, &next);
+        remember(file, &next, counts ? CHECKED : BROKEN);
+        if (!counts) return false;
+    }
+    *out = next;
+    return true;
+}
+
+/* Whether head's record is in the file as it was made. */
+static bool recorded(int fd, const struct stat *file, const Head *head) {
+    unsigned char stored[SLOT], made[SLOT];
+    if (head->start < DATA + SLOT || head->end > (uint64_t)file->st_size) return false;
+    encode_head(made, record_magic, head);
+    return read_at(fd, stored, SLOT, head->start - SLOT) && !memcmp(stored, made, SLOT);
+}
+
+/* The newest commit: from the newer slot's head that counts, or from the
+ * newest this process has seen in the file when its record is still there,
+ * along the records that follow. sequences gets the slots' (0: none). */
+static bool read_head(int fd, struct stat *file, bool writer, Head *out, uint64_t sequences[2]) {
+    unsigned char slots[BLOCK + SLOT];
+    Head head, seen;
+    if (fstat(fd, file) || !S_ISREG(file->st_mode) || file->st_size < 0 || !read_at(fd, slots, sizeof(slots), 0) ||
+        !pick(fd, file, (const unsigned char *[2]){slots, slots + BLOCK}, writer, &head, sequences)) return false;
+    if (newest_known(file, &seen) && seen.seq > head.seq && recorded(fd, file, &seen)) head = seen;
+    while (follow(fd, file, writer, &head, &head)) {}
+    *out = head;
+    return true;
+}
+
+static bool map_tree(int fd, const struct stat *file, const Head *head, Tree *tree) {
+    *tree = (Tree){0};
+    if ((uint64_t)file->st_size < head->end) return false;
+    void *map = mmap(NULL, (size_t)head->end, PROT_READ, MAP_SHARED, fd, 0);
+    if (map == MAP_FAILED) return false;
+    *tree = (Tree){.map = map, .end = head->end};
+    return true;
+}
+
+/* A lent mapping may be released on another thread (a host that answers on
+ * its connection threads): its count is atomic. */
+static void release_mapping(Mapping *mapping) {
+    if (!mapping || __atomic_sub_fetch(&mapping->refs, 1, __ATOMIC_ACQ_REL)) return;
+    munmap((void *)mapping->map, (size_t)mapping->end);
+    free(mapping);
+}
+
+static void release_tree(Tree *tree) {
+    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);
+    }
+    free(tree->nodes);
+    *tree = (Tree){0};
+}
+
+static const unsigned char *leaf_bytes(const Tree *tree, Entry entry) {
+    if (!(entry.ref & LEAF_BIT) || !entry.bytes || entry.bytes > LEAF_MAX) return NULL;
+    uint64_t at = entry.ref & LOCATION;
+    if (entry.ref & MEMORY_BIT) {
+        if (at >= tree->count) return NULL;
+        const Node *node = &tree->nodes[at];
+        return node->bytes && node->size == entry.bytes ? node->bytes : NULL;
+    }
+    if (at < DATA || at > tree->end || entry.bytes > tree->end - at) return NULL;
+    return tree->map + at;
+}
+
+/* Returns the child count of a checked branch at height, or 0. */
+static size_t children(const Tree *tree, Entry entry, unsigned height, Entry out[FAN_MAX]) {
+    if ((entry.ref & LEAF_BIT) || !height) return 0;
+    uint64_t at = entry.ref & LOCATION;
+    size_t count;
+    if (entry.ref & MEMORY_BIT) {
+        if (at >= tree->count) return 0;
+        const Node *node = &tree->nodes[at];
+        if (node->bytes || node->height != height || !node->size || node->size > FAN_MAX) return 0;
+        count = node->size;
+        memcpy(out, node->children, count * sizeof(*out));
+    } else {
+        if (at < DATA || at % 8 || at > tree->end || tree->end - at < 8) return 0;
+        const unsigned char *p = tree->map + at;
+        count = (size_t)p[4] | (size_t)p[5] << 8;
+        unsigned stored = (unsigned)p[6] | (unsigned)p[7] << 8;
+        if (memcmp(p, branch_magic, sizeof(branch_magic)) || stored != height || !count ||
+            count > FAN_MAX || (tree->end - at - 8) / 16 < count) return 0;
+        for (size_t i = 0; i < count; ++i)
+            out[i] = (Entry){decode64(p + 8 + 16 * i), decode64(p + 16 + 16 * i)};
+    }
+    uint64_t sum = 0;
+    for (size_t i = 0; i < count; ++i) {
+        bool leaf = (out[i].ref & LEAF_BIT) != 0;
+        if (!out[i].bytes || out[i].bytes > UINT64_MAX - sum || leaf != (height == 1)) return 0;
+        sum += out[i].bytes;
+    }
+    return sum == entry.bytes ? count : 0;
+}
+
+static bool root_height(const Tree *tree, Entry root, unsigned *height) {
+    *height = 0;
+    if (!root.bytes) return !root.ref;
+    if (root.ref & LEAF_BIT) return true;
+    uint64_t at = root.ref & LOCATION;
+    if (root.ref & MEMORY_BIT) {
+        if (at >= tree->count || tree->nodes[at].bytes) return false;
+        *height = tree->nodes[at].height;
+    } else {
+        if (at < DATA || at > tree->end || tree->end - at < 8) return false;
+        *height = (unsigned)tree->map[at + 6] | (unsigned)tree->map[at + 7] << 8;
+    }
+    return *height >= 1 && *height <= HEIGHT_MAX;
+}
+
+/* Visit a range as borrowed extents: 1 done, 0 damaged, -1 stopped by visit. */
+static int visit_range(const Tree *tree, Entry entry, unsigned height, uint64_t offset,
+                       uint64_t length, bool (*visit)(void *, PureSpan), void *context) {
+    if (!height) {
+        const unsigned char *bytes = leaf_bytes(tree, entry);
+        if (!bytes || offset > entry.bytes || length > entry.bytes - offset) return 0;
+        return visit(context, (PureSpan){bytes + offset, (size_t)length}) ? 1 : -1;
+    }
+    Entry kids[FAN_MAX];
+    size_t count = children(tree, entry, height, kids);
+    if (!count) return 0;
+    uint64_t start = 0;
+    for (size_t i = 0; i < count && length; ++i) {
+        uint64_t end = start + kids[i].bytes;
+        if (offset < end) {
+            uint64_t inside = offset - start, take = kids[i].bytes - inside;
+            if (take > length) take = length;
+            int result = visit_range(tree, kids[i], height - 1, inside, take, visit, context);
+            if (result != 1) return result;
+            offset += take;
+            length -= take;
+        }
+        start = end;
+    }
+    return length ? 0 : 1;
+}
+
+static bool copy_span(void *context, PureSpan span) {
+    unsigned char **out = context;
+    memcpy(*out, span.data, span.size);
+    *out += span.size;
+    return true;
+}
+
+static bool copy_range(const Tree *tree, Entry entry, unsigned height, uint64_t offset,
+                       uint64_t length, unsigned char *out) {
+    return !length || visit_range(tree, entry, height, offset, length, copy_span, &out) == 1;
+}
+
+/* The bytes of a range held by one leaf, or NULL when it spans leaves. */
+static const unsigned char *single_leaf(const Tree *tree, Entry entry, unsigned height,
+                                        uint64_t offset, uint64_t length) {
+    while (height) {
+        Entry kids[FAN_MAX];
+        size_t count = children(tree, entry, height, kids), i = 0;
+        uint64_t start = 0;
+        if (!count) return NULL;
+        while (i + 1 < count && offset >= start + kids[i].bytes) start += kids[i++].bytes;
+        if (offset - start > kids[i].bytes || length > kids[i].bytes - (offset - start)) return NULL;
+        entry = kids[i];
+        offset -= start;
+        --height;
+    }
+    const unsigned char *bytes = leaf_bytes(tree, entry);
+    return bytes && offset <= entry.bytes && length <= entry.bytes - offset ? bytes + offset : NULL;
+}
+
+static bool push(List *list, Entry entry) {
+    if (list->count == list->capacity) {
+        size_t capacity = list->capacity ? 2 * list->capacity : 8;
+        Entry *items = realloc(list->items, capacity * sizeof(*items));
+        if (!items) return false;
+        list->items = items;
+        list->capacity = capacity;
+    }
+    list->items[list->count++] = entry;
+    return true;
+}
+
+static size_t node_bytes(const Node *node) {
+    return node->bytes ? node->size : node->size * sizeof(Entry);
+}
+
+static bool add_node(Tree *tree, Node node, uint64_t *index) {
+    tree->bytes += node_bytes(&node);
+    if (tree->free_head) {
+        *index = tree->free_head - 1;
+        tree->free_head = tree->nodes[*index].size;
+        --tree->free_count;
+        tree->nodes[*index] = node;
+        return true;
+    }
+    if (tree->count == tree->capacity) {
+        size_t capacity = tree->capacity ? 2 * tree->capacity : 64;
+        Node *nodes = realloc(tree->nodes, capacity * sizeof(*nodes));
+        if (!nodes) { tree->bytes -= node_bytes(&node); return false; }
+        tree->nodes = nodes;
+        tree->capacity = capacity;
+    }
+    tree->nodes[tree->count] = node;
+    *index = tree->count++;
+    return true;
+}
+
+static void take(Source *source, unsigned char *out, size_t length) {
+    while (length) {
+        PureSpan part = source->part[source->index];
+        size_t left = part.size - source->offset;
+        if (!left) { ++source->index; source->offset = 0; continue; }
+        size_t n = left < length ? left : length;
+        memcpy(out, part.data + source->offset, n);
+        out += n;
+        length -= n;
+        source->offset += n;
+    }
+}
+
+/* Balanced leaves: every leaf of a multi-leaf result holds >= LEAF_MAX / 2. */
+static bool chunk_bytes(Tree *tree, Source *source, uint64_t total, List *out) {
+    uint64_t pieces = (total + LEAF_MAX - 1) / LEAF_MAX;
+    for (uint64_t i = 0; i < pieces; ++i) {
+        size_t size = (size_t)(total / pieces + (i < total % pieces));
+        unsigned char *bytes = malloc(size);
+        uint64_t index;
+        if (!bytes) return false;
+        take(source, bytes, size);
+        if (!add_node(tree, (Node){.bytes = bytes, .size = size}, &index)) { free(bytes); return false; }
+        if (!push(out, (Entry){LEAF_BIT | MEMORY_BIT | index, size})) return false;
+    }
+    return true;
+}
+
+static bool chunk_entries(Tree *tree, const Entry *items, size_t count, unsigned height, List *out) {
+    size_t pieces = (count + FAN_MAX - 1) / FAN_MAX;
+    for (size_t i = 0, at = 0; i < pieces; ++i) {
+        size_t size = count / pieces + (i < count % pieces);
+        Entry *kids = malloc(size * sizeof(*kids));
+        uint64_t sum = 0, index;
+        if (!kids) return false;
+        memcpy(kids, items + at, size * sizeof(*kids));
+        for (size_t k = 0; k < size; ++k) sum += kids[k].bytes;
+        if (!add_node(tree, (Node){.children = kids, .size = size, .height = height}, &index)) {
+            free(kids);
+            return false;
+        }
+        if (!push(out, (Entry){MEMORY_BIT | index, sum})) return false;
+        at += size;
+    }
+    return true;
+}
+
+/* Only nodes created by this edit can be underfull; stored siblings are left as they are. */
+static bool underfull(const Tree *tree, Entry entry, unsigned height) {
+    if (!(entry.ref & MEMORY_BIT)) return false;
+    return height ? tree->nodes[entry.ref & LOCATION].size < FAN_MIN : entry.bytes < LEAF_MIN;
+}
+
+static bool replace(List *list, size_t at, size_t removed, const List *with) {
+    size_t count = list->count - removed + with->count;
+    if (count > list->capacity) {
+        Entry *items = realloc(list->items, count * sizeof(*items));
+        if (!items) return false;
+        list->items = items;
+        list->capacity = count;
+    }
+    memmove(list->items + at + with->count, list->items + at + removed,
+            (list->count - at - removed) * sizeof(*list->items));
+    memcpy(list->items + at, with->items, with->count * sizeof(*list->items));
+    list->count = count;
+    return true;
+}
+
+static bool merge(Tree *tree, Entry a, Entry b, unsigned height, List *out);
+
+/* Merge each new underfull node with a neighbour. A merge removes a node or
+ * yields two nodes of at least half capacity, so the loop terminates. */
+static bool repair(Tree *tree, List *all, unsigned height) {
+    bool ok = true;
+    for (size_t i = 0; ok && all->count > 1 && i < all->count;) {
+        if (!underfull(tree, all->items[i], height)) { ++i; continue; }
+        size_t at = i + 1 < all->count ? i : i - 1;
+        List merged = {0};
+        ok = merge(tree, all->items[at], all->items[at + 1], height, &merged) &&
+             replace(all, at, 2, &merged);
+        free(merged.items);
+        i = at;
+    }
+    return ok;
+}
+
+static bool merge(Tree *tree, Entry a, Entry b, unsigned height, List *out) {
+    if (!height) {
+        const unsigned char *x = leaf_bytes(tree, a), *y = leaf_bytes(tree, b);
+        if (!x || !y) return false;
+        Source source = {.part = {{x, (size_t)a.bytes}, {y, (size_t)b.bytes}}};
+        return chunk_bytes(tree, &source, a.bytes + b.bytes, out);
+    }
+    Entry kids[FAN_MAX];
+    List all = {0};
+    size_t count = children(tree, a, height, kids);
+    bool ok = count != 0;
+    for (size_t i = 0; ok && i < count; ++i) ok = push(&all, kids[i]);
+    count = ok ? children(tree, b, height, kids) : 0;
+    ok = count != 0;
+    for (size_t i = 0; ok && i < count; ++i) ok = push(&all, kids[i]);
+    /* A small child carried in from a one-child parent is repaired here too. */
+    ok = ok && repair(tree, &all, height - 1) && chunk_entries(tree, all.items, all.count, height, out);
+    free(all.items);
+    return ok;
+}
+
+/* Replace [offset, offset + remove) of entry with insert. Appends the
+ * replacement nodes, all at entry's height, to out (possibly none). */
+static bool edit(Tree *tree, Entry entry, unsigned height, uint64_t offset, uint64_t remove,
+                 PureSpan insert, List *out) {
+    if (offset > entry.bytes || remove > entry.bytes - offset) return false;
+    if (!height) {
+        const unsigned char *bytes = leaf_bytes(tree, entry);
+        if (!bytes) return false;
+        Source source = {.part = {{bytes, (size_t)offset}, insert,
+            {bytes + offset + remove, (size_t)(entry.bytes - offset - remove)}}};
+        return chunk_bytes(tree, &source, entry.bytes - remove + insert.size, out);
+    }
+    Entry kids[FAN_MAX];
+    size_t count = children(tree, entry, height, kids), first = count, final = count;
+    if (!count) return false;
+    uint64_t last = remove ? offset + remove - 1 : offset, start = 0, first_start = 0, final_start = 0;
+    for (size_t i = 0; i < count; ++i) {
+        uint64_t end = start + kids[i].bytes;
+        bool tail = i + 1 == count;
+        if (first == count && (offset < end || tail)) { first = i; first_start = start; }
+        if (first != count && (last < end || tail)) { final = i; final_start = start; break; }
+        start = end;
+    }
+    List middle = {0}, all = {0};
+    bool ok = first == final ?
+        edit(tree, kids[first], height - 1, offset - first_start, remove, insert, &middle) :
+        edit(tree, kids[first], height - 1, offset - first_start,
+             first_start + kids[first].bytes - offset, insert, &middle) &&
+        edit(tree, kids[final], height - 1, 0, offset + remove - final_start, (PureSpan){0}, &middle);
+    for (size_t i = 0; ok && i < first; ++i) ok = push(&all, kids[i]);
+    for (size_t i = 0; ok && i < middle.count; ++i) ok = push(&all, middle.items[i]);
+    for (size_t i = final + 1; ok && i < count; ++i) ok = push(&all, kids[i]);
+    free(middle.items);
+    ok = ok && repair(tree, &all, height - 1) && chunk_entries(tree, all.items, all.count, height, out);
+    free(all.items);
+    return ok;
+}
+
+/* The new root of root with [offset, offset + remove) replaced by insert. */
+static bool rebuild(Tree *tree, Entry root, uint64_t offset, uint64_t remove,
+                    PureSpan insert, Entry *out) {
+    unsigned height;
+    List list = {0};
+    bool ok = root_height(tree, root, &height);
+    if (ok && !root.bytes) {
+        Source source = {.part = {insert}};
+        ok = !offset && !remove && chunk_bytes(tree, &source, insert.size, &list);
+    } else if (ok) ok = edit(tree, root, height, offset, remove, insert, &list);
+    while (ok && list.count > 1) {
+        List up = {0};
+        ok = height < HEIGHT_MAX && chunk_entries(tree, list.items, list.count, ++height, &up);
+        free(list.items);
+        list = up;
+    }
+    Entry result = ok && list.count ? list.items[0] : (Entry){0};
+    free(list.items);
+    while (ok && height && result.bytes) {
+        Entry kids[FAN_MAX];
+        size_t count = children(tree, result, height, kids);
+        if (count != 1) { ok = count > 1; break; }
+        result = kids[0];
+        --height;
+    }
+    *out = result;
+    return ok;
+}
+
+static bool flush(Writer *writer) {
+    if (writer->used && !writer->failed) {
+        digest_add(&writer->digest, writer->buffer, writer->used);
+        if (!write_at(writer->fd, writer->buffer, writer->used, writer->at - writer->used)) writer->failed = true;
+    }
+    writer->used = 0;
+    return !writer->failed;
+}
+
+/* Zeros from..to; a failure only costs speed. */
+static void zero_fill(int fd, uint64_t from, uint64_t to) {
+    static const unsigned char zeros[BUFFER];
+    while (from < to) {
+        size_t n = to - from < BUFFER ? (size_t)(to - from) : BUFFER;
+        if (!write_at(fd, zeros, n, from)) return;
+        from += n;
+    }
+}
+
+/* Where the written zeros after a published end should reach. */
+static uint64_t tail_end(uint64_t end) {
+    return end + ((end - DATA) / 8 > TAIL_STEP ? (end - DATA) / 8 : TAIL_STEP);
+}
+
+static void put_bytes(Writer *writer, const unsigned char *bytes, size_t length) {
+    while (length && !writer->failed) {
+        if (writer->used == BUFFER && !flush(writer)) return;
+        size_t n = BUFFER - writer->used < length ? BUFFER - writer->used : length;
+        memcpy(writer->buffer + writer->used, bytes, n);
+        writer->used += n;
+        writer->at += n;
+        bytes += n;
+        length -= n;
+    }
+}
+
+static uint64_t put_branch(Writer *writer, const Entry *kids, size_t count, unsigned height) {
+    static const unsigned char zero[8];
+    put_bytes(writer, zero, (size_t)((8 - writer->at % 8) % 8));
+    uint64_t at = writer->at;
+    unsigned char header[8];
+    memcpy(header, branch_magic, sizeof(branch_magic));
+    header[4] = (unsigned char)count;
+    header[5] = (unsigned char)(count >> 8);
+    header[6] = (unsigned char)height;
+    header[7] = (unsigned char)(height >> 8);
+    put_bytes(writer, header, sizeof(header));
+    for (size_t i = 0; i < count; ++i) {
+        unsigned char pair[16];
+        encode64(pair, kids[i].ref);
+        encode64(pair + 8, kids[i].bytes);
+        put_bytes(writer, pair, sizeof(pair));
+    }
+    return at;
+}
+
+/* Append a subtree's arena nodes, children first, and set entry to the
+ * stored reference. The arena is not changed, so a failed commit leaves a
+ * draft intact. With relocate, stored nodes are copied too (compaction, or a
+ * draft whose base file was replaced by compaction). */
+static bool write_subtree(const Tree *tree, Writer *writer, Entry *entry, unsigned height,
+                          bool relocate) {
+    if (!entry->bytes || (!(entry->ref & MEMORY_BIT) && !relocate)) return true;
+    if (!height) {
+        const unsigned char *bytes = leaf_bytes(tree, *entry);
+        if (!bytes) return false;
+        entry->ref = LEAF_BIT | writer->at;
+        put_bytes(writer, bytes, (size_t)entry->bytes);
+        return !writer->failed;
+    }
+    Entry kids[FAN_MAX];
+    size_t count = children(tree, *entry, height, kids);
+    if (!count) return false;
+    for (size_t i = 0; i < count; ++i)
+        if (!write_subtree(tree, writer, &kids[i], height - 1, relocate)) return false;
+    entry->ref = put_branch(writer, kids, count, height);
+    return !writer->failed;
+}
+
+static bool write_root(const Tree *tree, Writer *writer, Entry *root, bool relocate) {
+    unsigned height;
+    return root_height(tree, *root, &height) && write_subtree(tree, writer, root, height, relocate);
+}
+
+static bool read_record(const Tree *tree, Entry catalog, uint64_t index,
+                        unsigned char record[RECORD], Entry *document, int64_t *revision) {
+    unsigned height;
+    if (!root_height(tree, catalog, &height) ||
+        !copy_range(tree, catalog, height, index * RECORD, RECORD, record)) return false;
+    *document = (Entry){decode64(record + KEY), decode64(record + KEY + 8)};
+    uint64_t stored = decode64(record + KEY + 16);
+    *revision = (int64_t)stored;
+    return stored && stored <= (uint64_t)INT64_MAX && !document->ref == !document->bytes &&
+           document->bytes <= PURE_OBJECT_LIMIT;
+}
+
+/* Binary search over records ordered by their zero-padded key bytes. */
+static bool find(const Tree *tree, Entry catalog, const char *key, uint64_t *index,
+                 bool *found, Entry *document, int64_t *revision) {
+    unsigned char wanted[KEY] = {0}, record[RECORD];
+    memcpy(wanted, key, strlen(key));
+    uint64_t low = 0, high = catalog.bytes / RECORD;
+    *found = false;
+    *document = (Entry){0};
+    *revision = 0;
+    while (low < high) {
+        uint64_t middle = low + (high - low) / 2;
+        if (!read_record(tree, catalog, middle, record, document, revision)) return false;
+        int order = memcmp(wanted, record, KEY);
+        if (!order) { *index = middle; *found = true; return true; }
+        if (order < 0) high = middle; else low = middle + 1;
+    }
+    *document = (Entry){0};
+    *revision = 0;
+    *index = low;
+    return true;
+}
+
+static int create_temporary(int directory, char name[96]) {
+    for (unsigned attempt = 0; attempt < 64; ++attempt) {
+        if (temporary_sequence == UINT64_MAX) break;
+        snprintf(name, 96, ".pending-%ld-%" PRIu64, (long)getpid(), ++temporary_sequence);
+        int fd = openat(directory, name, O_RDWR | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC, 0600);
+        if (fd >= 0) return fd;
+        if (errno != EEXIST) break;
+    }
+    name[0] = 0; /* Never remove an existing file on a name collision. */
+    return -1;
+}
+
+/* Publish a fresh file with both slots pointing at head. It is synced whole
+ * before it is renamed into place, so its head names no commit bytes. */
+static bool install(int directory, char temporary[96], int fd, Head *head) {
+    unsigned char slot[SLOT];
+    struct stat file;
+    head->start = head->end;
+    head->check = digest_end(&(Digest){0});
+    encode_head(slot, head_magic, head);
+    zero_fill(fd, head->end, tail_end(head->end));
+    bool ok = write_at(fd, slot, SLOT, 0) && write_at(fd, slot, SLOT, BLOCK) && sync_fd(fd) && !fstat(fd, &file);
+    if (close(fd)) ok = false;
+    ok = ok && !renameat(directory, temporary, directory, tree_name);
+    if (ok) temporary[0] = 0;
+    ok = ok && sync_fd(directory);
+    if (ok) remember(&file, head, DURABLE);
+    return ok;
+}
+
+/* Rewrite only reachable nodes. Open readers keep their mapping of the
+ * previous file. A failure leaves the published file untouched. */
+static void compact(int directory, int source, const struct stat *file, const Head *head) {
+    Tree tree;
+    if (!map_tree(source, file, head, &tree)) return;
+    char temporary[96];
+    int fd = create_temporary(directory, temporary);
+    Writer *writer = fd >= 0 ? malloc(sizeof(*writer)) : NULL;
+    uint64_t count = head->catalog_bytes / RECORD;
+    unsigned char *records = count ? malloc((size_t)(count * RECORD)) : NULL;
+    Entry catalog = {head->catalog, head->catalog_bytes};
+    unsigned height;
+    bool ok = writer && (!count || records) && root_height(&tree, catalog, &height) &&
+              (!count || copy_range(&tree, catalog, height, 0, count * RECORD, records));
+    if (writer) *writer = (Writer){.fd = fd, .at = DATA};
+    for (uint64_t i = 0; ok && i < count; ++i) {
+        unsigned char *record = records + i * RECORD;
+        Entry document = {decode64(record + KEY), decode64(record + KEY + 8)};
+        unsigned document_height;
+        ok = root_height(&tree, document, &document_height) &&
+             write_subtree(&tree, writer, &document, document_height, true);
+        encode64(record + KEY, document.ref);
+    }
+    Entry fresh = {0};
+    ok = ok && rebuild(&tree, (Entry){0}, 0, 0, (PureSpan){records, (size_t)(count * RECORD)}, &fresh) &&
+         write_root(&tree, writer, &fresh, false) && flush(writer);
+    if (ok) {
+        Head compacted = {head->seq + 1, writer->at, fresh.ref, fresh.bytes, writer->at - DATA, 0, 0};
+        ok = install(directory, temporary, fd, &compacted);
+        fd = -1;
+    }
+    if (fd >= 0) close(fd);
+    if (*temporary) unlinkat(directory, temporary, 0);
+    free(records);
+    free(writer);
+    release_tree(&tree);
+}
+
+static bool lock_writer(int lock, bool wait) {
+    struct flock hold = {.l_type = F_WRLCK, .l_whence = SEEK_SET, .l_start = 0, .l_len = 0};
+    int rc;
+    do { rc = fcntl(lock, wait ? F_SETLKW : F_SETLK, &hold); } while (rc && wait && errno == EINTR);
+    return rc == 0;
+}
+
+PureResult pure_store_init(const char *path) {
+    if (!path || !*path) return PURE_STORE_ERROR;
+    bool created = mkdir(path, 0700) == 0;
+    if (!created && errno != EEXIST) return PURE_STORE_ERROR;
+    int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    if (directory < 0) return PURE_STORE_ERROR; /* Never reinterpret an old DB file. */
+    pthread_mutex_lock(&writers);
+    int lock = openat(directory, ".writer.lock", O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
+    struct stat info;
+    bool ok = lock >= 0 && fstat(lock, &info) == 0 && S_ISREG(info.st_mode) && lock_writer(lock, true);
+    if (ok) {
+        Head head;
+        struct stat file;
+        int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+        if (fd >= 0) {
+            ok = read_head(fd, &file, true, &head, NULL); /* An unreadable file is never replaced. */
+            close(fd);
+        } else if (errno != ENOENT) ok = false;
+        else {
+            char temporary[96];
+            int created_fd = create_temporary(directory, temporary);
+            head = (Head){.seq = 1, .end = DATA};
+            ok = created_fd >= 0 && install(directory, temporary, created_fd, &head);
+            if (*temporary) unlinkat(directory, temporary, 0);
+        }
+    }
+    ok = ok && sync_fd(lock) && sync_fd(directory);
+    if (lock >= 0) close(lock);
+    pthread_mutex_unlock(&writers);
+    close(directory);
+    if (ok && created) {
+        char *parent = strdup(path);
+        if (!parent) return PURE_STORE_ERROR;
+        size_t length = strlen(parent);
+        while (length > 1 && parent[length - 1] == '/') parent[--length] = 0;
+        char *slash = strrchr(parent, '/');
+        if (!slash) strcpy(parent, ".");
+        else if (slash == parent) parent[1] = 0;
+        else *slash = 0;
+        int fd = open(parent, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+        free(parent);
+        ok = fd >= 0 && sync_fd(fd);
+        if (fd >= 0) close(fd);
+    }
+    return ok ? PURE_OK : PURE_STORE_ERROR;
+}
+
+void pure_store_release(PureStoredView *view) {
+    if (view->mapped_size) munmap(view->mapping, view->mapped_size);
+    *view = (PureStoredView){0};
+}
+
+PureResult pure_store_select(const char *path, const char *key, size_t offset,
+                             size_t length, bool whole, size_t minimum,
+                             size_t available, PureStoredView *out) {
+    *out = (PureStoredView){0};
+    if (!valid_key(key)) return PURE_STORE_ERROR;
+    int directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    if (directory < 0) return PURE_STORE_ERROR;
+    int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+    close(directory);
+    if (fd < 0) return PURE_STORE_ERROR;
+    Tree tree = {0};
+    Head head;
+    struct stat file;
+    Entry document;
+    uint64_t index;
+    int64_t revision;
+    bool found;
+    unsigned height;
+    PureResult result = PURE_STORE_ERROR;
+    if (!read_head(fd, &file, false, &head, NULL) || !map_tree(fd, &file, &head, &tree) ||
+        !find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found,
+              &document, &revision)) goto finish;
+    if (!found) { result = PURE_NOT_FOUND; goto finish; }
+    size_t size = (size_t)document.bytes;
+    if (size < minimum || offset > size) { result = PURE_LIMIT; goto finish; }
+    if (whole) length = size - offset;
+    if (length > size - offset) { result = PURE_LIMIT; goto finish; }
+    if (length > available) { result = PURE_BUDGET; goto finish; }
+    if (length) {
+        if (!root_height(&tree, document, &height)) goto finish;
+        const unsigned char *direct = single_leaf(&tree, document, height, offset, length);
+        void *mapping;
+        size_t mapped_size, prefix = 0;
+        if (direct) {
+            /* One extent: map it in place, as the flat store does. */
+            long page = sysconf(_SC_PAGESIZE);
+            if (page <= 0) goto finish;
+            size_t start = (size_t)(direct - tree.map), aligned = start - start % (size_t)page;
+            prefix = start - aligned;
+            mapped_size = prefix + length;
+            mapping = mmap(NULL, mapped_size, PROT_READ, MAP_PRIVATE, fd, (off_t)aligned);
+            if (mapping == MAP_FAILED) goto finish;
+        } else {
+            /* A contiguous view of several extents is assembled for this API. */
+            mapped_size = length;
+            mapping = mmap(NULL, mapped_size, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
+            if (mapping == MAP_FAILED) goto finish;
+            if (!copy_range(&tree, document, height, offset, length, mapping) ||
+                mprotect(mapping, mapped_size, PROT_READ)) {
+                munmap(mapping, mapped_size);
+                goto finish;
+            }
+        }
+        out->mapping = mapping;
+        out->mapped_size = mapped_size;
+        out->body = (PureSpan){(const unsigned char *)mapping + prefix, length};
+    }
+    out->revision = revision;
+    out->object_size = size;
+    result = PURE_OK;
+finish:
+    release_tree(&tree);
+    close(fd);
+    return result;
+}
+
+PureResult pure_store_get(const char *path, const char *key, size_t available, PureStoredView *out) {
+    return pure_store_select(path, key, 0, 0, true, 0, available, out);
+}
+
+PureResult pure_store_get_range(const char *path, const char *key, size_t offset,
+                                size_t length, size_t available, PureStoredView *out) {
+    return pure_store_select(path, key, offset, length, false, 0, available, out);
+}
+
+/* A locked writer over the current published file. */
+typedef struct {
+    int directory, lock, fd;
+    Head head;
+    uint64_t slots[2]; /* the sequences of the slots' heads */
+    Tree tree;
+    struct stat file;
+    bool turn; /* holds this process's writer turn */
+} Session;
+
+/* Keep written zeros after the published end, where commits and their
+ * unpublished leftovers go. A sync of blocks the file already has writes
+ * only them; a new size or new blocks add a journal commit (on an SD card
+ * 2.4 ms against 6 ms per commit). The zeros become durable with the next
+ * commit's sync. */
+static void prepare_tail(Session *session) {
+    uint64_t size = (uint64_t)session->file.st_size, end = session->head.end;
+    if (size < end + TAIL_MIN) zero_fill(session->fd, size > end ? size : end, tail_end(end));
+}
+
+static void close_session(Session *session) {
+    release_tree(&session->tree);
+    if (session->fd >= 0) close(session->fd);
+    if (session->lock >= 0) close(session->lock);
+    if (session->directory >= 0) close(session->directory);
+    if (session->turn) pthread_mutex_unlock(&writers);
+    session->turn = false;
+}
+
+static PureResult open_session(const char *path, Session *session) {
+    *session = (Session){.directory = -1, .lock = -1, .fd = -1};
+    pthread_mutex_lock(&writers); /* another thread of this process waits, not fails */
+    session->turn = true;
+    session->directory = open(path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    if (session->directory < 0) return PURE_STORE_ERROR;
+    session->lock = openat(session->directory, ".writer.lock", O_RDWR | O_NOFOLLOW | O_CLOEXEC);
+    struct stat info;
+    if (session->lock < 0 || fstat(session->lock, &info) || !S_ISREG(info.st_mode)) return PURE_STORE_ERROR;
+    if (!lock_writer(session->lock, false))
+        return errno == EACCES || errno == EAGAIN ? PURE_BUSY : PURE_STORE_ERROR;
+    session->fd = openat(session->directory, tree_name, O_RDWR | O_NOFOLLOW | O_CLOEXEC);
+    if (session->fd < 0 || !read_head(session->fd, &session->file, true, &session->head, session->slots))
+        return PURE_STORE_ERROR;
+    /* A head this process did not make durable may be the last write of a
+     * writer that died before its sync: make it durable before building on it. */
+    if (recall(&session->file, &session->head) != DURABLE) {
+        if (!sync_data(session->fd)) return PURE_STORE_ERROR;
+        remember(&session->file, &session->head, DURABLE);
+    }
+    prepare_tail(session);
+    return map_tree(session->fd, &session->file, &session->head, &session->tree) ? PURE_OK : PURE_STORE_ERROR;
+}
+
+/* A commit's bytes start after room for its record at the published end. */
+static uint64_t commit_start(const Session *session) { return session->head.end + SLOT; }
+
+/* The documents' new nodes are already written at writer. Record each in
+ * the catalog (revisions come in as the base and leave as the new one), then
+ * write the head as the record before them and sync once: every document
+ * becomes visible at once, or none does. Readers of this process see the
+ * head only once it is durable. */
+#ifdef PURE_STORE_TEST
+static unsigned long publications;
+unsigned long pure_tree_test_publications(void) { return publications; }
+#endif
+
+static PureResult publish(Session *session, Writer *writer, size_t count, const char *const *keys,
+                          const Entry *documents, int64_t *revisions) {
+#ifdef PURE_STORE_TEST
+    ++publications;
+#endif
+    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);
+    /* Unpublished bytes stay in the tail; the next commit overwrites them. */
+    if (!written || fault(PURE_AFTER_HEADER) || fault(PURE_BEFORE_FILE_SYNC) || fault(PURE_BEFORE_RENAME))
+        return PURE_STORE_ERROR;
+    Head published = {session->head.seq + 1, writer->at, catalog.ref, catalog.bytes, session->head.live,
+                      commit_start(session), digest_end(&writer->digest)};
+    unsigned char record[SLOT], slot[SLOT];
+    encode_head(record, record_magic, &published);
+    encode_head(slot, head_magic, &published);
+    /* Every few commits the head goes into the older slot as well: one more
+     * range for that sync, and fewer records for a reader to follow. */
+    uint64_t newer = session->slots[0] > session->slots[1] ? session->slots[0] : session->slots[1];
+    unsigned older = session->slots[0] > session->slots[1];
+    bool checkpoint = published.seq - newer >= CHECKPOINT;
+    for (size_t i = 0; i < count; ++i) ++revisions[i]; /* Past this point the head may already be visible. */
+    remember(&session->file, &published, PUBLISHING);
+    bool durable = write_at(session->fd, record, SLOT, published.start - SLOT) &&
+                   (!checkpoint || write_at(session->fd, slot, SLOT, older * BLOCK)) &&
+                   !fault(PURE_BEFORE_DIR_SYNC) && sync_data(session->fd);
+    remember(&session->file, &published, durable ? DURABLE : CHECKED);
+    if (!durable) return PURE_UNCERTAIN;
+    if (published.end - DATA > 4 * published.live + COMPACT_SLACK && !fstat(session->fd, &session->file))
+        compact(session->directory, session->fd, &session->file, &published);
+    return PURE_OK;
+}
+
+static PureResult mutate(const char *path, const char *key, int64_t expected,
+                         PureSpan body, bool splice, size_t offset, size_t remove,
+                         size_t max_size, int64_t *revision) {
+    *revision = 0;
+    if (!valid_key(key) || (body.size && !body.data)) return PURE_STORE_ERROR;
+    if (body.size > PURE_OBJECT_LIMIT || max_size > PURE_OBJECT_LIMIT || expected < 0) return PURE_LIMIT;
+    Session session;
+    Writer *writer = NULL;
+    Entry document, next;
+    uint64_t index;
+    bool found;
+    PureResult result = open_session(path, &session);
+    if (result != PURE_OK) goto finish;
+    result = PURE_STORE_ERROR;
+    if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, key,
+              &index, &found, &document, revision)) goto finish;
+    if (*revision != expected) { result = PURE_CONFLICT; goto finish; }
+    if (*revision == INT64_MAX) { result = PURE_LIMIT; goto finish; }
+    uint64_t old_size = document.bytes;
+    if (splice && (offset > old_size || remove > old_size - offset ||
+                   old_size - remove > max_size ||
+                   body.size > max_size - (old_size - remove))) { result = PURE_LIMIT; goto finish; }
+    writer = malloc(sizeof(*writer));
+    if (!writer || !rebuild(&session.tree, splice ? document : (Entry){0}, splice ? offset : 0,
+                            splice ? remove : 0, body, &next)) goto finish;
+    *writer = (Writer){.fd = session.fd, .at = commit_start(&session)};
+    if (!write_root(&session.tree, writer, &next, false)) goto finish;
+    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. 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);
+    if (directory < 0) return PURE_STORE_ERROR;
+    int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+    close(directory);
+    if (fd < 0) return PURE_STORE_ERROR;
+    Tree tree = {0};
+    Head head;
+    struct stat file;
+    PureResult result = PURE_STORE_ERROR;
+    if (read_head(fd, &file, false, &head, NULL) && map_tree(fd, &file, &head, &tree)) {
+        Entry catalog = {head.catalog, head.catalog_bytes}, document;
+        unsigned char record[RECORD];
+        char key[KEY + 1] = {0};
+        int64_t revision;
+        result = PURE_OK;
+        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;
+        }
+    }
+    release_tree(&tree);
+    close(fd);
+    return result;
+}
+
+/* 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;
+    bool pinned; /* drafts open at the version of the last check */
+};
+
+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 */
+    ino_t inode;
+    Entry root;
+    int64_t revision;
+    size_t live;      /* arena bytes reachable at the last collection */
+    bool held;        /* a transaction still reads earlier versions */
+};
+
+/* Rebase the draft on the committed version of its key. */
+static PureResult load_draft(PureDraft *draft) {
+    int directory = open(draft->path, O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC);
+    if (directory < 0) return PURE_STORE_ERROR;
+    int fd = openat(directory, tree_name, O_RDONLY | O_NOFOLLOW | O_CLOEXEC);
+    close(directory);
+    if (fd < 0) return PURE_STORE_ERROR;
+    Tree tree = {0};
+    Head head;
+    struct stat info;
+    Entry document;
+    uint64_t index;
+    int64_t revision;
+    bool found;
+    bool ok = read_head(fd, &info, false, &head, NULL) && map_tree(fd, &info, &head, &tree) &&
+              find(&tree, (Entry){head.catalog, head.catalog_bytes}, draft->key, &index, &found,
+                   &document, &revision);
+    close(fd);
+    if (!ok) { release_tree(&tree); return PURE_STORE_ERROR; }
+    release_tree(&draft->tree);
+    draft->tree = tree;
+    draft->device = info.st_dev;
+    draft->inode = info.st_ino;
+    draft->root = document;
+    draft->revision = revision;
+    draft->live = 0;
+    return PURE_OK;
+}
+
+PureResult pure_draft_open(const char *path, const char *key, PureDraft **out) {
+    *out = NULL;
+    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);
+    PureResult result = load_draft(draft);
+    if (result != PURE_OK) { pure_draft_close(draft); return result; }
+    *out = draft;
+    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 */
+}
+
+PureDraft *pure_draft_freeze(const PureDraft *draft) {
+    if (!draft->tree.shared || (draft->root.ref & MEMORY_BIT)) return NULL;
+    PureDraft *frozen = calloc(1, sizeof(*frozen));
+    if (!frozen) return NULL;
+    Mapping *mapping = draft->tree.shared;
+    __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED);
+    *frozen = (PureDraft){.tree = {.map = mapping->map, .end = mapping->end, .shared = mapping},
+                          .device = draft->device, .inode = draft->inode, .root = draft->root,
+                          .revision = draft->revision};
+    memcpy(frozen->key, draft->key, sizeof(frozen->key));
+    return frozen;
+}
+
+void pure_draft_close(PureDraft *draft) {
+    if (!draft) return;
+    release_tree(&draft->tree);
+    if (!draft->store) free(draft->path);
+    free(draft);
+}
+
+/* Bring the handle to the newest published version: an unchanged store costs
+ * one system call (the mapping shows the head slots); 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};
+    }
+    if (store->head.seq) {
+        /* The next commit's record follows the head's bytes; a slot shows a
+         * newer head only for a commit of the earlier format. */
+        const unsigned char *map = store->mapping->map;
+        unsigned char next[16];
+        bool newer = decode64(map + 8) > store->head.seq || decode64(map + BLOCK + 8) > store->head.seq ||
+                     (read_at(store->fd, next, sizeof(next), store->head.end) && !memcmp(next, record_magic, 8) &&
+                      decode64(next + 8) == store->head.seq + 1);
+        if (!newer) return PURE_OK;
+    }
+    Head head;
+    if (!read_head(store->fd, &info, false, &head, NULL)) return PURE_STORE_ERROR;
+    if (store->mapping && head.seq == store->head.seq && head.check == store->head.check) return PURE_OK;
+    Tree tree;
+    Mapping *mapping = malloc(sizeof(*mapping));
+    if (!mapping) return PURE_ALLOCATION;
+    if (!map_tree(store->fd, &info, &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;
+    __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED);
+    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);
+}
+
+PureMapping *pure_draft_mapping(const PureDraft *draft) {
+    Mapping *mapping = draft->tree.shared;
+    if (mapping) __atomic_add_fetch(&mapping->refs, 1, __ATOMIC_RELAXED);
+    return mapping;
+}
+
+bool pure_mapping_holds(const PureMapping *mapping, PureSpan span) {
+    uintptr_t base = (uintptr_t)mapping->map, at = (uintptr_t)span.data;
+    return !span.size || (span.data && at >= base && at - base <= mapping->end && span.size <= mapping->end - (at - base));
+}
+
+void pure_mapping_release(PureMapping *mapping) { release_mapping(mapping); }
+
+PureResult pure_store_pin(PureStore *store, bool pinned) {
+    store->pinned = false;
+    PureResult result = pinned ? refresh(store) : PURE_OK;
+    store->pinned = pinned && result == PURE_OK;
+    return result;
+}
+
+static PureResult open_through(PureStore *store, const char *key, PureDraft **out) {
+    *out = NULL;
+    PureResult result = store->pinned ? PURE_OK : 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; }
+
+static size_t mark(const Tree *tree, Entry entry, unsigned char *marks) {
+    if (!(entry.ref & MEMORY_BIT)) return 0;
+    size_t at = (size_t)(entry.ref & LOCATION);
+    if (marks[at]) return 0;
+    marks[at] = 1;
+    const Node *node = &tree->nodes[at];
+    size_t bytes = node_bytes(node);
+    if (!node->bytes)
+        for (size_t i = 0; i < node->size; ++i) bytes += mark(tree, node->children[i], marks);
+    return bytes;
+}
+
+/* Every edit copies a leaf and path, so reclaim arena nodes the root no
+ * longer reaches once they outweigh the reachable ones (and 32 KiB). */
+static void collect(PureDraft *draft) {
+    Tree *tree = &draft->tree;
+    if (draft->held) return;
+    size_t allowance = draft->live > 32768 ? draft->live : 32768;
+    if (tree->bytes <= draft->live || tree->bytes - draft->live <= allowance) return;
+    unsigned char *marks = calloc(tree->count, 1);
+    if (!marks) return;
+    draft->live = mark(tree, draft->root, marks);
+    for (size_t i = 0; i < tree->count; ++i) {
+        Node *node = &tree->nodes[i];
+        if (marks[i] || node->height == FREE) continue;
+        tree->bytes -= node_bytes(node);
+        free(node->bytes);
+        free(node->children);
+        *node = (Node){.size = tree->free_head, .height = FREE};
+        tree->free_head = i + 1;
+        ++tree->free_count;
+    }
+    free(marks);
+}
+
+PureResult pure_draft_splice(PureDraft *draft, size_t offset, size_t remove, PureSpan insert) {
+    uint64_t size = draft->root.bytes;
+    if (insert.size && !insert.data) return PURE_INVALID;
+    if (offset > size || remove > size - offset || insert.size > PURE_OBJECT_LIMIT - (size - remove))
+        return PURE_LIMIT;
+    if (!remove && !insert.size) return PURE_OK;
+    Entry next;
+    bool ok = rebuild(&draft->tree, draft->root, offset, remove, insert, &next);
+    if (ok) draft->root = next;
+    collect(draft);
+    return ok ? PURE_OK : PURE_STORE_ERROR;
+}
+
+PureVersion pure_draft_version(const PureDraft *draft) {
+    return (PureVersion){draft->root.ref, draft->root.bytes};
+}
+
+PureResult pure_draft_read_version(const PureDraft *draft, PureVersion version, size_t offset,
+                                   size_t length, bool (*visit)(void *, PureSpan), void *context) {
+    Entry root = {version.ref, version.bytes};
+    unsigned height;
+    if (offset > root.bytes || length > root.bytes - offset) return PURE_LIMIT;
+    if (!length) return PURE_OK;
+    if (!root_height(&draft->tree, root, &height)) return PURE_STORE_ERROR;
+    return visit_range(&draft->tree, root, height, offset, length, visit, context) ?
+           PURE_OK : PURE_STORE_ERROR;
+}
+
+PureResult pure_draft_read(const PureDraft *draft, size_t offset, size_t length,
+                           bool (*visit)(void *, PureSpan), void *context) {
+    return pure_draft_read_version(draft, pure_draft_version(draft), offset, length, visit, context);
+}
+
+void pure_draft_restore(PureDraft *draft, PureVersion version) {
+    draft->root = (Entry){version.ref, version.bytes};
+}
+
+void pure_draft_hold(PureDraft *draft, bool held) {
+    draft->held = held;
+    if (!held) collect(draft);
+}
+
+/* ---- Group commits ------------------------------------------------------
+ * Threads that commit at the same time share one commit: they queue, the
+ * first becomes the leader and publishes every queued request of the same
+ * store with one head (one pair of syncs); the others wait for its result.
+ * Each request stays all or nothing on its own: a request with a stale draft
+ * is left out, and so is one whose key an earlier request of the same group
+ * already changes. The leader touches the waiters' drafts and handles only
+ * while they wait. */
+typedef struct Waiter {
+    PureDraft *const *drafts;
+    size_t count;
+    int64_t *revisions;
+    PureResult result;
+    bool accepted, done;
+    struct Waiter *next;
+} Waiter;
+
+static struct {
+    pthread_mutex_t lock;
+    pthread_cond_t finished, joined;
+    Waiter *head, *tail;
+    size_t queued, expected; /* expected: how many requests groups have held lately */
+    bool leading;
+    unsigned long groups, requests;
+} group = {PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER, PTHREAD_COND_INITIALIZER, NULL, NULL, 0, 1, false, 0, 0};
+
+void pure_store_group_counts(unsigned long *groups, unsigned long *requests) {
+    pthread_mutex_lock(&group.lock);
+    *groups = group.groups;
+    *requests = group.requests;
+    pthread_mutex_unlock(&group.lock);
+}
+
+enum { GATHER_NS = 2000000 };
+
+/* Publish the queued requests of the store at path in one session. */
+static void commit_group(const char *path, Waiter *first) {
+    size_t total = 0;
+    for (Waiter *w = first; w; w = w->next)
+        if (!strcmp(w->drafts[0]->path, path)) { total += w->count; w->accepted = false; }
+    Session session;
+    Writer *writer = NULL;
+    Entry *roots = malloc(total * sizeof(*roots));
+    const char **keys = malloc(total * sizeof(*keys));
+    int64_t *revisions = malloc(total * sizeof(*revisions));
+    PureDraft **drafts = malloc(total * sizeof(*drafts));
+    bool allocated = roots && keys && revisions && drafts, checked = false;
+    PureResult result = allocated ? open_session(path, &session) : PURE_ALLOCATION;
+    size_t count = 0;
+    if (result == PURE_OK) {
+        /* A request holds when each of its keys is at its draft's base
+         * revision and no earlier request of this group changes it. */
+        for (Waiter *w = first; w; w = w->next) {
+            if (strcmp(w->drafts[0]->path, path)) continue;
+            w->result = PURE_OK;
+            for (size_t i = 0; i < w->count && w->result == PURE_OK; ++i) {
+                PureDraft *d = w->drafts[i];
+                Entry current;
+                uint64_t index;
+                int64_t stored;
+                bool found;
+                if (!find(&session.tree, (Entry){session.head.catalog, session.head.catalog_bytes}, d->key, &index,
+                          &found, &current, &stored)) { w->result = PURE_STORE_ERROR; break; }
+                for (size_t k = 0; k < count; ++k)
+                    if (!strcmp(keys[k], d->key)) { ++stored; break; } /* it moves with the earlier request */
+                if (stored != d->revision) { w->revisions[i] = stored; w->result = PURE_CONFLICT; }
+                else if (stored == INT64_MAX) w->result = PURE_LIMIT;
+            }
+            if (w->result != PURE_OK) continue;
+            w->accepted = true;
+            for (size_t i = 0; i < w->count; ++i) {
+                drafts[count] = w->drafts[i];
+                roots[count] = w->drafts[i]->root;
+                keys[count] = w->drafts[i]->key;
+                revisions[count++] = w->drafts[i]->revision;
+            }
+        }
+        checked = true;
+        if (count) {
+            result = PURE_ALLOCATION;
+            if ((writer = malloc(sizeof(*writer)))) {
+                *writer = (Writer){.fd = session.fd, .at = commit_start(&session)};
+                bool written = true;
+                for (size_t i = 0; i < count && written; ++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;
+                    written = write_root(&drafts[i]->tree, writer, &roots[i], relocate);
+                }
+                result = written ? publish(&session, writer, count, keys, roots, revisions) : PURE_STORE_ERROR;
+            }
+        }
+        /* Rebase while still holding the writer lock; the arenas are no longer
+         * needed. Drafts of a store handle share one new mapping. If remapping
+         * fails a draft keeps equal content at its new revision. */
+        PureStore *refreshed[8] = {NULL};
+        size_t fresh = 0;
+        for (size_t i = 0; result == PURE_OK && i < count; ++i) {
+            PureStore *store = drafts[i]->store;
+            bool rebased;
+            if (store) {
+                bool seen = false;
+                for (size_t k = 0; k < fresh && !seen; ++k) seen = refreshed[k] == store;
+                if (!seen && refresh(store) == PURE_OK) {
+                    seen = true;
+                    if (fresh < 8) refreshed[fresh++] = store;
+                }
+                rebased = seen && rebase(drafts[i]) == PURE_OK;
+            } else rebased = load_draft(drafts[i]) == PURE_OK;
+            if (!rebased) drafts[i]->revision = revisions[i];
+        }
+    }
+    /* Before the checks every request shares the failure; after them an
+     * accepted request takes the publication's result and new revisions, and
+     * a refused one keeps its own result. */
+    size_t at = 0;
+    for (Waiter *w = first; w; w = w->next) {
+        if (strcmp(w->drafts[0]->path, path)) continue;
+        if (!checked) { w->result = result; continue; }
+        if (!w->accepted) continue;
+        for (size_t i = 0; i < w->count; ++i) w->revisions[i] = revisions[at++];
+        w->result = result;
+    }
+    free(writer);
+    free(roots);
+    free((void *)keys);
+    free(revisions);
+    free(drafts);
+    if (allocated) close_session(&session);
+}
+
+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;
+    Waiter me = {.drafts = drafts, .count = count, .revisions = revisions};
+    pthread_mutex_lock(&group.lock);
+    if (group.tail) group.tail->next = &me; else group.head = &me;
+    group.tail = &me;
+    ++group.queued;
+    pthread_cond_signal(&group.joined);
+    while (!me.done) {
+        if (group.leading) { pthread_cond_wait(&group.finished, &group.lock); continue; }
+        group.leading = true;
+        /* When several threads have been committing together, give the others
+         * a moment (at most 2 ms) to join, or they would miss this group and
+         * wait for the next one: a sync costs far more. One thread never waits. */
+        if (group.expected > 1 && group.queued < group.expected) {
+            struct timespec deadline;
+            clock_gettime(CLOCK_REALTIME, &deadline);
+            deadline.tv_nsec += GATHER_NS;
+            if (deadline.tv_nsec >= 1000000000) { ++deadline.tv_sec; deadline.tv_nsec -= 1000000000; }
+            while (group.queued < group.expected &&
+                   pthread_cond_timedwait(&group.joined, &group.lock, &deadline) == 0) {}
+        }
+        Waiter *batch = group.head;
+        size_t size = group.queued;
+        group.head = group.tail = NULL;
+        group.queued = 0;
+        ++group.groups;
+        group.requests += size;
+        /* Remember the largest recent group, forgetting one member per group. */
+        group.expected = size > group.expected - 1 ? size : group.expected - 1;
+        if (!group.expected) group.expected = 1;
+        pthread_mutex_unlock(&group.lock);
+        /* One session per store among the queued requests. */
+        for (Waiter *w = batch; w; w = w->next) {
+            bool earlier = false;
+            for (Waiter *v = batch; v != w && !earlier; v = v->next) earlier = !strcmp(v->drafts[0]->path, w->drafts[0]->path);
+            if (!earlier) commit_group(w->drafts[0]->path, batch);
+        }
+        pthread_mutex_lock(&group.lock);
+        for (Waiter *w = batch, *next; w; w = next) { next = w->next; w->done = true; }
+        group.leading = false;
+        pthread_cond_broadcast(&group.finished);
+    }
+    pthread_mutex_unlock(&group.lock);
+    return me.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);
+}
+
+PureResult pure_store_splice(const char *path, const char *key, int64_t expected,
+                             size_t offset, size_t remove, PureSpan insert,
+                             size_t max_size, int64_t *revision) {
+    return mutate(path, key, expected, insert, true, offset, remove, max_size, revision);
+}
+
+#ifdef PURE_STORE_TEST
+static bool shape(const Tree *tree, Entry entry, unsigned height, bool root,
+                  uint64_t *leaves, uint64_t *small) {
+    if (!height) {
+        ++*leaves;
+        if (!root && entry.bytes < LEAF_MIN) ++*small;
+        return leaf_bytes(tree, entry) != NULL;
+    }
+    Entry kids[FAN_MAX];
+    size_t count = children(tree, entry, height, kids);
+    if (!count || (root && count < 2)) return false;
+    if (!root && count < FAN_MIN) ++*small;
+    for (size_t i = 0; i < count; ++i)
+        if (!shape(tree, kids[i], height - 1, false, leaves, small)) return false;
+    return true;
+}
+
+/* Test-only: live arena nodes and their payload bytes. */
+void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes) {
+    *nodes = *bytes = 0;
+    for (size_t i = 0; i < draft->tree.count; ++i) {
+        const Node *node = &draft->tree.nodes[i];
+        if (node->height == FREE) continue;
+        ++*nodes;
+        *bytes += node->bytes ? node->size : node->size * sizeof(Entry);
+    }
+}
+
+/* Test-only: the stored height, leaf count and number of underfull nodes. */
+bool pure_tree_test_shape(const char *path, const char *key, unsigned *height,
+                          uint64_t *leaves, uint64_t *small, uint64_t *file_end) {
+    int directory = open(path, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
+    int fd = directory >= 0 ? openat(directory, tree_name, O_RDONLY | O_CLOEXEC) : -1;
+    if (directory >= 0) close(directory);
+    Tree tree = {0};
+    Head head;
+    struct stat file;
+    Entry document;
+    uint64_t index;
+    int64_t revision;
+    bool found = false;
+    *leaves = *small = 0;
+    bool ok = fd >= 0 && read_head(fd, &file, false, &head, NULL) && map_tree(fd, &file, &head, &tree) &&
+              find(&tree, (Entry){head.catalog, head.catalog_bytes}, key, &index, &found,
+                   &document, &revision) && found && root_height(&tree, document, height) &&
+              (!document.bytes || shape(&tree, document, *height, true, leaves, small));
+    if (ok) *file_end = head.end;
+    release_tree(&tree);
+    if (fd >= 0) close(fd);
+    return ok;
+}
+#endif
diff --git a/store/test-tree-store.c b/store/test-tree-store.c
new file mode 100644
index 0000000..1130acd
--- /dev/null
+++ b/store/test-tree-store.c
@@ -0,0 +1,1131 @@
+#define _POSIX_C_SOURCE 200809L
+#include "store.h"
+#include "state.h"
+#include <assert.h>
+#include <dirent.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/stat.h>
+#include <pthread.h>
+#include <sys/wait.h>
+#include <time.h>
+#include <unistd.h>
+
+#define CHECK(x) do { if (!(x)) { fprintf(stderr, "%s:%d: %s\n", __FILE__, __LINE__, #x); abort(); } } while (0)
+
+static int fail_stage = -1;
+static int crash;
+static int slow_once; /* the next publication pauses before its data sync */
+static int held_sync;  /* 1: the next publication waits after its head write; 2: it is waiting */
+static void pause_ms(long ms) {
+    struct timespec wait = {ms / 1000, (ms % 1000) * 1000000};
+    nanosleep(&wait, NULL);
+}
+int pure_store_test_fault(PureStoreStage stage) {
+    if (stage == PURE_BEFORE_FILE_SYNC && __atomic_exchange_n(&slow_once, 0, __ATOMIC_SEQ_CST)) pause_ms(100);
+    int one = 1;
+    if (stage == PURE_BEFORE_DIR_SYNC &&
+        __atomic_compare_exchange_n(&held_sync, &one, 2, false, __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST))
+        while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) == 2) pause_ms(1);
+    if ((int)stage != fail_stage) return 0;
+    if (crash) _exit(77);
+    return 1;
+}
+bool pure_tree_test_shape(const char *path, const char *key, unsigned *height,
+                          uint64_t *leaves, uint64_t *small, uint64_t *file_end);
+
+void pure_draft_test_arena(const PureDraft *draft, size_t *nodes, size_t *bytes);
+unsigned long pure_tree_test_publications(void);
+unsigned long pure_tree_test_syncs(void);
+void pure_tree_test_forget(void);
+
+/* The file: head slots in blocks 0 and 1 (sequence at +8, end at +16,
+ * catalog at +24), commits from DATA_START, each a 128-byte record (its head)
+ * and then its bytes. */
+enum { KEYS = 4, CAPACITY = 1 << 20, SLOT_BLOCK = 4096, DATA_START = 8192, RECORD_SIZE = 128 };
+typedef struct { unsigned char *bytes; size_t size; int64_t revision; } Oracle;
+
+static uint32_t random_value(uint32_t *state) {
+    *state ^= *state << 13;
+    *state ^= *state >> 17;
+    *state ^= *state << 5;
+    return *state;
+}
+
+static void expect(const char *path, const char *key, const Oracle *o) {
+    PureStoredView view;
+    CHECK(pure_store_get(path, key, PURE_OBJECT_LIMIT, &view) == PURE_OK);
+    CHECK(view.revision == o->revision && view.body.size == o->size);
+    CHECK(!o->size || !memcmp(view.body.data, o->bytes, o->size));
+    pure_store_release(&view);
+}
+
+static uint64_t file_size(const char *path) {
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    struct stat info;
+    CHECK(stat(name, &info) == 0);
+    return (uint64_t)info.st_size;
+}
+
+/* Where the newest commit ends. */
+static uint64_t published_end(const char *path, const char *key) {
+    unsigned height;
+    uint64_t leaves, small, end;
+    CHECK(pure_tree_test_shape(path, key, &height, &leaves, &small, &end));
+    return end;
+}
+
+static uint64_t field_at(int fd, uint64_t at) {
+    unsigned char bytes[8];
+    CHECK(pread(fd, bytes, 8, (off_t)at) == 8);
+    uint64_t value = 0;
+    for (unsigned i = 0; i < 8; ++i) value |= (uint64_t)bytes[i] << (8 * i);
+    return value;
+}
+
+static uint64_t slot_field(int fd, unsigned slot, unsigned offset) { return field_at(fd, slot * SLOT_BLOCK + offset); }
+
+/* Where the newest head is: the newer slot's, or the last record after it. */
+static uint64_t newest_head(int fd) {
+    uint64_t at = slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? 0 : SLOT_BLOCK;
+    for (;;) {
+        uint64_t end = field_at(fd, at + 16);
+        unsigned char magic[8];
+        if (pread(fd, magic, 8, (off_t)end) != 8 || memcmp(magic, "PUREREC1", 8) ||
+            field_at(fd, end + 8) != field_at(fd, at + 8) + 1) return at;
+        at = end;
+    }
+}
+
+static void fill(unsigned char *bytes, size_t size, uint32_t *state) {
+    for (size_t i = 0; i < size; ++i) bytes[i] = (unsigned char)random_value(state);
+}
+
+/* Random edits against a flat oracle, including views held across later
+ * commits and compactions. Every stored node must stay balanced. */
+static void oracle(const char *path) {
+    static const char *keys[KEYS] = {"alpha", "b", "gamma-long_key-0123456789", "z"};
+    Oracle docs[KEYS] = {{0}};
+    unsigned char *insert = malloc(65536);
+    for (unsigned k = 0; k < KEYS; ++k) CHECK((docs[k].bytes = malloc(CAPACITY)));
+    CHECK(insert);
+    uint32_t state = 0x9e3779b9u;
+    PureStoredView held = {0};
+    unsigned char *held_copy = malloc(CAPACITY);
+    size_t held_size = 0;
+    unsigned max_height = 0;
+    size_t max_size = 0, splices = 0;
+    for (unsigned step = 0; step < 6000; ++step) {
+        unsigned k = random_value(&state) % KEYS;
+        Oracle *o = &docs[k];
+        uint32_t kind = random_value(&state) % 100;
+        size_t count = kind < 70 ? random_value(&state) % 24 :
+                       kind < 95 ? random_value(&state) % 6000 : random_value(&state) % 65536;
+        int64_t revision;
+        fill(insert, count, &state);
+        if (kind >= 98 || !o->revision) {
+            CHECK(pure_store_put(path, keys[k], o->revision, (PureSpan){insert, count}, &revision) == PURE_OK);
+            memcpy(o->bytes, insert, count);
+            o->size = count;
+        } else {
+            size_t offset = o->size ? random_value(&state) % (o->size + 1) : 0;
+            size_t remove = o->size - offset ? random_value(&state) % (o->size - offset + 1) : 0;
+            if (random_value(&state) % 24) remove = remove % 64;
+            if (o->size - remove + count > CAPACITY) { count = 0; remove = o->size - offset; }
+            CHECK(pure_store_splice(path, keys[k], o->revision, offset, remove,
+                                    (PureSpan){insert, count}, PURE_OBJECT_LIMIT, &revision) == PURE_OK);
+            memmove(o->bytes + offset + count, o->bytes + offset + remove, o->size - offset - remove);
+            memcpy(o->bytes + offset, insert, count);
+            o->size = o->size - remove + count;
+            ++splices;
+        }
+        if (o->size > max_size) max_size = o->size;
+        CHECK(revision == o->revision + 1);
+        o->revision = revision;
+        expect(path, keys[k], o);
+        unsigned height;
+        uint64_t leaves, small, end;
+        CHECK(pure_tree_test_shape(path, keys[k], &height, &leaves, &small, &end));
+        CHECK(small == 0);
+        if (height > max_height) max_height = height;
+        if (step % 97 == 0) {
+            if (held.mapped_size) {
+                CHECK(held.body.size == held_size && !memcmp(held.body.data, held_copy, held_size));
+                pure_store_release(&held);
+            }
+            CHECK(pure_store_get(path, keys[k], PURE_OBJECT_LIMIT, &held) == PURE_OK);
+            held_size = held.body.size;
+            if (held_size) memcpy(held_copy, held.body.data, held_size);
+        }
+    }
+    if (held.mapped_size) {
+        CHECK(held.body.size == held_size && !memcmp(held.body.data, held_copy, held_size));
+        pure_store_release(&held);
+    }
+    for (unsigned k = 0; k < KEYS; ++k) { expect(path, keys[k], &docs[k]); free(docs[k].bytes); }
+    CHECK(max_height >= 2 && max_height <= 4 && max_size > 262144);
+    printf("{\"oracle_steps\":6000,\"splices\":%zu,\"max_document_bytes\":%zu,\"max_height\":%u}\n",
+           splices, max_size, max_height);
+    free(insert);
+    free(held_copy);
+}
+
+/* A one-byte insert, delete or replacement in a 1 MiB document appends only
+ * its changed leaf and path, not another copy of the document. */
+static void amplification(const char *path) {
+    size_t size = 1048576;
+    unsigned char *bytes = malloc(size + 1);
+    uint32_t state = 7;
+    CHECK(bytes);
+    fill(bytes, size, &state);
+    int64_t revision;
+    CHECK(pure_store_put(path, "large", 0, (PureSpan){bytes, size}, &revision) == PURE_OK);
+    Oracle o = {bytes, size, revision};
+    unsigned char byte = 'Q';
+    const size_t edits[][3] = {{size / 2, 0, 1}, {size / 3, 1, 0}, {size - 1, 1, 1}, {0, 0, 1}};
+    uint64_t worst = 0;
+    for (size_t i = 0; i < sizeof(edits) / sizeof(*edits); ++i) {
+        uint64_t before = published_end(path, "large");
+        size_t offset = edits[i][0], remove = edits[i][1], count = edits[i][2];
+        CHECK(pure_store_splice(path, "large", o.revision, offset, remove, (PureSpan){&byte, count},
+                                PURE_OBJECT_LIMIT, &revision) == PURE_OK);
+        memmove(bytes + offset + count, bytes + offset + remove, o.size - offset - remove);
+        memcpy(bytes + offset, &byte, count);
+        o.size = o.size - remove + count;
+        o.revision = revision;
+        uint64_t after = published_end(path, "large");
+        if (after > before && after - before > worst) worst = after - before;
+        expect(path, "large", &o);
+    }
+    /* Compaction may shrink the file; growth per edit is the claim. */
+    CHECK(worst > 0 && worst < 16384);
+    unsigned height;
+    uint64_t leaves, small, end;
+    CHECK(pure_tree_test_shape(path, "large", &height, &leaves, &small, &end));
+    CHECK(small == 0 && height == 2 && leaves >= 256 && leaves <= 512);
+    printf("{\"one_byte_edit_max_appended_bytes\":%llu,\"height\":%u,\"leaves\":%llu}\n",
+           (unsigned long long)worst, height, (unsigned long long)leaves);
+    free(bytes);
+}
+
+static void faults(const char *path) {
+    unsigned char text[] = "fault-free";
+    int64_t revision;
+    CHECK(pure_store_put(path, "fault", 0, (PureSpan){text, 10}, &revision) == PURE_OK);
+    Oracle o = {text, 10, revision};
+    uint64_t size = file_size(path), end = published_end(path, "fault");
+    for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; ++fail_stage) {
+        CHECK(pure_store_splice(path, "fault", o.revision, 1, 3, (PureSpan){text, 5},
+                                PURE_OBJECT_LIMIT, &revision) == PURE_STORE_ERROR);
+        expect(path, "fault", &o);
+        /* The bytes stayed unpublished, in the written zeros after the end. */
+        CHECK(published_end(path, "fault") == end && file_size(path) == size);
+    }
+    fail_stage = PURE_BEFORE_DIR_SYNC;
+    CHECK(pure_store_splice(path, "fault", o.revision, 10, 0, (PureSpan){text, 2},
+                            PURE_OBJECT_LIMIT, &revision) == PURE_UNCERTAIN);
+    fail_stage = -1;
+    unsigned char next[] = "fault-freefa";
+    o = (Oracle){next, 12, o.revision + 1};
+    CHECK(revision == o.revision);
+    expect(path, "fault", &o);
+}
+
+/* A process that dies at each publication stage leaves either the previous
+ * or the next complete version; the next commit overwrites its unpublished
+ * bytes. A head it wrote but did not sync is made durable before the next
+ * writer builds on it. */
+static void crashes(const char *path) {
+    unsigned char text[] = "crash-base";
+    int64_t revision;
+    CHECK(pure_store_put(path, "crash", 0, (PureSpan){text, 10}, &revision) == PURE_OK);
+    Oracle o = {text, 10, revision};
+    for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) {
+        uint64_t before = published_end(path, "crash");
+        pid_t child = fork();
+        CHECK(child >= 0);
+        if (!child) {
+            crash = 1;
+            fail_stage = stage;
+            pure_store_splice(path, "crash", o.revision, 0, 1, (PureSpan){(const unsigned char *)"C", 1},
+                              PURE_OBJECT_LIMIT, &revision);
+            _exit(0);
+        }
+        int status;
+        CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77);
+        if (stage == PURE_BEFORE_DIR_SYNC) { text[0] = 'C'; ++o.revision; }
+        expect(path, "crash", &o);
+        uint64_t published = published_end(path, "crash");
+        CHECK(stage == PURE_BEFORE_DIR_SYNC ? published > before : published == before);
+        unsigned long syncs = pure_tree_test_syncs();
+        CHECK(pure_store_splice(path, "crash", o.revision, 10, 0, (PureSpan){(const unsigned char *)"", 0},
+                                PURE_OBJECT_LIMIT, &revision) == PURE_OK);
+        o.revision = revision;
+        expect(path, "crash", &o);
+        CHECK(pure_tree_test_syncs() - syncs == (stage == PURE_BEFORE_DIR_SYNC ? 2u : 1u));
+        CHECK(published_end(path, "crash") > published);
+    }
+}
+
+static void tear(const char *path, uint64_t at) {
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    int fd = open(name, O_RDWR);
+    unsigned char torn;
+    CHECK(fd >= 0 && pread(fd, &torn, 1, (off_t)at) == 1);
+    torn ^= 0x5a;
+    CHECK(pwrite(fd, &torn, 1, (off_t)at) == 1 && close(fd) == 0);
+}
+
+/* A torn newest record exposes the previous commit to a new process, and the
+ * next writer overwrites that record and the commit's bytes. A torn slot
+ * costs only a longer walk: the other slot's head leads along the records. */
+static void torn_head(const char *path) {
+    unsigned char first[] = "first", second[] = "second";
+    int64_t revision;
+    CHECK(pure_store_put(path, "torn", 0, (PureSpan){first, 5}, &revision) == PURE_OK);
+    Oracle o = {first, 5, revision};
+    uint64_t record = published_end(path, "torn"); /* where the next commit's record goes */
+    CHECK(pure_store_put(path, "torn", o.revision, (PureSpan){second, 6}, &revision) == PURE_OK);
+    tear(path, SLOT_BLOCK + 20);
+    pure_tree_test_forget();
+    expect(path, "torn", &(Oracle){second, 6, revision});
+    tear(path, record + 20);
+    pure_tree_test_forget();
+    expect(path, "torn", &o);
+    CHECK(pure_store_put(path, "torn", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT);
+    CHECK(pure_store_put(path, "torn", o.revision, (PureSpan){second, 3}, &revision) == PURE_OK);
+    o = (Oracle){second, 3, revision};
+    CHECK(published_end(path, "torn") < record + RECORD_SIZE + 4096); /* it went where the torn one was */
+    pure_tree_test_forget();
+    expect(path, "torn", &o);
+}
+
+/* Commits are found along their records: from a slot's head, which every
+ * CHECKPOINT commits moves on, by a new process too; a damaged record ends
+ * the walk at the commit before it, which the next writer builds on. */
+static void records(const char *path) {
+    int64_t revision;
+    uint64_t ends[41];
+    char text[16];
+    CHECK(pure_store_put(path, "walk", 0, (PureSpan){(const unsigned char *)"0", 1}, &revision) == PURE_OK);
+    for (unsigned i = 1; i <= 40; ++i) {
+        ends[i] = published_end(path, "walk");
+        snprintf(text, sizeof(text), "%u", i);
+        CHECK(pure_store_put(path, "walk", revision, (PureSpan){(const unsigned char *)text, strlen(text)},
+                             &revision) == PURE_OK);
+    }
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    int fd = open(name, O_RDONLY);
+    CHECK(fd >= 0);
+    uint64_t newest_slot = slot_field(fd, 0, 8) > slot_field(fd, 1, 8) ? slot_field(fd, 0, 8) : slot_field(fd, 1, 8);
+    CHECK(close(fd) == 0);
+    /* 42 commits (the empty file's first): heads in slots at 17 and 33. */
+    CHECK(newest_slot == 33);
+    pure_tree_test_forget();
+    expect(path, "walk", &(Oracle){(unsigned char *)"40", 2, 41});
+    pid_t child = fork();
+    CHECK(child >= 0);
+    if (!child) {
+        pure_tree_test_forget(); /* a process that never saw them */
+        PureStoredView view;
+        _exit(pure_store_get(path, "walk", 16, &view) == PURE_OK && view.body.size == 2 &&
+              !memcmp(view.body.data, "40", 2) && view.revision == 41 ? 0 : 1);
+    }
+    int status;
+    CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0);
+    /* Commit 37 ("37", revision 38) has a damaged record: the walk ends at 36. */
+    tear(path, ends[37] + 30);
+    pure_tree_test_forget();
+    expect(path, "walk", &(Oracle){(unsigned char *)"36", 2, 37});
+    CHECK(pure_store_put(path, "walk", 37, (PureSpan){(const unsigned char *)"x", 1}, &revision) == PURE_OK);
+    CHECK(revision == 38 && published_end(path, "walk") < ends[38]);
+    pure_tree_test_forget();
+    expect(path, "walk", &(Oracle){(unsigned char *)"x", 1, 38});
+}
+
+/* A reader's view stays byte-identical while another process commits and
+ * compacts; readers take no lock. */
+static void isolation(const char *path) {
+    size_t size = 300000;
+    unsigned char *bytes = malloc(size);
+    uint32_t state = 99;
+    CHECK(bytes);
+    fill(bytes, size, &state);
+    int64_t revision;
+    CHECK(pure_store_put(path, "shared", 0, (PureSpan){bytes, size}, &revision) == PURE_OK);
+    PureStoredView view;
+    CHECK(pure_store_get(path, "shared", size, &view) == PURE_OK);
+    uint64_t before = file_size(path);
+    pid_t child = fork();
+    CHECK(child >= 0);
+    if (!child) {
+        int64_t at = revision;
+        unsigned char byte = 'x';
+        for (unsigned i = 0; i < 400; ++i)
+            if (pure_store_splice(path, "shared", at, (i * 7919u) % size, 1, (PureSpan){&byte, 1},
+                                  PURE_OBJECT_LIMIT, &at) != PURE_OK) _exit(1);
+        _exit(0);
+    }
+    int status;
+    CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0);
+    CHECK(view.revision == revision && view.body.size == size && !memcmp(view.body.data, bytes, size));
+    pure_store_release(&view);
+    PureStoredView latest;
+    CHECK(pure_store_get(path, "shared", size, &latest) == PURE_OK && latest.revision == revision + 400);
+    pure_store_release(&latest);
+    /* About 2 MB of path copies without compaction would exceed this. */
+    CHECK(file_size(path) < before + 3 * 524288);
+    free(bytes);
+}
+
+/* Damaged branch bytes are reported as storage errors, never trusted. */
+static void corruption(const char *path) {
+    unsigned char *bytes = malloc(200000);
+    uint32_t state = 3;
+    CHECK(bytes);
+    fill(bytes, 200000, &state);
+    int64_t revision;
+    CHECK(pure_store_put(path, "damaged", 0, (PureSpan){bytes, 200000}, &revision) == PURE_OK);
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    uint64_t end = published_end(path, "damaged");
+    int fd = open(name, O_RDWR);
+    CHECK(fd >= 0);
+    unsigned errors = 0;
+    for (unsigned i = 0; i < 3000; ++i) {
+        off_t at = (off_t)(DATA_START + random_value(&state) % (end - DATA_START));
+        unsigned char original, damaged;
+        CHECK(pread(fd, &original, 1, at) == 1);
+        damaged = original ^ (unsigned char)(1u << (random_value(&state) % 8));
+        CHECK(pwrite(fd, &damaged, 1, at) == 1);
+        PureStoredView view;
+        PureResult result = pure_store_get(path, "damaged", PURE_OBJECT_LIMIT, &view);
+        CHECK(result == PURE_OK || result == PURE_STORE_ERROR || result == PURE_NOT_FOUND);
+        if (result == PURE_OK) pure_store_release(&view); else ++errors;
+        CHECK(pwrite(fd, &original, 1, at) == 1);
+    }
+    CHECK(close(fd) == 0 && errors > 0);
+    free(bytes);
+}
+
+static void many_keys(const char *path) {
+    int64_t revision;
+    char key[32];
+    for (unsigned i = 0; i < 1024; ++i) {
+        unsigned id = (i * 617u) % 1024u;
+        snprintf(key, sizeof(key), "doc-%04u", id);
+        CHECK(pure_store_put(path, key, 0, (PureSpan){(const unsigned char *)key, strlen(key)}, &revision) == PURE_OK);
+    }
+    for (unsigned id = 0; id < 1024; ++id) {
+        snprintf(key, sizeof(key), "doc-%04u", id);
+        Oracle o = {(unsigned char *)key, strlen(key), 1};
+        expect(path, key, &o);
+    }
+    PureStoredView view;
+    CHECK(pure_store_get(path, "doc-1024", 64, &view) == PURE_NOT_FOUND);
+    CHECK(pure_store_get(path, "doc-0001", 3, &view) == PURE_BUDGET);
+    CHECK(pure_store_splice(path, "doc-0002", 1, 9, 0, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_LIMIT);
+    CHECK(pure_store_put(path, "doc-0003", 0, (PureSpan){0}, &revision) == PURE_CONFLICT && revision == 1);
+}
+
+/* The newest catalog of a one-key store is one leaf; forge its revision. */
+static void revision_limit(const char *path) {
+    int64_t revision;
+    CHECK(pure_store_put(path, "work", 0, (PureSpan){(const unsigned char *)"w", 1}, &revision) == PURE_OK);
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    int fd = open(name, O_RDWR);
+    CHECK(fd >= 0);
+    uint64_t catalog = field_at(fd, newest_head(fd) + 24);
+    CHECK(catalog >> 63);
+    const unsigned char maximum[] = {255,255,255,255,255,255,255,127};
+    off_t record = (off_t)(catalog & ((UINT64_C(1) << 62) - 1));
+    CHECK(pwrite(fd, maximum, 8, record + 80) == 8 && close(fd) == 0);
+    /* The forged commit no longer matches its digest; this process wrote it
+     * and does not read it again, a new process would pass over it. */
+    PureStoredView view;
+    CHECK(pure_store_get(path, "work", 1, &view) == PURE_OK && view.revision == INT64_MAX);
+    pure_store_release(&view);
+    CHECK(pure_store_splice(path, "work", INT64_MAX, 0, 0, (PureSpan){0}, PURE_OBJECT_LIMIT, &revision) == PURE_LIMIT);
+}
+
+static bool count_key(void *context, const char *key) {
+    (void)key;
+    ++*(unsigned *)context;
+    return true;
+}
+
+static void keys(const char *path) {
+    unsigned count = 0;
+    CHECK(pure_store_keys(path, count_key, &count) == PURE_OK && count == 0);
+    int64_t revision;
+    const char *names[] = {"m", "a", "zz", "a-b", "A"};
+    for (unsigned i = 0; i < 5; ++i)
+        CHECK(pure_store_put(path, names[i], 0, (PureSpan){0}, &revision) == PURE_OK);
+    CHECK(pure_store_keys(path, count_key, &count) == PURE_OK && count == 5);
+}
+
+typedef struct { unsigned char *out; size_t spans, largest, limit; } Gather;
+
+static bool gather(void *context, PureSpan span) {
+    Gather *g = context;
+    memcpy(g->out, span.data, span.size);
+    g->out += span.size;
+    if (span.size > g->largest) g->largest = span.size;
+    return ++g->spans != g->limit;
+}
+
+static void expect_draft(const PureDraft *draft, const Oracle *o, unsigned char *scratch) {
+    Gather g = {scratch, 0, 0, 0};
+    CHECK(pure_draft_size(draft) == o->size && pure_draft_revision(draft) == o->revision);
+    CHECK(pure_draft_read(draft, 0, o->size, gather, &g) == PURE_OK);
+    CHECK((size_t)(g.out - scratch) == o->size && (!o->size || !memcmp(scratch, o->bytes, o->size)));
+    CHECK(g.largest <= 4096);
+}
+
+static void apply(Oracle *o, size_t offset, size_t remove, const unsigned char *insert, size_t count) {
+    memmove(o->bytes + offset + count, o->bytes + offset + remove, o->size - offset - remove);
+    if (count) memcpy(o->bytes + offset, insert, count);
+    o->size = o->size - remove + count;
+}
+
+/* Edits live only in the draft until commit; each commit continues from the
+ * published version with an empty arena. */
+static void drafts(const char *path) {
+    Oracle o = {malloc(CAPACITY), 0, 0};
+    unsigned char *scratch = malloc(CAPACITY), *insert = malloc(65536);
+    CHECK(o.bytes && scratch && insert);
+    PureDraft *draft;
+    PureStoredView view;
+    CHECK(pure_draft_open(path, "draft", &draft) == PURE_OK);
+    expect_draft(draft, &o, scratch);
+    uint32_t state = 0x1234567u;
+    for (unsigned step = 0; step < 20000; ++step) {
+        uint32_t kind = random_value(&state) % 100;
+        size_t count = kind < 80 ? random_value(&state) % 8 :
+                       kind < 98 ? random_value(&state) % 4000 : random_value(&state) % 65536;
+        size_t offset = o.size ? random_value(&state) % (o.size + 1) : 0;
+        size_t remove = o.size - offset ? random_value(&state) % (o.size - offset + 1) : 0;
+        if (random_value(&state) % 64) remove %= 16;
+        if (o.size - remove + count > CAPACITY) count = 0;
+        fill(insert, count, &state);
+        CHECK(pure_draft_splice(draft, offset, remove, (PureSpan){insert, count}) == PURE_OK);
+        apply(&o, offset, remove, insert, count);
+        if (step % 97 == 0) expect_draft(draft, &o, scratch);
+        if (step % 1000 == 999) {
+            if (o.revision) CHECK(pure_store_get(path, "draft", PURE_OBJECT_LIMIT, &view) == PURE_OK &&
+                                  view.revision == o.revision);
+            if (o.revision) pure_store_release(&view);
+            int64_t revision;
+            CHECK(pure_draft_commit(draft, &revision) == PURE_OK && revision == o.revision + 1);
+            o.revision = revision;
+            expect(path, "draft", &o);
+            expect_draft(draft, &o, scratch);
+            size_t nodes, bytes;
+            pure_draft_test_arena(draft, &nodes, &bytes);
+            CHECK(nodes == 0);
+        }
+    }
+    CHECK(o.size > 100000);
+    Gather g = {scratch, 0, 0, 3};
+    CHECK(pure_draft_read(draft, 0, o.size, gather, &g) == PURE_OK && g.spans == 3);
+    CHECK(pure_draft_splice(draft, o.size + 1, 0, (PureSpan){0}) == PURE_LIMIT);
+    CHECK(pure_draft_splice(draft, 0, o.size + 1, (PureSpan){0}) == PURE_LIMIT);
+    CHECK(pure_draft_read(draft, o.size, 1, gather, &g) == PURE_LIMIT);
+    expect_draft(draft, &o, scratch);
+    pure_draft_close(draft);
+    printf("{\"draft_steps\":20000,\"draft_commits\":%lld,\"draft_bytes\":%zu}\n",
+           (long long)o.revision, o.size);
+    free(o.bytes);
+    free(scratch);
+    free(insert);
+}
+
+/* Long typing and deleting at one place keeps only the live path in memory,
+ * and its commit appends the typed bytes and path, not the document. */
+static void typing(const char *path) {
+    size_t size = 1048576;
+    Oracle o = {malloc(size + 40000), size, 0};
+    unsigned char *scratch = malloc(size + 40000);
+    uint32_t state = 5;
+    CHECK(o.bytes && scratch);
+    fill(o.bytes, size, &state);
+    CHECK(pure_store_put(path, "typed", 0, (PureSpan){o.bytes, size}, &o.revision) == PURE_OK);
+    PureDraft *draft;
+    CHECK(pure_draft_open(path, "typed", &draft) == PURE_OK);
+    size_t cursor = size / 2, nodes, bytes, peak = 0;
+    for (unsigned i = 0; i < 30000; ++i) {
+        unsigned char byte = (unsigned char)('a' + i % 26);
+        if (i % 3 == 2) {
+            CHECK(pure_draft_splice(draft, cursor - 1, 1, (PureSpan){0}) == PURE_OK);
+            apply(&o, --cursor, 1, NULL, 0);
+        } else {
+            CHECK(pure_draft_splice(draft, cursor, 0, (PureSpan){&byte, 1}) == PURE_OK);
+            apply(&o, cursor++, 0, &byte, 1);
+        }
+        pure_draft_test_arena(draft, &nodes, &bytes);
+        if (bytes > peak) peak = bytes;
+    }
+    expect_draft(draft, &o, scratch);
+    /* Reachable: 10,000 net typed bytes, the leaves they landed in and one
+     * path. Unreachable copies are reclaimed past max(reachable, 32 KiB). */
+    CHECK(peak < 2 * (10000 + 4 * 4096 + 2048) + 32768 + 8192);
+    uint64_t before = published_end(path, "typed");
+    int64_t revision;
+    CHECK(pure_draft_commit(draft, &revision) == PURE_OK);
+    o.revision = revision;
+    uint64_t appended = published_end(path, "typed") - before;
+    CHECK(appended < 10000 + 16384);
+    expect(path, "typed", &o);
+    pure_draft_close(draft);
+    printf("{\"typing_keystrokes\":30000,\"peak_arena_bytes\":%zu,\"commit_appended_bytes\":%llu}\n",
+           peak, (unsigned long long)appended);
+    free(o.bytes);
+    free(scratch);
+}
+
+/* Two drafts of one key: the second commit conflicts and keeps its edits. */
+static void draft_conflict(const char *path) {
+    unsigned char scratch[64];
+    int64_t revision;
+    CHECK(pure_store_put(path, "shared", 0, (PureSpan){(const unsigned char *)"base", 4}, &revision) == PURE_OK);
+    PureDraft *first, *second;
+    CHECK(pure_draft_open(path, "shared", &first) == PURE_OK && pure_draft_open(path, "shared", &second) == PURE_OK);
+    CHECK(pure_draft_splice(first, 4, 0, (PureSpan){(const unsigned char *)"-one", 4}) == PURE_OK);
+    CHECK(pure_draft_splice(second, 0, 0, (PureSpan){(const unsigned char *)"two-", 4}) == PURE_OK);
+    CHECK(pure_draft_commit(first, &revision) == PURE_OK && revision == 2);
+    CHECK(pure_draft_commit(second, &revision) == PURE_CONFLICT && revision == 2);
+    Oracle kept = {(unsigned char *)"two-base", 8, 1};
+    expect_draft(second, &kept, scratch);
+    Oracle stored = {(unsigned char *)"base-one", 8, 2};
+    expect(path, "shared", &stored);
+    pure_draft_close(second);
+    CHECK(pure_draft_open(path, "shared", &second) == PURE_OK);
+    expect_draft(second, &stored, scratch);
+    CHECK(pure_store_put(path, "shared", 2, (PureSpan){(const unsigned char *)"x", 1}, &revision) == PURE_OK);
+    CHECK(pure_draft_commit(first, &revision) == PURE_CONFLICT && revision == 3);
+    pure_draft_close(first);
+    pure_draft_close(second);
+}
+
+/* Another key's writes compact the file under an open draft; the draft's
+ * stored extents are copied from its old mapping on commit. */
+static void draft_relocation(const char *path) {
+    unsigned char *big = malloc(300000), scratch[20000];
+    uint32_t state = 11;
+    CHECK(big);
+    fill(big, 300000, &state);
+    Oracle o = {malloc(20000), 12000, 0};
+    CHECK(o.bytes);
+    fill(o.bytes, o.size, &state);
+    CHECK(pure_store_put(path, "kept", 0, (PureSpan){o.bytes, o.size}, &o.revision) == PURE_OK);
+    PureDraft *draft;
+    CHECK(pure_draft_open(path, "kept", &draft) == PURE_OK);
+    CHECK(pure_draft_splice(draft, 6000, 10, (PureSpan){(const unsigned char *)"relocated", 9}) == PURE_OK);
+    apply(&o, 6000, 10, (const unsigned char *)"relocated", 9);
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    struct stat before, after;
+    CHECK(stat(name, &before) == 0);
+    int64_t revision = 0;
+    for (unsigned i = 0; i < 8; ++i)
+        CHECK(pure_store_put(path, "other", revision, (PureSpan){big, 300000 - i}, &revision) == PURE_OK);
+    CHECK(stat(name, &after) == 0 && after.st_ino != before.st_ino);
+    CHECK(pure_draft_commit(draft, &revision) == PURE_OK && revision == o.revision + 1);
+    o.revision = revision;
+    expect(path, "kept", &o);
+    expect_draft(draft, &o, scratch);
+    Oracle other = {big, 300000 - 7, 8};
+    expect(path, "other", &other);
+    pure_draft_close(draft);
+    free(o.bytes);
+    free(big);
+}
+
+/* A held draft keeps earlier versions readable across later edits. */
+static void draft_versions(const char *path) {
+    static unsigned char bytes[200000], scratch[200000];
+    uint32_t state = 17;
+    fill(bytes, sizeof(bytes), &state);
+    int64_t revision;
+    CHECK(pure_store_put(path, "versioned", 0, (PureSpan){bytes, sizeof(bytes)}, &revision) == PURE_OK);
+    PureDraft *draft;
+    CHECK(pure_draft_open(path, "versioned", &draft) == PURE_OK);
+    pure_draft_hold(draft, true);
+    PureVersion first = pure_draft_version(draft);
+    for (unsigned i = 0; i < 400; ++i) {
+        unsigned char byte = (unsigned char)i;
+        CHECK(pure_draft_splice(draft, (i * 977u) % 150000, 3, (PureSpan){&byte, 1}) == PURE_OK);
+    }
+    Gather g = {scratch, 0, 0, 0};
+    CHECK(pure_draft_read_version(draft, first, 0, sizeof(bytes), gather, &g) == PURE_OK);
+    CHECK((size_t)(g.out - scratch) == sizeof(bytes) && !memcmp(scratch, bytes, sizeof(bytes)));
+    CHECK(pure_draft_size(draft) == sizeof(bytes) - 800);
+    size_t nodes, held_bytes, after;
+    pure_draft_test_arena(draft, &nodes, &held_bytes);
+    pure_draft_hold(draft, false);
+    pure_draft_test_arena(draft, &nodes, &after);
+    CHECK(after < held_bytes);
+    pure_draft_close(draft);
+}
+
+/* A draft without a store: edits, versions, restore and reclamation. */
+static void anonymous_draft(const char *path) {
+    (void)path;
+    Oracle o = {malloc(CAPACITY), 0, 0};
+    unsigned char *scratch = malloc(CAPACITY), insert[3000];
+    PureDraft *draft = pure_draft_new();
+    CHECK(draft && o.bytes && scratch && pure_draft_size(draft) == 0 && pure_draft_revision(draft) == 0);
+    uint32_t state = 42;
+    for (unsigned step = 0; step < 4000; ++step) {
+        size_t count = random_value(&state) % (step % 50 ? 12 : sizeof(insert));
+        size_t offset = o.size ? random_value(&state) % (o.size + 1) : 0;
+        size_t remove = o.size - offset ? random_value(&state) % (o.size - offset + 1) % 32 : 0;
+        fill(insert, count, &state);
+        bool abandon = random_value(&state) % 5 == 0;
+        pure_draft_hold(draft, true);
+        PureVersion before = pure_draft_version(draft);
+        CHECK(pure_draft_splice(draft, offset, remove, (PureSpan){insert, count}) == PURE_OK);
+        if (abandon) pure_draft_restore(draft, before);
+        else apply(&o, offset, remove, insert, count);
+        pure_draft_hold(draft, false);
+        if (step % 37 == 0) expect_draft(draft, &o, scratch);
+    }
+    expect_draft(draft, &o, scratch);
+    int64_t revision;
+    CHECK(pure_draft_commit(draft, &revision) == PURE_INVALID);
+    size_t nodes, bytes;
+    pure_draft_test_arena(draft, &nodes, &bytes);
+    CHECK(bytes < 2 * o.size + 65536);
+    pure_draft_close(draft);
+    free(o.bytes);
+    free(scratch);
+}
+
+/* A failed commit changes neither the store nor the draft. */
+static void draft_faults(const char *path) {
+    unsigned char scratch[64];
+    int64_t revision;
+    CHECK(pure_store_put(path, "faulty", 0, (PureSpan){(const unsigned char *)"stable", 6}, &revision) == PURE_OK);
+    PureDraft *draft;
+    CHECK(pure_draft_open(path, "faulty", &draft) == PURE_OK);
+    CHECK(pure_draft_splice(draft, 6, 0, (PureSpan){(const unsigned char *)"+edit", 5}) == PURE_OK);
+    Oracle stored = {(unsigned char *)"stable", 6, 1}, edited = {(unsigned char *)"stable+edit", 11, 1};
+    uint64_t size = file_size(path);
+    for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; ++fail_stage) {
+        CHECK(pure_draft_commit(draft, &revision) == PURE_STORE_ERROR);
+        expect(path, "faulty", &stored);
+        expect_draft(draft, &edited, scratch);
+        CHECK(file_size(path) == size);
+    }
+    fail_stage = PURE_BEFORE_DIR_SYNC;
+    CHECK(pure_draft_commit(draft, &revision) == PURE_UNCERTAIN && revision == 2);
+    fail_stage = -1;
+    expect_draft(draft, &edited, scratch);
+    CHECK(pure_draft_commit(draft, &revision) == PURE_CONFLICT && revision == 2);
+    pure_draft_close(draft);
+    CHECK(pure_draft_open(path, "faulty", &draft) == PURE_OK);
+    edited.revision = 2;
+    expect_draft(draft, &edited, scratch);
+    CHECK(pure_draft_splice(draft, 0, 6, (PureSpan){0}) == PURE_OK);
+    CHECK(pure_draft_commit(draft, &revision) == PURE_OK && revision == 3);
+    Oracle final = {(unsigned char *)"+edit", 5, 3};
+    expect(path, "faulty", &final);
+    pure_draft_close(draft);
+    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]);
+    }
+}
+
+/* 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);
+}
+
+/* Threads of one process commit through their own handles at once: none is
+ * refused as busy, every commit lands, and edits of one shared key conflict
+ * instead of overwriting each other. */
+typedef struct { const char *path; unsigned number, conflicts; } Committer;
+
+static void *commit_many(void *context) {
+    Committer *c = context;
+    PureStore *store;
+    CHECK(pure_store_open(c->path, &store) == PURE_OK);
+    char key[16];
+    snprintf(key, sizeof(key), "t%u", c->number);
+    for (unsigned i = 0; i < 40; ++i) {
+        PureDraft *own, *shared;
+        int64_t revisions[2];
+        CHECK(pure_store_draft(store, key, &own) == PURE_OK && pure_store_draft(store, "shared", &shared) == PURE_OK);
+        CHECK(pure_draft_splice(own, pure_draft_size(own), 0, (PureSpan){(const unsigned char *)"x", 1}) == PURE_OK);
+        CHECK(pure_draft_splice(shared, pure_draft_size(shared), 0, (PureSpan){(const unsigned char *)"y", 1}) == PURE_OK);
+        PureDraft *batch[2] = {own, shared};
+        PureResult result = pure_draft_commit_all(batch, 2, revisions);
+        CHECK(result == PURE_OK || result == PURE_CONFLICT);
+        if (result == PURE_CONFLICT) {
+            ++c->conflicts;
+            CHECK(pure_draft_commit(own, &revisions[0]) == PURE_OK);
+        }
+        pure_draft_close(own);
+        pure_draft_close(shared);
+    }
+    pure_store_close(store);
+    return NULL;
+}
+
+static void threads(const char *path) {
+    pthread_t thread[4];
+    Committer committer[4];
+    for (unsigned i = 0; i < 4; ++i) {
+        committer[i] = (Committer){path, i, 0};
+        CHECK(pthread_create(&thread[i], NULL, commit_many, &committer[i]) == 0);
+    }
+    unsigned conflicts = 0;
+    for (unsigned i = 0; i < 4; ++i) {
+        CHECK(pthread_join(thread[i], NULL) == 0);
+        conflicts += committer[i].conflicts;
+    }
+    PureStoredView view;
+    char key[16];
+    for (unsigned i = 0; i < 4; ++i) {
+        snprintf(key, sizeof(key), "t%u", i);
+        CHECK(pure_store_get(path, key, PURE_OBJECT_LIMIT, &view) == PURE_OK && view.body.size == 40 && view.revision == 40);
+        pure_store_release(&view);
+    }
+    CHECK(pure_store_get(path, "shared", PURE_OBJECT_LIMIT, &view) == PURE_OK);
+    CHECK(view.body.size == 160 - conflicts && view.revision == (int64_t)(160 - conflicts));
+    pure_store_release(&view);
+}
+
+/* Commits that arrive while another is being published share the next
+ * publication: one head for all of them. */
+typedef struct { const char *path; unsigned number; PureResult result; } Single;
+
+static void *commit_one(void *context) {
+    Single *s = context;
+    PureStore *store;
+    PureDraft *draft;
+    char key[16];
+    int64_t revision;
+    snprintf(key, sizeof(key), "g%u", s->number);
+    CHECK(pure_store_open(s->path, &store) == PURE_OK && pure_store_draft(store, key, &draft) == PURE_OK);
+    CHECK(pure_draft_splice(draft, 0, 0, (PureSpan){(const unsigned char *)key, strlen(key)}) == PURE_OK);
+    s->result = pure_draft_commit(draft, &revision);
+    CHECK(s->result != PURE_OK || revision == 1);
+    pure_draft_close(draft);
+    pure_store_close(store);
+    return NULL;
+}
+
+static void grouped(const char *path) {
+    pthread_t thread[5];
+    Single single[5];
+    unsigned long before = pure_tree_test_publications();
+    __atomic_store_n(&slow_once, 1, __ATOMIC_SEQ_CST);
+    for (unsigned i = 0; i < 5; ++i) {
+        single[i] = (Single){path, i, PURE_INVALID};
+        CHECK(pthread_create(&thread[i], NULL, commit_one, &single[i]) == 0);
+        if (!i) pause_ms(20); /* the first is now publishing, slowly */
+    }
+    for (unsigned i = 0; i < 5; ++i) CHECK(pthread_join(thread[i], NULL) == 0 && single[i].result == PURE_OK);
+    CHECK(pure_tree_test_publications() - before == 2);
+    char key[16];
+    for (unsigned i = 0; i < 5; ++i) {
+        snprintf(key, sizeof(key), "g%u", i);
+        expect(path, key, &(Oracle){(unsigned char *)key, strlen(key), 1});
+    }
+}
+
+/* A crash during a commit's only sync can leave its head on the device
+ * without all of its bytes. A process that starts after it passes over that
+ * head to the previous commit, and the next writer overwrites both. */
+static void lost_commit(const char *path) {
+    unsigned char first[] = "first", second[] = "second";
+    int64_t revision;
+    CHECK(pure_store_put(path, "lost", 0, (PureSpan){first, 5}, &revision) == PURE_OK);
+    Oracle o = {first, 5, revision};
+    uint64_t start = published_end(path, "lost");
+    CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 6}, &revision) == PURE_OK);
+    uint64_t end = published_end(path, "lost");
+    char name[512];
+    snprintf(name, sizeof(name), "%s/pure.tree", path);
+    int fd = open(name, O_RDWR);
+    off_t at = (off_t)(start + (end - start) / 2);
+    unsigned char byte;
+    CHECK(fd >= 0 && pread(fd, &byte, 1, at) == 1);
+    byte ^= 0xff;
+    CHECK(pwrite(fd, &byte, 1, at) == 1 && close(fd) == 0);
+    pure_tree_test_forget();
+    expect(path, "lost", &o);
+    PureStore *store;
+    PureDraft *draft;
+    CHECK(pure_store_open(path, &store) == PURE_OK && pure_store_draft(store, "lost", &draft) == PURE_OK);
+    CHECK(pure_draft_revision(draft) == o.revision);
+    pure_draft_close(draft);
+    CHECK(pure_store_put(path, "lost", o.revision + 1, (PureSpan){first, 1}, &revision) == PURE_CONFLICT);
+    CHECK(pure_store_put(path, "lost", o.revision, (PureSpan){second, 3}, &revision) == PURE_OK);
+    o = (Oracle){second, 3, revision};
+    CHECK(published_end(path, "lost") < end); /* it started where the lost commit did */
+    CHECK(pure_store_draft(store, "lost", &draft) == PURE_OK && pure_draft_revision(draft) == o.revision);
+    pure_draft_close(draft);
+    pure_store_close(store);
+    pure_tree_test_forget();
+    expect(path, "lost", &o);
+}
+
+/* While a commit's sync is pending, this process keeps reading the previous
+ * version: a crash could still take the new one back. */
+static void *commit_held(void *context) {
+    int64_t revision;
+    CHECK(pure_store_put(context, "pending", 1, (PureSpan){(const unsigned char *)"new", 3}, &revision) == PURE_OK);
+    return NULL;
+}
+
+static void pending_head(const char *path) {
+    int64_t revision;
+    CHECK(pure_store_put(path, "pending", 0, (PureSpan){(const unsigned char *)"old", 3}, &revision) == PURE_OK);
+    PureStore *store;
+    PureDraft *draft;
+    CHECK(pure_store_open(path, &store) == PURE_OK);
+    __atomic_store_n(&held_sync, 1, __ATOMIC_SEQ_CST);
+    pthread_t writer;
+    CHECK(pthread_create(&writer, NULL, commit_held, (void *)path) == 0);
+    while (__atomic_load_n(&held_sync, __ATOMIC_SEQ_CST) != 2) pause_ms(1);
+    CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 1);
+    pure_draft_close(draft);
+    expect(path, "pending", &(Oracle){(unsigned char *)"old", 3, 1});
+    __atomic_store_n(&held_sync, 0, __ATOMIC_SEQ_CST);
+    CHECK(pthread_join(writer, NULL) == 0);
+    CHECK(pure_store_draft(store, "pending", &draft) == PURE_OK && pure_draft_revision(draft) == 2);
+    pure_draft_close(draft);
+    pure_store_close(store);
+    expect(path, "pending", &(Oracle){(unsigned char *)"new", 3, 2});
+}
+
+/* A commit costs one sync, and commits into the written zeros after the end
+ * leave the file size alone (no journal commit for a new size). */
+static void one_sync(const char *path) {
+    int64_t revision;
+    CHECK(pure_store_put(path, "sync", 0, (PureSpan){(const unsigned char *)"0", 1}, &revision) == PURE_OK);
+    uint64_t size = file_size(path);
+    unsigned long syncs = pure_tree_test_syncs();
+    for (unsigned i = 0; i < 20; ++i)
+        CHECK(pure_store_put(path, "sync", revision, (PureSpan){(const unsigned char *)"x", 1}, &revision) == PURE_OK);
+    CHECK(pure_tree_test_syncs() - syncs == 20 && file_size(path) == size);
+}
+
+static void clean(const char *path) {
+    DIR *entries = opendir(path);
+    CHECK(entries);
+    struct dirent *entry;
+    while ((entry = readdir(entries))) {
+        if (!strcmp(entry->d_name, ".") || !strcmp(entry->d_name, "..")) continue;
+        CHECK(strncmp(entry->d_name, ".pending-", 9));
+        CHECK(unlinkat(dirfd(entries), entry->d_name, 0) == 0);
+    }
+    CHECK(closedir(entries) == 0 && rmdir(path) == 0);
+}
+
+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, store_handle, threads, grouped,
+                                      lost_commit, pending_head, one_sync, records};
+    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);
+        checks[i](path);
+        CHECK(pure_store_init(path) == PURE_OK);
+        clean(path);
+    }
+    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-store-handle-threads-group-commit-lost-commit-pending-head-one-sync-records\"}");
+    return 0;
+}