Bound what an app holds (--memory): refuse what would grow past it

pure-serve --memory BYTES gives the app a budget of state bytes: its
lists' items as they are stored, its texts, the outbox and the numbers
heard (docs/design/013-memory-budget.md). The count is deterministic,
the same on any device. An event after which the states hold more than
the budget, and more than before it, fails with the fault memory and
changes nothing; HTTP answers 503 memory. One that frees or keeps them
goes on, so an app over its budget can always shrink.

Sends waiting for their receivers are state: a full outbox refuses the
next send until some are stored, which is the backpressure a slow
receiver puts on clients. A delivery refused for memory is not heard:
the link keeps it (c9c0edf) and HTTP senders try again after 503.

An answer (--reply) now takes its send out of the outbox in its own
event (pure_app_answer), so it has the room its send held. Apart, an app
whose outbox filled the budget could never apply an answer that keeps
anything, and the send leaves only after its answer: both would stay for
good. An answer that still does not fit waits, with the wire's after it,
and is tried again every 100 ms: pure_wire_answer leaves it in place
until pure_wire_answered.

Checks: test-module memory_budget (refused and undone, freeing over the
budget, a full outbox, a delivery not heard, answers that fit or wait,
across a restart); check.py server_memory (a sender refused 503 memory
with its receiver down and taking buys once it is up; a receiver
restarted below what it keeps, refusing 7 times in 2 s, then taking all)
and server_answers_memory (answers wait while receipts fill the budget,
then apply once each; the API called once per order). The stats report
state_bytes and memory_refused.

3 files changed, 190 insertions, 24 deletions

diff --git a/core/module.c b/core/module.c
index 79d3c5b..cbf0efe 100644
--- a/core/module.c
+++ b/core/module.c
@@ -1179,7 +1179,7 @@ static PureResult fault(PureApp *app, PureFault kind, uint32_t pc) {
     app->fault = kind;
     app->fault_pc = pc;
     return kind == PURE_FAULT_LIMIT ? PURE_LIMIT : kind == PURE_FAULT_NOT_FOUND ? PURE_NOT_FOUND :
-           kind == PURE_FAULT_EXISTS ? PURE_CONFLICT : PURE_INVALID;
+           kind == PURE_FAULT_EXISTS ? PURE_CONFLICT : kind == PURE_FAULT_MEMORY ? PURE_BUDGET : PURE_INVALID;
 }
 
 static bool reserve(PureArena *arena, size_t more) {
@@ -2645,7 +2645,7 @@ void pure_app_stop(PureApp *app) {
 typedef struct {
     PureVersion texts[PURE_MODULE_STATES];
     size_t lists[PURE_MODULE_STATES], heard, outbox; /* stored lists' journal lengths */
-    size_t undo;
+    size_t undo, sent_saved;
     uint64_t last_sent[PURE_MODULE_INPUTS];
     bool opened;
 } Mark;
@@ -2660,6 +2660,7 @@ static void mark_event(PureApp *app, Mark *mark) {
     mark->heard = app->heard ? pure_list_changes(app->heard).size : 0;
     mark->outbox = app->outbox ? pure_list_changes(app->outbox).size : 0;
     mark->undo = app->undo.count;
+    mark->sent_saved = app->sent_saved;
     memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent));
 }
 
@@ -2671,6 +2672,7 @@ static void undo_event(PureApp *app, const Mark *mark) {
     if (app->heard) pure_list_forget(app->heard, mark->heard);
     if (app->outbox) pure_list_forget(app->outbox, mark->outbox);
     pure_undo_rollback(&app->undo, mark->undo);
+    app->sent_saved = mark->sent_saved;
     memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent));
     if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
 }
@@ -2696,6 +2698,24 @@ static PureResult hear(PureApp *app, const char *source, uint64_t seq) {
     return pure_list_insert(app->heard, at, item, &app->undo);
 }
 
