Make an API's answers the app's events (--reply OUTPUT=INPUT)

The final answer to each send of a wired output (a 2xx, or a 4xx but
408 and 429) becomes a message event of the given input: fields from
the answer's JSON object by name, other members skipped; a field the
answer lacks from the send's field of that name and type, so the
request's own id is there; a u32 http_status field is the status. It is
applied with the send's number from the wire's source, before the send
is trimmed and in the same save, so an answer that comes again after a
restart (the API repeating itself for the Idempotency-Key) applies once.
Sends that wait for answers go one per request; answers are saved with
the next round, or at once when none waits.

A server that writes an answer in pieces holds the second until the
first is acknowledged, and a delayed acknowledgement stalled every
answer 40 ms: the client now acknowledges at once (TCP_QUICKACK) after
each read. Against a payments API that answers 200 or 402, with the
sender killed (-9) midway, paid and declined orders matched the API's
answers one to one.

3 files changed, 255 insertions, 31 deletions

diff --git a/prototypes/connections/deliver.c b/prototypes/connections/deliver.c
index c4136a2..0d3da64 100644
--- a/prototypes/connections/deliver.c
+++ b/prototypes/connections/deliver.c
@@ -1,5 +1,6 @@
 #define _POSIX_C_SOURCE 200809L
 #include "deliver.h"
+#include "message-json.h"
 #include "module-text.h"
 #include <errno.h>
 #include <netdb.h>
@@ -18,10 +19,14 @@
 #include <time.h>
 #include <unistd.h>
 
-enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256 };
-enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED };
+enum { TIMEOUT_S = 10, FIRST_PAUSE_MS = 100, LONGEST_PAUSE_MS = 10000, GROUPS = PURE_MODULE_NAME / 2, BATCH_MOST = 256,
+       ANSWER_MOST = 1 << 20 };
+enum { DELIVERED, AGAIN, REFUSED_BY_PEER, TLS_FAILED, ANSWER_UNREAD };
 
-typedef struct { char *body; size_t size; uint64_t seq; } Pending;
+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;
 
 struct PureWire {
     const PureModule *module;
@@ -40,6 +45,11 @@ struct PureWire {
     int fd;   /* the connection kept for the next send, or -1; changed under the lock */
     bool tls; /* https: the connection is TLS */
     SSL *ssl; /* its TLS session */
+    int answer_input; /* the input answers become events of, or -1 */
+    Answer *answers;  /* made, for the owner to take in order */
+    size_t answer_head, answer_count, answer_capacity;
+    Answer taken;     /* the one the owner took last, kept until it takes the next */
+    uint32_t placed[2 * PURE_MODULE_FIELDS];
 };
 
 /* ---- TLS: one client context, verifying the receiver's certificate against
@@ -226,12 +236,42 @@ PureWire *pure_wire_open(const PureModule *m, const char *output, const char *ur
     memcpy(w->source, source, length + 1);
     w->fd = -1;
     w->tls = tls;
+    w->answer_input = -1;
     pthread_mutex_init(&w->lock, NULL);
     pthread_cond_init(&w->changed, NULL);
     return w;
 }
 
 uint32_t pure_wire_output(const PureWire *w) { return w->output; }
+
+bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) {
+    const PureModule *m = w->module;
+    if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed) {
+        snprintf(error, size, "an answer goes to a message input without a key");
+        return false;
+    }
+    w->answer_input = (int)input;
+    return true;
+}
+
+bool pure_wire_answer(PureWire *w, PureWireAnswer *out) {
+    free(w->taken.bytes);
+    w->taken = (Answer){0};
+    pthread_mutex_lock(&w->lock);
+    bool some = w->answer_count > 0;
+    if (some) {
+        w->taken = w->answers[w->answer_head++];
+        --w->answer_count;
+        w->delivered = w->taken.seq; /* trimmed after the owner applies it */
+    }
+    pthread_mutex_unlock(&w->lock);
+    if (!some) return false;
+    const PureInputShape *in = &w->module->input[w->answer_input];
+    for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i);
+    *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq,
+                            {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source};
+    return true;
+}
 uint64_t pure_wire_queued(const PureWire *w) { return w->queued; }
 
 uint64_t pure_wire_delivered(PureWire *w) {
@@ -275,7 +315,14 @@ bool pure_wire_queue(PureWire *w, uint64_t seq, PureSpan payload) {
             w->capacity = capacity;
         }
     }
