#define _GNU_SOURCE
/* pure-serve: the HTTP host for aup modules (docs/research/075).
 *
 * An app's views and inputs are its HTTP interface, with no routes to write:
 * GET /views/NAME answers a view (a number or text as text/plain, a table as
 * JSON rows, ?from=N&count=M for a page of them) and POST /inputs/NAME
 * delivers an event (a text input's text is the body; a key input is
 * pressed, down then up; a message takes its u32 fields from the query and
 * its bytes field from the body, and answers what its handler replies, a
 * field of the message). A failed event answers with its fault.
 *
 * A module with keyed messages named show and edit also serves the document
 * contract of server.c (GET and PUT /objects/KEY, If-Match or If-None-Match:
 * *, ETag = revision, X-Pure-Error): GET is its show message and PUT its edit
 * message. HTTP/1.1 connections persist and may pipeline; an epoll thread
 * accepts, reads and writes without blocking. Requests go through the
 * runtime core (a work slot per connection, the payload budget, ingress
 * owned by the core, write replies borrowed from the request). After each
 * round of events every ready request is applied to the module with
 * deferred saving, one save stores the burst, and only then are its
 * requests answered: an answer is durable, and a failed save fails the
 * whole burst, reads included.
 *
 * With --threads N each thread has its own listener on the same port
 * (SO_REUSEPORT), epoll set and runtime domain; the kernel spreads
 * connections over them. A module of keyed documents alone runs one app per
 * thread: they share nothing but the store, whose commits take turns and
 * whose revisions settle concurrent edits of a key. States have one owner:
 * for a module with them, one more thread owns the app, and the connection
 * threads hand it their ready requests and answer what it hands back. Its
 * bursts take the requests of every thread, one save each. */
#include "deliver.h"
#include "wait.h"
#include "link.h"
#include "seal.h"
#include "surface.h"
#include "message-json.h"
#include "module-store.h"
#include "module-text.h"
#include "runtime.h"
#include "store.h"
#include <arpa/inet.h>
#include <dlfcn.h>
#include <errno.h>
#include <fcntl.h>
#include <inttypes.h>
#include <limits.h>
#include <math.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <pthread.h>
#include <signal.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <poll.h>
#include <sched.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/socket.h>
#include <sys/uio.h>
#include <time.h>
#include <unistd.h>

enum { HEADER_LIMIT = 4096, EVENTS = 256, BATCH = 256 };
typedef enum { CLOSED, READING_HEAD, READING_BODY, WAITING, WRITING } Phase;
typedef struct Host Host;

/* What a request names: a document, a view (a table's rows from, count) or
 * an input. Kept per work slot, so an accepted event runs even if its
 * connection is gone before the burst. */
typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT, ROUTE_QUERY } RouteKind;
/* A view's (or a query's) arguments come as the request's query: argued
 * says so, and given is its text, kept on the request's work slot. */
typedef struct { RouteKind kind; uint32_t target; size_t from, count; bool argued; const char *given; } Route;

/* This node's key, for deliveries sealed to it (--key), and whether it
 * takes only those (--sealed-only), or only those from the senders named,
 * each by the key it seals with (--sender NAME=KEY). */
static PureSealKey *seal_key;
static bool sealed_only;
static struct Sender { char name[PURE_SOURCE + 1]; unsigned char key[PURE_SEAL_KEY]; } senders[PURE_MODULE_INPUTS];
static unsigned sender_count;
static size_t memory_budget; /* --memory: the app's budget of what its states hold (0: none) */
static uint64_t work_budget;  /* --work: what an event or a view's evaluation may run (0: none) */

/* A delivery's X-Pure-Seal: the suite, the sender's session (its
 * encapsulated key), the seal's number in it and, when the sender sealed
 * with a key of its own, that key. */
typedef struct {
    char suite[32]; /* empty: not sealed */
    unsigned char enc[PURE_SEAL_KEY], from[PURE_SEAL_KEY];
    uint64_t n;
    bool authenticated;
} Seal;

/* Sessions a connection thread has opened, the least used replaced first:
 * a sender's key agreement is made once a session, not once a request. */
enum { SESSIONS = 16 };
typedef struct { Seal seal; PureOpener *opener; uint64_t used; } Session;

/* Allocated when accepted and reused after closing: memory follows the
 * connections that are open, not the configured maximum. */
typedef struct Connection {
    Host *host;
    struct Connection *next; /* on the spare or the closed list */
    size_t index;            /* in the host's table */
    int fd;
    Phase phase;
    bool readable, writable, keep_alive, put, post;
    bool hung_up; /* the peer closed its side: read until the end, never stop at a short read */
    uint64_t deadline;
    char key[65]; /* the document's key, or the view's or input's name */
    RouteKind route;
    /* An app input's message: its fixed fields from the query, which fields
     * were given there, and whether the body is a JSON object of the rest.
     * Its ingress is [u32le payload size][payload]; with JSON it is [u32le
     * payload size | PLACED][u32le start and size of each field][payload: the
     * prefix, then the JSON as it came], its strings read where they are. */
    unsigned char fields[8 * PURE_MODULE_FIELDS];
    size_t field_bytes;
    bool given[PURE_MODULE_FIELDS], json;
    uint32_t input;
    unsigned char *ingress;
    char source[PURE_SOURCE + 1]; /* a delivered message's sender (X-Pure-Source), or empty */
    uint64_t sequence;            /* and its number (X-Pure-Sequence) */
    unsigned batch;               /* how many numbered messages the body holds (X-Pure-Count) */
    /* A sealed delivery: its seal (X-Pure-Seal), what the seal is bound to
     * (source, number, count), and how many plain bytes the body holds once
     * opened, cleared when its ticket goes. */
    Seal seal;
    char aad[PURE_SOURCE + 48];
    size_t opened;
    int64_t expected;
    size_t body_size, received, in_used;
    unsigned char *body;
    PureTicket ticket;
    struct iovec *out, small[4]; /* the answer: its head, then its body's extents */
    size_t out_count, out_at;
    char head[320];
    char in[HEADER_LIMIT]; /* the header being read, then pipelined bytes */
} Connection;

/* One request applied in a burst, for the connection thread that answers
 * it. A read answered from the store lends its extents: they point into a
 * store mapping, which the resource holds and the runtime charges and
 * releases with the ticket. */
/* A surface's event, waiting for its save: done, and how it went. */
typedef struct { PureResult status; bool done; } Waiter;

typedef struct {
    Host *host;       /* NULL for a surface's event */
    Waiter *waiter;   /* and then its waiter */
    PureTicket ticket;
    PureInput input;
    PureResult status;
    int64_t revision;
    PureResource resource;
    PureSpan reply;
    struct iovec *extents;
    size_t extent_count;
    char key[65];
    RouteKind route;
    PureFault fault;  /* a failed event's */
    const char *type; /* a view's content type */
    struct LinkJob *link; /* records a link delivered (link/link.h) */
    uint64_t heard;       /* then the source's last number heard after them */
    bool gap;             /* and whether they stopped at a gap */
} Applied;

/* The app and its store. A connection thread runs it itself (an app of
 * documents only), or one owner thread runs it for all connection threads.
 * The owner publishes a snapshot of the states after each save, and the
 * connection threads show views from it themselves: reads never wait for
 * the owner, which may be applying events or waiting for a sync. */
typedef struct {
    const PureModule *module;
    PureStore *store;
    PureApp app;
    uint32_t show, edit;
    uint64_t bursts, burst_events;
    Applied *applied; /* a round's requests, grown to fit */
    size_t applied_capacity;
    bool wrote;       /* the last round saved an event or an edit */
    uint64_t save_waited; /* ns the last save waited for the device (a sync), not working */
    bool reads_first; /* the owner's: a round's reads before its writes */
    uint64_t next[PURE_MODULE_INPUTS]; /* when each timer ticks next (monotonic clock, ms; 0: not yet begun) */
    uint64_t ticks;
    uint64_t memory_refused; /* events refused for the app's memory budget */
    bool answers_wait; /* an answer the app had no room for: tried again after ANSWER_WAIT_MS */
    /* The owner's: the last save's snapshot (NULL: the owner shows views),
     * taken under the lock, and the app's changes it shows. */
    bool publishes;
    pthread_mutex_t published_lock;
    PureSnapshot *published;
    uint64_t published_changes, publications;
    struct PureSurfaceHost *surface; /* told of each snapshot published */
    /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */
    PureWire *wires[PURE_MODULE_INPUTS];
    unsigned wire_count;
    uint64_t trimmed[PURE_MODULE_INPUTS], answers;
    bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */
    /* The owner's: outputs linked to Pure receivers (pure://), which output
     * each is, and how far each was trimmed. */
    PureLink *links[PURE_MODULE_INPUTS];
    unsigned link_count;
    uint32_t link_output[PURE_MODULE_INPUTS];
    uint64_t link_trimmed[PURE_MODULE_INPUTS];
    /* Each outlet's place in the outbox, wires first: its sends before it
     * are queued (feed_outlets); and the app's sends_gone when they were
     * last moved back. How often an outlet's window was full. */
    size_t outlet_from[2 * PURE_MODULE_INPUTS];
    uint64_t gone_seen, windows_full;
    struct Writer *writer; /* the thread writing the app's checkpoints, once one was due */
    uint64_t checkpoints;  /* images it wrote */
} Executor;

/* The images of the app's lists are written on a thread of their own
 * (pure_app_checkpoint): the owner hands one over after a save, and takes
 * it back, written, after a later one. */
typedef struct Writer {
    pthread_mutex_t lock;
    pthread_cond_t wake;
    pthread_t thread;
    PureCheckpoint *taken, *written;
    PureResult result;
    bool stopping;
} Writer;

static void *run_writer(void *context) {
    Writer *w = context;
    pthread_mutex_lock(&w->lock);
    for (;;) {
        while (!w->taken && !w->stopping) pthread_cond_wait(&w->wake, &w->lock);
        if (!w->taken) break;
        PureCheckpoint *c = w->taken;
        pthread_mutex_unlock(&w->lock);
        PureResult result = pure_checkpoint_write(c);
        pthread_mutex_lock(&w->lock);
        w->taken = NULL;
        w->written = c;
        w->result = result;
    }
    pthread_mutex_unlock(&w->lock);
    return NULL;
}

/* After a save: a checkpoint written is handed back, and one due handed over. */
static void checkpoint_round(Executor *ex) {
    Writer *w = ex->writer;
    if (w) {
        pthread_mutex_lock(&w->lock);
        PureCheckpoint *written = w->written;
        PureResult result = w->result;
        w->written = NULL;
        pthread_mutex_unlock(&w->lock);
        if (written) {
            pure_app_checkpointed(&ex->app, written, result);
            ex->checkpoints += result == PURE_OK;
        }
    }
    PureCheckpoint *due = pure_app_checkpoint(&ex->app);
    if (!due) return;
    if (!w) {
        w = calloc(1, sizeof(*w));
        if (!w || pthread_mutex_init(&w->lock, NULL) || pthread_cond_init(&w->wake, NULL) ||
            pthread_create(&w->thread, NULL, run_writer, w)) {
            free(w); /* written here instead */
            pure_app_checkpointed(&ex->app, due, pure_checkpoint_write(due));
            return;
        }
        pthread_setname_np(w->thread, "pure-checkpoint"); /* its CPU, seen apart */
        ex->writer = w;
    }
    pthread_mutex_lock(&w->lock);
    w->taken = due;
    pthread_cond_signal(&w->wake);
    pthread_mutex_unlock(&w->lock);
}

/* The writer finishes the image it writes, which goes back to the app. */
static void stop_writer(Executor *ex) {
    Writer *w = ex->writer;
    if (!w) return;
    pthread_mutex_lock(&w->lock);
    w->stopping = true;
    pthread_cond_signal(&w->wake);
    pthread_mutex_unlock(&w->lock);
    pthread_join(w->thread, NULL);
    if (w->written) {
        pure_app_checkpointed(&ex->app, w->written, w->result);
        ex->checkpoints += w->result == PURE_OK;
    }
    pthread_mutex_destroy(&w->lock);
    pthread_cond_destroy(&w->wake);
    free(w);
    ex->writer = NULL;
}

/* A request made ready: its effect, what it names, and the thread that
 * answers it (or, for a surface's event, the waiter it wakes). */
typedef struct LinkJob { PureLinkListener *listener; void *token; unsigned char *ingress; } LinkJob;
typedef struct { Host *host; PureEffect effect; Route route; Waiter *waiter; LinkJob *link; } Job;

/* The owner of an app with states served by several connection threads:
 * their requests wait here, and it runs them in bursts, one save each
 * however many threads they came from, handing each thread back its
 * applied requests to answer (and a surface its events). */
typedef struct Owner {
    Executor executor;
    pthread_mutex_t lock;
    pthread_cond_t waiting;
    Job *jobs;
    size_t count, capacity;
    uint64_t arrivals; /* requests handed to the owner, ever */
    bool stopping;
    bool delivered; /* a wire delivered a send: trim it */
    Host *hosts;
    unsigned host_count;
} Owner;

struct Host {
    const PureModule *module;
    bool documents; /* the module serves /objects: it has show and edit */
    Executor local; /* the app, when this thread runs it */
    Owner *owner;   /* or the owner that runs it */
    PureApp reader; /* shows the owner's snapshot here */
    Applied *views; /* a round's views shown here, answered together */
    size_t views_capacity;
    uint64_t shown; /* views shown from snapshots */
    Route *routes;  /* by work slot */
    struct Given { char *text; size_t capacity; } *given; /* a view's arguments as asked, by work slot */
    PureRuntime runtime;
    PureWork *work;
    Connection **table;   /* open connections by index */
    Connection **by_slot; /* the connection waiting on a work slot */
    Connection *spare, *closed;
    Job *jobs; /* a round's requests: at most one per work slot */
    size_t capacity, active;
    int epoll, listener;
    uint64_t timeout_ms;
    uint64_t now; /* the monotonic clock in ms, read once a round: deadlines count from it */
    uint64_t accepted, requests, expired;
    Session sessions[SESSIONS];
    uint64_t session_uses;
    /* What the owner hands back to answer (swapped for answering), and the
     * eventfd that says so; pending counts requests at the owner. */
    pthread_mutex_t lock;
    Applied *done, *answering;
    size_t done_count, done_capacity, answering_capacity, pending;
    int wake;
};

/* Grow an array to hold `need` items; false when memory runs out. */
static bool grow(void *array, size_t *capacity, size_t need, size_t size) {
    if (need <= *capacity) return true;
    size_t next = *capacity ? *capacity : 64;
    while (next < need) next *= 2;
    void *grown = realloc(*(void **)array, next * size);
    if (!grown) return false;
    *(void **)array = grown;
    *capacity = next;
    return true;
}

static atomic_bool stopping; /* lock-free, so a signal handler may set it */
static void stop(int signal_number) { (void)signal_number; atomic_store(&stopping, true); }

static uint64_t now_ms(void) {
    struct timespec t;
    clock_gettime(CLOCK_MONOTONIC, &t);
    return (uint64_t)t.tv_sec * 1000 + (uint64_t)t.tv_nsec / 1000000;
}

/* The time events see: the wall clock, in milliseconds since 1970 (UTC). */
static double wall_ms(void) {
    struct timespec t;
    clock_gettime(CLOCK_REALTIME, &t);
    return floor((double)t.tv_sec * 1000 + (double)t.tv_nsec / 1e6);
}

static bool number(const char *text, uint64_t maximum, uint64_t *value) {
    if (!*text) return false;
    uint64_t n = 0;
    for (; *text; ++text) {
        if (*text < '0' || *text > '9') return false;
        unsigned digit = (unsigned)(*text - '0');
        if (n > (maximum - digit) / 10) return false;
        n = n * 10 + digit;
    }
    *value = n;
    return true;
}

/* ---- Connections ------------------------------------------------------ */

static void release_ticket(void *owner) {
    Connection *c = owner;
    if (c->opened && c->body) explicit_bzero(c->body, c->opened); /* an opened seal's plain text is gone with it */
    c->opened = 0;
    if (c->ticket.runtime && !pure_runtime_release(c->ticket)) abort();
    c->ticket = (PureTicket){0};
    c->body = NULL;
}

/* Everything an answer borrowed goes back with its ticket: the request's
 * payload, a copied reply or the lent store extents. */
static void finish_output(Connection *c) {
    release_ticket(c);
    if (c->out != c->small) free(c->out);
    c->out = NULL;
    c->out_count = c->out_at = 0;
}

/* Events of this round may still name c: it is reused after the round. */
static void close_connection(Connection *c) {
    if (c->phase == CLOSED) return;
    Host *host = c->host;
    finish_output(c);
    close(c->fd); /* also leaves the epoll set */
    c->phase = CLOSED;
    host->table[c->index] = NULL;
    c->next = host->closed;
    host->closed = c;
    --host->active;
}

/* An answer's head is written in place, part by part: printf's scanning of
 * its format took a sixth of a small read's time. Clipped at the end of the
 * buffer, which its parts, all short, never reach. */
typedef struct { char *at, *end; } Head;

