Take messages of number and text fields, several variable ones, as JSON

A message field is now a u32, a number (binary64), bytes or text, and
variable fields may be several and anywhere: the payload's prefix holds
the fixed fields and each variable field's length but the last one's,
which takes the rest; a module whose one bytes field came last is laid
out as before. On arrival a length past the payload, a text that is not
UTF-8 or a number that is not finite fails the message.

pure-serve fills an input's message from a JSON object body
(Content-Type: application/json) on the connection thread before the
request is ready, so the owner never parses and replies are still
borrowed: the ingress is [u32 payload size][payload][the JSON as it
came]. A missing, unknown, doubled or mistyped field, or text that is not
JSON, answers 400 and the request never runs. Numbers may come in the
query too.

4 files changed, 368 insertions, 48 deletions

diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index 3490ba9..7f22c49 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -315,7 +315,10 @@ PURE_COLD static const char *verify_block(const PureModule *m, PureBlock block,
             of = O_ARG;
             owner = arg >> 8 & 0xFF;
         }
-        if (out == T_FIELD) out = m->field[shape->field_start + arg].type == PURE_FIELD_U32 ? T_NUM : T_BYTES;
+        if (out == T_FIELD) {
+            uint8_t type = m->field[shape->field_start + arg].type;
+            out = type == PURE_FIELD_TEXT ? T_TEXT : type == PURE_FIELD_BYTES ? T_BYTES : T_NUM;
+        }
         if (out == T_ITEM) out = field_type(m, list_of(arg), item_field(arg));
         if (out) {
             if (height == PURE_MODULE_STACK) REFUSE("stack overflow");
@@ -445,19 +448,24 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
         }
         if (shape->kind != PURE_KIND_MESSAGE || read32(at + 8) || shape->field_start != next_field ||
             shape->field_count > out->fields - next_field) REFUSE("message shape", i);
-        /* Fixed u32le fields in order, then at most one bytes tail. */
+        /* The prefix holds the fixed fields and the variable fields' lengths
+         * in order, but the last variable field's: it takes the rest. */
+        uint32_t last = UINT32_MAX;
+        for (uint32_t f = 0; f < shape->field_count; ++f) {
+            uint32_t type = read32(fields + (size_t)(next_field + f) * FIELD_BYTES + 4);
+            if (type == PURE_FIELD_BYTES || type == PURE_FIELD_TEXT) last = f;
+        }
         for (uint32_t f = 0; f < shape->field_count; ++f) {
             const unsigned char *entry = fields + (size_t)(next_field + f) * FIELD_BYTES;
             PureField *field = &out->field[next_field + f];
             field->type = (uint8_t)read32(entry + 4);
-            if (!name_at(text, strings, read32(entry), true, field->name) ||
-                (field->type != PURE_FIELD_U32 && field->type != PURE_FIELD_BYTES) || read32(entry + 4) > 2 ||
-                (field->type == PURE_FIELD_BYTES && f + 1 != shape->field_count))
+            if (!name_at(text, strings, read32(entry), true, field->name) || field->type < PURE_FIELD_U32 ||
+                read32(entry + 4) > PURE_FIELD_TEXT)
                 REFUSE("message field", next_field + f);
             for (uint32_t k = 0; k < f; ++k)
                 if (!strcmp(out->field[next_field + k].name, field->name)) REFUSE("duplicate field", next_field + f);
-            field->offset = shape->prefix;
-            if (field->type == PURE_FIELD_U32) shape->prefix += 4;
+            field->offset = f == last ? PURE_FIELD_REST : shape->prefix;
+            if (f != last) shape->prefix += field->type == PURE_FIELD_NUMBER ? 8 : 4;
         }
         next_field += shape->field_count;
     }