-    w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq};
+    unsigned char *sent = NULL;
+    if (w->answer_input >= 0 && !(sent = malloc(payload.size ? payload.size : 1))) {
+        pthread_mutex_unlock(&w->lock);
+        free(t.bytes);
+        return false;
+    }
+    if (sent && payload.size) memcpy(sent, payload.data, payload.size);
+    w->queue[w->head + w->count++] = (Pending){t.bytes, t.size, seq, sent, payload.size};
     w->queued = seq;
     pthread_cond_signal(&w->changed);
     pthread_mutex_unlock(&w->lock);
@@ -294,6 +341,13 @@ static bool fill(Reader *r) {
     if (w->ssl) n = SSL_read(w->ssl, 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;
 }
@@ -320,14 +374,17 @@ static bool read_line(Reader *r, char *line, size_t size) {
     }
 }
 
-static bool skip(Reader *r, uint64_t size) {
+/* 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 true;
+    return !into || !into->failed;
 }
 
 static bool send_all(PureWire *w, const char *bytes, size_t size) {
@@ -397,7 +454,7 @@ static bool refusal(const char *error) {
 
 /* 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) {
+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;
@@ -414,15 +471,18 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns
                     if (!read_line(&r, line, sizeof(line))) return false;
                     chunk = strtoull(line, NULL, 16);
                     if (!chunk) break;
-                    if (!skip(&r, chunk) || !read_line(&r, line, sizeof(line))) return false;
+                    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 skip(&r, length);
+            if (sized) return take(&r, length, body);
             *keep = false; /* the body runs to the close */
-            while (fill(&r)) r.at = r.end;
+            while (fill(&r)) {
+                if (body && !take(&r, r.end - r.at, body)) return false;
+                r.at = r.end;
+            }
             return true;
         }
         char *value = strchr(line, ':');
@@ -438,9 +498,64 @@ static bool read_answer(PureWire *w, int *status, bool *refused, bool *keep, uns
     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:
+ * fields from the body's JSON object by name, others skipped; a field the
+ * body lacks from the send's field of that name and type; a u32 field named
+ * http_status from the status. False when a field has none of these. */
+static bool form_answer(const PureWire *w, int status, const Text *body, const Pending *p, Answer *out) {
+    const PureModule *m = w->module;
+    const PureInputShape *in = &m->input[w->answer_input], *sent = &m->input[w->output];
+    size_t table = 8 * (size_t)in->field_count, total = table + in->prefix + body->size + p->sent_size;
+    unsigned char *bytes = calloc(1, total ? total : 1);
+    if (!bytes) return false;
+    unsigned char *prefix = bytes + table, *text = prefix + in->prefix, *copy = text + body->size;
+    if (body->size) memcpy(text, body->bytes, body->size);
+    if (p->sent_size) memcpy(copy, p->sent, p->sent_size);
+    PureJsonMessage message = {.module = m, .shape = in, .payload = prefix, .table = bytes, .lenient = true};
+    size_t at = pure_json_space(text, 0, body->size);
+    if (at < body->size && text[at] == '{' && pure_json_message(&message, text, body->size, &at)) {
+        free(bytes);
+        return false;
+    }
+    for (uint32_t f = 0; f < in->field_count; ++f) {
+        if (message.seen[f]) continue;
+        const PureField *field = &m->field[in->field_start + f];
+        if (field->type == PURE_FIELD_U32 && !strcmp(field->name, "http_status")) {
+            write_u32(prefix + field->offset, (uint32_t)status);
+            continue;
+        }
+        /* The send's field of the same name and type, and where its bytes are. */
+        size_t start = sent->prefix, length = 0;
+        const PureField *from = NULL;
+        for (uint32_t g = 0; g < sent->field_count && !from; ++g) {
+            const PureField *other = &m->field[sent->field_start + g];
+            bool variable = other->type == PURE_FIELD_BYTES || other->type == PURE_FIELD_TEXT;
+            if (variable) length = other->offset == PURE_FIELD_REST ? p->sent_size - start : read_u32(p->sent + other->offset);
+            if (!strcmp(other->name, field->name) && other->type == field->type) from = other;
+            else if (variable) start += length;
+        }
+        bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT;
+        if (!from || (variable && (start > p->sent_size || length > p->sent_size - start))) {
+            free(bytes);
+            return false;
+        }
+        if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER)
+            memcpy(prefix + field->offset, p->sent + from->offset, field->type == PURE_FIELD_U32 ? 4 : 8);
+        else {
+            write_u32(bytes + 8 * f, (uint32_t)(copy + start - prefix));
+            write_u32(bytes + 8 * f + 4, (uint32_t)length);
+        }
+    }
+    *out = (Answer){p->seq, bytes, total, table};
+    return true;
+}
+
 /* One request for count sends from p (numbered in a row): one object, or
- * an array of them for a receiver that takes batches. */
-static int post(PureWire *w, const Pending *p, unsigned count) {
+ * 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) {
     size_t body = count > 1 ? count + 1 : 0;
     for (unsigned i = 0; i < count; ++i) body += p[i].size;
     char head[2048], batch[32] = "";
@@ -473,19 +588,30 @@ static int post(PureWire *w, const Pending *p, unsigned count) {
         int status;
         bool refused, keep, sent = send_all(w, head, (size_t)size);
         unsigned batch_size = 0;
+        Text body = {0};
         for (unsigned i = 0; sent && i < count; ++i)
             sent = (!i || count == 1 || send_all(w, ",", 1)) && (i || count == 1 || send_all(w, "[", 1)) &&
                    send_all(w, p[i].body, p[i].size);
         if (sent && count > 1) sent = send_all(w, "]", 1);
-        if (!sent || !read_answer(w, &status, &refused, &keep, &batch_size)) {
+        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;
             continue;
         }
         if (!keep) drop_connection(w);
-        if (batch_size) w->batch = batch_size;
+        if (batch_size && w->answer_input < 0) w->batch = batch_size;
+        if (status == 408 || status == 429 || status >= 500) {
+            free(body.bytes);
+            return AGAIN;
+        }
+        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;
+        }
+        free(body.bytes);
         if ((status >= 200 && status < 300) || (status >= 400 && status < 500 && refused)) return DELIVERED;
-        if (status == 408 || status == 429 || status >= 500) return AGAIN;
         return REFUSED_BY_PEER;
     }
     return AGAIN;
