Module executor and pure-serve: 027's 4a, a message's rows holding lists: rows whose records hold lists are read, as the message arrives, by stage 3's decoder into the event's lists (value_take into the rows' list), the event beginning with them; the JSON edge reads an object's array as its record's objects, 64 lists deep at most (items_of: an item's lists encoded first and the item moved down over them); the load no longer refuses them; a hole closed: the decoder read past the end where fewer than 4 bytes were left for a length (size - at - 4 wrapped around), unseen while it read only what was stored; unit test nested_rows (three events, four rows not whole changing nothing); check server_trees (300 random forests over HTTP against a Python model walking them in the same order, a require refusing some, a list field not an array, a field the record lacks and 65 lists deep refused, a restart); counts: b30 against b29 within 0.001% but BFS with batches +0.12%, the bank's queries -0.00% to +0.23%; 027's stage 4 planned in six steps (4a to 4f, the counts it starts from, 4b in two commits) and 4a as made; 008 on nested rows

4 files changed, 98 insertions, 21 deletions

diff --git a/core/module-run.h b/core/module-run.h
index 4ae94a4..2ad87c6 100644
--- a/core/module-run.h
+++ b/core/module-run.h
@@ -102,6 +102,7 @@ typedef struct {
     PureValue *locals; /* the event's locals (a handler's, and what runs in it), or NULL */
     bool arguments; /* FIELD reads the arguments given for a view, not the message */
     bool modified, exists;
+    bool nested; /* its rows made some of the event's lists in records (027's 4a): the event begins with them */
     uint32_t (*spans)[2]; /* a message's variable fields: start and size in the payload */
 } Event;
 
diff --git a/core/module.c b/core/module.c
index 6804a4c..03cc434 100644
--- a/core/module.c
+++ b/core/module.c
@@ -1822,16 +1822,14 @@ PURE_COLD PureResult pure_module_load(const void *bytes, size_t size, PureModule
                out->record_field_type[out->record[out->state_ref[i]].first + out->state_key[i] - 1u] == PURE_RECORD_BOOL))) ||
             (out->state_type[i] == PURE_STATE_INDEX && !state_of(out, out->state_ref[i], PURE_STATE_LIST)))
             REFUSE("state reference", i);
-    /* A list in a record is an event's alone (027's stage 2b): no stored, transient or message's list holds one, and no
-     * list is keyed by one; a register is no message's rows. */
+    /* No list is keyed by a list in its records; a register is no message's rows. */
     for (uint32_t i = 0; i < out->states; ++i) {
         if (out->state_type[i] != PURE_STATE_LIST) continue;
         if (out->state_key[i] && out->record_field_type[out->record[out->state_ref[i]].first + out->state_key[i] - 1u] == PURE_RECORD_LIST)
             REFUSE("list in a record as a key", i);
     }
     for (uint32_t f = 0; f < out->fields; ++f)
-        if (out->field[f].type == PURE_FIELD_ROWS &&
-            (out->record_nested[out->state_ref[out->field[f].rows]] || out->state_flags[out->field[f].rows] & PURE_LIST_BOUND >> 8))
+        if (out->field[f].type == PURE_FIELD_ROWS && out->state_flags[out->field[f].rows] & PURE_LIST_BOUND >> 8)
             REFUSE("rows field", f);
     /* A stored position into what is not stored would come back meaning nothing. */
     for (uint32_t i = 0; i < out->states; ++i)
@@ -2518,28 +2516,31 @@ static unsigned char *value_put(const PureApp *app, uint32_t record, uint64_t li
     return out;
 }
 
