Keep a link's two sides as state alone, the sockets and threads apart

link/link-protocol.h is the protocol with no socket, thread or clock of
its own: packets and time in, packets out. link/link.h carries it over
UDP as before. A sender starts small (8 flights, 16 records, a 4 KiB
ring) so one thread can hold many. The transport's CPU a record is the
same within 2% (five rounds each way, 61 B to 16 KiB).

test-link runs both sides through a network of its own: loss, packets
held back and doubled, failed saves, an idle sweep, and both sides
starting again, every record applied once as sent.

8 files changed, 2054 insertions, 1555 deletions

diff --git a/Makefile b/Makefile
index 10a5f0e..e3e26c5 100644
--- a/Makefile
+++ b/Makefile
@@ -20,12 +20,12 @@ BUILD ?= build
 # surfaces link it); libpure-link.a is the links with the crypto loader.
 # prototypes/ keeps the earlier document runtime (000-077).
 # Unit tests sit by their part (core/test-module.c, store/test-tree-store.c,
-# store/test-log.c, serve/test-message-json.c);
+# store/test-log.c, link/test-link.c, serve/test-message-json.c);
 # tests/check.py checks the hosts end to end.
 INCLUDES = -Icore
 MODEL = core/module.c core/module-list.c core/module-save.c core/module-text.c core/module-store.c core/flow.c
 STORE = store/store-tree.c store/store-log.c