@@ -507,13 +633,31 @@ static void *deliver(void *context) {
         memcpy(next, w->queue + w->head, count * sizeof(*next));
         ++w->attempts;
         pthread_mutex_unlock(&w->lock);
-        int outcome = post(w, next, count);
+        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; }
+                }
+            }
+            if (outcome == DELIVERED) w->answers[w->answer_head + w->answer_count++] = answer;
+        }
         if (outcome == DELIVERED) {
-            for (unsigned i = 0; i < count; ++i) free(w->queue[w->head + i].body);
+            for (unsigned i = 0; i < count; ++i) {
+                free(w->queue[w->head + i].body);
+                free(w->queue[w->head + i].sent);
+            }
             w->head += count;
             w->count -= count;
-            w->delivered = next[count - 1].seq;
+            if (!answer.bytes) w->delivered = next[count - 1].seq;
             w->deliveries += count;
             pause_ms = 0;
             pthread_mutex_unlock(&w->lock);
@@ -521,9 +665,10 @@ static void *deliver(void *context) {
             pthread_mutex_lock(&w->lock);
             continue;
         }
-        if ((outcome == REFUSED_BY_PEER || outcome == TLS_FAILED) && reported != next[0].seq) {
+        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;
         }
@@ -557,7 +702,13 @@ void pure_wire_close(PureWire *w) {
     if (w->started) pthread_join(w->thread, NULL);
     if (w->ssl) SSL_free(w->ssl);
     if (w->fd >= 0) close(w->fd);
-    for (size_t i = 0; i < w->count; ++i) free(w->queue[w->head + i].body);
+    for (size_t i = 0; i < w->count; ++i) {
+        free(w->queue[w->head + i].body);
+        free(w->queue[w->head + i].sent);
+    }
+    for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes);
+    free(w->answers);
+    free(w->taken.bytes);
     free(w->queue);
     pthread_cond_destroy(&w->changed);
     pthread_mutex_destroy(&w->lock);
diff --git a/prototypes/connections/deliver.h b/prototypes/connections/deliver.h
index 13048fc..2d1da29 100644
--- a/prototypes/connections/deliver.h
+++ b/prototypes/connections/deliver.h
@@ -34,6 +34,27 @@ uint64_t pure_wire_delivered(PureWire *);
 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 *);