static void put_text(Head *h, const char *text) {
    size_t length = strlen(text), room = (size_t)(h->end - h->at);
    if (length > room) length = room;
    memcpy(h->at, text, length);
    h->at += length;
}

static void put_decimal(Head *h, uint64_t value) {
    char digits[20];
    size_t count = 0;
    do digits[count++] = (char)('0' + value % 10); while (value /= 10);
    while (count && h->at < h->end) *h->at++ = digits[--count];
}

/* Answer with a body of `count` extents (`size` bytes in all); the ticket
 * keeps them alive. One writev sends as much as the socket takes. */
static void respond_typed(Connection *c, int status, const char *reason, const char *error, int64_t revision,
                          const char *type, const struct iovec *extents, size_t count, size_t size) {
    Head h = {c->head, c->head + sizeof(c->head)};
    put_text(&h, "HTTP/1.1 ");
    put_decimal(&h, (uint64_t)status);
    put_text(&h, " ");
    put_text(&h, reason);
    put_text(&h, "\r\nContent-Length: ");
    put_decimal(&h, size);
    put_text(&h, c->keep_alive ? "\r\n" : "\r\nConnection: close\r\n");
    put_text(&h, "Content-Type: ");
    put_text(&h, type);
    put_text(&h, "\r\nCache-Control: no-store\r\nX-Pure-Error: ");
    put_text(&h, error);
    put_text(&h, "\r\n");
    if (revision > 0) {
        put_text(&h, "ETag: \"");
        put_decimal(&h, (uint64_t)revision);
        put_text(&h, "\"\r\n");
    }
    /* A delivery's answer tells how many numbered messages one request may carry. */
    if (*c->source) {
        put_text(&h, "X-Pure-Batch: ");
        put_decimal(&h, BATCH);
        put_text(&h, "\r\n");
    }
    put_text(&h, "\r\n");
    size_t length = (size_t)(h.at - c->head);
    c->out = count < sizeof(c->small) / sizeof(*c->small) ? c->small : malloc((count + 1) * sizeof(*c->out));
    c->phase = WRITING;
    if (!c->out) { c->keep_alive = false; c->out = c->small; count = 0; } /* cannot send it: the close tells */
    c->out[0] = (struct iovec){c->head, length};
    if (count) memcpy(c->out + 1, extents, count * sizeof(*extents));
    c->out_count = count + 1;
    c->out_at = 0;
}

static void respond_extents(Connection *c, int status, const char *reason, const char *error, int64_t revision,
                            const struct iovec *extents, size_t count, size_t size) {
    respond_typed(c, status, reason, error, revision, "application/octet-stream", extents, count, size);
}

static void respond(Connection *c, int status, const char *reason, const char *error, int64_t revision,
                    PureSpan body) {
    struct iovec extent = {(void *)body.data, body.size};
    respond_extents(c, status, reason, error, revision, &extent, body.size ? 1 : 0, body.size);
}

/* A request refused before it has a ticket; a body it came with is not read. */
static void reject(Connection *c, int status, const char *reason, const char *error) {
    if (c->body_size) c->keep_alive = false;
    respond(c, status, reason, error, 0, (PureSpan){0});
}

/* A request the host cannot read further: answer and close. */
static void refuse(Connection *c, int status, const char *reason, const char *error) {
    release_ticket(c);
    c->keep_alive = false;
    respond(c, status, reason, error, 0, (PureSpan){0});
}

static void answer(Connection *c, PureOutcome result) {
    switch (result.status) {
    case PURE_OK: respond(c, 200, "OK", "none", result.revision, result.body); return;
    case PURE_CONFLICT: respond(c, 412, "Precondition Failed", "conflict", result.revision, (PureSpan){0}); return;
    case PURE_NOT_FOUND: respond(c, 404, "Not Found", "not_found", 0, (PureSpan){0}); break;
    case PURE_BUSY: respond(c, 503, "Service Unavailable", "store_busy", 0, (PureSpan){0}); break;
    case PURE_BUDGET: respond(c, 503, "Service Unavailable", "payload_budget", 0, (PureSpan){0}); break;
    case PURE_ALLOCATION: respond(c, 503, "Service Unavailable", "allocation", 0, (PureSpan){0}); break;
    case PURE_CLOSED: respond(c, 503, "Service Unavailable", "stopped", 0, (PureSpan){0}); break;
    case PURE_DENIED: respond(c, 403, "Forbidden", "denied", 0, (PureSpan){0}); break;
    case PURE_LIMIT: respond(c, 413, "Content Too Large", "limit", 0, (PureSpan){0}); break;
    case PURE_UNCERTAIN: respond(c, 500, "Internal Server Error", "durability_unknown", 0, (PureSpan){0}); break;
    default: respond(c, 500, "Internal Server Error", "storage", 0, (PureSpan){0}); break;
    }
    /* A failed request carries no body: the payload can go now. */
    if (!result.body.size) release_ticket(c);
}

/* An app route's answer: a view's value, or what an event replied. A failed
 * event names its fault: 404 not_found, 409 exists, 503 memory (the app's
 * budget: again later), 422 for the others; a view without a value is 404
 * no_value. */
static void answer_app(Connection *c, PureOutcome result, bool view, PureFault fault, const char *type) {
    const char *named = fault != PURE_FAULT_NONE ? pure_fault_name(fault) : NULL;
    if (result.status == PURE_OK) {
        struct iovec extent = {(void *)result.body.data, result.body.size};
        respond_typed(c, 200, "OK", "none", 0, type ? type : "application/octet-stream", &extent,
                      result.body.size ? 1 : 0, result.body.size);
    } else if (result.status == PURE_NOT_FOUND) {
        respond(c, 404, "Not Found", view ? "no_value" : named ? named : "not_found", 0, (PureSpan){0});
    } else if (result.status == PURE_CONFLICT) {
        respond(c, 409, "Conflict", named ? named : "conflict", 0, (PureSpan){0});
    } else if (fault == PURE_FAULT_MEMORY) {
        respond(c, 503, "Service Unavailable", named, 0, (PureSpan){0});
    } else if (named && (result.status == PURE_LIMIT || result.status == PURE_INVALID)) {
        respond(c, 422, "Unprocessable Content", named, 0, (PureSpan){0});
    } else if (result.status == PURE_INVALID) {
        respond(c, 400, "Bad Request", "invalid", 0, (PureSpan){0}); /* such as a text that is not UTF-8 */
    } else {
        answer(c, result);
        return;
    }
    if (!result.body.size) release_ticket(c);
}

/* "SUITE ENC N[ FROM]": the keys as 64 hex digits, the number in decimal. */
static bool read_seal(const char *value, Seal *seal) {
    char copy[256], *rest, *suite = NULL, *enc = NULL, *n = NULL, *from = NULL, *extra = NULL;
    if (strlen(value) >= sizeof(copy)) return false;
    strcpy(copy, value);
    suite = strtok_r(copy, " ", &rest);
    if (suite) enc = strtok_r(NULL, " ", &rest);
    if (enc) n = strtok_r(NULL, " ", &rest);
    if (n) from = strtok_r(NULL, " ", &rest);
    if (from) extra = strtok_r(NULL, " ", &rest);
    uint64_t sequence;
    if (!n || extra || strlen(suite) >= sizeof(seal->suite) || !pure_seal_peer(enc, seal->enc) ||
        !number(n, UINT64_MAX, &sequence) || (from && !pure_seal_peer(from, seal->from)))
        return false;
    strcpy(seal->suite, suite);
    seal->n = sequence;
    seal->authenticated = from != NULL;
    return true;
}

/* A sender this node names, sealing with the key it was named with. */
static bool known_sender(const char *source, const Seal *seal) {
    for (unsigned i = 0; i < sender_count; ++i)
        if (!strcmp(senders[i].name, source))
            return seal->authenticated && !memcmp(senders[i].key, seal->from, PURE_SEAL_KEY);
    return false;
}

/* Parse one request head of `size` bytes (through the blank line): the same
 * bounded HTTP/1.1 subset as server.c, plus Connection, POST and the app
 * routes. A query stays in head for start_request; an unknown path is
 * answered 404 once its head is read. */
