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.

1 files changed, 36 insertions, 7 deletions

diff --git a/serve/serve.c b/serve/serve.c
index d06f430..319f165 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -843,8 +843,17 @@ static const char *convert(Connection *c) {
         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; !why && f < shape->field_count; ++f)
-            if (!b.seen[f] && !c->given[f]) why = "missing_field";
+        /* A delivery from another node may lack fields its sender's module has
+         * not (docs/design/014-fields-by-name.md): they are 0 or empty. A
+         * member naming no field is refused all the same: it would be dropped. */
+        for (uint32_t f = 0; !why && f < shape->field_count; ++f) {
+            if (b.seen[f] || c->given[f]) continue;
+            const PureField *field = &c->host->module->field[shape->field_start + f];
+            if (!delivered || field->type == PURE_FIELD_ROWS) why = "missing_field";
+            else if (field->type == PURE_FIELD_U32 || field->type == PURE_FIELD_NUMBER)
+                memset(b.payload + field->offset, 0, field->type == PURE_FIELD_NUMBER ? 8 : 4);
+            else put_u32(b.table + 8 * f, 0), put_u32(b.table + 8 * f + 4, 0);
+        }
         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";
@@ -1374,6 +1383,9 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
         uint64_t first = get_u64(delivery), heard = pure_app_heard(app, source);
         unsigned count = get_u32(delivery + 8);
         const unsigned char *at = effect->value.data + 4 + DELIVERY;
+        /* Records laid out with other fields (named after them) are laid out anew for the input. */
+        size_t records = header & (STORED_RECORDS - 1), named = effect->value.size - 4 - DELIVERY - records;
+        const unsigned char *fields = named ? at + records : NULL;
         out->gap = false;
         out->fault = PURE_FAULT_NONE;
         for (unsigned i = 0; i < count; ++i) {
@@ -1385,7 +1397,14 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
                 out->gap = true;
                 break;
             }
-            PureResult status = pure_app_deliver(app, route->target, source, first + i, record, NULL);
+            unsigned char *recast = NULL;
+            size_t recast_size = 0;
+            if (fields && pure_module_recast(app->module, route->target, fields, named, record, &recast, &recast_size) !=
+                              PURE_OK)
+                break; /* not laid out as its fields were named: not applied, sent again */
+            PureResult status = pure_app_deliver(app, route->target, source, first + i,
+                                                 recast ? (PureSpan){recast, recast_size} : record, NULL);
+            free(recast);
             if (status != PURE_OK && !app_refused(app)) { /* not applied: sent again */
                 out->fault = app->fault;
                 break;
@@ -2375,12 +2394,15 @@ static void stop_shard(Host *host) {
 }
 
 /* The input a link's stream goes to: one taking messages (not keyed) of the
- * sender's shape. */
-static int link_input(void *context, const char *name, uint32_t shape) {
+ * sender's shape, or, named its fields, one that takes them by name
+ * (pure_module_takes: each record is laid out anew for it). */
+static int link_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
     const PureModule *m = ((Owner *)context)->executor.app.module;
     int index = pure_module_input(m, name);
     if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE || m->input[index].keyed) return -PURE_LINK_NO_INPUT;
-    return pure_module_shape(m, (uint32_t)index) == shape ? index : -PURE_LINK_OTHER_SHAPE;
+    if (pure_module_shape(m, (uint32_t)index) == shape) return index;
+    if (!fields) return -PURE_LINK_OTHER_SHAPE;
+    return pure_module_takes(m, (uint32_t)index, fields, size) ? index : -PURE_LINK_FIELDS_LACKED;
 }
 
 /* Records a link delivered, queued at the owner as a request is: applied
@@ -2388,7 +2410,7 @@ static int link_input(void *context, const char *name, uint32_t shape) {
 static PureLinkListener *link_listener;
 static bool link_deliver(void *context, const PureLinkDelivery *d) {
     Owner *owner = context;
-    size_t size = 4 + DELIVERY + d->size, length = strlen(d->source);
+    size_t size = 4 + DELIVERY + d->size + d->fields_size, length = strlen(d->source);
     unsigned char *ingress = malloc(size);
     LinkJob *link = malloc(sizeof(*link));
     if (!ingress || !link || d->size > (STORED_RECORDS - 1) || length > PURE_SOURCE) {
@@ -2403,6 +2425,7 @@ static bool link_deliver(void *context, const PureLinkDelivery *d) {
     ingress[16] = (unsigned char)length;
     memcpy(ingress + 17, d->source, length);
     memcpy(ingress + 4 + DELIVERY, d->records, d->size);
+    if (d->fields_size) memcpy(ingress + 4 + DELIVERY + d->size, d->fields, d->fields_size); /* laid out with those */
     *link = (LinkJob){link_listener, d->token, ingress};
     Job job = {.route = {.kind = ROUTE_INPUT, .target = d->input}, .link = link,
                .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}};
@@ -2451,6 +2474,12 @@ static bool open_wires(Executor *ex, const PureModule *m, const char *const *sen
                                  ? pure_link_open(equals + 1, source, pure_module_shape(m, (uint32_t)index), seal_key,
                                                   own_public, error, sizeof(error))
                                  : NULL;
+            unsigned char named[PURE_FIELDS_NAMED];
+            if (link && !pure_link_fields(link, named, pure_module_fields(m, (uint32_t)index, named))) {
+                pure_link_close(link);
+                link = NULL;
+                snprintf(error, sizeof(error), "no memory");
+            }
             if (!link) {
                 fprintf(stderr, "--send %s: %s\n", sends[i], written > PURE_SOURCE ? "the name and output are longer than 64 bytes" : error);
                 return false;