-/* A list value's stored form (bytes) read into a new one of the event's lists of record's items, its lists in fields
- * too: its number (0: none), or 0 with *made false where the bytes are not that form or memory ran out (*bad: they are
- * not). */
-static uint64_t value_take(PureApp *app, uint32_t record, const unsigned char *bytes, size_t size, bool *made, bool *bad) {
+/* A list value's stored form (bytes) read into a new one of the event's lists of record's items (or appended to into, a
+ * message's rows), its lists in fields too: its number (0: none, or into), or 0 with *made false where the bytes are not
+ * that form or memory ran out (*bad: they are not). */
+static uint64_t value_take(PureApp *app, uint32_t record, const unsigned char *bytes, size_t size, PureList *into, bool *made,
+                           bool *bad) {
     const PureModule *m = app->module;
     const PureRecord *r = &m->record[record];
     *made = true;
     if (!size) return 0;
-    uint64_t list = nest(app, record, NULL);
-    if (!list) return *made = false, 0;
+    uint64_t list = into ? 0 : nest(app, record, NULL);
+    if (!into && !list) return *made = false, 0;
     for (size_t at = 0; at < size;) {
         PureCell cells[PURE_RECORD_FIELDS];
         for (unsigned f = 0; f < r->count; ++f) {
             uint32_t field = r->first + f;
             uint8_t type = m->record_field_type[field];
             if (type == PURE_RECORD_TEXT || type == PURE_RECORD_LIST) {
-                uint32_t length = size - at >= 4 ? read32(bytes + at) : UINT32_MAX;
+                if (size - at < 4) return *bad = true, *made = false, 0;
+                uint32_t length = read32(bytes + at);
                 if (length > size - at - 4 || (type == PURE_RECORD_TEXT && !pure_utf8_valid((PureSpan){bytes + at + 4, length})))
                     return *bad = true, *made = false, 0;
                 cells[f] = (PureCell){.text = (unsigned char *)bytes + at + 4, .size = length};
                 if (type == PURE_RECORD_LIST) {
-                    cells[f] = (PureCell){.i64 = (int64_t)value_take(app, nested_record(m, field), bytes + at + 4, length, made, bad)};
+                    cells[f] = (PureCell){.i64 = (int64_t)value_take(app, nested_record(m, field), bytes + at + 4, length, NULL, made,
+                                                                     bad)};
                     if (!*made) return 0;
                 }
                 at += 4 + length;
@@ -2550,7 +2551,7 @@ static uint64_t value_take(PureApp *app, uint32_t record, const unsigned char *b
                 if (!pure_cell_fits(pure_record_kind(m, field), cells[f])) return *bad = true, *made = false, 0;
             }
         }
-        PureList *items = app->nested.list[list - 1]; /* (the lists in its fields may have moved the table) */
+        PureList *items = into ? into : app->nested.list[list - 1]; /* (the lists in its fields may have moved the table) */
         if (pure_list_insert(items, pure_list_count(items), cells, NULL) != PURE_OK) return *made = false, 0;
         if (m->record_nested[record]) share_cells(app, m, record, cells);
     }
@@ -2604,7 +2605,8 @@ static __attribute__((cold, noinline)) PureResult kept_value(PureApp *app, uint3
                                                              int64_t *out) {
     const PureModule *m = app->module;
     bool made, bad = false;
-    uint64_t value = value_take(app, nested_record(m, m->record[m->state_ref[list]].first + field), cell.text, cell.size, &made, &bad);
+    uint32_t record = nested_record(m, m->record[m->state_ref[list]].first + field);
+    uint64_t value = value_take(app, record, cell.text, cell.size, NULL, &made, &bad);
     if (!made) return bad ? PURE_STORE_ERROR : PURE_ALLOCATION;
     if (!grows(app, (Held){0, 0}, (Held){0, cell.size})) return PURE_BUDGET;
     *out = (int64_t)value;
@@ -5118,7 +5120,7 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
     app->grown = app->spent = 0;
     app->room = app->memory > PURE_GROWTH_DEFAULT ? app->memory : PURE_GROWTH_DEFAULT;
     for (unsigned i = 0; m->held && i < PURE_MODULE_HELD; ++i) app->held[i].size = 0, app->held[i].by_ref = false;
-    if (app->nested.count) unbind(app), drop_nested(app); /* a view's body bound some on the app itself */
+    if (app->nested.count && !ev->nested) unbind(app), drop_nested(app); /* a view's body bound some on the app itself */
     Mark mark;
     mark_event(app, &mark);
     memcpy(pending, app->values, sizeof(app->values));
@@ -5310,19 +5312,32 @@ static PureResult message(PureApp *app, uint32_t input, const char *key, int64_t
         if (opened != PURE_OK) { app->document = NULL; return opened; }
         if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; }
     }
-    /* Its rows fill their lists (emptied when the event ends); rows that are not whole fail it. */
+    /* Its rows fill their lists (emptied when the event ends); rows that are not whole fail it. Rows whose records hold
+     * lists (027's 4a) read those into the event's lists as a kept list's cell is read, in the event's lists begun anew. */
+    bool nested = false;
     for (uint32_t f = 0; f < shape->field_count; ++f) {
         const PureField *field = &m->field[shape->field_start + f];
         if (field->type != PURE_FIELD_ROWS) continue;
-        PureResult filled = pure_list_read(app->lists[field->rows], payload.data + spans[f][0], spans[f][1]);
+        uint32_t record = m->state_ref[field->rows];
+        const unsigned char *rows = payload.data + spans[f][0];
+        PureResult filled;
+        if (m->record_nested[record]) {
+            if (!nested && app->nested.count) unbind(app), drop_nested(app);
+            nested = true;
+            bool made, bad = false;
+            value_take(app, record, rows, spans[f][1], app->lists[field->rows], &made, &bad);
+            filled = made ? PURE_OK : bad ? PURE_STORE_ERROR : PURE_ALLOCATION;
+        } else
+            filled = pure_list_read(app->lists[field->rows], rows, spans[f][1]);
         if (filled == PURE_OK) continue;
         clear_rows(app);
+        if (nested) drop_nested(app);
         pure_app_finish(app);
         return filled == PURE_ALLOCATION ? filled : fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
     }
     app->payload = payload;
     Event ev = {.app = app, .shape = shape, .key = key, .base = base, .arena = &app->arena, .exists = app->revision > 0,
-                .spans = spans};
+                .nested = nested, .spans = spans};
     PureResult result = dispatch(app, 1 + 2 * input, &ev);
     if (result != PURE_OK) pure_app_finish(app);
     return result;
diff --git a/core/module.h b/core/module.h
index 15109b6..349c4d2 100644
--- a/core/module.h
+++ b/core/module.h
@@ -98,9 +98,11 @@ enum { PURE_MODULE_HELD = 32, PURE_HELD_TEXT = 16, PURE_HELD_APPEND = 1 << 8 };
  * an empty one), an i64 in its cell, so moves and reads (ITEM_GET) are an
  * integer's; such lists go as the event or a view's body ends. A list is a
  * value: storing one in a second cell shares it, and a list shared or bound
- * is copied before it changes (so no other value sees the change). Since
- * stage 3 a stored or a transient list's records may hold lists too (a
- * message's rows not, nor as a key): there the cell holds the value in its
+ * is copied before it changes (so no other value sees the change). A
+ * message's rows may hold lists (027's 4a): each comes in as one of the
+ * event's lists, read from its stored form (below) before the handler runs.
+ * Since stage 3 a stored or a transient list's records may hold lists too
+ * (never as a key): there the cell holds the value in its
  * items' stored form (PURE_CELL_LIST), read into one of the event's lists
  * as ITEM_GET reads it and written from one as it is set (LIST_APPEND,
  * LIST_INSERT, ITEM_SET; APPEND_IN reads, appends and writes back), so