static bool parse_head(Connection *c, char *head, size_t size, char **query) {
    for (size_t i = 0; i < size; ++i) {
        unsigned char b = (unsigned char)head[i];
        if ((b < 32 && b != '\r' && b != '\n' && b != '\t') || b == 127) return false;
    }
    head[size - 2] = 0; /* keep the last CRLF of the last header */
    char *line = head, *next = strstr(line, "\r\n");
    if (!next) return false;
    *next = 0;
    char *path = strchr(line, ' ');
    if (!path) return false;
    *path++ = 0;
    c->put = !strcmp(line, "PUT");
    c->post = !strcmp(line, "POST");
    if (!c->put && !c->post && strcmp(line, "GET")) return false;
    char *version = strchr(path, ' ');
    if (!version) return false;
    *version++ = 0;
    if (strcmp(version, "HTTP/1.1")) return false;
    if ((*query = strchr(path, '?'))) *(*query)++ = 0;
    static const struct { const char *prefix; RouteKind kind; } routes[] = {
        {"/objects/", ROUTE_OBJECT}, {"/views/", ROUTE_VIEW}, {"/inputs/", ROUTE_INPUT}, {"/queries/", ROUTE_QUERY}};
    const char *key = "";
    c->route = ROUTE_NONE;
    for (size_t i = 0; i < sizeof(routes) / sizeof(*routes) && c->route == ROUTE_NONE; ++i)
        if (!strncmp(path, routes[i].prefix, strlen(routes[i].prefix))) {
            c->route = routes[i].kind;
            key = path + strlen(routes[i].prefix);
        }
    size_t key_size = strlen(key);
    if (c->route != ROUTE_NONE && (!key_size || key_size > 64)) return false;
    for (size_t i = 0; i < key_size; ++i) {
        char b = key[i];
        if (!((b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') || b == '-' || b == '_'))
            return false;
    }
    memcpy(c->key, key, key_size + 1);
    /* Each route has its method: documents GET and PUT, views GET, inputs
     * POST, queries PUT (registered) and GET (shown). */
    if ((c->route == ROUTE_OBJECT && (c->post || *query)) || (c->route == ROUTE_VIEW && (c->put || c->post)) ||
        (c->route == ROUTE_INPUT && !c->post) || (c->route == ROUTE_QUERY && (c->post || (c->put && *query))))
        return false;
    bool length_seen = false, condition_seen = false, host_seen = false;
    c->source[0] = 0;
    c->sequence = 0;
    c->batch = 1;
    c->seal.suite[0] = 0;
    bool counted = false;
    c->body_size = 0;
    c->json = false;
    c->expected = -1;
    c->keep_alive = true;
    line = next + 2;
    while (*line) {
        next = strstr(line, "\r\n");
        if (!next) return false;
        *next = 0;
        char *value = strchr(line, ':');
        if (!value || value == line) return false;
        *value++ = 0;
        for (char *p = line; *p; ++p)
            if (!((*p >= 'a' && *p <= 'z') || (*p >= 'A' && *p <= 'Z') || (*p >= '0' && *p <= '9') || *p == '-'))
                return false;
        while (*value == ' ' || *value == '\t') ++value;
        char *tail = value + strlen(value);
        while (tail > value && (tail[-1] == ' ' || tail[-1] == '\t')) *--tail = 0;
        uint64_t n;
        if (!strcasecmp(line, "Content-Length")) {
            if (length_seen || !number(value, SIZE_MAX, &n)) return false;
            length_seen = true;
            c->body_size = (size_t)n;
        } else if (!strcasecmp(line, "Host")) {
            if (host_seen || !*value) return false;
            host_seen = true;
        } else if (!strcasecmp(line, "If-Match")) {
            if (condition_seen || tail - value < 3 || *value != '"' || tail[-1] != '"') return false;
            tail[-1] = 0;
            if (!number(value + 1, INT64_MAX, &n) || !n) return false;
            c->expected = (int64_t)n;
            condition_seen = true;
        } else if (!strcasecmp(line, "If-None-Match")) {
            if (condition_seen || strcmp(value, "*")) return false;
            c->expected = 0;
            condition_seen = true;
        } else if (!strcasecmp(line, "Content-Type")) {
            c->json = !strncasecmp(value, "application/json", 16);
        } else if (!strcasecmp(line, "Connection")) {
            if (!strcasecmp(value, "close")) c->keep_alive = false;
            else if (strcasecmp(value, "keep-alive")) return false;
        } else if (!strcasecmp(line, "X-Pure-Source")) {
            /* A message delivered from another app's output: applied once per number. */
            size_t length = strlen(value);
            if (*c->source || !length || length > PURE_SOURCE) return false;
            for (size_t i = 0; i < length; ++i) if (value[i] <= ' ' || value[i] > '~') return false;
            memcpy(c->source, value, length + 1);
        } else if (!strcasecmp(line, "X-Pure-Sequence")) {
            if (c->sequence || !number(value, UINT64_MAX, &n) || !n) return false;
            c->sequence = n;
        } else if (!strcasecmp(line, "X-Pure-Seal")) {
            /* A delivery sealed to this node's key: suite, session, number[, sender]. */
            if (*c->seal.suite || !read_seal(value, &c->seal)) return false;
        } else if (!strcasecmp(line, "X-Pure-Count")) {
            /* Numbered messages from X-Pure-Sequence on, as a JSON array. */
            if (counted || !number(value, BATCH, &n) || !n) return false;
            c->batch = (unsigned)n;
            counted = true;
        } else if (!strcasecmp(line, "Transfer-Encoding") || !strcasecmp(line, "Expect")) return false;
        line = next + 2;
    }
    if (!*c->source != !c->sequence || (*c->source && c->route != ROUTE_INPUT) || (counted && !*c->source) ||
        c->sequence > UINT64_MAX - c->batch) return false;
    /* A document's PUT names the revision it was made against; a query's replaces what was registered. */
    if (c->put) return host_seen && length_seen && condition_seen == (c->route != ROUTE_QUERY);
    if (c->post) return host_seen && !condition_seen; /* no Content-Length: no body */
    return host_seen && !condition_seen && c->body_size == 0;
}

static void put_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
static void put_u64(unsigned char *p, uint64_t v) { for (unsigned i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
static uint32_t get_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; }
static uint64_t get_u64(const unsigned char *p) { return get_u32(p) | (uint64_t)get_u32(p + 4) << 32; }

static int hex_value(char c) {
    return c >= '0' && c <= '9' ? c - '0' : c >= 'a' && c <= 'f' ? c - 'a' + 10 : c >= 'A' && c <= 'F' ? c - 'A' + 10 : -1;
}

/* %XX decoded in place (never longer): the length after, or -1 for a bad escape. */
static long percent_decoded(char *text, size_t length) {
    size_t o = 0;
    for (size_t i = 0; i < length; ++i) {
        if (text[i] != '%') { text[o++] = text[i]; continue; }
        int high = i + 2 < length ? hex_value(text[i + 1]) : -1, low = high >= 0 ? hex_value(text[i + 2]) : -1;
        if (low < 0) return -1;
        text[o++] = (char)(high << 4 | low);
        i += 2;
    }
    return (long)o;
}

/* What an app route names in the module, with a message's fixed fields
 * taken from the query (%XX decoded, as a view's arguments are) into
 * c->fields. NULL, or the error to answer (404 for a name the module lacks,
 * 400 otherwise). */
static const char *resolve(const Host *host, Connection *c, char *query, Route *route, int *status) {
    const PureModule *m = host->module;
    *status = 404;
    c->field_bytes = 0;
    memset(c->given, 0, sizeof(c->given));
    if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents) ||
        (route->kind == ROUTE_QUERY && !host->owner)) return "route"; /* a query reads an app's states */
    if (route->kind == ROUTE_OBJECT) return NULL;
    if (route->kind == ROUTE_QUERY) { /* the view is the query module's, found when shown */
        route->from = 0;
        route->count = SIZE_MAX;
        *status = 400;
        for (char *pair = query; pair && *pair;) {
            char *end = strchr(pair, '&'), *value = strchr(pair, '=');
            if (end) *end = 0;
            uint64_t n;
            if (!value || (end && value > end)) return "query";
            *value++ = 0;
            bool from = !strcmp(pair, "from"), paging = from || !strcmp(pair, "count");
            if (paging && !number(value, SIZE_MAX, &n)) return "query";
            if (!paging) route->argued = true; /* the query's view reads it, if it takes arguments */
            else if (from) route->from = (size_t)n;
            else route->count = (size_t)n;
            pair = end ? end + 1 : NULL;
        }
        return NULL;
    }
    int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key);
    /* An output is the app's to send and a timer ticks by itself: neither is posted. */
    if (found >= 0 && route->kind == ROUTE_INPUT &&
        (m->input[found].kind == PURE_KIND_OUTPUT || m->input[found].kind == PURE_KIND_TIMER ||
         m->input[found].kind == PURE_KIND_ARGUMENTS)) found = -1;
    if (found < 0) return route->kind == ROUTE_VIEW ? "no_view" : "no_input";
    *status = 400;
    route->target = (uint32_t)found;
    route->from = 0;
    route->count = SIZE_MAX;
    const PureInputShape *shape = route->kind == ROUTE_INPUT ? &m->input[found] : NULL;
    bool message = shape && shape->kind == PURE_KIND_MESSAGE;
    if (message && shape->keyed) return "keyed";
    if (*c->source && !message) return "delivery";
    if (!host->owner) c->source[0] = 0; /* an app without states or outputs applies it as a message */
    if (shape) {
        /* A body is a text input's text, a message's one variable field (rows
         * in their list's stored form), or a JSON object of a message's fields
         * (two variable fields need one). */
        unsigned variables = 0;
        for (uint32_t f = 0; message && f < shape->field_count; ++f)
            variables += m->field[shape->field_start + f].type == PURE_FIELD_BYTES ||
                         m->field[shape->field_start + f].type == PURE_FIELD_TEXT ||
                         m->field[shape->field_start + f].type == PURE_FIELD_ROWS;
        if (c->json && !message) return "json";
        if (!c->json && (variables > 1 || (c->body_size && shape->kind == PURE_KIND_KEY) ||
                         (c->body_size && message && !variables))) return variables > 1 ? "json" : "body";
        c->input = (uint32_t)found;
        c->field_bytes = message ? shape->prefix : 0;
    }
    /* name=value pairs: a table's from and count, a message's fixed fields. */
    for (char *pair = query; pair && *pair;) {
        char *end = strchr(pair, '&'), *value = strchr(pair, '=');
        if (end) *end = 0;
        if (!value || (end && value > end)) return "query";
        *value++ = 0;
        if (route->kind == ROUTE_VIEW) {
            uint64_t n;
            bool from = !strcmp(pair, "from"), paging = from || !strcmp(pair, "count");
            if (!paging && m->view_arguments[found]) route->argued = true; /* an argument, read when shown */
            else if (!number(value, SIZE_MAX, &n) || m->view_type[found] != PURE_VIEW_TABLE || !paging) return "query";
            else if (from) route->from = (size_t)n;
            else route->count = (size_t)n;
        } else {
            uint32_t f = 0;
            while (f < (message ? shape->field_count : 0) && strcmp(m->field[shape->field_start + f].name, pair)) ++f;
            const PureField *field = message && f < shape->field_count ? &m->field[shape->field_start + f] : NULL;
            long length = percent_decoded(value, strlen(value)); /* as a view's arguments are */
            if (!field || c->given[f] || !pure_field_fixed(field->type) ||
                length < 0 || !pure_json_fixed(field, (const unsigned char *)value, (size_t)length, c->fields))
                return "query";
            c->given[f] = true;
        }
        pair = end ? end + 1 : NULL;
    }
    for (uint32_t f = 0; message && !c->json && f < shape->field_count; ++f) {
        uint8_t type = m->field[shape->field_start + f].type;
        if (pure_field_fixed(type) && !c->given[f]) return "missing_field";
    }
    for (uint32_t f = 0; c->batch > 1 && f < shape->field_count; ++f)
        if (c->given[f]) return "delivery"; /* a batch's fields are all in its objects */
    if (c->batch > 1 && !c->json) return "delivery";
    if (!*c->source) c->batch = 1;
    return NULL;
}

/* Parsing a JSON body in place: the payload it is in, the placed fields'
 * table, and which fields it gave. */
#define PLACED (UINT32_C(1) << 31)
/* After the size: a delivered message's number (u64le), how many follow it
 * (u32le: a batch) and its source (u8 length and the bytes), in DELIVERY
 * bytes before the tables; then a table for each message, a prefix for each
 * and the JSON. */
#define DELIVERED (UINT32_C(1) << 30)
/* With DELIVERED: the batch's messages are records a link delivered, each
 * as stored (u32le size, then its bytes), applied in order from the one
 * after the source's last heard number, stopping at a gap. */
#define STORED_RECORDS (UINT32_C(1) << 29)
enum { DELIVERY = 80 };
/* An app input's JSON body becomes its message: an object with a member for
 * each field the query did not give (a u32, a number, or a string for bytes
 * and text), groups as objects inside it, and nothing else. Fixed fields go
 * into the prefix; strings stay in the body, and the table says where.
 * NULL, or the error to answer. */
static const char *convert(Connection *c) {
    const PureInputShape *shape = &c->host->module->input[c->input];
    uint32_t delivered = get_u32(c->ingress) & DELIVERED;
    size_t table = 8 * (size_t)shape->field_count, payload = c->batch * c->field_bytes + c->body_size;
    unsigned char *tables = c->ingress + 4 + (delivered ? DELIVERY : 0), *prefixes = tables + c->batch * table;
    unsigned char *json = c->body;
    size_t size = c->body_size, at = pure_json_space(json, 0, size);
    /* A message's rows go after the body: they are encoded, then placed. */
    PureJsonMessage b = {.module = c->host->module, .shape = shape, .given = c->given, .rows_base = payload};
    const char *why = NULL;
    /* A batch is an array of its messages' objects. */
    if (c->batch > 1) {
        if (at == size || json[at++] != '[') return "json";
        at = pure_json_space(json, at, size);
    }
    for (unsigned i = 0; i < c->batch && !why; ++i) {
        if (i && (at == size || json[at++] != ',')) why = "json";
        at = pure_json_space(json, at, size);
        b.table = tables + i * table;
        b.payload = prefixes + i * c->field_bytes;
        memset(b.seen, 0, sizeof(b.seen));
        if (!why) why = pure_json_message(&b, json, size, &at);
        at = pure_json_space(json, at, size);
        /* A delivery from another node may lack fields its sender's module has
         * not (docs/design/014-fields-by-name.md): they are 0 or empty. A
         * member naming no field is refused all the same: it would be dropped. */
        for (uint32_t f = 0; !why && f < shape->field_count; ++f) {
            if (b.seen[f] || c->given[f]) continue;
            const PureField *field = &c->host->module->field[shape->field_start + f];
            if (!delivered || field->type == PURE_FIELD_ROWS) why = "missing_field";
            else if (pure_field_fixed(field->type))
                memset(b.payload + field->offset, 0, pure_field_size(field->type));
            else put_u32(b.table + 8 * f, 0), put_u32(b.table + 8 * f + 4, 0);
        }
        if (!why && c->batch > 1 && b.rows_size) why = "delivery"; /* an output sends no rows */
    }
    if (!why && c->batch > 1 && (at == size || json[at++] != ']')) why = "json";
    if (!why && pure_json_space(json, at, size) != size) why = "json";
    if (!why && b.rows_size) {
        size_t body = (size_t)(c->body - c->ingress);
        PureResult grown = pure_runtime_grow(c->ticket, body + c->body_size + b.rows_size);
        if (grown != PURE_OK) why = grown == PURE_BUDGET ? "payload_budget" : "memory";
        else {
            c->ingress = pure_runtime_ingress(c->ticket, NULL);
            c->body = c->ingress + body;
            memcpy(c->body + c->body_size, b.rows, b.rows_size);
        }
    }
    free(b.rows);
    if (why) return why;
    put_u32(c->ingress, (uint32_t)(payload + b.rows_size) | PLACED | delivered);
    return NULL;
}

/* The session a seal was made in: one this thread opened before, or opened
 * now in place of the least used; NULL when it does not open. */
static PureOpener *session(Host *host, const Seal *seal) {
    Session *least = &host->sessions[0];
    for (unsigned i = 0; i < SESSIONS; ++i) {
        Session *k = &host->sessions[i];
        if (k->opener && !memcmp(k->seal.enc, seal->enc, PURE_SEAL_KEY) && !strcmp(k->seal.suite, seal->suite) &&
            k->seal.authenticated == seal->authenticated &&
            (!seal->authenticated || !memcmp(k->seal.from, seal->from, PURE_SEAL_KEY))) {
            k->used = ++host->session_uses;
            return k->opener;
        }
        if (k->used < least->used) least = k;
    }
    PureOpener *opener = pure_opener(seal_key, seal->suite, seal->enc, seal->authenticated ? seal->from : NULL);
    if (!opener) return NULL;
    pure_opener_free(least->opener);
    *least = (Session){*seal, opener, ++host->session_uses};
    return opener;
}

/* A sealed body opened where it lies, or false (another key, another
 * session or sender, another delivery's binding, or changed on the way). */
static bool open_seal(Connection *c) {
    if (c->body_size < PURE_SEAL_OVERHEAD) return false;
    PureOpener *opener = session(c->host, &c->seal);
    size_t size = c->body_size - PURE_SEAL_OVERHEAD;
    unsigned char *plain = opener ? malloc(size ? size : 1) : NULL;
    bool opened = plain && pure_unseal(opener, c->seal.n, c->aad, c->body, c->body_size, plain);
    if (opened) {
        memcpy(c->body, plain, size);
        c->body_size = size;
        c->opened = size;
    }
    if (plain) explicit_bzero(plain, size);
    free(plain);
    return opened;
}

/* A request's body is in: an app input's JSON becomes its payload, and the
 * request is ready, or it is answered 400 and never runs. */
static void body_done(Connection *c) {
    if (*c->seal.suite && !open_seal(c)) {
        release_ticket(c); /* never made ready: it never runs */
        respond(c, 400, "Bad Request", "seal", 0, (PureSpan){0});
        return;
    }
    const char *why = c->route == ROUTE_INPUT && c->json ? convert(c) : NULL;
    if (why) {
        release_ticket(c); /* never made ready: it never runs */
        if (!strcmp(why, "payload_budget") || !strcmp(why, "memory")) respond(c, 503, "Service Unavailable", why, 0, (PureSpan){0});
        else respond(c, 400, "Bad Request", why, 0, (PureSpan){0});
        return;
    }
    if (!pure_runtime_ready(c->ticket)) abort();
    c->phase = WAITING;
}

/* A whole head is in `in`: admit the request and take any body bytes that
 * came with it; later bytes stay for the next request. */
static void start_request(Connection *c, size_t head_size) {
    Host *host = c->host;
    char head[HEADER_LIMIT], *query = NULL;
    memcpy(head, c->in, head_size);
    size_t extra = c->in_used - head_size;
    memmove(c->in, c->in + head_size, extra);
    c->in_used = extra;
    if (!parse_head(c, head, head_size, &query)) { refuse(c, 400, "Bad Request", "request"); return; }
    /* The query as it came: resolving takes it apart, and a view's arguments are read from it when shown. */
    char raw[HEADER_LIMIT];
    size_t raw_size = query ? strlen(query) : 0;
    if (raw_size) memcpy(raw, query, raw_size + 1);
    /* A node that keeps its deliveries sealed refuses the others, and one
     * that names its senders refuses any other sender; a seal needs this
     * node's key, and binds the delivery's source, number and count. */
    if (*c->source && (sealed_only || sender_count) && !*c->seal.suite) { refuse(c, 403, "Forbidden", "unsealed"); return; }
    if (*c->seal.suite) {
        if (!*c->source || !seal_key || !c->json) { refuse(c, 400, "Bad Request", "seal"); return; }
        if (sender_count && !known_sender(c->source, &c->seal)) { refuse(c, 403, "Forbidden", "sender"); return; }
        snprintf(c->aad, sizeof(c->aad), "%s %llu %u", c->source, (unsigned long long)c->sequence, c->batch);
    }
    if (c->body_size > PURE_OBJECT_LIMIT) { refuse(c, 413, "Content Too Large", "limit"); return; }
    ++host->requests;
    Route route = {.kind = c->route};
    int status;
    const char *why = resolve(host, c, query, &route, &status);
    if (why) { reject(c, status, status == 404 ? "Not Found" : "Bad Request", why); return; }
    /* A document's PUT carries its body; an input's event its payload's size,
     * the payload (the fixed fields, then the body), and room for its JSON. */
    bool write = route.kind == ROUTE_INPUT || c->put;
    size_t table = route.kind == ROUTE_INPUT && c->json ? 8 * (size_t)host->module->input[c->input].field_count : 0;
    size_t lead = 4 + (route.kind == ROUTE_INPUT && *c->source ? DELIVERY : 0);
    /* A batch has a table and a prefix for each of its messages. */
    size_t input_size = route.kind == ROUTE_INPUT ? lead + c->batch * (table + c->field_bytes) + c->body_size :
                        write ? c->body_size : 0;
    PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key,
                                              route.kind == ROUTE_INPUT || route.kind == ROUTE_QUERY ? 0 : c->expected,
                                              input_size, &c->ticket);
    if (admitted != PURE_OK) {
        c->ticket = (PureTicket){0};
        /* An unread body would desynchronize the stream. */
        if (c->body_size) c->keep_alive = false;
        answer(c, (PureOutcome){.status = admitted});
        return;
    }
    host->by_slot[c->ticket.slot] = c;
    if (route.argued) {
        struct Given *given = &host->given[c->ticket.slot];
        if (given->capacity <= raw_size) {
            char *grown = realloc(given->text, raw_size + 1);
            if (!grown) abort();
            given->text = grown;
            given->capacity = raw_size + 1;
        }
        memcpy(given->text, raw, raw_size + 1);
        route.given = given->text;
    }
    host->routes[c->ticket.slot] = route;
    if (!write) { c->phase = WAITING; return; }
    unsigned char *ingress = c->ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes */
    c->body = ingress;
    if (route.kind == ROUTE_INPUT) {
        put_u32(ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | (lead > 4 ? DELIVERED : 0));
        if (lead > 4) {
            size_t length = strlen(c->source);
            put_u64(ingress + 4, c->sequence);
            put_u32(ingress + 12, c->batch);
            ingress[16] = (unsigned char)length;
            memcpy(ingress + 17, c->source, length);
        }
        if (c->field_bytes) memcpy(ingress + lead + c->batch * table, c->fields, c->field_bytes);
        c->body = ingress + lead + c->batch * (table + c->field_bytes);
    }
    size_t take = c->in_used < c->body_size ? c->in_used : c->body_size;
    if (take) memcpy(c->body, c->in, take);
    memmove(c->in, c->in + take, c->in_used - take);
    c->in_used -= take;
    c->received = take;
    c->phase = READING_BODY;
    if (c->received == c->body_size) body_done(c);
}

static size_t head_end(const Connection *c) {
    for (size_t i = 3; i < c->in_used; ++i)
        if (!memcmp(c->in + i - 3, "\r\n\r\n", 4)) return i + 1;
    return 0;
}

/* Advance a connection as far as its socket allows without blocking. */
static void drive(Connection *c) {
    for (;;) {
        if (c->phase == READING_HEAD) {
            size_t end = head_end(c);
            if (end) { start_request(c, end); continue; }
            if (c->in_used == HEADER_LIMIT) { refuse(c, 431, "Request Header Fields Too Large", "headers"); continue; }
            if (!c->readable) return;
            size_t room = HEADER_LIMIT - c->in_used;
            ssize_t n = recv(c->fd, c->in + c->in_used, room, 0);
            if (n > 0) {
                c->in_used += (size_t)n;
                c->deadline = c->host->now + c->host->timeout_ms;
                /* A short read emptied the socket; the edge-triggered epoll
                 * tells when more comes. Not asking again spares a recv that
                 * would fail (EAGAIN) after every request. */
                if ((size_t)n < room && !c->hung_up) c->readable = false;
                continue;
            }
            if (n < 0 && errno == EAGAIN) { c->readable = false; return; }
            if (n < 0 && errno == EINTR) continue;
            close_connection(c);
            return;
        }
        if (c->phase == READING_BODY) {
            if (!c->readable) return;
            size_t room = c->body_size - c->received;
            ssize_t n = recv(c->fd, c->body + c->received, room, 0);
            if (n > 0) {
                c->received += (size_t)n;
                c->deadline = c->host->now + c->host->timeout_ms;
                if ((size_t)n < room && !c->hung_up) c->readable = false;
                if (c->received == c->body_size) body_done(c);
                continue;
            }
            if (n < 0 && errno == EAGAIN) { c->readable = false; return; }
            if (n < 0 && errno == EINTR) continue;
            close_connection(c);
            return;
        }
        if (c->phase == WRITING) {
            if (!c->writable) return;
            size_t left = c->out_count - c->out_at;
            ssize_t n = writev(c->fd, c->out + c->out_at, (int)(left < IOV_MAX ? left : IOV_MAX));
            if (n < 0 && errno == EAGAIN) { c->writable = false; return; }
            if (n < 0 && errno == EINTR) continue;
            if (n < 0) { close_connection(c); return; }
            c->deadline = c->host->now + c->host->timeout_ms; /* a slow reader is fine while it reads */
            for (size_t sent = (size_t)n; c->out_at < c->out_count;) {
                struct iovec *v = &c->out[c->out_at];
                if (sent < v->iov_len) {
                    v->iov_base = (char *)v->iov_base + sent;
                    v->iov_len -= sent;
                    break;
                }
                sent -= v->iov_len;
                ++c->out_at;
            }
            if (c->out_at < c->out_count) continue;
            /* Answered: the next request may already be here. */
            finish_output(c);
            c->deadline = c->host->now + c->host->timeout_ms;
            if (!c->keep_alive) { close_connection(c); return; }
            c->phase = READING_HEAD;
            continue;
        }
        return; /* WAITING: the next burst answers */
    }
}

