1 files changed, 171 insertions, 20 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);