+/* The output's saved sends through seq leave the outbox: their receiver
+ * stored them, or answered them (in the answer's event). */
+static PureResult let_sends_go(PureApp *app, uint32_t output, uint64_t through) {
+    if (!app->outbox) return PURE_OK;
+    const char *name = app->module->input_names[output];
+    size_t length = strlen(name);
+    for (size_t i = 0; i < app->sent_saved;) {
+        PureCell of = pure_list_get(app->outbox, i, 0);
+        if (of.size != length || memcmp(of.text, name, length)) { ++i; continue; }
+        if ((uint64_t)pure_list_get(app->outbox, i, 1).num > through) break;
+        PureResult removed = pure_list_remove(app->outbox, i, &app->undo);
+        if (removed != PURE_OK) return removed;
+        --app->sent_saved;
+        app->links_changed = true;
+    }
+    return PURE_OK;
+}
+
 static PureResult keep_reply(PureApp *app);
 
 /* Every list of a message's rows emptied: an event's rows are its own. */
@@ -2705,14 +2725,29 @@ static void clear_rows(PureApp *app) {
         if (m->state_type[i] == PURE_STATE_LIST && m->state_flags[i] & PURE_LIST_ROWS >> 8) pure_list_clear(app->lists[i]);
 }
 
+void pure_app_memory(PureApp *app, size_t bytes) { app->memory = bytes; }
+
+size_t pure_app_state_bytes(const PureApp *app) {
+    const PureModule *m = app->module;
+    size_t bytes = (app->outbox ? pure_list_size(app->outbox) : 0) + (app->heard ? pure_list_size(app->heard) : 0);
+    for (uint32_t i = 0; i < m->states; ++i) {
+        if (app->lists[i] && !(m->state_flags[i] & PURE_LIST_ROWS >> 8)) bytes += pure_list_size(app->lists[i]);
+        else if (app->texts[i]) bytes += pure_draft_size(app->texts[i]);
+    }
+    return bytes; /* a message's rows are its own, gone when it ends */
+}
+
 /* One event: every handler for it runs in module order against pending
  * states, all or nothing. A key, text or message event saves before it
  * returns unless saving is deferred; a frame leaves that to the next event
- * (frames come 60 times a second). A message's document joins the burst. */
+ * (frames come 60 times a second). A message's document joins the burst.
+ * With a memory budget, an event that leaves the states over it, and
+ * holding more than before, fails. */
 static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
     const PureModule *m = app->module;
     double pending[PURE_MODULE_STATES], locals[PURE_MODULE_LOCALS] = {0};
     ev->locals = locals;
+    size_t held = app->memory ? pure_app_state_bytes(app) : 0;
     Mark mark;
     mark_event(app, &mark);
     memcpy(pending, app->values, sizeof(app->values));
@@ -2728,8 +2763,14 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
     if (result == PURE_OK && app->has_reply &&
         ((app->reply.source == FROM_DOCUMENT && ev->modified) || app->reply.source == FROM_TEXT))
         result = keep_reply(app);
-    /* A delivery's number is heard with its changes. */
+    /* A delivery's number is heard with its changes, and an answer's sends go with it. */
     if (result == PURE_OK && app->delivering.pending) result = hear(app, app->delivering.source, app->delivering.seq);
