Deliver an app's saved sends over HTTP, in order and in batches (083)

pure-serve --send OUTPUT=http://host:port/path wires an output: after
every save the owner queues the new sends as JSON (fields and groups)
and a thread per output posts them in order with X-Pure-Source,
X-Pure-Sequence and Idempotency-Key. A 2xx, or a Pure receiver's
refusal, delivers; the owner then trims. 408, 429, 5xx and no answer
are sent again after a pause growing to 10 s, other answers every 10 s:
nothing is dropped. A Pure receiver applies a delivered message with
pure_app_deliver (now taking placed fields) and advertises
X-Pure-Batch: 256; the sender then posts up to that many numbers in a
row as one JSON array (X-Pure-Count), applied in order in one burst
with one save.

Delivery kept up with orders once batched (391 deliveries/s one at a
time; a backlog of 20,000 went at 60k/s). With four clients ordering,
killing the receiver and then the sender (-9) left the receiver's count
and total equal to what the sender saved, five times and under ASan.

7 files changed, 763 insertions, 40 deletions

diff --git a/prototypes/connections/Makefile b/prototypes/connections/Makefile
index 6775e5a..60d83d7 100644
--- a/prototypes/connections/Makefile
+++ b/prototypes/connections/Makefile
@@ -44,8 +44,8 @@ $(BUILD)/pure-connections-async-gated: server.c $(HOST) flow.c program.c $(CORE)
 	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_ASYNC_EFFECTS -DPURE_HTTP_ASYNC -DPURE_STORE_TEST server.c $(HOST) flow.c program.c $(CORE) test-async-server-gate.c $(LDFLAGS) -pthread $(LDLIBS) -o $@
 
 # The HTTP host for aup modules: persistent connections, epoll, group commits.
-$(BUILD)/pure-serve: serve.c flow.c module-store.c $(MODULE) $(CORE) module-store.h $(MODULE_HEADERS) | $(BUILD)
-	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@
+$(BUILD)/pure-serve: serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) deliver.h module-store.h $(MODULE_HEADERS) | $(BUILD)
+	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_ASYNC_EFFECTS serve.c deliver.c flow.c module-store.c $(MODULE) $(CORE) $(LDFLAGS) -lm $(LDLIBS) -o $@
 
 $(BUILD)/pure-connections-fixed: server.c $(HOST) flow.c $(CORE) $(HEADERS) $(BUILD)/embedded-plan.h | $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) -DPURE_RUNTIME_METRICS -DPURE_EMBEDDED_PROGRAM -I$(BUILD) server.c $(HOST) flow.c $(CORE) $(LDFLAGS) $(LDLIBS) -o $@
diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c
new file mode 100644
index 0000000..5f32978
--- /dev/null
+++ b/prototypes/connections/deliver.c
@@ -0,0 +1,500 @@
+#define _POSIX_C_SOURCE 200809L
+#include "deliver.h"
+#include "module-text.h"
+#include <errno.h>
+#include <netdb.h>
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <strings.h>
+#include <sys/socket.h>
+#include <sys/time.h>
+#include <time.h>
+#include <unistd.h>
+
+enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 };
+enum { DELIVERED, AGAIN, REFUSED_BY_PEER };
+
+typedef struct { char *body; size_t size; uint64_t seq; } Pending;
+
+struct PureWire {
+    const PureModule *module;
+    uint32_t output;
+    char host[256], port[8], path[1024], source[PURE_SOURCE + 1];
+    pthread_mutex_t lock;
+    pthread_cond_t changed;
+    Pending *queue; /* the sends to deliver, in order, from head */
+    size_t head, count, capacity;
+    uint64_t queued, delivered, attempts, deliveries;
+    unsigned batch; /* how many a request may carry, as the receiver said (X-Pure-Batch); 0: one */
+    bool stopping, started, stuck;
+    pthread_t thread;
+    void (*notify)(void *);
+    void *context;
+    int fd; /* the connection kept for the next send, or -1; changed under the lock */
+};
+
+/* ---- A send as JSON: its fields in order, a group's fields as an object. */
+
+typedef struct { char *bytes; size_t size, capacity; bool failed; } Text;
+
+static void put(Text *t, const void *bytes, size_t size) {
+    if (t->failed) return;
+    if (size > t->capacity - t->size) {
+        size_t capacity = t->capacity ? t->capacity : 256;
+        while (size > capacity - t->size) capacity *= 2;
+        char *grown = realloc(t->bytes, capacity);
+        if (!grown) { t->failed = true; return; }
+        t->bytes = grown;
+        t->capacity = capacity;
+    }
+    memcpy(t->bytes + t->size, bytes, size);
+    t->size += size;
+}
+
+static void put_string(Text *t, const unsigned char *bytes, size_t size) {
+    put(t, "\"", 1);
+    size_t run = 0;
+    for (size_t i = 0; i < size; ++i) {
+        unsigned char b = bytes[i];
+        if (b != '"' && b != '\\' && b >= 0x20) continue;
+        put(t, bytes + run, i - run);
+        char escaped[8];
+        if (b == '"' || b == '\\') put(t, (char[]){'\\', (char)b}, 2);
+        else put(t, escaped, (size_t)snprintf(escaped, sizeof(escaped), "\\u%04x", b));
+        run = i + 1;
+    }
+    put(t, bytes + run, size - run);
+    put(t, "\"", 1);
+}
+
+static uint32_t read_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; }
+
+/* The group a field is in: the length of its name before the last dot (0: none). */
+static size_t group_of(const char *name) {
+    const char *dot = strrchr(name, '.');
+    return dot ? (size_t)(dot - name) : 0;
+}
+
+/* Whether the group path open (length open_size) holds the group of name (length size). */
+static bool holds(const char *open, size_t open_size, const char *name, size_t size) {
+    return open_size <= size && !memcmp(open, name, open_size) && (open_size == size || !open_size || name[open_size] == '.');
+}
+
+static bool encode(const PureWire *w, PureSpan payload, Text *t) {
+    const PureModule *m = w->module;
+    const PureInputShape *shape = &m->input[w->output];
+    if (payload.size < shape->prefix) return false;
+    char open[PURE_MODULE_NAME] = "";
+    size_t open_size = 0, at = shape->prefix;
+    unsigned depth = 0;
+    bool first[GROUPS + 1] = {true};
+    put(t, "{", 1);
+    for (uint32_t f = 0; f < shape->field_count; ++f) {
+        const PureField *field = &m->field[shape->field_start + f];
+        size_t group = group_of(field->name);
+        while (depth && !holds(open, open_size, field->name, group)) {
+            put(t, "}", 1);
+            --depth;
+            open_size = group_of(open) < open_size ? group_of(open) : 0;
+            open[open_size] = 0;
+        }
+        while (open_size < group) {
+            size_t start = open_size ? open_size + 1 : 0, end = start;
+            while (end < group && field->name[end] != '.') ++end;
+            if (!first[depth]) put(t, ",", 1);
+            first[depth] = false;
+            put_string(t, (const unsigned char *)field->name + start, end - start);
+            put(t, ":{", 2);
+            first[++depth] = true;
+            memcpy(open, field->name, end);
+            open[open_size = end] = 0;
+        }
+        if (!first[depth]) put(t, ",", 1);
+        first[depth] = false;
+        const char *key = field->name + (group ? group + 1 : 0);
+        put_string(t, (const unsigned char *)key, strlen(key));
+        put(t, ":", 1);
+        char digits[32];
+        if (field->type == PURE_FIELD_U32) {
+            put(t, digits, (size_t)snprintf(digits, sizeof(digits), "%u", (unsigned)read_u32(payload.data + field->offset)));
+        } else if (field->type == PURE_FIELD_NUMBER) {
+            uint64_t bits = read_u32(payload.data + field->offset) | (uint64_t)read_u32(payload.data + field->offset + 4) << 32;
+            double value;
+            memcpy(&value, &bits, sizeof(value));
+            put(t, digits, pure_format_number(value, digits));
+        } else {
+            size_t size = field->offset == PURE_FIELD_REST ? payload.size - at : read_u32(payload.data + field->offset);
+            if (size > payload.size - at) return false;
+            PureSpan value = {payload.data + at, size};
+            if (!pure_utf8_valid(value)) return false; /* bytes that are not text have no JSON form */
+            put_string(t, value.data, size);
+            at += size;
+        }
+    }
+    while (depth--) put(t, "}", 1);
+    put(t, "}", 1);
+    return !t->failed;
+}
+
+/* ---- The wire. */
+
+PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source,
+                         char *error, size_t size) {
+    int index = pure_module_input(m, output);
+    if (index < 0 || m->input[index].kind != PURE_KIND_OUTPUT) {
+        snprintf(error, size, "no output '%s'", output);
+        return NULL;
+    }
+    /* A group's fields are together, so each group is one object. */
+    const PureInputShape *shape = &m->input[index];
+    for (uint32_t f = 1; f < shape->field_count; ++f) {
+        const char *name = m->field[shape->field_start + f].name;
+        for (uint32_t e = 0; e + 1 < f; ++e) {
+            const char *earlier = m->field[shape->field_start + e].name, *between = m->field[shape->field_start + f - 1].name;
+            size_t shared = 0;
+            while (earlier[shared] && earlier[shared] == name[shared]) ++shared;
+            while (shared && earlier[shared - 1] != '.') --shared; /* the groups both are in */
+            if (shared && strncmp(between, name, shared)) {
+                snprintf(error, size, "the fields of output '%s' in one group are not together", output);
+                return NULL;
+            }
+        }
+    }
+    size_t length = strlen(source);
+    if (!length || length > PURE_SOURCE) {
+        snprintf(error, size, "a source name has 1 to %d bytes", PURE_SOURCE);
+        return NULL;
+    }
+    if (strncmp(url, "http://", 7)) {
+        snprintf(error, size, strncmp(url, "https://", 8) ? "a URL is http://host[:port]/path" :
+                                                          "https is not delivered to yet: use http");
+        return NULL;
+    }
+    const char *host = url + 7, *slash = strchr(host, '/'), *colon = memchr(host, ':', slash ? (size_t)(slash - host) : strlen(host));
+    const char *host_end = colon ? colon : slash ? slash : host + strlen(host);
+    const char *port = colon ? colon + 1 : "80";
+    size_t port_size = colon ? (size_t)((slash ? slash : host + strlen(host)) - port) : 2;
+    if (host_end == host || (size_t)(host_end - host) >= 256 || !port_size || port_size >= 8 ||
+        (slash && strlen(slash) >= 1024)) {
+        snprintf(error, size, "a URL is http://host[:port]/path");
+        return NULL;
+    }
+    PureWire *w = calloc(1, sizeof(*w));
+    if (!w) { snprintf(error, size, "out of memory"); return NULL; }
+    w->module = m;
+    w->output = (uint32_t)index;
+    memcpy(w->host, host, (size_t)(host_end - host));
+    memcpy(w->port, port, port_size);
+    snprintf(w->path, sizeof(w->path), "%s", slash ? slash : "/");
+    memcpy(w->source, source, length + 1);
+    w->fd = -1;
+    pthread_mutex_init(&w->lock, NULL);
+    pthread_cond_init(&w->changed, NULL);
+    return w;
+}
+
+uint32_t pure_wire_output(const PureWire *w) { return w->output; }
+uint64_t pure_wire_queued(const PureWire *w) { return w->queued; }
+
+uint64_t pure_wire_delivered(PureWire *w) {
+    pthread_mutex_lock(&w->lock);
+    uint64_t delivered = w->delivered;
+    pthread_mutex_unlock(&w->lock);
+    return delivered;
+}
+
+void pure_wire_counts(PureWire *w, uint64_t *delivered, uint64_t *attempts) {
+    pthread_mutex_lock(&w->lock);
+    *delivered = w->deliveries;
+    *attempts = w->attempts;
+    pthread_mutex_unlock(&w->lock);
+}
+
+bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
+    if (w->stuck) return false;
+    Text t = {0};
+    if (!encode(w, payload, &t)) {
+        free(t.bytes);
+        w->stuck = true;
+        fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"no JSON form: its bytes are not text\"}\n", w->source,
+                (unsigned long long)seq);
+        return false;
+    }
+    pthread_mutex_lock(&w->lock);
+    if (w->head + w->count == w->capacity) {
+        if (w->head) {
+            memmove(w->queue, w->queue + w->head, w->count * sizeof(*w->queue));
+            w->head = 0;
+        } else {
+            size_t capacity = w->capacity ? 2 * w->capacity : 16;
+            Pending *grown = realloc(w->queue, capacity * sizeof(*grown));
+            if (!grown) {
+                pthread_mutex_unlock(&w->lock);
+                free(t.bytes);
+                return false; /* it stays in the outbox: queued after the next save */
+            }
+            w->queue = grown;
+            w->capacity = capacity;
+        }
+    }
+    w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq};
+    w->queued = seq;
+    pthread_cond_signal(&w->changed);
+    pthread_mutex_unlock(&w->lock);
+    return true;
+}
+
+/* ---- HTTP/1.1, one request at a time on a kept connection. */
+
+typedef struct { int fd; char bytes[4096]; size_t at, end; } Reader;
+
+static bool fill(Reader *r) {
+    if (r->at == r->end) r->at = r->end = 0;
+    if (r->end == sizeof(r->bytes)) return false;
+    ssize_t n;
+    do n = recv(r->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR);
+    if (n <= 0) return false;
+    r->end += (size_t)n;
+    return true;
+}
+
+/* A line without its CRLF, in line (at most size - 1 bytes). */
+static bool read_line(Reader *r, char *line, size_t size) {
+    for (;;) {
+        char *end = memchr(r->bytes + r->at, '\n', r->end - r->at);
+        if (end) {
+            size_t length = (size_t)(end - (r->bytes + r->at));
+            if (length && end[-1] == '\r') --length;
+            if (length >= size) return false;
+            memcpy(line, r->bytes + r->at, length);
+            line[length] = 0;
+            r->at = (size_t)(end - r->bytes) + 1;
+            return true;
+        }
+        if (r->at) {
+            memmove(r->bytes, r->bytes + r->at, r->end - r->at);
+            r->end -= r->at;
+            r->at = 0;
+        }
+        if (!fill(r)) return false;
+    }
+}
+
+static bool skip(Reader *r, uint64_t size) {
+    while (size) {
+        if (r->at == r->end && !fill(r)) return false;
+        size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size;
+        r->at += n;
+        size -= n;
+    }
+    return true;
+}
+
+static bool send_all(int fd, const char *bytes, size_t size) {
+    while (size) {
+        ssize_t n = send(fd, bytes, size, MSG_NOSIGNAL);
+        if (n < 0 && errno == EINTR) continue;
+        if (n <= 0) return false;
+        bytes += n;
+        size -= (size_t)n;
+    }
+    return true;
+}
+
+static int connect_to(const PureWire *w) {
+    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM}, *list;
+    if (getaddrinfo(w->host, w->port, &hints, &list)) return -1;
+    int fd = -1;
+    for (struct addrinfo *a = list; a && fd < 0; a = a->ai_next) {
+        fd = socket(a->ai_family, a->ai_socktype | SOCK_CLOEXEC, a->ai_protocol);
+        if (fd < 0) continue;
+        struct timeval limit = {TIMEOUT_S, 0};
+        int one = 1;
+        setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &limit, sizeof(limit));
+        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &limit, sizeof(limit));
+        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+        if (connect(fd, a->ai_addr, a->ai_addrlen)) {
+            close(fd);
+            fd = -1;
+        }
+    }
+    freeaddrinfo(list);
+    return fd;
+}
+
+static void drop_connection(PureWire *w) {
+    pthread_mutex_lock(&w->lock);
+    if (w->fd >= 0) close(w->fd);
+    w->fd = -1;
+    pthread_mutex_unlock(&w->lock);
+}
+
+/* A fault a Pure receiver answers when its app refused the message. */
+static bool refusal(const char *error) {
+    static const char *const faults[] = {"divide", "non_finite", "clamp", "limit", "not_found", "encoding", "exists"};
+    for (size_t i = 0; i < sizeof(faults) / sizeof(*faults); ++i) if (!strcmp(error, faults[i])) return true;
+    return false;
+}
+
+/* The answer to one request: its status, whether a Pure receiver refused it,
+ * and whether the connection may carry the next. */
+static bool read_answer(int fd, int *status, bool *refused, bool *keep, unsigned *batch) {
+    Reader r = {.fd = fd};
+    char line[1024];
+    if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false;
+    *status = atoi(line + 9);
+    *keep = line[7] == '1';
+    *refused = false;
+    uint64_t length = 0;
+    bool sized = false, chunked = false;
+    while (read_line(&r, line, sizeof(line))) {
+        if (!*line) {
+            if (chunked) {
+                for (;;) {
+                    uint64_t chunk;
+                    if (!read_line(&r, line, sizeof(line))) return false;
+                    chunk = strtoull(line, NULL, 16);
+                    if (!chunk) break;
+                    if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false;
+                }
+                while (read_line(&r, line, sizeof(line)) && *line) {}
+                return true;
+            }
+            if (*status == 204 || *status == 304 || *status < 200) return true;
+            if (sized) return skip(&r, length);
+            *keep = false; /* the body runs to the close */
+            while (fill(&r)) r.at = r.end;
+            return true;
+        }
+        char *value = strchr(line, ':');
+        if (!value) return false;
+        *value++ = 0;
+        while (*value == ' ' || *value == '\t') ++value;
+        if (!strcasecmp(line, "Content-Length")) { length = strtoull(value, NULL, 10); sized = true; }
+        else if (!strcasecmp(line, "Transfer-Encoding")) chunked = !strcasecmp(value, "chunked");
+        else if (!strcasecmp(line, "Connection")) *keep = strcasecmp(value, "close") != 0 && *keep;
+        else if (!strcasecmp(line, "X-Pure-Error")) *refused = refusal(value);
+        else if (!strcasecmp(line, "X-Pure-Batch")) *batch = (unsigned)strtoul(value, NULL, 10);
+    }
+    return false;
+}
+
+/* One request for count sends from p (numbered in a row): one object, or
+ * an array of them for a receiver that takes batches. */
+static int post(PureWire *w, const Pending *p, unsigned count) {
+    size_t body = count > 1 ? count + 1 : 0;
+    for (unsigned i = 0; i < count; ++i) body += p[i].size;
+    char head[2048], batch[32] = "";
+    if (count > 1) snprintf(batch, sizeof(batch), "X-Pure-Count: %u\r\n", count);
+    int size = snprintf(head, sizeof(head),
+                        "POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n"
+                        "X-Pure-Source: %s\r\nX-Pure-Sequence: %llu\r\n%sIdempotency-Key: %s-%llu\r\n\r\n",
+                        w->path, w->host, w->port, body, w->source, (unsigned long long)p->seq, batch, w->source,
+                        (unsigned long long)p->seq);
+    /* A kept connection the peer closed is tried once more on a new one. */
+    for (int attempt = 0; attempt < 2; ++attempt) {
+        bool fresh = w->fd < 0;
+        if (fresh) {
+            int fd = connect_to(w);
+            if (fd < 0) return AGAIN;
+            pthread_mutex_lock(&w->lock);
+            w->fd = fd;
+            bool stopping = w->stopping;
+            pthread_mutex_unlock(&w->lock);
+            if (stopping) return AGAIN;
+        }
+        int status;
+        bool refused, keep, sent = send_all(w->fd, head, (size_t)size);
+        unsigned batch_size = 0;
+        for (unsigned i = 0; sent && i < count; ++i)
+            sent = (!i || count == 1 || send_all(w->fd, ",", 1)) && (i || count == 1 || send_all(w->fd, "[", 1)) &&
+                   send_all(w->fd, p[i].body, p[i].size);
+        if (sent && count > 1) sent = send_all(w->fd, "]", 1);
+        if (!sent || !read_answer(w->fd, &status, &refused, &keep, &batch_size)) {
+            drop_connection(w);
+            if (fresh) return AGAIN;
+            continue;
+        }
+        if (!keep) drop_connection(w);
+        if (batch_size) w->batch = batch_size;
+        if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED;
+        if (status == 408 || status == 429 || status >= 500) return AGAIN;
+        return REFUSED_BY_PEER;
+    }
+    return AGAIN;
+}
+
+static void *deliver(void *context) {
+    PureWire *w = context;
+    long pause_ms = 0;
+    uint64_t reported = 0;
+    pthread_mutex_lock(&w->lock);
+    for (;;) {
+        while (!w->stopping && !w->count) pthread_cond_wait(&w->changed, &w->lock);
+        if (w->stopping) break;
+        /* As many as the receiver takes at once; their bodies stay put while
+         * the queue only grows behind them. */
+        unsigned count = w->batch > 1 ? (unsigned)(w->count < w->batch ? w->count : w->batch) : 1;
+        Pending next[BATCH_MOST];
+        if (count > BATCH_MOST) count = BATCH_MOST;
+        memcpy(next, w->queue + w->head, count * sizeof(*next));
+        ++w->attempts;
+        pthread_mutex_unlock(&w->lock);
+        int outcome = post(w, next, count);
+        pthread_mutex_lock(&w->lock);
+        if (outcome == DELIVERED) {
+            for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body);
+            w->head += count;
+            w->count -= count;
+            w->delivered = next[count - 1].seq;
+            w->deliveries += count;
+            pause_ms = 0;
+            pthread_mutex_unlock(&w->lock);
+            if (w->notify) w->notify(w->context);
+            pthread_mutex_lock(&w->lock);
+            continue;
+        }
+        if (outcome == REFUSED_BY_PEER && reported != next[0].seq) {
+            fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"the receiver does not take it; trying again\"}\n",
+                    w->source, (unsigned long long)next[0].seq);
+            reported = next[0].seq;
+        }
+        pause_ms = outcome == REFUSED_BY_PEER ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS;
+        if (pause_ms > LONGEST_PAUSE_MS) pause_ms = LONGEST_PAUSE_MS;
+        struct timespec until;
+        clock_gettime(CLOCK_REALTIME, &until);
+        until.tv_sec += pause_ms / 1000;
+        until.tv_nsec += (pause_ms % 1000) * 1000000;
+        if (until.tv_nsec >= 1000000000) { ++until.tv_sec; until.tv_nsec -= 1000000000; }
+        while (!w->stopping && pthread_cond_timedwait(&w->changed, &w->lock, &until) == 0) {}
+    }
+    pthread_mutex_unlock(&w->lock);
+    return NULL;
+}
+
+bool pure_wire_start(PureWire *w, void (*notify)(void *), void *context) {
+    w->notify = notify;
+    w->context = context;
+    w->started = pthread_create(&w->thread, NULL, deliver, w) == 0;
+    return w->started;
+}
+
+void pure_wire_close(PureWire *w) {
+    if (!w) return;
+    pthread_mutex_lock(&w->lock);
+    w->stopping = true;
+    if (w->fd >= 0) shutdown(w->fd, SHUT_RDWR); /* a send or an answer waited for ends now */
+    pthread_cond_broadcast(&w->changed);
+    pthread_mutex_unlock(&w->lock);
+    if (w->started) pthread_join(w->thread, NULL);
+    if (w->fd >= 0) close(w->fd);
+    for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body);
+    free(w->queue);
+    pthread_cond_destroy(&w->changed);
+    pthread_mutex_destroy(&w->lock);
+    free(w);
+}
diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h
new file mode 100644
index 0000000..f720296
--- /dev/null
+++ b/prototypes/connections/deliver.h
@@ -0,0 +1,36 @@
+#ifndef PURE_DELIVER_H
+#define PURE_DELIVER_H
+#include "module.h"
+
+/* Delivering an app's saved sends over HTTP (docs/research/083-delivery-over-http.md).
+ * A wire takes one output's sends in order: each goes as a JSON object of
+ * its fields (groups as objects inside) by POST to a URL, with the headers
+ * X-Pure-Source (the sender's name for the output), X-Pure-Sequence and
+ * Idempotency-Key, one at a time, again and again until it is delivered:
+ * a 2xx answer, or a Pure receiver's refusal (a 4xx with X-Pure-Error
+ * naming a fault: it heard the message and refused it). 408, 429, 5xx and
+ * no answer are tried again after a pause (0.1 s doubling to 10 s); another
+ * 4xx is logged and tried again every 10 s. Nothing is ever dropped: what is
+ * not delivered stays in the app's outbox, and a receiver applies a number
+ * once, so a send delivered twice (its answer lost) is harmless. */
+typedef struct PureWire PureWire;
+
+/* A wire for the output named output of m to url (http://host[:port]/path),
+ * as source; NULL with why in error when it cannot be. */
+PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source,
+                         char *error, size_t size);
+uint32_t pure_wire_output(const PureWire *);
+/* The owner's side: queue a saved send numbered after the last queued one
+ * (it is encoded now); false when it cannot be (a bytes field that is not
+ * UTF-8 has no JSON form), and the wire then stops at it. */
+bool pure_wire_queue(PureWire *, uint64_t seq, PureSpan payload);
+uint64_t pure_wire_queued(const PureWire *);
+/* The last number delivered; any thread. */
+uint64_t pure_wire_delivered(PureWire *);
+/* Deliver on a thread of the wire's own, calling notify after each delivery. */
+bool pure_wire_start(PureWire *, void (*notify)(void *), void *context);
+/* Stop (a delivery under way is finished or abandoned) and free. */
+void pure_wire_close(PureWire *);
+/* Counts for a host's report. */
+void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts);
+#endif
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index ee72494..5cb7f11 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -1676,7 +1676,8 @@ void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) {
         }
 }
 
-PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload) {
+PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
+                            const uint32_t *placed) {
     const PureModule *m = app->module;
     size_t length = source ? strlen(source) : 0;
     app->duplicate = false;
@@ -1700,7 +1701,7 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     app->delivering.source = source;
     app->delivering.seq = seq;
     app->delivering.pending = true;
-    PureResult result = message(app, input, NULL, -1, payload, NULL);
+    PureResult result = message(app, input, NULL, -1, payload, placed);
     if (!app->delivering.pending) return result;
     /* Not applied. Refused by the app (a fault), the message was delivered
      * all the same: its number is heard, saved like a key event's changes. */
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index ed40c15..3441c03 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -316,8 +316,9 @@ void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
  * before is not applied again: it succeeds with app->duplicate set. The
  * number counts as heard with the event's changes, and also when the app
  * refuses the message (a fault: it was delivered, and refused); not when a
- * save fails. */
-PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload);
+ * save fails. placed is as for pure_app_message_fields, or NULL. */
+PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
+                            const uint32_t *placed);
 
 /* Readers on other threads. Right after a save an app can make a snapshot of
  * its saved states, for evaluating views anywhere while it keeps applying
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 0029164..2ecb780 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -29,6 +29,7 @@
  * 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 "module-store.h"
 #include "module-text.h"
 #include "runtime.h"
@@ -57,7 +58,7 @@
 #include <time.h>
 #include <unistd.h>
 
-enum { HEADER_LIMIT = 4096, EVENTS = 256 };
+enum { HEADER_LIMIT = 4096, EVENTS = 256, BATCH = 256 };
 typedef enum { CLOSED, READING_HEAD, READING_BODY, WAITING, WRITING } Phase;
 typedef struct Host Host;
 
@@ -89,6 +90,9 @@ typedef struct Connection {
     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) */
     int64_t expected;
     size_t body_size, received, in_used;
     unsigned char *body;