+/* Answers as events: the final answer to each send (a 2xx, or a 4xx but 408
+ * and 429) becomes a message event of input (a message without a key), in
+ * the order sent: its fields come from the answer's JSON object by name
+ * (other members are skipped), a field the answer lacks from the send's
+ * field of that name and type (as a request's id), and a u32 field named
+ * http_status is the answer's status. An answer that makes no message is
+ * logged and the send tried again every 10 s. Answers go one per request. */
+bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size);
+/* The owner's side: the next answer in order, or false. Taking it counts
+ * its send delivered (the owner applies the answer, then trims); what out
+ * points to is the wire's until the next call. The answer is applied with
+ * pure_app_deliver from the wire's source with the send's number, so an
+ * answer that comes again after a restart is applied once. */
+typedef struct {
+    uint32_t input;
+    uint64_t seq;
+    PureSpan payload;
+    const uint32_t *placed;
+    const char *source;
+} PureWireAnswer;
+bool pure_wire_answer(PureWire *, PureWireAnswer *out);
 /* Counts for a host's report. */
 void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts);
 #endif
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 88a9a6f..60771e1 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -150,7 +150,8 @@ typedef struct {
     /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */
     PureWire *wires[PURE_MODULE_INPUTS];
     unsigned wire_count;
-    uint64_t trimmed[PURE_MODULE_INPUTS];
+    uint64_t trimmed[PURE_MODULE_INPUTS], answers;
+    bool held[PURE_MODULE_INPUTS]; /* an answer failed to apply: the wire's sends stay until a restart */
 } Executor;
 
 /* A request made ready: its effect, what it names, and the thread that answers it. */
@@ -1051,14 +1052,32 @@ static void after_save(Executor *ex) {
     if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q);
 }
 
-/* What the wires delivered leaves the outbox (with the next save). */
-static void trim_delivered(Executor *ex) {
+/* The answers the wires made become events, in order; then what the wires
+ * delivered leaves the outbox (with the next save). True when an answer was
+ * applied (the caller saves). */
+static bool take_deliveries(Executor *ex) {
+    bool answered = false;
     for (unsigned i = 0; i < ex->wire_count; ++i) {
+        PureWireAnswer answer;
+        while (pure_wire_answer(ex->wires[i], &answer)) {
+            ex->app.now = wall_ms();
+            PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload,
+                                                 answer.placed);
+            if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) {
+                ex->held[i] = true; /* not applied and not refused: sent again after a restart */
+                fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n",
+                        answer.source, answer.seq);
+            }
+            pure_app_finish(&ex->app);
+            answered = true;
+            ++ex->answers;
+        }
         uint64_t delivered = pure_wire_delivered(ex->wires[i]);
-        if (delivered <= ex->trimmed[i]) continue;
+        if (delivered <= ex->trimmed[i] || ex->held[i]) continue;
         pure_app_trim(&ex->app, pure_wire_output(ex->wires[i]), delivered);
         ex->trimmed[i] = delivered;
     }
+    return answered;
 }
 
 /* One save for a burst: its requests are durable with it or fail with it. */
@@ -1290,8 +1309,14 @@ static void *run_owner(void *context) {
             owner->delivered = false;
             pthread_mutex_unlock(&owner->lock);
             long due = run_timers(&owner->executor);
-            if (delivered) trim_delivered(&owner->executor);
+            bool answered = delivered && take_deliveries(&owner->executor);
             pthread_mutex_lock(&owner->lock);
+            /* Answers are saved with the next round, or now if none waits. */
+            if (answered && !owner->count && !owner->stopping) {
+                pthread_mutex_unlock(&owner->lock);
+                if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor);
+                pthread_mutex_lock(&owner->lock);
+            }
             if (owner->count || owner->stopping) break;
             if (owner->delivered) continue;
             if (due < 0) {
@@ -1604,6 +1629,30 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen
     return true;
 }
 
