Take a delivery's fields by name, dropping none

A message from another node was taken only in its input's exact shape
(the link's STREAM shape, fields and order alike; over HTTP, every
field). So a board whose nodes are upgraded one after the other stopped
both ways until both were, and an output and its input whose fields
differ only in order were refused for good (docs/design/014).

An input now takes a message when it has every field the sender sends,
by name, of the same type or one that holds its values (a u32 in a
number, text in bytes); a field only the input has is 0 or empty. One
that sends a field the input lacks is refused: taking it would drop the
value, and with its number heard the value would never come again (two
boards apart for good). So add fields to receivers first and take them
from senders first, and nothing stops or is lost.

The link carries a sender's fields in a new frame, FIELDS (0x0d: names,
shape, u8 count, each {type, name length, name}): after a REFUSED for
another shape the sender puts it in its packets (after OFFER, before
the records, so a receiver that knows no FIELDS reads what it read
before) until the stream is taken, the first time at once. A receiver
seeing another shape waits for the packet's FIELDS before refusing, and
asks its host with them (PureLinkHost.input takes them); REFUSED says 4
when the input lacks a field. Deliveries carry the fields and the host
lays each record out anew (pure_module_takes, pure_module_recast,
pure_module_fields, over pure_send_recast's rule). A stream whose sender
starts again in another shape lets go what it held; the sender sends it
again laid out anew. A refusal already queued for a stream now says the
last reason judged, not the first. Over HTTP a delivery (X-Pure-Source)
may lack fields; an unknown member is still refused.

test-link named: a stream of another shape naming its fields is taken,
200 stored in 47 ms with every delivery naming them; an input that does
not take them refuses (4) and the sender backs off. check.py
server_fields: five outputs, three linked and two over HTTP, fields in
another order taken, an input with a field more taking messages without
it, one lacking a field refusing until it starts again with it, then
taking all. server_upgrade's new receiver declares its fields in another
order. 012 names the frame and the rules.

2 files changed, 82 insertions, 1 deletions

diff --git a/core/module.c b/core/module.c
index 97270ef..ea6809e 100644
--- a/core/module.c
+++ b/core/module.c
@@ -1211,6 +1211,71 @@ uint32_t pure_module_shape(const PureModule *m, uint32_t input) {
     return h;
 }
 
+size_t pure_module_fields(const PureModule *m, uint32_t input, unsigned char out[PURE_FIELDS_NAMED]) {
+    const PureInputShape *s = &m->input[input];
+    size_t at = 0;
+    out[at++] = (unsigned char)s->field_count;
+    for (uint32_t f = 0; f < s->field_count; ++f) {
+        const PureField *field = &m->field[s->field_start + f];
+        size_t length = strlen(field->name);
+        out[at++] = field->type;
+        out[at++] = (unsigned char)length;
+        memcpy(out + at, field->name, length);
+        at += length;
+    }
+    return at;
+}
+
+/* Fields as a sender names them, laid out: their count, or -1 when they are
+ * not well named (no rows: a sender sends none). */
+static int named_fields(const unsigned char *bytes, size_t size, PureField *fields) {
+    if (!size || bytes[0] > PURE_MODULE_FIELDS) return -1;
+    unsigned count = bytes[0];
+    size_t at = 1;
+    for (unsigned f = 0; f < count; ++f) {
+        size_t length = size - at >= 2 ? bytes[at + 1] : 0;
+        if (size - at < 2 || !length || length >= PURE_MODULE_NAME || size - at < 2 + length || !bytes[at] ||
+            bytes[at] >= PURE_FIELD_ROWS)
+            return -1;
+        fields[f] = (PureField){.type = bytes[at]};
+        memcpy(fields[f].name, bytes + at + 2, length);
+        for (unsigned k = 0; k < f; ++k)
+            if (!strcmp(fields[k].name, fields[f].name)) return -1;
+        at += 2 + length;
+    }
+    if (at != size) return -1;
+    pure_fields_lay_out(fields, count);
+    return (int)count;
+}
+
+bool pure_module_takes(const PureModule *m, uint32_t input, const unsigned char *fields, size_t size) {
+    PureField sent[PURE_MODULE_FIELDS];
+    int count = named_fields(fields, size, sent);
+    if (count < 0) return false;
+    const PureInputShape *s = &m->input[input];
+    for (int k = 0; k < count; ++k) {
+        uint32_t f = 0;
+        while (f < s->field_count && strcmp(m->field[s->field_start + f].name, sent[k].name)) ++f;
+        if (f == s->field_count) return false; /* its value would be dropped */
+        uint8_t type = m->field[s->field_start + f].type, from = sent[k].type;
+        if (type != from && !(type == PURE_FIELD_NUMBER && from == PURE_FIELD_U32) &&
+            !(type == PURE_FIELD_BYTES && from == PURE_FIELD_TEXT))
+            return false;
+    }
+    return true;
+}
+
+PureResult pure_module_recast(const PureModule *m, uint32_t input, const unsigned char *fields, size_t size,
+                              PureSpan payload, unsigned char **out, size_t *out_size) {
+    PureField sent[PURE_MODULE_FIELDS];
+    int count = named_fields(fields, size, sent);
+    *out = NULL;
+    *out_size = 0;
+    if (count < 0) return PURE_STORE_ERROR;
+    const PureInputShape *s = &m->input[input];
+    return pure_send_recast(sent, (uint32_t)count, &m->field[s->field_start], s->field_count, payload, out, out_size);
+}
+
 int pure_module_view(const PureModule *m, const char *name) {
     for (uint32_t i = 0; i < m->views; ++i)
         if (!m->view_table[i] && !strcmp(m->view_names[i], name)) return (int)i;
diff --git a/core/module.h b/core/module.h
index ad31821..2cac3b3 100644
--- a/core/module.h
+++ b/core/module.h
@@ -217,8 +217,24 @@ int pure_module_view(const PureModule *, const char *name);
 int pure_module_column(const PureModule *, uint32_t table, const char *name);
 /* What a message of an input looks like, as 32 bits: its fields' names,
  * types and places (FNV-1a). A sender's output and the input it sends to
- * across nodes must agree on it (link/link.h). */
+ * across nodes agree on it, or the input takes the output's fields by name
+ * (link/link.h, docs/design/014-fields-by-name.md). */
 uint32_t pure_module_shape(const PureModule *, uint32_t input);
+/* An input's (or output's) fields as a sender names them to a receiver: u8
+ * count, each field's {u8 type, u8 name length, name}. Its size (at most
+ * PURE_FIELDS_NAMED), written to out. */
+enum { PURE_FIELDS_NAMED = 1 + PURE_MODULE_FIELDS * (2 + PURE_MODULE_NAME) };
+size_t pure_module_fields(const PureModule *, uint32_t input, unsigned char out[PURE_FIELDS_NAMED]);
+/* Whether input takes messages laid out with the fields named (as
+ * pure_module_fields names them): it has every one by name, of the same type
+ * or one that holds all its values (a u32 in a number, text in bytes). A
+ * field it has that they lack is 0 or empty; none of theirs is dropped. */
+bool pure_module_takes(const PureModule *, uint32_t input, const unsigned char *fields, size_t size);
+/* A payload laid out with the fields named, laid out anew for input by
+ * name (pure_module_takes says it can be; out is malloc'd). STORE_ERROR
+ * when payload is not laid out with them. */
+PureResult pure_module_recast(const PureModule *, uint32_t input, const unsigned char *fields, size_t size,
+                              PureSpan payload, unsigned char **out, size_t *out_size);
 /* The index state a number view shows as its position (a view that is just
  * `at.position`), or -1: what a host moves with pure_app_point. */
 int pure_module_position(const PureModule *, uint32_t view);