/* ---- Bursts ----------------------------------------------------------- */

static void release_copy(void *allocation, size_t size) {
    (void)size;
    free(allocation);
}

static void release_loan(void *allocation, size_t size) {
    (void)size;
    pure_mapping_release(allocation);
}

static bool allocate_ingress(void *context, size_t size, PureResource *out) {
    (void)context;
    void *bytes = malloc(size ? size : 1); /* a key press or an empty text has no payload */
    if (!bytes) return false;
    *out = (PureResource){.body = {bytes, size}, .allocation = bytes, .allocation_size = size, .release = release_copy};
    return true;
}

typedef struct { unsigned char *bytes; size_t size; const unsigned char *first; size_t spans; } Reply;

static bool measure_reply(void *context, PureSpan span) {
    Reply *r = context;
    if (!r->spans++) r->first = span.data;
    r->size += span.size;
    return true;
}

/* A value's extents, if the store mapping holds every one. */
typedef struct { struct iovec *extents; size_t count, capacity; const PureMapping *mapping; bool held; } Gather;

static bool gather_extents(void *context, PureSpan span) {
    Gather *g = context;
    if (!span.size) return true;
    g->held = g->held && pure_mapping_holds(g->mapping, span) &&
              grow(&g->extents, &g->capacity, g->count + 1, sizeof(*g->extents));
    if (g->held) g->extents[g->count++] = (struct iovec){(void *)span.data, span.size};
    return g->held;
}

/* A gathered value goes out as the mapping's extents, not copied: the answer
 * holds the mapping (charged to the budget) until it is sent and its ticket
 * released. false (and nothing held) when some extent was not stored. */
static bool lend(Applied *out, PureMapping *mapping, Gather *gather, bool read, size_t size) {
    if (!read || !gather->held) {
        free(gather->extents);
        pure_mapping_release(mapping);
        return false;
    }
    out->extents = gather->extents;
    out->extent_count = gather->count;
    out->resource = (PureResource){.body = {NULL, size}, .allocation = mapping, .allocation_size = size,
                                   .release = release_loan};
    return true;
}

static bool copy_reply(void *context, PureSpan span) {
    Reply *r = context;
    if (span.size) memcpy(r->bytes + r->size, span.data, span.size);
    r->size += span.size;
    return true;
}

/* ---- App routes -------------------------------------------------------- */

typedef struct { unsigned char *bytes; size_t size, capacity; bool failed; } Output;

static bool put_bytes(Output *o, const void *bytes, size_t size) {
    if (o->failed) return false;
    if (size > o->capacity - o->size) {
        size_t capacity = o->capacity ? o->capacity : 256;
        while (capacity - o->size < size) {
            if (capacity > SIZE_MAX / 2) { o->failed = true; return false; }
            capacity *= 2;
        }
        unsigned char *grown = realloc(o->bytes, capacity);
        if (!grown) { o->failed = true; return false; }
        o->bytes = grown;
        o->capacity = capacity;
    }
    if (size) memcpy(o->bytes + o->size, bytes, size);
    o->size += size;
    return true;
}

static bool put_span(void *context, PureSpan span) { return put_bytes(context, span.data, span.size); }

/* A JSON string's contents: quotes, backslashes and control bytes escaped,
 * other bytes (UTF-8, which Pure keeps valid) as they are. */
static bool put_json_span(void *context, PureSpan span) {
    Output *o = context;
    const unsigned char *p = span.data;
    size_t plain = 0;
    for (size_t i = 0; i < span.size; ++i) {
        unsigned char b = p[i];
        if (b >= 0x20 && b != '"' && b != '\\') continue;
        char escape[8];
        int length = b == '"' || b == '\\' ? snprintf(escape, sizeof(escape), "\\%c", b) :
                                               snprintf(escape, sizeof(escape), "\\u%04x", b);
        put_bytes(o, p + plain, i - plain);
        put_bytes(o, escape, (size_t)length);
        plain = i + 1;
    }
    return put_bytes(o, p + plain, span.size - plain);
}

static void put_number(Output *o, double value) {
    char digits[32];
    put_bytes(o, digits, pure_format_number(value, digits));
}

/* An i64's every digit (a JSON number too: a reader keeping binary64s loses those past 2^53). */
static void put_integer(Output *o, int64_t value) {
    char digits[24];
    put_bytes(o, digits, (size_t)snprintf(digits, sizeof(digits), "%" PRId64, value));
}

/* A view's arguments from a query's name=value pairs (values %XX decoded;
 * from and count page a table): a payload of their input's shape, the
 * fixed fields in its prefix, the texts and bytes after it in order, each's
 * length in the prefix but the last's. NULL for a pair naming no field or
 * given twice, a fixed field missing or not of its type, or no memory. */
static unsigned char *arguments_payload(const PureModule *m, int input, const char *given, size_t *size) {
    const PureInputShape *shape = &m->input[input];
    size_t length = given ? strlen(given) : 0;
    char *copy = malloc(length + 1);
    unsigned char *payload = malloc(shape->prefix + length + 1);
    const char *values[PURE_MODULE_FIELDS] = {0};
    long lengths[PURE_MODULE_FIELDS] = {0};
    bool ok = copy && payload;
    if (ok) {
        memcpy(copy, given ? given : "", length + 1);
        memset(payload, 0, shape->prefix);
    }
    for (char *pair = copy; ok && pair && *pair;) {
        char *end = strchr(pair, '&'), *value = strchr(pair, '=');
        if (end) *end = 0;
        if (!value || (end && value > end)) { ok = false; break; }
        *value++ = 0;
        uint32_t f = 0;
        while (f < shape->field_count && strcmp(m->field[shape->field_start + f].name, pair)) ++f;
        if (f == shape->field_count) ok = !strcmp(pair, "from") || !strcmp(pair, "count");
        else if (values[f]) ok = false;
        else if ((lengths[f] = percent_decoded(value, strlen(value))) < 0) ok = false;
        else values[f] = value;
        pair = end ? end + 1 : NULL;
    }
    size_t at = shape->prefix;
    for (uint32_t f = 0; ok && f < shape->field_count; ++f) {
        const PureField *field = &m->field[shape->field_start + f];
        if (pure_field_fixed(field->type)) {
            ok = values[f] && pure_json_fixed(field, (const unsigned char *)values[f], (size_t)lengths[f], payload);
            continue;
        }
        if (field->offset != PURE_FIELD_REST) put_u32(payload + field->offset, (uint32_t)lengths[f]);
        if (lengths[f]) memcpy(payload + at, values[f], (size_t)lengths[f]);
        at += (size_t)lengths[f];
    }
    free(copy);
    if (!ok) {
        free(payload);
        return NULL;
    }
    *size = at;
    return payload;
}

static PureResult show_given(PureApp *app, const Route *route, size_t available, Applied *answer);

/* A view shown with its arguments, when it takes some (they are the request's to give). */
static PureResult show_view(PureApp *app, const Route *route, size_t available, Applied *answer) {
    int input = pure_module_arguments(app->module, route->target);
    if (input < 0) return route->argued ? PURE_INVALID : show_given(app, route, available, answer);
    size_t size;
    unsigned char *payload = arguments_payload(app->module, input, route->given, &size);
    if (!payload) return PURE_INVALID;
    PureResult given = pure_app_arguments(app, route->target, (PureSpan){payload, size}, NULL);
    PureResult status = given == PURE_OK ? show_given(app, route, available, answer) : PURE_INVALID;
    free(payload);
    return status;
}

/* A view's value: a number in Pure's shortest form or a text, as text; a
 * table as {"rows":N,"from":F,"items":[{column: value}...]} for the rows
 * asked, each cell computed only for them (a cell without a value is null). */
static PureResult show_given(PureApp *app, const Route *route, size_t available, Applied *answer) {
    const PureModule *m = app->module;
    uint32_t view = route->target;
    PureResource *out = &answer->resource;
    const char **type = &answer->type;
    /* Room for a page of short rows at once. */
    Output o = {0};
    PureResult status = PURE_OK;
    double value;
    if (m->view_type[view] == PURE_VIEW_TABLE && !(o.bytes = malloc(o.capacity = 1024))) return PURE_ALLOCATION;
    if (m->view_type[view] == PURE_VIEW_I64) {
        int64_t integer;
        *type = "text/plain; charset=utf-8";
        if (!pure_app_view_i64(app, view, &integer)) status = PURE_NOT_FOUND;
        else put_integer(&o, integer);
    } else if (m->view_type[view] == PURE_VIEW_NUM || m->view_type[view] == PURE_VIEW_BOOL) {
        *type = "text/plain; charset=utf-8";
        if (!pure_app_view(app, view, &value)) status = PURE_NOT_FOUND;
        else if (m->view_type[view] == PURE_VIEW_BOOL) put_bytes(&o, value ? "true" : "false", value ? 4 : 5);
        else put_number(&o, value);
    } else if (m->view_type[view] == PURE_VIEW_TEXT) {
        *type = "text/plain; charset=utf-8";
        /* A text state shown as it is goes out as its stored extents. */
        int state = pure_module_text_state(m, view);
        PureDraft *text = state >= 0 ? app->texts[state] : NULL;
        PureVersion version = text ? pure_draft_version(text) : (PureVersion){0, 0};
        if (version.bytes > available) return PURE_BUDGET;
        PureMapping *mapping = version.bytes ? pure_draft_mapping(text) : NULL;
        if (mapping) {
            Gather gather = {NULL, 0, 0, mapping, true};
            bool read = pure_draft_read_version(text, version, 0, (size_t)version.bytes, gather_extents, &gather) == PURE_OK;
            if (lend(answer, mapping, &gather, read, (size_t)version.bytes)) return PURE_OK;
        }
        status = pure_app_view_text(app, view, put_span, &o);
    } else {
        *type = "application/json";
        size_t rows = pure_app_rows(app, view), from = route->from < rows ? route->from : rows;
        if (pure_app_view_fault(app) == PURE_FAULT_WORK) status = PURE_LIMIT, rows = from = 0; /* its filter's */
        size_t end = from + (route->count < rows - from ? route->count : rows - from);
        /* Each column's key, ,"name": (the first row's first without the comma), made once. */
        struct { uint32_t view; size_t length; char key[PURE_MODULE_NAME + 4]; } columns[PURE_MODULE_VIEWS];
        unsigned count = 0;
        for (uint32_t column = 0; column < m->views; ++column)
            if (m->view_table[column] == view + 1) {
                columns[count].view = column;
                columns[count].length = (size_t)snprintf(columns[count].key, sizeof(columns[count].key), ",\"%s\":",
                                                         m->view_names[column]);
                ++count;
            }
        char head[80];
        put_bytes(&o, head, (size_t)snprintf(head, sizeof(head), "{\"rows\":%zu,\"from\":%zu,\"items\":[", rows, from));
        for (size_t row = from; row < end && !o.failed; ++row) {
            put_bytes(&o, ",{" + (row == from), 2 - (row == from));
            for (unsigned k = 0; k < count; ++k) {
                put_bytes(&o, columns[k].key + !k, columns[k].length - !k);
                uint8_t kind = m->view_type[columns[k].view];
                if (kind == PURE_VIEW_I64) {
                    int64_t integer;
                    if (!pure_app_cell_i64(app, columns[k].view, row, &integer)) put_bytes(&o, "null", 4);
                    else put_integer(&o, integer);
                } else if (kind == PURE_VIEW_NUM || kind == PURE_VIEW_BOOL) {
                    if (!pure_app_cell(app, columns[k].view, row, &value)) put_bytes(&o, "null", 4);
                    else if (kind == PURE_VIEW_BOOL) put_bytes(&o, value ? "true" : "false", value ? 4 : 5);
                    else put_number(&o, value);
                } else {
                    size_t mark = o.size;
                    put_bytes(&o, "\"", 1);
                    if (pure_app_cell_text(app, columns[k].view, row, put_json_span, &o) == PURE_OK) put_bytes(&o, "\"", 1);
                    else if (!o.failed) {
                        o.size = mark;
                        put_bytes(&o, "null", 4);
                    }
                }
                if (pure_app_view_fault(app) == PURE_FAULT_WORK) { /* a cell past the work budget fails the table */
                    status = PURE_LIMIT;
                    end = row;
                    break;
                }
            }
            put_bytes(&o, "}", 1);
        }
        put_bytes(&o, "]}", 2);
    }
    if (status != PURE_OK && pure_app_view_fault(app) == PURE_FAULT_WORK) { /* past the work budget: 422 work */
        answer->fault = PURE_FAULT_WORK;
        status = PURE_LIMIT;
    }
    if (status == PURE_OK && !o.bytes && !(o.bytes = malloc(1))) o.failed = true; /* an empty value has a buffer too */
    if (status == PURE_OK && o.failed) status = PURE_ALLOCATION;
    if (status == PURE_OK && o.size > available) status = PURE_BUDGET;
    if (status != PURE_OK) {
        free(o.bytes);
        return status;
    }
    *out = (PureResource){.body = {o.bytes, o.size}, .allocation = o.bytes, .allocation_size = o.size,
                          .release = release_copy};
    return PURE_OK;
}

/* The app refused a message it was given (a fault of its own): it was
 * delivered, and is heard. Refused for memory it was not. */
static bool app_refused(const PureApp *app) { return app->fault != PURE_FAULT_NONE && app->fault != PURE_FAULT_MEMORY; }

/* An input's event: a text, a key pressed (down, then up), or a message
 * whose reply (a field of it) is borrowed from the request. */
static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) {
    const PureInputShape *shape = &app->module->input[route->target];
    uint32_t header = get_u32(effect->value.data);
    const unsigned char *delivery = effect->value.data + 4;
    if (header & STORED_RECORDS) {
        char source[PURE_SOURCE + 1];
        memcpy(source, delivery + 13, delivery[12]);
        source[delivery[12]] = 0;
        uint64_t first = get_u64(delivery), heard = pure_app_heard(app, source);
        unsigned count = get_u32(delivery + 8);
        const unsigned char *at = effect->value.data + 4 + DELIVERY;
        /* Records laid out with other fields (named after them) are laid out anew for the input. */
        size_t records = header & (STORED_RECORDS - 1), named = effect->value.size - 4 - DELIVERY - records;
        const unsigned char *fields = named ? at + records : NULL;
        out->gap = false;
        out->fault = PURE_FAULT_NONE;
        for (unsigned i = 0; i < count; ++i) {
            uint32_t size = get_u32(at);
            PureSpan record = {at + 4, size};
            at += 4 + size;
            if (first + i <= heard) continue;
            if (first + i > heard + 1) { /* one before it is missing: the sender sends from there again */
                out->gap = true;
                break;
            }
            unsigned char *recast = NULL;
            size_t recast_size = 0;
            if (fields && pure_module_recast(app->module, route->target, fields, named, record, &recast, &recast_size) !=
                              PURE_OK)
                break; /* not laid out as its fields were named: not applied, sent again */
            PureResult status = pure_app_deliver(app, route->target, source, first + i,
                                                 recast ? (PureSpan){recast, recast_size} : record, NULL);
            free(recast);
            if (status != PURE_OK && !app_refused(app)) { /* not applied: sent again */
                out->fault = app->fault;
                break;
            }
            heard = first + i;
        }
        out->heard = pure_app_heard(app, source);
        return PURE_OK; /* the save says whether they are stored */
    }
    unsigned batch = header & DELIVERED ? get_u32(delivery + 8) : 1;
    size_t lead = 4 + (header & DELIVERED ? DELIVERY : 0), table = header & PLACED ? 8 * (size_t)shape->field_count : 0;
    PureSpan payload = {effect->value.data + lead + batch * table, header & ~(PLACED | DELIVERED)};
    PureResult status;
    if (shape->kind == PURE_KIND_TEXT) {
        status = pure_app_text(app, route->target, payload);
    } else if (shape->kind == PURE_KIND_KEY) {
        status = pure_app_input(app, route->target, PURE_EDGE_DOWN);
        if (status == PURE_OK) status = pure_app_input(app, route->target, PURE_EDGE_UP);
    } else {
        uint32_t placed[2 * PURE_MODULE_FIELDS];
        for (size_t i = 0; table && i < 2 * (size_t)shape->field_count; ++i)
            placed[i] = get_u32(effect->value.data + lead + 4 * i);
        if (header & DELIVERED) {
            char source[PURE_SOURCE + 1];
            memcpy(source, delivery + 13, delivery[12]);
            source[delivery[12]] = 0;
            status = pure_app_deliver(app, route->target, source, get_u64(delivery), payload, table ? placed : NULL);
            /* The rest of a batch, in order: an app's refusal was heard and the
             * batch goes on; another failure (the budget's too) stops it, to
             * be sent again. */
            for (unsigned i = 1; i < batch && (status == PURE_OK || app_refused(app)); ++i) {
                const unsigned char *at = effect->value.data + lead + i * table;
                for (size_t k = 0; k < 2 * (size_t)shape->field_count; ++k) placed[k] = get_u32(at + 4 * k);
                PureSpan next = {payload.data + i * shape->prefix, payload.size - i * shape->prefix};
                status = pure_app_deliver(app, route->target, source, get_u64(delivery) + i, next, placed);
            }
            if (batch > 1 && app_refused(app)) status = PURE_OK;
        } else
            status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) :
                             pure_app_message(app, route->target, NULL, -1, payload);
        Reply reply = {0};
        if (status == PURE_OK) pure_app_reply(app, measure_reply, &reply);
        out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size};
    }
    out->fault = app->fault;
    return status;
}