+    if (result == PURE_OK && app->delivering.pending && app->delivering.release)
+        result = let_sends_go(app, app->delivering.release - 1, app->delivering.seq);
+    if (result == PURE_OK && app->memory) {
+        size_t holds = pure_app_state_bytes(app);
+        if (holds > app->memory && holds > held) result = fault(app, PURE_FAULT_MEMORY, UINT32_MAX);
+    }
     if (result != PURE_OK) {
         undo_event(app, &mark);
         clear_rows(app);
@@ -2950,23 +2991,14 @@ void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *conte
  * store with the next save; a failed save brings them back. Its sends are
  * in order among the others: the look stops at its first one kept. */
 void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) {
-    const PureModule *m = app->module;
-    if (!app->outbox || output >= m->inputs) return;
-    const char *name = m->input_names[output];
-    size_t length = strlen(name);
+    if (!app->outbox || output >= app->module->inputs) return;
     open_burst(app);
-    for (size_t i = 0; i < app->sent_saved;) {
-        PureCell of = pure_list_get(app->outbox, i, 0);
-        if (of.size != length || memcmp(of.text, name, length)) { ++i; continue; }
-        if ((uint64_t)pure_list_get(app->outbox, i, 1).num > through) break;
-        if (pure_list_remove(app->outbox, i, &app->undo) != PURE_OK) break;
-        --app->sent_saved;
-        app->links_changed = true;
-    }
+    let_sends_go(app, output, through);
 }
 
-PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
-                            const uint32_t *placed) {
+/* A delivery, or an answer to the sends of output (release: its number + 1). */
+static PureResult deliver(PureApp *app, uint32_t input, uint32_t release, const char *source, uint64_t seq,
+                          PureSpan payload, const uint32_t *placed) {
     const PureModule *m = app->module;
     size_t length = source ? strlen(source) : 0;
     app->duplicate = false;
@@ -2983,16 +3015,19 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     }
     app->delivering.source = source;
     app->delivering.seq = seq;
+    app->delivering.release = release;
     app->delivering.pending = true;
     PureResult result = message(app, input, NULL, -1, payload, placed);
     if (!app->delivering.pending) return result;
     /* Not applied. Refused by the app (a fault), the message was delivered
-     * all the same: its number is heard, saved like a key event's changes. */
+     * all the same: its number is heard, saved like a key event's changes.
+     * Refused for memory it was not: its sender sends it again. */
     app->delivering.pending = false;
-    if (app->fault == PURE_FAULT_NONE) return result;
+    if (app->fault == PURE_FAULT_NONE || app->fault == PURE_FAULT_MEMORY) return result;
     Mark mark;
     mark_event(app, &mark);
     PureResult heard = hear(app, source, seq);
+    if (heard == PURE_OK && release) heard = let_sends_go(app, release - 1, seq);
     if (heard != PURE_OK) {
         undo_event(app, &mark);
         return heard;
@@ -3001,6 +3036,17 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     return saved == PURE_OK ? result : saved;
 }
 
+PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
+                            const uint32_t *placed) {
+    return deliver(app, input, 0, source, seq, payload, placed);
+}
+
+PureResult pure_app_answer(PureApp *app, uint32_t input, uint32_t output, const char *source, uint64_t seq,
+                           PureSpan payload, const uint32_t *placed) {
+    if (output >= app->module->inputs || app->module->input[output].kind != PURE_KIND_OUTPUT) return PURE_INVALID;
+    return deliver(app, input, output + 1, source, seq, payload, placed);
+}
+
 PureResult pure_app_reply(const PureApp *app, bool (*visit)(void *, PureSpan), void *context) {
     if (!app->has_reply) return PURE_OK;
     return visit_value(app, &app->arena, app->reply, visit, context);
diff --git a/core/module.h b/core/module.h
index 0b45ba2..c11c682 100644
--- a/core/module.h
+++ b/core/module.h
@@ -109,10 +109,12 @@ typedef enum { PURE_VIEW_NUM = 0, PURE_VIEW_TEXT = 1, PURE_VIEW_TABLE = 2 } Pure
 typedef enum { PURE_UNIT_CHAR = 0, PURE_UNIT_LINE = 1 } PureUnit;
 typedef enum { PURE_EDGE_DOWN, PURE_EDGE_UP } PureEdge;
 /* EXISTS: an item with that key is already in the keyed list (the event
- * reports PURE_CONFLICT). */
+ * reports PURE_CONFLICT). MEMORY: the event would leave the states holding
+ * more than the app's budget (pure_app_memory; it reports PURE_BUDGET). */
 typedef enum {
     PURE_FAULT_NONE, PURE_FAULT_DIVIDE, PURE_FAULT_NON_FINITE, PURE_FAULT_CLAMP,
-    PURE_FAULT_LIMIT, PURE_FAULT_NOT_FOUND, PURE_FAULT_ENCODING, PURE_FAULT_EXISTS
+    PURE_FAULT_LIMIT, PURE_FAULT_NOT_FOUND, PURE_FAULT_ENCODING, PURE_FAULT_EXISTS,
+    PURE_FAULT_MEMORY
 } PureFault;
 
 /* plan, shape and range are the runtime's, set at load: a fold whose block
@@ -310,6 +312,7 @@ typedef struct {
     bool document_pending; /* the current event's document is among them */
     PureFault fault;
     uint32_t fault_pc;