-LINK = link/link.c link/crypto.c
+LINK = link/link-protocol.c link/link.c link/crypto.c
 SERVE = serve/serve.c serve/deliver.c serve/message-json.c
 HEADERS = $(wildcard core/*.h) $(wildcard link/*.h) $(wildcard serve/*.h) $(wildcard store/*.h)
 OBJECTS = $(patsubst %.c,$(BUILD)/obj/%.o,$(MODEL) $(STORE))
@@ -75,6 +75,11 @@ $(BUILD)/test-module: core/test-module.c $(MODEL) $(STORE) $(HEADERS)
 	@mkdir -p $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -DPURE_STORE_TEST core/test-module.c $(MODEL) $(STORE) $(LDFLAGS) -pthread -lm $(LDLIBS) -o $@
 
+# A link's two sides through a network of the test's own (pure-crypto.so beside it).
+$(BUILD)/test-link: link/test-link.c $(BUILD)/libpure-link.a $(BUILD)/pure-crypto.so $(wildcard link/*.h)
+	@mkdir -p $(BUILD)
+	$(CC) $(CPPFLAGS) $(CFLAGS) link/test-link.c $(BUILD)/libpure-link.a $(LDFLAGS) -pthread -ldl $(LDLIBS) -o $@
+
 $(BUILD)/test-message-json: serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a $(HEADERS)
 	@mkdir -p $(BUILD)
 	$(CC) $(CPPFLAGS) $(CFLAGS) $(INCLUDES) -Iserve serve/test-message-json.c serve/message-json.c $(BUILD)/libpure.a \
@@ -91,10 +96,11 @@ $(BUILD)/test-tree-store: store/test-tree-store.c $(STORE) $(HEADERS)
 # The HTTP host's checks build their apps with aup (the sibling checkout) and
 # compare documents with the earlier document server.
 AUP ?= ../aup
-check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-message-json
+check: all $(BUILD)/test-module $(BUILD)/test-tree-store $(BUILD)/test-log $(BUILD)/test-link $(BUILD)/test-message-json
 	$(BUILD)/test-tree-store
 	$(BUILD)/test-log
 	$(BUILD)/test-module
+	$(BUILD)/test-link
 	$(BUILD)/test-message-json
 	$(MAKE) -C $(AUP) build/aup
 	$(MAKE) -C prototypes/connections BUILD=build/tree STORE=store-tree.c build/tree/pure-connections
diff --git a/link/link-protocol.c b/link/link-protocol.c
new file mode 100644
index 0000000..94a965b
--- /dev/null
+++ b/link/link-protocol.c
@@ -0,0 +1,1494 @@
+#define _GNU_SOURCE
+/* Links without connections as state alone (link/link-protocol.h,
+ * docs/design/011-pure-transport.md). */
+#include "link-protocol.h"
+#include <fcntl.h>
+#include <malloc.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/mman.h>
+#include <sys/random.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+enum { VERSION = 0x70, HEADER = 21, DATAGRAM = PURE_LINK_DATAGRAM, BODY = DATAGRAM - HEADER - PURE_AEAD_TAG, PIECE = 1000 };
+enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, STORED_GAP = 1 };
+enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */
+enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 };
+enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM };
+#define MS UINT64_C(1000000)
+/* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of
+ * the Internet at large); a probe at least every second while a receiver is
+ * away; a record received but not stored after a second is sent again. */
+static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS,
+                      STORED_WAIT = 1000 * MS, RECORD_MOST = UINT64_C(64) << 20, HELD_MOST = UINT64_C(16) << 20;
+
+static void put16(unsigned char *p, uint32_t v) { p[0] = (unsigned char)v, p[1] = (unsigned char)(v >> 8); }
+static void put32(unsigned char *p, uint32_t v) { for (int i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> 8 * i); }
+static void put64(unsigned char *p, uint64_t v) { for (int i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> 8 * i); }
+static uint32_t get16(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8; }
+static uint32_t get32(const unsigned char *p) {
+    return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24;
+}
+static uint64_t get64(const unsigned char *p) { return (uint64_t)get32(p) | (uint64_t)get32(p + 4) << 32; }
+/* LEB128: seven bits a byte, low first. */
+static size_t put_varint(unsigned char *p, uint64_t v) {
+    size_t n = 0;
+    for (; v >= 0x80; v >>= 7) p[n++] = (unsigned char)(v | 0x80);
+    p[n++] = (unsigned char)v;
+    return n;
+}
+static size_t varint_size(uint64_t v) {
+    size_t n = 1;
+    for (; v >= 0x80; v >>= 7) ++n;
+    return n;
+}
+/* A varint from p (size bytes): how many bytes it took, or 0. */
+static size_t get_varint(const unsigned char *p, size_t size, uint64_t *v) {
+    *v = 0;
+    for (size_t n = 0; n < size && n < 10; ++n) {
+        *v |= (uint64_t)(p[n] & 0x7f) << (7 * n);
+        if (!(p[n] & 0x80)) return n + 1;
+    }
+    return 0;
+}
+
+static uint64_t least(uint64_t a, uint64_t b) { return a < b ? a : b; }
+static uint64_t most(uint64_t a, uint64_t b) { return a > b ? a : b; }
+
+static uint32_t fnv(uint32_t h, const void *bytes, size_t size) {
+    const unsigned char *p = bytes;
+    for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u;
+    return h;
+}
+
+static uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); }
+
+static uint64_t random64(void) {
+    uint64_t value = 0;
+    while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {}
+    return value;
+}
+
+/* The key and nonce base of packets from `from` (in its run from_run) to
+ * `to`, answering to_run (0 for none). */
+static bool derive(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
+                   const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
+                   unsigned char key[PURE_AEAD_KEY], unsigned char base[PURE_AEAD_NONCE]) {
+    unsigned char info[16 + 2 * PURE_SEAL_KEY + 16], out[PURE_AEAD_KEY + PURE_AEAD_NONCE];
+    memcpy(info, "pure link 0 keys", 16);
+    memcpy(info + 16, from, PURE_SEAL_KEY);
+    memcpy(info + 16 + PURE_SEAL_KEY, to, PURE_SEAL_KEY);
+    put64(info + 16 + 2 * PURE_SEAL_KEY, from_run);
+    put64(info + 24 + 2 * PURE_SEAL_KEY, to_run);
+    if (!pure_link_derive(secret, info, sizeof(info), out, sizeof(out))) return false;
+    memcpy(key, out, PURE_AEAD_KEY);
+    memcpy(base, out + PURE_AEAD_KEY, PURE_AEAD_NONCE);
+    explicit_bzero(out, sizeof(out));
+    return true;
+}
+
+/* The same as a context of its own, for a sender's one link. */
+static PureAead *aead_of(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
+                         const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
+                         unsigned char base[PURE_AEAD_NONCE]) {
+    unsigned char key[PURE_AEAD_KEY];
+    PureAead *aead = derive(secret, from, to, from_run, to_run, key, base) ? pure_aead(key) : NULL;
+    explicit_bzero(key, sizeof(key));
+    return aead;
+}
+
+static void nonce_of(unsigned char out[PURE_AEAD_NONCE], const unsigned char base[PURE_AEAD_NONCE], uint64_t pn) {
+    memcpy(out, base, PURE_AEAD_NONCE);
+    for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i));
+}
+
+/* A packet: the header, then body sealed (with key, or the context's own
+ * when NULL); its size, or 0. */
+static size_t seal_packet(PureAead *aead, const unsigned char *key, const unsigned char base[PURE_AEAD_NONCE], uint32_t id,
+                          uint64_t run, uint64_t pn, const unsigned char *body, size_t size, unsigned char *out) {
+    unsigned char nonce[PURE_AEAD_NONCE];
+    out[0] = VERSION;
+    put32(out + 1, id);
+    put64(out + 5, run);
+    put64(out + 13, pn);
+    nonce_of(nonce, base, pn);
+    return pure_aead_seal(aead, key, nonce, out, HEADER, body, size, out + HEADER) ? HEADER + size + PURE_AEAD_TAG : 0;
+}
+
+static size_t put_names(unsigned char *p, const char *source, const char *input) {
+    size_t a = strlen(source), b = strlen(input);
+    p[0] = (unsigned char)a;
+    memcpy(p + 1, source, a);
+    p[1 + a] = (unsigned char)b;
+    memcpy(p + 2 + a, input, b);
+    return 2 + a + b;
+}
+
+/* A source and an input's names from p (size bytes); how many it takes, or 0. */
+static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1],
+                        char input[PURE_LINK_NAME + 1]) {
+    if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0;
+    size_t a = p[0], b = p[1 + a];
+    if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0;
+    memcpy(source, p + 1, a);
+    source[a] = 0;
+    memcpy(input, p + 2 + a, b);
+    input[b] = 0;
+    return 2 + a + b;
+}
+
+/* ---- The sending side */
+
+enum { UNSENT, FLYING, RECEIVED };
+typedef struct { uint64_t pn; uint8_t state; } Piece;
+/* A record queued: its bytes at `at` in the link's byte ring, its pieces'
+ * states (one kept here, more beside). */
+typedef struct { uint64_t seq; size_t at; uint32_t size, pieces; Piece *more, one; } Record;
+static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one : &r->more[i]; }
+typedef struct { uint64_t seq; uint32_t piece; } Unit;
+typedef struct {
+    uint64_t pn, sent; /* its number; when it went (ns) */
+    uint32_t bytes;
+    bool eliciting, done; /* acknowledged or lost */
+    uint16_t count;
+    Unit units[UNITS];
+} Flight;
+
+struct PureLinkSender {
+    uint32_t shape, own_id, peer_id;
+    char input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1];
+    const PureSealKey *own;
+    unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32];
+    Record *records; /* the sends not yet stored, oldest first (a ring of consecutive numbers) */
+    size_t head, count, capacity;
+    /* Their bytes, one after another as queued (a record's never cut by the
+     * ring's end): taken back in order as they are stored. */
+    unsigned char *ring;
+    size_t ring_capacity;
+    uint64_t queued, stored; /* the last queued; saved by the receiver */
+    uint64_t run, next_pn, in_run;
+    PureAead *out, *in; /* made once the receiver's key is known */
+    unsigned char out_base[PURE_AEAD_NONCE], in_base[PURE_AEAD_NONCE];
+    Flight *flights; /* one for each number sent, oldest first (a ring) */
+    size_t flight_head, flight_count, flight_capacity;
+    uint64_t in_flight, window, threshold, recovery_start, eliciting;
+    uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt;
+    bool acked_any, sampled;
+    uint64_t loss_time, last_eliciting, credit, stored_changed;
+    uint64_t received;    /* pieces received and not yet stored */
+    uint64_t unsent_from; /* no record before this number has a piece unsent */
+    unsigned pto_count, probes;
+    PureLinkCounts counts;
+};
+
+static Record *record_at(PureLinkSender *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; }
+
+/* The queued record numbered seq, or NULL (stored already, or not queued). */
+static Record *record_of(PureLinkSender *l, uint64_t seq) {
+    if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL;
+    return record_at(l, (size_t)(seq - record_at(l, 0)->seq));
+}
+
+/* Every change of a piece's state goes through here: the link knows at once
+ * how many pieces wait to be stored, and where the unsent ones begin. */
+static void set_piece(PureLinkSender *l, Record *r, uint32_t i, uint8_t state) {
+    uint8_t was = (*piece_of(r, i)).state;
+    if (was == state) return;
+    if (was == RECEIVED) --l->received;
+    if (state == RECEIVED) ++l->received;
+    if (state == UNSENT && r->seq < l->unsent_from) l->unsent_from = r->seq;
+    (*piece_of(r, i)).state = state;
+}
+
+/* Where size bytes go at the end of the byte ring (after the newest record,
+ * or from its start when that is free and the end is not), or SIZE_MAX. */
+static size_t ring_place(PureLinkSender *l, size_t size) {
+    if (!l->count) return size <= l->ring_capacity ? 0 : SIZE_MAX;
+    Record *oldest = record_at(l, 0), *newest = record_at(l, l->count - 1);
+    size_t head = oldest->at, tail = newest->at + newest->size;
+    if (newest->at >= head) /* the bytes are [head, tail): after them, or before them from the start */
+        return l->ring_capacity - tail >= size ? tail : size <= head ? 0 : SIZE_MAX;
+    return head - tail >= size ? tail : SIZE_MAX; /* around the end: between the newest and the oldest */
+}
+
+/* A ring room for size bytes more: twice as large (or more), the records
+ * moved into it one after another. */
+static bool ring_grow(PureLinkSender *l, size_t size) {
+    size_t used = 0;
+    for (size_t k = 0; k < l->count; ++k) used += record_at(l, k)->size;
+    size_t capacity = l->ring_capacity ? 2 * l->ring_capacity : 4 << 10;
+    while (capacity < used + size + 1) capacity *= 2;
+    unsigned char *ring = malloc(capacity);
+    if (!ring) return false;
+    for (size_t k = 0, at = 0; k < l->count; ++k) {
+        Record *r = record_at(l, k);
+        if (r->size) memcpy(ring + at, l->ring + r->at, r->size);
+        r->at = at;
+        at += r->size;
+    }
+    free(l->ring);
+    l->ring = ring;
+    l->ring_capacity = capacity;
+    return true;
+}
+
+bool pure_link_sender_queue(PureLinkSender *l, uint64_t seq, const unsigned char *bytes, size_t size) {
+    Record r = {.seq = seq, .size = (uint32_t)size};
+    r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1;
+    if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false;
+    bool ok = seq > l->queued && (!l->count || seq == l->queued + 1);
+    size_t at = ok ? ring_place(l, size) : SIZE_MAX;
+    if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX;
+    if (ok) {
+        r.at = at;
+        if (size) memcpy(l->ring + at, bytes, size);
+    }
+    if (ok && l->count == l->capacity) {
+        size_t capacity = l->capacity ? 2 * l->capacity : 16;
+        Record *grown = malloc(capacity * sizeof(*grown));
+        if (grown) {
+            for (size_t k = 0; k < l->count; ++k) grown[k] = *record_at(l, k);
+            free(l->records);
+            l->records = grown;
+            l->capacity = capacity;
+            l->head = 0;
+        } else
+            ok = false;
+    }
+    if (!ok) {
+        free(r.more);
+        return false;
+    }
+    *record_at(l, l->count++) = r;
+    l->queued = seq;
+    return true;
+}
+
+static Flight *flight_of(PureLinkSender *l, uint64_t pn) {
+    if (!l->flight_count) return NULL;
+    uint64_t first = l->flights[l->flight_head].pn;
+    if (pn < first || pn - first >= l->flight_count) return NULL;
+    return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity];
+}
+
+/* Room for one more flight: made before a packet is, so every number sent
+ * has its flight (they are found by number). */
+static bool flight_room(PureLinkSender *l) {
+    if (l->flight_count < l->flight_capacity) return true;
+    size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 8;
+    Flight *grown = malloc(capacity * sizeof(*grown));
+    if (!grown) return false;
+    for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity];
+    free(l->flights);
+    l->flights = grown;
+    l->flight_capacity = capacity;
+    l->flight_head = 0;
+    return true;
+}
+
+static void push_flight(PureLinkSender *l, const Flight *f) {
+    l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f;
+}
+
+static void settle(PureLinkSender *l, Flight *f) {
+    f->done = true;
+    if (f->eliciting) {
+        --l->eliciting;
+        l->in_flight -= f->bytes;
+    }
+}
+
+/* A packet lost: what it carried and was not received otherwise goes again. */
+static void lose(PureLinkSender *l, Flight *f, uint64_t now) {
+    settle(l, f);
+    ++l->counts.lost;
+    for (uint16_t u = 0; u < f->count; ++u) {
+        Record *r = record_of(l, f->units[u].seq);
+        Piece *p = r ? &(*piece_of(r, f->units[u].piece)) : NULL;
+        if (p && p->state == FLYING && p->pn == f->pn) {
+            set_piece(l, r, f->units[u].piece, UNSENT);
+            ++l->counts.resent;
+        }
+    }
+    if (f->sent > l->recovery_start) { /* one congestion event per round trip */
+        l->recovery_start = now;
+        l->window = most(l->window / 2, WINDOW_LEAST);
+        l->threshold = l->window;
+    }
+}
+
+static void detect_lost(PureLinkSender *l, uint64_t now) {
+    l->loss_time = 0;
+    if (!l->acked_any) return;
+    uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT);
+    uint64_t delay = most(rtt * 9 / 8, GRANULARITY);
+    for (size_t k = 0; k < l->flight_count; ++k) {
+        Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity];
+        if (f->pn >= l->largest_acked) break;
+        if (f->done) continue;
+        if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now);
+        else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay;
+    }
+}
+
+static void sample_rtt(PureLinkSender *l, uint64_t latest, uint64_t ack_delay) {
+    l->latest_rtt = latest;
+    if (!l->sampled) {
+        l->min_rtt = l->smoothed_rtt = latest;
+        l->rtt_var = latest / 2;
+        l->sampled = true;
+        return;
+    }
+    l->min_rtt = least(l->min_rtt, latest);
+    ack_delay = least(ack_delay, ACK_DELAY_MOST);
+    uint64_t adjusted = latest >= l->min_rtt + ack_delay ? latest - ack_delay : latest;
+    uint64_t difference = l->smoothed_rtt > adjusted ? l->smoothed_rtt - adjusted : adjusted - l->smoothed_rtt;
+    l->rtt_var = (3 * l->rtt_var + difference) / 4;
+    l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8;
+}
+
+static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint64_t now) {
+    if (size < 13 || size < 13 + 16u * p[12]) return;
+    uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000;
+    Flight *top = NULL;
+    for (unsigned k = 0; k < p[12]; ++k) {
+        uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k);
+        if (first > last || !l->flight_count) continue;
+        uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1;
+        for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) {
+            Flight *f = flight_of(l, pn);
+            if (!f || f->done) continue;
+            settle(l, f);
+            ++l->counts.acknowledged;
+            for (uint16_t u = 0; u < f->count; ++u) {
+                Record *r = record_of(l, f->units[u].seq);
+                if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) {
+                    set_piece(l, r, f->units[u].piece, RECEIVED);
+                    l->stored_changed = now; /* waits for its save from now */
+                }
+            }
+            if (f->eliciting && f->sent > l->recovery_start)
+                l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window;
+            if (pn == largest) top = f;
+        }
+    }
+    if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest;
+    l->acked_any = true;
+    if (top && top->eliciting) sample_rtt(l, now - top->sent, delay);
+    l->pto_count = 0;
+    detect_lost(l, now);
+    while (l->flight_count && l->flights[l->flight_head].done) {
+        l->flight_head = (l->flight_head + 1) % l->flight_capacity;
+        --l->flight_count;
+    }
+}
+
+static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, uint64_t now) {
+    char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
+    size_t names = get_names(p, size, source, input);
+    if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return;
+    uint64_t heard = get64(p + names);
+    unsigned flags = p[names + 8];
+    uint32_t credit = get32(p + names + 9);
+    if (heard > l->stored) {
+        while (l->count && record_at(l, 0)->seq <= heard) {
+            Record *r = record_at(l, 0);
+            for (uint32_t i = 0; i < r->pieces; ++i) l->received -= (*piece_of(r, i)).state == RECEIVED;
+            free(r->more); /* its bytes go back with the ring's head */
+            l->head = (l->head + 1) % l->capacity;
+            --l->count;
+        }
+        l->stored = heard;
+        l->stored_changed = now;
+    }
+    if (flags & STORED_GAP) /* it did not keep what came after a gap: send it again */
+        for (size_t k = 0; k < l->count && l->received; ++k) {
+            Record *r = record_at(l, k);
+            for (uint32_t i = 0; i < r->pieces; ++i)
+                if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT);
+        }
+    l->credit = credit ? least(credit, 65536) : 1;
+}
+
+void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_t n, uint64_t now) {
+    unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
+    if (!l->out || n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || get32(packet + 1) != l->peer_id)
+        return; /* not ours, or an error from an earlier send (nobody listening) */
+    uint64_t run = get64(packet + 5), pn = get64(packet + 13);
+    if (!l->in || l->in_run != run) {
+        pure_aead_free(l->in);
+        l->in = aead_of(l->secret, l->peer, l->own_public, run, l->run, l->in_base);
+        l->in_run = run;
+        if (!l->in) return;
+    }
+    nonce_of(nonce, l->in_base, pn);
+    size_t size = n - HEADER - PURE_AEAD_TAG;
+    if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) return;
+    ++l->counts.received;
+    for (size_t at = 0; at < size;) {
+        unsigned type = plain[at++];
+        if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) {
+            on_ack(l, plain + at, size - at, now);
+            at += 13 + 16u * plain[at + 12];
+        } else if (type == F_STORED) {
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
+            size_t names = get_names(plain + at, size - at, source, input);
+            if (!names || at + names + 13 > size) break;
+            on_stored(l, plain + at, size - at, now);
+            at += names + 13;
+        } else if (type != F_PING)
+            break; /* padding, or what this side does not know: the rest */
+    }
+}
+
+/* A packet of pieces not sent yet, in order, within the receiver's credit:
+ * its body's size, or 0 when there is none. */
+static size_t fill(PureLinkSender *l, unsigned char *body, Flight *f, bool pad) {
+    size_t at = 0;
+    body[at++] = F_STREAM;
+    at += put_names(body + at, l->source, l->input);
+    put32(body + at, l->shape);
+    at += 4;
+    /* The credit counts from the oldest record not stored: after a restart
+     * the sender does not know yet how far the receiver stored. */
+    uint64_t base = l->count ? record_at(l, 0)->seq - 1 : l->stored;
+    /* From the first record that may have an unsent piece: the ones before
+     * it are sent, and every one found sent moves that mark on. */
+    size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0;
+    bool before = true;
+    /* A record none of whose pieces went yet goes whole when it fits what is
+     * left of the packet: whole records whose numbers follow on share a
+     * RECORDS frame (its count at run). Else its unsent pieces go, each in a
+     * RECORD frame. */
+    size_t run = 0;
+    uint64_t next = 0;
+    for (; k < l->count && f->count < UNITS; ++k) {
+        Record *r = record_at(l, k);
+        if (r->seq > base + l->credit) break;
+        bool unsent = false, fresh = true, joins = run && r->seq == next;
+        for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT;
+        size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size;
+        if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) {
+            if (!joins) {
+                body[at] = F_RECORDS;
+                put64(body + at + 1, r->seq);
+                body[at + 9] = 0;
+                run = at + 9;
+                at += RECORDS_HEAD;
+            }
+            at += put_varint(body + at, r->size);
+            if (r->size) memcpy(body + at, l->ring + r->at, r->size);
+            at += r->size;
+            ++body[run];
+            next = r->seq + 1;
+            for (uint32_t i = 0; i < r->pieces; ++i) {
+                set_piece(l, r, i, FLYING);
+                piece_of(r, i)->pn = f->pn;
+                f->units[f->count++] = (Unit){r->seq, i};
+            }
+            ++l->counts.records;
+        } else if (fresh && r->pieces == 1)
+            unsent = true; /* too large for what is left: the next packet */
+        else
+            for (uint32_t i = 0; i < r->pieces; ++i) {
+                if (piece_of(r, i)->state != UNSENT) continue;
+                uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset);
+                if (f->count == UNITS || at + RECORD_HEAD + length > BODY) {
+                    unsent = true;
+                    break;
+                }
+                body[at] = F_RECORD;
+                put64(body + at + 1, r->seq);
+                put32(body + at + 9, r->size);
+                put32(body + at + 13, offset);
+                put16(body + at + 17, length);
+                if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length);
+                at += RECORD_HEAD + length;
+                run = 0;
+                set_piece(l, r, i, FLYING);
+                piece_of(r, i)->pn = f->pn;
+                f->units[f->count++] = (Unit){r->seq, i};
+                ++l->counts.records;
+            }
+        if (before && !unsent) l->unsent_from = r->seq + 1;
+        before = before && !unsent;
+        if (unsent) { /* the packet is full: whole, so it can go in a burst */
+            if (pad && f->count) {
+                memset(body + at, F_PADDING, BODY - at);
+                at = BODY;
+            }
+            break;
+        }
+    }
+    return f->count ? at : 0;
+}
+
+/* Its number is spent all the same. */
+void pure_link_sender_refused(PureLinkSender *l, uint64_t pn) {
+    Flight *f = flight_of(l, pn);
+    if (!f || f->done) return;
+    for (uint16_t u = 0; u < f->count; ++u) {
+        Record *r = record_of(l, f->units[u].seq);
+        if (r && (*piece_of(r, f->units[u].piece)).pn == pn) set_piece(l, r, f->units[u].piece, UNSENT);
+    }
+    settle(l, f);
+    f->eliciting = false;
+    f->count = 0;
+    --l->counts.packets;
+}
+
+/* Seal f with body into out, and count it sent: its size, or 0 (what it
+ * carried is then unsent again). Its number is spent either way. */
+static size_t transmit(PureLinkSender *l, Flight *f, const unsigned char *body, size_t size, unsigned char *out,
+                       uint64_t now) {
+    size_t total = seal_packet(l->out, NULL, l->out_base, l->own_id, l->run, f->pn, body, size, out);
+    ++l->next_pn;
+    f->sent = now;
+    f->bytes = (uint32_t)total;
+    if (!total) { /* not sealed: what it carried is unsent again */
+        for (uint16_t u = 0; u < f->count; ++u) {
+            Record *r = record_of(l, f->units[u].seq);
+            if (r && (*piece_of(r, f->units[u].piece)).pn == f->pn) set_piece(l, r, f->units[u].piece, UNSENT);
+        }
+        f->done = true;
+        f->eliciting = false;
+        f->count = 0;
+        push_flight(l, f);
+        return 0;
+    }
+    push_flight(l, f);
+    ++l->counts.packets;
+    if (f->eliciting) {
+        ++l->eliciting;
+        l->in_flight += f->bytes;
+        l->last_eliciting = now;
+    }
+    return total;
+}
+
+size_t pure_link_sender_packet(PureLinkSender *l, unsigned char out[PURE_LINK_DATAGRAM], bool whole, uint64_t now,
+                               uint64_t *number) {
+    bool probe = l->probes > 0;
+    if (!l->out || (!probe && l->in_flight + DATAGRAM > l->window) || !flight_room(l)) return 0;
+    unsigned char body[BODY];
+    Flight f = {.pn = l->next_pn, .eliciting = true};
+    size_t size = fill(l, body, &f, whole);
+    if (!size) {
+        if (!probe) return 0;
+        body[0] = F_PING; /* nothing to send again: a probe all the same */
+        size = 1;
+    }
+    size_t total = transmit(l, &f, body, size, out, now);
+    if (!total) return 0;
+    if (probe) --l->probes;
+    *number = f.pn;
+    return total;
+}
+
+static uint64_t pto_period(const PureLinkSender *l) {
+    uint64_t smoothed = l->sampled ? l->smoothed_rtt : FIRST_RTT, var = l->sampled ? l->rtt_var : FIRST_RTT / 2;
+    uint64_t period = smoothed + most(4 * var, GRANULARITY) + ACK_DELAY_MOST;
+    for (unsigned i = 0; i < l->pto_count && period < PTO_MOST; ++i) period *= 2;
+    return least(period, PTO_MOST);
+}
+
+static bool waiting_for_save(const PureLinkSender *l) { return l->received > 0; }
+
+uint64_t pure_link_sender_due(const PureLinkSender *l) {
+    uint64_t due = l->loss_time;
+    if (l->eliciting) {
+        uint64_t pto = l->last_eliciting + pto_period(l);
+        if (!due || pto < due) due = pto;
+    }
+    if (waiting_for_save(l)) {
+        uint64_t stored = l->stored_changed + STORED_WAIT;
+        if (!due || stored < due) due = stored;
+    }
+    return due;
+}
+
+void pure_link_sender_tick(PureLinkSender *l, uint64_t now) {
+    if (l->loss_time && l->loss_time <= now) detect_lost(l, now);
+    else if (l->eliciting && l->last_eliciting + pto_period(l) <= now) {
+        /* Nothing heard for a while: a probe, what flies longest sent again. */
+        ++l->pto_count;
+        l->probes = 1;
+        for (size_t k = 0; k < l->count; ++k) {
+            Record *r = record_at(l, k);
+            uint32_t i = 0;
+            while (i < r->pieces && (*piece_of(r, i)).state != FLYING) ++i;
+            if (i < r->pieces) {
+                set_piece(l, r, i, UNSENT);
+                ++l->counts.resent;
+                break;
+            }
+        }
+    }
+    if (waiting_for_save(l) && l->stored_changed + STORED_WAIT <= now) {
+        /* Received long ago and still not stored: the receiver lost them (it
+         * restarted) or its word got lost. Sent again: it takes a number once. */
+        for (size_t k = 0; k < l->count && l->received; ++k) {
+            Record *r = record_at(l, k);
+            for (uint32_t i = 0; i < r->pieces; ++i)
+                if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT), ++l->counts.resent;
+        }
+        l->stored_changed = now;
+    }
+}
+
+PureLinkSender *pure_link_sender(const char *source, const char *input, uint32_t shape, const PureSealKey *own,
+                                 const unsigned char own_public[PURE_SEAL_KEY]) {
+    size_t a = strlen(source), b = strlen(input);
+    PureLinkSender *l = a && a <= PURE_LINK_NAME && b && b <= PURE_LINK_NAME ? calloc(1, sizeof(*l)) : NULL;
+    if (!l) return NULL;
+    memcpy(l->source, source, a);
+    memcpy(l->input, input, b);
+    l->shape = shape;
+    l->own = own;
+    memcpy(l->own_public, own_public, PURE_SEAL_KEY);
+    l->own_id = key_id(own_public);
+    l->run = random64();
+    l->window = WINDOW_FIRST;
+    l->threshold = UINT64_MAX;
+    l->credit = CREDIT;
+    return l;
+}
+
+bool pure_link_sender_peer(PureLinkSender *l, const unsigned char peer[PURE_SEAL_KEY]) {
+    memcpy(l->peer, peer, PURE_SEAL_KEY);
+    l->peer_id = key_id(peer);
+    pure_aead_free(l->out);
+    pure_aead_free(l->in);
+    l->out = l->in = NULL;
+    return l->own && pure_link_secret(l->own, l->peer, l->secret) &&
+           (l->out = aead_of(l->secret, l->own_public, l->peer, l->run, 0, l->out_base));
+}
+
+uint64_t pure_link_sender_queued(const PureLinkSender *l) { return l->queued; }
+
+uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; }
+
+void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) {
+    *out = l->counts;
+    out->stored = l->stored;
+}
+
+void pure_link_sender_free(PureLinkSender *l) {
+    if (!l) return;
+    for (size_t k = 0; k < l->count; ++k) free(record_at(l, k)->more);
+    free(l->records);
+    free(l->ring);
+    free(l->flights);
+    pure_aead_free(l->out);
+    pure_aead_free(l->in);
+    explicit_bzero(l->secret, sizeof(l->secret));
+    free(l);
+}
+
+/* ---- The receiving side
+ *
+ * A sender allowed is a peer: its key and the sources it may send as, a few
+ * bytes each, found by its key's id. What a sending peer needs besides (the
+ * secret its key shares with the node's, its runs' keys and addresses, its
+ * streams) is its active state: made at its first packet (one X25519), let
+ * go once the peer has been quiet for the idle period with nothing of it
+ * waiting here. So a link at rest costs the node its allowing, and the next
+ * packet goes on without a handshake. */
+
+typedef struct { uint64_t first, last; } Range;
+typedef struct Active Active;
+typedef struct {
+    Active *active;
+    uint64_t run, seen; /* the sender's run; when it last sent */
+    bool used, ack_due, due;
+    unsigned char in_key[PURE_AEAD_KEY], out_key[PURE_AEAD_KEY], in_base[PURE_AEAD_NONCE], out_base[PURE_AEAD_NONCE];
+    uint64_t next_pn, largest_at;
+    _Alignas(8) unsigned char address[PURE_LINK_ADDRESS]; /* where it sends from: a socket's address */
+    size_t address_size;
+    Range got[RANGES]; /* the numbers received, newest first */
+    unsigned got_count;
+} Run;
+typedef struct {
+    uint64_t seq;
+    unsigned char *bytes; /* its size, then a byte a piece: had */
+    uint32_t size, pieces, got;
+} Held;
+typedef struct Stream {
+    Active *active;
+    char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1];
+    uint32_t input, shape; /* the host's number for the input; the messages' shape */
+    uint64_t heard;              /* the last number heard (stored), once known */
+    bool known, out, stored_due, gap, ready;
+    /* The line: once the last heard is known, whole records from the one
+     * after it on, in order, as a delivery carries them (u32 size, bytes),
+     * from line_head: handed over as they are, let go by moving the head.
+     * Records that come early or in pieces, or before the last heard is
+     * known, wait in held until the line reaches them. */
+    unsigned char *line;
+    size_t line_head, line_size, line_capacity;
+    uint64_t line_first;
+    unsigned line_count;
+    Held *held;                  /* records not yet stored, by number: held_count from held_first */
+    size_t held_first, held_count, held_capacity, held_bytes;
+    Run *run;                    /* the run that sent to it last: where STORED goes */
+    uint64_t run_id;
+} Stream;
+struct Active {
+    uint32_t peer;               /* its place among the peers */
+    unsigned char secret[32];
+    uint64_t seen;
+    size_t place;                /* among the active */
+    Run runs[RUNS];
+    Stream **streams;
+    unsigned stream_count, stream_capacity;
+};
+typedef struct {
+    unsigned char key[PURE_SEAL_KEY];
+    uint32_t id, allowed; /* allowed: the place + 1 of its first source (0: none) */
+    Active *active;
+} Peer;
+typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 length, bytes); next: a place + 1 */
+
+/* A peer quiet this long lets its active state go; the active are looked at
+ * that often; what they held goes back to the system once TRIM_AFTER have
+ * gone, or TRIM_WAIT after any went. */
+static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS;
+enum { TRIM_AFTER = 1024 };
+
+struct PureLinkReceiver {
+    PureLinkHost host;
+    const PureSealKey *own;
+    unsigned char own_public[PURE_SEAL_KEY];
+    uint32_t own_id;
+    uint64_t run, idle, swept, trimmed, untrimmed;
+    PureAead *aead; /* keyed by each packet's run */
+    unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */
+    bool open_keyed, seal_keyed;
+    unsigned char *memo; /* the secrets remembered (pure_link_remember), mapped */
+    size_t memo_size;
+    Peer *peers;
+    size_t peer_count, peer_capacity;
+    uint32_t *index; /* the peers by key id: open addressing, a place + 1 (0: empty) */
+    size_t index_size;
+    Allowed *allowed;
+    size_t allowed_count, allowed_capacity;
+    unsigned char *names;
+    size_t names_size, names_capacity;
+    Active **actives;
+    size_t active_count, active_capacity;
+    Stream **ready; /* streams that may have whole records to hand over */
+    size_t ready_count, ready_capacity;
+    Run **due; /* runs to answer */
+    size_t due_count, due_capacity;
+    unsigned char *batch;
+    size_t batch_capacity;
+    PureLinkCounts counts;
+};
+
+PureLinkReceiver *pure_link_receiver(const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
+                                     const PureLinkHost *host, char *error, size_t size) {
+    if (!own) {
+        snprintf(error, size, "a receiver needs the node's key");
+        return NULL;
+    }
+    if (!pure_seal_ready(error, size)) return NULL;
+    PureLinkReceiver *s = calloc(1, sizeof(*s));
+    unsigned char key[PURE_AEAD_KEY] = {0}; /* each packet brings its run's */
+    if (!s || !(s->aead = pure_aead(key))) {
+        free(s);
+        snprintf(error, size, "out of memory");
+        return NULL;
+    }
+    s->host = *host;
+    s->own = own;
+    memcpy(s->own_public, own_public, PURE_SEAL_KEY);
+    s->own_id = key_id(own_public);
+    s->run = random64();
+    s->idle = IDLE_FIRST;
+    return s;
+}
+
+void pure_link_idle(PureLinkReceiver *s, uint64_t ms) { s->idle = ms * MS; }
+
+/* items (count used of capacity) with room for one more: the same or grown,
+ * or NULL when memory runs out (items stay as they were). */
+static void *room(void *items, size_t *capacity, size_t count, size_t item_size, size_t first) {
+    if (count < *capacity) return items;
+    size_t more = *capacity ? 2 * *capacity : first;
+    void *grown = realloc(items, more * item_size);
+    if (grown) *capacity = more;
+    return grown;
+}
+
+/* The peers by key id again, in an index twice as large. */
+static bool grow_index(PureLinkReceiver *s) {
+    size_t size = s->index_size ? 2 * s->index_size : 1024;
+    uint32_t *index = calloc(size, sizeof(*index));
+    if (!index) return false;
+    for (size_t p = 0; p < s->peer_count; ++p) {
+        size_t slot = s->peers[p].id & (size - 1);
+        while (index[slot]) slot = (slot + 1) & (size - 1);
+        index[slot] = (uint32_t)p + 1;
+    }
+    free(s->index);
+    s->index = index;
+    s->index_size = size;
+    return true;
+}
+
+static bool same_name(const PureLinkReceiver *s, uint32_t name, const char *source, size_t length) {
+    return s->names[name] == length && !memcmp(s->names + name + 1, source, length);
+}
+
+/* Keys are found by their ids, which are 32 bits: two keys may share one. */
+bool pure_link_allow(PureLinkReceiver *s, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
+                     size_t size) {
+    size_t length = strlen(source), p = SIZE_MAX;
+    uint32_t id = key_id(key);
+    if (!length || length > PURE_LINK_NAME) {
+        snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME);
+        return false;
+    }
+    for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot];
+         slot = (slot + 1) & (s->index_size - 1)) {
+        const Peer *peer = &s->peers[s->index[slot] - 1];
+        if (peer->id == id && !memcmp(peer->key, key, PURE_SEAL_KEY)) {
+            p = s->index[slot] - 1;
+            break;
+        }
+    }
+    if (p == SIZE_MAX) {
+        Peer *peers = s->peer_count < UINT32_MAX - 1 && ((s->peer_count + 1) * 2 <= s->index_size || grow_index(s)) ?
+                      room(s->peers, &s->peer_capacity, s->peer_count, sizeof(Peer), 64) : NULL;
+        if (!peers) goto full;
+        s->peers = peers;
+        p = s->peer_count++;
+        s->peers[p] = (Peer){.id = id};
+        memcpy(s->peers[p].key, key, PURE_SEAL_KEY);
+        size_t slot = id & (s->index_size - 1);
+        while (s->index[slot]) slot = (slot + 1) & (s->index_size - 1);
+        s->index[slot] = (uint32_t)p + 1;
+    }
+    for (uint32_t a = s->peers[p].allowed; a; a = s->allowed[a - 1].next)
+        if (same_name(s, s->allowed[a - 1].name, source, length)) return true;
+    Allowed *kept = s->allowed_count < UINT32_MAX - 1 && s->names_size + 1 + length <= UINT32_MAX ?
+                    room(s->allowed, &s->allowed_capacity, s->allowed_count, sizeof(Allowed), 64) : NULL;
+    if (!kept) goto full;
+    s->allowed = kept;
+    if (s->names_size + 1 + length > s->names_capacity) {
+        size_t more = s->names_capacity ? 2 * s->names_capacity : 4096;
+        unsigned char *names = realloc(s->names, more);
+        if (!names) goto full;
+        s->names = names;
+        s->names_capacity = more;
+    }
+    s->names[s->names_size] = (unsigned char)length;
+    memcpy(s->names + s->names_size + 1, source, length);
+    s->allowed[s->allowed_count] = (Allowed){(uint32_t)s->names_size, s->peers[p].allowed};
+    s->peers[p].allowed = (uint32_t)++s->allowed_count;
+    s->names_size += 1 + length;
+    return true;
+full:
+    snprintf(error, size, "out of memory");
+    return false;
+}
+
+static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; }
+
+static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; }
+
+/* Room for more bytes at the line's end: its front let go given back
+ * first when that is at least half of it (so no byte moves twice in a
+ * doubling), else a longer line. */
+static bool line_room(Stream *st, size_t more) {
+    if (st->line_size + more <= st->line_capacity) return true;
+    if (st->line_head && st->line_head >= st->line_capacity / 2) {
+        memmove(st->line, st->line + st->line_head, st->line_size - st->line_head);
+        st->line_size -= st->line_head;
+        st->line_head = 0;
+        if (st->line_size + more <= st->line_capacity) return true;
+    }
+    size_t capacity = st->line_capacity ? st->line_capacity : 4096;
+    while (capacity < st->line_size + more) capacity *= 2;
+    unsigned char *grown = realloc(st->line, capacity);
+    if (!grown) return false;
+    st->line = grown;
+    st->line_capacity = capacity;
+    return true;
+}
+
+static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) {
+    if (!line_room(st, 4 + (size_t)size)) return false;
+    put32(st->line + st->line_size, size);
+    if (size) memcpy(st->line + st->line_size + 4, bytes, size);
+    st->line_size += 4 + (size_t)size;
+    ++st->line_count;
+    return true;
+}
+
+/* The line's records up to heard let go (all of them, when heard is past it). */
+static void line_drop(Stream *st, uint64_t heard) {
+    if (heard < st->line_first) return;
+    if (heard >= line_end(st)) {
+        st->line_head = st->line_size = 0;
+        st->line_count = 0;
+        st->line_first = heard + 1;
+        return;
+    }
+    for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first)
+        st->line_head += 4 + (size_t)get32(st->line + st->line_head);
+}
+
+/* Records held that the line reached, moved onto it; ones behind it let go. */
+static void drop_first(Stream *st);
+static void line_pull(Stream *st) {
+    while (st->held_count) {
+        Held *h = held_at(st, 0);
+        if (h->seq < line_end(st)) drop_first(st);
+        else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st);
+        else break;
+    }
+}
+
+/* The first record held let go: the ones after it stay where they are. */
+static void drop_first(Stream *st) {
+    st->held_bytes -= held_at(st, 0)->size;
+    free(held_at(st, 0)->bytes);
+    ++st->held_first;
+    if (!--st->held_count) st->held_first = 0;
+}
+
+/* The memo of secrets (pure_link_remember): a header, then a place a peer
+ * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a
+ * of its whole key and the secret) and the secret. A place whose check
+ * fails (never made, another key, cut short by a crash) is made again. */
+enum { MEMO_HEAD = 64, MEMO_PLACE = 64 };
+static const unsigned char memo_magic[8] = {'P', 'L', 'M', '1', 0, 0, 0, 0};
+
+static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) {
+    return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32);
+}
+
+bool pure_link_remember(PureLinkReceiver *s, const char *path, char *error, size_t size) {
+    int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
+    struct stat info;
+    unsigned char head[MEMO_HEAD];
+    size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
+    bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077);
+    /* Another node's key, or none yet: begun again. */
+    bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) &&
+                !memcmp(head + 8, s->own_public, PURE_SEAL_KEY);
+    if (ok && !ours) {
+        memset(head, 0, MEMO_HEAD);
+        memcpy(head, memo_magic, 8);
+        memcpy(head + 8, s->own_public, PURE_SEAL_KEY);
+        ok = !ftruncate(fd, 0) && pwrite(fd, head, MEMO_HEAD, 0) == MEMO_HEAD;
+    }
+    ok = ok && (ours && (size_t)info.st_size >= needed ? true : !ftruncate(fd, (off_t)needed));
+    void *map = ok ? mmap(NULL, needed, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0) : MAP_FAILED;
+    if (fd >= 0) close(fd);
+    if (map == MAP_FAILED) {
+        snprintf(error, size, "%s: cannot keep secrets there (a file only its owner may read)", path);
+        return false;
+    }
+    s->memo = map;
+    s->memo_size = needed;
+    return true;
+}
+
+/* A peer's active state, made at its first packet: the secret its key
+ * shares with ours, remembered or made (one X25519, then remembered: the
+ * secret first, the key that makes the place its own last). */
+static Active *activate(PureLinkReceiver *s, Peer *peer, uint64_t now) {
+    Active *a = calloc(1, sizeof(*a)), **actives = room(s->actives, &s->active_capacity, s->active_count, sizeof(*actives), 64);
+    if (actives) s->actives = actives;
+    if (!a || !actives) {
+        free(a);
+        return NULL;
+    }
+    size_t index = (size_t)(peer - s->peers);
+    unsigned char *place = s->memo && MEMO_HEAD + (index + 1) * MEMO_PLACE <= s->memo_size ?
+                           s->memo + MEMO_HEAD + index * MEMO_PLACE : NULL;
+    if (place && !memcmp(place, peer->key, 28) && get32(place + 28) == memo_check(peer->key, place + 32)) {
+        memcpy(a->secret, place + 32, 32);
+        ++s->counts.remembered;
+    } else if (pure_link_secret(s->own, peer->key, a->secret)) {
+        if (place) {
+            memset(place, 0, 28);
+            __atomic_thread_fence(__ATOMIC_RELEASE);
+            memcpy(place + 32, a->secret, 32);
+            put32(place + 28, memo_check(peer->key, a->secret));
+            __atomic_thread_fence(__ATOMIC_RELEASE);
+            memcpy(place, peer->key, 28);
+        }
+    } else {
+        explicit_bzero(a, sizeof(*a));
+        free(a);
+        return NULL;
+    }
+    a->peer = (uint32_t)(peer - s->peers);
+    a->seen = now;
+    a->place = s->active_count;
+    s->actives[s->active_count++] = a;
+    peer->active = a;
+    return a;
+}
+
+static void let_go(PureLinkReceiver *s, Active *a) {
+    for (unsigned k = 0; k < a->stream_count; ++k) {
+        while (a->streams[k]->held_count) drop_first(a->streams[k]);
+        free(a->streams[k]->held);
+        free(a->streams[k]->line);
+        free(a->streams[k]);
+    }
+    free(a->streams);
+    s->peers[a->peer].active = NULL;
+    s->actives[a->place] = s->actives[--s->active_count];
+    s->actives[a->place]->place = a->place;
+    explicit_bzero(a, sizeof(*a));
+    free(a);
+}
+
+/* Whether any of a peer's records or answers waits here. */
+static bool busy(const Active *a) {
+    for (unsigned k = 0; k < a->stream_count; ++k)
+        if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true;
+    for (unsigned i = 0; i < RUNS; ++i)
+        if (a->runs[i].due) return true;
+    return false;
+}
+
+/* Let go the active state of peers quiet for the idle period; what idle
+ * links held goes back to the system. */
+static void sweep(PureLinkReceiver *s, uint64_t now) {
+    if (now - s->swept < SWEEP_EVERY) return;
+    s->swept = now;
+    size_t freed = 0;
+    for (size_t k = 0; k < s->active_count;) {
+        Active *a = s->actives[k];
+        if (now - a->seen < s->idle || busy(a)) ++k;
+        else let_go(s, a), ++freed;
+    }
+    s->counts.swept += freed;
+    s->untrimmed += freed;
+    if (s->untrimmed >= TRIM_AFTER || (s->untrimmed && now - s->trimmed >= TRIM_WAIT)) {
+        malloc_trim(0);
+        s->untrimmed = 0;
+        s->trimmed = now;
+    }
+}
+
+static void mark_ready(PureLinkReceiver *s, Stream *st) {
+    Stream **ready = st->ready ? NULL : room(s->ready, &s->ready_capacity, s->ready_count, sizeof(*ready), 64);
+    if (!ready) return;
+    s->ready = ready;
+    s->ready[s->ready_count++] = st;
+    st->ready = true;
+}
+
+static void mark_due(PureLinkReceiver *s, Run *r) {
+    Run **due = r->due ? NULL : room(s->due, &s->due_capacity, s->due_count, sizeof(*due), 64);
+    if (!due) return;
+    s->due = due;
+    s->due[s->due_count++] = r;
+    r->due = true;
+}
+
+static Run *run_found(Active *a, uint64_t run) {
+    for (unsigned i = 0; i < RUNS; ++i)
+        if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i];
+    return NULL;
+}
+
+/* A new run's keys, in the slot unused longest: only once its first packet
+ * opened with them, so a forged one displaces nothing. */
+static Run *install_run(Active *a, const Run *made) {
+    Run *slot = &a->runs[0];
+    for (unsigned i = 0; i < RUNS; ++i) {
+        Run *r = &a->runs[i];
+        if (!r->used || (slot->used && r->seen < slot->seen)) slot = r;
+    }
+    bool due = slot->due; /* the answer list may hold the slot */
+    *slot = *made;
+    slot->due = due;
+    return slot;
+}
+
+static bool had(const Run *r, uint64_t pn) {
+    for (unsigned i = 0; i < r->got_count; ++i)
+        if (pn >= r->got[i].first && pn <= r->got[i].last) return true;
+    return false;
+}
+
+static void note(Run *r, uint64_t pn, uint64_t now) {
+    unsigned i = 0, n = r->got_count;
+    while (i < n && r->got[i].first > pn + 1) ++i;
+    if (i < n && r->got[i].first == pn + 1) { /* joins the range above it */
+        r->got[i].first = pn;
+        if (i + 1 < n && r->got[i + 1].last + 1 == pn) {
+            r->got[i].first = r->got[i + 1].first;
+            memmove(&r->got[i + 1], &r->got[i + 2], (n - i - 2) * sizeof(Range));
+            --r->got_count;
+        }
+    } else if (i < n && r->got[i].last + 1 == pn) {
+        r->got[i].last = pn;
+    } else if (i < RANGES) {
+        if (n == RANGES) --n; /* the oldest range is forgotten */
+        memmove(&r->got[i + 1], &r->got[i], (n - i) * sizeof(Range));
+        r->got[i] = (Range){pn, pn};
+        r->got_count = n + 1;
+    }
+    if (r->got_count && r->got[0].last == pn) r->largest_at = now;
+}
+
+static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *source) {
+    for (uint32_t a = peer->allowed; a; a = s->allowed[a - 1].next)
+        if (same_name(s, s->allowed[a - 1].name, source, strlen(source))) return true;
+    return false;
+}
+
+/* The stream of a peer's source to input, made the first time: NULL when the
+ * source is not allowed from the peer, or the host has no input that takes
+ * such messages. */
+static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape) {
+    for (unsigned k = 0; k < a->stream_count; ++k) {
+        Stream *st = a->streams[k];
+        if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL;
+    }
+    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1;
+    if (index < 0) return NULL;
+    size_t capacity = a->stream_capacity;
+    Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
+    if (!streams) return NULL;
+    a->streams = streams;
+    a->stream_capacity = (unsigned)capacity;
+    if (!(st = calloc(1, sizeof(*st)))) return NULL;
+    st->active = a;
+    st->input = (uint32_t)index;
+    st->shape = shape;
+    snprintf(st->source, sizeof(st->source), "%s", source);
+    snprintf(st->input_name, sizeof(st->input_name), "%s", input);
+    a->streams[a->stream_count++] = st;
+    return st;
+}
+
+/* A record's piece, kept until the record is whole and handed over. */
+static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const unsigned char *bytes, uint32_t length) {
+    if (st->known && seq <= st->heard) {
+        st->stored_due = true; /* sent again: say again how far it is stored */
+        return;
+    }
+    if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return;
+    if (st->known && seq < line_end(st)) return; /* on the line already */
+    if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */
+        if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size))
+            line_pull(st);
+        return;
+    }
+    /* In order it goes last; out of order, its place is found by halving. */
+    size_t k = st->held_count;
+    if (k && held_at(st, k - 1)->seq >= seq) {
+        size_t low = 0;
+        while (low < k) {
+            size_t middle = low + (k - low) / 2;
+            if (held_at(st, middle)->seq < seq) low = middle + 1;
+            else k = middle;
+        }
+    }
+    if (k == st->held_count || held_at(st, k)->seq != seq) {
+        if (st->held_bytes + size > HELD_MOST) return;
+        if (st->held_first && st->held_first + st->held_count == st->held_capacity) { /* the front is free: used again */
+            memmove(st->held, held_at(st, 0), st->held_count * sizeof(Held));
+            st->held_first = 0;
+        }
+        Held *kept = room(st->held, &st->held_capacity, st->held_first + st->held_count, sizeof(Held), 16);
+        if (!kept) return;
+        st->held = kept;
+        Held h = {.seq = seq, .size = size, .pieces = size ? (size + PIECE - 1) / PIECE : 1};
+        if (!(h.bytes = malloc(size + h.pieces))) return;
+        memset(h.bytes + size, 0, h.pieces);
+        memmove(held_at(st, k + 1), held_at(st, k), (st->held_count - k) * sizeof(Held));
+        *held_at(st, k) = h;
+        ++st->held_count;
+        st->held_bytes += size;
+    }
+    Held *h = held_at(st, k);
+    if (h->size != size) return;
+    if (!offset && length == size) { /* whole: every piece had */
+        if (size) memcpy(h->bytes, bytes, size);
+        memset(h->bytes + size, 1, h->pieces);
+        h->got = h->pieces;
+        return;
+    }
+    uint32_t piece = offset / PIECE;
+    if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return;
+    if (length) memcpy(h->bytes + offset, bytes, length);
+    h->bytes[size + piece] = 1;
+    ++h->got;
+}
+
+/* Whole records from the one after the last heard, in order, to the host:
+ * one delivery a stream at a time. Before the last heard is known (after a
+ * restart) they go from the first held: the host says how far it heard.
+ * False when the host could not take them now. */
+static bool hand_over(PureLinkReceiver *s, Stream *st) {
+    if (st->out) return true;
+    if (st->known) { /* the line as it is, up to BATCH records */
+        line_pull(st);
+        if (!st->line_count) return true;
+        unsigned n = 0;
+        size_t bytes = 0;
+        for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes);
+        PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st};
+        if (!s->host.deliver(s->host.context, &d)) return false;
+        st->out = true;
+        ++s->counts.deliveries;
+        return true;
+    }
+    if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true;
+    size_t n = 0, bytes = 0;
+    while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces &&
+           held_at(st, n)->seq == held_at(st, 0)->seq + n)
+        bytes += 4 + held_at(st, n++)->size;
+    if (!n) return true;
+    if (bytes > s->batch_capacity) {
+        unsigned char *grown = realloc(s->batch, bytes);
+        if (!grown) return false;
+        s->batch = grown;
+        s->batch_capacity = bytes;
+    }
+    size_t at = 0;
+    for (size_t k = 0; k < n; ++k) {
+        put32(s->batch + at, held_at(st, k)->size);
+        memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size);
+        at += 4 + held_at(st, k)->size;
+    }
+    PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st};
+    if (!s->host.deliver(s->host.context, &d)) return false;
+    st->out = true;
+    ++s->counts.deliveries;
+    return true;
+}
+
+/* A packet from peer, if it opens with the keys of peer's run (made for a
+ * run not seen yet, and for a peer at rest its active state): its frames. */
+static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *packet, size_t n, const void *from,
+                      size_t from_size, uint64_t now) {
+    uint64_t pn = get64(packet + 13), run = get64(packet + 5);
+    Active *a = peer->active;
+    Run *r = a ? run_found(a, run) : NULL;
+    if (r && had(r, pn)) {
+        r->ack_due = true; /* its acknowledgment was lost, perhaps */
+        mark_due(s, r);
+        return true;
+    }
+    bool woke = !a;
+    if (woke && !(a = activate(s, peer, now))) return false;
+    Run made = {.active = a, .run = run, .seen = now, .used = true};
+    bool keyed = r || (derive(a->secret, peer->key, s->own_public, run, 0, made.in_key, made.in_base) &&
+                       derive(a->secret, s->own_public, peer->key, s->run, run, made.out_key, made.out_base));
+    unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
+    nonce_of(nonce, r ? r->in_base : made.in_base, pn);
+    size_t size = n - HEADER - PURE_AEAD_TAG;
+    const unsigned char *key = r ? r->in_key : made.in_key, *give = key;
+    if (s->open_keyed && !memcmp(s->open_key, key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
+    else if (keyed) memcpy(s->open_key, key, PURE_AEAD_KEY), s->open_keyed = true;
+    if (!keyed || !pure_aead_open(s->aead, give, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) {
+        explicit_bzero(&made, sizeof(made));
+        if (woke) let_go(s, a);
+        return false;
+    }
+    if (woke) ++s->counts.woken;
+    if (!r) r = install_run(a, &made);
+    a->seen = r->seen = now;
+    note(r, pn, now);
+    r->address_size = from_size <= sizeof(r->address) ? from_size : sizeof(r->address);
+    memcpy(r->address, from, r->address_size); /* where it is now: answers go there */
+    ++s->counts.received;
+    Stream *st = NULL;
+    bool eliciting = false;
+    for (size_t at = 0; at < size;) {
+        unsigned type = plain[at++];
+        if (type == F_STREAM) {
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
+            size_t names = get_names(plain + at, size - at, source, input);
+            if (!names || at + names + 4 > size) break;
+            st = stream_of(s, a, source, input, get32(plain + at + names));
+            if (st) {
+                st->run = r;
+                st->run_id = r->run;
+            }
+            at += names + 4;
+            eliciting = true;
+        } else if (type == F_RECORD) {
+            if (at + RECORD_HEAD - 1 > size) break;
+            uint32_t length = get16(plain + at + 16);
+            if (at + RECORD_HEAD - 1 + length > size) break;
+            if (st) {
+                hold(st, get64(plain + at), get32(plain + at + 8), get32(plain + at + 12), plain + at + 18, length);
+                mark_ready(s, st);
+            }
+            at += RECORD_HEAD - 1 + length;
+            eliciting = true;
+        } else if (type == F_RECORDS) {
+            if (at + RECORDS_HEAD - 1 > size) break;
+            uint64_t first = get64(plain + at), length = 0;
+            unsigned count = plain[at + 8];
+            at += RECORDS_HEAD - 1;
+            unsigned i = 0;
+            for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) &&
+                              length <= size - at - used; ++i) {
+                if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length);
+                at += used + length;
+            }
+            if (st) mark_ready(s, st);
+            eliciting = true;
+            if (i < count) break; /* cut short: the rest is not read */
+        } else if (type == F_PING) {
+            eliciting = true;
+        } else
+            break; /* padding, or what this side does not know: the rest */
+    }
+    if (eliciting) {
+        r->ack_due = true;
+        mark_due(s, r);
+    }
+    return true;
+}
+
+void pure_link_receiver_take(PureLinkReceiver *s, const unsigned char *packet, size_t n, const void *from,
+                             size_t from_size, uint64_t now) {
+    if (n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || !s->index_size) return;
+    uint32_t id = get32(packet + 1);
+    size_t mask = s->index_size - 1;
+    uint64_t run = get64(packet + 5);
+    /* The peers with the packet's key id: first one whose run is known (no
+     * key to make), then any. */
+    for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
+        Peer *peer = &s->peers[s->index[slot] - 1];
+        if (peer->id == id && peer->active && run_found(peer->active, run) &&
+            take_from(s, peer, packet, n, from, from_size, now))
+            return;
+    }
+    for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
+        Peer *peer = &s->peers[s->index[slot] - 1];
+        if (peer->id == id && !(peer->active && run_found(peer->active, run)) &&
+            take_from(s, peer, packet, n, from, from_size, now))
+            return;
+    }
+}
+
+/* Acknowledgments and how far each stream is stored, a packet to each run
+ * that has any. How far a stream is stored goes with every acknowledgment
+ * to its run, so a lost word is said again at the next packet. */
+static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *context) {
+    unsigned char body[BODY], packet[DATAGRAM];
+    for (size_t k = 0; k < s->due_count; ++k) {
+        Run *r = s->due[k];
+        Active *a = r->active;
+        r->due = false;
+        if (!r->used) continue;
+        size_t at = 0;
+        if (r->ack_due && r->got_count) {
+            body[at++] = F_ACK;
+            put64(body + at, r->got[0].last);
+            put32(body + at + 8, (uint32_t)least((now - r->largest_at) / 1000, UINT32_MAX));
+            body[at + 12] = (unsigned char)r->got_count;
+            at += 13;
+            for (unsigned i = 0; i < r->got_count; ++i, at += 16) {
+                put64(body + at, r->got[i].first);
+                put64(body + at + 8, r->got[i].last);
+            }
+        }
+        for (unsigned j = 0; j < a->stream_count; ++j) {
+            Stream *st = a->streams[j];
+            if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue;
+            if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */
+            body[at++] = F_STORED;
+            at += put_names(body + at, st->source, st->input_name);
+            put64(body + at, st->heard);
+            body[at + 8] = st->gap ? STORED_GAP : 0;
+            size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
+            put32(body + at + 9, (uint32_t)room);
+            at += 13;
+            st->stored_due = st->gap = false;
+        }
+        r->ack_due = false;
+        if (!at) continue;
+        const unsigned char *give = r->out_key;
+        if (s->seal_keyed && !memcmp(s->seal_key, r->out_key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
+        else memcpy(s->seal_key, r->out_key, PURE_AEAD_KEY), s->seal_keyed = true;
+        size_t total = seal_packet(s->aead, give, r->out_base, s->own_id, s->run, r->next_pn++, body, at, packet);
+        if (total && send(context, packet, total, r->address, r->address_size)) ++s->counts.packets;
+    }
+    s->due_count = 0;
+}
+
+/* What the host said of its deliveries: the saved records let go, and how
+ * far they are stored said. */
+static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t count) {
+    for (size_t k = 0; k < count; ++k) {
+        Stream *st = done[k].token;
+        st->out = false;
+        mark_ready(s, st);
+        if (!done[k].stored) continue; /* not saved: handed over again */
+        if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
+        if (!st->known) st->line_first = st->heard + 1; /* the line begins */
+        line_drop(st, st->heard);
+        while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st);
+        st->known = true;
+        st->gap = done[k].gap;
+        st->stored_due = true;
+        if (st->run && st->run->used && st->run_id == st->run->run) mark_due(s, st->run);
+        s->counts.stored = st->heard;
+    }
+}
+
+/* The streams ready, each handed over; one the host could not take now is
+ * tried again at the next turn. */
+static void hand_over_ready(PureLinkReceiver *s) {
+    size_t count = s->ready_count;
+    for (size_t k = 0; k < count; ++k) {
+        Stream *st = s->ready[k];
+        st->ready = false;
+        if (!hand_over(s, st)) mark_ready(s, st);
+    }
+    memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready));
+    s->ready_count -= count;
+}
+
+void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size_t count, uint64_t now,
+                             PureLinkSend *send, void *context) {
+    if (!s->swept) s->swept = s->trimmed = now;
+    apply_done(s, done, count);
+    hand_over_ready(s);
+    answer(s, now, send, context);
+    sweep(s, now);
+}
+
+uint64_t pure_link_receiver_due(const PureLinkReceiver *s) {
+    return s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0;
+}
+
+void pure_link_receiver_counts(const PureLinkReceiver *s, PureLinkCounts *counts) {
+    *counts = s->counts;
+    counts->active = s->active_count;
+}
+
+void pure_link_receiver_free(PureLinkReceiver *s) {
+    if (!s) return;
+    while (s->active_count) let_go(s, s->actives[0]);
+    pure_aead_free(s->aead);
+    if (s->memo) munmap(s->memo, s->memo_size);
+    free(s->actives);
+    free(s->peers);
+    free(s->index);
+    free(s->allowed);
+    free(s->names);
+    free(s->ready);
+    free(s->due);
+    free(s->batch);
+    free(s);
+}
diff --git a/link/link-protocol.h b/link/link-protocol.h
new file mode 100644
index 0000000..b0f8ef3
--- /dev/null
+++ b/link/link-protocol.h
@@ -0,0 +1,160 @@
+#ifndef PURE_LINK_PROTOCOL_H
+#define PURE_LINK_PROTOCOL_H
+#include "seal.h"
+
+/* Links without connections (docs/design/011-pure-transport.md), as state
+ * alone: no socket, thread or clock of their own. The owner gives each side
+ * the packets that came and the time (ns, monotonic), and sends the packets
+ * each side makes; link/link.h is such an owner over UDP. So one thread may
+ * carry many links, and a test may carry both sides through a network of its
+ * own making.
+ *
+ * A link's lasting state is the apps' own: the sender's outbox (what to send
+ * and send again) and the receiver's heard numbers (what it applied, saved
+ * with the events). What either side keeps in memory is a cache, so either
+ * may restart and the other goes on with the next packet.
+ *
+ * A packet is a header in the clear, then frames sealed with AES-128-GCM
+ * (the header as associated data):
+ *   u8 0x70 (version 0), u32le the key id of the node that sent it (FNV-1a
+ *   of its public key; two keys may share an id, and a receiver tries the
+ *   keys it allows with it), u64le that node's run (random, new each time it
+ *   starts, so no nonce repeats under a key across restarts), u64le the
+ *   packet's number in its run.
+ * Keys come from the secret the two nodes' keys share (X25519) by
+ * HKDF-SHA256, with both public keys and the sending node's run: a sender's
+ * packets use its own run; a receiver's answers add the run they answer,
+ * so an answer opens only in that run. The nonce is the key's base with the
+ * packet's number in its last eight bytes (XOR, big-endian, as QUIC does).
+ *
+ * Frames (numbers little-endian):
+ *   0x01 STREAM  source (u8 length, bytes), input (u8 length, bytes), u32 the
+ *                shape of the messages (the sender's output and the
+ *                receiver's input must have the same): the stream the
+ *                records after it are of
+ *   0x02 RECORD  u64 number, u32 size, u32 offset, u16 length, bytes: a piece
+ *                of a saved send (pieces are 1000 bytes, the last shorter)
+ *   0x03 ACK     u64 largest number, u32 delay (µs), u8 ranges, each as u64
+ *                first and u64 last, newest first
+ *   0x04 STORED  source, input (as STREAM), u64 the last number heard
+ *                (applied and saved), u8 flags (1: a record after a gap was
+ *                not taken: send what follows it again), u32 credit (records
+ *                it takes past that)
+ *   0x05 PING    to be acknowledged
+ *   0x06 RECORDS u64 the first one's number, u8 count, then whole records one
+ *                after another (numbers following on), each its size (a
+ *                varint: LEB128) and its bytes: a send that fits the packet
+ *                whole, with 1 to 3 bytes of framing, not the 19 of RECORD
+ *   0x00 PADDING to the end of the packet: a full packet sent in a burst is
+ *                a whole datagram
+ * Losses are detected as RFC 9002 says; congestion control is its NewReno
+ * (RFC 9002 §7). */
+
+/* A packet is at most PURE_LINK_DATAGRAM bytes; a source's or an input's
+ * name is 1 to PURE_LINK_NAME bytes; a receiver keeps where a packet came
+ * from (the socket's address) in PURE_LINK_ADDRESS bytes. */
+enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 28 };
+
+/* Counts for a host's report: packets and records sent, records sent again,
+ * packets lost and acknowledged, packets received, deliveries handed over,
+ * the last number stored; a receiver's senders woken (their active state
+ * made), of them with a secret remembered (no X25519), let go when idle,
+ * and active now. */
+typedef struct {
+    uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active;
+} PureLinkCounts;
+
+/* ---- The sending side: one output's saved sends to a receiver's input, as
+ * source (the receiver's name for it), their messages of shape (the
+ * receiver's input must have the same: core/module.h's pure_module_shape),
+ * from own (this node's key, whose public half is own_public). NULL when
+ * the names are not 1 to PURE_LINK_NAME bytes, or memory runs out. */
+typedef struct PureLinkSender PureLinkSender;
+PureLinkSender *pure_link_sender(const char *source, const char *input, uint32_t shape, const PureSealKey *own,
+                                 const unsigned char own_public[PURE_SEAL_KEY]);
+/* The receiver's public key: the keys made from it (one X25519). A sender
+ * makes no packet before it has one. */
+bool pure_link_sender_peer(PureLinkSender *, const unsigned char peer[PURE_SEAL_KEY]);
+/* A saved send, numbered one past the last queued (any number when none
+ * waits); false when it cannot be kept. */
+bool pure_link_sender_queue(PureLinkSender *, uint64_t seq, const unsigned char *bytes, size_t size);
+/* A packet that came (the receiver's answers; others are let be). */
+void pure_link_sender_take(PureLinkSender *, const unsigned char *packet, size_t size, uint64_t now);
+/* The timers due by now: losses found, probes, records sent again. */
+void pure_link_sender_tick(PureLinkSender *, uint64_t now);
+/* The next packet to send into out, and its number: its size, or 0 when
+ * there is none (nothing to send, or the window is full). whole: a full
+ * packet is padded to PURE_LINK_DATAGRAM (so packets may go in a burst of
+ * whole datagrams, as UDP GSO sends them). */
+size_t pure_link_sender_packet(PureLinkSender *, unsigned char out[PURE_LINK_DATAGRAM], bool whole, uint64_t now,
+                               uint64_t *number);
+/* A packet made but not sent (the socket refused it): what it carried is
+ * sent again, and it counts as never sent. */
+void pure_link_sender_refused(PureLinkSender *, uint64_t number);
+/* When the next timer is due (ns), or 0 for none. */
+uint64_t pure_link_sender_due(const PureLinkSender *);
+/* The last number queued; the last the receiver stored. */
+uint64_t pure_link_sender_queued(const PureLinkSender *);
+uint64_t pure_link_sender_stored(const PureLinkSender *);
+void pure_link_sender_counts(const PureLinkSender *, PureLinkCounts *);
+void pure_link_sender_free(PureLinkSender *);
+
+/* ---- The receiving side: packets from the senders allowed (a source name
+ * and the key it seals with, a few bytes each however many there are: see
+ * pure_link_idle). A stream (a source to an input) goes to the host's input
+ * its name and shape find (its number, or -1: the stream is refused). Whole
+ * records of a stream, in order, go to deliver as one delivery; the host
+ * applies them from the one after the source's last heard number, stops at
+ * a gap, saves, and then says so with a done (pure_link_receiver_turn).
+ * Records wait here until then, and only the saved ones are acknowledged as
+ * stored. */
+typedef struct {
+    uint32_t input;
+    const char *source;
+    uint64_t first;
+    unsigned count;
+    const unsigned char *records; /* count of: u32le size, bytes */
+    size_t size;
+    void *token; /* for its done */
+} PureLinkDelivery;
+typedef struct {
+    int (*input)(void *context, const char *name, uint32_t shape);
+    bool (*deliver)(void *context, const PureLinkDelivery *);
+    void *context;
+} PureLinkHost;
+/* A delivery's end: heard, the last number heard; stored, its save
+ * succeeded (else it is handed over again); gap, it stopped at a gap. */
+typedef struct { void *token; uint64_t heard; bool stored, gap; } PureLinkDone;
+/* A packet to send to an address a packet came from: whether it went. */
+typedef bool PureLinkSend(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
+                          size_t to_size);
+
+typedef struct PureLinkReceiver PureLinkReceiver;
+PureLinkReceiver *pure_link_receiver(const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
+                                     const PureLinkHost *host, char *error, size_t size);
+bool pure_link_allow(PureLinkReceiver *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
+                     size_t size);
+/* A sender quiet for ms (30 s at first) with nothing of it waiting lets its
+ * keys, addresses and streams go: an idle link costs the receiver only its
+ * allowing, and the sender's next packet makes them again (one X25519). */
+void pure_link_idle(PureLinkReceiver *, uint64_t ms);
+/* Remember the secret each sender's key shares with the node's in a file
+ * (path, 0600; begun again for another node's key): a place a sender in
+ * the order allowed, mapped, so a sender at rest takes no memory for it.
+ * A node that starts again then makes no X25519 for a sender it heard from
+ * before. After the senders are allowed, before the first packet. */
+bool pure_link_remember(PureLinkReceiver *, const char *path, char *error, size_t size);
+/* A packet that came from an address (from_size bytes, at most
+ * PURE_LINK_ADDRESS kept): answers to its sender go there. */
+void pure_link_receiver_take(PureLinkReceiver *, const unsigned char *packet, size_t size, const void *from,
+                             size_t from_size, uint64_t now);
+/* After the packets that came: the dones (count of them) applied, whole
+ * records handed over, answers sent through send, and senders at rest let
+ * go. */
+void pure_link_receiver_turn(PureLinkReceiver *, const PureLinkDone *done, size_t count, uint64_t now,
+                             PureLinkSend *send, void *context);
+/* When a turn is due with no packet (ns), or 0 for none. */
+uint64_t pure_link_receiver_due(const PureLinkReceiver *);
+void pure_link_receiver_counts(const PureLinkReceiver *, PureLinkCounts *);
+void pure_link_receiver_free(PureLinkReceiver *);
+#endif
diff --git a/link/link.c b/link/link.c
index 103dcf5..0588633 100644
--- a/link/link.c
+++ b/link/link.c
@@ -1,10 +1,8 @@
 #define _GNU_SOURCE