@@ -610,6 +618,7 @@ typedef struct {
     double dt;
     size_t row; /* the row a column is evaluated for */
     bool modified, exists;
+    uint32_t (*spans)[2]; /* a message's variable fields: start and size in the payload */
 } Event;
 
 static PureResult fault(PureApp *app, PureFault kind, uint32_t pc) {
@@ -776,7 +785,8 @@ static PureResult run(const PureModule *m, double *values, PureBlock block, Even
         case PURE_OP_FIELD: {
             const PureField *field = &m->field[ev->shape->field_start + arg];
             if (field->type == PURE_FIELD_U32) stack[sp++].num = read32(app->payload.data + field->offset);
-            else stack[sp++].bytes = (PureBytes){FROM_PAYLOAD, 0, {0, 0}, field->offset, app->payload.size - field->offset};
+            else if (field->type == PURE_FIELD_NUMBER) stack[sp++].num = read_double(app->payload.data + field->offset);
+            else stack[sp++].bytes = (PureBytes){FROM_PAYLOAD, 0, {0, 0}, ev->spans[arg][0], ev->spans[arg][1]};
             continue;
         }
         case PURE_OP_DOC: case PURE_OP_DOC_SIZE: {
@@ -1313,6 +1323,23 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
         return PURE_INVALID;
     const PureInputShape *shape = &m->input[input];
     if (payload.size < shape->prefix) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
+    /* Where each variable field is; a length past the payload, a text that
+     * is not UTF-8 or a number that is not finite fails the message. */
+    uint32_t spans[PURE_MODULE_FIELDS][2];
+    size_t at = shape->prefix;
+    for (uint32_t f = 0; f < shape->field_count; ++f) {
+        const PureField *field = &m->field[shape->field_start + f];
+        if (field->type == PURE_FIELD_NUMBER && !isfinite(read_double(payload.data + field->offset)))
+            return fault(app, PURE_FAULT_NON_FINITE, UINT32_MAX);
+        if (field->type != PURE_FIELD_BYTES && field->type != PURE_FIELD_TEXT) continue;
+        size_t size = field->offset == PURE_FIELD_REST ? payload.size - at : read32(payload.data + field->offset);
+        if (size > payload.size - at) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
+        if (field->type == PURE_FIELD_TEXT && !pure_utf8_valid((PureSpan){size ? payload.data + at : NULL, size}))
+            return fault(app, PURE_FAULT_ENCODING, UINT32_MAX);
+        spans[f][0] = (uint32_t)at;
+        spans[f][1] = (uint32_t)size;
+        at += size;
+    }
     if (shape->keyed && m->collections) {
         if (!app->data.open) return PURE_INVALID;
         /* A key joins a burst once, so its revision moves once per save. */
@@ -1334,7 +1361,8 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
         if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; }
     }
     app->payload = payload;
-    Event ev = {.app = app, .shape = shape, .key = key, .base = base, .arena = &app->arena, .exists = app->revision > 0};
+    Event ev = {.app = app, .shape = shape, .key = key, .base = base, .arena = &app->arena, .exists = app->revision > 0,
+                .spans = spans};
     PureResult result = dispatch(app, 1 + 2 * input, &ev);
     if (result != PURE_OK) pure_app_finish(app);
     return result;
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index 09a8540..52671f0 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -51,7 +51,14 @@ typedef enum { PURE_FOLD_COUNT, PURE_FOLD_SUM, PURE_FOLD_MAX, PURE_FOLD_MIN, PUR
 typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PURE_EQUAL, PURE_NOT_EQUAL } PureRelation;
 
 typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3 } PureInputKind;
-typedef enum { PURE_FIELD_U32 = 1, PURE_FIELD_BYTES = 2 } PureFieldType;
+/* A message field: a u32le or a binary64 (fixed, at its offset in the
+ * payload's prefix), or bytes or text of any length. The prefix also holds
+ * each variable field's u32le length but the last one's (its offset is
+ * PURE_FIELD_REST), which takes what follows the others; the variable
+ * fields' bytes follow the prefix in declaration order. A text is UTF-8 and
+ * a number finite, or the message fails. */
+typedef enum { PURE_FIELD_U32 = 1, PURE_FIELD_BYTES = 2, PURE_FIELD_NUMBER = 3, PURE_FIELD_TEXT = 4 } PureFieldType;
+enum { PURE_FIELD_REST = UINT32_MAX };
 typedef enum {
     PURE_STATE_NUM = 0, PURE_STATE_TEXT = 1, PURE_STATE_CURSOR = 2, PURE_STATE_LIST = 3, PURE_STATE_INDEX = 4
 } PureStateType;
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 2a1587d..2cb69c1 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -38,6 +38,7 @@
 #include <fcntl.h>
 #include <inttypes.h>
 #include <limits.h>
+#include <math.h>
 #include <netinet/in.h>
 #include <netinet/tcp.h>
 #include <pthread.h>
@@ -77,8 +78,14 @@ typedef struct Connection {
     uint64_t deadline;
     char key[65]; /* the document's key, or the view's or input's name */
     RouteKind route;
-    unsigned char fields[4 * PURE_MODULE_FIELDS]; /* a message's u32 fields, from the query */
+    /* An app input's message: its fixed fields from the query, which fields
+     * were given there, and whether the body is a JSON object of the rest.
+     * Its ingress is [u32le payload size][payload][the JSON as it came]. */
+    unsigned char fields[8 * PURE_MODULE_FIELDS];
     size_t field_bytes;
+    bool given[PURE_MODULE_FIELDS], json;
+    uint32_t input;
+    unsigned char *ingress;
     int64_t expected;
     size_t body_size, received, in_used;
     unsigned char *body;
@@ -364,6 +371,7 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
         (c->route == ROUTE_INPUT && !c->post)) return false;
     bool length_seen = false, condition_seen = false, host_seen = false;
     c->body_size = 0;
+    c->json = false;
     c->expected = -1;
     c->keep_alive = true;
     line = next + 2;
@@ -398,6 +406,8 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
             if (condition_seen || strcmp(value, "*")) return false;
             c->expected = 0;
             condition_seen = true;
+        } else if (!strcasecmp(line, "Content-Type")) {
+            c->json = !strncasecmp(value, "application/json", 16);
         } else if (!strcasecmp(line, "Connection")) {
             if (!strcasecmp(value, "close")) c->keep_alive = false;
             else if (strcasecmp(value, "keep-alive")) return false;
@@ -409,13 +419,59 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
     return host_seen && !condition_seen && c->body_size == 0;
 }
 