static void apply_effect(Executor *ex, const Job *job, Applied *out) {
    PureApp *app = &ex->app;
    const PureEffect *effect = &job->effect;
    *out = (Applied){.host = job->host, .waiter = job->waiter, .ticket = effect->ticket, .input = effect->input,
                     .route = job->route.kind, .link = job->link};
    memcpy(out->key, effect->key, strlen(effect->key) + 1);
    if (job->route.kind == ROUTE_VIEW) {
        out->status = show_view(app, &job->route, effect->available, out);
        return;
    }
    if (job->route.kind == ROUTE_INPUT) {
        out->status = apply_input(app, &job->route, effect, out);
        if (out->status == PURE_OK) ++ex->burst_events;
        ex->memory_refused += out->fault == PURE_FAULT_MEMORY;
        return;
    }
    bool read = effect->input == PURE_INPUT_READ;
    out->status = pure_app_message(app, read ? ex->show : ex->edit, effect->key,
                                   read ? -1 : effect->expected, read ? (PureSpan){0} : effect->value);
    out->revision = app->revision;
    if (out->status != PURE_OK) return;
    ++ex->burst_events;
    Reply reply = {0};
    pure_app_reply(app, measure_reply, &reply);
    if (!read) {
        /* Checked at start: an edit replies a field of its request. */
        out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size};
        return;
    }
    if (reply.size > effect->available) { out->status = PURE_BUDGET; return; }
    if (!reply.size) return;
    /* A stored document is lent, extent by extent, not copied. */
    PureMapping *mapping = app->document ? pure_draft_mapping(app->document) : NULL;
    if (mapping) {
        Gather gather = {NULL, 0, 0, mapping, true};
        bool read = pure_app_reply(app, gather_extents, &gather) == PURE_OK;
        if (lend(out, mapping, &gather, read, reply.size)) return;
    }
    unsigned char *copy = malloc(reply.size);
    if (!copy) { out->status = PURE_ALLOCATION; return; }
    reply = (Reply){.bytes = copy};
    pure_app_reply(app, copy_reply, &reply);
    out->resource = (PureResource){.body = {copy, reply.size}, .allocation = copy, .allocation_size = reply.size,
                                   .release = release_copy};
}

/* ---- A surface in the node (serve/surface.h): one, run in its own thread,
 * reading published snapshots and giving events to the owner. */
struct PureSurfaceHost {
    struct Owner *owner;
    pthread_mutex_t lock;
    pthread_cond_t applied; /* a waiter's event was applied and saved (or not) */
    PureSurface table;
    PureSurfaceRun run;
    int argc, result;
    char **argv;
    pthread_t thread;
};

static void surface_changed(PureSurfaceHost *s) {
    uint64_t one = 1;
    if (write(s->table.changes, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
}

/* After a save, before its answers go: from now on views show what it
 * saved, so a read made after a write's answer sees the write. A failed save
 * returns the app to the last save's states, which the snapshot still shows. */
static void publish(Executor *ex) {
    if (!ex->publishes || (ex->published && ex->published_changes == ex->app.changes)) return;
    PureSnapshot *made = pure_app_snapshot(&ex->app), *old;
    pthread_mutex_lock(&ex->published_lock);
    old = ex->published;
    ex->published = made;
    pthread_mutex_unlock(&ex->published_lock);
    pure_snapshot_release(old);
    ex->published_changes = ex->app.changes;
    ++ex->publications;
    if (ex->surface) surface_changed(ex->surface);
}

/* An output's outlets: its wires first, then its links. Each holds a
 * window of the output's saved sends, WINDOW_SENDS and --window bytes past
 * what its receiver has (at least one): the outbox keeps the rest, and the
 * outlet takes them as its receiver takes what it holds. */
enum { WINDOW_SENDS = 65536 };
static size_t window_bytes = 4u << 20;

static uint32_t outlet_output(Executor *ex, unsigned i) {
    return i < ex->wire_count ? pure_wire_output(ex->wires[i]) : ex->link_output[i - ex->wire_count];
}

/* The saved sends not yet queued go to their outlets, in order, as far as
 * each window goes; an outlet that cannot take one takes it later. Each
 * outlet looks on from its place in the outbox, moved back by the sends
 * that left the outbox since (they may have been before it), so the sends
 * it queued before are not looked at again, however many wait. */
static void feed_outlets(Executor *ex) {
    unsigned outlets = ex->wire_count + ex->link_count;
    uint64_t gone = ex->app.sends_gone - ex->gone_seen;
    ex->gone_seen = ex->app.sends_gone;
    for (unsigned i = 0; i < outlets; ++i) {
        PureWire *wire = i < ex->wire_count ? ex->wires[i] : NULL;
        PureLink *link = wire ? NULL : ex->links[i - ex->wire_count];
        uint32_t output = outlet_output(ex, i);
        uint64_t queued = wire ? pure_wire_queued(wire) : pure_link_queued(link),
                 delivered = wire ? pure_wire_delivered(wire) : pure_link_delivered(link);
        size_t held = wire ? pure_wire_held(wire) : pure_link_held(link), at = ex->outlet_from[i];
        at = at > gone ? at - (size_t)gone : 0;
        for (; at < ex->app.sent_saved; ++at) {
            uint32_t of;
            uint64_t seq;
            PureSpan payload;
            if (!pure_app_sent(&ex->app, at, &of, &seq, &payload) || of != output || seq <= queued) continue;
            if (queued > delivered && (queued - delivered >= WINDOW_SENDS || held + payload.size > window_bytes)) {
                ++ex->windows_full;
                break;
            }
            if (wire ? !pure_wire_queue(wire, seq, payload) : !pure_link_queue(link, seq, payload.data, payload.size))
                break;
            queued = seq;
            held += payload.size;
        }
        ex->outlet_from[i] = at;
    }
}

static void after_save(Executor *ex) {
    publish(ex);
    checkpoint_round(ex);
    feed_outlets(ex);
}

enum { ANSWER_WAIT_MS = 100 };

/* The answers the wires made become events, in order, each taking its send
 * out of the outbox; one the app has no room for waits, with the wire's
 * after it, and is tried again after ANSWER_WAIT_MS. Then what the wires
 * delivered leaves the outbox (with the next save). True when an answer was
 * applied (the caller saves). */
static bool take_deliveries(Executor *ex) {
    bool answered = false;
    ex->answers_wait = false;
    for (unsigned i = 0; i < ex->wire_count; ++i) {
        PureWireAnswer answer;
        while (pure_wire_answer(ex->wires[i], &answer)) {
            ex->app.now = wall_ms();
            PureResult result = pure_app_answer(&ex->app, answer.input, pure_wire_output(ex->wires[i]), answer.source,
                                                answer.seq, answer.payload, answer.placed);
            if (result != PURE_OK && ex->app.fault == PURE_FAULT_MEMORY) {
                pure_app_finish(&ex->app);
                ++ex->memory_refused;
                ex->answers_wait = true;
                break;
            }
            if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) {
                ex->held[i] = true; /* not applied and not refused: sent again after a restart */
                fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n",
                        answer.source, answer.seq);
            }
            pure_wire_answered(ex->wires[i]);
            pure_app_finish(&ex->app);
            answered = true;
            ++ex->answers;
        }
        uint64_t delivered = pure_wire_delivered(ex->wires[i]);
        if (delivered <= ex->trimmed[i] || ex->held[i]) continue;
        pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered);
        ex->trimmed[i] = delivered;
    }
    for (unsigned i = 0; i < ex->link_count; ++i) {
        uint64_t stored = pure_link_delivered(ex->links[i]);
        if (stored <= ex->link_trimmed[i]) continue;
        pure_app_trim(&ex->app, ex->link_output[i], stored);
        ex->link_trimmed[i] = stored;
    }
    feed_outlets(ex); /* their windows moved on */
    return answered;
}

/* One save for a burst: its requests are durable with it or fail with it. */
static void save_burst(Executor *ex, Applied *applied, size_t count) {
    if (!count) return;
    for (size_t i = 0; i < count; ++i)
        ex->wrote = ex->wrote || (applied[i].status == PURE_OK && applied[i].input == PURE_INPUT_WRITE);
    struct timespec began, ended, worked, work_ended;
    clock_gettime(CLOCK_MONOTONIC, &began);
    clock_gettime(CLOCK_THREAD_CPUTIME_ID, &worked);
    PureResult saved = pure_app_save(&ex->app);
    clock_gettime(CLOCK_MONOTONIC, &ended);
    clock_gettime(CLOCK_THREAD_CPUTIME_ID, &work_ended);
    int64_t waited = (int64_t)(ended.tv_sec - began.tv_sec) * 1000000000 + (ended.tv_nsec - began.tv_nsec) -
                     ((int64_t)(work_ended.tv_sec - worked.tv_sec) * 1000000000 + (work_ended.tv_nsec - worked.tv_nsec));
    ex->save_waited = waited > 0 ? (uint64_t)waited : 0;
    ++ex->bursts;
    if (saved == PURE_OK) after_save(ex);
    for (size_t i = 0; i < count; ++i)
        if (applied[i].status == PURE_OK) applied[i].status = saved;
}

/* Run a round's requests in bursts. A burst ends before a document's key
 * repeats or before its documents would outgrow one save, so every request
 * is stored by its burst's save or fails with it; views and events of the
 * app's states join any burst. Each group of answers goes to finish.
 *
 * The owner takes a round's reads first and answers them at once: a
 * round's requests come from different connections (each has one request
 * at a time), so nothing orders them, and placed before the writes the
 * reads see only saved states (a round starts from a save). With writes
 * going on this halves the reads' wait and doubles both rates. A single
 * thread does not: it waits out each save itself, and reads answered first
 * come back first, splitting its next writes over more saves (the owner
 * gathers after a save instead). */
static void execute(Executor *ex, const Job *jobs, size_t count, void (*finish)(void *, Applied *, size_t),
                    void *context) {
    if (!count) return;
    if (!grow(&ex->applied, &ex->applied_capacity, count, sizeof(*ex->applied))) abort();
    Applied *applied = ex->applied;
    /* One look at the store per round: a request made after another shard's
     * answer was sent is read after that answer's commit. */
    bool pinned = pure_store_pin(ex->store, true) == PURE_OK;
    ex->app.now = wall_ms(); /* a round's events share its time */
    size_t reads = 0;
    for (size_t i = 0; ex->reads_first && i < count; ++i)
        if (jobs[i].effect.input == PURE_INPUT_READ) apply_effect(ex, &jobs[i], &applied[reads++]);
    if (reads) finish(context, applied, reads);
    size_t n = reads, start = reads, keys = 0;
    for (size_t i = 0; i < count; ++i) {
        if (ex->reads_first && jobs[i].effect.input == PURE_INPUT_READ) continue;
        /* A document joins a burst once (its revision moves once a save). */
        bool document = jobs[i].route.kind == ROUTE_OBJECT;
        if (document && (keys == PURE_APP_PENDING || pure_app_pending(&ex->app, jobs[i].effect.key))) {
            save_burst(ex, applied + start, n - start);
            finish(context, applied + start, n - start);
            start = n;
            keys = 0;
        }
        apply_effect(ex, &jobs[i], &applied[n++]);
        keys += document;
    }
    save_burst(ex, applied + start, n - start);
    finish(context, applied + start, n - start);
    if (pinned) pure_store_pin(ex->store, false);
}

/* Tick every timer that is due, with one save for them. A tick can change
 * states and send (the outbox is the app's state too): an app with neither
 * has nothing a tick could do, and runs on every thread, so it ticks none.
 * The milliseconds until the next is due, or -1 without timers. Missed
 * ticks are not made up. When a timer is due is the monotonic clock's
 * business (the wall clock may be set back); events see the wall clock. */
static long run_timers(Executor *ex) {
    const PureModule *m = ex->module;
    bool changes = m->states > 0;
    for (uint32_t i = 0; i < m->inputs && !changes; ++i) changes = m->input[i].kind == PURE_KIND_OUTPUT;
    if (!changes) return -1;
    uint64_t now = now_ms(), soonest = UINT64_MAX;
    bool ticked = false;
    for (uint32_t i = 0; i < m->inputs; ++i) {
        if (m->input[i].kind != PURE_KIND_TIMER) continue;
        if (!ex->next[i]) ex->next[i] = now + m->input[i].period; /* the first period starts now */
        if (ex->next[i] <= now) {
            ex->app.now = wall_ms();
            if (pure_app_tick(&ex->app, i) == PURE_OK) ++ex->ticks;
            ticked = true;
            ex->next[i] = now + m->input[i].period;
        }
        if (ex->next[i] - now < soonest) soonest = ex->next[i] - now;
    }
    if (ticked && pure_app_save(&ex->app) == PURE_OK) after_save(ex); /* a failed save undoes the ticks */
    return soonest == UINT64_MAX ? -1 : (long)soonest;
}

/* On the request's connection thread: complete its ticket and answer it if
 * its connection still waits for it. */
static void answer_applied(Host *host, Applied *a) {
    if (!pure_runtime_complete(a->ticket, a->status, a->input == PURE_INPUT_READ ? &a->resource : NULL,
                               a->status == PURE_OK ? a->reply : (PureSpan){0}, a->revision)) abort();
    Connection *c = host->by_slot[a->ticket.slot];
    if (c && c->phase == WAITING && c->ticket.generation == a->ticket.generation) {
        PureOutcome result;
        if (!pure_runtime_result(a->ticket, &result)) abort();
        /* A lent answer's body is its extents (the runtime charged them). */
        if (result.status == PURE_OK && a->extents)
            respond_typed(c, 200, "OK", "none", result.revision, a->type ? a->type : "application/octet-stream",
                          a->extents, a->extent_count, result.body.size);
        else if (a->route != ROUTE_OBJECT) answer_app(c, result, a->route == ROUTE_VIEW, a->fault, a->type);
        else answer(c, result);
        drive(c); /* most answers fit the socket buffer at once */
    }
    free(a->extents);
}

static void answer_burst(void *context, Applied *applied, size_t count) {
    (void)context;
    for (size_t i = 0; i < count; ++i) answer_applied(applied[i].host, &applied[i]);
}

/* On the owner: give each connection thread its applied requests, and wake it. */
static void hand_back(void *context, Applied *applied, size_t count) {
    Owner *owner = context;
    PureSurfaceHost *surface = owner->executor.surface;
    bool events = false;
    for (size_t i = 0; i < count; ++i) {
        events = events || applied[i].waiter;
        LinkJob *link = applied[i].link;
        if (!link) continue; /* records a link delivered: stored now, or not */
        pure_link_done(link->listener, (PureLinkDone){link->token, applied[i].heard, applied[i].status == PURE_OK,
                                                     applied[i].gap, applied[i].fault == PURE_FAULT_MEMORY});
        free(link->ingress);
        free(link);
    }
    if (events) { /* a surface's: its waiter wakes */
        pthread_mutex_lock(&surface->lock);
        for (size_t i = 0; i < count; ++i)
            if (applied[i].waiter) applied[i].waiter->status = applied[i].status, applied[i].waiter->done = true;
        pthread_cond_broadcast(&surface->applied);
        pthread_mutex_unlock(&surface->lock);
    }
    for (unsigned h = 0; h < owner->host_count; ++h) {
        Host *host = &owner->hosts[h];
        size_t mine = 0;
        for (size_t i = 0; i < count; ++i) mine += applied[i].host == host;
        if (!mine) continue;
        pthread_mutex_lock(&host->lock);
        if (!grow(&host->done, &host->done_capacity, host->done_count + mine, sizeof(*host->done))) abort();
        for (size_t i = 0; i < count; ++i)
            if (applied[i].host == host) host->done[host->done_count++] = applied[i];
        pthread_mutex_unlock(&host->lock);
        uint64_t one = 1;
        if (write(host->wake, &one, sizeof(one)) != sizeof(one)) abort();
    }
}

/* On a connection thread: answer what the owner handed back. */
static void answer_handed(Host *host) {
    uint64_t signals;
    if (read(host->wake, &signals, sizeof(signals)) < 0 && errno != EAGAIN) abort();
    pthread_mutex_lock(&host->lock);
    Applied *done = host->done;
    size_t count = host->done_count, capacity = host->done_capacity;
    host->done = host->answering;
    host->done_capacity = host->answering_capacity;
    host->done_count = 0;
    pthread_mutex_unlock(&host->lock);
    host->answering = done;
    host->answering_capacity = capacity;
    for (size_t i = 0; i < count; ++i) answer_applied(host, &done[i]);
    host->pending -= count;
}

/* The owner's newest snapshot, held, shown by this thread's reader; NULL
 * when there is none. Taken for each round, after the round's answers from
 * the owner: those were handed back after their saves' snapshots. */