-/* Links without connections (link/link.h, docs/design/011-pure-transport.md). */
+/* Links over UDP (link/link.h): link/link-protocol.h's two sides with
+ * sockets and threads. */
 #include "link.h"
-#include <arpa/inet.h>
 #include <errno.h>
-#include <fcntl.h>
-#include <malloc.h>
 #include <netdb.h>
 #include <netinet/in.h>
 #include <netinet/udp.h>
@@ -14,16 +12,11 @@
 #include <stdlib.h>
 #include <string.h>
 #include <sys/eventfd.h>
-#include <sys/mman.h>
-#include <sys/random.h>
 #include <sys/socket.h>
-#include <sys/stat.h>
 #include <time.h>
 #include <unistd.h>
 
-enum { VERSION = 0x70, HEADER = 21, DATAGRAM = 1200, BODY = DATAGRAM - HEADER - PURE_AEAD_TAG, PIECE = 1000 };
-enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, STORED_GAP = 1 };
-enum { RECORDS_HEAD = 10 }; /* type, the first number, the count */
+enum { DATAGRAM = PURE_LINK_DATAGRAM };
 /* Packets sent at once with UDP GSO: each but the last is a whole datagram
  * (a full one padded), and 54 of them fit a UDP datagram's 65,507 bytes. */
 enum { BURST = 54 };