@@ -142,6 +146,10 @@ typedef struct {
     pthread_mutex_t published_lock;
     PureSnapshot *published;
     uint64_t published_changes, publications;
+    /* 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];
 } Executor;
 
 /* A request made ready: its effect, what it names, and the thread that answers it. */
@@ -159,6 +167,7 @@ typedef struct {
     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;
@@ -269,10 +278,13 @@ static void respond_typed(Connection *c, int status, const char *reason, const c
                           const char *type, const struct iovec *extents, size_t count, size_t size) {
     char tag[48] = "";
     if (revision > 0) snprintf(tag, sizeof(tag), "ETag: \"%" PRId64 "\"\r\n", revision);
+    /* A delivery's answer tells how many numbered messages one request may carry. */
+    char batch[32] = "";
+    if (*c->source) snprintf(batch, sizeof(batch), "X-Pure-Batch: %d\r\n", BATCH);
     int length = snprintf(c->head, sizeof(c->head),
         "HTTP/1.1 %d %s\r\nContent-Length: %zu\r\n%sContent-Type: %s\r\n"
-        "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s\r\n",
-        status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", type, error, tag);
+        "Cache-Control: no-store\r\nX-Pure-Error: %s\r\n%s%s\r\n",
+        status, reason, size, c->keep_alive ? "" : "Connection: close\r\n", type, error, tag, batch);
     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 */
@@ -396,6 +408,10 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
     if ((c->route == ROUTE_OBJECT && (c->post || *query)) || (c->route == ROUTE_VIEW && (c->put || c->post)) ||
         (c->route == ROUTE_INPUT && !c->post)) return false;
     bool length_seen = false, condition_seen = false, host_seen = false;
+    c->source[0] = 0;
+    c->sequence = 0;
+    c->batch = 1;
+    bool counted = false;
     c->body_size = 0;
     c->json = false;
     c->expected = -1;
@@ -437,16 +453,34 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
         } 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-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;
     if (c->put) return host_seen && length_seen && condition_seen;
     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; }
 
 /* A JSON number's text: -?(0|[1-9][0-9]*)(.[0-9]+)?([eE][+-]?[0-9]+)?, its
  * length (0 if there is none). An integer has neither part after it. */
@@ -512,6 +546,8 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     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, or a
          * JSON object of a message's fields (two variable fields need one). */
@@ -552,6 +588,10 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
         uint8_t type = m->field[shape->field_start + f].type;
         if ((type == PURE_FIELD_U32 || type == PURE_FIELD_NUMBER) && !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;
 }
 
@@ -636,6 +676,12 @@ static size_t json_space(const unsigned char *p, size_t at, size_t size) {
 /* 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)
+enum { DELIVERY = 80 };
 typedef struct {
     Connection *c;
     const PureInputShape *shape;
@@ -735,16 +781,31 @@ static const char *members(Body *b, unsigned char *json, size_t size, size_t *at
  * NULL, or the error to answer. */
 static const char *convert(Connection *c) {
     const PureInputShape *shape = &c->host->module->input[c->input];
-    Body b = {.c = c, .shape = shape, .table = c->ingress + 4};
-    b.payload = b.table + 8 * shape->field_count;
+    uint32_t delivered = get_u32(c->ingress) & DELIVERED;
+    size_t table = 8 * (size_t)shape->field_count;
+    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 = json_space(json, 0, size);
-    const char *why = members(&b, json, size, &at, "", 0);
-    if (why) return why;
+    /* A batch is an array of its messages' objects. */
+    if (c->batch > 1) {
+        if (at == size || json[at++] != '[') return "json";
+        at = json_space(json, at, size);
+    }
+    for (unsigned i = 0; i < c->batch; ++i) {
+        if (i) {
+            if (at == size || json[at++] != ',') return "json";
+            at = json_space(json, at, size);
+        }
+        Body b = {.c = c, .shape = shape, .table = tables + i * table, .payload = prefixes + i * c->field_bytes};
+        const char *why = members(&b, json, size, &at, "", 0);
+        if (why) return why;
+        at = json_space(json, at, size);
+        for (uint32_t f = 0; f < shape->field_count; ++f)
+            if (!b.seen[f] && !c->given[f]) return "missing_field";
+    }
+    if (c->batch > 1 && (at == size || json[at++] != ']')) return "json";
     if (json_space(json, at, size) != size) return "json";
-    for (uint32_t f = 0; f < shape->field_count; ++f)
-        if (!b.seen[f] && !c->given[f]) return "missing_field";
-    put_u32(c->ingress, (uint32_t)(c->field_bytes + c->body_size) | PLACED);
+    put_u32(c->ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | PLACED | delivered);
     return NULL;
 }
 
@@ -781,7 +842,10 @@ static void start_request(Connection *c, size_t head_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 input_size = route.kind == ROUTE_INPUT ? 4 + table + c->field_bytes + c->body_size : write ? c->body_size : 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 ? 0 : c->expected, input_size, &c->ticket);
     if (admitted != PURE_OK) {
@@ -797,9 +861,16 @@ static void start_request(Connection *c, size_t head_size) {
     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->field_bytes + c->body_size));
-        if (c->field_bytes) memcpy(ingress + 4 + table, c->fields, c->field_bytes);
-        c->body = ingress + 4 + table + c->field_bytes;
+        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);
@@ -1072,8 +1143,10 @@ static PureResult show_view(PureApp *app, const Route *route, size_t available,
 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);
-    size_t table = header & PLACED ? 8 * (size_t)shape->field_count : 0;
-    PureSpan payload = {effect->value.data + 4 + table, header & ~PLACED};
+    const unsigned char *delivery = effect->value.data + 4;
+    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);
@@ -1083,9 +1156,24 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
     } 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 + 4 + 4 * i);
