Deliver wires on the sending loop, a step at a time (017, 2b-2)

A wire is now a sender of link/sending.h: its request goes on a
non-blocking socket as far as the socket takes it (connect, TLS
handshake, write, answer read), and its round waits for the socket or
its deadline (10 s with nothing moved). The answer is read as it comes:
the head whole, then a sized body counted down, a chunked one (few and
small) parsed again as it grows, or a body to the close. The crypto
table (version 6) takes the TLS handshake a step at a time, and reads
and writes say what they wait for. The owner pushes a wire once a turn,
as it does a link. A host whose app sends to ten links and ten wires
has 5 threads where it had 24 (none: 3).

7 files changed, 512 insertions, 291 deletions

diff --git a/link/crypto-openssl.c b/link/crypto-openssl.c
index 96daa82..16651a2 100644
--- a/link/crypto-openssl.c
+++ b/link/crypto-openssl.c
@@ -7,6 +7,7 @@
 #include <errno.h>
 #include <fcntl.h>
 #include <openssl/crypto.h>
+#include <openssl/err.h>
 #include <openssl/evp.h>
 #include <openssl/core_names.h>
 #include <openssl/hpke.h>
@@ -34,13 +35,13 @@ static bool tls_start(const char *ca_file) {
         return false;
     }
     SSL_CTX_set_verify(context, SSL_VERIFY_PEER, NULL);
-    SSL_CTX_set_mode(context, SSL_MODE_AUTO_RETRY);
+    SSL_CTX_set_mode(context, SSL_MODE_ENABLE_PARTIAL_WRITE | SSL_MODE_ACCEPT_MOVING_WRITE_BUFFER);
     tls_context = context;
     return true;
 }
 
-/* TLS over a connected socket: the receiver's certificate must name the host
- * (a DNS name, or an IP address written as one). */
+/* TLS over a connected, non-blocking socket: the receiver's certificate
+ * must name the host (a DNS name, or an IP address written as one). */
 static PureTls *tls_open(const char *host, int fd) {
     SSL *ssl = SSL_new(tls_context);
     if (!ssl) return NULL;
@@ -48,15 +49,44 @@ static PureTls *tls_open(const char *host, int fd) {
     bool literal = inet_pton(AF_INET, host, address) == 1 || inet_pton(AF_INET6, host, address) == 1;
     bool named = literal ? X509_VERIFY_PARAM_set1_ip_asc(SSL_get0_param(ssl), host) == 1 :
                            SSL_set_tlsext_host_name(ssl, host) == 1 && SSL_set1_host(ssl, host) == 1;
-    if (!named || SSL_set_fd(ssl, fd) != 1 || SSL_connect(ssl) != 1 || SSL_get_verify_result(ssl) != X509_V_OK) {
+    if (!named || SSL_set_fd(ssl, fd) != 1) {
         SSL_free(ssl);
         return NULL;
     }
+    SSL_set_connect_state(ssl);
     return (PureTls *)ssl;
 }
 
-static int tls_read(PureTls *tls, void *bytes, int size) { return SSL_read((SSL *)tls, bytes, size); }
-static int tls_write(PureTls *tls, const void *bytes, int size) { return SSL_write((SSL *)tls, bytes, size); }
+/* What a call that moved nothing waits for (link/tls.h). */
+static int wants(SSL *ssl, int result) {
+    switch (SSL_get_error(ssl, result)) {
+    case SSL_ERROR_WANT_READ: return PURE_TLS_WANTS_READ;
+    case SSL_ERROR_WANT_WRITE: return PURE_TLS_WANTS_WRITE;
+    case SSL_ERROR_ZERO_RETURN: return PURE_TLS_CLOSED;
+    default: return PURE_TLS_FAILED;
+    }
+}
+
+static int tls_step(PureTls *tls) {
+    SSL *ssl = (SSL *)tls;
+    ERR_clear_error(); /* SSL_get_error reads the thread's queue */
+    int result = SSL_do_handshake(ssl);
+    if (result == 1) return SSL_get_verify_result(ssl) == X509_V_OK ? 1 : PURE_TLS_FAILED;
+    int waits = wants(ssl, result);
+    return waits == PURE_TLS_CLOSED ? PURE_TLS_FAILED : waits;
+}
+
+static int tls_read(PureTls *tls, void *bytes, int size) {
+    ERR_clear_error();
+    int n = SSL_read((SSL *)tls, bytes, size);
+    return n > 0 ? n : wants((SSL *)tls, n);
+}
+
+static int tls_write(PureTls *tls, const void *bytes, int size) {
+    ERR_clear_error();
+    int n = SSL_write((SSL *)tls, bytes, size);
+    return n > 0 ? n : wants((SSL *)tls, n);
+}
 static void tls_close(PureTls *tls) { SSL_free((SSL *)tls); }
 
 /* ---- Sealing: HPKE (RFC 9180), X25519 and HKDF-SHA256, with the AEAD the
@@ -305,7 +335,7 @@ static bool aead_open(PureAead *a, const unsigned char *key, const unsigned char
            EVP_DecryptFinal_ex(a->open, plain + body, &last) == 1;
 }
 
-const PureCrypto pure_crypto = {PURE_CRYPTO_VERSION, tls_start, tls_open, tls_read, tls_write, tls_close,
+const PureCrypto pure_crypto = {PURE_CRYPTO_VERSION, tls_start, tls_open, tls_step, tls_read, tls_write, tls_close,
                                 seal_key, seal_key_free, sealer, seal, sealer_free, opener, unseal, opener_free,
                                 link_secret, link_ephemeral, link_agree, link_derive, aead, aead_seal, aead_open,
                                 aead_free};
diff --git a/link/crypto.c b/link/crypto.c
index c8fb9bd..bec57c5 100644
--- a/link/crypto.c
+++ b/link/crypto.c
@@ -58,6 +58,7 @@ bool pure_tls_start(const char *ca_file, char *error, size_t size) {
 
 /* A session exists only once the table is loaded. */
 PureTls *pure_tls_open(const char *host, int fd) { return crypto ? crypto->tls_open(host, fd) : NULL; }
+int pure_tls_step(PureTls *tls) { return crypto->tls_step(tls); }
 int pure_tls_read(PureTls *tls, void *bytes, int size) { return crypto->tls_read(tls, bytes, size); }
 int pure_tls_write(PureTls *tls, const void *bytes, int size) { return crypto->tls_write(tls, bytes, size); }
 void pure_tls_close(PureTls *tls) { if (tls) crypto->tls_close(tls); }
diff --git a/link/crypto.h b/link/crypto.h
index ede5fb9..4d3fff0 100644
--- a/link/crypto.h
+++ b/link/crypto.h
@@ -8,15 +8,17 @@
  * first time a server seals, opens a seal or connects to an https receiver
  * (link/crypto.c): one that does none of these maps no OpenSSL, 2.3 MiB
  * less resident and 0.6 MiB less of its own. */
-enum { PURE_CRYPTO_VERSION = 5 };
+enum { PURE_CRYPTO_VERSION = 6 };
 
 typedef struct {
     unsigned version;
     /* TLS client: the context (verifying against ca_file, or the system's
-     * CAs when NULL), sessions over connected sockets, their reads and
-     * writes (as SSL_read and SSL_write: bytes, or <= 0), and their end. */
+     * CAs when NULL), sessions over connected non-blocking sockets, their
+     * handshakes' steps, reads and writes (as link/tls.h says), and their
+     * end. */
     bool (*tls_start)(const char *ca_file);
     PureTls *(*tls_open)(const char *host, int fd);
+    int (*tls_step)(PureTls *);
     int (*tls_read)(PureTls *, void *bytes, int size);
     int (*tls_write)(PureTls *, const void *bytes, int size);
     void (*tls_close)(PureTls *);
diff --git a/link/tls.h b/link/tls.h
index 3939f21..232e192 100644
--- a/link/tls.h
+++ b/link/tls.h
@@ -8,10 +8,19 @@
 typedef struct PureTls PureTls;
 /* The context, made once; false with why in error when it cannot be. */
 bool pure_tls_start(const char *ca_file, char *error, size_t size);
-/* A session over a connected socket to host (a DNS name, or an IP address
- * written as one), its certificate verified; NULL when it fails. */
+/* A session over a connected, non-blocking socket to host (a DNS name, or
+ * an IP address written as one); NULL when it cannot be. Its handshake is
+ * made a step at a time, as the socket is ready, by pure_tls_step. */
 PureTls *pure_tls_open(const char *host, int fd);
-/* As SSL_read and SSL_write: the bytes moved, or <= 0. */
+/* What a step, a read or a write says when it moved no bytes: the session
+ * ended (the peer closed it), failed, or waits for the socket to be
+ * readable or writable (then the same again). */
+enum { PURE_TLS_CLOSED = 0, PURE_TLS_FAILED = -1, PURE_TLS_WANTS_READ = -2, PURE_TLS_WANTS_WRITE = -3 };
+/* The handshake: 1 once it is made and the receiver's certificate
+ * verified, else what it waits for, or PURE_TLS_FAILED. */
+int pure_tls_step(PureTls *);
+/* As SSL_read and SSL_write: the bytes moved (a write may move part), or
+ * one of the above. */
 int pure_tls_read(PureTls *, void *bytes, int size);
 int pure_tls_write(PureTls *, const void *bytes, int size);
 void pure_tls_close(PureTls *);
diff --git a/serve/deliver.c b/serve/deliver.c
index 9ed4529..d56932d 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -1,8 +1,8 @@
 #define _POSIX_C_SOURCE 200809L
 #include "deliver.h"
 #include <inttypes.h>
-#include "wait.h"
 #include "seal.h"
+#include "sending.h"
 #include "message-json.h"
 #include "module-text.h"
 #include <errno.h>
@@ -15,36 +15,42 @@
 #include <stdlib.h>
 #include <string.h>
 #include <strings.h>
+#include <sys/epoll.h>
 #include <sys/socket.h>
-#include <sys/time.h>
 #include <time.h>
 #include <unistd.h>
 
 enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256,
-       ANSWER_MOST = 1 << 20 };
+       ANSWER_MOST = 1 << 20, HEAD_MOST = 64 << 10 };
 enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD };
+#define MS UINT64_C(1000000)
+
+typedef struct { char *bytes; size_t size, capacity; bool failed; } Text;
 
 typedef struct { char *body; size_t size; uint64_t seq; unsigned char *sent; size_t sent_size; } Pending;
 /* An answer made into its input's message: a table of the fields' starts and
  * sizes (u32le), then the payload from its prefix on. */
 typedef struct { uint64_t seq; unsigned char *bytes; size_t size, table; } Answer;
 
+/* Where a request is: none under way, to connect, connecting, in its TLS
+ * handshake, written, its answer read. */
+enum { IDLE, CONNECT, CONNECTING, HANDSHAKE, WRITING, READING };
+
 struct PureWire {
+    PureSending sending; /* its place in the sending loop (link/sending.h) */
     const PureModule *module;
     uint32_t output;
     char host[256], port[8], path[1024], source[PURE_SOURCE + 1];
     pthread_mutex_t lock;
-    pthread_cond_t changed;
     Pending *queue; /* the sends to deliver, in order, from head */
     size_t held;    /* their bytes */
     size_t head, count, capacity;
     uint64_t queued, delivered, attempts, deliveries;
     unsigned batch; /* how many a request may carry, as the receiver said (X-Pure-Batch); 0: one */
-    bool stopping, started, stuck;
-    pthread_t thread;
+    bool started, stuck;
     void (*notify)(void *);
     void *context;
-    int fd;   /* the connection kept for the next send, or -1; changed under the lock */
+    int fd;           /* the connection kept for the next send, or -1 */
     bool tls;         /* https: the connection is TLS */
     PureTls *session; /* its TLS session */
     int answer_input; /* the input answers become events of, or -1 */
@@ -58,6 +64,29 @@ struct PureWire {
     unsigned char peer[PURE_SEAL_KEY], from[PURE_SEAL_KEY], enc[PURE_SEAL_KEY];
     const PureSealKey *own;
     PureSealer *sealer;
+    /* Its rounds' own, on the sending loop: the request under way. */
+    int phase;
+    struct addrinfo *addresses, *trying; /* the receiver's, while it connects */
+    bool fresh;                          /* its connection was made for it */
+    char *request;
+    size_t request_size, written;
+    Pending first;     /* the first send it carries */
+    unsigned carried;  /* how many */
+    uint64_t last;     /* the last one's number */
+    uint64_t deadline; /* nothing moved by then: it failed */
+    /* Its answer as it comes: got, its head (got_head bytes) and a chunked
+     * body after it; what the head says; the body kept when answers are
+     * wired. */
+    Text got, body;
+    size_t got_head;
+    bool headed, keep, refused, sized, chunked;
+    int status;
+    unsigned batch_said;
+    uint64_t left; /* a sized body's bytes to come */
+    /* After a request failed: none till then. */
+    uint64_t pause_until;
+    long pause_ms;
+    uint64_t reported; /* the first send the last failure was said of */
 };
 
 /* TLS verifies a receiver's certificate against the system's CAs or the
@@ -68,8 +97,6 @@ void pure_wire_trust(const char *ca_file) { tls_ca = ca_file; }
 
 /* ---- A send as JSON: its fields in order, a group's fields as an object. */
 
-typedef struct { char *bytes; size_t size, capacity; bool failed; } Text;
-
 static void put(Text *t, const void *bytes, size_t size) {
     if (t->failed) return;
     if (size > t->capacity - t->size) {
@@ -176,6 +203,8 @@ static bool encode(const PureWire *w, PureSpan payload, Text *t) {
 
 /* ---- The wire. */
 
+static uint64_t wire_round(PureSending *, uint32_t ready);
+
 PureWire *pure_wire_open(const PureModule *m, const char *output, const char *url, const char *source,
                          char *error, size_t size) {
     int index = pure_module_input(m, output);
@@ -230,7 +259,7 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur
     w->tls = tls;
     w->answer_input = -1;
     pthread_mutex_init(&w->lock, NULL);
-    pure_cond_monotonic(&w->changed);
+    pure_sending_init(&w->sending, wire_round);
     return w;
 }
 
@@ -327,110 +356,13 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
     w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size};
     w->queued = seq;
     w->held += t.size + (sent ? payload.size : 0);
-    pthread_cond_signal(&w->changed);
     pthread_mutex_unlock(&w->lock);
     return true;
 }
 
-/* ---- HTTP/1.1, one request at a time on a kept connection. */
+void pure_wire_push(PureWire *w) { pure_sending_push(&w->sending); }
 
-typedef struct { PureWire *wire; char bytes[4096]; size_t at, end; } Reader;
-
-static bool fill(Reader *r) {
-    if (r->at == r->end) r->at = r->end = 0;
-    if (r->end == sizeof(r->bytes)) return false;
-    PureWire *w = r->wire;
-    ssize_t n;
-    if (w->session) n = pure_tls_read(w->session, r->bytes + r->end, (int)(sizeof(r->bytes) - r->end));
-    else do n = recv(w->fd, r->bytes + r->end, sizeof(r->bytes) - r->end, 0); while (n < 0 && errno == EINTR);
-    if (n <= 0) return false;
-#ifdef TCP_QUICKACK
-    /* Acknowledge at once: a server that writes an answer in pieces holds the
-     * next piece until the first is acknowledged (Nagle), which a delayed
-     * acknowledgement would stall for 40 ms. The flag does not stay set. */
-    int one = 1;
-    setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one));
-#endif
-    r->end += (size_t)n;
-    return true;
-}
-
-/* A line without its CRLF, in line (at most size - 1 bytes). */
-static bool read_line(Reader *r, char *line, size_t size) {
-    for (;;) {
-        char *end = memchr(r->bytes + r->at, '\n', r->end - r->at);
-        if (end) {
-            size_t length = (size_t)(end - (r->bytes + r->at));
-            if (length && end[-1] == '\r') --length;
-            if (length >= size) return false;
-            memcpy(line, r->bytes + r->at, length);
-            line[length] = 0;
-            r->at = (size_t)(end - r->bytes) + 1;
-            return true;
-        }
-        if (r->at) {
-            memmove(r->bytes, r->bytes + r->at, r->end - r->at);
-            r->end -= r->at;
-            r->at = 0;
-        }
-        if (!fill(r)) return false;
-    }
-}
-
-/* size bytes of the body: kept in into (up to ANSWER_MOST), or skipped when it is NULL. */
-static bool take(Reader *r, uint64_t size, Text *into) {
-    if (into && size > ANSWER_MOST - into->size) return false;
-    while (size) {
-        if (r->at == r->end && !fill(r)) return false;
-        size_t n = r->end - r->at < size ? r->end - r->at : (size_t)size;
-        if (into) put(into, r->bytes + r->at, n);
-        r->at += n;
-        size -= n;
-    }
-    return !into || !into->failed;
-}
-
-static bool send_all(PureWire *w, const char *bytes, size_t size) {
-    while (size) {
-        ssize_t n = w->session ? pure_tls_write(w->session, bytes, (int)(size < INT32_MAX ? size : INT32_MAX)) :
-                                 send(w->fd, bytes, size, MSG_NOSIGNAL);
-        if (n < 0 && !w->session && errno == EINTR) continue;
-        if (n <= 0) return false;
-        bytes += n;
-        size -= (size_t)n;
-    }
-    return true;
-}
-
-static int connect_to(const PureWire *w) {
-    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM}, *list;
-    if (getaddrinfo(w->host, w->port, &hints, &list)) return -1;
-    int fd = -1;
-    for (struct addrinfo *a = list; a && fd < 0; a = a->ai_next) {
-        fd = socket(a->ai_family, a->ai_socktype | SOCK_CLOEXEC, a->ai_protocol);
-        if (fd < 0) continue;
-        struct timeval limit = {TIMEOUT_S, 0};
-        int one = 1;
-        setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &limit, sizeof(limit));
-        setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &limit, sizeof(limit));
-        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
-        if (connect(fd, a->ai_addr, a->ai_addrlen)) {
-            close(fd);
-            fd = -1;
-        }
-    }
-    freeaddrinfo(list);
-    return fd;
-}
-
-static void drop_connection(PureWire *w) {
-    pthread_mutex_lock(&w->lock);
-    pure_tls_close(w->session);
-    if (w->fd >= 0) close(w->fd);
-    w->session = NULL;
-    w->fd = -1;
-    pthread_mutex_unlock(&w->lock);
-}
+/* ---- Requests. */
 
 /* A fault a Pure receiver answers when its app refused the message: any
  * but memory (its budget: it may take the message later). */
