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.

7 files changed, 512 insertions, 59 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();
diff --git a/serve/deliver.c b/serve/deliver.c
index b4d9e57..68c5f14 100644
--- a/serve/deliver.c
+++ b/serve/deliver.c
@@ -47,7 +47,6 @@ struct PureWire {
     int answer_input; /* the input answers become events of, or -1 */
     Answer *answers;  /* made, for the owner to take in order */
     size_t answer_head, answer_count, answer_capacity;
-    Answer taken;     /* the one the owner took last, kept until it takes the next */
     uint32_t placed[2 * PURE_MODULE_FIELDS];
     /* Sealed sends: to peer's key, authenticated by the node's own key when
      * it has one (its public half is from), in one session made at the first
@@ -240,23 +239,26 @@ bool pure_wire_reply(PureWire *w, uint32_t input, char *error, size_t size) {
 }
 
 bool pure_wire_answer(PureWire *w, PureWireAnswer *out) {
-    free(w->taken.bytes);
-    w->taken = (Answer){0};
     pthread_mutex_lock(&w->lock);
     bool some = w->answer_count > 0;
-    if (some) {
-        w->taken = w->answers[w->answer_head++];
-        --w->answer_count;
-        w->delivered = w->taken.seq; /* trimmed after the owner applies it */
-    }
+    Answer next = some ? w->answers[w->answer_head] : (Answer){0}; /* its bytes stay put while the array moves */
     pthread_mutex_unlock(&w->lock);
     if (!some) return false;
     const PureInputShape *in = &w->module->input[w->answer_input];
-    for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(w->taken.bytes + 4 * i);
-    *out = (PureWireAnswer){(uint32_t)w->answer_input, w->taken.seq,
-                            {w->taken.bytes + w->taken.table, w->taken.size - w->taken.table}, w->placed, w->source};
+    for (size_t i = 0; i < 2 * (size_t)in->field_count; ++i) w->placed[i] = read_u32(next.bytes + 4 * i);
+    *out = (PureWireAnswer){(uint32_t)w->answer_input, next.seq, {next.bytes + next.table, next.size - next.table},
+                            w->placed, w->source};
     return true;
 }