+/* --reply OUTPUT=INPUT: the answers to that output's sends become INPUT's events. */
+static bool wire_replies(Executor *ex, const PureModule *m, const char *const *replies, unsigned count) {
+    for (unsigned i = 0; i < count; ++i) {
+        const char *equals = strchr(replies[i], '=');
+        char output[PURE_MODULE_NAME], error[160] = "";
+        size_t length = equals ? (size_t)(equals - replies[i]) : 0;
+        int input = equals ? pure_module_input(m, equals + 1) : -1;
+        PureWire *wire = NULL;
+        if (length && length < sizeof(output)) {
+            memcpy(output, replies[i], length);
+            output[length] = 0;
+            int index = pure_module_input(m, output);
+            for (unsigned k = 0; k < ex->wire_count && index >= 0; ++k)
+                if (pure_wire_output(ex->wires[k]) == (uint32_t)index) wire = ex->wires[k];
+        }
+        if (!wire || input < 0 || !pure_wire_reply(wire, (uint32_t)input, error, sizeof(error))) {
+            fprintf(stderr, "--reply %s: %s\n", replies[i], !wire ? "takes OUTPUT=INPUT of an output given --send" :
+                                                         input < 0 ? "no such input" : error);
+            return false;
+        }
+    }
+    return true;
+}
+
 /* --threads auto: a connection thread for each CPU this process may run on,
  * less the one the owner of an app with states takes. */
 static unsigned auto_threads(bool owner) {
@@ -1617,8 +1666,8 @@ static unsigned auto_threads(bool owner) {
 int main(int argc, char **argv) {
     static PureModule module;
     static Owner owner;
-    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS];
-    unsigned send_count = 0;
+    const char *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
+    unsigned send_count = 0, reply_count = 0;
     unsigned port = 8767, threads = 1;
     size_t budget = 64u << 20, capacity = 1024;
     uint64_t timeout_ms = 10000;
@@ -1630,6 +1679,7 @@ int main(int argc, char **argv) {
         else if (!strcmp(argv[i], "--send") && send_count < PURE_MODULE_INPUTS) sends[send_count++] = argv[i + 1];
         else if (!strcmp(argv[i], "--name")) name = argv[i + 1];
         else if (!strcmp(argv[i], "--send-ca")) pure_wire_trust(argv[i + 1]);
+        else if (!strcmp(argv[i], "--reply") && reply_count < PURE_MODULE_INPUTS) replies[reply_count++] = argv[i + 1];
         else if (!strcmp(argv[i], "--port") && number(argv[i + 1], 65535, &value)) port = (unsigned)value;
         else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
         else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
@@ -1671,7 +1721,8 @@ int main(int argc, char **argv) {
             return 1;
         owner.executor.reads_first = true;
         owner.executor.publishes = true;
-        if (!open_wires(&owner.executor, &module, sends, send_count, name)) return 2;
+        if (!open_wires(&owner.executor, &module, sends, send_count, name) ||
+            !wire_replies(&owner.executor, &module, replies, reply_count)) return 2;
         after_save(&owner.executor); /* the snapshot, and the sends stored before */
         for (unsigned i = 0; i < owner.executor.wire_count; ++i)
             if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
@@ -1731,15 +1782,16 @@ int main(int argc, char **argv) {
     fprintf(stderr, "{\"active\":%zu,\"retained_payload_bytes\":%zu,\"accepted\":%" PRIu64 ",\"requests\":%" PRIu64
             ",\"bursts\":%" PRIu64 ",\"burst_events\":%" PRIu64 ",\"ticks\":%" PRIu64 ",\"expired\":%" PRIu64
             ",\"threads\":%u,\"owner\":%s,\"snapshots\":%" PRIu64 ",\"views_from_snapshots\":%" PRIu64
-            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
+            ",\"delivered\":%" PRIu64 ",\"delivery_attempts\":%" PRIu64 ",\"answers\":%" PRIu64
+            ",\"commit_groups\":%lu,\"grouped_commits\":%lu}\n",
             active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
-            owner.executor.publications, shown, delivered, attempts, groups, grouped);
+            owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped);
     free(hosts);
     free(thread);
     return 0;
 usage:
     fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
           "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=http[s]://host:port/path ...] [--name NAME] "
-          "[--send-ca FILE]\n", stderr);
+          "[--send-ca FILE] [--reply OUTPUT=INPUT ...]\n", stderr);
     return 2;
 }