-/* What an app route names in the module, with a message's u32 fields taken
- * from the query into c->fields. NULL, or the error to answer (404 for a
- * name the module lacks, 400 otherwise). */
+static void put_u32(unsigned char *p, uint32_t v) { for (unsigned i = 0; i < 4; ++i) p[i] = (unsigned char)(v >> (8 * i)); }
+static uint32_t get_u32(const unsigned char *p) { return p[0] | p[1] << 8 | (uint32_t)p[2] << 16 | (uint32_t)p[3] << 24; }
+
+/* A JSON number's text: -?(0|[1-9][0-9]*)(.[0-9]+)?([eE][+-]?[0-9]+)?, its
+ * length (0 if there is none). An integer has neither part after it. */
+static size_t json_number(const unsigned char *p, size_t size, bool *integer) {
+    size_t i = 0;
+    *integer = true;
+    if (i < size && p[i] == '-') ++i;
+    if (i == size || p[i] < '0' || p[i] > '9') return 0;
+    if (p[i++] != '0') while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+    if (i < size && p[i] == '.') {
+        *integer = false;
+        if (++i == size || p[i] < '0' || p[i] > '9') return 0;
+        while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+    }
+    if (i < size && (p[i] == 'e' || p[i] == 'E')) {
+        *integer = false;
+        if (++i < size && (p[i] == '+' || p[i] == '-')) ++i;
+        if (i == size || p[i] < '0' || p[i] > '9') return 0;
+        while (i < size && p[i] >= '0' && p[i] <= '9') ++i;
+    }
+    return i;
+}
+
+/* A fixed field's value from its text (a query value or a JSON number) into
+ * the prefix: a u32 in range, or a finite number. */
+static bool fixed_field(const PureField *field, const unsigned char *text, size_t size, unsigned char *prefix) {
+    bool integer;
+    if (!size || json_number(text, size, &integer) != size || size > 40) return false;
+    char digits[48];
+    memcpy(digits, text, size);
+    digits[size] = 0;
+    if (field->type == PURE_FIELD_U32) {
+        uint64_t n;
+        if (!integer || !number(digits, UINT32_MAX, &n)) return false;
+        put_u32(prefix + field->offset, (uint32_t)n);
+        return true;
+    }
+    double value = strtod(digits, NULL);
+    if (!isfinite(value)) return false;
+    memcpy(prefix + field->offset, &value, 8);
+    return true;
+}
+
+/* What an app route names in the module, with a message's fixed fields
+ * taken from the query into c->fields. NULL, or the error to answer (404
+ * for a name the module lacks, 400 otherwise). */
 static const char *resolve(const Host *host, Connection *c, char *query, Route *route, int *status) {
     const PureModule *m = host->module;
     *status = 404;
     c->field_bytes = 0;
+    memset(c->given, 0, sizeof(c->given));
     if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents)) return "route";
     if (route->kind == ROUTE_OBJECT) return NULL;
     int found = route->kind == ROUTE_VIEW ? pure_module_view(m, c->key) : pure_module_input(m, c->key);