diff --git a/core/test-module.c b/core/test-module.c
index eb6752d..9d925dd 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -6628,6 +6628,64 @@ static void nested_replace(void) {
     printf("{\"nested_replace\":{\"lists\":2,\"items\":3}}\n");
 }
 
+/* 027's 4a: a message's rows whose records hold lists. Each node's kids come in as one of the event's lists, read
+ * through a register; the rows and their lists go as the event ends, and rows that are not whole fail it with nothing
+ * changed. */
+static size_t put_node(unsigned char *out, int64_t name, const int64_t *kids, uint32_t count) {
+    memcpy(out, &name, 8);
+    put32(out + 8, 8 * count);
+    for (uint32_t k = 0; k < count; ++k) memcpy(out + 12 + 8 * k, &kids[k], 8);
+    return 12 + 8 * (size_t)count;
+}
+
+static void nested_rows(void) {
+    enum { LC_ = PURE_OP_LIST_COUNT, BIND_ = PURE_OP_LIST_BIND, NODES = 1, KIDS = 2, TOTAL = 3 };
+    static const S nums[] = {{"n", 0}};
+    static const RF leaf_fields[] = {{"v", 3, 0}};
+    static const RF node_fields[] = {{"name", 3, 0}, {"kids", 4, 0}};
+    static const R records[] = {{"Leaf", leaf_fields, 1}, {"Node", node_fields, 2}};
+    static const T others[] = {{"order.nodes", 3 | (uint32_t)PURE_LIST_ROWS << 8, 1},
+                               {"let.kids", 3 | (uint32_t)(PURE_LIST_ROWS | PURE_LIST_BOUND) << 8, 0}, {"total", PURE_STATE_I64, 0}};
+    static const F fields[] = {{"nodes", PURE_FIELD_ROWS | NODES << 8}};
+    static const I inputs[] = {{"order", NULL, PURE_KIND_MESSAGE, false, fields, 1}};
+    static const O code[] = {
+        /* 0: for x in order.nodes { total += sum(k.v for k in x.kids) } */
+        {EACH, WALKING(NODES, 10), 0}, {ROW, NODES, 0}, {IG, ITEM(NODES, 1), 0}, {BIND_, KIDS, 0},
+        {L, TOTAL, 0}, {FOLD, FOLDING(PURE_FOLD_SUM | PURE_FOLD_INTEGER, KIDS, 3), 0}, {ROW, KIDS, 0}, {IG, ITEM(KIDS, 0), 0},
+        {C, 1, 1}, {IADD, 0, 0}, {ST, TOTAL, 0},
+        /* 11: n = n * 10 + order.nodes.count */
+        {L, 0, 0}, {C, 0, 10}, {MUL, 0, 0}, {LC_, NODES, 0}, {ADD, 0, 0}, {ST, 0, 0},
+    };
+    H handler = {1, 0, LEN(code)};
+    Spec spec = {nums, 1, inputs, 1, NULL, 0, &handler, 1, code, LEN(code), NULL, 0, others, LEN(others), NULL, 0, NULL, 0,
+                 records, 2, NULL, 0};
+    if (load(spec, &module_a) != PURE_OK) { fprintf(stderr, "nested rows: %s\n", error); abort(); }
+    PureApp app;
+    CHECK(pure_app_start(&app, &module_a, (PureData){0}) == PURE_OK);
+    static const int64_t first[] = {7, 9}, second[] = {5};
+    unsigned char payload[64];
+    size_t size = put_node(payload, 1, first, 2);
+    size += put_node(payload + size, 2, second, 1);
+    size += put_node(payload + size, 3, NULL, 0);
+    for (int round = 1; round <= 3; ++round) { /* each event begins with its lists and theirs empty */
+        CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, size}) == PURE_OK);
+        CHECK(app.values[TOTAL].i64 == 21 * round && app.values[0].num == (round == 1 ? 3 : round == 2 ? 33 : 333));
+        CHECK(pure_list_count(app.lists[NODES]) == 0 && app.nested.count == 2); /* the third node's kids are none */
+    }
+    /* Not whole: a node's kids past the end, a kid cut short, a node cut short. */
+    const size_t cuts[] = {size - 1, size - 12 - 4, 4};
+    unsigned char broken[64];
+    memcpy(broken, payload, size);
+    put32(broken + 8, 1000);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){broken, size}) == PURE_LIMIT && app.fault == PURE_FAULT_LIMIT);
+    for (size_t k = 0; k < LEN(cuts); ++k)
+        CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, cuts[k]}) == PURE_LIMIT && app.fault == PURE_FAULT_LIMIT);
+    CHECK(app.values[TOTAL].i64 == 63 && app.values[0].num == 333 && pure_list_count(app.lists[NODES]) == 0 && !app.nested.count);
+    CHECK(pure_app_message(&app, 0, NULL, -1, (PureSpan){payload, size}) == PURE_OK && app.values[TOTAL].i64 == 84);
+    pure_app_stop(&app);
+    printf("{\"nested_rows\":{\"rounds\":3,\"refused\":%zu}}\n", LEN(cuts) + 1);
+}
+
 /* A timer's tick sees the time the host set; floor and a floored modulo
  * take it apart. now is for handlers only. */
 enum { NOW = PURE_OP_NOW, FLOOR = PURE_OP_FLOOR, MOD = PURE_OP_MOD };
@@ -7007,6 +7065,7 @@ int main(void) {
     functions_and_characters();
     nested_crashes();
     nested_replace();
+    nested_rows();
     memory_budget();
     work_budget();
     fold_mutations();