static PureSnapshot *read_published(Host *host) {
    Executor *ex = &host->owner->executor;
    pthread_mutex_lock(&ex->published_lock);
    PureSnapshot *snapshot = ex->published;
    if (snapshot) pure_snapshot_hold(snapshot);
    pthread_mutex_unlock(&ex->published_lock);
    if (snapshot) pure_app_read(&host->reader, snapshot);
    return snapshot;
}

/* ---- Queries: PUT /queries/NAME registers a module of the app's states and
 * views of its own (aup query), checked when it comes: it must load (be
 * verified) with its states laid out as the app's, and have a view NAME.
 * GET /queries/NAME shows that view from the last save's snapshot, on the
 * connection thread, as the app's views are: a query never waits for, nor
 * holds up, the owner's writes. */
enum { QUERIES = 64 };
typedef struct { PureModule module; unsigned refs; } Held;
static struct { char name[65]; Held *held; } queries[QUERIES];
static pthread_mutex_t queries_lock = PTHREAD_MUTEX_INITIALIZER;

static void let_go_query(Held *held) {
    pthread_mutex_lock(&queries_lock);
    bool last = !--held->refs;
    pthread_mutex_unlock(&queries_lock);
    if (!last) return;
    pure_module_release(&held->module);
    free(held);
}

/* Whether a's states are laid out as b's: names, kinds, what each refers to,
 * keys, and the records and document collections behind them. */
static bool same_states(const PureModule *a, const PureModule *b) {
    if (a->states != b->states || a->records != b->records || a->record_fields != b->record_fields ||
        a->collections != b->collections)
        return false;
    for (uint32_t i = 0; i < a->states; ++i)
        if (strcmp(a->state_names[i], b->state_names[i]) || a->state_type[i] != b->state_type[i] ||
            a->state_ref[i] != b->state_ref[i] || a->state_flags[i] != b->state_flags[i] || a->state_key[i] != b->state_key[i])
            return false;
    for (uint32_t i = 0; i < a->records; ++i)
        if (strcmp(a->record_names[i], b->record_names[i]) || a->record[i].first != b->record[i].first ||
            a->record[i].count != b->record[i].count)
            return false;
    for (uint32_t i = 0; i < a->record_fields; ++i)
        if (strcmp(a->record_field_names[i], b->record_field_names[i]) || a->record_field_type[i] != b->record_field_type[i])
            return false;
    for (uint32_t i = 0; i < a->collections; ++i)
        if (strcmp(a->collection_names[i], b->collection_names[i])) return false;
    return true;
}

static PureResult register_query(const Host *host, const Job *job) {
    Held *held = calloc(1, sizeof(*held));
    char error[128];
    if (!held) return PURE_ALLOCATION;
    if (pure_module_load(job->effect.value.data, job->effect.value.size, &held->module, error, sizeof(error)) != PURE_OK) {
        free(held);
        return PURE_INVALID;
    }
    PureResult status = !same_states(&held->module, host->module)          ? PURE_CONFLICT
                        : pure_module_view(&held->module, job->effect.key) < 0 ? PURE_NOT_FOUND
                                                                               : PURE_OK;
    Held *old = NULL;
    held->refs = 1;
    if (status == PURE_OK) {
        pthread_mutex_lock(&queries_lock);
        int slot = -1;
        for (int i = 0; i < QUERIES && slot < 0; ++i)
            if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) slot = i;
        for (int i = 0; i < QUERIES && slot < 0; ++i)
            if (!queries[i].held) slot = i;
        if (slot < 0) status = PURE_LIMIT;
        else {
            old = queries[slot].held;
            snprintf(queries[slot].name, sizeof(queries[slot].name), "%s", job->effect.key);
            queries[slot].held = held;
        }
        pthread_mutex_unlock(&queries_lock);
    }
    if (status != PURE_OK) let_go_query(held);
    if (old) let_go_query(old);
    return status;
}

static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a) {
    Held *held = NULL;
    pthread_mutex_lock(&queries_lock);
    for (int i = 0; i < QUERIES && !held; ++i)
        if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) (held = queries[i].held)->refs++;
    pthread_mutex_unlock(&queries_lock);
    if (!held) return PURE_NOT_FOUND;
    if (!snapshot) {
        let_go_query(held);
        return PURE_BUSY;
    }
    PureApp reader = {0};
    pure_app_read(&reader, snapshot);
    pure_app_work(&reader, work_budget);
    reader.module = &held->module; /* the app's states, the query's views */
    Route route = job->route;
    route.target = (uint32_t)pure_module_view(&held->module, job->effect.key);
    PureResult status = show_view(&reader, &route, job->effect.available, a);
    pure_app_stop(&reader);
    let_go_query(held);
    return status;
}

/* Every request made ready since the last round: run here, or handed to the
 * owner. Views are shown here from the snapshot when there is one, after the
 * rest went to the owner, which meanwhile applies and saves them. Answering
 * may make more ready; they form the next round. */
static void serve_bursts(Host *host) {
    for (;;) {
        size_t count = 0;
        PureEffect effect;
        while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT)
            host->jobs[count++] = (Job){host, effect, host->routes[effect.ticket.slot], NULL, NULL};
        if (!count) return;
        if (!host->owner) {
            execute(&host->local, host->jobs, count, answer_burst, NULL);
            continue;
        }
        PureSnapshot *snapshot = NULL;
        size_t views = 0, queried = 0;
        for (size_t i = 0; i < count; ++i) {
            views += host->jobs[i].route.kind == ROUTE_VIEW;
            queried += host->jobs[i].route.kind == ROUTE_QUERY;
        }
        if ((views || queried) && !(snapshot = read_published(host))) views = 0; /* the owner shows them */
        size_t handed = count - views - queried; /* queries are all this thread's */
        if (handed) {
            Owner *owner = host->owner;
            pthread_mutex_lock(&owner->lock);
            if (!grow(&owner->jobs, &owner->capacity, owner->count + handed, sizeof(*owner->jobs))) abort();
            for (size_t i = 0; i < count; ++i)
                if ((!snapshot || host->jobs[i].route.kind != ROUTE_VIEW) && host->jobs[i].route.kind != ROUTE_QUERY)
                    owner->jobs[owner->count++] = host->jobs[i];
            owner->arrivals += handed;
            pthread_cond_signal(&owner->waiting);
            pthread_mutex_unlock(&owner->lock);
            host->pending += handed;
        }
        if (!views && !queried) return;
        /* Shown first and answered after, as a burst is. */
        if (!grow(&host->views, &host->views_capacity, views + queried, sizeof(*host->views))) abort();
        size_t shown = 0;
        for (size_t i = 0; i < count; ++i) {
            const Job *job = &host->jobs[i];
            if (job->route.kind != ROUTE_QUERY && (job->route.kind != ROUTE_VIEW || !snapshot)) continue;
            Applied *a = &host->views[shown++];
            bool registering = job->route.kind == ROUTE_QUERY && job->effect.input == PURE_INPUT_WRITE;
            *a = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input,
                           .route = registering ? ROUTE_QUERY : ROUTE_VIEW};
            a->status = job->route.kind == ROUTE_VIEW ? show_view(&host->reader, &job->route, job->effect.available, a)
                        : registering              ? register_query(host, job)
                                                   : run_query(job, snapshot, a);
        }
        pure_snapshot_release(snapshot); /* what answers lend holds its own mapping */
        for (size_t i = 0; i < shown; ++i) answer_applied(host, &host->views[i]);
        host->shown += shown;
    }
}

/* The owner's thread: take every waiting request of every connection thread
 * as one round, until they have all stopped. */
static struct timespec later(struct timespec t, long nanoseconds) {
    t.tv_sec += nanoseconds / 1000000000;
    t.tv_nsec += nanoseconds % 1000000000;
    if (t.tv_nsec >= 1000000000) { ++t.tv_sec; t.tv_nsec -= 1000000000; }
    return t;
}

static bool before(struct timespec a, struct timespec b) {
    return a.tv_sec < b.tv_sec || (a.tv_sec == b.tv_sec && a.tv_nsec < b.tv_nsec);
}

static void *run_owner(void *context) {
    Owner *owner = context;
    Job *spare = NULL;
    size_t spare_capacity = 0;
    bool gather = false;
    uint64_t answered_at = 0; /* arrivals when the last round's answers were handed back */
    for (;;) {
        pthread_mutex_lock(&owner->lock);
        /* Wait for requests, or until the next timer is due; trim what the
         * wires delivered meanwhile. */
        for (;;) {
            bool delivered = owner->delivered;
            owner->delivered = false;
            pthread_mutex_unlock(&owner->lock);
            long due = run_timers(&owner->executor);
            bool answered = (delivered || owner->executor.answers_wait) && take_deliveries(&owner->executor);
            if (owner->executor.answers_wait && (due < 0 || due > ANSWER_WAIT_MS)) due = ANSWER_WAIT_MS;
            pthread_mutex_lock(&owner->lock);
            /* Answers are saved with the next round, or now if none waits. */
            if (answered && !owner->count && !owner->stopping) {
                pthread_mutex_unlock(&owner->lock);
                if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor);
                pthread_mutex_lock(&owner->lock);
            }
            if (owner->count || owner->stopping) break;
            if (owner->delivered) continue;
            if (due < 0) {
                pthread_cond_wait(&owner->waiting, &owner->lock);
                continue;
            }
            struct timespec wake = pure_wait_until(due * 1000000LL);
            pthread_cond_timedwait(&owner->waiting, &owner->lock, &wake);
        }
        /* After a round that saved, its answers bring requests back from the
         * connection threads: gather them until none has come for 100 µs (a
         * millisecond at most), so they share one save as they would on a
         * single thread, where answering takes that long anyway. The quiet
         * counts only once a request came after the answers: requests that
         * waited out the save alone would otherwise start the next round
         * without the clients just answered, and the clients would settle
         * into two halves taking turns, each save carrying half of them.
         * Gathering never outlasts the time the last save waited for its
         * device, nor does the quiet: waiting pays only while it costs less
         * than the wait a save of the late requests would take on its own
         * (a sync of milliseconds on a flash card; in RAM next to none, and
         * there gathering would only leave the owner idle). */
        uint64_t waited = owner->executor.save_waited, budget = waited < 1000000 ? waited : 1000000;
        struct timespec now, end;
        clock_gettime(PURE_WAIT_CLOCK, &now);
        end = later(now, (long)budget);
        while (gather && owner->count && !owner->stopping && before(now, end)) {
            size_t seen = owner->count;
            bool returned = owner->arrivals != answered_at;
            struct timespec quiet = later(now, budget < 100000 ? (long)budget : 100000);
            pthread_cond_timedwait(&owner->waiting, &owner->lock, returned && before(quiet, end) ? &quiet : &end);
            if (returned && owner->count == seen) break;
            clock_gettime(PURE_WAIT_CLOCK, &now);
        }
        if (!owner->count) {
            pthread_mutex_unlock(&owner->lock);
            break;
        }
        Job *jobs = owner->jobs;
        size_t count = owner->count, capacity = owner->capacity;
        owner->jobs = spare;
        owner->capacity = spare_capacity;
        owner->count = 0;
        pthread_mutex_unlock(&owner->lock);
        owner->executor.wrote = false;
        execute(&owner->executor, jobs, count, hand_back, owner);
        gather = owner->executor.wrote;
        pthread_mutex_lock(&owner->lock);
        answered_at = owner->arrivals;
        pthread_mutex_unlock(&owner->lock);
        spare = jobs;
        spare_capacity = capacity;
    }
    free(spare);
    return NULL;
}

/* The surface's table (serve/surface.h). */
static PureSnapshot *surface_read(PureSurfaceHost *s, PureApp *reader) {
    Executor *ex = &s->owner->executor;
    pthread_mutex_lock(&ex->published_lock);
    PureSnapshot *snapshot = ex->published;
    if (snapshot) pure_snapshot_hold(snapshot);
    pthread_mutex_unlock(&ex->published_lock);
    if (snapshot) pure_app_read(reader, snapshot);
    return snapshot;
}

static void surface_release(PureSurfaceHost *s, PureSnapshot *snapshot) {
    (void)s;
    pure_snapshot_release(snapshot);
}

/* An event queued at the owner as a request is, waited for until its save. */
static PureResult surface_event(PureSurfaceHost *s, uint32_t input, PureSpan text) {
    Owner *owner = s->owner;
    const PureModule *m = owner->executor.module;
    if (input >= m->inputs || (m->input[input].kind != PURE_KIND_KEY && m->input[input].kind != PURE_KIND_TEXT) ||
        (text.size && !text.data))
        return PURE_INVALID;
    if (text.size > PURE_OBJECT_LIMIT) return PURE_LIMIT;
    unsigned char *value = malloc(4 + text.size); /* as a request's ingress: its size, then its bytes */
    if (!value) return PURE_ALLOCATION;
    put_u32(value, (uint32_t)text.size);
    if (text.size) memcpy(value + 4, text.data, text.size);
    Waiter waiter = {PURE_INVALID, false};
    Job job = {.route = {.kind = ROUTE_INPUT, .target = input}, .waiter = &waiter,
               .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {value, 4 + text.size}}};
    pthread_mutex_lock(&owner->lock);
    bool queued = !owner->stopping && grow(&owner->jobs, &owner->capacity, owner->count + 1, sizeof(*owner->jobs));
    if (queued) {
        owner->jobs[owner->count++] = job;
        ++owner->arrivals;
        pthread_cond_signal(&owner->waiting);
    }
    pthread_mutex_unlock(&owner->lock);
    pthread_mutex_lock(&s->lock);
    while (queued && !waiter.done) pthread_cond_wait(&s->applied, &s->lock);
    pthread_mutex_unlock(&s->lock);
    free(value);
    return queued ? waiter.status : PURE_INVALID;
}

static PureResult surface_press(PureSurfaceHost *s, uint32_t input) {
    return input < s->owner->executor.module->inputs &&
                   s->owner->executor.module->input[input].kind == PURE_KIND_KEY
               ? surface_event(s, input, (PureSpan){0})
               : PURE_INVALID;
}

static PureResult surface_type(PureSurfaceHost *s, uint32_t input, PureSpan text) {
    return input < s->owner->executor.module->inputs &&
                   s->owner->executor.module->input[input].kind == PURE_KIND_TEXT
               ? surface_event(s, input, text)
               : PURE_INVALID;
}

static bool surface_stopping(PureSurfaceHost *s) {
    (void)s;
    return atomic_load(&stopping);
}

static void *run_surface(void *context) {
    PureSurfaceHost *s = context;
    s->result = s->run(&s->table, s->argc, s->argv);
    fprintf(stderr, "{\"surface\":\"%s\",\"returned\":%d}\n", s->argv[0], s->result);
    return NULL;
}

/* --surface LIBRARY [-- ARGUMENTS]: mapped now, run in its own thread. */
static bool start_surface(PureSurfaceHost *s, Owner *owner, int argc, char **argv) {
    void *library = dlopen(argv[0], RTLD_NOW | RTLD_LOCAL);
    PureSurfaceRun run = library ? (PureSurfaceRun)(uintptr_t)dlsym(library, "pure_surface_run") : NULL;
    if (!run) {
        fprintf(stderr, "--surface %s: %s\n", argv[0], library ? "no pure_surface_run in it" : dlerror());
        return false;
    }
    *s = (PureSurfaceHost){.owner = owner, .run = run, .argc = argc, .argv = argv};
    s->table = (PureSurface){PURE_SURFACE_VERSION, sizeof(PureApp), sizeof(PureModule), owner->executor.module, s,
                             eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC), surface_read, surface_release, surface_press,
                             surface_type, surface_stopping};
    if (s->table.changes < 0 || pthread_mutex_init(&s->lock, NULL) || pthread_cond_init(&s->applied, NULL)) return false;
    owner->executor.surface = s; /* before the thread: its first snapshot may come any time */
    if (pthread_create(&s->thread, NULL, run_surface, s)) {
        owner->executor.surface = NULL;
        return false;
    }
    return true;
}

/* A write's reply is borrowed from its request, so the handler of an input
 * served by PUT or POST may reply only a field of its message. */
static bool replies_borrow(const PureModule *m, uint32_t input) {
    for (uint32_t h = 0; h < m->handlers; ++h) {
        if (m->handler[h].event != 1 + 2 * input) continue;
        const PureBlock *code = &m->handler[h].code;
        /* A field replied as it is, or a text field as its bytes. */
        for (uint32_t pc = code->start; pc < code->start + code->length; ++pc) {
            if (m->code[pc].op != PURE_OP_REPLY) continue;
            uint32_t from = pc > code->start && m->code[pc - 1].op == PURE_OP_BYTES_FROM ? pc - 1 : pc;
            if (from == code->start || m->code[from - 1].op != PURE_OP_FIELD) return false;
        }
    }
    return true;
}

/* ---- Main ------------------------------------------------------------- */