-        status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) :
-                         pure_app_message(app, route->target, NULL, -1, payload);
+            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 stops it, to be sent again. */
+            for (unsigned i = 1; i < batch && (status == PURE_OK || app->fault != PURE_FAULT_NONE); ++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->fault != PURE_FAULT_NONE) 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};
@@ -1154,6 +1242,35 @@ static void publish(Executor *ex) {
     ++ex->publications;
 }
 
+/* The saved sends of wired outputs not queued yet go to their wires, in order;
+ * a wire that cannot take one takes the rest at a later save. */
+typedef struct { Executor *executor; bool blocked[PURE_MODULE_INPUTS]; } Queueing;
+
+static bool queue_send(void *context, uint32_t output, uint64_t seq, PureSpan payload) {
+    Queueing *q = context;
+    Executor *ex = q->executor;
+    for (unsigned i = 0; i < ex->wire_count; ++i)
+        if (pure_wire_output(ex->wires[i]) == output && !q->blocked[i] && seq > pure_wire_queued(ex->wires[i]))
+            q->blocked[i] = !pure_wire_queue(ex->wires[i], seq, payload);
+    return true;
+}
+
+static void after_save(Executor *ex) {
+    publish(ex);
+    Queueing q = {.executor = ex};
+    if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q);
+}
+
+/* What the wires delivered leaves the outbox (with the next save). */
+static void trim_delivered(Executor *ex) {
+    for (unsigned i = 0; i < ex->wire_count; ++i) {
+        uint64_t delivered = pure_wire_delivered(ex->wires[i]);
+        if (delivered <= ex->trimmed[i]) continue;
+        pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered);
+        ex->trimmed[i] = delivered;
+    }
+}
+
 /* 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;
@@ -1161,7 +1278,7 @@ static void save_burst(Executor *ex, Applied *applied, size_t count) {
         ex->wrote = ex->wrote || (applied[i].status == PURE_OK && applied[i].input == PURE_INPUT_WRITE);
     PureResult saved = pure_app_save(&ex->app);
     ++ex->bursts;
-    if (saved == PURE_OK) publish(ex);
+    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;
 }
@@ -1230,7 +1347,7 @@ static long run_timers(Executor *ex) {
         }
         if (soonest < 0 || ex->next[i] - now < soonest) soonest = ex->next[i] - now;
     }
-    if (ticked && pure_app_save(&ex->app) == PURE_OK) publish(ex); /* a failed save undoes the ticks */
+    if (ticked && pure_app_save(&ex->app) == PURE_OK) after_save(ex); /* a failed save undoes the ticks */
     return soonest < 0 ? -1 : (long)ceil(soonest);
 }
 