+    size_t memory; /* the budget of what the states hold (0: none): pure_app_memory */
     uint64_t ticks, events;
     /* The time events see as now, in milliseconds since 1970 (UTC). The host
      * sets it, so a script can replay any time; views never see it. */
@@ -346,7 +349,7 @@ typedef struct {
     PureDraft *stored_links;
     bool links_changed; /* since the last save */
     bool duplicate;     /* the last delivery was heard before and not applied */
-    struct { const char *source; uint64_t seq; bool pending; } delivering;
+    struct { const char *source; uint64_t seq; uint32_t release; bool pending; } delivering;
     /* The arguments given for a view (pure_app_arguments): its input + 1 (0:
      * none), where their fields are, and a count of the times they were given. */
     unsigned argued;
@@ -443,13 +446,31 @@ void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
  * before is not applied again: it succeeds with app->duplicate set. The
  * number counts as heard with the event's changes, and also when the app
  * refuses the message (a fault: it was delivered, and refused); not when a
- * save fails. placed is as for pure_app_message_fields, or NULL. */
+ * save fails, nor when the app has no memory for it (pure_app_memory: it
+ * comes again). placed is as for pure_app_message_fields, or NULL. */
 PureResult pure_app_deliver(PureApp *, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
                             const uint32_t *placed);
+/* An answer to the send numbered seq of output (serve/deliver.h's --reply),
+ * delivered as pure_app_deliver delivers from source; in the same event the
+ * output's saved sends through seq leave the outbox, so the answer has the
+ * room its send held. */
+PureResult pure_app_answer(PureApp *, uint32_t input, uint32_t output, const char *source, uint64_t seq,
+                           PureSpan payload, const uint32_t *placed);
 /* The last number heard from source (0 when none): a host that gets
  * numbered messages out of order hands them over from the one after it. */
 uint64_t pure_app_heard(const PureApp *, const char *source);
 
+/* An app's memory budget (docs/design/013-memory-budget.md): what its
+ * states hold, as they are stored (their lists' items, their texts, the
+ * outbox and the numbers heard), at most bytes (0: no budget, at first). An
+ * event after which they hold more than that, and more than before it,
+ * fails with the fault PURE_FAULT_MEMORY (PURE_BUDGET) and changes nothing;
+ * one that frees or keeps them goes on. A delivery refused so is not heard:
+ * its sender sends it again. */
+void pure_app_memory(PureApp *, size_t bytes);
+/* What the states hold now, as the budget counts it. */
+size_t pure_app_state_bytes(const PureApp *);
+
 /* Readers on other threads. Right after a save an app can make a snapshot of
  * its saved states, for evaluating views anywhere while it keeps applying
  * events: numbers are copied; text states share their stored versions and
diff --git a/core/test-module.c b/core/test-module.c
index 1446e73..70cc02d 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -4209,6 +4209,104 @@ static void links(void) {
     CHECK(load(bad, &module_b) == PURE_INVALID && strstr(error, "handler event"));
 }
 
+/* The app's memory budget (docs/design/013-memory-budget.md): an event that
+ * would leave the states over it and holding more is refused and changes
+ * nothing; one that frees them goes on; sends waiting for their receivers
+ * count, so a full outbox refuses more until some are trimmed; a delivery
+ * refused so is not heard, and comes again; an answer has its send's room. */
+static void memory_budget(void) {
+    enum { ADD, DROP, SEND_KEY, OUT, TAKE, ANSWER };
+    static const S nums[] = {{"n", 0}};
+    static const T states[] = {{"rows", 3, 0}};
+    static const RF fields[] = {{"value", 0, 0}};
+    static const R records[] = {{"row", fields, 1}};
+    static const F value[] = {{"value", PURE_FIELD_NUMBER}};
+    static const I inputs[] = {KEY("add", "KeyA"), KEY("drop", "KeyD"), KEY("send", "KeyS"),
+                               {"out", NULL, PURE_KIND_OUTPUT, false, value, 1},
+                               {"take", NULL, PURE_KIND_MESSAGE, false, value, 1},
+                               {"answer", NULL, PURE_KIND_MESSAGE, false, value, 1}};
+    static const O code[] = {
+        {L, 0, 0}, {LA, 1, 0},     /* add: rows.append(row { n }) */
+        {XF, 1, 0}, {LR, 1, 0},    /* drop: rows.remove(rows.first) */
+        {C, 0, 1}, {SEND, 3, 0},   /* send: send out { value: 1 } */
+        {FIELD, 0, 0}, {LA, 1, 0}, /* take: rows.append(row { take.value }) */
+        {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, /* answer: six rows */
+        {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0}, {FIELD, 0, 0}, {LA, 1, 0},
+    };
+    static const H handlers[] = {{1 + 2 * ADD, 0, 2}, {1 + 2 * DROP, 2, 2}, {1 + 2 * SEND_KEY, 4, 2}, {1 + 2 * TAKE, 6, 2},
+                                 {1 + 2 * ANSWER, 8, 12}};
+    Spec spec = {nums, 1, inputs, 6, NULL, 0, handlers, 5, code, LEN(code), NULL, 0, states, 1, NULL, 0, NULL, 0,
+                 records, 1, NULL, 0};
+    CHECK(load(spec, &module_a) == PURE_OK);
+    char *path = new_store();
+    PureApp app;
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK);
+    for (int k = 0; k < 3; ++k) CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_OK);
+    size_t three = pure_app_state_bytes(&app);
+    CHECK(three == 3 * 8); /* a row's number as it is stored */
+    /* Room for one row more: the next is refused, and nothing of it stays. */
+    pure_app_memory(&app, three + 8);
+    CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.lists[1]) == 4);
+    CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_BUDGET && app.fault == PURE_FAULT_MEMORY &&
+          pure_list_count(app.lists[1]) == 4 && pure_app_state_bytes(&app) == three + 8);
+    /* Over the budget (made smaller), what frees goes on and what grows does not. */
+    pure_app_memory(&app, 8);
+    CHECK(pure_app_input(&app, ADD, PURE_EDGE_DOWN) == PURE_BUDGET);
+    CHECK(pure_app_input(&app, DROP, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.lists[1]) == 3);
+    /* Sends count while they wait: a full outbox refuses the next until some are trimmed. */
+    size_t before = pure_app_state_bytes(&app);
+    pure_app_memory(&app, before);
+    CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_BUDGET && !pure_list_count(app.outbox));
+    pure_app_memory(&app, 0);
+    CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK && pure_list_count(app.outbox) == 1);
+    size_t one = pure_app_state_bytes(&app) - before;
+    pure_app_memory(&app, before + 2 * one);
+    CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK);
+    CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_BUDGET && pure_list_count(app.outbox) == 2);
+    pure_app_trim(&app, OUT, 1);
+    CHECK(pure_app_save(&app) == PURE_OK && pure_list_count(app.outbox) == 1);
+    CHECK(pure_app_input(&app, SEND_KEY, PURE_EDGE_DOWN) == PURE_OK && outbox_of(&app).items[1].seq == 3);
+    /* A delivery refused for memory is not heard: given again with room, it is applied once. */
+    double taken = 7;
+    pure_app_memory(&app, pure_app_state_bytes(&app));
+    CHECK(pure_app_deliver(&app, TAKE, "far.out", 1, (PureSpan){(const unsigned char *)&taken, 8}, NULL) == PURE_BUDGET &&
+          app.fault == PURE_FAULT_MEMORY && pure_app_heard(&app, "far.out") == 0 && pure_list_count(app.lists[1]) == 3);
+    pure_app_finish(&app);
+    pure_app_memory(&app, 0);
+    CHECK(pure_app_deliver(&app, TAKE, "far.out", 1, (PureSpan){(const unsigned char *)&taken, 8}, NULL) == PURE_OK &&
+          !app.duplicate && pure_app_heard(&app, "far.out") == 1 && pure_list_count(app.lists[1]) == 4);
+    pure_app_finish(&app);
+    CHECK(pure_list_get(app.lists[1], 3, 0).num == 7);
+    /* An answer takes its send out of the outbox in its own event: one that
+     * keeps more than the send held is refused whole at the budget (the send
+     * stays), and goes on with room; one that keeps less goes on at it. */
+    size_t full = pure_app_state_bytes(&app);
+    CHECK(one < 6 * 8 && one > 8);
+    double paid = 9;
+    PureSpan nine = {(const unsigned char *)&paid, 8};
+    pure_app_memory(&app, full);
+    CHECK(pure_app_answer(&app, ANSWER, OUT, "wire.out", 2, nine, NULL) == PURE_BUDGET &&
+          app.fault == PURE_FAULT_MEMORY && pure_list_count(app.outbox) == 2 && pure_list_count(app.lists[1]) == 4 &&
+          pure_app_heard(&app, "wire.out") == 0 && pure_app_state_bytes(&app) == full);
+    pure_app_finish(&app);
+    size_t heard = 4 + strlen("wire.out") + 8; /* its source among the numbers heard */
+    pure_app_memory(&app, full + 6 * 8 + heard - one);
+    CHECK(pure_app_answer(&app, ANSWER, OUT, "wire.out", 2, nine, NULL) == PURE_OK && pure_list_count(app.outbox) == 1 &&
+          outbox_of(&app).items[0].seq == 3 && pure_list_count(app.lists[1]) == 10 &&
+          pure_app_heard(&app, "wire.out") == 2 && pure_app_state_bytes(&app) == full + 6 * 8 + heard - one);
+    pure_app_finish(&app);
+    pure_app_memory(&app, pure_app_state_bytes(&app));
+    CHECK(pure_app_answer(&app, TAKE, OUT, "wire.out", 3, nine, NULL) == PURE_OK && !pure_list_count(app.outbox) &&
+          pure_list_count(app.lists[1]) == 11);
+    pure_app_finish(&app);
+    CHECK(pure_app_answer(&app, TAKE, TAKE, "wire.out", 4, nine, NULL) == PURE_INVALID); /* not an output */
+    pure_app_stop(&app);
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_list_count(app.lists[1]) == 11 &&
+          !pure_list_count(app.outbox) && pure_app_heard(&app, "far.out") == 1 && pure_app_heard(&app, "wire.out") == 3);
+    pure_app_stop(&app);
+    remove_store(path);
+}
+
 /* An output whose fields are named otherwise (here two numbers swapped) is
  * another output: its waiting sends' bytes would mean something else. A
  * module that has it so is refused while sends wait, replacing or starting
@@ -4360,6 +4458,7 @@ int main(void) {
     snapshots();
     links();
     output_meaning();
+    memory_budget();
     fold_mutations();
     pointing();
     list_mutations();