static bool load_module(PureModule *m, const char *path, bool *documents, uint32_t *show_out, uint32_t *edit_out) {
    char error[160];
    if (pure_module_read(path, m, error, sizeof(error)) != PURE_OK) {
        fprintf(stderr, "Module: %s\n", error);
        return false;
    }
    /* Documents are served when the module keeps a keyed collection: its keyed
     * messages show and edit read and write one by key. Without one, inputs
     * named show and edit are messages as any other. */
    int show = pure_module_input(m, "show"), edit = pure_module_input(m, "edit");
    *documents = m->collections > 0;
    if (*documents && (show < 0 || edit < 0 || m->input[show].kind != PURE_KIND_MESSAGE || !m->input[show].keyed ||
                       m->input[edit].kind != PURE_KIND_MESSAGE || !m->input[edit].keyed)) {
        fputs("Module: documents need keyed messages named show and edit\n", stderr);
        return false;
    }
    for (uint32_t i = 0; i < m->inputs; ++i) {
        bool served = (*documents && i == (uint32_t)edit) || (m->input[i].kind == PURE_KIND_MESSAGE && !m->input[i].keyed);
        if (served && !replies_borrow(m, i)) {
            fprintf(stderr, "Module: %s may reply only a field of its message\n", m->input_names[i]);
            return false;
        }
    }
    *show_out = (uint32_t)(show >= 0 ? show : 0);
    *edit_out = (uint32_t)(edit >= 0 ? edit : 0);
    return true;
}

static void accept_all(Host *host) {
    for (;;) {
        int fd = accept4(host->listener, NULL, NULL, SOCK_NONBLOCK | SOCK_CLOEXEC);
        if (fd < 0) return;
        if (host->active == host->capacity) { close(fd); continue; } /* full: the peer sees the close */
        Connection *c = host->spare;
        if (c) host->spare = c->next;
        else if (!(c = malloc(sizeof(*c)))) { close(fd); continue; }
        size_t index = 0;
        while (host->table[index]) ++index; /* a free index exists: active < capacity */
        int one = 1;
        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
        c->host = host;
        c->index = index;
        c->fd = fd;
        c->phase = READING_HEAD;
        c->readable = c->writable = c->keep_alive = true;
        c->hung_up = false;
        c->deadline = host->now + host->timeout_ms;
        c->ticket = (PureTicket){0};
        c->out = NULL;
        c->out_count = c->out_at = 0;
        c->body = NULL;
        c->opened = 0;
        c->in_used = 0;
        struct epoll_event event = {.events = EPOLLIN | EPOLLOUT | EPOLLRDHUP | EPOLLET, .data.ptr = c};
        if (epoll_ctl(host->epoll, EPOLL_CTL_ADD, fd, &event)) {
            close(fd);
            c->next = host->spare;
            host->spare = c;
            continue;
        }
        host->table[index] = c;
        ++host->active;
        ++host->accepted;
        drive(c);
    }
}


static bool start_executor(Executor *ex, const PureModule *module, uint32_t show, uint32_t edit, const char *store) {
    ex->module = module;
    ex->show = show;
    ex->edit = edit;
    if (pure_store_open(store, &ex->store) != PURE_OK) { fputs("Cannot open the store\n", stderr); return false; }
    if (pure_app_start(&ex->app, module, pure_store_data(ex->store)) != PURE_OK || !pure_app_defer(&ex->app, true)) {
        fprintf(stderr, "Cannot start the module on the store%s%s\n", ex->app.why[0] ? ": " : "", ex->app.why);
        return false;
    }
    pure_app_background_checkpoints(&ex->app, true);
    pure_app_memory(&ex->app, memory_budget);
    pure_app_work(&ex->app, work_budget);
    return true;
}

static void stop_executor(Executor *ex) {
    stop_writer(ex);
    pure_snapshot_release(ex->published);
    pure_app_stop(&ex->app);
    pure_store_close(ex->store);
    free(ex->applied);
}

/* One connection thread: its own listener on the shared port, epoll set,
 * runtime domain and, unless an owner runs the app for all of them, app and
 * store handle. */
static bool start_shard(Host *host, const PureModule *module, bool documents, uint32_t show, uint32_t edit,
                        const char *store, size_t budget, unsigned *port) {
    host->module = module;
    host->documents = documents;
    host->wake = -1;
    pure_app_work(&host->reader, work_budget); /* the views it shows from snapshots */
    if (!host->owner && !start_executor(&host->local, module, show, edit, store)) return false;
    host->work = calloc(host->capacity, sizeof(*host->work));
    host->table = calloc(host->capacity, sizeof(*host->table));
    host->by_slot = calloc(host->capacity, sizeof(*host->by_slot));
    host->jobs = calloc(host->capacity, sizeof(*host->jobs));
    host->routes = calloc(host->capacity, sizeof(*host->routes));
    host->given = calloc(host->capacity, sizeof(*host->given));
    if (!host->work || !host->table || !host->by_slot || !host->jobs || !host->routes || !host->given) {
        fputs("Cannot allocate connections\n", stderr);
        return false;
    }
    if (!pure_runtime_init(&host->runtime, host->work, host->capacity, host, (PureHost){.allocate = allocate_ingress},
                           budget) ||
        !pure_runtime_enable_async(&host->runtime)) return false;
    host->listener = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
    int one = 1;
    setsockopt(host->listener, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
    setsockopt(host->listener, SOL_SOCKET, SO_REUSEPORT, &one, sizeof(one));
    struct sockaddr_in address = {.sin_family = AF_INET, .sin_port = htons((uint16_t)*port),
                                  .sin_addr.s_addr = htonl(INADDR_LOOPBACK)};
    socklen_t address_size = sizeof(address);
    if (host->listener < 0 || bind(host->listener, (struct sockaddr *)&address, sizeof(address)) ||
        listen(host->listener, 4096) || getsockname(host->listener, (struct sockaddr *)&address, &address_size)) {
        perror("listen");
        return false;
    }
    *port = ntohs(address.sin_port); /* the first shard picks it, the others join it */
    host->epoll = epoll_create1(EPOLL_CLOEXEC);
    struct epoll_event listen_event = {.events = EPOLLIN, .data.ptr = NULL};
    if (host->epoll < 0 || epoll_ctl(host->epoll, EPOLL_CTL_ADD, host->listener, &listen_event)) {
        perror("epoll");
        return false;
    }
    /* The owner says it handed requests back through an eventfd. */
    if (host->owner) {
        struct epoll_event wake_event = {.events = EPOLLIN, .data.ptr = &host->wake};
        if (pthread_mutex_init(&host->lock, NULL) || (host->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) < 0 ||
            epoll_ctl(host->epoll, EPOLL_CTL_ADD, host->wake, &wake_event)) {
            perror("eventfd");
            return false;
        }
    }
    return true;
}

static void *run_shard(void *context) {
    Host *host = context;
    struct epoll_event events[EVENTS];
    host->now = now_ms();
    uint64_t next_sweep = host->now + 1000;
    while (!atomic_load(&stopping)) {
        /* A thread that runs the app ticks its timers on time. */
        long due = host->owner ? -1 : run_timers(&host->local);
        int ready = epoll_wait(host->epoll, events, EVENTS, due >= 0 && due < 1000 ? (int)due : 1000);
        host->now = now_ms();
        for (int e = 0; e < ready; ++e) {
            Connection *c = events[e].data.ptr;
            if (!c) { accept_all(host); continue; }
            if (events[e].data.ptr == &host->wake) { answer_handed(host); continue; }
            if (c->phase == CLOSED) continue;
            if (events[e].events & EPOLLERR) { close_connection(c); continue; }
            if (events[e].events & (EPOLLIN | EPOLLRDHUP | EPOLLHUP)) c->readable = true;
            if (events[e].events & (EPOLLRDHUP | EPOLLHUP)) c->hung_up = true;
            if (events[e].events & EPOLLOUT) c->writable = true;
            drive(c);
        }
        serve_bursts(host);
        uint64_t now = host->now;
        if (now >= next_sweep) {
            next_sweep = now + 1000;
            for (size_t i = 0; i < host->capacity; ++i) {
                Connection *c = host->table[i];
                if (c && c->phase != WAITING && now >= c->deadline) { ++host->expired; close_connection(c); }
            }
        }
        /* The round is over: closed connections can be reused. */
        while (host->closed) {
            Connection *c = host->closed;
            host->closed = c->next;
            c->next = host->spare;
            host->spare = c;
        }
    }
    pure_runtime_stop(&host->runtime);
    serve_bursts(host);
    /* What the owner still runs is answered before the connections close. */
    while (host->pending) {
        struct pollfd wake = {host->wake, POLLIN, 0};
        if (poll(&wake, 1, -1) > 0) answer_handed(host);
    }
    for (size_t i = 0; i < host->capacity; ++i) if (host->table[i]) close_connection(host->table[i]);
    return NULL;
}

static void stop_shard(Host *host) {
    if (!host->owner) stop_executor(&host->local);
    pure_app_stop(&host->reader);
    if (host->listener >= 0) close(host->listener);
    if (host->epoll >= 0) close(host->epoll);
    if (host->wake >= 0) close(host->wake);
    for (unsigned i = 0; i < SESSIONS; ++i) pure_opener_free(host->sessions[i].opener);
    while (host->closed) {
        Connection *c = host->closed;
        host->closed = c->next;
        c->next = host->spare;
        host->spare = c;
    }
    for (Connection *c = host->spare, *next; c; c = next) { next = c->next; free(c); }
    free(host->work);
    free(host->table);
    free(host->by_slot);
    free(host->jobs);
    free(host->routes);
    for (size_t i = 0; host->given && i < host->capacity; ++i) free(host->given[i].text);
    free(host->given);
    free(host->done);
    free(host->answering);
    free(host->views);
}

/* The input a link's stream goes to: one taking messages (not keyed) of the
 * sender's shape, or, named its fields, one that takes them by name
 * (pure_module_takes: each record is laid out anew for it). */
static int link_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
    const PureModule *m = ((Owner *)context)->executor.app.module;
    int index = pure_module_input(m, name);
    if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE || m->input[index].keyed) return -PURE_LINK_NO_INPUT;
    if (pure_module_shape(m, (uint32_t)index) == shape) return index;
    if (!fields) return -PURE_LINK_OTHER_SHAPE;
    return pure_module_takes(m, (uint32_t)index, fields, size) ? index : -PURE_LINK_FIELDS_LACKED;
}

/* Records a link delivered, queued at the owner as a request is: applied
 * with the next burst and done (pure_link_done) once its save is over. */
static PureLinkListener *link_listener;
static bool link_deliver(void *context, const PureLinkDelivery *d) {
    Owner *owner = context;
    size_t size = 4 + DELIVERY + d->size + d->fields_size, length = strlen(d->source);
    unsigned char *ingress = malloc(size);
    LinkJob *link = malloc(sizeof(*link));
    if (!ingress || !link || d->size > (STORED_RECORDS - 1) || length > PURE_SOURCE) {
        free(ingress);
        free(link);
        return false;
    }
    memset(ingress, 0, 4 + DELIVERY);
    put_u32(ingress, (uint32_t)d->size | DELIVERED | STORED_RECORDS);
    put_u64(ingress + 4, d->first);
    put_u32(ingress + 12, d->count);
    ingress[16] = (unsigned char)length;
    memcpy(ingress + 17, d->source, length);
    memcpy(ingress + 4 + DELIVERY, d->records, d->size);
    if (d->fields_size) memcpy(ingress + 4 + DELIVERY + d->size, d->fields, d->fields_size); /* laid out with those */
    *link = (LinkJob){link_listener, d->token, ingress};
    Job job = {.route = {.kind = ROUTE_INPUT, .target = d->input}, .link = link,
               .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}};
    pthread_mutex_lock(&owner->lock);
    bool queued = !owner->stopping && grow(&owner->jobs, &owner->capacity, owner->count + 1, sizeof(*owner->jobs));
    if (queued) {
        owner->jobs[owner->count++] = job;
        ++owner->arrivals;
        pthread_cond_signal(&owner->waiting);
    }
    pthread_mutex_unlock(&owner->lock);
    if (!queued) {
        free(ingress);
        free(link);
    }
    return queued;
}

/* A wire delivered a send: the owner trims it. */
static void wire_delivered(void *context) {
    Owner *owner = context;
    pthread_mutex_lock(&owner->lock);
    owner->delivered = true;
    pthread_cond_signal(&owner->waiting);
    pthread_mutex_unlock(&owner->lock);
}

/* --send OUTPUT=URL: that output's saved sends go to url, from NAME.OUTPUT. */
static unsigned char own_public[PURE_SEAL_KEY]; /* this node's key's public half, when it has one */
static bool open_wires(Executor *ex, const PureModule *m, const char *const *sends, unsigned count, const char *name) {
    for (unsigned i = 0; i < count; ++i) {
        const char *equals = strchr(sends[i], '=');
        char output[PURE_MODULE_NAME], source[PURE_SOURCE + 2], error[160];
        size_t length = equals ? (size_t)(equals - sends[i]) : 0;
        if (!length || length >= sizeof(output)) {
            fprintf(stderr, "--send takes OUTPUT=URL, not %s\n", sends[i]);
            return false;
        }
        memcpy(output, sends[i], length);
        output[length] = 0;
        int written = snprintf(source, sizeof(source), "%s.%s", name, output);
        if (!strncmp(equals + 1, "pure://", 7)) { /* a link: no connection, sealed with the node's key */
            int index = pure_module_input(m, output);
            if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) snprintf(error, sizeof(error), "no output '%s'", output);
            PureLink *link = index >= 0 && m->input[index].kind == PURE_KIND_OUTPUT && written > 0 && written <= PURE_SOURCE
                                 ? pure_link_open(equals + 1, source, pure_module_shape(m, (uint32_t)index), seal_key,
                                                  own_public, error, sizeof(error))
                                 : NULL;
            unsigned char named[PURE_FIELDS_NAMED];
            if (link && !pure_link_fields(link, named, pure_module_fields(m, (uint32_t)index, named))) {
                pure_link_close(link);
                link = NULL;
                snprintf(error, sizeof(error), "no memory");
            }
            if (!link) {
                fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error);
                return false;
            }
            ex->link_output[ex->link_count] = (uint32_t)index;
            ex->links[ex->link_count++] = link;
            continue;
        }
        PureWire *wire = written > 0 && written <= PURE_SOURCE ? pure_wire_open(m, output, equals + 1, source, error, sizeof(error)) : NULL;
        if (!wire) {
            fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error);
            return false;
        }
        for (unsigned k = 0; k < ex->wire_count; ++k)
            if (pure_wire_output(ex->wires[k]) == pure_wire_output(wire)) {
                fprintf(stderr, "--send %s: the output is wired already\n", sends[i]);
                pure_wire_close(wire);
                return false;
            }
        ex->wires[ex->wire_count++] = wire;
    }
    return true;
}

/* --seal OUTPUT=KEY: that output's sends sealed to the receiver's public key (hex). */
static bool wire_seals(Executor *ex, const PureModule *m, const char *const *seals, unsigned count,
                       const unsigned char *own_key) {
    char error[512];
    if (count && !pure_seal_ready(error, sizeof(error))) {
        fprintf(stderr, "--seal: %s\n", error);
        return false;
    }
    for (unsigned i = 0; i < count; ++i) {
        const char *equals = strchr(seals[i], '=');
        unsigned char peer[PURE_SEAL_KEY];
        char output[PURE_MODULE_NAME];
        size_t length = equals ? (size_t)(equals - seals[i]) : 0;
        PureWire *wire = NULL;
        PureLink *link = NULL;
        if (length && length < sizeof(output)) {
            memcpy(output, seals[i], length);
            output[length] = 0;
            for (unsigned k = 0; k < ex->wire_count; ++k)
                if (!strcmp(m->input_names[pure_wire_output(ex->wires[k])], output)) wire = ex->wires[k];
            for (unsigned k = 0; k < ex->link_count; ++k)
                if (!strcmp(m->input_names[ex->link_output[k]], output)) link = ex->links[k];
        }
        if (link && pure_seal_peer(equals + 1, peer)) { /* the link's receiver */
            pure_link_peer(link, peer);
            continue;
        }
        if (!wire || !pure_seal_peer(equals + 1, peer)) {
            fprintf(stderr, "--seal takes OUTPUT=KEY for an output given to --send, the key as 64 hex digits: %s\n", seals[i]);
            return false;
        }
        pure_wire_seal(wire, peer, seal_key, own_key); /* sealed with this node's key when it has one */
    }
    return true;
}

/* --sender NAME=KEY: deliveries from source NAME are taken only sealed with
 * that key (hex); with any, only named senders are taken. */
static bool name_senders(const char *const *named, unsigned count) {
    for (unsigned i = 0; i < count; ++i) {
        const char *equals = strchr(named[i], '=');
        size_t length = equals ? (size_t)(equals - named[i]) : 0;
        struct Sender *sender = &senders[sender_count];
        if (!length || length > PURE_SOURCE || !pure_seal_peer(equals + 1, sender->key)) {
            fprintf(stderr, "--sender takes NAME=KEY, the key as 64 hex digits: %s\n", named[i]);
            return false;
        }
        memcpy(sender->name, named[i], length);
        sender->name[length] = 0;
        ++sender_count;
    }
    return true;
}