@@ -428,39 +484,195 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     bool message = shape && shape->kind == PURE_KIND_MESSAGE;
     if (message && shape->keyed) return "keyed";
     if (shape) {
-        const PureField *last = message && shape->field_count ? &m->field[shape->field_start + shape->field_count - 1] : NULL;
-        if (c->body_size && shape->kind != PURE_KIND_TEXT && !(last && last->type == PURE_FIELD_BYTES)) return "body";
+        /* A body is a text input's text, a message's one variable field, or a
+         * JSON object of a message's fields (two variable fields need one). */
+        unsigned variables = 0;
+        for (uint32_t f = 0; message && f < shape->field_count; ++f)
+            variables += m->field[shape->field_start + f].type == PURE_FIELD_BYTES ||
+                         m->field[shape->field_start + f].type == PURE_FIELD_TEXT;
+        if (c->json && !message) return "json";
+        if (!c->json && (variables > 1 || (c->body_size && shape->kind == PURE_KIND_KEY) ||
+                         (c->body_size && message && !variables))) return variables > 1 ? "json" : "body";
+        c->input = (uint32_t)found;
         c->field_bytes = message ? shape->prefix : 0;
     }
-    /* name=value pairs: a table's from and count, a message's u32 fields. */
-    bool given[PURE_MODULE_FIELDS] = {false};
+    /* name=value pairs: a table's from and count, a message's fixed fields. */
     for (char *pair = query; pair && *pair;) {
         char *end = strchr(pair, '&'), *value = strchr(pair, '=');
         if (end) *end = 0;
         if (!value || (end && value > end)) return "query";
         *value++ = 0;
-        uint64_t n;
-        if (!number(value, route->kind == ROUTE_VIEW ? SIZE_MAX : UINT32_MAX, &n)) return "query";
         if (route->kind == ROUTE_VIEW) {
+            uint64_t n;
             bool from = !strcmp(pair, "from");
-            if (m->view_type[found] != PURE_VIEW_TABLE || (!from && strcmp(pair, "count"))) return "query";
+            if (!number(value, SIZE_MAX, &n) || m->view_type[found] != PURE_VIEW_TABLE || (!from && strcmp(pair, "count")))
+                return "query";
             if (from) route->from = (size_t)n;
             else route->count = (size_t)n;
         } else {
             uint32_t f = 0;
             while (f < (message ? shape->field_count : 0) && strcmp(m->field[shape->field_start + f].name, pair)) ++f;
             const PureField *field = message && f < shape->field_count ? &m->field[shape->field_start + f] : NULL;
-            if (!field || field->type != PURE_FIELD_U32 || given[f]) return "query";
-            given[f] = true;
-            for (unsigned i = 0; i < 4; ++i) c->fields[field->offset + i] = (unsigned char)(n >> (8 * i));
+            if (!field || c->given[f] || (field->type != PURE_FIELD_U32 && field->type != PURE_FIELD_NUMBER) ||
+                !fixed_field(field, (const unsigned char *)value, strlen(value), c->fields)) return "query";
+            c->given[f] = true;
         }
         pair = end ? end + 1 : NULL;
     }