+
+void pure_wire_answered(PureWire *w) {
+    pthread_mutex_lock(&w->lock);
+    Answer done = w->answers[w->answer_head++];
+    --w->answer_count;
+    w->delivered = done.seq; /* trimmed after the owner applies it */
+    pthread_mutex_unlock(&w->lock);
+    free(done.bytes);
+}
 uint64_t pure_wire_queued(const PureWire *w) { return w->queued; }
 
 uint64_t pure_wire_delivered(PureWire *w) {
@@ -771,7 +773,6 @@ void pure_wire_close(PureWire *w) {
     }
     for (size_t i = 0; i < w->answer_count; ++i) free(w->answers[w->answer_head + i].bytes);
     free(w->answers);
-    free(w->taken.bytes);
     free(w->queue);
     pthread_cond_destroy(&w->changed);
     pthread_mutex_destroy(&w->lock);
diff --git a/serve/deliver.h b/serve/deliver.h
index 71cc38e..a7ccac3 100644
--- a/serve/deliver.h
+++ b/serve/deliver.h
@@ -49,11 +49,10 @@ void pure_wire_close(PureWire *);
  * http_status is the answer's status. An answer that makes no message is
  * logged and the send tried again every 10 s. Answers go one per request. */
 bool pure_wire_reply(PureWire *, uint32_t input, char *error, size_t size);
-/* The owner's side: the next answer in order, or false. Taking it counts
- * its send delivered (the owner applies the answer, then trims); what out
- * points to is the wire's until the next call. The answer is applied with
- * pure_app_deliver from the wire's source with the send's number, so an
- * answer that comes again after a restart is applied once. */
+/* The owner's side: the next answer in order, or false; it stays the next,
+ * and what out points to the wire's, until pure_wire_answered. The answer
+ * is applied with pure_app_answer from the wire's source with the send's
+ * number, so an answer that comes again after a restart is applied once. */
 typedef struct {
     uint32_t input;
     uint64_t seq;
@@ -62,6 +61,9 @@ typedef struct {
     const char *source;
 } PureWireAnswer;
 bool pure_wire_answer(PureWire *, PureWireAnswer *out);
+/* The next answer was applied, or refused by the app: it goes, and its send
+ * counts delivered (the owner trims it). */
+void pure_wire_answered(PureWire *);
 /* Counts for a host's report. */
 void pure_wire_counts(PureWire *, uint64_t *delivered, uint64_t *attempts);
 #endif
diff --git a/serve/serve.c b/serve/serve.c
index 95ac37f..ab386ae 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -185,6 +185,8 @@ typedef struct {
     bool reads_first; /* the owner's: a round's reads before its writes */
     uint64_t next[PURE_MODULE_INPUTS]; /* when each timer ticks next (monotonic clock, ms; 0: not yet begun) */
     uint64_t ticks;
+    uint64_t memory_refused; /* events refused for the app's memory budget */
+    bool answers_wait; /* an answer the app had no room for: tried again after ANSWER_WAIT_MS */
     /* The owner's: the last save's snapshot (NULL: the owner shows views),
      * taken under the lock, and the app's changes it shows. */
     bool publishes;
@@ -517,11 +519,12 @@ static void answer(Connection *c, PureOutcome result) {
 }
 
 static const char *const fault_names[] = {"none", "divide", "non_finite", "clamp", "limit", "not_found", "encoding",
-                                          "exists"};
+                                          "exists", "memory"};
 
 /* An app route's answer: a view's value, or what an event replied. A failed
- * event names its fault: 404 not_found, 409 exists, 422 for the others;
- * a view without a value is 404 no_value. */
+ * event names its fault: 404 not_found, 409 exists, 503 memory (the app's
+ * budget: again later), 422 for the others; a view without a value is 404
+ * no_value. */
 static void answer_app(Connection *c, PureOutcome result, bool view, PureFault fault, const char *type) {
     const char *named = fault != PURE_FAULT_NONE ? fault_names[fault] : NULL;
     if (result.status == PURE_OK) {
@@ -532,6 +535,8 @@ static void answer_app(Connection *c, PureOutcome result, bool view, PureFault f
         respond(c, 404, "Not Found", view ? "no_value" : named ? named : "not_found", 0, (PureSpan){0});
     } else if (result.status == PURE_CONFLICT) {
         respond(c, 409, "Conflict", named ? named : "conflict", 0, (PureSpan){0});
+    } else if (fault == PURE_FAULT_MEMORY) {
+        respond(c, 503, "Service Unavailable", named, 0, (PureSpan){0});
     } else if (named && (result.status == PURE_LIMIT || result.status == PURE_INVALID)) {
         respond(c, 422, "Unprocessable Content", named, 0, (PureSpan){0});
     } else if (result.status == PURE_INVALID) {
@@ -1346,6 +1351,10 @@ static PureResult show_given(PureApp *app, const Route *route, size_t available,
     return PURE_OK;
 }
 
+/* The app refused a message it was given (a fault of its own): it was
+ * delivered, and is heard. Refused for memory it was not. */
+static bool app_refused(const PureApp *app) { return app->fault != PURE_FAULT_NONE && app->fault != PURE_FAULT_MEMORY; }
+
 /* An input's event: a text, a key pressed (down, then up), or a message
  * whose reply (a field of it) is borrowed from the request. */
 static PureResult apply_input(PureApp *app, const Route *route, const PureEffect *effect, Applied *out) {
@@ -1360,6 +1369,7 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
         unsigned count = get_u32(delivery + 8);
         const unsigned char *at = effect->value.data + 4 + DELIVERY;
         out->gap = false;
+        out->fault = PURE_FAULT_NONE;
         for (unsigned i = 0; i < count; ++i) {
             uint32_t size = get_u32(at);
             PureSpan record = {at + 4, size};
@@ -1370,11 +1380,13 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
                 break;
             }
             PureResult status = pure_app_deliver(app, route->target, source, first + i, record, NULL);
-            if (status != PURE_OK && app->fault == PURE_FAULT_NONE) break; /* not applied: sent again */
+            if (status != PURE_OK && !app_refused(app)) { /* not applied: sent again */
+                out->fault = app->fault;
+                break;
+            }
             heard = first + i;
         }
         out->heard = pure_app_heard(app, source);
-        out->fault = PURE_FAULT_NONE;
         return PURE_OK; /* the save says whether they are stored */
     }
     unsigned batch = header & DELIVERED ? get_u32(delivery + 8) : 1;
@@ -1396,14 +1408,15 @@ static PureResult apply_input(PureApp *app, const Route *route, const PureEffect
             source[delivery[12]] = 0;
             status = pure_app_deliver(app, route->target, source, get_u64(delivery), payload, table ? placed : NULL);
             /* The rest of a batch, in order: an app's refusal was heard and the
-             * batch goes on; another failure stops it, to be sent again. */
-            for (unsigned i = 1; i < batch && (status == PURE_OK || app->fault != PURE_FAULT_NONE); ++i) {
+             * batch goes on; another failure (the budget's too) stops it, to
+             * be sent again. */
+            for (unsigned i = 1; i < batch && (status == PURE_OK || app_refused(app)); ++i) {
                 const unsigned char *at = effect->value.data + lead + i * table;
                 for (size_t k = 0; k < 2 * (size_t)shape->field_count; ++k) placed[k] = get_u32(at + 4 * k);
                 PureSpan next = {payload.data + i * shape->prefix, payload.size - i * shape->prefix};
                 status = pure_app_deliver(app, route->target, source, get_u64(delivery) + i, next, placed);
             }
-            if (batch > 1 && app->fault != PURE_FAULT_NONE) status = PURE_OK;
+            if (batch > 1 && app_refused(app)) status = PURE_OK;
         } else
             status = table ? pure_app_message_fields(app, route->target, NULL, -1, payload, placed) :
                              pure_app_message(app, route->target, NULL, -1, payload);
@@ -1428,6 +1441,7 @@ static void apply_effect(Executor *ex, const Job *job, Applied *out) {
     if (job->route.kind == ROUTE_INPUT) {
         out->status = apply_input(app, &job->route, effect, out);
         if (out->status == PURE_OK) ++ex->burst_events;
+        ex->memory_refused += out->fault == PURE_FAULT_MEMORY;
         return;
     }
     bool read = effect->input == PURE_INPUT_READ;
@@ -1549,22 +1563,34 @@ static void after_save(Executor *ex) {
     free(q.fresh);
 }
 
-/* The answers the wires made become events, in order; then what the wires
+enum { ANSWER_WAIT_MS = 100 };
+
+/* The answers the wires made become events, in order, each taking its send
+ * out of the outbox; one the app has no room for waits, with the wire's
+ * after it, and is tried again after ANSWER_WAIT_MS. Then what the wires
  * delivered leaves the outbox (with the next save). True when an answer was
  * applied (the caller saves). */
 static bool take_deliveries(Executor *ex) {
     bool answered = false;
+    ex->answers_wait = false;
     for (unsigned i = 0; i < ex->wire_count; ++i) {
         PureWireAnswer answer;
         while (pure_wire_answer(ex->wires[i], &answer)) {
             ex->app.now = wall_ms();
-            PureResult result = pure_app_deliver(&ex->app, answer.input, answer.source, answer.seq, answer.payload,
-                                                 answer.placed);
+            PureResult result = pure_app_answer(&ex->app, answer.input, pure_wire_output(ex->wires[i]), answer.source,
+                                                answer.seq, answer.payload, answer.placed);
+            if (result != PURE_OK && ex->app.fault == PURE_FAULT_MEMORY) {
+                pure_app_finish(&ex->app);
+                ++ex->memory_refused;
+                ex->answers_wait = true;
+                break;
+            }
             if (result != PURE_OK && ex->app.fault == PURE_FAULT_NONE) {
                 ex->held[i] = true; /* not applied and not refused: sent again after a restart */
                 fprintf(stderr, "{\"wire\":\"%s\",\"seq\":%" PRIu64 ",\"error\":\"its answer was not applied\"}\n",
                         answer.source, answer.seq);
             }
+            pure_wire_answered(ex->wires[i]);
             pure_app_finish(&ex->app);
             answered = true;
             ++ex->answers;
@@ -1711,7 +1737,7 @@ static void hand_back(void *context, Applied *applied, size_t count) {
         LinkJob *link = applied[i].link;
         if (!link) continue; /* records a link delivered: stored now, or not */
         pure_link_done(link->listener, (PureLinkDone){link->token, applied[i].heard, applied[i].status == PURE_OK,
-                                                     applied[i].gap, false});
+                                                     applied[i].gap, applied[i].fault == PURE_FAULT_MEMORY});
         free(link->ingress);
         free(link);
     }
@@ -1948,7 +1974,8 @@ static void *run_owner(void *context) {
             owner->delivered = false;
             pthread_mutex_unlock(&owner->lock);
             long due = run_timers(&owner->executor);
-            bool answered = delivered && take_deliveries(&owner->executor);
+            bool answered = (delivered || owner->executor.answers_wait) && take_deliveries(&owner->executor);
+            if (owner->executor.answers_wait && (due < 0 || due > ANSWER_WAIT_MS)) due = ANSWER_WAIT_MS;
             pthread_mutex_lock(&owner->lock);
             /* Answers are saved with the next round, or now if none waits. */
             if (answered && !owner->count && !owner->stopping) {
@@ -2193,6 +2220,8 @@ static void accept_all(Host *host) {
     }
 }
 
+static size_t memory_budget; /* --memory: the app's budget of what its states hold (0: none) */
+
 static bool start_executor(Executor *ex, const PureModule *module, uint32_t show, uint32_t edit, const char *store) {
     ex->module = module;
     ex->show = show;
@@ -2203,6 +2232,7 @@ static bool start_executor(Executor *ex, const PureModule *module, uint32_t show
         return false;
     }
     pure_app_background_checkpoints(&ex->app, true);
+    pure_app_memory(&ex->app, memory_budget);
     return true;
 }
 
@@ -2600,6 +2630,7 @@ int main(int argc, char **argv) {
         else if (!strcmp(argv[i], "--senders")) senders_path = argv[i + 1];
         else if (!strcmp(argv[i], "--link-idle-ms") && number(argv[i + 1], 86400000, &value) && value) link_idle_ms = value;
         else if (!strcmp(argv[i], "--budget") && number(argv[i + 1], SIZE_MAX, &value) && value) budget = (size_t)value;
+        else if (!strcmp(argv[i], "--memory") && number(argv[i + 1], SIZE_MAX, &value) && value) memory_budget = (size_t)value;
         else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
         else if (!strcmp(argv[i], "--threads") && !strcmp(argv[i + 1], "auto")) threads = 0;
         else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value;
@@ -2721,7 +2752,7 @@ int main(int argc, char **argv) {
     pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */
     size_t active = 0, retained = 0;
     uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0,
-             ticks = owner.executor.ticks, shown = 0, checkpoints = 0;
+             ticks = owner.executor.ticks, shown = 0, checkpoints = 0, memory_refused = owner.executor.memory_refused;
     for (unsigned i = 0; i < threads; ++i) {
         shown += hosts[i].shown;
         active += hosts[i].active;
@@ -2731,6 +2762,7 @@ int main(int argc, char **argv) {
         bursts += hosts[i].local.bursts;
         events += hosts[i].local.burst_events;
         ticks += hosts[i].local.ticks;
+        memory_refused += hosts[i].local.memory_refused;
         expired += hosts[i].expired;
         stop_shard(&hosts[i]);
     }
@@ -2750,6 +2782,7 @@ int main(int argc, char **argv) {
         sent.refused += c.refused;
         pure_link_close(owner.executor.links[i]); /* what was not stored stays in the outbox */
     }
+    size_t state_bytes = owned ? pure_app_state_bytes(&owner.executor.app) : 0;
     if (owned) {
         checkpoints = owner.executor.checkpoints;
         stop_executor(&owner.executor);
@@ -2768,12 +2801,13 @@ int main(int argc, char **argv) {
             ",\"link_refused\":%" PRIu64 ",\"listener_packets\":%" PRIu64 ",\"listener_received\":%" PRIu64
             ",\"listener_deliveries\":%" PRIu64 ",\"listener_woken\":%" PRIu64 ",\"listener_remembered\":%" PRIu64
             ",\"listener_swept\":%" PRIu64 ",\"listener_active\":%" PRIu64 ",\"listener_keyed\":%" PRIu64
-            ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64 "}\n",
+            ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64
+            ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 "}\n",
             active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
             owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped,
             checkpoints, sent.packets, sent.records, sent.resent, sent.lost, sent.acknowledged, sent.stored,
             sent.refused, heard.packets, heard.received, heard.deliveries, heard.woken, heard.remembered, heard.swept,
-            heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused);
+            heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused);
     free(hosts);
     free(thread);
     return 0;
@@ -2781,11 +2815,13 @@ usage:
     fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
           "[--connections 1024] [--timeout-ms 10000] [--send OUTPUT=URL ...] [--name NAME] "
           "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] "
-          "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] "
+          "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] "
           "[--surface LIBRARY [-- ARGUMENTS]]\n"
           "  --send OUTPUT=pure://host:port/INPUT: to a Pure node's --link-port, the way between nodes (UDP, no\n"
           "      connection, docs/design/012-link-protocol.md; its key by --seal OUTPUT=KEY)\n"
           "  --send OUTPUT=http[s]://host:port/path: to anything else, over HTTP\n"
-          "  --senders FILE: more senders the --link-port takes, a line each: NAME KEY\n", stderr);
+          "  --senders FILE: more senders the --link-port takes, a line each: NAME KEY\n"
+          "  --memory BYTES: what the app's states may hold (lists' items, texts, sends not yet stored): an event\n"
+          "      past it is refused, 503 memory (docs/design/013-memory-budget.md)\n", stderr);
     return 2;
 }
diff --git a/tests/check.py b/tests/check.py
index 3c77dc4..e15a996 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -444,6 +444,20 @@ view leaders = each e in board {
 """
 
 
+def post(port, path, body):
+    """A JSON body posted on its own connection: status, X-Pure-Error."""
+    import http.client
+    connection = http.client.HTTPConnection('127.0.0.1', port, timeout=30)
+    try:
+        data = json.dumps(body).encode()
+        connection.request('POST', path, data, {'Content-Length': str(len(data)), 'Content-Type': 'application/json'})
+        response = connection.getresponse()
+        response.read()
+        return response.status, response.getheader('X-Pure-Error')
+    finally:
+        connection.close()
+
+
 def http(port, method, path, body=b''):
     """One request on its own connection: status, X-Pure-Error, content type, body."""
     import http.client
@@ -1074,6 +1088,131 @@ def server_links_refused(aup, pure, work):
     return {'said': said[0]['error'], 'packets': stats['a']['link_packets']}
 
 
+
+MEMORY_RECEIVER = """record Order {
+  item: text
+  count = 0
+}
+state kept: list of Order
+state got = 0
+input order = message {
+  item: text
+  count: u32
+  to: {
+    city: text
+  }
+}
+input clean = message {
+  below: u32
+}
+on order {
+  kept.append(Order { item: order.item, count: order.count })
+  got = got + 1
+}
+on clean {
+  for o in kept where o.count < clean.below {
+    kept.remove(o)
+  }
+}
+view count = got
+view held = kept.count
+"""
+
+
+def server_memory(aup, pure, work):
+    """What an app holds, bounded (--memory, docs/design/013-memory-budget.md).
+    A sender whose link's receiver is down fills its outbox: its next buy is
+    refused, 503 memory, and changes nothing, while its views still read;
+    once the receiver is up and stores what waited, it takes buys again. The
+    receiver, started again with a budget below what it keeps, refuses the
+    link's deliveries (and a client's order, 503 memory) but not the clean
+    that frees them; meanwhile its link hands the records over a few times a
+    second, not at every turn, and once freed it takes them all, once."""
+    import select
+    import socket
+    import time
+    sender_module, receiver_module = work / 'memory-sender.pmod', work / 'memory-receiver.pmod'
+    for module, text in ((sender_module, LINK_SENDER), (receiver_module, MEMORY_RECEIVER)):
+        source = module.with_suffix('.aup')
+        source.write_text(text)
+        compile_module(aup, source, module)
+
+    def free_port(kind=socket.SOCK_STREAM):
+        with socket.socket(socket.AF_INET, kind) as probe:
+            probe.bind(('127.0.0.1', 0))
+            return probe.getsockname()[1]
+
+    def start(args, port):
+        process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        ready, _, _ = select.select([process.stdout], [], [], 10)
+        line = process.stdout.readline() if ready else b''
+        assert line, f'host did not start: {process.stderr.read(2000) if process.poll() is not None else ""}'
+        return process, json.loads(line)
+
+    def stop(process):
+        process.terminate()
+        _, error = process.communicate(timeout=30)
+        assert process.returncode == 0, error[-1000:]
+        return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1]
+
+    def number(port, view):
+        return int(http(port, 'GET', f'/views/{view}')[3])
+
+    def until(test, seconds):  # test once more at most after the deadline: it may post
+        deadline = time.time() + seconds
+        while not test():
+            if time.time() > deadline:
+                return False
+            time.sleep(0.02)
+        return True
+
+    keys = {}
+    for name in ('sender', 'receiver'):
+        process, info = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'memory-key',
+                               '--key', work / f'memory-{name}.key'], 0)
+        keys[name] = info['public_key']
+        stop(process)
+    port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM)
+    receiving = [pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'memory-b',
+                 '--key', work / 'memory-receiver.key', '--link-port', link_port,
+                 '--sender', f'memory-sender.ordered={keys["sender"]}']
+    sender = start([pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'memory-a',
+                    '--key', work / 'memory-sender.key', '--send', f'ordered=pure://127.0.0.1:{link_port}/order',
+                    '--seal', f'ordered={keys["receiver"]}', '--memory', 1000], port_a)[0]
+    receiver, stats, item = None, {}, 'x' * 40
+    try:
+        accepted, refused = 0, None
+        while accepted < 200 and not refused:
+            status = post(port_a, '/inputs/buy', {'item': item, 'count': 1})
+            accepted += status[0] == 200
+            refused = status if status[0] != 200 else None
+        assert refused == (503, 'memory') and 3 <= accepted < 200, (accepted, refused)
+        assert number(port_a, 'sent') == accepted
+        receiver = start(receiving, port_b)[0]
+        assert until(lambda: number(port_b, 'count') == accepted, 20), number(port_b, 'count')
+        assert until(lambda: post(port_a, '/inputs/buy', {'item': item, 'count': 1})[0] == 200, 10)
+        assert until(lambda: number(port_b, 'count') == accepted + 1, 10)
+        stop(receiver)
+        receiver = start(receiving + ['--memory', 300], port_b)[0]  # below what it keeps
+        order = {'item': item, 'count': 1, 'to': {'city': 'Seoul'}}
+        assert post(port_b, '/inputs/order', order) == (503, 'memory')
+        for _ in range(3):
+            assert post(port_a, '/inputs/buy', {'item': item, 'count': 1})[0] == 200
+        time.sleep(2.0)
+        assert number(port_b, 'count') == accepted + 1 and number(port_a, 'sent') == accepted + 4
+        assert post(port_b, '/inputs/clean', {'below': 100})[0] == 200
+        assert until(lambda: number(port_b, 'count') == accepted + 4, 5), number(port_b, 'count')
+        assert number(port_b, 'held') == 3
+    finally:
+        for name, process in (('b', receiver), ('a', sender)):
+            if process:
+                stats[name] = stop(process)
+    assert stats['a']['memory_refused'] >= 1 and stats['a']['state_bytes'] <= 1000, stats['a']
+    assert 2 <= stats['b']['memory_refused'] <= 40 and stats['b']['state_bytes'] <= 300, stats['b']
+    return {'sends_held': accepted, 'receiver_refused': stats['b']['memory_refused'],
+            'link_resent': stats['a']['link_resent']}
+
 def server_links_independent(aup, pure, work):
     """A receiver written from docs/design/012-link-protocol.md alone, in
     Python with the cryptography package and no code of link/: a Pure node
@@ -2312,6 +2451,113 @@ def server_answers(aup, pure, work):
     return {'orders': sent, 'paid': paid, 'declined': declined, 'answers_after_restart': stats['answers']}
 
 
+
+RECEIPT_SOURCE = """record Receipt {
+  order: text
+  note: text
+  status = 0
+}
+state receipts: list of Receipt
+state sent = 0
+state answered = 0
+input buy = message {
+  order: text
+}
+output charge = message {
+  order: text
+}
+input charged = message {
+  order: text
+  note: text
+  http_status: u32
+}
+input clean = message {
+  below: u32
+}
+on buy {
+  sent = sent + 1
+  send charge { order: buy.order }
+}
+on charged {
+  answered = answered + 1
+  receipts.append(Receipt { order: charged.order, note: charged.note, status: charged.http_status })
+}
+on clean {
+  for r in receipts where r.status < clean.below {
+    receipts.remove(r)
+  }
+}
+view sent_view = sent
+view answered_view = answered
+view receipts_view = receipts.count
+"""
+
+
+def server_answers_memory(aup, pure, work):
+    """Answers (--reply) under a memory budget: an answer keeps a receipt
+    larger than its send, so once receipts fill the budget the next answer
+    waits (it is not lost, and not applied twice) and buys are refused, 503
+    memory. A clean that frees the receipts lets the answers in, each once."""
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+    import threading
+    import time
+    module = work / 'receipt.pmod'
+    module.with_suffix('.aup').write_text(RECEIPT_SOURCE)
+    compile_module(aup, module.with_suffix('.aup'), module)
+    seen, lock = {}, threading.Lock()
+
+    class Api(BaseHTTPRequestHandler):
+        protocol_version = 'HTTP/1.1'
+
+        def do_POST(self):
+            body = json.loads(self.rfile.read(int(self.headers['Content-Length'])))
+            with lock:
+                seen[body['order']] = seen.get(body['order'], 0) + 1
+            data = json.dumps({'note': 'n' * 200}).encode()
+            self.send_response(200)
+            self.send_header('Content-Type', 'application/json')
+            self.send_header('Content-Length', str(len(data)))
+            self.end_headers()
+            self.wfile.write(data)
+
+        def log_message(self, *args):
+            pass
+
+    server = ThreadingHTTPServer(('127.0.0.1', 0), Api)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    host = HttpHost([pure / 'build/pure-serve', '--module', module, '--store', work / 'receipt-store', '--memory', 3000,
+                     '--send', f'charge=http://127.0.0.1:{server.server_address[1]}/charge', '--reply', 'charge=charged'])
+
+    def number(view):
+        return int(http(host.port, 'GET', f'/views/{view}')[3])
+
+    try:
+        bought, refusals = 0, 0
+        while refusals < 3 and bought < 200:
+            status = post(host.port, '/inputs/buy', {'order': f'o{bought}'})
+            if status[0] == 200:
+                bought += 1
+            else:
+                assert status == (503, 'memory'), status
+                refusals += 1
+            time.sleep(0.02)
+        time.sleep(0.5)
+        assert 3 <= bought < 200 and number('sent_view') == bought, bought
+        waiting = bought - number('answered_view')
+        assert waiting >= 1, (bought, number('answered_view'))
+        assert post(host.port, '/inputs/clean', {'below': 1000})[0] == 200
+        deadline = time.time() + 10
+        while number('answered_view') < bought and time.time() < deadline:
+            time.sleep(0.02)
+        assert number('answered_view') == bought and number('receipts_view') == waiting, (number('answered_view'), waiting)
+    finally:
+        stats = host.stop()
+        server.shutdown()
+    with lock:
+        assert len(seen) == bought and all(count == 1 for count in seen.values()), seen
+    assert stats[-1]['memory_refused'] >= refusals + 1, stats[-1]
+    return {'bought': bought, 'answers_waited': waiting, 'refused': stats[-1]['memory_refused']}
+
 BEAT_SOURCE = """state beats = 0
 state last = 0
 state stamped = 0
@@ -2605,6 +2851,7 @@ def main():
         report['server_links_udp'] = server_links_udp(aup, pure, work)
         report['server_links_udp_lossy'] = server_links_udp_lossy(aup, pure, work)
         report['server_links_refused'] = server_links_refused(aup, pure, work)
+        report['server_memory'] = server_memory(aup, pure, work)
         report['server_links_independent'] = server_links_independent(aup, pure, work)
         report['server_links_moved'] = server_links_moved(aup, pure, work)
         report['server_webhook'] = server_webhook(aup, pure, work)
@@ -2614,6 +2861,7 @@ def main():
         report['server_rows'] = server_rows(aup, pure, work)
         report['server_arguments'] = server_arguments(aup, pure, work)
         report['server_answers'] = server_answers(aup, pure, work)
+        report['server_answers_memory'] = server_answers_memory(aup, pure, work)
     print(json.dumps(report))