@@ -33,47 +26,7 @@ enum { BURST = 54 };
 #ifndef UDP_GRO
 #define UDP_GRO 104
 #endif
-enum { RANGES = 32, RUNS = 2, UNITS = 64, CREDIT = 1024, BATCH = 256, RECORD_HEAD = 19 };
-enum { WINDOW_FIRST = 10 * DATAGRAM, WINDOW_LEAST = 2 * DATAGRAM };
 #define MS UINT64_C(1000000)
-/* RFC 9002's values, but a first RTT of 100 ms (a LAN's, not the 333 ms of
- * the Internet at large); a probe at least every second while a receiver is
- * away; a record received but not stored after a second is sent again. */
-static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS,
-                      STORED_WAIT = 1000 * MS, RECORD_MOST = UINT64_C(64) << 20, HELD_MOST = UINT64_C(16) << 20;
-
-static void put16(unsigned char *p, uint32_t v) { p[0] = (unsigned char)v, p[1] = (unsigned char)(v >> 8); }
-static void put32(unsigned char *p, uint32_t v) { for (int i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> 8 * i); }
-static void put64(unsigned char *p, uint64_t v) { for (int i = 0; i < 8; ++i) p[i] = (unsigned char)(v >> 8 * i); }
-static uint32_t get16(const unsigned char *p) { return (uint32_t)p[0] | (uint32_t)p[1] << 8; }
-static uint32_t get32(const unsigned char *p) {
-    return (uint32_t)p[0] | (uint32_t)p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24;
-}
-static uint64_t get64(const unsigned char *p) { return (uint64_t)get32(p) | (uint64_t)get32(p + 4) << 32; }
-/* LEB128: seven bits a byte, low first. */
-static size_t put_varint(unsigned char *p, uint64_t v) {
-    size_t n = 0;
-    for (; v >= 0x80; v >>= 7) p[n++] = (unsigned char)(v | 0x80);
-    p[n++] = (unsigned char)v;
-    return n;
-}
-static size_t varint_size(uint64_t v) {
-    size_t n = 1;
-    for (; v >= 0x80; v >>= 7) ++n;
-    return n;
-}
-/* A varint from p (size bytes): how many bytes it took, or 0. */
-static size_t get_varint(const unsigned char *p, size_t size, uint64_t *v) {
-    *v = 0;
-    for (size_t n = 0; n < size && n < 10; ++n) {
-        *v |= (uint64_t)(p[n] & 0x7f) << (7 * n);
-        if (!(p[n] & 0x80)) return n + 1;
-    }
-    return 0;
-}
-
-static uint64_t least(uint64_t a, uint64_t b) { return a < b ? a : b; }
-static uint64_t most(uint64_t a, uint64_t b) { return a > b ? a : b; }
 
 static uint64_t now_ns(void) {
     struct timespec t;
@@ -81,88 +34,6 @@ static uint64_t now_ns(void) {
     return (uint64_t)t.tv_sec * 1000000000u + (uint64_t)t.tv_nsec;
 }
 
-static uint32_t fnv(uint32_t h, const void *bytes, size_t size) {
-    const unsigned char *p = bytes;
-    for (size_t i = 0; i < size; ++i) h = (h ^ p[i]) * 16777619u;
-    return h;
-}
-
-static uint32_t key_id(const unsigned char key[PURE_SEAL_KEY]) { return fnv(2166136261u, key, PURE_SEAL_KEY); }
-
-static uint64_t random64(void) {
-    uint64_t value = 0;
-    while (getrandom(&value, sizeof(value), 0) != sizeof(value) || !value) {}
-    return value;
-}
-
-/* The key and nonce base of packets from `from` (in its run from_run) to
- * `to`, answering to_run (0 for none). */
-static bool derive(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
-                   const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
-                   unsigned char key[PURE_AEAD_KEY], unsigned char base[PURE_AEAD_NONCE]) {
-    unsigned char info[16 + 2 * PURE_SEAL_KEY + 16], out[PURE_AEAD_KEY + PURE_AEAD_NONCE];
-    memcpy(info, "pure link 0 keys", 16);
-    memcpy(info + 16, from, PURE_SEAL_KEY);
-    memcpy(info + 16 + PURE_SEAL_KEY, to, PURE_SEAL_KEY);
-    put64(info + 16 + 2 * PURE_SEAL_KEY, from_run);
-    put64(info + 24 + 2 * PURE_SEAL_KEY, to_run);
-    if (!pure_link_derive(secret, info, sizeof(info), out, sizeof(out))) return false;
-    memcpy(key, out, PURE_AEAD_KEY);
-    memcpy(base, out + PURE_AEAD_KEY, PURE_AEAD_NONCE);
-    explicit_bzero(out, sizeof(out));
-    return true;
-}
-
-/* The same as a context of its own, for a sender's one link. */
-static PureAead *aead_of(const unsigned char secret[32], const unsigned char from[PURE_SEAL_KEY],
-                         const unsigned char to[PURE_SEAL_KEY], uint64_t from_run, uint64_t to_run,
-                         unsigned char base[PURE_AEAD_NONCE]) {
-    unsigned char key[PURE_AEAD_KEY];
-    PureAead *aead = derive(secret, from, to, from_run, to_run, key, base) ? pure_aead(key) : NULL;
-    explicit_bzero(key, sizeof(key));
-    return aead;
-}
-
-static void nonce_of(unsigned char out[PURE_AEAD_NONCE], const unsigned char base[PURE_AEAD_NONCE], uint64_t pn) {
-    memcpy(out, base, PURE_AEAD_NONCE);
-    for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i));
-}
-
-/* A packet: the header, then body sealed (with key, or the context's own
- * when NULL); its size, or 0. */
-static size_t seal_packet(PureAead *aead, const unsigned char *key, const unsigned char base[PURE_AEAD_NONCE], uint32_t id,
-                          uint64_t run, uint64_t pn, const unsigned char *body, size_t size, unsigned char *out) {
-    unsigned char nonce[PURE_AEAD_NONCE];
-    out[0] = VERSION;
-    put32(out + 1, id);
-    put64(out + 5, run);
-    put64(out + 13, pn);
-    nonce_of(nonce, base, pn);
-    return pure_aead_seal(aead, key, nonce, out, HEADER, body, size, out + HEADER) ? HEADER + size + PURE_AEAD_TAG : 0;
-}
-
-static size_t put_names(unsigned char *p, const char *source, const char *input) {
-    size_t a = strlen(source), b = strlen(input);
-    p[0] = (unsigned char)a;
-    memcpy(p + 1, source, a);
-    p[1 + a] = (unsigned char)b;
-    memcpy(p + 2 + a, input, b);
-    return 2 + a + b;
-}
-
-/* A source and an input's names from p (size bytes); how many it takes, or 0. */
-static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LINK_NAME + 1],
-                        char input[PURE_LINK_NAME + 1]) {
-    if (size < 1 || !p[0] || p[0] > PURE_LINK_NAME || size < 2u + p[0]) return 0;
-    size_t a = p[0], b = p[1 + a];
-    if (!b || b > PURE_LINK_NAME || size < 2 + a + b) return 0;
-    memcpy(source, p + 1, a);
-    source[a] = 0;
-    memcpy(input, p + 2 + a, b);
-    input[b] = 0;
-    return 2 + a + b;
-}
-
 /* Datagrams taken from a socket at once (recvmmsg). */
 enum { TAKE = 64, SLOT = DATAGRAM + 64 };
 typedef struct {
@@ -221,63 +92,27 @@ static int take_datagrams(int fd, Taking *t) {
 
 /* ---- The sending side */
 
-enum { UNSENT, FLYING, RECEIVED };
-typedef struct { uint64_t pn; uint8_t state; } Piece;
-/* A record queued: its bytes at `at` in the link's byte ring, its pieces'
- * states (one kept here, more beside). */
-typedef struct { uint64_t seq; size_t at; uint32_t size, pieces; Piece *more, one; } Record;
-static Piece *piece_of(Record *r, uint32_t i) { return r->pieces == 1 ? &r->one : &r->more[i]; }
-typedef struct { uint64_t seq; uint32_t piece; } Unit;
-typedef struct {
-    uint64_t pn, sent; /* its number; when it went (ns) */
-    uint32_t bytes;
-    bool eliciting, done; /* acknowledged or lost */
-    uint16_t count;
-    Unit units[UNITS];
-} Flight;
-
 struct PureLink {
-    uint32_t shape, own_id, peer_id;
-    char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1];
-    const PureSealKey *own;
-    unsigned char own_public[PURE_SEAL_KEY], peer[PURE_SEAL_KEY], secret[32];
-    bool peered;
+    PureLinkSender *sender; /* under the lock */
+    char host[256], port[8], input[PURE_LINK_NAME + 1];
+    bool peered, keyed; /* given the receiver's key; the keys made from it */
     pthread_mutex_t lock;
-    Record *records; /* the sends not yet stored, oldest first (a ring of consecutive numbers) */
-    size_t head, count, capacity;
-    /* Their bytes, one after another as queued (a record's never cut by the
-     * ring's end): taken back in order as they are stored. */
-    unsigned char *ring;
-    size_t ring_capacity;
-    uint64_t queued, stored, told; /* the last queued; saved by the receiver; told the owner */
     bool stopping, started;
     /* The thread sleeps (in poll), and has been woken since; records were
      * queued since it last sent. A queue wakes it only when it sleeps, once. */
     bool sleeping, signaled, queued_new;
+    uint64_t told; /* the last stored the owner was told of */
     int fd, wake;
     pthread_t thread;
     void (*notify)(void *);
     void *context;
-    /* The thread's own, under the lock. */
-    uint64_t run, next_pn, in_run;
-    PureAead *out, *in;
-    unsigned char out_base[PURE_AEAD_NONCE], in_base[PURE_AEAD_NONCE];
-    Flight *flights; /* one for each number sent, oldest first (a ring) */
-    size_t flight_head, flight_count, flight_capacity;
-    uint64_t in_flight, window, threshold, recovery_start, eliciting;
-    uint64_t largest_acked, latest_rtt, smoothed_rtt, rtt_var, min_rtt;
-    bool acked_any, sampled;
-    uint64_t loss_time, last_eliciting, credit, stored_changed;
-    uint64_t received;    /* pieces received and not yet stored */
-    uint64_t unsent_from; /* no record before this number has a piece unsent */
+    /* The thread's own. */
     Taking *taking;       /* answers, received at once */
     unsigned char *burst; /* packets sealed and not yet sent (BURST datagrams) */
     size_t burst_size;
     unsigned burst_count;
     uint64_t burst_pn[BURST];
     bool gso;
-    unsigned pto_count, probes;
-    PureLinkCounts counts;
 };
 
 PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
@@ -300,41 +135,22 @@ PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, co
         snprintf(error, size, "out of memory");
         return NULL;
     }
-    l->shape = shape;
     memcpy(l->host, host, (size_t)(colon - host));
     memcpy(l->port, colon + 1, (size_t)(slash - colon - 1));
     strcpy(l->input, slash + 1);
-    strcpy(l->source, source);
-    l->own = own;
-    memcpy(l->own_public, own_public, PURE_SEAL_KEY);
-    l->own_id = key_id(own_public);
     l->fd = l->wake = -1;