-    for (uint32_t f = 0; message && f < shape->field_count; ++f)
-        if (m->field[shape->field_start + f].type == PURE_FIELD_U32 && !given[f]) return "missing_field";
+    for (uint32_t f = 0; message && !c->json && f < shape->field_count; ++f) {
+        uint8_t type = m->field[shape->field_start + f].type;
+        if ((type == PURE_FIELD_U32 || type == PURE_FIELD_NUMBER) && !c->given[f]) return "missing_field";
+    }
+    return NULL;
+}
+
+/* A JSON string's contents (after its opening quote) up to the closing one:
+ * its length, or 0 when it does not close or holds a control byte. */
+static size_t json_string_end(const unsigned char *p, size_t size) {
+    for (size_t i = 0; i < size; ++i) {
+        if (p[i] == '"') return i + 1;
+        if (p[i] < 0x20) return 0;
+        if (p[i] == '\\' && ++i == size) return 0;
+    }
+    return 0;
+}
+
+static int hex_digit(unsigned char c) {
+    return c >= '0' && c <= '9' ? c - '0' : c >= 'a' && c <= 'f' ? c - 'a' + 10 : c >= 'A' && c <= 'F' ? c - 'A' + 10 : -1;
+}
+
+static bool json_code(const unsigned char *p, size_t left, uint32_t *code) {
+    if (left < 4) return false;
+    *code = 0;
+    for (int i = 0; i < 4; ++i) {
+        int d = hex_digit(p[i]);
+        if (d < 0) return false;
+        *code = *code << 4 | (uint32_t)d;
+    }
+    return true;
+}
+
+/* Unescape a JSON string's contents into out (never longer); false for a
+ * bad escape or a lone surrogate. The result is checked as UTF-8 after. */
+static bool json_unescape(const unsigned char *in, size_t size, unsigned char *out, size_t *written) {
+    size_t o = 0;
+    for (size_t i = 0; i < size;) {
+        unsigned char b = in[i++];
+        if (b != '\\') { out[o++] = b; continue; }
+        if (i == size) return false;
+        b = in[i++];
+        const char *plain = strchr("\"\\/bfnrt", b);
+        if (b && plain) { out[o++] = (unsigned char)"\"\\/\b\f\n\r\t"[plain - "\"\\/bfnrt"]; continue; }
+        uint32_t code, low;
+        if (b != 'u' || !json_code(in + i, size - i, &code)) return false;
+        i += 4;
+        if (code >= 0xDC00 && code <= 0xDFFF) return false;
+        if (code >= 0xD800 && code <= 0xDBFF) {
+            if (size - i < 6 || in[i] != '\\' || in[i + 1] != 'u' || !json_code(in + i + 2, size - i - 2, &low) ||
+                low < 0xDC00 || low > 0xDFFF) return false;
+            i += 6;
+            code = 0x10000 + ((code - 0xD800) << 10) + (low - 0xDC00);
+        }
+        if (code < 0x80) out[o++] = (unsigned char)code;
+        else if (code < 0x800) { out[o++] = (unsigned char)(0xC0 | code >> 6); out[o++] = (unsigned char)(0x80 | (code & 0x3F)); }
+        else if (code < 0x10000) {
+            out[o++] = (unsigned char)(0xE0 | code >> 12);
+            out[o++] = (unsigned char)(0x80 | (code >> 6 & 0x3F));
+            out[o++] = (unsigned char)(0x80 | (code & 0x3F));
+        } else {
+            out[o++] = (unsigned char)(0xF0 | code >> 18);
+            out[o++] = (unsigned char)(0x80 | (code >> 12 & 0x3F));
+            out[o++] = (unsigned char)(0x80 | (code >> 6 & 0x3F));
+            out[o++] = (unsigned char)(0x80 | (code & 0x3F));
+        }
+    }
+    *written = o;
+    return true;
+}
+
+static size_t json_space(const unsigned char *p, size_t at, size_t size) {
+    while (at < size && (p[at] == ' ' || p[at] == '\t' || p[at] == '\n' || p[at] == '\r')) ++at;
+    return at;
+}
+
+/* An app input's JSON body, as it came after the payload's room, becomes its
+ * message's payload: an object with a member for each field the query did
+ * not give (a u32, a number, or a string for bytes and text), and no other.
+ * NULL, or the error to answer. */
+static const char *convert(Connection *c) {
+    const PureModule *m = c->host->module;
+    const PureInputShape *shape = &m->input[c->input];
+    const unsigned char *json = c->body;
+    size_t size = c->body_size;
+    unsigned char *payload = c->ingress + 4;
+    struct { const unsigned char *value; size_t size; bool string; } member[PURE_MODULE_FIELDS];
+    bool seen[PURE_MODULE_FIELDS] = {false};
+    size_t at = json_space(json, 0, size);
+    if (at == size || json[at++] != '{') return "json";
+    at = json_space(json, at, size);
+    for (bool first = true; at < size && json[at] != '}'; first = false) {
+        if (!first) {
+            if (json[at++] != ',') return "json";
+            at = json_space(json, at, size);
+        }
+        size_t key_end = at < size && json[at] == '"' ? json_string_end(json + at + 1, size - at - 1) : 0;
+        if (!key_end) return "json";
+        const unsigned char *key = json + at + 1;
+        size_t key_size = key_end - 1;
+        at = json_space(json, at + 1 + key_end, size);
+        if (at == size || json[at++] != ':') return "json";
+        at = json_space(json, at, size);
+        uint32_t f = 0;
+        while (f < shape->field_count && (strlen(m->field[shape->field_start + f].name) != key_size ||
+                                          memcmp(m->field[shape->field_start + f].name, key, key_size))) ++f;
+        if (f == shape->field_count || seen[f] || c->given[f]) return "json_field";
+        seen[f] = true;
+        bool integer, string = at < size && json[at] == '"';
+        size_t length = string ? json_string_end(json + at + 1, size - at - 1) : json_number(json + at, size - at, &integer);
+        if (!length) return "json";
+        member[f].value = json + at + string;
+        member[f].size = length - string;
+        member[f].string = string;
+        at = json_space(json, at + string + length, size);
+    }
+    if (at == size || json_space(json, at + 1, size) != size) return "json";
+    /* The fields in order: fixed ones into the prefix, strings after it. */
+    size_t used = shape->prefix;
+    for (uint32_t f = 0; f < shape->field_count; ++f) {
+        const PureField *field = &m->field[shape->field_start + f];
+        bool variable = field->type == PURE_FIELD_BYTES || field->type == PURE_FIELD_TEXT;
+        if (!seen[f]) {
+            if (!c->given[f]) return "missing_field";
+            continue;
+        }
+        if (member[f].string != variable) return "json_type";
+        if (!variable) {
+            if (!fixed_field(field, member[f].value, member[f].size, payload)) return "json_type";
+            continue;
+        }
+        size_t written;
+        if (!json_unescape(member[f].value, member[f].size, payload + used, &written) ||
+            !pure_utf8_valid((PureSpan){payload + used, written})) return "json";
+        if (field->offset != PURE_FIELD_REST) put_u32(payload + field->offset, (uint32_t)written);
+        used += written;
+    }
+    put_u32(c->ingress, (uint32_t)used);
     return NULL;
 }
 