/* --senders FILE: senders a --link-port takes besides those named, a line
 * each (NAME KEY, the key as 64 hex digits), as many as memory holds. */
static bool allow_senders(PureLinkReceiver *receiver, const char *path, char *error, size_t size) {
    if (!path) return true;
    FILE *file = fopen(path, "r");
    if (!file) {
        snprintf(error, size, "--senders %s: %s", path, strerror(errno));
        return false;
    }
    char *line = NULL;
    size_t capacity = 0, count = 0;
    bool ok = true;
    for (ssize_t length; ok && (length = getline(&line, &capacity, file)) >= 0; ++count) {
        while (length && (line[length - 1] == '\n' || line[length - 1] == '\r')) line[--length] = 0;
        char *space = strchr(line, ' ');
        unsigned char key[PURE_SEAL_KEY];
        if (!length) continue;
        if (!space || space == line || (size_t)(space - line) > PURE_SOURCE || !pure_seal_peer(space + 1, key)) {
            snprintf(error, size, "--senders %s: line %zu is not NAME KEY", path, count + 1);
            ok = false;
            break;
        }
        *space = 0;
        for (const char *c = line; *c && ok; ++c) ok = *c > ' ' && *c <= '~';
        if (!ok) snprintf(error, size, "--senders %s: line %zu names no source", path, count + 1);
        else ok = pure_link_allow(receiver, line, key, error, size);
    }
    free(line);
    fclose(file);
    return ok;
}

/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */
static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) {
    for (unsigned i = 0; i < count; ++i) {
        const char *equals = strchr(replies[i], '=');
        char output[PURE_MODULE_NAME], error[160] = "";
        size_t length = equals ? (size_t)(equals - replies[i]) : 0;
        int input = equals ? pure_module_input(m, equals + 1) : -1;
        PureWire *wire = NULL;
        if (length && length < sizeof(output)) {
            memcpy(output, replies[i], length);
            output[length] = 0;
            int index = pure_module_input(m, output);
            for (unsigned k = 0; k < ex->wire_count && index >= 0; ++k)
                if (pure_wire_output(ex->wires[k]) == (uint32_t)index) wire = ex->wires[k];
        }
        if (!wire || input < 0 || !pure_wire_reply(wire, (uint32_t)input, error, sizeof(error))) {
            fprintf(stderr, "--reply %s: %s\n", replies[i], !wire ? "takes OUTPUT=INPUT of an output given --send" :
                                                         input < 0 ? "no such input" : error);
            return false;
        }
    }
    return true;
}

/* --threads auto: a connection thread for each CPU this process may run on,
 * less the one the owner of an app with states takes. */
static unsigned auto_threads(bool owner) {
    cpu_set_t set;
    long cpus = sched_getaffinity(0, sizeof(set), &set) ? sysconf(_SC_NPROCESSORS_ONLN) : CPU_COUNT(&set);
    if (cpus < 1) cpus = 1;
    if (owner && cpus > 1) --cpus;
    return cpus > 256 ? 256 : (unsigned)cpus;
}

int main(int argc, char **argv) {
    static PureModule module;
    static Owner owner;
    static PureSurfaceHost surface;
    /* What follows -- is the surface's: its library first, then its arguments. */
    int options = argc, surface_argc = 0;
    char *surface_argv[64];
    for (int i = 1; i < argc; ++i)
        if (!strcmp(argv[i], "--")) {
            options = i;
            break;
        }
    const char *surface_path = NULL, *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
    const char *seals[PURE_MODULE_INPUTS], *named[PURE_MODULE_INPUTS], *key_path = NULL, *senders_path = NULL;
    unsigned send_count = 0, reply_count = 0, seal_count = 0, named_count = 0;
    unsigned port = 8767, threads = 1, link_port = 0;
    size_t budget = 64u << 20, capacity = 1024;
    uint64_t timeout_ms = 10000, link_idle_ms = 0;
    for (int i = 1; i < options; i += 2) {
        uint64_t value;
        if (!strcmp(argv[i], "--sealed-only")) { sealed_only = true; --i; continue; }
        if (i + 1 == options) goto usage;
        if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
        else if (!strcmp(argv[i], "--surface")) surface_path = argv[i + 1];
        else if (!strcmp(argv[i], "--key")) key_path = argv[i + 1];
        else if (!strcmp(argv[i], "--seal") && seal_count < PURE_MODULE_INPUTS) seals[seal_count++] = argv[i + 1];
        else if (!strcmp(argv[i], "--store")) store = argv[i + 1];
        else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1];
        else if (!strcmp(argv[i], "--name")) name = argv[i + 1];
        else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]);
        else if (!strcmp(argv[i], "--reply") && reply_count < PURE_MODULE_INPUTS) replies[reply_count++] = argv[i + 1];
        else if (!strcmp(argv[i], "--sender") && named_count < PURE_MODULE_INPUTS) named[named_count++] = argv[i + 1];
        else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
        else if (!strcmp(argv[i], "--link-port") && number(argv[i + 1], 65535, &value) && value) link_port = (unsigned)value;
        else if (!strcmp(argv[i], "--senders")) senders_path = argv[i + 1];
        else if (!strcmp(argv[i], "--link-idle-ms") && number(argv[i + 1], 86400000, &value) && value) link_idle_ms = value;
        else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
        else if (!strcmp(argv[i], "--memory") && number(argv[i + 1], SIZE_MAX, &value) && value) memory_budget = (size_t)value;
        else if (!strcmp(argv[i], "--work") && number(argv[i + 1], UINT64_MAX, &value) && value) work_budget = value;
        else if (!strcmp(argv[i], "--window") && number(argv[i + 1], SIZE_MAX, &value) && value) window_bytes = (size_t)value;
        else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
        else if (!strcmp(argv[i], "--threads") && !strcmp(argv[i + 1], "auto")) threads = 0;
        else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value;
        else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) timeout_ms = value;
        else goto usage;
    }
    if (surface_path) surface_argv[surface_argc++] = (char *)surface_path;
    for (int i = options + 1; i < argc && surface_argc < 63; ++i) surface_argv[surface_argc++] = argv[i];
    surface_argv[surface_argc] = NULL;
    if (options < argc && !surface_path) goto usage;
    uint32_t show, edit;
    bool documents;
    if (!module_path || !store || !load_module(&module, module_path, &documents, &show, &edit)) goto usage;
    if (!threads) threads = auto_threads(module.states);
    char public_key[2 * PURE_SEAL_KEY + 1] = "";
    if (key_path) {
        char error[256];
        if (!(seal_key = pure_seal_key(key_path, public_key, error, sizeof(error)))) { fprintf(stderr, "--key %s\n", error); return 2; }
    }
    if ((sealed_only || named_count) && !seal_key) { fputs("--sealed-only and --sender need --key\n", stderr); return 2; }
    if (!name_senders(named, named_count)) return 2;
    if (seal_key && !pure_seal_peer(public_key, own_public)) return 2;
    if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; }
    signal(SIGPIPE, SIG_IGN);
    struct sigaction stop_action = {.sa_handler = stop};
    sigemptyset(&stop_action.sa_mask);
    sigaction(SIGINT, &stop_action, NULL);
    sigaction(SIGTERM, &stop_action, NULL);
    Host *hosts = calloc(threads, sizeof(*hosts));
    pthread_t *thread = calloc(threads, sizeof(*thread)), owner_thread;
    if (!hosts || !thread) return 1;
    /* States (and outputs, whose sends are the app's state) have one owner on
     * a thread of its own, whose snapshots the connection threads read. */
    bool outputs = false;
    for (uint32_t i = 0; i < module.inputs; ++i) outputs = outputs || module.input[i].kind == PURE_KIND_OUTPUT;
    bool owned = module.states || outputs;
    /* The sender's name: --name, or the module file's name without .pmod. */
    char stem[PURE_SOURCE + 1];
    if (!name) {
        const char *base = strrchr(module_path, '/') ? strrchr(module_path, '/') + 1 : module_path;
        size_t length = strlen(base);
        if (length > 5 && !strcmp(base + length - 5, ".pmod")) length -= 5;
        snprintf(stem, sizeof(stem), "%.*s", (int)length, base);
        name = stem;
    }
    if (owned) {
        if (!start_executor(&owner.executor, &module, show, edit, store) || pthread_mutex_init(&owner.lock, NULL) ||
            pure_cond_monotonic(&owner.waiting) || pthread_mutex_init(&owner.executor.published_lock, NULL))
            return 1;
        owner.executor.reads_first = true;
        owner.executor.publishes = true;
        if (!open_wires(&owner.executor, &module, sends, send_count, name) ||
            !wire_replies(&owner.executor, &module, replies, reply_count) ||
            !wire_seals(&owner.executor, &module, seals, seal_count, seal_key ? own_public : NULL)) return 2;
        after_save(&owner.executor); /* the snapshot, and the sends stored before */
        for (unsigned i = 0; i < owner.executor.wire_count; ++i)
            if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
        for (unsigned i = 0; i < owner.executor.link_count; ++i) {
            char error[256];
            if (!pure_link_start(owner.executor.links[i], wire_delivered, &owner, error, sizeof(error))) {
                fprintf(stderr, "--send %s\n", error);
                return 2;
            }
        }
        owner.hosts = hosts;
        owner.host_count = threads;
    }
    for (unsigned i = 0; i < threads; ++i) {
        hosts[i].capacity = capacity;
        hosts[i].timeout_ms = timeout_ms;
        hosts[i].listener = hosts[i].epoll = -1;
        hosts[i].owner = owned ? &owner : NULL;
        if (!start_shard(&hosts[i], &module, documents, show, edit, store, budget, &port)) return 1;
    }
    if (owned && pthread_create(&owner_thread, NULL, run_owner, &owner)) { fputs("Cannot start the owner\n", stderr); return 1; }
    if (owned) pthread_setname_np(owner_thread, "pure-owner"); /* its CPU, seen apart: one core at most */
    if (link_port) { /* records from the senders named, over UDP */
        char error[256];
        if (!owned || !seal_key) {
            fputs("--link-port needs an app with states and the node's --key\n", stderr);
            return 2;
        }
        PureLinkReceiver *receiver =
            pure_link_receiver(seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner}, error, sizeof(error));
        bool allowed = receiver != NULL && allow_senders(receiver, senders_path, error, sizeof(error));
        for (unsigned i = 0; allowed && i < sender_count; ++i)
            allowed = pure_link_allow(receiver, senders[i].name, senders[i].key, error, sizeof(error));
        if (allowed && link_idle_ms) pure_link_idle(receiver, link_idle_ms);
        char memo[4096];
        snprintf(memo, sizeof(memo), "%s.links", key_path); /* beside the key the secrets come from */
        allowed = allowed && pure_link_remember(receiver, memo, error, sizeof(error)) &&
                  (link_listener = pure_link_listen(receiver, link_port, error, sizeof(error)));
        if (!allowed || !pure_link_serve(link_listener, error, sizeof(error))) {
            fprintf(stderr, "--link-port: %s\n", error);
            return 2;
        }
    }
    /* Its name too: its sends' sources are NAME.OUTPUT, which a receiver's --sender must name. */
    printf("{\"port\":%u,\"threads\":%u,\"name\":\"", port, threads);
    for (const char *c = name; *c; ++c) printf(*c == '"' || *c == '\\' ? "\\%c" : "%c", *c);
    if (*public_key) printf("\",\"public_key\":\"%s\"}\n", public_key);
    else printf("\"}\n");
    fflush(stdout);
    /* A surface reads the owner's snapshots and gives it events. */
    if (surface_path && !owned) { fputs("--surface needs an app with states\n", stderr); return 2; }
    if (surface_path && !start_surface(&surface, &owner, surface_argc, surface_argv)) return 1;
    for (unsigned i = 1; i < threads; ++i)
        if (pthread_create(&thread[i], NULL, run_shard, &hosts[i])) { fputs("Cannot start a thread\n", stderr); return 1; }
    run_shard(&hosts[0]);
    for (unsigned i = 1; i < threads; ++i) pthread_join(thread[i], NULL);
    if (surface_path) { /* stopped before the owner, which may still apply its last event */
        surface_changed(&surface);
        pthread_join(surface.thread, NULL);
    }
    if (owned) {
        pthread_mutex_lock(&owner.lock);
        owner.stopping = true;
        pthread_cond_signal(&owner.waiting);
        pthread_mutex_unlock(&owner.lock);
        pthread_join(owner_thread, NULL);
    }
    PureLinkCounts sent = {0}, heard = {0}; /* what links sent, and what the listener took */
    pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */
    size_t active = 0, retained = 0;
    uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0,
             ticks = owner.executor.ticks, shown = 0, checkpoints = 0, memory_refused = owner.executor.memory_refused,
             windows_full = owner.executor.windows_full;
    for (unsigned i = 0; i < threads; ++i) {
        shown += hosts[i].shown;
        active += hosts[i].active;
        retained += hosts[i].runtime.retained;
        accepted += hosts[i].accepted;
        requests += hosts[i].requests;
        bursts += hosts[i].local.bursts;
        events += hosts[i].local.burst_events;
        ticks += hosts[i].local.ticks;
        memory_refused += hosts[i].local.memory_refused;
        expired += hosts[i].expired;
        stop_shard(&hosts[i]);
    }
    uint64_t delivered = 0, attempts = 0;
    for (unsigned i = 0; i < owner.executor.wire_count; ++i) {
        uint64_t d, a;
        pure_wire_counts(owner.executor.wires[i], &d, &a);
        delivered += d;
        attempts += a;
        pure_wire_close(owner.executor.wires[i]); /* what was not delivered stays in the outbox */
    }
    for (unsigned i = 0; i < owner.executor.link_count; ++i) {
        PureLinkCounts c;
        pure_link_counts(owner.executor.links[i], &c);
        sent.packets += c.packets, sent.records += c.records, sent.resent += c.resent, sent.lost += c.lost;
        sent.acknowledged += c.acknowledged, sent.received += c.received, sent.stored += c.stored;
        sent.refused += c.refused;
        pure_link_close(owner.executor.links[i]); /* what was not stored stays in the outbox */
    }
    size_t state_bytes = owned ? pure_app_state_bytes(&owner.executor.app) : 0;
    if (owned) {
        checkpoints = owner.executor.checkpoints;
        stop_executor(&owner.executor);
        checkpoints = owner.executor.checkpoints;
        free(owner.jobs);
    }
    unsigned long groups, grouped;
    pure_store_group_counts(&groups, &grouped);
    fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64
            ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64
            ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64
            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64
            ",\"commit_groups\":%lu,\"grouped_commits\":%lu,\"checkpoints\":%" PRIu64
            ",\"link_packets\":%" PRIu64 ",\"link_records\":%" PRIu64 ",\"link_resent\":%" PRIu64
            ",\"link_lost\":%" PRIu64 ",\"link_acknowledged\":%" PRIu64 ",\"link_stored\":%" PRIu64
            ",\"link_refused\":%" PRIu64 ",\"listener_packets\":%" PRIu64 ",\"listener_received\":%" PRIu64
            ",\"listener_deliveries\":%" PRIu64 ",\"listener_woken\":%" PRIu64 ",\"listener_remembered\":%" PRIu64
            ",\"listener_swept\":%" PRIu64 ",\"listener_active\":%" PRIu64 ",\"listener_keyed\":%" PRIu64
            ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64
            ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 ",\"windows_full\":%" PRIu64 "}\n",
            active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
            owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped,
            checkpoints, sent.packets, sent.records, sent.resent, sent.lost, sent.acknowledged, sent.stored,
            sent.refused, heard.packets, heard.received, heard.deliveries, heard.woken, heard.remembered, heard.swept,
            heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused,
            windows_full);
    free(hosts);
    free(thread);
    return 0;
usage:
    fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
          "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=URL ...] [--name NAME] "
          "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] "
          "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] [--work N] "
          "[--window 4194304] [--surface LIBRARY [-- ARGUMENTS]]\n"
          "  --send OUTPUT=pure://host:port/INPUT: to a Pure node's --link-port, the way between nodes (UDP, no\n"
          "      connection, docs/design/012-link-protocol.md; its key by --seal OUTPUT=KEY)\n"
          "  --send OUTPUT=http[s]://host:port/path: to anything else, over HTTP\n"
          "  --senders FILE: more senders the --link-port takes, a line each: NAME KEY\n"
          "  --name NAME: the node's name, its sends' sources being NAME.OUTPUT (the module file's name without\n"
          "      .pmod unless given; the first line out says it)\n"
          "  --memory BYTES: what the app's states may hold (lists' items, texts, sends not yet stored): an event\n"
          "      past it is refused, 503 memory (docs/design/013-memory-budget.md)\n"
          "  --work N: what an event, or one evaluation of a view, may run, in instructions counted as written: past\n"
          "      it the event is refused and the view has no value, 422 work (docs/design/017-apps-on-one-host.md)\n"
          "  --window BYTES: what each --send holds of its sends beyond what its receiver has (and 65,536 sends);\n"
          "      the outbox keeps the rest\n", stderr);
    return 2;
}