+    if (!(l->sender = pure_link_sender(source, l->input, shape, own, own_public))) {
+        pthread_mutex_destroy(&l->lock);
+        free(l);
+        snprintf(error, size, "out of memory");
+        return NULL;
+    }
     return l;
 }
 
 void pure_link_peer(PureLink *l, const unsigned char peer[PURE_SEAL_KEY]) {
-    memcpy(l->peer, peer, PURE_SEAL_KEY);
-    l->peer_id = key_id(peer);
     l->peered = true;
-}
-
-static Record *record_at(PureLink *l, size_t k) { return &l->records[(l->head + k) % l->capacity]; }
-
-/* The queued record numbered seq, or NULL (stored already, or not queued). */
-static Record *record_of(PureLink *l, uint64_t seq) {
-    if (!l->count || seq < record_at(l, 0)->seq || seq - record_at(l, 0)->seq >= l->count) return NULL;
-    return record_at(l, (size_t)(seq - record_at(l, 0)->seq));
-}
-
-/* Every change of a piece's state goes through here: the link knows at once
- * how many pieces wait to be stored, and where the unsent ones begin. */
-static void set_piece(PureLink *l, Record *r, uint32_t i, uint8_t state) {
-    uint8_t was = (*piece_of(r, i)).state;
-    if (was == state) return;
-    if (was == RECEIVED) --l->received;
-    if (state == RECEIVED) ++l->received;
-    if (state == UNSENT && r->seq < l->unsent_from) l->unsent_from = r->seq;
-    (*piece_of(r, i)).state = state;
+    l->keyed = pure_link_sender_peer(l->sender, peer);
 }
 
 static void wake(int fd) {
@@ -342,372 +158,39 @@ static void wake(int fd) {
     if (fd >= 0 && write(fd, &one, sizeof(one)) != sizeof(one)) { /* full: it is readable already */ }
 }
 
-/* Where size bytes go at the end of the byte ring (after the newest record,
- * or from its start when that is free and the end is not), or SIZE_MAX. */
-static size_t ring_place(PureLink *l, size_t size) {
-    if (!l->count) return size <= l->ring_capacity ? 0 : SIZE_MAX;
-    Record *oldest = record_at(l, 0), *newest = record_at(l, l->count - 1);
-    size_t head = oldest->at, tail = newest->at + newest->size;
-    if (newest->at >= head) /* the bytes are [head, tail): after them, or before them from the start */
-        return l->ring_capacity - tail >= size ? tail : size <= head ? 0 : SIZE_MAX;
-    return head - tail >= size ? tail : SIZE_MAX; /* around the end: between the newest and the oldest */
-}
-
-/* A ring room for size bytes more: twice as large (or more), the records
- * moved into it one after another. */
-static bool ring_grow(PureLink *l, size_t size) {
-    size_t used = 0;
-    for (size_t k = 0; k < l->count; ++k) used += record_at(l, k)->size;
-    size_t capacity = l->ring_capacity ? 2 * l->ring_capacity : 64 << 10;
-    while (capacity < used + size + 1) capacity *= 2;
-    unsigned char *ring = malloc(capacity);
-    if (!ring) return false;
-    for (size_t k = 0, at = 0; k < l->count; ++k) {
-        Record *r = record_at(l, k);
-        if (r->size) memcpy(ring + at, l->ring + r->at, r->size);
-        r->at = at;
-        at += r->size;
-    }
-    free(l->ring);
-    l->ring = ring;
-    l->ring_capacity = capacity;
-    return true;
-}
-
 bool pure_link_queue(PureLink *l, uint64_t seq, const unsigned char *bytes, size_t size) {
-    Record r = {.seq = seq, .size = (uint32_t)size};
-    r.pieces = size ? (uint32_t)((size + PIECE - 1) / PIECE) : 1;
-    if (size > RECORD_MOST || (r.pieces > 1 && !(r.more = calloc(r.pieces, sizeof(Piece))))) return false;
     pthread_mutex_lock(&l->lock);
-    bool ok = seq > l->queued && (!l->count || seq == l->queued + 1);
-    size_t at = ok ? ring_place(l, size) : SIZE_MAX;
-    if (ok && at == SIZE_MAX) ok = ring_grow(l, size) && (at = ring_place(l, size)) != SIZE_MAX;
-    if (ok) {
-        r.at = at;
-        if (size) memcpy(l->ring + at, bytes, size);
-    }
-    if (ok && l->count == l->capacity) {
-        size_t capacity = l->capacity ? 2 * l->capacity : 64;
-        Record *grown = malloc(capacity * sizeof(*grown));
-        if (grown) {
-            for (size_t k = 0; k < l->count; ++k) grown[k] = *record_at(l, k);
-            free(l->records);
-            l->records = grown;
-            l->capacity = capacity;
-            l->head = 0;
-        } else
-            ok = false;
-    }
-    bool signal = false;
+    bool ok = pure_link_sender_queue(l->sender, seq, bytes, size), signal = false;
     if (ok) {
-        *record_at(l, l->count++) = r;
-        l->queued = seq;
         l->queued_new = true;
         signal = l->sleeping && !l->signaled;
         l->signaled = l->signaled || signal;
     }
     pthread_mutex_unlock(&l->lock);
-    if (!ok) {
-        free(r.more);
-        return false;
-    }
     if (signal) wake(l->wake);
-    return true;
+    return ok;
 }
 
 uint64_t pure_link_queued(PureLink *l) {
     pthread_mutex_lock(&l->lock);
-    uint64_t queued = l->queued;
+    uint64_t queued = pure_link_sender_queued(l->sender);
     pthread_mutex_unlock(&l->lock);
     return queued;
 }
 
 uint64_t pure_link_delivered(PureLink *l) {
     pthread_mutex_lock(&l->lock);
-    uint64_t stored = l->stored;
+    uint64_t stored = pure_link_sender_stored(l->sender);
     pthread_mutex_unlock(&l->lock);
     return stored;
 }
 
-static Flight *flight_of(PureLink *l, uint64_t pn) {
-    if (!l->flight_count) return NULL;
-    uint64_t first = l->flights[l->flight_head].pn;
-    if (pn < first || pn - first >= l->flight_count) return NULL;
-    return &l->flights[(l->flight_head + (size_t)(pn - first)) % l->flight_capacity];
-}
-
-static bool push_flight(PureLink *l, const Flight *f) {
-    if (l->flight_count == l->flight_capacity) {
-        size_t capacity = l->flight_capacity ? 2 * l->flight_capacity : 256;
-        Flight *grown = malloc(capacity * sizeof(*grown));
-        if (!grown) return false;
-        for (size_t k = 0; k < l->flight_count; ++k) grown[k] = l->flights[(l->flight_head + k) % l->flight_capacity];
-        free(l->flights);
-        l->flights = grown;
-        l->flight_capacity = capacity;
-        l->flight_head = 0;
-    }
-    l->flights[(l->flight_head + l->flight_count++) % l->flight_capacity] = *f;
-    return true;
-}
-
-static void settle(PureLink *l, Flight *f) {
-    f->done = true;
-    if (f->eliciting) {
-        --l->eliciting;
-        l->in_flight -= f->bytes;
-    }
-}
-
-/* A packet lost: what it carried and was not received otherwise goes again. */
-static void lose(PureLink *l, Flight *f, uint64_t now) {
-    settle(l, f);
-    ++l->counts.lost;
-    for (uint16_t u = 0; u < f->count; ++u) {
-        Record *r = record_of(l, f->units[u].seq);
-        Piece *p = r ? &(*piece_of(r, f->units[u].piece)) : NULL;
-        if (p && p->state == FLYING && p->pn == f->pn) {
-            set_piece(l, r, f->units[u].piece, UNSENT);
-            ++l->counts.resent;
-        }
-    }
-    if (f->sent > l->recovery_start) { /* one congestion event per round trip */
-        l->recovery_start = now;
-        l->window = most(l->window / 2, WINDOW_LEAST);
-        l->threshold = l->window;
-    }
-}
-
-static void detect_lost(PureLink *l, uint64_t now) {
-    l->loss_time = 0;
-    if (!l->acked_any) return;
-    uint64_t rtt = most(l->latest_rtt, l->sampled ? l->smoothed_rtt : FIRST_RTT);
-    uint64_t delay = most(rtt * 9 / 8, GRANULARITY);
-    for (size_t k = 0; k < l->flight_count; ++k) {
-        Flight *f = &l->flights[(l->flight_head + k) % l->flight_capacity];
-        if (f->pn >= l->largest_acked) break;
-        if (f->done) continue;
-        if (f->sent + delay <= now || l->largest_acked >= f->pn + 3) lose(l, f, now);
-        else if (!l->loss_time || f->sent + delay < l->loss_time) l->loss_time = f->sent + delay;
-    }
-}
-
-static void sample_rtt(PureLink *l, uint64_t latest, uint64_t ack_delay) {
-    l->latest_rtt = latest;
-    if (!l->sampled) {
-        l->min_rtt = l->smoothed_rtt = latest;
-        l->rtt_var = latest / 2;
-        l->sampled = true;
-        return;
-    }
-    l->min_rtt = least(l->min_rtt, latest);
-    ack_delay = least(ack_delay, ACK_DELAY_MOST);
-    uint64_t adjusted = latest >= l->min_rtt + ack_delay ? latest - ack_delay : latest;
-    uint64_t difference = l->smoothed_rtt > adjusted ? l->smoothed_rtt - adjusted : adjusted - l->smoothed_rtt;
-    l->rtt_var = (3 * l->rtt_var + difference) / 4;
-    l->smoothed_rtt = (7 * l->smoothed_rtt + adjusted) / 8;
-}
-
-static void on_ack(PureLink *l, const unsigned char *p, size_t size, uint64_t now) {
-    if (size < 13 || size < 13 + 16u * p[12]) return;
-    uint64_t largest = get64(p), delay = (uint64_t)get32(p + 8) * 1000;
-    Flight *top = NULL;
-    for (unsigned k = 0; k < p[12]; ++k) {
-        uint64_t first = get64(p + 13 + 16 * k), last = get64(p + 21 + 16 * k);
-        if (first > last || !l->flight_count) continue;
-        uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1;
-        for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) {
-            Flight *f = flight_of(l, pn);
-            if (!f || f->done) continue;
-            settle(l, f);
-            ++l->counts.acknowledged;
-            for (uint16_t u = 0; u < f->count; ++u) {
-                Record *r = record_of(l, f->units[u].seq);
-                if (r && (*piece_of(r, f->units[u].piece)).state != RECEIVED) {
-                    set_piece(l, r, f->units[u].piece, RECEIVED);
-                    l->stored_changed = now; /* waits for its save from now */
-                }
-            }
-            if (f->eliciting && f->sent > l->recovery_start)
-                l->window += l->window < l->threshold ? f->bytes : (uint64_t)DATAGRAM * f->bytes / l->window;
-            if (pn == largest) top = f;
-        }
-    }
-    if (!l->acked_any || largest > l->largest_acked) l->largest_acked = largest;
-    l->acked_any = true;
-    if (top && top->eliciting) sample_rtt(l, now - top->sent, delay);
-    l->pto_count = 0;
-    detect_lost(l, now);
-    while (l->flight_count && l->flights[l->flight_head].done) {
-        l->flight_head = (l->flight_head + 1) % l->flight_capacity;
-        --l->flight_count;
-    }
-}
-
-static void on_stored(PureLink *l, const unsigned char *p, size_t size, uint64_t now) {
-    char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
-    size_t names = get_names(p, size, source, input);
-    if (!names || size < names + 13 || strcmp(source, l->source) || strcmp(input, l->input)) return;
-    uint64_t heard = get64(p + names);
-    unsigned flags = p[names + 8];
-    uint32_t credit = get32(p + names + 9);
-    if (heard > l->stored) {
-        while (l->count && record_at(l, 0)->seq <= heard) {
-            Record *r = record_at(l, 0);
-            for (uint32_t i = 0; i < r->pieces; ++i) l->received -= (*piece_of(r, i)).state == RECEIVED;
-            free(r->more); /* its bytes go back with the ring's head */
-            l->head = (l->head + 1) % l->capacity;
-            --l->count;
-        }
-        l->stored = heard;
-        l->stored_changed = now;
-    }
-    if (flags & STORED_GAP) /* it did not keep what came after a gap: send it again */
-        for (size_t k = 0; k < l->count && l->received; ++k) {
-            Record *r = record_at(l, k);
-            for (uint32_t i = 0; i < r->pieces; ++i)
-                if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT);
-        }
-    l->credit = credit ? least(credit, 65536) : 1;
-}
-
-static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now);
-
 static void take_answers(PureLink *l, uint64_t now) {
     for (unsigned rounds = 0, n = TAKE; rounds < 4 && n == TAKE; ++rounds) {
         n = (unsigned)take_datagrams(l->fd, l->taking);
-        for (unsigned k = 0; k < n; ++k) take_answer(l, l->taking->bytes[k], l->taking->messages[k].msg_len, now);
-    }
-}
-
-static void take_answer(PureLink *l, unsigned char *packet, ssize_t n, uint64_t now) {
-    unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
-    if (n < HEADER + PURE_AEAD_TAG || (size_t)n > DATAGRAM || packet[0] != VERSION || get32(packet + 1) != l->peer_id)
-        return; /* not ours, or an error from an earlier send (nobody listening) */
-    uint64_t run = get64(packet + 5), pn = get64(packet + 13);
-    if (!l->in || l->in_run != run) {
-        pure_aead_free(l->in);
-        l->in = aead_of(l->secret, l->peer, l->own_public, run, l->run, l->in_base);
-        l->in_run = run;
-        if (!l->in) return;
-    }
-    nonce_of(nonce, l->in_base, pn);
-    size_t size = (size_t)n - HEADER - PURE_AEAD_TAG;
-    if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, (size_t)n - HEADER, plain)) return;
-    ++l->counts.received;
-    for (size_t at = 0; at < size;) {
-        unsigned type = plain[at++];
-        if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) {
-            on_ack(l, plain + at, size - at, now);
-            at += 13 + 16u * plain[at + 12];
-        } else if (type == F_STORED) {
-            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
-            size_t names = get_names(plain + at, size - at, source, input);
-            if (!names || at + names + 13 > size) break;
-            on_stored(l, plain + at, size - at, now);
-            at += names + 13;
-        } else if (type != F_PING)
-            break; /* padding, or what this side does not know: the rest */
-    }
-}
-
-/* A packet of pieces not sent yet, in order, within the receiver's credit:
- * its body's size, or 0 when there is none. */
-static size_t fill(PureLink *l, unsigned char *body, Flight *f) {
-    size_t at = 0;
-    body[at++] = F_STREAM;
-    at += put_names(body + at, l->source, l->input);
-    put32(body + at, l->shape);
-    at += 4;
-    /* The credit counts from the oldest record not stored: after a restart
-     * the sender does not know yet how far the receiver stored. */
-    uint64_t base = l->count ? record_at(l, 0)->seq - 1 : l->stored;
-    /* From the first record that may have an unsent piece: the ones before
-     * it are sent, and every one found sent moves that mark on. */
-    size_t k = l->count && l->unsent_from > base + 1 ? (size_t)(l->unsent_from - base - 1) : 0;
-    bool before = true;
-    /* A record none of whose pieces went yet goes whole when it fits what is
-     * left of the packet: whole records whose numbers follow on share a
-     * RECORDS frame (its count at run). Else its unsent pieces go, each in a
-     * RECORD frame. */
-    size_t run = 0;
-    uint64_t next = 0;
-    for (; k < l->count && f->count < UNITS; ++k) {
-        Record *r = record_at(l, k);
-        if (r->seq > base + l->credit) break;
-        bool unsent = false, fresh = true, joins = run && r->seq == next;
-        for (uint32_t i = 0; i < r->pieces && fresh; ++i) fresh = piece_of(r, i)->state == UNSENT;
-        size_t whole = (joins ? 0 : RECORDS_HEAD) + varint_size(r->size) + r->size;
-        if (fresh && f->count + r->pieces <= UNITS && at + whole <= BODY) {
-            if (!joins) {
-                body[at] = F_RECORDS;
-                put64(body + at + 1, r->seq);
-                body[at + 9] = 0;
-                run = at + 9;
-                at += RECORDS_HEAD;
-            }
-            at += put_varint(body + at, r->size);
-            if (r->size) memcpy(body + at, l->ring + r->at, r->size);
-            at += r->size;
-            ++body[run];
-            next = r->seq + 1;
-            for (uint32_t i = 0; i < r->pieces; ++i) {
-                set_piece(l, r, i, FLYING);
-                piece_of(r, i)->pn = f->pn;
-                f->units[f->count++] = (Unit){r->seq, i};
-            }
-            ++l->counts.records;
-        } else if (fresh && r->pieces == 1)
-            unsent = true; /* too large for what is left: the next packet */
-        else
-            for (uint32_t i = 0; i < r->pieces; ++i) {
-                if (piece_of(r, i)->state != UNSENT) continue;
-                uint32_t offset = i * PIECE, length = (uint32_t)least(PIECE, r->size - offset);
-                if (f->count == UNITS || at + RECORD_HEAD + length > BODY) {
-                    unsent = true;
-                    break;
-                }
-                body[at] = F_RECORD;
-                put64(body + at + 1, r->seq);
-                put32(body + at + 9, r->size);
-                put32(body + at + 13, offset);
-                put16(body + at + 17, length);
-                if (length) memcpy(body + at + RECORD_HEAD, l->ring + r->at + offset, length);
-                at += RECORD_HEAD + length;
-                run = 0;
-                set_piece(l, r, i, FLYING);
-                piece_of(r, i)->pn = f->pn;
-                f->units[f->count++] = (Unit){r->seq, i};
-                ++l->counts.records;
-            }
-        if (before && !unsent) l->unsent_from = r->seq + 1;
-        before = before && !unsent;
-        if (unsent) { /* the packet is full: whole, so it can go in a burst */
-            if (l->gso && f->count) {
-                memset(body + at, F_PADDING, BODY - at);
-                at = BODY;
-            }
-            break;
-        }
-    }
-    return f->count ? at : 0;
-}
-
-/* Seal and send f with body; its number is spent even when the socket
- * refuses it (what it carried is then marked unsent again). */
-/* A packet the socket refused: what it carried is unsent again, and it
- * counts as never sent. Its number is spent all the same. */
-static void refused(PureLink *l, uint64_t pn) {
-    Flight *f = flight_of(l, pn);
-    if (!f || f->done) return;
-    for (uint16_t u = 0; u < f->count; ++u) {
-        Record *r = record_of(l, f->units[u].seq);
-        if (r && (*piece_of(r, f->units[u].piece)).pn == pn) set_piece(l, r, f->units[u].piece, UNSENT);
+        for (unsigned k = 0; k < n; ++k)
+            pure_link_sender_take(l->sender, l->taking->bytes[k], l->taking->messages[k].msg_len, now);
     }
-    settle(l, f);
-    f->eliciting = false;
-    f->count = 0;
-    --l->counts.packets;
 }
 
 static bool transient(void) { return errno == EAGAIN || errno == EWOULDBLOCK || errno == ENOBUFS || errno == EINTR; }
@@ -724,119 +207,32 @@ static void flush(PureLink *l) {
         if (send(l->fd, l->burst + at, size, 0) < 0 && transient()) break;
         at += size;
     }
-    for (unsigned k = sent; k < n; ++k) refused(l, l->burst_pn[k]);
+    for (unsigned k = sent; k < n; ++k) pure_link_sender_refused(l->sender, l->burst_pn[k]);
     l->burst_count = 0;
     l->burst_size = 0;
 }
 
-/* Seal f with body into the burst (a whole datagram each but the last), and
- * count it sent; its number is spent even when the socket then refuses it. */
-static bool transmit(PureLink *l, Flight *f, const unsigned char *body, size_t size, uint64_t now) {
-    if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */
-    size_t total = seal_packet(l->out, NULL, l->out_base, l->own_id, l->run, f->pn, body, size, l->burst + l->burst_size);
-    ++l->next_pn;
-    f->sent = now;
-    f->bytes = (uint32_t)total;
-    if (!total) { /* not sealed: what it carried is unsent again */
-        for (uint16_t u = 0; u < f->count; ++u) {
-            Record *r = record_of(l, f->units[u].seq);
-            if (r && (*piece_of(r, f->units[u].piece)).pn == f->pn) set_piece(l, r, f->units[u].piece, UNSENT);
-        }
-        f->done = true;
-        f->eliciting = false;
-        f->count = 0;
-        push_flight(l, f);
-        return false;
-    }
-    if (!push_flight(l, f)) return false;
-    l->burst_pn[l->burst_count++] = f->pn;
-    l->burst_size += total;
-    ++l->counts.packets;
-    if (f->eliciting) {
-        ++l->eliciting;
-        l->in_flight += f->bytes;
-        l->last_eliciting = now;
-    }
-    if (l->burst_count == BURST) flush(l);
-    return true;
-}
-
+/* Packets while the sender has them (up to 128 a turn), into bursts: a
+ * short packet ends one. */
 static void send_packets(PureLink *l, uint64_t now) {
-    unsigned char body[BODY];
     l->queued_new = false; /* all queued is looked at now */
     for (unsigned n = 0; n < 128; ++n) {
-        bool probe = l->probes > 0;
-        if (!probe && l->in_flight + DATAGRAM > l->window) break;
-        Flight f = {.pn = l->next_pn, .eliciting = true};
-        size_t size = fill(l, body, &f);
-        if (!size) {
-            if (!probe) break;
-            body[0] = F_PING; /* nothing to send again: a probe all the same */
-            size = 1;
-        }
-        if (!transmit(l, &f, body, size, now)) break;
-        if (probe) --l->probes;
+        if (l->burst_count && l->burst_size != (size_t)l->burst_count * DATAGRAM) flush(l); /* the last was short */
+        uint64_t pn;
+        size_t size = pure_link_sender_packet(l->sender, l->burst + l->burst_size, l->gso, now, &pn);
+        if (!size) break;
+        l->burst_pn[l->burst_count++] = pn;
+        l->burst_size += size;
+        if (l->burst_count == BURST) flush(l);
     }
     flush(l);
 }
 
-static uint64_t pto_period(const PureLink *l) {
-    uint64_t smoothed = l->sampled ? l->smoothed_rtt : FIRST_RTT, var = l->sampled ? l->rtt_var : FIRST_RTT / 2;
-    uint64_t period = smoothed + most(4 * var, GRANULARITY) + ACK_DELAY_MOST;
-    for (unsigned i = 0; i < l->pto_count && period < PTO_MOST; ++i) period *= 2;
-    return least(period, PTO_MOST);
-}
-
-static bool waiting_for_save(PureLink *l) { return l->received > 0; }
-
-/* The next time a timer is due (ns), or 0. */
-static uint64_t next_timer(PureLink *l) {
-    uint64_t due = l->loss_time;
-    if (l->eliciting) {
-        uint64_t pto = l->last_eliciting + pto_period(l);
-        if (!due || pto < due) due = pto;
-    }
-    if (waiting_for_save(l)) {
-        uint64_t stored = l->stored_changed + STORED_WAIT;
-        if (!due || stored < due) due = stored;
-    }
-    return due;
-}
-
-static void run_timers(PureLink *l, uint64_t now) {
-    if (l->loss_time && l->loss_time <= now) detect_lost(l, now);
-    else if (l->eliciting && l->last_eliciting + pto_period(l) <= now) {
-        /* Nothing heard for a while: a probe, what flies longest sent again. */
-        ++l->pto_count;
-        l->probes = 1;
-        for (size_t k = 0; k < l->count; ++k) {
-            Record *r = record_at(l, k);
-            uint32_t i = 0;
-            while (i < r->pieces && (*piece_of(r, i)).state != FLYING) ++i;
-            if (i < r->pieces) {
-                set_piece(l, r, i, UNSENT);
-                ++l->counts.resent;
-                break;
-            }
-        }
-    }
-    if (waiting_for_save(l) && l->stored_changed + STORED_WAIT <= now) {
-        /* Received long ago and still not stored: the receiver lost them (it
-         * restarted) or its word got lost. Sent again: it takes a number once. */
-        for (size_t k = 0; k < l->count && l->received; ++k) {
-            Record *r = record_at(l, k);
-            for (uint32_t i = 0; i < r->pieces; ++i)
-                if ((*piece_of(r, i)).state == RECEIVED) set_piece(l, r, i, UNSENT), ++l->counts.resent;
-        }
-        l->stored_changed = now;
-    }
-}
-
 static void *run_link(void *context) {
     PureLink *l = context;
     for (;;) {
         pthread_mutex_lock(&l->lock);
-        uint64_t now = now_ns(), due = next_timer(l);
+        uint64_t now = now_ns(), due = pure_link_sender_due(l->sender);
         bool stopping = l->stopping, more = l->queued_new; /* queued while it sent: no sleep */
         l->sleeping = !more;
         pthread_mutex_unlock(&l->lock);
@@ -852,10 +248,11 @@ static void *run_link(void *context) {
         l->sleeping = l->signaled = false;
         now = now_ns();
         if (waits[0].revents) take_answers(l, now);
-        run_timers(l, now);
+        pure_link_sender_tick(l->sender, now);
         send_packets(l, now);
-        bool told = l->stored != l->told;
-        l->told = l->stored;
+        uint64_t stored = pure_link_sender_stored(l->sender);
+        bool told = stored != l->told;
+        l->told = stored;
         pthread_mutex_unlock(&l->lock);
         if (told && l->notify) l->notify(l->context);
     }
@@ -883,13 +280,7 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e
     }
     int segment = DATAGRAM;
     l->gso = ok && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment));
