Take a message's rows over HTTP

A rows field's JSON value is an array of objects naming the record's
fields (numbers, strings; a field left out takes its default). The
connection thread encodes them as the list's stored form, which may be
longer than the JSON, grows the request's input by that much
(pure_runtime_grow: charged to the payload budget as admission charges an
input, 503 payload_budget when it does not fit) and places them after the
body, so the owner still applies a message it need not parse. A message
whose only variable field is rows takes a body that is not JSON as those
rows in their stored form: a bulk insert without JSON. An API's answer
(--reply) gives rows the same way. A delivered batch holds no rows.

server_rows loads 30,000 stock rows in one JSON request (some 250,000
rows a second, against 6,500 with a request a row) and 20,000 more as
bytes, refuses a load with a key already there whole (409 exists), then
checks 400 random orders against a model (294 applied, 106 refused with
none of their rows applied), texts with escapes and Korean, a walk that
removes, and a restart that keeps every state and no rows.

The format's document (008) describes walks, rows fields and lists of
rows, and OUTER_ROW.

1 files changed, 36 insertions, 19 deletions

diff --git a/serve/serve.c b/serve/serve.c
index 588feec..7fb3768 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -643,12 +643,14 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     if (*c->source && !message) return "delivery";
     if (!host->owner) c->source[0] = 0; /* an app without states or outputs applies it as a message */
     if (shape) {
-        /* 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). */
+        /* A body is a text input's text, a message's one variable field (rows
+         * in their list's stored form), 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;
+                         m->field[shape->field_start + f].type == PURE_FIELD_TEXT ||
+                         m->field[shape->field_start + f].type == PURE_FIELD_ROWS;
         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";
@@ -706,31 +708,45 @@ enum { DELIVERY = 80 };
 static const char *convert(Connection *c) {
     const PureInputShape *shape = &c->host->module->input[c->input];
     uint32_t delivered = get_u32(c->ingress) & DELIVERED;
-    size_t table = 8 * (size_t)shape->field_count;
+    size_t table = 8 * (size_t)shape->field_count, payload = c->batch * c->field_bytes + c->body_size;
     unsigned char *tables = c->ingress + 4 + (delivered ? DELIVERY : 0), *prefixes = tables + c->batch * table;
     unsigned char *json = c->body;
     size_t size = c->body_size, at = pure_json_space(json, 0, size);
+    /* A message's rows go after the body: they are encoded, then placed. */
+    PureJsonMessage b = {.module = c->host->module, .shape = shape, .given = c->given, .rows_base = payload};
+    const char *why = NULL;
     /* A batch is an array of its messages' objects. */
     if (c->batch > 1) {
         if (at == size || json[at++] != '[') return "json";
         at = pure_json_space(json, at, size);
     }
-    for (unsigned i = 0; i < c->batch; ++i) {
-        if (i) {
-            if (at == size || json[at++] != ',') return "json";
-            at = pure_json_space(json, at, size);
-        }
-        PureJsonMessage b = {.module = c->host->module, .shape = shape, .table = tables + i * table,
-                             .payload = prefixes + i * c->field_bytes, .given = c->given};
-        const char *why = pure_json_message(&b, json, size, &at);
-        if (why) return why;
+    for (unsigned i = 0; i < c->batch && !why; ++i) {
+        if (i && (at == size || json[at++] != ',')) why = "json";
+        at = pure_json_space(json, at, size);
+        b.table = tables + i * table;
+        b.payload = prefixes + i * c->field_bytes;
+        memset(b.seen, 0, sizeof(b.seen));
+        if (!why) why = pure_json_message(&b, json, size, &at);
         at = pure_json_space(json, at, size);
-        for (uint32_t f = 0; f < shape->field_count; ++f)
-            if (!b.seen[f] && !c->given[f]) return "missing_field";
+        for (uint32_t f = 0; !why && f < shape->field_count; ++f)
+            if (!b.seen[f] && !c->given[f]) why = "missing_field";
+        if (!why && c->batch > 1 && b.rows_size) why = "delivery"; /* an output sends no rows */
+    }
+    if (!why && c->batch > 1 && (at == size || json[at++] != ']')) why = "json";
+    if (!why && pure_json_space(json, at, size) != size) why = "json";
+    if (!why && b.rows_size) {
+        size_t body = (size_t)(c->body - c->ingress);
+        PureResult grown = pure_runtime_grow(c->ticket, body + c->body_size + b.rows_size);
+        if (grown != PURE_OK) why = grown == PURE_BUDGET ? "payload_budget" : "memory";
+        else {
+            c->ingress = pure_runtime_ingress(c->ticket, NULL);
+            c->body = c->ingress + body;
+            memcpy(c->body + c->body_size, b.rows, b.rows_size);
+        }
     }
-    if (c->batch > 1 && (at == size || json[at++] != ']')) return "json";
-    if (pure_json_space(json, at, size) != size) return "json";
-    put_u32(c->ingress, (uint32_t)(c->batch * c->field_bytes + c->body_size) | PLACED | delivered);
+    free(b.rows);
+    if (why) return why;
+    put_u32(c->ingress, (uint32_t)(payload + b.rows_size) | PLACED | delivered);
     return NULL;
 }
 
@@ -784,7 +800,8 @@ 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});
+        if (!strcmp(why, "payload_budget") || !strcmp(why, "memory")) respond(c, 503, "Service Unavailable", why, 0, (PureSpan){0});
+        else respond(c, 400, "Bad Request", why, 0, (PureSpan){0});
         return;
     }
     if (!pure_runtime_ready(c->ticket)) abort();