@@ -1376,12 +1493,17 @@ static void *run_owner(void *context) {
     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. */
+        /* 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);
+            if (delivered) trim_delivered(&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;
@@ -1654,6 +1776,44 @@ static void stop_shard(Host *host) {
     free(host->views);
 }
 
+/* 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 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);
+        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;
+}
+
 /* --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) {
@@ -1667,7 +1827,8 @@ static unsigned auto_threads(bool owner) {
 int main(int argc, char **argv) {
     static PureModule module;
     static Owner owner;
-    const char *module_path = NULL, *store = NULL;
+    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS];
+    unsigned send_count = 0;
     unsigned port = 8767, threads = 1;
     size_t budget = 64u << 20, capacity = 1024;
     uint64_t timeout_ms = 10000;
@@ -1676,6 +1837,8 @@ int main(int argc, char **argv) {
         if (i + 1 == argc) goto usage;
         if (!strcmp(argv[i], "--module")) module_path = 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], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
         else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
         else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
@@ -1697,16 +1860,30 @@ int main(int argc, char **argv) {
     Host *hosts = calloc(threads, sizeof(*hosts));
     pthread_t *thread = calloc(threads, sizeof(*thread)), owner_thread;
     if (!hosts || !thread) return 1;
-    /* States have one owner on a thread of its own, whose snapshots the
-     * connection threads read. */
-    bool owned = module.states;
+    /* 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) ||
             pthread_cond_init(&owner.waiting, NULL) || pthread_mutex_init(&owner.executor.published_lock, NULL))
             return 1;
         owner.executor.reads_first = true;
         owner.executor.publishes = true;
-        publish(&owner.executor);
+        if (!open_wires(&owner.executor, &module, sends, send_count, name)) 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; }
         owner.hosts = hosts;
         owner.host_count = threads;
     }
@@ -1746,6 +1923,14 @@ int main(int argc, char **argv) {
         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 */
+    }
     if (owned) {
         stop_executor(&owner.executor);
         free(owner.jobs);
@@ -1755,14 +1940,14 @@ int main(int argc, char **argv) {
     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
-            ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
+            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
             active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
-            owner.executor.publications, shown, groups, grouped);
+            owner.executor.publications, shown, delivered, attempts, groups, grouped);
     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]\n", stderr);
