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.

1 files changed, 394 insertions, 0 deletions

diff --git a/core/runtime.c b/core/runtime.c
new file mode 100644
index 0000000..4a857c7
--- /dev/null
+++ b/core/runtime.c
@@ -0,0 +1,394 @@
+#include "runtime.h"
+#include <string.h>
+
+#define NONE SIZE_MAX
+enum { FREE, RECEIVING, READY, RUNNING, DONE };
+#ifdef PURE_RUNTIME_METRICS
+#define METRIC(...) do { __VA_ARGS__; } while (0)
+#else
+#define METRIC(...) ((void)0)
+#endif
+
+static PureWork *lookup(PureTicket ticket, bool require_held) {
+    PureRuntime *r = ticket.runtime;
+    if (!r || ticket.slot >= r->capacity || !ticket.generation) return NULL;
+    PureWork *w = &r->work[ticket.slot];
+    if (w->state == FREE || w->generation != ticket.generation ||
+        (require_held && !w->held)) return NULL;
+    return w;
+}
+
+static bool valid_key(const char *key) {
+    if (!key || !*key) return false;
+    size_t n = 0;
+    for (; n <= 64 && key[n]; ++n) {
+        unsigned char c = (unsigned char)key[n];
+        if (!((c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') ||
+              (c >= '0' && c <= '9') || c == '-' || c == '_')) return false;
+    }
+    return n <= 64;
+}
+
+/* WRITE results are loans from the admitted input, not a second allocation.
+ * Integer offsets avoid relational comparisons between unrelated C pointers. */
+static bool borrowed_reply(PureSpan input, PureSpan reply) {
+    if (!reply.size) return true;
+    if (!input.data || !reply.data) return false;
+    uintptr_t base = (uintptr_t)input.data, start = (uintptr_t)reply.data;
+    return start >= base && start - base <= input.size &&
+           reply.size <= input.size - (start - base);
+}
+
+static void dispose(PureRuntime *r, PureResource *resource) {
+    PureResource old = *resource;
+    *resource = (PureResource){0};
+    if (old.release) {
+        bool running = r->running;
+        r->running = true;
+        old.release(old.allocation, old.allocation_size);
+        r->running = running;
+    }
+}
+
+static void payload_release(PureRuntime *r, PureWork *w);
+
+/* Iteratively retire a chain of forwarded values. An arbitrarily long chain
+ * must not turn completion into recursive stack growth. */
+static void retire(PureRuntime *r, PureWork *w) {
+    while (w && !w->held && !w->pins) {
+        PureTicket parent = w->parent;
+#ifdef PURE_RUNTIME_METRICS
+        bool released_payload = w->charged || w->resource.release || parent.runtime;
+#endif
+        dispose(r, &w->resource);
+        r->retained -= w->charged;
+        METRIC(if (released_payload) ++r->releases);
+        size_t slot = (size_t)(w - r->work);
+        *w = (PureWork){.next = r->free_head};
+        r->free_head = slot;
+        --r->active;
+        if (!parent.runtime) break;
+        r = parent.runtime;
+        w = lookup(parent, false);
+        /* Pins can only be created and removed by the owner-thread core. */
+        if (!w || !w->pins) break;
+        --w->pins;
+    }
+}
+
+static void payload_release(PureRuntime *r, PureWork *w) {
+    PureTicket parent = w->parent;
+#ifdef PURE_RUNTIME_METRICS
+    bool released_payload = w->charged || w->resource.release || parent.runtime;
+#endif
+    dispose(r, &w->resource);
+    r->retained -= w->charged;
+    METRIC(if (released_payload) ++r->releases);
+    w->charged = 0;
+    w->parent = (PureTicket){0};
+    if (parent.runtime) {
+        PureWork *source = lookup(parent, false);
+        if (source && source->pins) { --source->pins; retire(parent.runtime, source); }
+    }
+}
+
+bool pure_runtime_init(PureRuntime *r, PureWork *work, size_t capacity,
+                       const void *definition, PureHost host, size_t budget) {
+    if (!r || ((work == NULL) != (capacity == 0))) return false;
+    *r = (PureRuntime){.definition = definition, .host = host, .free_head = NONE,
+        .ready_head = NONE, .ready_tail = NONE,
+        .budget = budget, .accepting = true};
+    return pure_runtime_bind(r, work, capacity);
+}
+
+bool pure_runtime_bind(PureRuntime *r, PureWork *work, size_t capacity) {
+    if (!r || r->running || r->active || r->retained ||
+        r->ready_head != NONE || r->ready_tail != NONE ||
+        ((work == NULL) != (capacity == 0)))
+        return false;
+    r->work = work;
+    r->capacity = capacity;
+    r->free_head = capacity ? 0 : NONE;
+    for (size_t i = 0; i < capacity; ++i)
+        work[i] = (PureWork){.next = i + 1 < capacity ? i + 1 : NONE};
+    return true;
+}
+
+bool pure_runtime_restart(PureRuntime *r, const void *definition, PureHost host, size_t budget) {
+    if (!r || r->running || r->active) return false;
+    r->definition = definition;
+    r->host = host;
+    r->budget = budget;
+    r->accepting = true;
+    return true;
+}
+
+void pure_runtime_stop(PureRuntime *r) { if (r && !r->running) r->accepting = false; }
+
+static PureResult admission(PureRuntime *r, PureInput input, const char *key,
+                             int64_t expected, size_t bytes) {
+    if (!r || r->running || !valid_key(key) ||
+        (input != PURE_INPUT_READ && input != PURE_INPUT_WRITE) ||
+        (input == PURE_INPUT_READ ? bytes != 0 : expected < 0)) return PURE_INVALID;
+    if (!r->accepting) return PURE_CLOSED;
+    bool missing_effect = input == PURE_INPUT_READ ? !r->host.read : !r->host.write;
+#ifdef PURE_ASYNC_EFFECTS
+    if (missing_effect && !r->async_effects) return PURE_DENIED;
+#else
+    if (missing_effect) return PURE_DENIED;
+#endif
+    if (r->free_head == NONE) return PURE_BUSY;
+    if (r->serial == UINT64_MAX) return PURE_LIMIT;
+    if (bytes > r->budget - r->retained) { METRIC(++r->budget_rejections); return PURE_BUDGET; }
+    return PURE_OK;
+}
+
+static PureTicket occupy(PureRuntime *r, PureInput input, const char *key,
+                          int64_t expected, size_t bytes, PureResource resource) {
+    size_t slot = r->free_head;
+    PureWork *w = &r->work[slot];
+    r->free_head = w->next;
+    *w = (PureWork){.generation = ++r->serial, .next = NONE, .charged = bytes,
+        .state = RECEIVING, .held = true, .input = input, .expected = expected,
+        .resource = resource};
+    memcpy(w->key, key, strlen(key) + 1);
+    ++r->active;
+    METRIC(++r->accepted);
+    r->retained += bytes;
+    METRIC(if (r->active > r->peak_active) r->peak_active = r->active);
+    METRIC(if (r->retained > r->peak_retained) r->peak_retained = r->retained);
+    return (PureTicket){r, slot, w->generation};
+}
+
+bool pure_runtime_ready(PureTicket ticket) {
+    PureWork *w = lookup(ticket, true);
+    if (!w || ticket.runtime->running || w->state != RECEIVING) return false;
+    PureRuntime *r = ticket.runtime;
+    w->state = READY;
+    if (r->ready_tail == NONE) r->ready_head = ticket.slot;
+    else r->work[r->ready_tail].next = ticket.slot;
+    r->ready_tail = ticket.slot;
+    return true;
+}
+
+PureResult pure_runtime_accept(PureRuntime *r, PureInput input, const char *key,
+                               int64_t expected, size_t size, PureTicket *out) {
+    if (!out) return PURE_INVALID;
+    PureResult status = admission(r, input, key, expected, size);
+    if (status != PURE_OK) return status;
+    PureResource resource = {0};
+    if (size) {
+        if (!r->host.allocate) return PURE_ALLOCATION;
+        r->running = true;
+        bool ok = r->host.allocate(r->host.context, size, &resource);
+        r->running = false;
+        if (!ok || !resource.body.data || resource.body.size != size || !resource.release) {
+            dispose(r, &resource);
+            return PURE_ALLOCATION;
+        }
+    }
+    *out = occupy(r, input, key, expected, size, resource);
+    if (input == PURE_INPUT_READ) pure_runtime_ready(*out);
+    return PURE_OK;
+}
+
+unsigned char *pure_runtime_ingress(PureTicket ticket, size_t *size) {
+    PureWork *w = lookup(ticket, true);
+    if (size) *size = 0;
+    if (!w || ticket.runtime->running || w->state != RECEIVING ||
+        w->input != PURE_INPUT_WRITE || w->parent.runtime) return NULL;
+    if (size) *size = w->resource.body.size;
+    return (unsigned char *)w->resource.body.data;
+}
+
+bool pure_runtime_step(PureRuntime *r) {
+    if (!r || r->running || r->ready_head == NONE) return false;
+#ifdef PURE_ASYNC_EFFECTS
+    PureWork *head = &r->work[r->ready_head];
+    if (head->input == PURE_INPUT_READ ? !r->host.read : !r->host.write) return false;
+#endif
+    size_t slot = r->ready_head;
+    PureWork *w = &r->work[slot];
+    r->ready_head = w->next;
+    if (r->ready_head == NONE) r->ready_tail = NONE;
+    w->next = NONE;
+    w->state = RUNNING;
+    r->running = true;
+    PureOutcome outcome = {0};
+    if (w->input == PURE_INPUT_WRITE) {
+        PureSpan reply = {0};
+        outcome.status = r->host.write(r->host.context, r->definition, w->key,
+                                       w->expected, w->resource.body, &reply,
+                                       &outcome.revision);
+        if (outcome.status == PURE_OK) {
+            if (borrowed_reply(w->resource.body, reply)) outcome.body = reply;
+            else outcome.status = PURE_STORE_ERROR;
+        }
+        METRIC(if (outcome.status == PURE_OK) ++r->commits;
+               else if (outcome.status == PURE_CONFLICT) ++r->conflicts);
+    } else {
+        outcome.status = r->host.read(r->host.context, r->definition, w->key,
+            r->budget - r->retained, &w->resource, &outcome.revision);
+        if (outcome.status == PURE_OK) {
+            if (w->resource.body.size > r->budget - r->retained) outcome.status = PURE_BUDGET;
+            else if (w->resource.body.size && !w->resource.release)
+                outcome.status = PURE_STORE_ERROR;
+            else {
+                w->charged = w->resource.body.size;
+                r->retained += w->charged;
+                METRIC(if (r->retained > r->peak_retained) r->peak_retained = r->retained);
+                outcome.body = w->resource.body;
+                METRIC(++r->gets);
+            }
+        }
+        METRIC(if (outcome.status == PURE_BUDGET) ++r->budget_rejections);
+    }
+    if (outcome.status != PURE_OK || !outcome.body.size) {
+        outcome.body = (PureSpan){0};
+        payload_release(r, w);
+    }
+    METRIC(r->borrowed_output_bytes += outcome.body.size);
+    w->outcome = outcome;
+    w->state = DONE;
+    r->running = false;
+    return true;
+}
+
+#ifdef PURE_ASYNC_EFFECTS
+bool pure_runtime_enable_async(PureRuntime *r) {
+    if (!r || r->running || r->active) return false;
+    r->async_effects = true;
+    return true;
+}
+
+PureTake pure_runtime_take(PureRuntime *r, PureEffect *out) {
+    if (!r || !out || !r->async_effects || r->running || r->ready_head == NONE)
+        return PURE_TAKE_EMPTY;
+    size_t slot = r->ready_head;
+    PureWork *w = &r->work[slot];
+    r->ready_head = w->next;
+    if (r->ready_head == NONE) r->ready_tail = NONE;
+    w->next = NONE;
+    w->state = RUNNING;
+    *out = (PureEffect){.ticket = {r, slot, w->generation}, .input = w->input,
+                        .key = w->key, .expected = w->expected,
+                        .value = w->resource.body,
+                        .available = r->budget - r->retained};
+    return PURE_TAKE_EFFECT;
+}
+
+bool pure_runtime_complete(PureTicket ticket, PureResult status,
+                           PureResource *read_resource, PureSpan write_reply,
+                           int64_t revision) {
+    PureWork *w = lookup(ticket, false);
+    PureRuntime *r = ticket.runtime;
+    if (!w || r->running || w->state != RUNNING ||
+        (w->input == PURE_INPUT_WRITE && read_resource) ||
+        (w->input == PURE_INPUT_READ && write_reply.size) ||
+        (w->input == PURE_INPUT_WRITE && status == PURE_OK &&
+         !borrowed_reply(w->resource.body, write_reply)) ||
+        (read_resource && !read_resource->release &&
+         (read_resource->body.size || read_resource->allocation))) return false;
+    r->running = true;
+    PureOutcome outcome = {.status = status, .revision = revision};
+    if (w->input == PURE_INPUT_READ) {
+        PureResource acquired = {0};
+        if (read_resource) {
+            acquired = *read_resource;
+            *read_resource = (PureResource){0};
+        }
+        if (outcome.status == PURE_OK) {
+            if (acquired.body.size > r->budget - r->retained)
+                outcome.status = PURE_BUDGET;
+            else if (acquired.body.size && !acquired.release)
+                outcome.status = PURE_STORE_ERROR;
+            else {
+                w->resource = acquired;
+                acquired = (PureResource){0};
+                w->charged = w->resource.body.size;
+                r->retained += w->charged;
+                METRIC(if (r->retained > r->peak_retained) r->peak_retained = r->retained);
+                outcome.body = w->resource.body;
+                METRIC(++r->gets);
+            }
+        }
+        dispose(r, &acquired);
+        METRIC(if (outcome.status == PURE_BUDGET) ++r->budget_rejections);
+    } else {
+        if (status == PURE_OK) outcome.body = write_reply;
+        METRIC(if (outcome.status == PURE_OK) ++r->commits;
+               else if (outcome.status == PURE_CONFLICT) ++r->conflicts);
+    }
+    if (outcome.status != PURE_OK || !outcome.body.size) {
+        outcome.body = (PureSpan){0};
+        payload_release(r, w);
+    }
+    METRIC(r->borrowed_output_bytes += outcome.body.size);
+    w->outcome = outcome;
+    w->state = DONE;
+    r->running = false;
+    if (!w->held) retire(r, w);
+    return true;
+}
+#endif
+
+bool pure_runtime_result(PureTicket ticket, PureOutcome *out) {
+    PureWork *w = lookup(ticket, true);
+    if (!out || !w || ticket.runtime->running || w->state != DONE) return false;
+    *out = w->outcome;
+    return true;
+}
+
+PureResult pure_runtime_forward(PureTicket source, PureRuntime *r,
+                                const char *key, int64_t expected, PureTicket *out) {
+    PureWork *parent = lookup(source, true);
+    if (!out || !parent || source.runtime->running || parent->state != DONE ||
+        parent->outcome.status != PURE_OK || (parent->outcome.body.size && !parent->outcome.body.data))
+        return PURE_INVALID;
+    PureResult status = admission(r, PURE_INPUT_WRITE, key, expected, parent->outcome.body.size);
+    if (status != PURE_OK) return status;
+    if (parent->pins == SIZE_MAX) return PURE_LIMIT;
+    PureTicket ticket = occupy(r, PURE_INPUT_WRITE, key, expected, parent->outcome.body.size,
+                              (PureResource){.body = parent->outcome.body});
+    ++parent->pins;
+    r->work[ticket.slot].parent = source;
+    pure_runtime_ready(ticket);
+    *out = ticket;
+    return PURE_OK;
+}
+
+static void unlink_ready(PureRuntime *r, size_t slot) {
+    size_t previous = NONE;
+    for (size_t at = r->ready_head; at != NONE; at = r->work[at].next) {
+        if (at == slot) {
+            if (previous == NONE) r->ready_head = r->work[at].next;
+            else r->work[previous].next = r->work[at].next;
+            if (r->ready_tail == at) r->ready_tail = previous;
+            r->work[at].next = NONE;
+            return;
+        }
+        previous = at;
+    }
+}
+
+bool pure_runtime_cancel(PureTicket ticket) {
+    PureWork *w = lookup(ticket, true);
+    if (!w || ticket.runtime->running || (w->state != RECEIVING && w->state != READY)) return false;
+    if (w->state == READY) unlink_ready(ticket.runtime, ticket.slot);
+    payload_release(ticket.runtime, w);
+    w->outcome = (PureOutcome){.status = PURE_CANCELLED};
+    w->state = DONE;
+    return true;
+}
+
+bool pure_runtime_release(PureTicket ticket) {
+    PureWork *w = lookup(ticket, true);
+    if (!w || ticket.runtime->running) return false;
+    if (w->state == READY) unlink_ready(ticket.runtime, ticket.slot);
+    w->held = false;
+#ifdef PURE_ASYNC_EFFECTS
+    if (w->state == RUNNING) return true;
+#endif
+    retire(ticket.runtime, w);
+    return true;
+}