@@ -440,52 +372,6 @@ static bool refusal(const char *error) {
     return false;
 }
 
-/* The answer to one request: its status, whether a Pure receiver refused it,
- * and whether the connection may carry the next. */
-static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, unsigned *batch, Text *body) {
-    Reader r = {.wire = w};
-    char line[1024];
-    if (!read_line(&r, line, sizeof(line)) || strncmp(line, "HTTP/1.", 7) || strlen(line) < 12) return false;
-    *status = atoi(line + 9);
-    *keep = line[7] == '1';
-    *refused = false;
-    uint64_t length = 0;
-    bool sized = false, chunked = false;
-    while (read_line(&r, line, sizeof(line))) {
-        if (!*line) {
-            if (chunked) {
-                for (;;) {
-                    uint64_t chunk;
-                    if (!read_line(&r, line, sizeof(line))) return false;
-                    chunk = strtoull(line, NULL, 16);
-                    if (!chunk) break;
-                    if (!take(&r, chunk, body) || !read_line(&r, line, sizeof(line))) return false;
-                }
-                while (read_line(&r, line, sizeof(line)) && *line) {}
-                return true;
-            }
-            if (*status == 204 || *status == 304 || *status < 200) return true;
-            if (sized) return take(&r, length, body);
-            *keep = false; /* the body runs to the close */
-            while (fill(&r)) {
-                if (body && !take(&r, r.end - r.at, body)) return false;
-                r.at = r.end;
-            }
-            return true;
-        }
-        char *value = strchr(line, ':');
-        if (!value) return false;
-        *value++ = 0;
-        while (*value == ' ' || *value == '\t') ++value;
-        if (!strcasecmp(line, "Content-Length")) { length = strtoull(value, NULL, 10); sized = true; }
-        else if (!strcasecmp(line, "Transfer-Encoding")) chunked = !strcasecmp(value, "chunked");
-        else if (!strcasecmp(line, "Connection")) *keep = strcasecmp(value, "close") != 0 && *keep;
-        else if (!strcasecmp(line, "X-Pure-Error")) *refused = refusal(value);
-        else if (!strcasecmp(line, "X-Pure-Batch")) *batch = (unsigned)strtoul(value, NULL, 10);
-    }
-    return false;
-}
-
 static void write_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
 
 /* The answer (status, body) to the send p as a message of the answer input:
@@ -562,8 +448,6 @@ static bool form_answer(const PureWire *w, int status, const Text *body, const P
     return true;
 }
 
-static int send_request(PureWire *w, const char *request, size_t size, const Pending *p, Answer *answer);
-
 void pure_wire_seal(PureWire *w, const unsigned char peer[PURE_SEAL_KEY], const PureSealKey *own,
                     const unsigned char *from) {
     memcpy(w->peer, peer, PURE_SEAL_KEY);
@@ -616,16 +500,16 @@ static unsigned char *seal_body(PureWire *w, const Pending *p, unsigned count, s
 }
 
 /* One request for count sends from p (numbered in a row): one object, or
- * an array of them for a receiver that takes batches. A final answer to a
- * send whose answers are wired is made into *answer. */
-static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) {
+ * an array of them for a receiver that takes batches; NULL when it cannot
+ * be made (memory, a seal). */
+static char *make_request(PureWire *w, const Pending *p, unsigned count, size_t *size) {
     size_t body = count > 1 ? count + 1 : 0;
     for (unsigned i = 0; i < count; ++i) body += p[i].size;
     size_t length = body + (w->sealed ? PURE_SEAL_OVERHEAD : 0);
     char batch[32] = "", seal[256] = "";
     if (count > 1) snprintf(batch, sizeof(batch), "X-Pure-Count: %u\r\n", count);
     unsigned char *sealed = w->sealed ? seal_body(w, p, count, body, seal, sizeof(seal)) : NULL;
-    if (w->sealed && !sealed) return AGAIN;
+    if (w->sealed && !sealed) return NULL;
     /* The request in one piece, its head and then a batch in its brackets:
      * one write where there were two a send (a batch drained at 91k sends a
      * second, and at 470k in one piece). */
@@ -633,7 +517,7 @@ static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) {
     char *request = malloc(HEAD + length);
     if (!request) {
         free(sealed);
-        return AGAIN;
+        return NULL;
     }
     int head = snprintf(request, HEAD,
                         "POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n"
@@ -643,145 +527,435 @@ static int post(PureWire *w, const Pending *p, unsigned count, Answer *answer) {
     if (sealed) memcpy(request + head, sealed, length);
     else put_body((unsigned char *)request + head, p, count);
     free(sealed);
-    int result = send_request(w, request, (size_t)head + length, p, answer);
-    free(request);
-    return result;
-}
-
-static int send_request(PureWire *w, const char *request, size_t size, const Pending *p, Answer *answer) {
-    /* A kept connection the peer closed is tried once more on a new one. */
-    for (int attempt = 0; attempt < 2; ++attempt) {
-        bool fresh = w->fd < 0;
-        if (fresh) {
-            int fd = connect_to(w);
-            if (fd < 0) return AGAIN;
-            pthread_mutex_lock(&w->lock);
-            w->fd = fd;
-            bool stopping = w->stopping;
-            pthread_mutex_unlock(&w->lock);
-            if (stopping) return AGAIN;
-            /* The receiver's certificate must name the host. */
-            PureTls *session = w->tls ? pure_tls_open(w->host, fd) : NULL;
-            if (w->tls && !session) {
-                drop_connection(w);
-                return TLS_FAILED;
-            }
-            pthread_mutex_lock(&w->lock);
-            w->session = session;
-            pthread_mutex_unlock(&w->lock);
-        }
-        int status;
-        bool refused, keep, sent = send_all(w, request, size);
-        unsigned batch_size = 0;
-        Text body = {0};
-        if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size, w->answer_input >= 0 ? &body : NULL)) {
-            free(body.bytes);
-            drop_connection(w);
-            if (fresh) return AGAIN;
+    *size = (size_t)head + length;
+    return request;
+}
+
+/* ---- HTTP/1.1, one request at a time on a kept connection, in the wire's
+ * rounds on the sending loop: each goes as far as its socket lets it, and
+ * then waits for the socket (or its deadline, TIMEOUT_S after the last
+ * bytes moved). */
+
+/* Where the head ends in b (after its blank line), looking from from; 0
+ * while it has not. */
+static size_t head_end(const char *b, size_t size, size_t from) {
+    for (const char *at = memchr(b + from, '\n', size - from); at; at = memchr(at + 1, '\n', (size_t)(b + size - at - 1))) {
+        size_t i = (size_t)(at - b);
+        if (i + 1 < size && b[i + 1] == '\n') return i + 2;
+        if (i + 2 < size && b[i + 1] == '\r' && b[i + 2] == '\n') return i + 3;
+    }
+    return 0;
+}
+
+/* The head's lines (size bytes of b): the status, and the headers a wire
+ * heeds. False when it is no HTTP/1 answer. */
+static bool read_head(PureWire *w, const char *b, size_t size) {
+    w->sized = w->chunked = w->refused = false;
+    w->left = 0;
+    w->batch_said = 0;
+    for (size_t at = 0, n = 0; at < size; ++n) {
+        const char *end = memchr(b + at, '\n', size - at);
+        if (!end) return false;
+        size_t length = (size_t)(end - (b + at));
+        if (length && end[-1] == '\r') --length;
+        char line[1024];
+        if (length >= sizeof(line)) return false;
+        memcpy(line, b + at, length);
+        line[length] = 0;
+        at = (size_t)(end - b) + 1;
+        if (!n) {
+            if (length < 12 || strncmp(line, "HTTP/1.", 7)) return false;
+            w->status = atoi(line + 9);
+            w->keep = line[7] == '1';
             continue;
         }
-        if (!keep) drop_connection(w);
-        if (batch_size && w->answer_input < 0) w->batch = batch_size;
-        if (status == 408 || status == 429 || status >= 500) {
-            free(body.bytes);
-            return AGAIN;
+        if (!*line) break;
+        char *value = strchr(line, ':');
+        if (!value) return false;
+        *value++ = 0;
+        while (*value == ' ' || *value == '\t') ++value;
+        if (!strcasecmp(line, "Content-Length")) { w->left = strtoull(value, NULL, 10); w->sized = true; }
+        else if (!strcasecmp(line, "Transfer-Encoding")) w->chunked = !strcasecmp(value, "chunked");
+        else if (!strcasecmp(line, "Connection")) w->keep = strcasecmp(value, "close") != 0 && w->keep;
+        else if (!strcasecmp(line, "X-Pure-Error")) w->refused = refusal(value);
+        else if (!strcasecmp(line, "X-Pure-Batch")) w->batch_said = (unsigned)strtoul(value, NULL, 10);
+    }
+    return true;
+}
+
+/* A chunked body after the head, as far as it came (closed: no more
+ * comes), made again from its start each time (such answers are few and
+ * small): 1 once it is whole (kept when answers are wired), 0 while more
+ * must come, -1 when it cannot be. */
+static int chunks(PureWire *w, bool closed) {
+    const char *b = w->got.bytes + w->got_head;
+    size_t size = w->got.size - w->got_head, at = 0;
+    w->body.size = 0;
+    for (;;) {
+        const char *end = memchr(b + at, '\n', size - at);
+        if (!end) return closed ? -1 : 0;
+        uint64_t chunk = strtoull(b + at, NULL, 16);
+        at = (size_t)(end - b) + 1;
+        if (!chunk) break;
+        if (chunk > size - at) return closed ? -1 : 0;
+        if (w->answer_input >= 0) {
+            if (chunk > ANSWER_MOST - w->body.size) return -1;
+            put(&w->body, b + at, (size_t)chunk);
+        }
+        at += (size_t)chunk;
+        if (!(end = memchr(b + at, '\n', size - at))) return closed ? -1 : 0;
+        at = (size_t)(end - b) + 1;
+    }
+    for (;;) { /* its trailers, to a blank line (or the close) */
+        const char *end = memchr(b + at, '\n', size - at);
+        if (!end) return closed ? 1 : 0;
+        size_t length = (size_t)(end - (b + at));
+        at = (size_t)(end - b) + 1;
+        if (!length || (length == 1 && end[-1] == '\r')) return w->body.failed ? -1 : 1;
+    }
+}
+
+/* Bytes of the answer as they come (closed: no more come): 1 once it is
+ * whole, 0 while more must come, -1 when it is no answer or too large. */
+static int answer_bytes(PureWire *w, const char *bytes, size_t n, bool closed) {
+    if (!w->headed) {
+        size_t before = w->got.size;
+        put(&w->got, bytes, n);
+        size_t end = w->got.failed ? 0 : head_end(w->got.bytes, w->got.size, before >= 2 ? before - 2 : 0);
+        if (!end) return closed || w->got.failed || w->got.size > HEAD_MOST ? -1 : 0;
+        if (!read_head(w, w->got.bytes, end)) return -1;
+        w->headed = true;
+        w->got_head = end;
+        if (w->status == 204 || w->status == 304 || w->status < 200) return 1;
+        if (w->chunked) return chunks(w, closed);
+        bytes = w->got.bytes + end; /* the body's first bytes */
+        n = w->got.size - end;
+    } else if (w->chunked) {
+        put(&w->got, bytes, n);
+        return w->got.failed || w->got.size - w->got_head > ANSWER_MOST + HEAD_MOST ? -1 : chunks(w, closed);
+    }
+    size_t taken = w->sized && n > w->left ? (size_t)w->left : n;
+    if (w->answer_input >= 0) {
+        if (taken > ANSWER_MOST - w->body.size) return -1;
+        put(&w->body, bytes, taken);
+        if (w->body.failed) return -1;
+    }
+    if (w->sized) {
+        w->left -= taken;
+        return !w->left ? 1 : closed ? -1 : 0;
+    }
+    w->keep = false; /* the body runs to the close */
+    return closed ? 1 : 0;
+}
+
+static void drop(PureWire *w) {
+    pure_sending_watch(&w->sending, -1, 0); /* before its socket closes */
+    pure_tls_close(w->session);
+    w->session = NULL;
+    if (w->fd >= 0) close(w->fd);
+    w->fd = -1;
+}
+
+/* The request is done with: delivered, or to be tried again after a pause
+ * (0.1 s doubling to 10 s; 10 s for one the receiver does not take). */
+static void finish(PureWire *w, int outcome, Answer *answer) {
+    free(w->request);
+    w->request = NULL;
+    w->phase = IDLE;
+    w->headed = false;
+    w->got.size = w->body.size = 0;
+    w->got.failed = w->body.failed = false;
+    if (w->addresses) freeaddrinfo(w->addresses);
+    w->addresses = w->trying = NULL;
+    if (w->fd >= 0) pure_sending_watch(&w->sending, w->fd, EPOLLIN); /* kept: heard if it closes */
+    pthread_mutex_lock(&w->lock);
+    if (outcome == DELIVERED && answer && answer->bytes) { /* for the owner, who then counts it delivered */
+        if (w->answer_head + w->answer_count == w->answer_capacity) {
+            if (w->answer_head) {
+                memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers));
+                w->answer_head = 0;
+            } else {
+                size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16;
+                Answer *grown = realloc(w->answers, capacity * sizeof(*grown));
+                if (!grown) { free(answer->bytes); outcome = AGAIN; }
+                else { w->answers = grown; w->answer_capacity = capacity; }
+            }
         }
-        if (w->answer_input >= 0 && status >= 200 && status < 500) { /* a final answer: the app hears it */
-            bool formed = form_answer(w, status, &body, p, answer);
-            free(body.bytes);
-            return formed ? DELIVERED : ANSWER_UNREAD;
+        if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = *answer;
+    }
+    if (outcome == DELIVERED) {
+        for (unsigned i = 0; i < w->carried; ++i) {
+            const Pending *p = &w->queue[w->head + i];
+            w->held -= p->size + (p->sent ? p->sent_size : 0);
+            free(p->body);
+            free(p->sent);
         }
-        free(body.bytes);
-        if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED;
-        return REFUSED_BY_PEER;
+        w->head += w->carried;
+        w->count -= w->carried;
+        if (!answer || !answer->bytes) w->delivered = w->last;
+        w->deliveries += w->carried;
+        w->pause_ms = 0;
+        pthread_mutex_unlock(&w->lock);
+        if (w->notify) w->notify(w->context);
+        return;
+    }
+    pthread_mutex_unlock(&w->lock);
+    if (outcome != AGAIN && w->reported != w->first.seq) {
+        fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source, (unsigned long long)w->first.seq,
+                outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" :
+                outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" :
+                "the receiver does not take it; trying again");
+        w->reported = w->first.seq;
     }
-    return AGAIN;
+    w->pause_ms = outcome != AGAIN ? LONGEST_PAUSE_MS : w->pause_ms ? 2 * w->pause_ms : FIRST_PAUSE_MS;
+    if (w->pause_ms > LONGEST_PAUSE_MS) w->pause_ms = LONGEST_PAUSE_MS;
+    w->pause_until = pure_sending_now() + (uint64_t)w->pause_ms * MS;
 }
 
-static void *deliver(void *context) {
-    PureWire *w = context;
-    long pause_ms = 0;
-    uint64_t reported = 0;
+/* The request failed on its connection: the connection goes, and a request
+ * on one kept from before is tried once more on a new one. */
+static void failed(PureWire *w, int outcome) {
+    bool again = outcome == AGAIN && !w->fresh;
+    drop(w);
+    if (!again) {
+        finish(w, outcome, NULL);
+        return;
+    }
+    w->phase = CONNECT;
+    w->written = 0;
+    w->headed = false;
+    w->got.size = w->body.size = 0;
+    w->got.failed = w->body.failed = false;
+}
+
+/* The next request, from the sends queued (as many as the receiver takes
+ * at once: their bodies stay put while the queue only grows behind them):
+ * false when none waits. */
+static bool begin(PureWire *w, uint64_t now) {
+    Pending next[BATCH_MOST];
     pthread_mutex_lock(&w->lock);
+    unsigned count = w->batch > 1 ? (unsigned)(w->count < w->batch ? w->count : w->batch) : w->count ? 1 : 0;
+    if (count > BATCH_MOST) count = BATCH_MOST;
+    memcpy(next, w->queue + w->head, count * sizeof(*next));
+    if (count) ++w->attempts;
+    pthread_mutex_unlock(&w->lock);
+    if (!count) return false;
+    w->first = next[0];
+    w->carried = count;
+    w->last = next[count - 1].seq;
+    w->written = 0;
+    w->deadline = now + TIMEOUT_S * 1000 * MS;
+    if (!(w->request = make_request(w, next, count, &w->request_size))) finish(w, AGAIN, NULL);
+    else w->phase = w->fd >= 0 ? WRITING : CONNECT;
+    w->fresh = w->fd < 0;
+    return true;
+}
+
+/* A socket connecting to the next of the receiver's addresses (watched to
+ * be writable: connected); false when none is left. */
+static bool connect_next(PureWire *w) {
+    for (; w->trying; w->trying = w->trying->ai_next) {
+        struct addrinfo *a = w->trying;
+        int fd = socket(a->ai_family, a->ai_socktype | SOCK_NONBLOCK | SOCK_CLOEXEC, a->ai_protocol), one = 1;
+        if (fd < 0) continue;
+        setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
+        if (!connect(fd, a->ai_addr, a->ai_addrlen) || errno == EINPROGRESS) {
+            w->fd = fd;
+            pure_sending_watch(&w->sending, fd, EPOLLOUT);
+            return true;
+        }
+        close(fd);
+    }
+    return false;
+}
+
+/* The request's bytes to the socket as far as it takes them: 1 all
+ * written, 0 when it waits for the socket, -1 when it failed. */
+static int write_some(PureWire *w) {
+    while (w->written < w->request_size) {
+        size_t left = w->request_size - w->written;
+        ssize_t n;
+        if (w->session) {
+            int moved = pure_tls_write(w->session, w->request + w->written, left < INT32_MAX ? (int)left : INT32_MAX);
+            if (moved == PURE_TLS_WANTS_READ || moved == PURE_TLS_WANTS_WRITE) {
+                pure_sending_watch(&w->sending, w->fd, moved == PURE_TLS_WANTS_READ ? EPOLLIN : EPOLLOUT);
+                return 0;
+            }
+            n = moved;
+        } else {
+            n = send(w->fd, w->request + w->written, left, MSG_NOSIGNAL);
+            if (n < 0 && errno == EINTR) continue;
+            if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
+                pure_sending_watch(&w->sending, w->fd, EPOLLOUT);
+                return 0;
+            }
+        }
+        if (n <= 0) return -1;
+        w->written += (size_t)n;
+        w->deadline = pure_sending_now() + TIMEOUT_S * 1000 * MS;
+    }
+    return 1;
+}
+
+/* The answer's bytes from the socket as far as they came: 1 once it is
+ * whole, 0 when it waits for more, -1 when it failed. */
+static int read_some(PureWire *w) {
+    char bytes[16384];
     for (;;) {
-        while (!w->stopping && !w->count) pthread_cond_wait(&w->changed, &w->lock);
-        if (w->stopping) break;
-        /* As many as the receiver takes at once; their bodies stay put while
-         * the queue only grows behind them. */
-        unsigned count = w->batch > 1 ? (unsigned)(w->count < w->batch ? w->count : w->batch) : 1;
-        Pending next[BATCH_MOST];
-        if (count > BATCH_MOST) count = BATCH_MOST;
-        memcpy(next, w->queue + w->head, count * sizeof(*next));
-        ++w->attempts;
-        pthread_mutex_unlock(&w->lock);
-        Answer answer = {0};
-        int outcome = post(w, next, count, &answer);
-        pthread_mutex_lock(&w->lock);
-        if (outcome == DELIVERED && answer.bytes) { /* for the owner, who then counts it delivered */
-            if (w->answer_head + w->answer_count == w->answer_capacity) {
-                if (w->answer_head) {
-                    memmove(w->answers, w->answers + w->answer_head, w->answer_count * sizeof(*w->answers));
-                    w->answer_head = 0;
-                } else {
-                    size_t capacity = w->answer_capacity ? 2 * w->answer_capacity : 16;
-                    Answer *grown = realloc(w->answers, capacity * sizeof(*grown));
-                    if (!grown) { free(answer.bytes); outcome = AGAIN; }
-                    else { w->answers = grown; w->answer_capacity = capacity; }
-                }
+        ssize_t n;
+        if (w->session) {
+            int moved = pure_tls_read(w->session, bytes, (int)sizeof(bytes));
+            if (moved == PURE_TLS_WANTS_READ || moved == PURE_TLS_WANTS_WRITE) {
+                pure_sending_watch(&w->sending, w->fd, moved == PURE_TLS_WANTS_READ ? EPOLLIN : EPOLLOUT);
+                return 0;
+            }
+            n = moved; /* closed, or failed: as the close */
+        } else {
+            n = recv(w->fd, bytes, sizeof(bytes), 0);
+            if (n < 0 && errno == EINTR) continue;
+            if (n < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
+                pure_sending_watch(&w->sending, w->fd, EPOLLIN);
+                return 0;
+            }
+        }
+        if (n > 0) {
+#ifdef TCP_QUICKACK
+            /* Acknowledge at once: a server that writes an answer in pieces
+             * holds the next piece until the first is acknowledged (Nagle),
+             * which a delayed acknowledgement would stall for 40 ms. The flag
+             * does not stay set. */
+            int one = 1;
+            setsockopt(w->fd, IPPROTO_TCP, TCP_QUICKACK, &one, sizeof(one));
+#endif
+            w->deadline = pure_sending_now() + TIMEOUT_S * 1000 * MS;
+        }
+        int whole = answer_bytes(w, bytes, n > 0 ? (size_t)n : 0, n <= 0);
+        if (whole || n <= 0) return whole ? whole : -1;
+    }
+}
+
+/* The whole answer: what came of the request. */
+static void answered(PureWire *w) {
+    if (w->batch_said && w->answer_input < 0) w->batch = w->batch_said;
+    if (!w->keep) drop(w);
+    int status = w->status, outcome;
+    Answer answer = {0};
+    if (status == 408 || status == 429 || status >= 500) outcome = AGAIN;
+    else if (w->answer_input >= 0 && status >= 200 && status < 500) /* a final answer: the app hears it */
+        outcome = form_answer(w, status, &w->body, &w->first, &answer) ? DELIVERED : ANSWER_UNREAD;
+    else if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && w->refused)) outcome = DELIVERED;
+    else outcome = REFUSED_BY_PEER;
+    finish(w, outcome, &answer);
+}
+
+/* A round: the request under way taken as far as it goes, and the next
+ * begun when it is done. When the wire wants its next round. */
+static uint64_t wire_round(PureSending *s, uint32_t ready) {
+    PureWire *w = (PureWire *)s;
+    for (;;) {
+        uint64_t now = pure_sending_now();
+        if (w->phase >= CONNECTING && !ready && now >= w->deadline) { /* nothing moved for TIMEOUT_S */
+            failed(w, AGAIN);
+            continue;
+        }
+        switch (w->phase) {
+        case IDLE:
+            if ((ready & (EPOLLIN | EPOLLHUP | EPOLLERR)) && w->fd >= 0) drop(w); /* a kept connection that speaks unasked is closing */
+            ready = 0;
+            if (now < w->pause_until) return w->pause_until;
+            if (!begin(w, now)) return 0;
+            continue;
+        case CONNECT:
+            w->fresh = true;
+            if (!w->addresses) {
+                struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_STREAM};
+                if (getaddrinfo(w->host, w->port, &hints, &w->addresses)) w->addresses = NULL; /* a name looked up waits here */
+                w->trying = w->addresses;
+            }
+            if (!connect_next(w)) {
+                finish(w, AGAIN, NULL);
+                continue;
+            }
+            w->phase = CONNECTING;
+            w->deadline = now + TIMEOUT_S * 1000 * MS;
+            return w->deadline;
+        case CONNECTING: {
+            if (!ready) return w->deadline;
+            ready = 0;
+            int error = 0;
+            socklen_t size = sizeof(error);
+            if (getsockopt(w->fd, SOL_SOCKET, SO_ERROR, &error, &size) || error) { /* the next address */
+                drop(w);
+                w->trying = w->trying->ai_next;
+                w->phase = CONNECT;
+                continue;
+            }
+            freeaddrinfo(w->addresses);
+            w->addresses = w->trying = NULL;
+            pure_sending_watch(s, w->fd, EPOLLIN);
+            w->deadline = now + TIMEOUT_S * 1000 * MS;
+            if (!w->tls) w->phase = WRITING;
+            else if ((w->session = pure_tls_open(w->host, w->fd))) w->phase = HANDSHAKE; /* the certificate must name the host */
+            else failed(w, TLS_FAILED);
+            continue;
+        }
+        case HANDSHAKE: {
+            ready = 0;
+            int step = pure_tls_step(w->session);
+            if (step == PURE_TLS_WANTS_READ || step == PURE_TLS_WANTS_WRITE) {
+                pure_sending_watch(s, w->fd, step == PURE_TLS_WANTS_READ ? EPOLLIN : EPOLLOUT);
+                return w->deadline;
             }
-            if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer;
+            if (step != 1) {
+                failed(w, TLS_FAILED);
+                continue;
+            }
+            pure_sending_watch(s, w->fd, EPOLLIN);
+            w->phase = WRITING;
+            continue;
         }
-        if (outcome == DELIVERED) {
-            for (unsigned i = 0; i < count; ++i) {
-                const Pending *p = &w->queue[w->head + i];
-                w->held -= p->size + (p->sent ? p->sent_size : 0);
-                free(p->body);
-                free(p->sent);
+        case WRITING: {
+            ready = 0;
+            int written = write_some(w);
+            if (written < 0) failed(w, AGAIN);
+            else if (!written) return w->deadline;
+            else { /* the answer, once the socket has it */
+                pure_sending_watch(s, w->fd, EPOLLIN);
+                w->phase = READING;
+                return w->deadline;
             }
-            w->head += count;
-            w->count -= count;
-            if (!answer.bytes) w->delivered = next[count - 1].seq;
-            w->deliveries += count;
-            pause_ms = 0;
-            pthread_mutex_unlock(&w->lock);
-            if (w->notify) w->notify(w->context);
-            pthread_mutex_lock(&w->lock);
             continue;
         }
-        if (outcome != AGAIN && reported != next[0].seq) {
-            fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%llu,\"error\":\"%s\"}\n", w->source,
-                    (unsigned long long)next[0].seq, outcome == TLS_FAILED ? "the receiver's certificate is not trusted; trying again" :
-                    outcome == ANSWER_UNREAD ? "its answer does not make the answer input's message; trying again" :
-                    "the receiver does not take it; trying again");
-            reported = next[0].seq;
+        case READING: {
+            if (!ready) return w->deadline;
+            ready = 0;
+            int whole = read_some(w);
+            if (whole < 0) failed(w, AGAIN);
+            else if (!whole) return w->deadline;
+            else answered(w);
+            continue;
+        }
         }
-        pause_ms = outcome != AGAIN ? LONGEST_PAUSE_MS : pause_ms ? 2 * pause_ms : FIRST_PAUSE_MS;
-        if (pause_ms > LONGEST_PAUSE_MS) pause_ms = LONGEST_PAUSE_MS;
-        struct timespec until = pure_wait_until(pause_ms * 1000000LL);
-        while (!w->stopping && pthread_cond_timedwait(&w->changed, &w->lock, &until) == 0) {}
     }
-    pthread_mutex_unlock(&w->lock);
-    return NULL;
 }
 
 bool pure_wire_start(PureWire *w, void (*notify)(void *), void *context) {
     w->notify = notify;
     w->context = context;
-    w->started = pthread_create(&w->thread, NULL, deliver, w) == 0;
+    char why[128];
+    w->started = pure_sending_join(&w->sending, why, sizeof(why));
     return w->started;
 }
 
 void pure_wire_close(PureWire *w) {
     if (!w) return;
-    pthread_mutex_lock(&w->lock);
-    w->stopping = true;
-    if (w->fd >= 0) shutdown(w->fd, SHUT_RDWR); /* a send or an answer waited for ends now */
-    pthread_cond_broadcast(&w->changed);
-    pthread_mutex_unlock(&w->lock);
-    if (w->started) pthread_join(w->thread, NULL);
-    pure_tls_close(w->session);
+    pure_sending_leave(&w->sending); /* no round of it after */
+    drop(w);
+    if (w->addresses) freeaddrinfo(w->addresses);
+    free(w->request);
+    free(w->got.bytes);
+    free(w->body.bytes);
     pure_sealer_free(w->sealer);
-    if (w->fd >= 0) close(w->fd);
     for (size_t i = 0; i < w->count; ++i) {
         free(w->queue[w->head + i].body);
         free(w->queue[w->head + i].sent);
@@ -789,7 +963,6 @@ void pure_wire_close(PureWire *w) {
     for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes);
     free(w->answers);
     free(w->queue);
-    pthread_cond_destroy(&w->changed);
     pthread_mutex_destroy(&w->lock);
     free(w);
 }
diff --git a/serve/deliver.h b/serve/deliver.h
index 244363a..40f1e4c 100644
--- a/serve/deliver.h
+++ b/serve/deliver.h
@@ -32,15 +32,20 @@ uint32_t pure_wire_output(const PureWire *);
 void pure_wire_seal(PureWire *, const unsigned char peer[32], const PureSealKey *own, const unsigned char *from);
 /* The owner's side: queue a saved send numbered after the last queued one
  * (it is encoded now); false when it cannot be (a bytes field that is not
- * UTF-8 has no JSON form), and the wire then stops at it. */
+ * UTF-8 has no JSON form), and the wire then stops at it. What was queued
+ * goes once pushed, so a turn's sends go together. */
 bool pure_wire_queue(PureWire *, uint64_t seq, PureSpan payload);
+void pure_wire_push(PureWire *);
 uint64_t pure_wire_queued(const PureWire *);
 /* The last number delivered; any thread. */
 uint64_t pure_wire_delivered(PureWire *);
 /* The bytes of the sends queued and not yet delivered (their JSON, and the
  * payloads kept for answers); any thread. */
 size_t pure_wire_held(PureWire *);
-/* Deliver on a thread of the wire's own, calling notify after each delivery. */
+/* Deliver on the process's sending loop (link/sending.h), calling notify
+ * (on its thread) after each delivery. A request goes on a non-blocking
+ * socket, a step at a time as it is ready; a receiver's name is looked up
+ * (DNS) where it connects, and the loop waits for it. */
 bool pure_wire_start(PureWire *, void (*notify)(void *), void *context);
 /* Stop (a delivery under way is finished or abandoned) and free. */
 void pure_wire_close(PureWire *);
diff --git a/serve/serve.c b/serve/serve.c
index 86e38f8..81a338d 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -1845,7 +1845,8 @@ static void feed_outlets(Executor *ex) {
             queued = seq;
             held += payload.size;
         }
-        if (link && queued != before) pure_link_push(link); /* the turn's together */
+        if (queued != before && link) pure_link_push(link); /* the turn's together */
+        else if (queued != before && wire) pure_wire_push(wire);
         ex->outlet_from[i] = at;
     }
     uint64_t now = ex->local_count ? now_ms() : 0;