+/* A request's body is in: an app input's JSON becomes its payload, and the
+ * request is ready, or it is answered 400 and never runs. */
+static void body_done(Connection *c) {
+    const char *why = c->route == ROUTE_INPUT && c->json ? convert(c) : NULL;
+    if (why) {
+        release_ticket(c); /* never made ready: it never runs */
+        respond(c, 400, "Bad Request", why, 0, (PureSpan){0});
+        return;
+    }
+    if (!pure_runtime_ready(c->ticket)) abort();
+    c->phase = WAITING;
+}
+
 /* A whole head is in `in`: admit the request and take any body bytes that
  * came with it; later bytes stay for the next request. */
 static void start_request(Connection *c, size_t head_size) {
@@ -477,11 +689,13 @@ static void start_request(Connection *c, size_t head_size) {
     int status;
     const char *why = resolve(host, c, query, &route, &status);
     if (why) { reject(c, status, status == 404 ? "Not Found" : "Bad Request", why); return; }
-    /* A document's PUT and an input's event carry a payload: the message's fields, then the body. */
+    /* A document's PUT carries its body; an input's event its payload's size,
+     * the payload (the fixed fields, then the body), and room for its JSON. */
     bool write = route.kind == ROUTE_INPUT || c->put;
+    size_t input_size = route.kind == ROUTE_INPUT ? 4 + c->field_bytes + c->body_size * (c->json ? 2 : 1) :
+                        write ? c->body_size : 0;
     PureResult admitted = pure_runtime_accept(&host->runtime, write ? PURE_INPUT_WRITE : PURE_INPUT_READ, c->key,
-                                              route.kind == ROUTE_INPUT ? 0 : c->expected,
-                                              write ? c->field_bytes + c->body_size : 0, &c->ticket);
+                                              route.kind == ROUTE_INPUT ? 0 : c->expected, input_size, &c->ticket);
     if (admitted != PURE_OK) {
         c->ticket = (PureTicket){0};
         /* An unread body would desynchronize the stream. */
@@ -492,19 +706,20 @@ static void start_request(Connection *c, size_t head_size) {
     host->by_slot[c->ticket.slot] = c;
     host->routes[c->ticket.slot] = route;
     if (!write) { c->phase = WAITING; return; }
-    unsigned char *ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes, as a key press */
-    if (c->field_bytes) memcpy(ingress, c->fields, c->field_bytes);
-    c->body = ingress ? ingress + c->field_bytes : NULL;
+    unsigned char *ingress = c->ingress = pure_runtime_ingress(c->ticket, NULL); /* NULL for no bytes */
+    c->body = ingress;
+    if (route.kind == ROUTE_INPUT) {
+        put_u32(ingress, (uint32_t)(c->field_bytes + c->body_size));
+        if (c->field_bytes) memcpy(ingress + 4, c->fields, c->field_bytes);
+        c->body = ingress + 4 + c->field_bytes + (c->json ? c->body_size : 0);
+    }
     size_t take = c->in_used < c->body_size ? c->in_used : c->body_size;
     if (take) memcpy(c->body, c->in, take);
     memmove(c->in, c->in + take, c->in_used - take);
     c->in_used -= take;
     c->received = take;
     c->phase = READING_BODY;
-    if (c->received == c->body_size) {
-        if (!pure_runtime_ready(c->ticket)) abort();
-        c->phase = WAITING;
-    }
+    if (c->received == c->body_size) body_done(c);
 }
 
 static size_t head_end(const Connection *c) {
@@ -534,10 +749,7 @@ static void drive(Connection *c) {
             if (n > 0) {
                 c->received += (size_t)n;
                 c->deadline = now_ms() + c->host->timeout_ms;
-                if (c->received == c->body_size) {
-                    if (!pure_runtime_ready(c->ticket)) abort();
-                    c->phase = WAITING;
-                }
+                if (c->received == c->body_size) body_done(c);
                 continue;
             }
             if (n < 0 && errno == EAGAIN) { c->readable = false; return; }
@@ -741,14 +953,15 @@ static PureResult show_view(PureApp *app, const Route *route, size_t available,
  * whose reply (a field of it) is borrowed from the request. */
 static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) {
     const PureInputShape *shape = &app->module->input[route->target];
+    PureSpan payload = {effect->value.data + 4, get_u32(effect->value.data)};
     PureResult status;
     if (shape->kind == PURE_KIND_TEXT) {
-        status = pure_app_text(app, route->target, effect->value);
+        status = pure_app_text(app, route->target, payload);
     } else if (shape->kind == PURE_KIND_KEY) {
         status = pure_app_input(app, route->target, PURE_EDGE_DOWN);
         if (status == PURE_OK) status = pure_app_input(app, route->target, PURE_EDGE_UP);
     } else {
-        status = pure_app_message(app, route->target, NULL, -1, effect->value);
+        status = pure_app_message(app, route->target, NULL, -1, payload);
         Reply reply = {0};
         if (status == PURE_OK) pure_app_reply(app, measure_reply, &reply);
         out->reply = (PureSpan){reply.size ? reply.first : NULL, reply.size};
@@ -994,9 +1207,12 @@ static bool replies_borrow(const PureModule *m, uint32_t input) {
     for (uint32_t h = 0; h < m->handlers; ++h) {
         if (m->handler[h].event != 1 + 2 * input) continue;
         const PureBlock *code = &m->handler[h].code;
-        for (uint32_t pc = code->start; pc < code->start + code->length; ++pc)
-            if (m->code[pc].op == PURE_OP_REPLY && (pc == code->start || m->code[pc - 1].op != PURE_OP_FIELD))
-                return false;
+        /* A field replied as it is, or a text field as its bytes. */
+        for (uint32_t pc = code->start; pc < code->start + code->length; ++pc) {
+            if (m->code[pc].op != PURE_OP_REPLY) continue;
+            uint32_t from = pc > code->start && m->code[pc - 1].op == PURE_OP_BYTES_FROM ? pc - 1 : pc;
+            if (from == code->start || m->code[from - 1].op != PURE_OP_FIELD) return false;
+        }
     }
     return true;
 }
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index 11d0660..c881678 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -627,8 +627,8 @@ static void messages(void) {
     static const H note_handler[] = {{1, 0, 2}};
     bad = (Spec){states, 1, unkeyed, 1, views, 0, note_handler, 1, doc_read, 2, collections, 1, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
     CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "document operation outside"));
-    static const F tail_first[] = {{"insert", 2}, {"start", 1}};
-    static const I misplaced[] = {{"edit", NULL, PURE_KIND_MESSAGE, true, tail_first, 2}};
+    static const F unknown_type[] = {{"insert", 5}, {"start", 1}};
+    static const I misplaced[] = {{"edit", NULL, PURE_KIND_MESSAGE, true, unknown_type, 2}};
     bad = (Spec){states, 1, misplaced, 1, views, 0, NULL, 0, NULL, 0, collections, 1, NULL, 0, NULL, 0, NULL, 0, NULL, 0, NULL, 0};
     CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "message field"));
     static const H even[] = {{2, 0, 0}};
@@ -2191,12 +2191,81 @@ static void keyed_mutations(void) {
     printf("{\"keyed_module_mutations\":20000,\"accepted_after_damage\":%u}\n", accepted);
 }
 
+/* A message of a text, a number, bytes and a u32, in that order: the prefix
+ * holds the text's length, the number and the u32; the text's bytes come
+ * first after it and the bytes (the last variable field) take the rest. A
+ * length past the payload, a text that is not UTF-8 or a number that is not
+ * finite fails the message. */
+enum { TSET = PURE_OP_TEXT_SET };
+static void message_fields(void) {
+    static const S nums[] = {{"n", 0}};
+    static const T texts[] = {{"title", 1, 0}};
+    static const char *const constants[] = {""};
+    static const F fields[] = {{"title", PURE_FIELD_TEXT}, {"points", PURE_FIELD_NUMBER}, {"note", PURE_FIELD_BYTES},
+                               {"count", PURE_FIELD_U32}};
+    static const I inputs[] = {{"post", NULL, PURE_KIND_MESSAGE, false, fields, 4}};
+    static const O code[] = {{FIELD, 1, 0}, {FIELD, 3, 0}, {ADD, 0, 0}, {ST, 0, 0}, {FIELD, 0, 0}, {TSET, 1, 0},
+                             {FIELD, 2, 0}, {REPLY, 0, 0}};
+    static const H handlers[] = {{1, 0, 8}};
+    Spec spec = {nums, 1, inputs, 1, NULL, 0, handlers, 1, code, LEN(code), NULL, 0, texts, 1, constants, 1, NULL, 0,
+                 NULL, 0, NULL, 0};
+    if (load(spec, &module_a) != PURE_OK) { fprintf(stderr, "%s\n", error); abort(); }
+    CHECK(module_a.input[0].prefix == 16 && module_a.field[0].offset == 0 && module_a.field[1].offset == 4 &&
+          module_a.field[2].offset == PURE_FIELD_REST && module_a.field[3].offset == 12);
+    PureApp app;
+    CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK);
+    unsigned char payload[64];
+    /* title "가나" (6 bytes), points 2.5, count 3, then the title's bytes and note "xyz". */
+    size_t size = 0;
+    uint32_t title_size = 6, count = 3;
+    double points = 2.5;
+    memcpy(payload, &title_size, 4);
+    memcpy(payload + 4, &points, 8);
+    memcpy(payload + 12, &count, 4);
+    memcpy(payload + 16, "\xea\xb0\x80\xeb\x82\x98" "xyz", 9);
+    size = 25;
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, size}) == PURE_OK);
+    CHECK(app.values[0] == 5.5 && !strcmp(state_text(&app, 1), "\xea\xb0\x80\xeb\x82\x98") && replied(&app, "xyz"));
+    pure_app_finish(&app);
+    /* Empty variable fields. */
+    title_size = 0;
+    memcpy(payload, &title_size, 4);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 16}) == PURE_OK && !*state_text(&app, 1) && replied(&app, ""));
+    pure_app_finish(&app);
+    /* A title longer than what is there, a title that is not UTF-8, a number that is not finite, a short prefix. */
+    title_size = 10;
+    memcpy(payload, &title_size, 4);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 25}) == PURE_LIMIT && app.fault == PURE_FAULT_LIMIT);
+    title_size = 2;
+    memcpy(payload, &title_size, 4);
+    memcpy(payload + 16, "\xff\xfe", 2);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 18}) == PURE_INVALID && app.fault == PURE_FAULT_ENCODING);
+    title_size = 0;
+    points = NAN;
+    memcpy(payload, &title_size, 4);
+    memcpy(payload + 4, &points, 8);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 16}) == PURE_INVALID && app.fault == PURE_FAULT_NON_FINITE);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, 15}) == PURE_LIMIT);
+    CHECK(app.values[0] == 5.5 && !*state_text(&app, 1)); /* failed messages changed nothing */
+    pure_app_stop(&app);
+    /* A text field is text and a bytes field bytes. */
+    static const O text_as_bytes[] = {{FIELD, 0, 0}, {LEN_, 0, 0}, {ST, 0, 0}};
+    static const O bytes_as_text[] = {{FIELD, 2, 0}, {TSET, 1, 0}};
+    static const H three[] = {{1, 0, 3}}, two[] = {{1, 0, 2}};
+    Spec bad = spec;
+    bad.o = text_as_bytes; bad.no = 3; bad.h = three;
+    CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "operand type"));
+    bad.o = bytes_as_text; bad.no = 2; bad.h = two;
+    CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "operand type"));
+}
+
 int main(void) {
     behaviour();
     faults();
     rejections();
     mutations();
     messages();
+    message_fields();
     message_mutations();
     texts();
     marks();
@@ -2214,6 +2283,6 @@ int main(void) {
     stored_states();
     deferred();
     printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp));
-    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order\"}");
+    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts-pointing-marks-folds-filters-keyed-lists-text-order-message-fields\"}");
     return 0;
 }