Measure a link at rest: 132 bytes against a QUIC connection's 38,106 (094)

M1 of design 011. A Pure node allows 1,000 to 1,000,000 senders; each
sends one order and, once told it is stored, rests. At rest a link costs
the node 132 bytes (the sender allowed, 73; its number heard, about 60)
and no transport state; with a million the node runs in 139 MB. A quinn
0.11 service holding the same exchange's clients connected at rest costs
38,106 bytes a connection: 289 times as much, past the design's 1/100.
A link at rest takes its next record in 0.56 ms to stored, one X25519 on
the node included; a new QUIC connection takes 1.06 ms and 2.8 times the
receiver's CPU, without storing.

research/experiments/019-idle-links: links.c, many senders written from
serve/link.h's packet format alone (a second implementation), and the
quinn service and clients (quic/, built offline); run.py makes both
measurements.

1 files changed, 624 insertions, 0 deletions

diff --git a/research/experiments/019-idle-links/links.c b/research/experiments/019-idle-links/links.c
new file mode 100644
index 0000000..732429a
--- /dev/null
+++ b/research/experiments/019-idle-links/links.c
@@ -0,0 +1,624 @@
+#define _GNU_SOURCE
+/* Many senders to one Pure node over links without connections
+ * (docs/design/011-pure-transport.md, M1). Written from the packet format in
+ * serve/link.h alone, with OpenSSL: a second implementation of it.
+ *
+ *   links keys N FILE
+ *       N X25519 keys into FILE (each a private then a public half, 32 bytes
+ *       each) and FILE.senders, a line each for pure-serve --senders:
+ *       dNNNNNNN KEY
+ *   links send FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT SEQ [WINDOW]
+ *       devices FIRST to FIRST + COUNT - 1 each send record SEQ (an order)
+ *       to INPUT (whose shape MODULE says) and wait until the node says it
+ *       stored it, WINDOW devices at a time, each on a socket of its own;
+ *       then a line of JSON: how many, how long, and the time from each
+ *       device's first packet to its record stored.
+ *   links run FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT PERIOD_MS SECONDS
+ *       devices FIRST.. each make a record every PERIOD_MS for SECONDS and
+ *       send what is not stored (again after 200 ms, doubling to a second,
+ *       without an answer), however the node comes and goes; then wait until
+ *       all is stored and write a line of JSON: records made and stored,
+ *       the records stored in each 100 ms, and the time from a record made
+ *       to it stored. */
+#include "module.h"
+#include <arpa/inet.h>
+#include <errno.h>
+#include <inttypes.h>
+#include <math.h>
+#include <netdb.h>
+#include <openssl/core_names.h>
+#include <openssl/evp.h>
+#include <openssl/kdf.h>
+#include <poll.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/epoll.h>
+#include <sys/random.h>
+#include <sys/socket.h>
+#include <time.h>
+#include <unistd.h>
+
+enum { KEY = 32, HEADER = 21, TAG = 16, DATAGRAM = 1200, NONCE = 12 };
+
+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 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; }
+
+static uint64_t now_ns(void) {
+    struct timespec t;
+    clock_gettime(CLOCK_MONOTONIC, &t);
+    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 void fail(const char *what) {
+    fprintf(stderr, "links: %s\n", what);
+    exit(1);
+}
+
+/* ---- Keys */
+
+static void make_key(unsigned char out[2 * KEY]) {
+    EVP_PKEY *key = EVP_PKEY_Q_keygen(NULL, NULL, "X25519");
+    size_t a = KEY, b = KEY;
+    if (!key || EVP_PKEY_get_raw_private_key(key, out, &a) != 1 || EVP_PKEY_get_raw_public_key(key, out + KEY, &b) != 1)
+        fail("cannot make a key");
+    EVP_PKEY_free(key);
+}
+
+typedef struct { unsigned char *keys; size_t first, count; } Making;
+
+static void *make_keys(void *context) {
+    Making *m = context;
+    for (size_t i = 0; i < m->count; ++i) make_key(m->keys + (m->first + i) * 2 * KEY);
+    return NULL;
+}
+
+static int keys(size_t n, const char *path) {
+    unsigned char *all = malloc(n * 2 * KEY);
+    if (!all) fail("out of memory");
+    pthread_t threads[4];
+    Making parts[4];
+    for (int t = 0; t < 4; ++t) {
+        parts[t] = (Making){all, n * t / 4, n * (t + 1) / 4 - n * t / 4};
+        pthread_create(&threads[t], NULL, make_keys, &parts[t]);
+    }
+    for (int t = 0; t < 4; ++t) pthread_join(threads[t], NULL);
+    FILE *file = fopen(path, "wb");
+    if (!file || fwrite(all, 2 * KEY, n, file) != n || fclose(file)) fail("cannot write the keys");
+    char senders[4096];
+    snprintf(senders, sizeof(senders), "%s.senders", path);
+    FILE *lines = fopen(senders, "w");
+    if (!lines) fail("cannot write the senders");
+    for (size_t i = 0; i < n; ++i) {
+        fprintf(lines, "d%07zu ", i);
+        for (int b = 0; b < KEY; ++b) fprintf(lines, "%02x", all[i * 2 * KEY + KEY + b]);
+        fputc('\n', lines);
+    }
+    if (fclose(lines)) fail("cannot write the senders");
+    free(all);
+    return 0;
+}
+
+/* ---- Keys of a link (serve/link.h): X25519, HKDF-SHA256, AES-128-GCM */
+
+static bool secret_of(const unsigned char own[2 * KEY], const unsigned char peer[KEY], unsigned char out[32]) {
+    EVP_PKEY *mine = EVP_PKEY_new_raw_private_key(EVP_PKEY_X25519, NULL, own, KEY);
+    EVP_PKEY *theirs = EVP_PKEY_new_raw_public_key(EVP_PKEY_X25519, NULL, peer, KEY);
+    EVP_PKEY_CTX *c = mine && theirs ? EVP_PKEY_CTX_new(mine, NULL) : NULL;
+    size_t length = 32;
+    bool ok = c && EVP_PKEY_derive_init(c) == 1 && EVP_PKEY_derive_set_peer(c, theirs) == 1 &&
+              EVP_PKEY_derive(c, out, &length) == 1 && length == 32;
+    EVP_PKEY_CTX_free(c);
+    EVP_PKEY_free(mine);
+    EVP_PKEY_free(theirs);
+    return ok;
+}
+
+static EVP_KDF *hkdf;
+
+typedef struct { EVP_CIPHER_CTX *seal, *open; unsigned char base[NONCE]; } Aead;
+
+static bool derive(const unsigned char secret[32], const unsigned char from[KEY], const unsigned char to[KEY],
+                   uint64_t from_run, uint64_t to_run, Aead *out) {
+    unsigned char info[16 + 2 * KEY + 16], key[16 + NONCE];
+    memcpy(info, "pure link 0 keys", 16);
+    memcpy(info + 16, from, KEY);
+    memcpy(info + 16 + KEY, to, KEY);
+    put64(info + 16 + 2 * KEY, from_run);
+    put64(info + 24 + 2 * KEY, to_run);
+    static const char salt[] = "pure link 0";
+    EVP_KDF_CTX *c = EVP_KDF_CTX_new(hkdf);
+    OSSL_PARAM params[] = {OSSL_PARAM_construct_utf8_string(OSSL_KDF_PARAM_DIGEST, (char *)"SHA256", 0),
+                           OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_KEY, (void *)secret, 32),
+                           OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_SALT, (void *)salt, sizeof(salt) - 1),
+                           OSSL_PARAM_construct_octet_string(OSSL_KDF_PARAM_INFO, info, sizeof(info)),
+                           OSSL_PARAM_construct_end()};
+    bool ok = c && EVP_KDF_derive(c, key, sizeof(key), params) == 1;
+    EVP_KDF_CTX_free(c);
+    if (!ok) return false;
+    memcpy(out->base, key + 16, NONCE);
+    out->seal = EVP_CIPHER_CTX_new();
+    out->open = EVP_CIPHER_CTX_new();
+    return out->seal && out->open && EVP_EncryptInit_ex(out->seal, EVP_aes_128_gcm(), NULL, key, NULL) == 1 &&
+           EVP_DecryptInit_ex(out->open, EVP_aes_128_gcm(), NULL, key, NULL) == 1;
+}
+
+static void aead_free(Aead *a) {
+    EVP_CIPHER_CTX_free(a->seal);
+    EVP_CIPHER_CTX_free(a->open);
+    *a = (Aead){0};
+}
+
+static void nonce_of(unsigned char out[NONCE], const unsigned char base[NONCE], uint64_t pn) {
+    memcpy(out, base, NONCE);
+    for (int i = 0; i < 8; ++i) out[4 + i] ^= (unsigned char)(pn >> (56 - 8 * i));
+}
+
+static size_t seal(Aead *a, uint32_t id, uint64_t run, uint64_t pn, const unsigned char *body, size_t size,
+                   unsigned char *out) {
+    unsigned char nonce[NONCE];
+    int n = 0;
+    out[0] = 0x70;
+    put32(out + 1, id);
+    put64(out + 5, run);
+    put64(out + 13, pn);
+    nonce_of(nonce, a->base, pn);
+    if (EVP_EncryptInit_ex(a->seal, NULL, NULL, NULL, nonce) != 1 || EVP_EncryptUpdate(a->seal, NULL, &n, out, HEADER) != 1 ||
+        EVP_EncryptUpdate(a->seal, out + HEADER, &n, body, (int)size) != 1 ||
+        EVP_EncryptFinal_ex(a->seal, out + HEADER + n, &n) != 1 ||
+        EVP_CIPHER_CTX_ctrl(a->seal, EVP_CTRL_GCM_GET_TAG, TAG, out + HEADER + size) != 1)
+        return 0;
+    return HEADER + size + TAG;
+}
+
+static bool open_packet(Aead *a, const unsigned char *packet, size_t n, unsigned char *plain) {
+    unsigned char nonce[NONCE], tag[TAG];
+    int length = 0, rest = 0;
+    size_t size = n - HEADER - TAG;
+    memcpy(tag, packet + n - TAG, TAG);
+    nonce_of(nonce, a->base, get64(packet + 13));
+    return EVP_DecryptInit_ex(a->open, NULL, NULL, NULL, nonce) == 1 &&
+           EVP_DecryptUpdate(a->open, NULL, &length, packet, HEADER) == 1 &&
+           EVP_DecryptUpdate(a->open, plain, &length, packet + HEADER, (int)size) == 1 &&
+           EVP_CIPHER_CTX_ctrl(a->open, EVP_CTRL_GCM_SET_TAG, TAG, tag) == 1 &&
+           EVP_DecryptFinal_ex(a->open, plain + length, &rest) == 1;
+}
+
+/* ---- What is sent: an order as the receiving input stores it */
+
+static PureModule module;
+
+static uint32_t shape_of(const PureModule *m, uint32_t input) {
+    const PureInputShape *s = &m->input[input];
+    uint32_t h = fnv(2166136261u, &s->prefix, sizeof(s->prefix));
+    for (uint32_t f = 0; f < s->field_count; ++f) {
+        const PureField *field = &m->field[s->field_start + f];
+        unsigned char fixed[7] = {field->type, (unsigned char)field->rows, (unsigned char)(field->rows >> 8)};
+        put32(fixed + 3, field->offset);
+        h = fnv(fnv(h, field->name, strlen(field->name) + 1), fixed, sizeof(fixed));
+    }
+    return h;
+}
+
+/* A message of input: each u32 field 1, each number 0, each text or bytes "pen". */
+static size_t order(const PureModule *m, uint32_t input, unsigned char *out) {
+    const PureInputShape *s = &m->input[input];
+    memset(out, 0, s->prefix);
+    size_t at = s->prefix;
+    for (uint32_t f = 0; f < s->field_count; ++f) {
+        const PureField *field = &m->field[s->field_start + f];
+        if (field->type == PURE_FIELD_U32) put32(out + field->offset, 1);
+        else if (field->type == PURE_FIELD_TEXT || field->type == PURE_FIELD_BYTES) {
+            if (field->offset != PURE_FIELD_REST) put32(out + field->offset, 3);
+            memcpy(out + at, "pen", 3);
+            at += 3;
+        }
+    }
+    return at;
+}
+
+/* ---- Devices */
+
+typedef struct {
+    size_t device;
+    int fd;
+    uint64_t run, pn, started, sent, in_run;
+    Aead out, in;
+    unsigned char secret[32];
+    bool busy, in_ready;
+    unsigned tries;
+} Slot;
+
+static unsigned char *all_keys, node_key[KEY], payload[512];
+static size_t payload_size;
+static uint32_t shape, node_id;
+static const char *input_name;
+static uint64_t seq;
+static double *waited;
+static size_t waited_count, resent, failed;
+
+static uint64_t random64(void) {
+    uint64_t v = 0;
+    while (getrandom(&v, sizeof(v), 0) != sizeof(v) || !v) {}
+    return v;
+}
+
+static void send_record(Slot *s) {
+    unsigned char body[DATAGRAM], packet[DATAGRAM];
+    char source[16];
+    snprintf(source, sizeof(source), "d%07zu", s->device);
+    size_t at = 0, a = strlen(source), b = strlen(input_name);
+    body[at++] = 1; /* STREAM */
+    body[at++] = (unsigned char)a;
+    memcpy(body + at, source, a);
+    at += a;
+    body[at++] = (unsigned char)b;
+    memcpy(body + at, input_name, b);
+    at += b;
+    put32(body + at, shape);
+    at += 4;
+    body[at++] = 2; /* RECORD */
+    put64(body + at, seq);
+    put32(body + at + 8, (uint32_t)payload_size);
+    put32(body + at + 12, 0);
+    put16(body + at + 16, (uint32_t)payload_size);
+    memcpy(body + at + 18, payload, payload_size);
+    at += 18 + payload_size;
+    const unsigned char *own = all_keys + s->device * 2 * KEY;
+    size_t n = seal(&s->out, fnv(2166136261u, own + KEY, KEY), s->run, s->pn++, body, at, packet);
+    if (!n || send(s->fd, packet, n, 0) < 0) { /* sent again at its time */ }
+    s->sent = now_ns();
+}
+
+static void start_device(Slot *s, size_t device) {
+    const unsigned char *own = all_keys + device * 2 * KEY;
+    s->device = device;
+    s->run = random64();
+    s->pn = 0;
+    s->tries = 0;
+    s->busy = true;
+    s->in_ready = false;
+    s->started = now_ns();
+    if (!secret_of(own, node_key, s->secret) || !derive(s->secret, own + KEY, node_key, s->run, 0, &s->out))
+        fail("cannot make a device's keys");
+    send_record(s);
+}
+
+/* An answer: done once it says the record is stored. */
+static bool take_answer(Slot *s, const unsigned char *packet, size_t n) {
+    unsigned char plain[DATAGRAM];
+    if (n < HEADER + TAG || packet[0] != 0x70 || get32(packet + 1) != node_id) return false;
+    const unsigned char *own = all_keys + s->device * 2 * KEY;
+    uint64_t run = get64(packet + 5);
+    if (!s->in_ready || s->in_run != run) {
+        aead_free(&s->in);
+        if (!derive(s->secret, node_key, own + KEY, run, s->run, &s->in)) return false;
+        s->in_ready = true;
+        s->in_run = run;
+    }
+    if (!open_packet(&s->in, packet, n, plain)) return false;
+    size_t size = n - HEADER - TAG;
+    for (size_t at = 0; at < size;) {
+        unsigned type = plain[at++];
+        if (type == 3) { /* ACK */
+            if (at + 13 > size) return false;
+            at += 13 + 16u * plain[at + 12];
+        } else if (type == 4) { /* STORED */
+            if (at + 1 > size) return false;
+            size_t a = plain[at], b = at + 1 + a < size ? plain[at + 1 + a] : 0;
+            if (at + 2 + a + b + 13 > size) return false;
+            uint64_t heard = get64(plain + at + 2 + a + b);
+            at += 2 + a + b + 13;
+            if (heard >= seq) return true;
+        } else
+            return false;
+    }
+    return false;
+}
+
+static int by_value(const void *a, const void *b) {
+    double x = *(const double *)a, y = *(const double *)b;
+    return x < y ? -1 : x > y;
+}
+
+static int send_all(int argc, char **argv) {
+    const char *path = argv[2], *module_path = argv[3];
+    input_name = argv[4];
+    size_t first = strtoull(argv[8], NULL, 10), count = strtoull(argv[9], NULL, 10), window = argc > 11 ? strtoull(argv[11], NULL, 10) : 256;
+    seq = strtoull(argv[10], NULL, 10);
+    char error[256];
+    if (pure_module_read(module_path, &module, error, sizeof(error)) != PURE_OK) fail(error);
+    int input = pure_module_input(&module, input_name);
+    if (input < 0) fail("no such input");
+    shape = shape_of(&module, (uint32_t)input);
+    payload_size = order(&module, (uint32_t)input, payload);
+    for (int i = 0; i < KEY; ++i) sscanf(argv[7] + 2 * i, "%2hhx", &node_key[i]);
+    node_id = fnv(2166136261u, node_key, KEY);
+    FILE *file = fopen(path, "rb");
+    if (!file) fail("cannot read the keys");
+    fseek(file, 0, SEEK_END);
+    size_t n = (size_t)ftell(file) / (2 * KEY);
+    if (first + count > n) fail("not so many keys");
+    all_keys = malloc(n * 2 * KEY);
+    fseek(file, 0, SEEK_SET);
+    if (!all_keys || fread(all_keys, 2 * KEY, n, file) != n) fail("cannot read the keys");
+    fclose(file);
+    hkdf = EVP_KDF_fetch(NULL, "HKDF", NULL);
+    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found;
+    if (getaddrinfo(argv[5], argv[6], &hints, &found)) fail("no such host");
+    if (window > count) window = count;
+    Slot *slots = calloc(window, sizeof(*slots));
+    waited = malloc(count * sizeof(*waited));
+    int poller = epoll_create1(0);
+    for (size_t k = 0; k < window; ++k) {
+        slots[k].fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+        if (slots[k].fd < 0 || connect(slots[k].fd, found->ai_addr, found->ai_addrlen)) fail("no socket");
+        struct epoll_event e = {.events = EPOLLIN, .data.u64 = k};
+        epoll_ctl(poller, EPOLL_CTL_ADD, slots[k].fd, &e);
+    }
+    uint64_t began = now_ns();
+    size_t next = 0, done = 0;
+    for (size_t k = 0; k < window; ++k) start_device(&slots[k], first + next++);
+    struct epoll_event events[256];
+    while (done < count) {
+        int ready = epoll_wait(poller, events, 256, 20);
+        uint64_t now = now_ns();
+        for (int e = 0; e < ready; ++e) {
+            Slot *s = &slots[events[e].data.u64];
+            unsigned char packet[DATAGRAM + 64];
+            for (ssize_t got; s->busy && (got = recv(s->fd, packet, sizeof(packet), 0)) > 0;) {
+                if (!take_answer(s, packet, (size_t)got)) continue;
+                waited[waited_count++] = (double)(now_ns() - s->started) / 1e6;
+                aead_free(&s->out);
+                aead_free(&s->in);
+                s->busy = false;
+                ++done;
+                if (next < count) start_device(s, first + next++);
+            }
+        }
+        for (size_t k = 0; k < window; ++k) { /* no answer: sent again, as a link does */
+            Slot *s = &slots[k];
+            if (!s->busy || now < s->sent + 200000000u) continue;
+            if (++s->tries > 50) {
+                ++failed, ++done;
+                aead_free(&s->out);
+                aead_free(&s->in);
+                s->busy = false;
+                if (next < count) start_device(s, first + next++);
+                continue;
+            }
+            ++resent;
+            send_record(s);
+        }
+    }
+    double seconds = (double)(now_ns() - began) / 1e9;
+    qsort(waited, waited_count, sizeof(*waited), by_value);
+    double p50 = waited_count ? waited[waited_count / 2] : 0, p99 = waited_count ? waited[waited_count * 99 / 100] : 0;
+    printf("{\"devices\":%zu,\"stored\":%zu,\"failed\":%zu,\"resent\":%zu,\"seconds\":%.3f,\"per_second\":%.0f,"
+           "\"stored_ms_p50\":%.3f,\"stored_ms_p99\":%.3f,\"stored_ms_max\":%.3f}\n",
+           count, waited_count, failed, resent, seconds, (double)count / seconds, p50, p99,
+           waited_count ? waited[waited_count - 1] : 0);
+    freeaddrinfo(found);
+    return failed ? 1 : 0;
+}
+
+/* ---- Devices that keep sending: an outbox each */
+
+enum { OUTBOX = 64, BATCH = 8 };
+typedef struct {
+    size_t device;
+    int fd;
+    uint64_t run, pn, in_run, made, stored; /* records made and stored: numbers 1..made, 1..stored */
+    uint64_t made_at[OUTBOX];                /* when each record not stored was made (ns), by number % OUTBOX */
+    uint64_t next_make, sent, wait;          /* when the next record is made; when last sent; the wait for an answer */
+    Aead out, in;
+    unsigned char secret[32];
+    bool in_ready;
+} Device;
+
+static void send_outbox(Device *d, uint64_t now) {
+    unsigned char body[DATAGRAM], packet[DATAGRAM];
+    char source[16];
+    snprintf(source, sizeof(source), "d%07zu", d->device);
+    size_t at = 0, a = strlen(source), b = strlen(input_name);
+    body[at++] = 1; /* STREAM */
+    body[at++] = (unsigned char)a;
+    memcpy(body + at, source, a);
+    at += a;
+    body[at++] = (unsigned char)b;
+    memcpy(body + at, input_name, b);
+    at += b;
+    put32(body + at, shape);
+    at += 4;
+    for (uint64_t seq = d->stored + 1; seq <= d->made && seq <= d->stored + BATCH; ++seq) {
+        body[at++] = 2; /* RECORD */
+        put64(body + at, seq);
+        put32(body + at + 8, (uint32_t)payload_size);
+        put32(body + at + 12, 0);
+        put16(body + at + 16, (uint32_t)payload_size);
+        memcpy(body + at + 18, payload, payload_size);
+        at += 18 + payload_size;
+    }
+    const unsigned char *own = all_keys + d->device * 2 * KEY;
+    size_t n = seal(&d->out, fnv(2166136261u, own + KEY, KEY), d->run, d->pn++, body, at, packet);
+    if (!n || send(d->fd, packet, n, 0) < 0) { /* sent again at its time */ }
+    d->sent = now;
+}
+
+/* An answer: how far the node stored (0 when it says nothing of it). */
+static uint64_t stored_in(Device *d, const unsigned char *packet, size_t n) {
+    unsigned char plain[DATAGRAM];
+    if (n < HEADER + TAG || packet[0] != 0x70 || get32(packet + 1) != node_id) return 0;
+    const unsigned char *own = all_keys + d->device * 2 * KEY;
+    uint64_t run = get64(packet + 5), heard = 0;
+    if (!d->in_ready || d->in_run != run) { /* the node started again: its answers come in another run */
+        Aead in = {0};
+        if (!derive(d->secret, node_key, own + KEY, run, d->run, &in)) return 0;
+        if (!open_packet(&in, packet, n, plain)) {
+            aead_free(&in);
+            return 0;
+        }
+        aead_free(&d->in);
+        d->in = in;
+        d->in_ready = true;
+        d->in_run = run;
+    } else if (!open_packet(&d->in, packet, n, plain))
+        return 0;
+    size_t size = n - HEADER - TAG;
+    for (size_t at = 0; at < size;) {
+        unsigned type = plain[at++];
+        if (type == 3) {
+            if (at + 13 > size) break;
+            at += 13 + 16u * plain[at + 12];
+        } else if (type == 4) {
+            if (at + 1 > size) break;
+            size_t a = plain[at], b = at + 1 + a < size ? plain[at + 1 + a] : 0;
+            if (at + 2 + a + b + 13 > size) break;
+            uint64_t said = get64(plain + at + 2 + a + b);
+            if (said > heard) heard = said;
+            at += 2 + a + b + 13;
+        } else
+            break;
+    }
+    return heard;
+}
+
+static int run_all(char **argv) {
+    const char *path = argv[2], *module_path = argv[3];
+    input_name = argv[4];
+    size_t first = strtoull(argv[8], NULL, 10), count = strtoull(argv[9], NULL, 10);
+    uint64_t period = strtoull(argv[10], NULL, 10) * 1000000u, seconds = strtoull(argv[11], NULL, 10);
+    char error[256];
+    if (pure_module_read(module_path, &module, error, sizeof(error)) != PURE_OK) fail(error);
+    int input = pure_module_input(&module, input_name);
+    if (input < 0) fail("no such input");
+    shape = shape_of(&module, (uint32_t)input);
+    payload_size = order(&module, (uint32_t)input, payload);
+    for (int i = 0; i < KEY; ++i) sscanf(argv[7] + 2 * i, "%2hhx", &node_key[i]);
+    node_id = fnv(2166136261u, node_key, KEY);
+    FILE *file = fopen(path, "rb");
+    if (!file) fail("cannot read the keys");
+    fseek(file, 0, SEEK_END);
+    size_t n = (size_t)ftell(file) / (2 * KEY);
+    if (first + count > n) fail("not so many keys");
+    all_keys = malloc(n * 2 * KEY);
+    fseek(file, 0, SEEK_SET);
+    if (!all_keys || fread(all_keys, 2 * KEY, n, file) != n) fail("cannot read the keys");
+    fclose(file);
+    hkdf = EVP_KDF_fetch(NULL, "HKDF", NULL);
+    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found;
+    if (getaddrinfo(argv[5], argv[6], &hints, &found)) fail("no such host");
+    Device *devices = calloc(count, sizeof(*devices));
+    int poller = epoll_create1(0);
+    uint64_t began = now_ns(), end = began + seconds * 1000000000u;
+    for (size_t k = 0; k < count; ++k) {
+        Device *d = &devices[k];
+        const unsigned char *own = all_keys + (first + k) * 2 * KEY;
+        d->device = first + k;
+        d->run = random64();
+        d->fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+        if (d->fd < 0 || connect(d->fd, found->ai_addr, found->ai_addrlen)) fail("no socket");
+        if (!secret_of(own, node_key, d->secret) || !derive(d->secret, own + KEY, node_key, d->run, 0, &d->out))
+            fail("cannot make a device's keys");
+        d->next_make = began + period * k / count; /* spread over the period */
+        d->wait = 200000000u;
+        struct epoll_event e = {.events = EPOLLIN, .data.u64 = k};
+        epoll_ctl(poller, EPOLL_CTL_ADD, d->fd, &e);
+    }
+    began = now_ns(); /* the keys took their time: the run starts now */
+    end = began + seconds * 1000000000u;
+    for (size_t k = 0; k < count; ++k) devices[k].next_make = began + period * k / count;
+    size_t buckets = (size_t)(seconds + 60) * 10;
+    uint32_t *timeline = calloc(buckets, sizeof(*timeline));
+    size_t latency_capacity = 1024, latency_count = 0;
+    double *latency = malloc(latency_capacity * sizeof(*latency));
+    uint64_t made = 0, stored = 0, sends = 0, again = 0, last_report = began, last_scan = 0;
+    struct epoll_event events[256];
+    for (;;) {
+        uint64_t now = now_ns();
+        bool making = now < end;
+        if (!making && stored == made) break;
+        if (now > end + 60000000000u) break; /* a minute past the end: what is left was not stored */
+        int ready = epoll_wait(poller, events, 256, 5);
+        now = now_ns();
+        for (int e = 0; e < ready; ++e) {
+            Device *d = &devices[events[e].data.u64];
+            unsigned char packet[DATAGRAM + 64];
+            for (ssize_t got; (got = recv(d->fd, packet, sizeof(packet), 0)) > 0;) {
+                uint64_t heard = stored_in(d, packet, (size_t)got);
+                if (heard > d->made) heard = d->made;
+                for (uint64_t seq = d->stored + 1; seq <= heard; ++seq) {
+                    size_t bucket = (size_t)((now - began) / 100000000u);
+                    if (bucket < buckets) ++timeline[bucket];
+                    if (latency_count == latency_capacity) latency = realloc(latency, (latency_capacity *= 2) * sizeof(*latency));
+                    latency[latency_count++] = (double)(now - d->made_at[seq % OUTBOX]) / 1e6;
+                    ++stored;
+                }
+                if (heard > d->stored) {
+                    d->stored = heard;
+                    d->wait = 200000000u;
+                }
+            }
+        }
+        if (now - last_scan < 5000000u) continue; /* the devices' clocks, every 5 ms */
+        last_scan = now;
+        for (size_t k = 0; k < count; ++k) {
+            Device *d = &devices[k];
+            if (making && now >= d->next_make && d->made - d->stored < OUTBOX) {
+                d->made_at[++d->made % OUTBOX] = now;
+                ++made;
+                d->next_make += period;
+                send_outbox(d, now);
+                ++sends;
+            } else if (d->stored < d->made && now >= d->sent + d->wait) { /* no answer: again, waiting longer */
+                send_outbox(d, now);
+                ++sends, ++again;
+                d->wait = d->wait * 2 > 1000000000u ? 1000000000u : d->wait * 2;
+            }
+        }
+        if (now - last_report >= 1000000000u) {
+            last_report = now;
+            fprintf(stderr, "{\"t\":%.1f,\"made\":%" PRIu64 ",\"stored\":%" PRIu64 "}\n", (double)(now - began) / 1e9, made, stored);
+        }
+    }
+    qsort(latency, latency_count, sizeof(*latency), by_value);
+    /* The longest time without a record stored, from the first to the last one stored. */
+    size_t last = 0, longest = 0, gap = 0, since = SIZE_MAX;
+    for (size_t b = 0; b < buckets; ++b)
+        if (timeline[b]) {
+            if (since != SIZE_MAX && b - since - 1 > longest) longest = b - since - 1, gap = since + 1;
+            since = b;
+            last = b;
+        }
+    printf("{\"devices\":%zu,\"made\":%" PRIu64 ",\"stored\":%" PRIu64 ",\"sends\":%" PRIu64 ",\"again\":%" PRIu64
+           ",\"stored_ms_p50\":%.3f,\"stored_ms_p99\":%.3f,\"stored_ms_max\":%.3f,\"longest_gap_ms\":%zu,\"gap_from_ms\":%zu,"
+           "\"timeline_100ms\":[",
+           count, made, stored, sends, again, latency_count ? latency[latency_count / 2] : 0,
+           latency_count ? latency[latency_count * 99 / 100] : 0, latency_count ? latency[latency_count - 1] : 0,
+           longest * 100, gap * 100);
+    for (size_t b = 0; b <= last; ++b) printf("%s%u", b ? "," : "", timeline[b]);
+    printf("]}\n");
+    freeaddrinfo(found);
+    return stored == made ? 0 : 1;
+}
+
+int main(int argc, char **argv) {
+    if (argc == 4 && !strcmp(argv[1], "keys")) return keys(strtoull(argv[2], NULL, 10), argv[3]);
+    if ((argc == 11 || argc == 12) && !strcmp(argv[1], "send")) return send_all(argc, argv);
+    if (argc == 12 && !strcmp(argv[1], "run")) return run_all(argv);
+    fputs("links keys N FILE | links send FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT SEQ [WINDOW]\n"
+          "      | links run FILE MODULE INPUT HOST PORT NODE_KEY FIRST COUNT PERIOD_MS SECONDS\n", stderr);
+    return 2;
+}