+          "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http://host:port/path ...] [--name NAME]\n", stderr);
     return 2;
 }
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index bb59174..ccb3371 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -2581,7 +2581,7 @@ static PureResult deliver(PureApp *app, const char *source, uint64_t seq, const
     memcpy(payload, &count, 4);
     memcpy(payload + 4, &total, 8);
     memcpy(payload + 12, item, length);
-    PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length});
+    PureResult result = pure_app_deliver(app, 0, source, seq, (PureSpan){payload, 12 + length}, NULL);
     pure_app_finish(app);
     return result;
 }
@@ -2618,10 +2618,10 @@ static void links(void) {
     Outbox o = outbox_of(&a);
     CHECK(o.count == 1 && o.items[0].output == 2 && o.items[0].seq == 1 && o.items[0].size == 15 && a.values[0] == 1);
     PureSpan sent = {o.items[0].payload, o.items[0].size};
-    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && !b.duplicate && b.values[0] == 2 &&
+    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == PURE_OK && !b.duplicate && b.values[0] == 2 &&
           !strcmp(state_text(&b, 1), "pen"));
     pure_app_finish(&b);
-    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent) == PURE_OK && b.duplicate && b.values[0] == 2);
+    CHECK(pure_app_deliver(&b, 0, "orders.ordered", 1, sent, NULL) == PURE_OK && b.duplicate && b.values[0] == 2);
     CHECK(deliver(&b, "other", 1, "cup", 5) == PURE_OK && !b.duplicate && b.values[0] == 7);
     CHECK(deliver(&b, "orders.ordered", 3, "hat", 1) == PURE_OK && b.values[0] == 8); /* numbers may skip */
     CHECK(deliver(&b, "orders.ordered", 2, "old", 1) == PURE_OK && b.duplicate && b.values[0] == 8);
@@ -2630,7 +2630,7 @@ static void links(void) {
     CHECK(deliver(&b, "other", 2, "big", 1) == PURE_OK && b.duplicate && b.values[0] == 8);
     /* Bad deliveries change nothing. */
     CHECK(deliver(&b, "", 4, "x", 1) == PURE_INVALID && deliver(&b, "a b", 4, "x", 1) == PURE_INVALID);
-    CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent) == PURE_INVALID);
+    CHECK(deliver(&b, "other", 0, "x", 1) == PURE_INVALID && pure_app_deliver(&a, 2, "s", 1, sent, NULL) == PURE_INVALID);
     CHECK(pure_app_message(&a, 2, NULL, -1, sent) == PURE_INVALID);
     /* Trimmed, a send is gone at once for the host and from the store at the next save. */
     pure_app_trim(&a, 2, 1);