-    ok = ok && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking)));
-    l->run = random64();
-    l->window = WINDOW_FIRST;
-    l->threshold = UINT64_MAX;
-    l->credit = CREDIT;
-    ok = ok && pure_link_secret(l->own, l->peer, l->secret) &&
-         (l->out = aead_of(l->secret, l->own_public, l->peer, l->run, 0, l->out_base)) &&
+    ok = ok && l->keyed && (l->burst = malloc((size_t)BURST * DATAGRAM)) && (l->taking = malloc(sizeof(*l->taking))) &&
          (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0;
     l->notify = notify;
     l->context = context;
@@ -904,8 +295,7 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e
 
 void pure_link_counts(PureLink *l, PureLinkCounts *out) {
     pthread_mutex_lock(&l->lock);
-    *out = l->counts;
-    out->stored = l->stored;
+    pure_link_sender_counts(l->sender, out);
     pthread_mutex_unlock(&l->lock);
 }
 
@@ -918,803 +308,46 @@ void pure_link_close(PureLink *l) {
         wake(l->wake);
         pthread_join(l->thread, NULL);
     }
-    for (size_t k = 0; k < l->count; ++k) {
-        free(record_at(l, k)->more);
-    }
-    free(l->records);
-    free(l->ring);
-    free(l->flights);
+    pure_link_sender_free(l->sender);
     free(l->burst);
     free(l->taking);
-    pure_aead_free(l->out);
-    pure_aead_free(l->in);
-    explicit_bzero(l->secret, sizeof(l->secret));
     if (l->fd >= 0) close(l->fd);
     if (l->wake >= 0) close(l->wake);
     pthread_mutex_destroy(&l->lock);
     free(l);
 }
 
-/* ---- The receiving side
- *
- * A sender allowed is a peer: its key and the sources it may send as, a few
- * bytes each, found by its key's id. What a sending peer needs besides (the
- * secret its key shares with the node's, its runs' keys and addresses, its
- * streams) is its active state: made at its first packet (one X25519), let
- * go once the peer has been quiet for the idle period with nothing of it
- * waiting here. So a link at rest costs the node its allowing, and the next
- * packet goes on without a handshake. */
-
-typedef struct { uint64_t first, last; } Range;
-typedef struct Active Active;
-typedef struct {
-    Active *active;
-    uint64_t run, seen; /* the sender's run; when it last sent */
-    bool used, ack_due, due;
-    unsigned char in_key[PURE_AEAD_KEY], out_key[PURE_AEAD_KEY], in_base[PURE_AEAD_NONCE], out_base[PURE_AEAD_NONCE];
-    uint64_t next_pn, largest_at;
-    struct sockaddr_in6 address; /* or an IPv4 one */
-    socklen_t address_size;
-    Range got[RANGES]; /* the numbers received, newest first */
-    unsigned got_count;
-} Run;
-typedef struct {
-    uint64_t seq;
-    unsigned char *bytes; /* its size, then a byte a piece: had */
-    uint32_t size, pieces, got;
-} Held;
-typedef struct Stream {
-    Active *active;
-    char source[PURE_LINK_NAME + 1], input_name[PURE_LINK_NAME + 1];
-    uint32_t input, shape; /* the host's number for the input; the messages' shape */
-    uint64_t heard;              /* the last number heard (stored), once known */
-    bool known, out, stored_due, gap, ready;
-    /* The line: once the last heard is known, whole records from the one
-     * after it on, in order, as a delivery carries them (u32 size, bytes),
-     * from line_head: handed over as they are, let go by moving the head.
-     * Records that come early or in pieces, or before the last heard is
-     * known, wait in held until the line reaches them. */
-    unsigned char *line;
-    size_t line_head, line_size, line_capacity;
-    uint64_t line_first;
-    unsigned line_count;
-    Held *held;                  /* records not yet stored, by number: held_count from held_first */
-    size_t held_first, held_count, held_capacity, held_bytes;
-    Run *run;                    /* the run that sent to it last: where STORED goes */
-    uint64_t run_id;
-} Stream;
-struct Active {
-    uint32_t peer;               /* its place among the peers */
-    unsigned char secret[32];
-    uint64_t seen;
-    size_t place;                /* among the active */
-    Run runs[RUNS];
-    Stream **streams;
-    unsigned stream_count, stream_capacity;
-};
-typedef struct {
-    unsigned char key[PURE_SEAL_KEY];
-    uint32_t id, allowed; /* allowed: the place + 1 of its first source (0: none) */
-    Active *active;
-} Peer;
-typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 length, bytes); next: a place + 1 */
-typedef struct { Stream *stream; uint64_t heard; bool stored, gap; } Done;
-
-/* A peer quiet this long lets its active state go; the active are looked at
- * that often; what they held goes back to the system once TRIM_AFTER have
- * gone, or TRIM_WAIT after any went. */
-static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS;
-enum { TRIM_AFTER = 1024 };
+/* ---- The receiving side */
 
 struct PureLinkListener {
-    PureLinkHost host;
-    const PureSealKey *own;
-    unsigned char own_public[PURE_SEAL_KEY];
-    uint32_t own_id;
+    PureLinkReceiver *receiver; /* the thread's */
     unsigned port;
-    uint64_t run, idle, swept, trimmed, untrimmed;
-    PureAead *aead; /* keyed by each packet's run */
-    unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */
-    bool open_keyed, seal_keyed;
-    unsigned char *memo; /* the secrets remembered (pure_link_remember), mapped */
-    size_t memo_size;
     int fd, wake;
     pthread_t thread;
     bool started;
     pthread_mutex_t lock; /* the done list and stopping */
-    Done *done;
+    PureLinkDone *done;
     size_t done_count, done_capacity;
     bool stopping;
-    Peer *peers;
-    size_t peer_count, peer_capacity;
-    uint32_t *index; /* the peers by key id: open addressing, a place + 1 (0: empty) */
-    size_t index_size;
-    Allowed *allowed;
-    size_t allowed_count, allowed_capacity;
-    unsigned char *names;
-    size_t names_size, names_capacity;
-    Active **actives;
-    size_t active_count, active_capacity;
-    Stream **ready; /* streams that may have whole records to hand over */
-    size_t ready_count, ready_capacity;
-    Run **due; /* runs to answer */
-    size_t due_count, due_capacity;
-    unsigned char *batch;
-    size_t batch_capacity;
-    PureLinkCounts counts;
 };
 
-PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
-                                   const PureLinkHost *host, char *error, size_t size) {
-    PureLinkListener *s = own ? calloc(1, sizeof(*s)) : NULL;
+PureLinkListener *pure_link_listen(PureLinkReceiver *r, unsigned port, char *error, size_t size) {
+    PureLinkListener *s = calloc(1, sizeof(*s));
     if (!s || pthread_mutex_init(&s->lock, NULL)) {
         free(s);
-        snprintf(error, size, own ? "out of memory" : "--link-port needs the node's --key");
+        snprintf(error, size, "out of memory");
         return NULL;
     }
-    s->host = *host;
-    s->own = own;
-    memcpy(s->own_public, own_public, PURE_SEAL_KEY);
-    s->own_id = key_id(own_public);
+    s->receiver = r;
     s->port = port;
-    s->idle = IDLE_FIRST;
     s->fd = s->wake = -1;
     return s;
 }
 
-void pure_link_idle(PureLinkListener *s, uint64_t ms) { s->idle = ms * MS; }
-
-/* items (count used of capacity) with room for one more: the same or grown,
- * or NULL when memory runs out (items stay as they were). */
-static void *room(void *items, size_t *capacity, size_t count, size_t item_size, size_t first) {
-    if (count < *capacity) return items;
-    size_t more = *capacity ? 2 * *capacity : first;
-    void *grown = realloc(items, more * item_size);
-    if (grown) *capacity = more;
-    return grown;
-}
-
-/* The peers by key id again, in an index twice as large. */
-static bool grow_index(PureLinkListener *s) {
-    size_t size = s->index_size ? 2 * s->index_size : 1024;
-    uint32_t *index = calloc(size, sizeof(*index));
-    if (!index) return false;
-    for (size_t p = 0; p < s->peer_count; ++p) {
-        size_t slot = s->peers[p].id & (size - 1);
-        while (index[slot]) slot = (slot + 1) & (size - 1);
-        index[slot] = (uint32_t)p + 1;
-    }
-    free(s->index);
-    s->index = index;
-    s->index_size = size;
-    return true;
-}
-
-static bool same_name(const PureLinkListener *s, uint32_t name, const char *source, size_t length) {
-    return s->names[name] == length && !memcmp(s->names + name + 1, source, length);
-}
-
-/* Keys are found by their ids, which are 32 bits: two keys may share one. */
-bool pure_link_allow(PureLinkListener *s, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
-                     size_t size) {
-    size_t length = strlen(source), p = SIZE_MAX;
-    uint32_t id = key_id(key);
-    if (!length || length > PURE_LINK_NAME) {
-        snprintf(error, size, "a source's name is 1 to %d bytes", PURE_LINK_NAME);
-        return false;
-    }
-    for (size_t slot = s->index_size ? id & (s->index_size - 1) : 0; s->index_size && s->index[slot];
-         slot = (slot + 1) & (s->index_size - 1)) {
-        const Peer *peer = &s->peers[s->index[slot] - 1];
-        if (peer->id == id && !memcmp(peer->key, key, PURE_SEAL_KEY)) {
-            p = s->index[slot] - 1;
-            break;
-        }
-    }
-    if (p == SIZE_MAX) {
-        Peer *peers = s->peer_count < UINT32_MAX - 1 && ((s->peer_count + 1) * 2 <= s->index_size || grow_index(s)) ?
-                      room(s->peers, &s->peer_capacity, s->peer_count, sizeof(Peer), 64) : NULL;
-        if (!peers) goto full;
-        s->peers = peers;
-        p = s->peer_count++;
-        s->peers[p] = (Peer){.id = id};
-        memcpy(s->peers[p].key, key, PURE_SEAL_KEY);
-        size_t slot = id & (s->index_size - 1);
-        while (s->index[slot]) slot = (slot + 1) & (s->index_size - 1);
-        s->index[slot] = (uint32_t)p + 1;
-    }
-    for (uint32_t a = s->peers[p].allowed; a; a = s->allowed[a - 1].next)
-        if (same_name(s, s->allowed[a - 1].name, source, length)) return true;
-    Allowed *kept = s->allowed_count < UINT32_MAX - 1 && s->names_size + 1 + length <= UINT32_MAX ?
-                    room(s->allowed, &s->allowed_capacity, s->allowed_count, sizeof(Allowed), 64) : NULL;
-    if (!kept) goto full;
-    s->allowed = kept;
-    if (s->names_size + 1 + length > s->names_capacity) {
-        size_t more = s->names_capacity ? 2 * s->names_capacity : 4096;
-        unsigned char *names = realloc(s->names, more);
-        if (!names) goto full;
-        s->names = names;
-        s->names_capacity = more;
-    }
-    s->names[s->names_size] = (unsigned char)length;
-    memcpy(s->names + s->names_size + 1, source, length);
-    s->allowed[s->allowed_count] = (Allowed){(uint32_t)s->names_size, s->peers[p].allowed};
-    s->peers[p].allowed = (uint32_t)++s->allowed_count;
-    s->names_size += 1 + length;
-    return true;
-full:
-    snprintf(error, size, "out of memory");
-    return false;
-}
-
-static Held *held_at(Stream *st, size_t k) { return &st->held[st->held_first + k]; }
-
-static uint64_t line_end(const Stream *st) { return st->line_first + st->line_count; }
-
-/* Room for more bytes at the line's end: its front let go given back
- * first when that is at least half of it (so no byte moves twice in a
- * doubling), else a longer line. */
-static bool line_room(Stream *st, size_t more) {
-    if (st->line_size + more <= st->line_capacity) return true;
-    if (st->line_head && st->line_head >= st->line_capacity / 2) {
-        memmove(st->line, st->line + st->line_head, st->line_size - st->line_head);
-        st->line_size -= st->line_head;
-        st->line_head = 0;
-        if (st->line_size + more <= st->line_capacity) return true;
-    }
-    size_t capacity = st->line_capacity ? st->line_capacity : 4096;
-    while (capacity < st->line_size + more) capacity *= 2;
-    unsigned char *grown = realloc(st->line, capacity);
-    if (!grown) return false;
-    st->line = grown;
-    st->line_capacity = capacity;
-    return true;
-}
-
-static bool line_append(Stream *st, const unsigned char *bytes, uint32_t size) {
-    if (!line_room(st, 4 + (size_t)size)) return false;
-    put32(st->line + st->line_size, size);
-    if (size) memcpy(st->line + st->line_size + 4, bytes, size);
-    st->line_size += 4 + (size_t)size;
-    ++st->line_count;
-    return true;
-}
-
-/* The line's records up to heard let go (all of them, when heard is past it). */
-static void line_drop(Stream *st, uint64_t heard) {
-    if (heard < st->line_first) return;
-    if (heard >= line_end(st)) {
-        st->line_head = st->line_size = 0;
-        st->line_count = 0;
-        st->line_first = heard + 1;
-        return;
-    }
-    for (uint64_t n = heard - st->line_first + 1; n; --n, --st->line_count, ++st->line_first)
-        st->line_head += 4 + (size_t)get32(st->line + st->line_head);
-}
-
-/* Records held that the line reached, moved onto it; ones behind it let go. */
-static void drop_first(Stream *st);
-static void line_pull(Stream *st) {
-    while (st->held_count) {
-        Held *h = held_at(st, 0);
-        if (h->seq < line_end(st)) drop_first(st);
-        else if (h->seq == line_end(st) && h->got == h->pieces && line_append(st, h->bytes, h->size)) drop_first(st);
-        else break;
-    }
-}
-
-/* The first record held let go: the ones after it stay where they are. */
-static void drop_first(Stream *st) {
-    st->held_bytes -= held_at(st, 0)->size;
-    free(held_at(st, 0)->bytes);
-    ++st->held_first;
-    if (!--st->held_count) st->held_first = 0;
-}
-
-/* The memo of secrets (pure_link_remember): a header, then a place a peer
- * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a
- * of its whole key and the secret) and the secret. A place whose check
- * fails (never made, another key, cut short by a crash) is made again. */
-enum { MEMO_HEAD = 64, MEMO_PLACE = 64 };
-static const unsigned char memo_magic[8] = {'P', 'L', 'M', '1', 0, 0, 0, 0};
-
-static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) {
-    return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32);
-}
-
-bool pure_link_remember(PureLinkListener *s, const char *path, char *error, size_t size) {
-    int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
-    struct stat info;
-    unsigned char head[MEMO_HEAD];
-    size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
-    bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077);
-    /* Another node's key, or none yet: begun again. */
-    bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) &&
-                !memcmp(head + 8, s->own_public, PURE_SEAL_KEY);
-    if (ok && !ours) {
-        memset(head, 0, MEMO_HEAD);
-        memcpy(head, memo_magic, 8);
-        memcpy(head + 8, s->own_public, PURE_SEAL_KEY);
-        ok = !ftruncate(fd, 0) && pwrite(fd, head, MEMO_HEAD, 0) == MEMO_HEAD;
-    }
-    ok = ok && (ours && (size_t)info.st_size >= needed ? true : !ftruncate(fd, (off_t)needed));
-    void *map = ok ? mmap(NULL, needed, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0) : MAP_FAILED;
-    if (fd >= 0) close(fd);
-    if (map == MAP_FAILED) {
-        snprintf(error, size, "%s: cannot keep secrets there (a file only its owner may read)", path);
-        return false;
-    }
-    s->memo = map;
-    s->memo_size = needed;
-    return true;
-}
-
-/* A peer's active state, made at its first packet: the secret its key
- * shares with ours, remembered or made (one X25519, then remembered: the
- * secret first, the key that makes the place its own last). */
-static Active *activate(PureLinkListener *s, Peer *peer, uint64_t now) {
-    Active *a = calloc(1, sizeof(*a)), **actives = room(s->actives, &s->active_capacity, s->active_count, sizeof(*actives), 64);
-    if (actives) s->actives = actives;
-    if (!a || !actives) {
-        free(a);
-        return NULL;
-    }
-    size_t index = (size_t)(peer - s->peers);
-    unsigned char *place = s->memo && MEMO_HEAD + (index + 1) * MEMO_PLACE <= s->memo_size ?
-                           s->memo + MEMO_HEAD + index * MEMO_PLACE : NULL;
-    if (place && !memcmp(place, peer->key, 28) && get32(place + 28) == memo_check(peer->key, place + 32)) {
-        memcpy(a->secret, place + 32, 32);
-        ++s->counts.remembered;
-    } else if (pure_link_secret(s->own, peer->key, a->secret)) {
-        if (place) {
-            memset(place, 0, 28);
-            __atomic_thread_fence(__ATOMIC_RELEASE);
-            memcpy(place + 32, a->secret, 32);
-            put32(place + 28, memo_check(peer->key, a->secret));
-            __atomic_thread_fence(__ATOMIC_RELEASE);
-            memcpy(place, peer->key, 28);
-        }
-    } else {
-        explicit_bzero(a, sizeof(*a));
-        free(a);
-        return NULL;
-    }
-    a->peer = (uint32_t)(peer - s->peers);
-    a->seen = now;
-    a->place = s->active_count;
-    s->actives[s->active_count++] = a;
-    peer->active = a;
-    return a;
-}
-
-static void let_go(PureLinkListener *s, Active *a) {
-    for (unsigned k = 0; k < a->stream_count; ++k) {
-        while (a->streams[k]->held_count) drop_first(a->streams[k]);
-        free(a->streams[k]->held);
-        free(a->streams[k]->line);
-        free(a->streams[k]);
-    }
-    free(a->streams);
-    s->peers[a->peer].active = NULL;
-    s->actives[a->place] = s->actives[--s->active_count];
-    s->actives[a->place]->place = a->place;
-    explicit_bzero(a, sizeof(*a));
-    free(a);
-}
-
-/* Whether any of a peer's records or answers waits here. */
-static bool busy(const Active *a) {
-    for (unsigned k = 0; k < a->stream_count; ++k)
-        if (a->streams[k]->out || a->streams[k]->held_count || a->streams[k]->line_count || a->streams[k]->ready) return true;
-    for (unsigned i = 0; i < RUNS; ++i)
-        if (a->runs[i].due) return true;
-    return false;
-}
-
-/* Let go the active state of peers quiet for the idle period; what idle
- * links held goes back to the system. */
-static void sweep(PureLinkListener *s, uint64_t now) {
-    if (now - s->swept < SWEEP_EVERY) return;
-    s->swept = now;
-    size_t freed = 0;
-    for (size_t k = 0; k < s->active_count;) {
-        Active *a = s->actives[k];
-        if (now - a->seen < s->idle || busy(a)) ++k;
-        else let_go(s, a), ++freed;
-    }
-    s->counts.swept += freed;
-    s->untrimmed += freed;
-    if (s->untrimmed >= TRIM_AFTER || (s->untrimmed && now - s->trimmed >= TRIM_WAIT)) {
-        malloc_trim(0);
-        s->untrimmed = 0;
-        s->trimmed = now;
-    }
-}
-
-static void mark_ready(PureLinkListener *s, Stream *st) {
-    Stream **ready = st->ready ? NULL : room(s->ready, &s->ready_capacity, s->ready_count, sizeof(*ready), 64);
-    if (!ready) return;
-    s->ready = ready;
-    s->ready[s->ready_count++] = st;
-    st->ready = true;
-}
-
-static void mark_due(PureLinkListener *s, Run *r) {
-    Run **due = r->due ? NULL : room(s->due, &s->due_capacity, s->due_count, sizeof(*due), 64);
-    if (!due) return;
-    s->due = due;
-    s->due[s->due_count++] = r;
-    r->due = true;
-}
-
-static Run *run_found(Active *a, uint64_t run) {
-    for (unsigned i = 0; i < RUNS; ++i)
-        if (a->runs[i].used && a->runs[i].run == run) return &a->runs[i];
-    return NULL;
-}
-
-/* A new run's keys, in the slot unused longest: only once its first packet
- * opened with them, so a forged one displaces nothing. */
-static Run *install_run(Active *a, const Run *made) {
-    Run *slot = &a->runs[0];
-    for (unsigned i = 0; i < RUNS; ++i) {
-        Run *r = &a->runs[i];
-        if (!r->used || (slot->used && r->seen < slot->seen)) slot = r;
-    }
-    bool due = slot->due; /* the answer list may hold the slot */
-    *slot = *made;
-    slot->due = due;
-    return slot;
-}
-
-static bool had(const Run *r, uint64_t pn) {
-    for (unsigned i = 0; i < r->got_count; ++i)
-        if (pn >= r->got[i].first && pn <= r->got[i].last) return true;
-    return false;
-}
-
-static void note(Run *r, uint64_t pn, uint64_t now) {
-    unsigned i = 0, n = r->got_count;
-    while (i < n && r->got[i].first > pn + 1) ++i;
-    if (i < n && r->got[i].first == pn + 1) { /* joins the range above it */
-        r->got[i].first = pn;
-        if (i + 1 < n && r->got[i + 1].last + 1 == pn) {
-            r->got[i].first = r->got[i + 1].first;
-            memmove(&r->got[i + 1], &r->got[i + 2], (n - i - 2) * sizeof(Range));
-            --r->got_count;
-        }
-    } else if (i < n && r->got[i].last + 1 == pn) {
-        r->got[i].last = pn;
-    } else if (i < RANGES) {
-        if (n == RANGES) --n; /* the oldest range is forgotten */
-        memmove(&r->got[i + 1], &r->got[i], (n - i) * sizeof(Range));
-        r->got[i] = (Range){pn, pn};
-        r->got_count = n + 1;
-    }
-    if (r->got_count && r->got[0].last == pn) r->largest_at = now;
-}
-
-static bool allowed(const PureLinkListener *s, const Peer *peer, const char *source) {
-    for (uint32_t a = peer->allowed; a; a = s->allowed[a - 1].next)
-        if (same_name(s, s->allowed[a - 1].name, source, strlen(source))) return true;
-    return false;
-}
-
-/* The stream of a peer's source to input, made the first time: NULL when the
- * source is not allowed from the peer, or the host has no input that takes
- * such messages. */
-static Stream *stream_of(PureLinkListener *s, Active *a, const char *source, const char *input, uint32_t shape) {
-    for (unsigned k = 0; k < a->stream_count; ++k) {
-        Stream *st = a->streams[k];
-        if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL;
-    }
-    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1;
-    if (index < 0) return NULL;
-    size_t capacity = a->stream_capacity;
-    Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
-    if (!streams) return NULL;
-    a->streams = streams;
-    a->stream_capacity = (unsigned)capacity;
-    if (!(st = calloc(1, sizeof(*st)))) return NULL;
-    st->active = a;
-    st->input = (uint32_t)index;
-    st->shape = shape;
-    snprintf(st->source, sizeof(st->source), "%s", source);
-    snprintf(st->input_name, sizeof(st->input_name), "%s", input);
-    a->streams[a->stream_count++] = st;
-    return st;
-}
-
-/* A record's piece, kept until the record is whole and handed over. */
-static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const unsigned char *bytes, uint32_t length) {
-    if (st->known && seq <= st->heard) {
-        st->stored_due = true; /* sent again: say again how far it is stored */
-        return;
-    }
-    if ((st->known && seq > st->heard + CREDIT) || size > RECORD_MOST) return;
-    if (st->known && seq < line_end(st)) return; /* on the line already */
-    if (st->known && seq == line_end(st) && !offset && length == size) { /* the next, whole */
-        if (st->held_bytes + (st->line_size - st->line_head) + size <= HELD_MOST && line_append(st, bytes, size))
-            line_pull(st);
-        return;
-    }
-    /* In order it goes last; out of order, its place is found by halving. */
-    size_t k = st->held_count;
-    if (k && held_at(st, k - 1)->seq >= seq) {
-        size_t low = 0;
-        while (low < k) {
-            size_t middle = low + (k - low) / 2;
-            if (held_at(st, middle)->seq < seq) low = middle + 1;
-            else k = middle;
-        }
-    }
-    if (k == st->held_count || held_at(st, k)->seq != seq) {
-        if (st->held_bytes + size > HELD_MOST) return;
-        if (st->held_first && st->held_first + st->held_count == st->held_capacity) { /* the front is free: used again */
-            memmove(st->held, held_at(st, 0), st->held_count * sizeof(Held));
-            st->held_first = 0;
-        }
-        Held *kept = room(st->held, &st->held_capacity, st->held_first + st->held_count, sizeof(Held), 16);
-        if (!kept) return;
-        st->held = kept;
-        Held h = {.seq = seq, .size = size, .pieces = size ? (size + PIECE - 1) / PIECE : 1};
-        if (!(h.bytes = malloc(size + h.pieces))) return;
-        memset(h.bytes + size, 0, h.pieces);
-        memmove(held_at(st, k + 1), held_at(st, k), (st->held_count - k) * sizeof(Held));
-        *held_at(st, k) = h;
-        ++st->held_count;
-        st->held_bytes += size;
-    }
-    Held *h = held_at(st, k);
-    if (h->size != size) return;
-    if (!offset && length == size) { /* whole: every piece had */
-        if (size) memcpy(h->bytes, bytes, size);
-        memset(h->bytes + size, 1, h->pieces);
-        h->got = h->pieces;
-        return;
-    }
-    uint32_t piece = offset / PIECE;
-    if (offset % PIECE || piece >= h->pieces || length != least(PIECE, size - offset) || h->bytes[size + piece]) return;
-    if (length) memcpy(h->bytes + offset, bytes, length);
-    h->bytes[size + piece] = 1;
-    ++h->got;
-}
-
-/* Whole records from the one after the last heard, in order, to the host:
- * one delivery a stream at a time. Before the last heard is known (after a
- * restart) they go from the first held: the host says how far it heard.
- * False when the host could not take them now. */
-static bool hand_over(PureLinkListener *s, Stream *st) {
-    if (st->out) return true;
-    if (st->known) { /* the line as it is, up to BATCH records */
-        line_pull(st);
-        if (!st->line_count) return true;
-        unsigned n = 0;
-        size_t bytes = 0;
-        for (; n < st->line_count && n < BATCH; ++n) bytes += 4 + (size_t)get32(st->line + st->line_head + bytes);
-        PureLinkDelivery d = {st->input, st->source, st->line_first, n, st->line + st->line_head, bytes, st};
-        if (!s->host.deliver(s->host.context, &d)) return false;
-        st->out = true;
-        ++s->counts.deliveries;
-        return true;
-    }
-    if (!st->held_count || (st->known && held_at(st, 0)->seq != st->heard + 1)) return true;
-    size_t n = 0, bytes = 0;
-    while (n < st->held_count && n < BATCH && held_at(st, n)->got == held_at(st, n)->pieces &&
-           held_at(st, n)->seq == held_at(st, 0)->seq + n)
-        bytes += 4 + held_at(st, n++)->size;
-    if (!n) return true;
-    if (bytes > s->batch_capacity) {
-        unsigned char *grown = realloc(s->batch, bytes);
-        if (!grown) return false;
-        s->batch = grown;
-        s->batch_capacity = bytes;
-    }
-    size_t at = 0;
-    for (size_t k = 0; k < n; ++k) {
-        put32(s->batch + at, held_at(st, k)->size);
-        memcpy(s->batch + at + 4, held_at(st, k)->bytes, held_at(st, k)->size);
-        at += 4 + held_at(st, k)->size;
-    }
-    PureLinkDelivery d = {st->input, st->source, held_at(st, 0)->seq, (unsigned)n, s->batch, bytes, st};
-    if (!s->host.deliver(s->host.context, &d)) return false;
-    st->out = true;
-    ++s->counts.deliveries;
-    return true;
-}
-
-/* A packet from peer, if it opens with the keys of peer's run (made for a
- * run not seen yet, and for a peer at rest its active state): its frames. */
-static bool take_from(PureLinkListener *s, Peer *peer, unsigned char *packet, size_t n,
-                      const struct sockaddr_storage *from, socklen_t from_size, uint64_t now) {
-    uint64_t pn = get64(packet + 13), run = get64(packet + 5);
-    Active *a = peer->active;
-    Run *r = a ? run_found(a, run) : NULL;
-    if (r && had(r, pn)) {
-        r->ack_due = true; /* its acknowledgment was lost, perhaps */
-        mark_due(s, r);
-        return true;
-    }
-    bool woke = !a;
-    if (woke && !(a = activate(s, peer, now))) return false;
-    Run made = {.active = a, .run = run, .seen = now, .used = true};
-    bool keyed = r || (derive(a->secret, peer->key, s->own_public, run, 0, made.in_key, made.in_base) &&
-                       derive(a->secret, s->own_public, peer->key, s->run, run, made.out_key, made.out_base));
-    unsigned char plain[DATAGRAM], nonce[PURE_AEAD_NONCE];
-    nonce_of(nonce, r ? r->in_base : made.in_base, pn);
-    size_t size = n - HEADER - PURE_AEAD_TAG;
-    const unsigned char *key = r ? r->in_key : made.in_key, *give = key;
-    if (s->open_keyed && !memcmp(s->open_key, key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
-    else if (keyed) memcpy(s->open_key, key, PURE_AEAD_KEY), s->open_keyed = true;
-    if (!keyed || !pure_aead_open(s->aead, give, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) {
-        explicit_bzero(&made, sizeof(made));
-        if (woke) let_go(s, a);
-        return false;
-    }
-    if (woke) ++s->counts.woken;
-    if (!r) r = install_run(a, &made);
-    a->seen = r->seen = now;
-    note(r, pn, now);
-    r->address_size = from_size <= sizeof(r->address) ? from_size : sizeof(r->address);
-    memcpy(&r->address, from, r->address_size); /* where it is now: answers go there */
-    ++s->counts.received;
-    Stream *st = NULL;
-    bool eliciting = false;
-    for (size_t at = 0; at < size;) {
-        unsigned type = plain[at++];
-        if (type == F_STREAM) {
-            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
-            size_t names = get_names(plain + at, size - at, source, input);
-            if (!names || at + names + 4 > size) break;
-            st = stream_of(s, a, source, input, get32(plain + at + names));
-            if (st) {
-                st->run = r;
-                st->run_id = r->run;
-            }
-            at += names + 4;
-            eliciting = true;
-        } else if (type == F_RECORD) {
-            if (at + RECORD_HEAD - 1 > size) break;
-            uint32_t length = get16(plain + at + 16);
-            if (at + RECORD_HEAD - 1 + length > size) break;
-            if (st) {
-                hold(st, get64(plain + at), get32(plain + at + 8), get32(plain + at + 12), plain + at + 18, length);
-                mark_ready(s, st);
-            }
-            at += RECORD_HEAD - 1 + length;
-            eliciting = true;
-        } else if (type == F_RECORDS) {
-            if (at + RECORDS_HEAD - 1 > size) break;
-            uint64_t first = get64(plain + at), length = 0;
-            unsigned count = plain[at + 8];
-            at += RECORDS_HEAD - 1;
-            unsigned i = 0;
-            for (size_t used; i < count && (used = get_varint(plain + at, size - at, &length)) &&
-                              length <= size - at - used; ++i) {
-                if (st) hold(st, first + i, (uint32_t)length, 0, plain + at + used, (uint32_t)length);
-                at += used + length;
-            }
-            if (st) mark_ready(s, st);
-            eliciting = true;
-            if (i < count) break; /* cut short: the rest is not read */
-        } else if (type == F_PING) {
-            eliciting = true;
-        } else
-            break; /* padding, or what this side does not know: the rest */
-    }
-    if (eliciting) {
-        r->ack_due = true;
-        mark_due(s, r);
-    }
-    return true;
-}
-
-static void take_packet(PureLinkListener *s, unsigned char *packet, size_t n, const struct sockaddr_storage *from,
-                        socklen_t from_size, uint64_t now) {
-    if (n < HEADER + PURE_AEAD_TAG || n > DATAGRAM || packet[0] != VERSION || !s->index_size) return;
-    uint32_t id = get32(packet + 1);
-    size_t mask = s->index_size - 1;
-    uint64_t run = get64(packet + 5);
-    /* The peers with the packet's key id: first one whose run is known (no
-     * key to make), then any. */
-    for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
-        Peer *peer = &s->peers[s->index[slot] - 1];
-        if (peer->id == id && peer->active && run_found(peer->active, run) &&
-            take_from(s, peer, packet, n, from, from_size, now))
-            return;
-    }
-    for (size_t slot = id & mask; s->index[slot]; slot = (slot + 1) & mask) {
-        Peer *peer = &s->peers[s->index[slot] - 1];
-        if (peer->id == id && !(peer->active && run_found(peer->active, run)) &&
-            take_from(s, peer, packet, n, from, from_size, now))
-            return;
-    }
-}
-
-/* Acknowledgments and how far each stream is stored, a packet to each run
- * that has any. How far a stream is stored goes with every acknowledgment
- * to its run, so a lost word is said again at the next packet. */
-static void answer(PureLinkListener *s, uint64_t now) {
-    unsigned char body[BODY], packet[DATAGRAM];
-    for (size_t k = 0; k < s->due_count; ++k) {
-        Run *r = s->due[k];
-        Active *a = r->active;
-        r->due = false;
-        if (!r->used) continue;
-        size_t at = 0;
-        if (r->ack_due && r->got_count) {
-            body[at++] = F_ACK;
-            put64(body + at, r->got[0].last);
-            put32(body + at + 8, (uint32_t)least((now - r->largest_at) / 1000, UINT32_MAX));
-            body[at + 12] = (unsigned char)r->got_count;
-            at += 13;
-            for (unsigned i = 0; i < r->got_count; ++i, at += 16) {
-                put64(body + at, r->got[i].first);
-                put64(body + at + 8, r->got[i].last);
-            }
-        }
-        for (unsigned j = 0; j < a->stream_count; ++j) {
-            Stream *st = a->streams[j];
-            if (!(st->stored_due || r->ack_due) || st->run != r || st->run_id != r->run || !st->known) continue;
-            if (at + strlen(st->source) + strlen(st->input_name) + 16 > BODY) break; /* type, names, 13 */
-            body[at++] = F_STORED;
-            at += put_names(body + at, st->source, st->input_name);
-            put64(body + at, st->heard);
-            body[at + 8] = st->gap ? STORED_GAP : 0;
-            size_t room = st->held_bytes + (st->line_size - st->line_head) < HELD_MOST ? CREDIT : 1;
-            put32(body + at + 9, (uint32_t)room);
-            at += 13;
-            st->stored_due = st->gap = false;
-        }
-        r->ack_due = false;
-        if (!at) continue;
-        const unsigned char *give = r->out_key;
-        if (s->seal_keyed && !memcmp(s->seal_key, r->out_key, PURE_AEAD_KEY)) give = NULL; /* the key it has */
-        else memcpy(s->seal_key, r->out_key, PURE_AEAD_KEY), s->seal_keyed = true;
-        size_t total = seal_packet(s->aead, give, r->out_base, s->own_id, s->run, r->next_pn++, body, at, packet);
-        if (total && sendto(s->fd, packet, total, 0, (const struct sockaddr *)&r->address, r->address_size) >= 0)
-            ++s->counts.packets;
-    }
-    s->due_count = 0;
-}
-
-static void take_done(PureLinkListener *s) {
-    pthread_mutex_lock(&s->lock);
-    Done *done = s->done;
-    size_t count = s->done_count;
-    s->done = NULL;
-    s->done_count = s->done_capacity = 0;
-    pthread_mutex_unlock(&s->lock);
-    for (size_t k = 0; k < count; ++k) {
-        Stream *st = done[k].stream;
-        st->out = false;
-        mark_ready(s, st);
-        if (!done[k].stored) continue; /* not saved: handed over again */
-        if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
-        if (!st->known) st->line_first = st->heard + 1; /* the line begins */
-        line_drop(st, st->heard);
-        while (st->held_count && held_at(st, 0)->seq <= st->heard) drop_first(st);
-        st->known = true;
-        st->gap = done[k].gap;
-        st->stored_due = true;
-        if (st->run && st->run->used && st->run_id == st->run->run) mark_due(s, st->run);
-        s->counts.stored = st->heard;
-    }
-    free(done);
-}
-
-/* The streams ready, each handed over; one the host could not take now is
- * tried again at the next turn. */
-static void hand_over_ready(PureLinkListener *s) {
-    size_t count = s->ready_count;
-    for (size_t k = 0; k < count; ++k) {
-        Stream *st = s->ready[k];
-        st->ready = false;
-        if (!hand_over(s, st)) mark_ready(s, st);
-    }
-    memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready));
-    s->ready_count -= count;
+static bool send_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
+                        size_t to_size) {
+    const PureLinkListener *s = context;
+    return sendto(s->fd, packet, size, 0, (const struct sockaddr *)(const void *)to, (socklen_t)to_size) >= 0;
 }
 
 static void *run_listener(void *context) {
@@ -1722,8 +355,9 @@ static void *run_listener(void *context) {
     Gathering *gathering = malloc(sizeof(*gathering));
     if (!gathering) return NULL;
     for (;;) {
+        uint64_t now = now_ns(), due = pure_link_receiver_due(s->receiver);
+        int timeout = due ? (int)((due > now ? due - now : 0) / MS) + 1 : -1;
         struct pollfd waits[2] = {{s->fd, POLLIN, 0}, {s->wake, POLLIN, 0}};
-        int timeout = s->active_count || s->ready_count ? (int)(SWEEP_EVERY / MS) : -1;
         if (poll(waits, 2, timeout) < 0 && errno != EINTR) break;
         pthread_mutex_lock(&s->lock);
         bool stopping = s->stopping;
@@ -1733,21 +367,25 @@ static void *run_listener(void *context) {
             uint64_t signals;
             if (read(s->wake, &signals, sizeof(signals)) < 0) { /* cleared already */ }
         }
-        uint64_t now = now_ns();
+        now = now_ns();
         for (unsigned rounds = 0, n = GATHER; rounds < 8 && n == GATHER; ++rounds) {
             n = (unsigned)gather(s->fd, gathering);
             for (unsigned k = 0; k < n; ++k) {
                 struct mmsghdr *m = &gathering->messages[k];
                 size_t length = m->msg_len, step = segment_of(&m->msg_hdr, length);
                 for (size_t at = 0; at < length; at += step)
-                    take_packet(s, gathering->bytes[k] + at, least(step, length - at), &gathering->from[k],
-                                m->msg_hdr.msg_namelen, now);
+                    pure_link_receiver_take(s->receiver, gathering->bytes[k] + at, step < length - at ? step : length - at,
+                                            &gathering->from[k], m->msg_hdr.msg_namelen, now);
             }
         }
-        take_done(s);
-        hand_over_ready(s);
-        answer(s, now);
-        sweep(s, now);
+        pthread_mutex_lock(&s->lock);
+        PureLinkDone *done = s->done;
+        size_t count = s->done_count;
+        s->done = NULL;
+        s->done_count = s->done_capacity = 0;
+        pthread_mutex_unlock(&s->lock);
+        pure_link_receiver_turn(s->receiver, done, count, now, send_answer, s);
+        free(done);
     }
     free(gathering);
     return NULL;
@@ -1773,11 +411,8 @@ bool pure_link_serve(PureLinkListener *s, char *error, size_t size) {
     setsockopt(s->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
     int one = 1; /* without GRO, a buffer is a datagram */
     setsockopt(s->fd, SOL_UDP, UDP_GRO, &one, sizeof(one));
-    s->run = random64();
-    s->swept = s->trimmed = now_ns();
     s->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
-    unsigned char key[PURE_AEAD_KEY] = {0}; /* each packet brings its run's */
-    if (s->wake < 0 || !(s->aead = pure_aead(key)) || pthread_create(&s->thread, NULL, run_listener, s)) {
+    if (s->wake < 0 || pthread_create(&s->thread, NULL, run_listener, s)) {
         snprintf(error, size, "--link-port %u: cannot start", s->port);
         return false;
     }
@@ -1790,12 +425,12 @@ void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool store
     pthread_mutex_lock(&s->lock);
     if (s->done_count == s->done_capacity) {
         size_t capacity = s->done_capacity ? 2 * s->done_capacity : 16;
-        Done *grown = realloc(s->done, capacity * sizeof(*grown));
+        PureLinkDone *grown = realloc(s->done, capacity * sizeof(*grown));
         if (!grown) abort(); /* a delivery's word cannot be lost */
         s->done = grown;
         s->done_capacity = capacity;
     }
-    s->done[s->done_count++] = (Done){token, heard, stored, gap};
+    s->done[s->done_count++] = (PureLinkDone){token, heard, stored, gap};
     pthread_mutex_unlock(&s->lock);
     wake(s->wake);
 }
@@ -1809,21 +444,8 @@ void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) {
         wake(s->wake);
         pthread_join(s->thread, NULL);
     }
-    if (counts) {
-        *counts = s->counts;
-        counts->active = s->active_count;
-    }
-    while (s->active_count) let_go(s, s->actives[0]);
-    pure_aead_free(s->aead);
-    if (s->memo) munmap(s->memo, s->memo_size);
-    free(s->actives);
-    free(s->peers);
-    free(s->index);
-    free(s->allowed);
-    free(s->names);
-    free(s->ready);
-    free(s->due);
-    free(s->batch);
+    if (counts) pure_link_receiver_counts(s->receiver, counts);
+    pure_link_receiver_free(s->receiver);
     free(s->done);
     if (s->fd >= 0) close(s->fd);
     if (s->wake >= 0) close(s->wake);
diff --git a/link/link.h b/link/link.h
index 1ee9473..5237262 100644
--- a/link/link.h
+++ b/link/link.h
@@ -1,53 +1,11 @@
 #ifndef PURE_LINK_H
 #define PURE_LINK_H
-#include "seal.h"
+#include "link-protocol.h"
 
-/* Links without connections (docs/design/011-pure-transport.md): one node's
- * saved sends to another node's input over UDP, sealed with a key the two
- * nodes' keys share, with no handshake: the first packet carries records.
- * A link's lasting state is the apps' own: the sender's outbox (what to send
- * and send again) and the receiver's heard numbers (what it applied, saved
- * with the events). What either side keeps in memory is a cache, so either
- * may restart and the other goes on with the next packet.
- *
- * A packet is a header in the clear, then frames sealed with AES-128-GCM
- * (the header as associated data):
- *   u8 0x70 (version 0), u32le the key id of the node that sent it (FNV-1a
- *   of its public key; two keys may share an id, and a receiver tries the
- *   keys it allows with it), u64le that node's run (random, new each time it
- *   starts, so no nonce repeats under a key across restarts), u64le the
- *   packet's number in its run.
- * Keys come from the secret the two nodes' keys share (X25519) by
- * HKDF-SHA256, with both public keys and the sending node's run: a sender's
- * packets use its own run; a receiver's answers add the run they answer,
- * so an answer opens only in that run. The nonce is the key's base with the
- * packet's number in its last eight bytes (XOR, big-endian, as QUIC does).
- *
- * Frames (numbers little-endian):
- *   0x01 STREAM  source (u8 length, bytes), input (u8 length, bytes), u32 the
- *                shape of the messages (the sender's output and the
- *                receiver's input must have the same): the stream the
- *                records after it are of
- *   0x02 RECORD  u64 number, u32 size, u32 offset, u16 length, bytes: a piece
- *                of a saved send (pieces are 1000 bytes, the last shorter)
- *   0x03 ACK     u64 largest number, u32 delay (µs), u8 ranges, each as u64
- *                first and u64 last, newest first
- *   0x04 STORED  source, input (as STREAM), u64 the last number heard
- *                (applied and saved), u8 flags (1: a record after a gap was
- *                not taken: send what follows it again), u32 credit (records
- *                it takes past that)
- *   0x05 PING    to be acknowledged
- *   0x06 RECORDS u64 the first one's number, u8 count, then whole records one
- *                after another (numbers following on), each its size (a
- *                varint: LEB128) and its bytes: a send that fits the packet
- *                whole, with 1 to 3 bytes of framing, not the 19 of RECORD
- *   0x00 PADDING to the end of the packet: a full packet sent in a burst is
- *                a whole datagram
- * Losses are detected as RFC 9002 says; congestion control is its NewReno
- * (RFC 9002 §7). */
-
-/* A source's or an input's name is 1 to PURE_LINK_NAME bytes. */
-enum { PURE_LINK_NAME = 64 };
+/* Links over UDP: the two sides of link/link-protocol.h (what goes on the
+ * wire, and what each side keeps) with sockets and threads. A sender sends
+ * on a thread of its own, packets in bursts with UDP GSO; a listener takes
+ * a receiver's packets on a port, with UDP GRO. */
 
 /* The sending side: one output's saved sends to url, pure://host:port/INPUT,
  * as source (a Pure receiver's name for it), their messages of shape (the
@@ -66,59 +24,17 @@ uint64_t pure_link_queued(PureLink *);
 uint64_t pure_link_delivered(PureLink *);
 /* Send on a thread of the link's own, calling notify when more was saved. */
 bool pure_link_start(PureLink *, void (*notify)(void *), void *context, char *error, size_t size);
+void pure_link_counts(PureLink *, PureLinkCounts *);
 void pure_link_close(PureLink *);
 
-/* The receiving side: packets on a UDP port, from the senders allowed (a
- * source name and the key it seals with, a few bytes each however many
- * there are: see pure_link_idle). A stream (a source to an input) goes to
- * the host's input its name and shape find (its number, or -1: the stream
- * is refused). Whole records of a stream, in order, go to deliver (both on
- * the listener's thread) as one delivery; the host applies them from the
- * one after the source's last heard number, stops at a gap, saves, and then
- * calls pure_link_done (any thread) with the last number heard. Records
- * wait here until then, and only the saved ones are acknowledged as
- * stored. */
-typedef struct {
-    uint32_t input;
-    const char *source;
-    uint64_t first;
-    unsigned count;
-    const unsigned char *records; /* count of: u32le size, bytes */
-    size_t size;
-    void *token; /* for pure_link_done */
-} PureLinkDelivery;
-typedef struct {
-    int (*input)(void *context, const char *name, uint32_t shape);
-    bool (*deliver)(void *context, const PureLinkDelivery *);
-    void *context;
-} PureLinkHost;
+/* The receiving side: a receiver's packets on a UDP port, taken on a thread
+ * of the listener's own, where the host's input and deliver are called. The
+ * listener takes the receiver: closing it frees the receiver, and gives its
+ * counts. pure_link_done (any thread) says how a delivery's save went. */
 typedef struct PureLinkListener PureLinkListener;
-PureLinkListener *pure_link_listen(unsigned port, const PureSealKey *own, const unsigned char own_public[PURE_SEAL_KEY],
-                                   const PureLinkHost *host, char *error, size_t size);
-bool pure_link_allow(PureLinkListener *, const char *source, const unsigned char key[PURE_SEAL_KEY], char *error,
-                     size_t size);
-/* A sender quiet for ms (30 s at first) with nothing of it waiting lets its
- * keys, addresses and streams go: an idle link costs the listener only its
- * allowing, and the sender's next packet makes them again (one X25519). */
-void pure_link_idle(PureLinkListener *, uint64_t ms);
-/* Remember the secret each sender's key shares with the node's in a file
- * (path, 0600; begun again for another node's key): a place a sender in
- * the order allowed, mapped, so a sender at rest takes no memory for it.
- * A node that starts again then makes no X25519 for a sender it heard from
- * before. After the senders are allowed, before serving. */
-bool pure_link_remember(PureLinkListener *, const char *path, char *error, size_t size);
+PureLinkListener *pure_link_listen(PureLinkReceiver *, unsigned port, char *error, size_t size);
 bool pure_link_serve(PureLinkListener *, char *error, size_t size);
 /* stored: the delivery's save succeeded; gap: it stopped at a gap. */
 void pure_link_done(PureLinkListener *, void *token, uint64_t heard, bool stored, bool gap);
-
-/* Counts for a host's report: packets and records sent, records sent again,
- * packets lost and acknowledged, packets received, deliveries handed over,
- * the last number stored; a listener's senders woken (their active state
- * made), of them with a secret remembered (no X25519), let go when idle,
- * and active when it closed. A listener's are given when it closes. */
-typedef struct {
-    uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, woken, remembered, swept, active;
-} PureLinkCounts;
-void pure_link_counts(PureLink *, PureLinkCounts *);
 void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts);
 #endif
diff --git a/link/test-link.c b/link/test-link.c
new file mode 100644
index 0000000..8be32a9
--- /dev/null
+++ b/link/test-link.c
@@ -0,0 +1,300 @@
+#define _GNU_SOURCE
+/* A link's two sides (link/link-protocol.h) through a network of the test's
+ * own: time is the test's, and packets are lost, held back (so they come out
+ * of order) and doubled at random, from a seed. The receiving host applies a
+ * record once, from the one after its last heard, and says so a turn later,
+ * now and then failing a save; the receiver and the sender start again
+ * midway, each from what its app kept. Every record must be applied once,
+ * in order, as it was sent. */
+#include "link-protocol.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <unistd.h>
+
+#define MS UINT64_C(1000000)
+enum { PACKETS = 1 << 16 };
+
+static uint64_t seed;
+static uint64_t next_random(void) {
+    seed ^= seed << 13, seed ^= seed >> 7, seed ^= seed << 17;
+    return seed;
+}
+static bool chance(unsigned percent) { return next_random() % 100 < percent; }
+
+static int failures;
+#define CHECK(condition, ...)                                                                                          \
+    do {                                                                                                               \
+        if (!(condition)) {                                                                                            \
+            fprintf(stderr, "test-link: " __VA_ARGS__);                                                                \
+            fputc('\n', stderr);                                                                                       \
+            ++failures;                                                                                                \
+        }                                                                                                              \
+    } while (0)
+
+/* A record's size and bytes, from its number: small ones, ones of a packet or
+ * so, and some of many pieces. */
+static size_t record_size(uint64_t seq) {
+    uint64_t h = seq * 0x9e3779b97f4a7c15u;
+    unsigned kind = (unsigned)(h >> 60) % 10;
+    return kind < 3 ? (size_t)(h >> 8) % 61 : kind < 8 ? 61 + (size_t)(h >> 8) % 1040 : 1101 + (size_t)(h >> 8) % 19000;
+}
+static unsigned char record_byte(uint64_t seq, size_t i) { return (unsigned char)(seq * 31 + i * 7 + (i >> 8)); }
+
+typedef struct {
+    uint64_t at;
+    bool to_receiver;
+    size_t size;
+    unsigned char bytes[PURE_LINK_DATAGRAM];
+} Packet;
+
+typedef struct {
+    /* The network: packets on their way, each arriving at its time. */
+    Packet *packets;
+    size_t count;
+    unsigned loss, doubled, spread_ms; /* percent, percent, the most a packet is held back */
+    uint64_t now, sent;
+    /* The receiving app: its last heard (kept across the receiver's
+     * restarts), what it said of deliveries (given at the next turn). */
+    uint64_t heard, applied, deliveries;
+    PureLinkDone done[64], told[64];
+    size_t done_count;
+    unsigned fail_every; /* a delivery's save fails every so many (0: never) */
+} World;
+
+static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) {
+    ++w->sent;
+    for (int copies = chance(w->doubled) ? 2 : 1; copies--;) {
+        if (chance(w->loss) || w->count == PACKETS) continue;
+        Packet *p = &w->packets[w->count++];
+        p->at = w->now + 5 * MS + (w->spread_ms ? next_random() % (w->spread_ms * MS) : 0);
+        p->to_receiver = to_receiver;
+        p->size = size;
+        memcpy(p->bytes, bytes, size);
+    }
+}
+
+static bool answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to, size_t to_size) {
+    CHECK(to_size == 6 && !memcmp(to, "sender", 6), "an answer to an address no packet came from");
+    put_packet(context, false, packet, size);
+    return true;
+}
+
+static int input_of(void *context, const char *name, uint32_t shape) {
+    (void)context;
+    return !strcmp(name, "order") && shape == 0x5eed ? 3 : -1;
+}
+
+/* Applied from the one after the last heard, in order; a gap stops it. */
+static bool deliver(void *context, const PureLinkDelivery *d) {
+    World *w = context;
+    CHECK(d->input == 3 && !strcmp(d->source, "test.ordered"), "a delivery to input %u from %s", d->input, d->source);
+    if (w->done_count == sizeof(w->done) / sizeof(w->done[0])) return false; /* busy: tried again */
+    ++w->deliveries;
+    bool fail = w->fail_every && w->deliveries % w->fail_every == 0, gap = false;
+    uint64_t heard = w->heard;
+    const unsigned char *at = d->records;
+    for (unsigned k = 0; k < d->count; ++k) {
+        uint32_t size = (uint32_t)at[0] | (uint32_t)at[1] << 8 | (uint32_t)at[2] << 16 | (uint32_t)at[3] << 24;
+        uint64_t seq = d->first + k;
+        if (seq == heard + 1) {
+            bool same = size == record_size(seq);
+            for (size_t i = 0; same && i < size; ++i) same = at[4 + i] == record_byte(seq, i);
+            CHECK(same, "record %llu is not as it was sent", (unsigned long long)seq);
+            heard = seq;
+        } else if (seq > heard + 1) {
+            gap = true;
+            break;
+        }
+        at += 4 + size;
+    }
+    if (!fail) {
+        w->applied += heard - w->heard;
+        w->heard = heard;
+    }
+    w->done[w->done_count++] = (PureLinkDone){d->token, w->heard, !fail, gap};
+    return true;
+}
+
+static PureSealKey *key(const char *dir, const char *name, unsigned char public_key[PURE_SEAL_KEY]) {
+    char path[512], hex[2 * PURE_SEAL_KEY + 1], error[256];
+    snprintf(path, sizeof(path), "%s/%s", dir, name);
+    PureSealKey *k = pure_seal_key(path, hex, error, sizeof(error));
+    if (!k || !pure_seal_peer(hex, public_key)) {
+        fprintf(stderr, "test-link: %s\n", error);
+        exit(1);
+    }
+    return k;
+}
+
+typedef struct {
+    const char *name;
+    uint64_t records, seed;
+    unsigned loss, doubled, spread_ms, fail_every;
+    uint64_t pause_after, receiver_restart_after, sender_restart_after, idle_ms;
+} Case;
+
+typedef struct { PureLinkCounts sent, heard; uint64_t packets, seconds_ms; } Outcome;
+
+static PureLinkReceiver *receiver_of(World *w, const PureSealKey *b, const unsigned char *public_b,
+                                     const unsigned char *public_a, const char *memo, uint64_t idle_ms) {
+    char error[256];
+    PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){input_of, deliver, w}, error, sizeof(error));
+    if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error)) ||
+        !pure_link_remember(r, memo, error, sizeof(error))) {
+        fprintf(stderr, "test-link: %s\n", error);
+        exit(1);
+    }
+    if (idle_ms) pure_link_idle(r, idle_ms);
+    return r;
+}
+
+/* The sender as its app makes it: the sends its outbox keeps, from the one
+ * after the last stored, up to limit. */
+static PureLinkSender *sender_of(const PureSealKey *a, const unsigned char *public_a, const unsigned char *public_b,
+                                 uint64_t from, uint64_t limit) {
+    PureLinkSender *s = pure_link_sender("test.ordered", "order", 0x5eed, a, public_a);
+    if (!s || !pure_link_sender_peer(s, public_b)) {
+        fprintf(stderr, "test-link: cannot make a sender\n");
+        exit(1);
+    }
+    static unsigned char bytes[20200];
+    for (uint64_t seq = from; seq <= limit; ++seq) {
+        size_t size = record_size(seq);
+        for (size_t i = 0; i < size; ++i) bytes[i] = record_byte(seq, i);
+        if (!pure_link_sender_queue(s, seq, bytes, size)) {
+            fprintf(stderr, "test-link: cannot queue %llu\n", (unsigned long long)seq);
+            exit(1);
+        }
+    }
+    return s;
+}
+
+static Outcome run(const Case *c, const char *dir) {
+    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY];
+    PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
+    char memo[512];
+    snprintf(memo, sizeof(memo), "%s/b.key.links", dir);
+    unlink(memo);
+    World w = {.packets = malloc(PACKETS * sizeof(Packet)), .loss = c->loss, .doubled = c->doubled,
+               .spread_ms = c->spread_ms, .fail_every = c->fail_every};
+    if (!w.packets) exit(1);
+    seed = c->seed;
+    uint64_t queued = c->pause_after ? c->pause_after : c->records, resumed = 0;
+    PureLinkSender *s = sender_of(a, public_a, public_b, 1, queued);
+    PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, c->idle_ms);
+    PureLinkCounts sent_before = {0}, heard_before = {0};
+    bool receiver_restarted = !c->receiver_restart_after, sender_restarted = !c->sender_restart_after;
+    unsigned char out[PURE_LINK_DATAGRAM];
+    while (w.heard < c->records || pure_link_sender_stored(s) < c->records) {
+        CHECK(w.now < 600000 * MS, "%s: not done after ten minutes (heard %llu of %llu)", c->name,
+              (unsigned long long)w.heard, (unsigned long long)c->records);
+        if (w.now >= 600000 * MS) break;
+        uint64_t number;
+        for (size_t size; (size = pure_link_sender_packet(s, out, false, w.now, &number));)
+            put_packet(&w, true, out, size);
+        /* Then the next thing to happen. */
+        uint64_t next = UINT64_MAX, due = pure_link_sender_due(s), turn = pure_link_receiver_due(r);
+        for (size_t k = 0; k < w.count; ++k)
+            if (w.packets[k].at < next) next = w.packets[k].at;
+        if (due && due < next) next = due;
+        if (turn && turn < next) next = turn;
+        if (w.done_count && w.now + MS < next) next = w.now + MS; /* the host's word, a turn later */
+        if (queued < c->records && pure_link_sender_stored(s) >= queued && !resumed) resumed = w.now + 2000 * MS;
+        if (resumed && queued < c->records && resumed < next) next = resumed;
+        if (next == UINT64_MAX) {
+            CHECK(false, "%s: nothing more happens (heard %llu of %llu)", c->name, (unsigned long long)w.heard,
+                  (unsigned long long)c->records);
+            break;
+        }
+        w.now = next > w.now ? next : w.now;
+        if (resumed && resumed <= w.now && queued < c->records) { /* after a pause: the rest */
+            static unsigned char bytes[20200];
+            for (uint64_t seq = queued + 1; seq <= c->records; ++seq) {
+                size_t size = record_size(seq);
+                for (size_t i = 0; i < size; ++i) bytes[i] = record_byte(seq, i);
+                CHECK(pure_link_sender_queue(s, seq, bytes, size), "%s: cannot queue %llu", c->name, (unsigned long long)seq);
+            }
+            queued = c->records;
+        }
+        for (size_t k = 0; k < w.count;) {
+            Packet *p = &w.packets[k];
+            if (p->at > w.now) {
+                ++k;
+                continue;
+            }
+            if (p->to_receiver) pure_link_receiver_take(r, p->bytes, p->size, "sender", 6, w.now);
+            else pure_link_sender_take(s, p->bytes, p->size, w.now);
+            *p = w.packets[--w.count];
+        }
+        pure_link_sender_tick(s, w.now);
+        size_t told = w.done_count;
+        memcpy(w.told, w.done, told * sizeof(*w.done));
+        w.done_count = 0;
+        pure_link_receiver_turn(r, w.told, told, w.now, answer, &w);
+        if (!receiver_restarted && w.heard >= c->receiver_restart_after) { /* kill -9: what it held is gone */
+            receiver_restarted = true;
+            pure_link_receiver_counts(r, &heard_before);
+            pure_link_receiver_free(r);
+            w.done_count = 0;
+            r = receiver_of(&w, b, public_b, public_a, memo, c->idle_ms);
+        }
+        if (!sender_restarted && pure_link_sender_stored(s) >= c->sender_restart_after) {
+            sender_restarted = true;
+            pure_link_sender_counts(s, &sent_before);
+            uint64_t stored = pure_link_sender_stored(s);
+            pure_link_sender_free(s);
+            s = sender_of(a, public_a, public_b, stored + 1, queued);
+        }
+    }
+    Outcome o = {.packets = w.sent, .seconds_ms = w.now / MS};
+    pure_link_sender_counts(s, &o.sent);
+    pure_link_receiver_counts(r, &o.heard);
+    o.sent.packets += sent_before.packets, o.sent.resent += sent_before.resent;
+    o.heard.woken += heard_before.woken, o.heard.swept += heard_before.swept;
+    CHECK(w.heard == c->records && w.applied == c->records, "%s: applied %llu, heard %llu of %llu", c->name,
+          (unsigned long long)w.applied, (unsigned long long)w.heard, (unsigned long long)c->records);
+    CHECK(pure_link_sender_stored(s) == c->records, "%s: the sender knows %llu stored of %llu", c->name,
+          (unsigned long long)pure_link_sender_stored(s), (unsigned long long)c->records);
+    printf("{\"case\":\"%s\",\"records\":%llu,\"virtual_ms\":%llu,\"packets\":%llu,\"resent\":%llu,\"lost\":%llu,"
+           "\"deliveries\":%llu,\"woken\":%llu,\"remembered\":%llu,\"swept\":%llu}\n",
+           c->name, (unsigned long long)c->records, (unsigned long long)o.seconds_ms, (unsigned long long)o.sent.packets,
+           (unsigned long long)o.sent.resent, (unsigned long long)o.sent.lost, (unsigned long long)w.deliveries,
+           (unsigned long long)o.heard.woken, (unsigned long long)o.heard.remembered, (unsigned long long)o.heard.swept);
+    pure_link_sender_free(s);
+    pure_link_receiver_free(r);
+    pure_seal_key_free(a);
+    pure_seal_key_free(b);
+    free(w.packets);
+    return o;
+}
+
+int main(void) {
+    char dir[] = "/tmp/pure-test-link-XXXXXX", error[256];
+    if (!mkdtemp(dir) || !pure_seal_ready(error, sizeof(error))) {
+        fprintf(stderr, "test-link: %s\n", error);
+        return 1;
+    }
+    /* A clean network: nothing is sent twice. */
+    Case clean = {"clean", 3000, 1, 0, 0, 0, 0, 0, 0, 0, 0};
+    Outcome o = run(&clean, dir);
+    CHECK(o.sent.resent == 0 && o.sent.lost == 0, "clean: %llu sent again, %llu lost", (unsigned long long)o.sent.resent,
+          (unsigned long long)o.sent.lost);
+    /* Loss, packets out of order and doubled, failed saves, a pause long
+     * enough for the receiver to let the sender's state go, and both sides
+     * starting again: the receiver remembers its secret, so it wakes the
+     * sender without an X25519. */
+    for (uint64_t s = 1; s <= 4; ++s) {
+        Case rough = {"rough", 3000, s * 7919, 10, 3, 20, 37, 1500, 800, 2200, 300};
+        o = run(&rough, dir);
+        CHECK(o.sent.resent > 0 && o.heard.swept >= 1 && o.heard.woken >= 1 && o.heard.remembered >= 1,
+              "rough: sent again %llu, swept %llu, woken %llu, remembered %llu", (unsigned long long)o.sent.resent,
+              (unsigned long long)o.heard.swept, (unsigned long long)o.heard.woken,
+              (unsigned long long)o.heard.remembered);
+    }
+    char command[600];
+    snprintf(command, sizeof(command), "rm -rf %s", dir);
+    if (system(command)) { /* left behind in /tmp */ }
+    if (failures) fprintf(stderr, "test-link: %d failed\n", failures);
+    return failures ? 1 : 0;
+}
diff --git a/research/experiments/019-idle-links/transport.c b/research/experiments/019-idle-links/transport.c
index c4c9de1..5827311 100644
--- a/research/experiments/019-idle-links/transport.c
+++ b/research/experiments/019-idle-links/transport.c
@@ -138,9 +138,9 @@ int main(int argc, char **argv) {
         return 1;
     unsigned port = ntohs(any.sin_port);
     close(probe);
-    if (!(listener = pure_link_listen(port, b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error))) ||
-        !pure_link_allow(listener, "bench.ordered", public_a, error, sizeof(error)) ||
-        !pure_link_serve(listener, error, sizeof(error)))
+    PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){input_of, deliver, NULL}, error, sizeof(error));
+    if (!r || !pure_link_allow(r, "bench.ordered", public_a, error, sizeof(error)) ||
+        !(listener = pure_link_listen(r, port, error, sizeof(error))) || !pure_link_serve(listener, error, sizeof(error)))
         return fprintf(stderr, "transport: %s\n", error), 1;
     char url[128];
     snprintf(url, sizeof(url), "pure://127.0.0.1:%u/order", port);
diff --git a/serve/serve.c b/serve/serve.c
index 5688d79..f466ee6 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -2493,7 +2493,7 @@ static bool name_senders(const char *const *named, unsigned count) {
 
 /* --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(PureLinkListener *listener, const char *path, char *error, size_t size) {
+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) {
@@ -2516,7 +2516,7 @@ static bool allow_senders(PureLinkListener *listener, const char *path, char *er
         *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(listener, line, key, error, size);
+        else ok = pure_link_allow(receiver, line, key, error, size);
     }
     free(line);
     fclose(file);
@@ -2675,15 +2675,16 @@ int main(int argc, char **argv) {
             fputs("--link-port needs an app with states and the node's --key\n", stderr);
             return 2;
         }
-        link_listener = pure_link_listen(link_port, seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner},
-                                         error, sizeof(error));
-        bool allowed = link_listener != NULL && allow_senders(link_listener, senders_path, error, sizeof(error));
+        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(link_listener, senders[i].name, senders[i].key, error, sizeof(error));
-        if (allowed && link_idle_ms) pure_link_idle(link_listener, link_idle_ms);
+            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(link_listener, memo, error, sizeof(error));
+        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;