Save app changes in bursts

A save now publishes everything since the last save in one commit: the
documents messages edited and the states that changed. If it fails, every
event since the last save is undone (values, text and list drafts, the
list undo log, event and tick counts), so an event reported as successful
is stored and a failed one leaves no trace.

By default each key, text and message event saves before it returns, as
before. With pure_app_defer events apply at once and a host saves a burst
with one commit; a key's document joins a burst once, so revisions move
one per save. On the Pi's SD card 600 events take 6.3 s saved one by one,
0.97 s in bursts of 8 and 0.34 s in bursts of 64. pure-run takes --batch N.

Tests compare a deferred stored app that saves at random and sometimes
stops unsaved with an app that replays only the saved events, undo a whole
burst on a failed save, and kill a process while it saves a burst.

9 files changed, 344 insertions, 98 deletions

diff --git a/prototypes/connections/module-list.c b/prototypes/connections/module-list.c
index ba67143..374bc78 100644
--- a/prototypes/connections/module-list.c
+++ b/prototypes/connections/module-list.c
@@ -234,9 +234,9 @@ PureCell pure_list_get(const PureList *list, size_t at, unsigned field) {
     return list->items[at][field];
 }
 
-void pure_undo_rollback(PureUndo *undo) {
+void pure_undo_rollback(PureUndo *undo, size_t mark) {
     Entry *entries = undo->entries;
-    for (size_t i = undo->count; i-- > 0;) {
+    for (size_t i = undo->count; i-- > mark;) {
         Entry *e = &entries[i];
         if (e->change == INSERTED) free_item(e->list, take(e->list, e->at));
         else if (e->change == REMOVED) place(e->list, e->at, e->item); /* capacity is still there */
@@ -245,7 +245,7 @@ void pure_undo_rollback(PureUndo *undo) {
             e->list->items[e->at][e->field] = e->old;
         }
     }
-    undo->count = 0;
+    undo->count = mark;
 }
 
 void pure_undo_commit(PureUndo *undo) {
diff --git a/prototypes/connections/module-list.h b/prototypes/connections/module-list.h
index e79ee46..b568a05 100644
--- a/prototypes/connections/module-list.h
+++ b/prototypes/connections/module-list.h
@@ -35,7 +35,8 @@ void pure_list_mirror(PureList *, PureDraft *, size_t header);
 PureResult pure_list_write(PureList *);
 /* Append the items stored in bytes; each field is checked. Before mirroring. */
 PureResult pure_list_read(PureList *, const unsigned char *bytes, size_t size);
-void pure_undo_rollback(PureUndo *);
+/* Undo the changes logged after mark (a count taken earlier), newest first. */
+void pure_undo_rollback(PureUndo *, size_t mark);
 void pure_undo_commit(PureUndo *);
 void pure_undo_free(PureUndo *);
 #endif
diff --git a/prototypes/connections/module-save.c b/prototypes/connections/module-save.c
index 40b3934..573ac3f 100644
--- a/prototypes/connections/module-save.c
+++ b/prototypes/connections/module-save.c
@@ -244,9 +244,15 @@ PureResult pure_save_open_values(PureApp *app, const PureModule *m, double *valu
 
 static bool same_version(PureVersion a, PureVersion b) { return a.ref == b.ref && a.bytes == b.bytes; }
 
-PureResult pure_save_commit(PureApp *app, const double *values, void *document, int64_t *revision) {
-    if (!pure_save_enabled(app))
-        return document ? app->data.commit(app->data.context, document, revision) : PURE_OK;
+PureResult pure_save_commit(PureApp *app, const double *values, void *const *documents, size_t document_count,
+                            int64_t *revisions) {
+    if (!pure_save_enabled(app)) {
+        /* Without a host that saves together, a save has one document at most. */
+        PureResult result = PURE_OK;
+        for (size_t i = 0; i < document_count && result == PURE_OK; ++i)
+            result = app->data.commit(app->data.context, documents[i], &revisions[i]);
+        return result;
+    }
     const PureModule *m = app->module;
     PureDraft *drafts[2 * PURE_MODULE_STATES + 1];
     uint32_t states[2 * PURE_MODULE_STATES + 1];
@@ -278,9 +284,8 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *document,
             drafts[count++] = app->stored_values;
         }
     }
-    if (!count && !document) return PURE_OK;
-    int64_t ignored;
-    PureResult result = app->data.save(app->data.context, document, drafts, count, document ? revision : &ignored);
+    if (!count && !document_count) return PURE_OK;
+    PureResult result = app->data.save(app->data.context, documents, document_count, drafts, count, revisions);
     for (size_t k = 0; result == PURE_OK && k < count; ++k)
         if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]);
     if (values_changed) {
diff --git a/prototypes/connections/module-save.h b/prototypes/connections/module-save.h
index 90e0701..b6db997 100644
--- a/prototypes/connections/module-save.h
+++ b/prototypes/connections/module-save.h
@@ -19,9 +19,10 @@ PureResult pure_save_open_list(PureApp *, const PureModule *m, uint32_t state, P
 PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, bool load);
 /* Store a list whose record changed in its draft, header and items anew. */
 PureResult pure_save_rewrite_list(const PureModule *m, uint32_t state, PureList *list, PureDraft *draft);
-/* Publish every state changed since the last save (values are the pending
- * numbers), with document when not NULL; revision is the document's. */
-PureResult pure_save_commit(PureApp *, const double *values, void *document, int64_t *revision);
+/* Publish the documents and every state changed since the last save
+ * (values are the states' numbers); revisions are the documents'. */
+PureResult pure_save_commit(PureApp *, const double *values, void *const *documents, size_t document_count,
+                            int64_t *revisions);
 
 /* A list with record's fields, taking from `from` each field of the same name
  * and kind and giving the others their defaults (module.c). */
diff --git a/prototypes/connections/module-store.c b/prototypes/connections/module-store.c
index 6335903..02d6361 100644
--- a/prototypes/connections/module-store.c
+++ b/prototypes/connections/module-store.c
@@ -42,17 +42,19 @@ static PureResult state(void *context, const char *name, PureDraft **out) {
     return pure_state_open(context, name, out);
 }
 
-/* The document and the app's changed states in one tree commit. */
-static PureResult save(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision) {
+enum { SAVED = PURE_APP_PENDING + 2 * PURE_MODULE_STATES + 1 };
+
+/* The documents and the app's changed states in one tree commit. */
+static PureResult save(void *context, void *const *documents, size_t document_count, PureDraft *const *states,
+                       size_t count, int64_t *revisions) {
     (void)context;
-    PureDraft *drafts[2 * PURE_MODULE_STATES + 2] = {0};
-    int64_t revisions[2 * PURE_MODULE_STATES + 2] = {0};
-    size_t n = 0;
-    if (count > 2 * PURE_MODULE_STATES + 1) return PURE_LIMIT;
-    if (document) drafts[n++] = document;
-    for (size_t i = 0; i < count; ++i) drafts[n++] = states[i];
-    PureResult result = pure_draft_commit_all(drafts, n, revisions);
-    if (document) *revision = revisions[0];
+    PureDraft *drafts[SAVED] = {0};
+    int64_t all[SAVED] = {0};
+    if (document_count > PURE_APP_PENDING || count > SAVED - document_count) return PURE_LIMIT;
+    for (size_t i = 0; i < document_count; ++i) drafts[i] = documents[i];
+    for (size_t i = 0; i < count; ++i) drafts[document_count + i] = states[i];
+    PureResult result = pure_draft_commit_all(drafts, document_count + count, all);
+    for (size_t i = 0; i < document_count; ++i) revisions[i] = all[i];
     return result;
 }
 
diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index 362919b..62f00c7 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -1075,13 +1075,14 @@ PureResult pure_app_start(PureApp *app, const PureModule *module, PureData data)
     return result;
 }
 
-PureResult pure_app_save(PureApp *app) {
-    int64_t revision;
-    return pure_save_commit(app, app->values, NULL, &revision);
+static bool pending(const PureApp *app, const void *document) {
+    for (unsigned i = 0; i < app->pending_count; ++i) if (app->pending[i].document == document) return true;
+    return false;
 }
 
 void pure_app_finish(PureApp *app) {
-    if (app->document) app->data.close(app->data.context, app->document);
+    /* A document in the unsaved burst stays open until the save. */
+    if (app->document && !pending(app, app->document)) app->data.close(app->data.context, app->document);
     free(app->copied);
     app->document = NULL;
     app->copied = NULL;
@@ -1090,8 +1091,75 @@ void pure_app_finish(PureApp *app) {
     app->arena.used = 0;
 }
 
+/* The start of a burst: what a failed save returns to. Text and stored list
+ * drafts stay held until the save, so every version since stays readable. */
+static void open_burst(PureApp *app) {
+    if (app->open) return;
+    const PureModule *m = app->module;
+    memcpy(app->durable_values, app->values, sizeof(app->values));
+    app->durable_ticks = app->ticks;
+    app->durable_events = app->events;
+    for (uint32_t i = 0; i < m->states; ++i) {
+        if (app->texts[i]) {
+            pure_draft_hold(app->texts[i], true);
+            app->durable_texts[i] = pure_draft_version(app->texts[i]);
+        }
+        if (app->list_drafts[i]) {
+            pure_draft_hold(app->list_drafts[i], true);
+            app->durable_lists[i] = pure_draft_version(app->list_drafts[i]);
+        }
+    }
+    app->open = true;
+}
+
+/* End a burst, keeping it (saved) or undoing all of it. The documents it held
+ * close, except the current event's, which pure_app_finish closes. */
+static void close_burst(PureApp *app, bool keep) {
+    for (unsigned i = 0; i < app->pending_count; ++i)
+        if (app->pending[i].document != app->document) app->data.close(app->data.context, app->pending[i].document);
+    app->pending_count = 0;
+    if (!app->open) return;
+    const PureModule *m = app->module;
+    for (uint32_t i = 0; i < m->states; ++i) {
+        if (app->texts[i]) {
+            if (!keep) pure_draft_restore(app->texts[i], app->durable_texts[i]);
+            pure_draft_hold(app->texts[i], false);
+        }
+        if (app->list_drafts[i]) {
+            if (!keep) pure_draft_restore(app->list_drafts[i], app->durable_lists[i]);
+            pure_draft_hold(app->list_drafts[i], false);
+        }
+    }
+    if (keep) pure_undo_commit(&app->undo);
+    else {
+        pure_undo_rollback(&app->undo, 0);
+        memcpy(app->values, app->durable_values, sizeof(app->values));
+        app->ticks = app->durable_ticks;
+        app->events = app->durable_events;
+    }
+    app->open = false;
+}
+
+PureResult pure_app_save(PureApp *app) {
+    void *documents[PURE_APP_PENDING];
+    int64_t revisions[PURE_APP_PENDING];
+    size_t count = 0;
+    for (unsigned i = 0; i < app->pending_count; ++i)
+        if (app->pending[i].edited) documents[count++] = app->pending[i].document;
+    PureResult result = pure_save_commit(app, app->values, documents, count, revisions);
+    close_burst(app, result == PURE_OK);
+    return result;
+}
+
+bool pure_app_defer(PureApp *app, bool deferred) {
+    if (deferred && !pure_save_enabled(app)) return false;
+    app->deferred = deferred;
+    return true;
+}
+
 void pure_app_stop(PureApp *app) {
     pure_app_finish(app);
+    close_burst(app, true); /* unsaved changes are dropped with the app */
     for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i) {
         pure_draft_close(app->texts[i]);
         pure_list_free(app->lists[i]);
@@ -1104,38 +1172,26 @@ void pure_app_stop(PureApp *app) {
     memset(app, 0, sizeof(*app));
 }
 
-/* Every text draft and stored list draft is held for the event; on failure
- * each returns to the version it had when the event began. */
-typedef struct { PureVersion before[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; } Texts;
+/* Where one event began inside the burst: a failed event returns there. */
+typedef struct { PureVersion texts[PURE_MODULE_STATES], lists[PURE_MODULE_STATES]; size_t undo; bool opened; } Mark;
 
-static void hold_texts(PureApp *app, Texts *texts) {
+static void mark_event(PureApp *app, Mark *mark) {
+    mark->opened = !app->open;
+    open_burst(app);
     for (uint32_t i = 0; i < app->module->states; ++i) {
-        if (app->texts[i]) {
-            pure_draft_hold(app->texts[i], true);
-            texts->before[i] = pure_draft_version(app->texts[i]);
-        }
-        if (app->list_drafts[i]) {
-            pure_draft_hold(app->list_drafts[i], true);
-            texts->lists[i] = pure_draft_version(app->list_drafts[i]);
-        }
+        if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]);
+        if (app->list_drafts[i]) mark->lists[i] = pure_draft_version(app->list_drafts[i]);
     }
+    mark->undo = app->undo.count;
 }
 
-/* Texts and stored lists return to their versions and lists undo their log,
- * or all keep what the event did. */
-static void settle(PureApp *app, const Texts *texts, bool keep) {
+static void undo_event(PureApp *app, const Mark *mark) {
     for (uint32_t i = 0; i < app->module->states; ++i) {
-        if (app->texts[i]) {
-            if (!keep) pure_draft_restore(app->texts[i], texts->before[i]);
-            pure_draft_hold(app->texts[i], false);
-        }
-        if (app->list_drafts[i]) {
-            if (!keep) pure_draft_restore(app->list_drafts[i], texts->lists[i]);
-            pure_draft_hold(app->list_drafts[i], false);
-        }
+        if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]);
+        if (app->list_drafts[i]) pure_draft_restore(app->list_drafts[i], mark->lists[i]);
     }
-    if (keep) pure_undo_commit(&app->undo);
-    else pure_undo_rollback(&app->undo);
+    pure_undo_rollback(&app->undo, mark->undo);
+    if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
 }
 
 /* All handlers of one event run in module order against pending states. */
@@ -1152,29 +1208,24 @@ static PureResult run_handlers(PureApp *app, uint32_t event, Event *ev, double *
     return PURE_OK;
 }
 
-/* A key or text event stores what it changed before it counts; a frame
- * leaves that to the next event (frames come 60 times a second). */
+/* A key or text event saves before it returns unless saving is deferred; a
+ * frame leaves that to the next event (frames come 60 times a second). */
 static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
     double pending[PURE_MODULE_STATES];
-    Texts texts;
-    hold_texts(app, &texts);
+    Mark mark;
+    mark_event(app, &mark);
     PureResult result = run_handlers(app, event, ev, pending);
-    int64_t revision;
-    if (result == PURE_OK && event) result = pure_save_commit(app, pending, NULL, &revision);
-    settle(app, &texts, result == PURE_OK);
-    if (result == PURE_OK) {
-        memcpy(app->values, pending, sizeof(pending));
-        ++app->events;
-    }
-    return result;
+    if (result != PURE_OK) { undo_event(app, &mark); return result; }
+    memcpy(app->values, pending, sizeof(pending));
+    ++app->events;
+    if (!event) ++app->ticks;
+    return event && !app->deferred ? pure_app_save(app) : PURE_OK;
 }
 
 PureResult pure_app_frame(PureApp *app) {
     pure_app_finish(app);
     Event ev = {.app = app, .arena = &app->arena, .dt = 1.0 / 60.0};
-    PureResult result = dispatch(app, 0, &ev);
-    if (result == PURE_OK) ++app->ticks;
-    return result;
+    return dispatch(app, 0, &ev);
 }
 
 PureResult pure_app_input(PureApp *app, uint32_t input, PureEdge edge) {
@@ -1220,32 +1271,47 @@ PureResult pure_app_message(PureApp *app, uint32_t input, const char *key, int64
         return PURE_INVALID;
     const PureInputShape *shape = &m->input[input];
     if (payload.size < shape->prefix) return fault(app, PURE_FAULT_LIMIT, UINT32_MAX);
-    app->payload = payload;
     if (shape->keyed && m->collections) {
         if (!app->data.open) return PURE_INVALID;
+        /* A key joins a burst once, so its revision moves once per save. */
+        bool seen = app->pending_count == PURE_APP_PENDING;
+        for (unsigned i = 0; i < app->pending_count && !seen; ++i) seen = !strcmp(app->pending[i].key, key);
+        if (seen) {
+            PureResult saved = pure_app_save(app);
+            if (saved != PURE_OK) return saved;
+        }
         PureResult opened = app->data.open(app->data.context, 0, key, &app->document, &app->revision);
-        if (opened != PURE_OK) { app->document = NULL; pure_app_finish(app); return opened; }
+        if (opened != PURE_OK) { app->document = NULL; return opened; }
         if (base >= 0 && base != app->revision) { pure_app_finish(app); return PURE_CONFLICT; }
     }
+    app->payload = payload;
     Event ev = {.app = app, .shape = shape, .arena = &app->arena, .exists = app->revision > 0};
-    double pending[PURE_MODULE_STATES];
-    Texts texts;
-    hold_texts(app, &texts);
-    PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending);
+    double pending_values[PURE_MODULE_STATES];
+    Mark mark;
+    mark_event(app, &mark);
+    PureResult result = run_handlers(app, 1 + 2 * input, &ev, pending_values);
     if (result == PURE_OK && ev.modified && base < 0) result = PURE_INVALID;
-    /* Commit rebases the working version and releasing texts may reclaim
+    /* A save rebases an edited document and releasing texts may reclaim
      * earlier text versions: keep such a reply by copying it first. */
     if (result == PURE_OK && app->has_reply &&
         ((app->reply.source == FROM_DOCUMENT && ev.modified) || app->reply.source == FROM_TEXT))
         result = keep_reply(app);
-    if (result == PURE_OK)
-        result = pure_save_commit(app, pending, ev.modified ? app->document : NULL, &app->revision);
-    /* States, texts and lists change only with a successful commit (or no document edit). */
-    settle(app, &texts, result == PURE_OK);
-    if (result == PURE_OK) {
-        memcpy(app->values, pending, sizeof(pending));
-        ++app->events;
-    } else if (result != PURE_UNCERTAIN) pure_app_finish(app);
+    if (result != PURE_OK) {
+        undo_event(app, &mark);
+        pure_app_finish(app);
+        return result;
+    }
+    memcpy(app->values, pending_values, sizeof(pending_values));
+    ++app->events;
+    if (app->document) {
+        app->pending[app->pending_count].document = app->document;
+        app->pending[app->pending_count].edited = ev.modified;
+        memcpy(app->pending[app->pending_count++].key, key, strlen(key) + 1);
+        app->revision += ev.modified; /* a save moves an edited document one revision */
+    }
+    if (app->deferred) return PURE_OK;
+    result = pure_app_save(app);
+    if (result != PURE_OK) pure_app_finish(app);
     return result;
 }
 
@@ -1359,6 +1425,10 @@ static PureResult keep_list(PureList **out, PureList *from, const PureModule *ol
  * a state the new module drops stays stored as it was. */
 PureResult pure_app_replace(PureApp *app, const PureModule *module) {
     pure_app_finish(app);
+    if (app->open || app->pending_count) {
+        PureResult saved = pure_app_save(app); /* the burst belongs to the module that made it */
+        if (saved != PURE_OK) return saved;
+    }
     const PureModule *old = app->module;
     bool stored = pure_save_enabled(app);
     double values[PURE_MODULE_STATES] = {0};
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index d2d4b7a..c5d4f52 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -19,7 +19,7 @@ enum {
     PURE_MODULE_NAME = 32, PURE_MODULE_STRINGS = 32768, PURE_MODULE_HEADER = 40,
     PURE_MODULE_FIELDS = 32, PURE_MODULE_COLLECTIONS = 1, PURE_MODULE_CONSTANTS = 64,
     PURE_MODULE_CONSTANT_BYTES = 8192, PURE_MODULE_RECORDS = 8, PURE_MODULE_RECORD_FIELDS = 64,
-    PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8
+    PURE_RECORD_FIELDS = 16, PURE_MODULE_JOINS = 8, PURE_APP_PENDING = 16
 };
 
 typedef enum {
@@ -123,10 +123,11 @@ typedef struct {
     void (*close)(void *context, void *document);
     /* Optional: the host keeps the app's states. state opens a runtime-owned
      * value by name as a draft (a missing one is empty at revision 0); save
-     * publishes the given state drafts together with document (or NULL),
-     * all or nothing, and names the document's new revision. */
+     * publishes the given documents and state drafts together, all or
+     * nothing, and names each document's new revision. */
     PureResult (*state)(void *context, const char *name, PureDraft **out);
-    PureResult (*save)(void *context, void *document, PureDraft *const *states, size_t count, int64_t *revision);
+    PureResult (*save)(void *context, void *const *documents, size_t document_count, PureDraft *const *states,
+                       size_t count, int64_t *revisions);
 } PureData;
 
 /* A byte or text value inside one event or view: a range of the payload, a
@@ -138,10 +139,15 @@ typedef struct { unsigned char *bytes; size_t used, capacity; } PureArena;
  * runtime-owned drafts; cursors are positions the runtime keeps on character
  * boundaries; lists are runtime-owned and an index always names an item
  * (or 0 in an empty list). Every event is atomic: on failure nothing changes.
- * When the data host keeps states, the app starts from the stored ones and
- * every key, text and message event stores what it changed with its own
- * commit (with the message's document); frame events are stored with the
- * next one, or by pure_app_save. */
+ *
+ * Saving. When the data host keeps states the app starts from the stored
+ * ones. A save publishes, in one commit, the documents edited by messages and
+ * the states changed since the last save; if it fails, every event since the
+ * last save is undone. By default each key, text and message event saves
+ * before it returns (a frame waits for the next one). Deferred, events apply
+ * at once and a host saves a burst with one commit: an event is durable once
+ * a later save succeeds, and each key's document joins a burst once (a second
+ * message for it saves the burst first). */
 typedef struct {
     const PureModule *module;
     PureData data;
@@ -152,6 +158,14 @@ typedef struct {
     PureDraft *list_drafts[PURE_MODULE_STATES]; /* stored lists, when kept */
     PureDraft *stored_values;                   /* numbers, cursors and indices, when kept */
     PureVersion saved[PURE_MODULE_STATES];      /* text and list versions at the last save */
+    /* Events since the last save: what a failed save returns to, and the
+     * documents messages opened (held until the save). */
+    bool deferred, open;
+    double durable_values[PURE_MODULE_STATES];
+    PureVersion durable_texts[PURE_MODULE_STATES], durable_lists[PURE_MODULE_STATES];
+    uint64_t durable_ticks, durable_events;
+    struct { void *document; char key[65]; bool edited; } pending[PURE_APP_PENDING];
+    unsigned pending_count;
     PureFault fault;
     uint32_t fault_pc;
     uint64_t ticks, events;
@@ -166,8 +180,11 @@ typedef struct {
 } PureApp;
 
 PureResult pure_app_start(PureApp *, const PureModule *, PureData data);
-/* Store every state changed since the last save (after frame events). */
+/* Publish every change since the last save in one commit, or undo them all. */
 PureResult pure_app_save(PureApp *);
+/* Defer saving to pure_app_save (false: save after each event). Needs a data
+ * host that saves documents together (PureData.save). */
+bool pure_app_defer(PureApp *, bool deferred);
 /* Leaves unsaved frame changes unsaved: call pure_app_save first to keep them. */
 void pure_app_stop(PureApp *);
 PureResult pure_app_frame(PureApp *);
@@ -176,9 +193,11 @@ PureResult pure_app_input(PureApp *, uint32_t input, PureEdge);
 PureResult pure_app_text(PureApp *, uint32_t input, PureSpan text);
 /* Deliver a message. base is the revision the sender edited, or -1 for none;
  * a handler that edits the document needs one. payload stays alive until
- * pure_app_finish. On return app->revision is the document's revision. */
+ * pure_app_finish. On return app->revision is the document's revision
+ * (deferred: the revision the next save gives it). */
 PureResult pure_app_message(PureApp *, uint32_t input, const char *key, int64_t base, PureSpan payload);
-/* Visit the reply of the last message in order, without copying. */
+/* Visit the reply of the last message in order, without copying, until the
+ * next event (deferred: copy it before a later save too). */
 PureResult pure_app_reply(const PureApp *, bool (*visit)(void *, PureSpan), void *context);
 void pure_app_finish(PureApp *);
 /* A number view; false for a text view or a failed evaluation. */
diff --git a/prototypes/connections/pure-run.c b/prototypes/connections/pure-run.c
index 7b6e253..6e4a6e1 100644
--- a/prototypes/connections/pure-run.c
+++ b/prototypes/connections/pure-run.c
@@ -8,7 +8,8 @@
 /* Test host for a verified module: reads commands from stdin and prints one
  * JSON snapshot per command. Not a product interface. With --store the app's
  * states live in that tree store: it starts from them, and frame changes are
- * saved at the end of input or by save.
+ * saved at the end of input or by save. --batch N defers saving and saves
+ * every N commands (and at the end), so N events share one commit.
  *   down NAME | up NAME | frame [N] | type NAME HEX|- | replace PATH | save | show
  * Text states and text views print as JSON strings; cursors as byte positions;
  * lists as arrays of objects; indices as positions; a table view as an array
@@ -134,9 +135,11 @@ static int hex(char c) {
 
 int main(int argc, char **argv) {
     static PureModule modules[2];
-    bool stored = argc == 4 && !strcmp(argv[2], "--store");
-    if ((argc != 2 && !stored) || !load(argv[1], &modules[0])) {
-        if (argc != 2 && !stored) fputs("pure-run MODULE [--store PATH] < commands\n", stderr);
+    bool stored = argc >= 4 && !strcmp(argv[2], "--store");
+    long batch = argc == 6 && stored && !strcmp(argv[4], "--batch") ? strtol(argv[5], NULL, 10) : 0;
+    bool usage = argc != 2 && !(stored && argc == 4) && !(batch > 0 && batch <= 100000);
+    if (usage || !load(argv[1], &modules[0])) {
+        if (usage) fputs("pure-run MODULE [--store PATH [--batch N]] < commands\n", stderr);
         return 2;
     }
     if (stored && pure_store_init(argv[3]) != PURE_OK) { fprintf(stderr, "cannot open store %s\n", argv[3]); return 2; }
@@ -144,6 +147,8 @@ int main(int argc, char **argv) {
     PureApp app;
     PureResult started = pure_app_start(&app, &modules[active], stored ? pure_tree_data(argv[3]) : (PureData){0});
     if (started != PURE_OK) { fprintf(stderr, "cannot start (%d)\n", (int)started); return 2; }
+    if (batch && !pure_app_defer(&app, true)) { fputs("cannot defer saving\n", stderr); return 2; }
+    long unsaved = 0;
     static char line[1 << 16];
     static unsigned char text[1 << 15];
     int status = 0;
@@ -176,6 +181,10 @@ int main(int argc, char **argv) {
         } else if (!strcmp(command, "save")) {
             if (pure_app_save(&app) != PURE_OK) failure = "save failed";
         } else if (strcmp(command, "show")) failure = "unknown command";
+        if (batch && strcmp(command, "show") && ++unsaved == batch) {
+            unsaved = 0;
+            if (pure_app_save(&app) != PURE_OK) failure = "save failed";
+        }
         if (failure) status = 1;
         show(&app, failure);
     }
diff --git a/prototypes/connections/test-module.c b/prototypes/connections/test-module.c
index 64409b2..e1d1ccf 100644
--- a/prototypes/connections/test-module.c
+++ b/prototypes/connections/test-module.c
@@ -1574,6 +1574,144 @@ static void stored_states(void) {
     stored_migration();
 }
 
+/* ---- Deferred saving ---------------------------------------------------- */
+
+typedef struct { uint32_t input; const char *word; } Step;
+
+static PureResult apply_step(PureApp *app, Step step) {
+    if (step.word) return pure_app_text(app, step.input, (PureSpan){(const unsigned char *)step.word, strlen(step.word)});
+    return pure_app_input(app, step.input, PURE_EDGE_DOWN);
+}
+
+/* Deferred events apply at once and last only through a save: an app that
+ * stops without saving comes back as it was at the last save. The oracle is
+ * an app in memory that replays exactly the saved events. */
+static void deferred_equivalence(Spec spec, const char *const *words, size_t word_count, uint32_t seed) {
+    char *path = new_store();
+    CHECK(load(spec, &module_a) == PURE_OK);
+    static Step history[2000];
+    size_t saved = 0, applied = 0;
+    PureApp memory, stored;
+    CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK);
+    CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK);
+    CHECK(pure_app_defer(&stored, true) && !pure_app_defer(&memory, true));
+    unsigned lost = 0, saves = 0;
+    for (unsigned step = 0; step < 600; ++step) {
+        Step next = {next_random(&seed) % module_a.inputs, NULL};
+        if (module_a.input[next.input].kind == PURE_KIND_TEXT) next.word = words[next_random(&seed) % word_count];
+        PureResult a = apply_step(&memory, next), b = apply_step(&stored, next);
+        CHECK(a == b);
+        if (a == PURE_OK) history[applied++] = next;
+        same_states(&memory, &stored);
+        uint32_t roll = next_random(&seed) % 16;
+        if (roll < 3) {
+            CHECK(pure_app_save(&stored) == PURE_OK);
+            saved = applied;
+            ++saves;
+        } else if (roll == 3) {
+            /* Stop without saving: only saved events come back. */
+            pure_app_stop(&stored);
+            pure_app_stop(&memory);
+            CHECK(pure_app_start(&stored, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&stored, true));
+            CHECK(pure_app_start(&memory, &module_a, (PureData){0}) == PURE_OK);
+            for (size_t k = 0; k < saved; ++k) CHECK(apply_step(&memory, history[k]) == PURE_OK);
+            lost += (unsigned)(applied - saved);
+            applied = saved;
+            same_states(&memory, &stored);
+        }
+    }
+    CHECK(saves > 50 && lost > 0);
+    pure_app_stop(&memory);
+    pure_app_stop(&stored);
+    remove_store(path);
+}
+
+/* A burst of events is saved with one commit: when that commit fails every
+ * event of the burst is undone, and a process that dies while saving comes
+ * back with all of the burst or none of it. */
+static void deferred_bursts(void) {
+    char *path = new_store();
+    Spec spec = {crash_nums, 1, crash_inputs, 1, NULL, 0, crash_handlers, 1, crash_code, LEN(crash_code), NULL, 0,
+                 crash_states, 4, crash_constants, 2, NULL, 0, crash_records, 1, NULL, 0};
+    CHECK(load(spec, &module_a) == PURE_OK);
+    PureApp app;
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true));
+    for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK);
+    CHECK(consistent(&app) == 5 && app.events == 5);
+    for (fail_stage = PURE_AFTER_HEADER; fail_stage <= PURE_BEFORE_RENAME; fail_stage += 2) {
+        CHECK(pure_app_save(&app) == PURE_STORE_ERROR && consistent(&app) == 0 && app.events == 0);
+        for (int i = 0; i < 5; ++i) CHECK(pure_app_input(&app, 0, PURE_EDGE_DOWN) == PURE_OK);
+    }
+    fail_stage = -1;
+    CHECK(pure_app_save(&app) == PURE_OK && consistent(&app) == 5);
+    pure_app_stop(&app);
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == 5);
+    pure_app_stop(&app);
+    unsigned expected = 5;
+    for (int stage = PURE_AFTER_HEADER; stage <= PURE_BEFORE_DIR_SYNC; ++stage) {
+        pid_t child = fork();
+        CHECK(child >= 0);
+        if (!child) {
+            PureApp doomed;
+            if (pure_app_start(&doomed, &module_a, pure_tree_data(path)) != PURE_OK || !pure_app_defer(&doomed, true)) _exit(1);
+            for (int i = 0; i < 7; ++i) if (pure_app_input(&doomed, 0, PURE_EDGE_DOWN) != PURE_OK) _exit(2);
+            crash = 1;
+            fail_stage = stage;
+            pure_app_save(&doomed);
+            _exit(0);
+        }
+        int status;
+        CHECK(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 77);
+        if (stage == PURE_BEFORE_DIR_SYNC) expected += 7;
+        CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && consistent(&app) == expected);
+        pure_app_stop(&app);
+    }
+    remove_store(path);
+}
+
+/* Messages in a burst: documents of different keys commit together with the
+ * states, each moving one revision; a second message for a key saves the
+ * burst first, and so does a full burst. */
+static void deferred_messages(void) {
+    char *path = new_store();
+    CHECK(load(editor_spec, &module_a) == PURE_OK);
+    PureApp app;
+    unsigned char buffer[64];
+    char key[8];
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && pure_app_defer(&app, true));
+    for (int i = 0; i < 3; ++i) {
+        snprintf(key, sizeof(key), "k%d", i);
+        CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "ab")) == PURE_OK && app.revision == 1);
+    }
+    CHECK(app.values[0] == 3 && app.pending_count == 3 && app.open);
+    /* The same key again: the burst is saved first, then this edit joins a new one. */
+    CHECK(pure_app_message(&app, 1, "k0", 1, edit_payload(buffer, 2, 0, "c")) == PURE_OK && app.revision == 2);
+    CHECK(app.pending_count == 1 && app.values[0] == 4);
+    CHECK(pure_app_message(&app, 1, "k1", 0, edit_payload(buffer, 0, 0, "x")) == PURE_CONFLICT);
+    CHECK(pure_app_save(&app) == PURE_OK && !app.open && !app.pending_count);
+    for (int i = 0; i < PURE_APP_PENDING + 1; ++i) {
+        snprintf(key, sizeof(key), "m%d", i);
+        CHECK(pure_app_message(&app, 1, key, 0, edit_payload(buffer, 0, 0, "y")) == PURE_OK);
+        CHECK(app.pending_count == (i < PURE_APP_PENDING ? (unsigned)i + 1 : 1)); /* full: saved on the way */
+    }
+    pure_app_stop(&app);           /* the last edit was never saved */
+    CHECK(pure_app_start(&app, &module_a, pure_tree_data(path)) == PURE_OK && app.values[0] == 4 + PURE_APP_PENDING);
+    CHECK(pure_app_message(&app, 0, "k0", -1, (PureSpan){0}) == PURE_OK && app.revision == 2 && replied(&app, "abc"));
+    CHECK(pure_app_message(&app, 0, "m16", -1, (PureSpan){0}) == PURE_NOT_FOUND);
+    CHECK(pure_app_message(&app, 0, "m15", -1, (PureSpan){0}) == PURE_OK && replied(&app, "y"));
+    pure_app_stop(&app);
+    remove_store(path);
+}
+
+static void deferred(void) {
+    static const char *const list_words[] = {"a", "\xea\xb0\x80", "", "\n", "\xf0\x9f\x98\x80"};
+    static const char *const text_words[] = {"a", "\xea\xb0\x80", "\n", "\xf0\x9f\x98\x80", "xy\nz", ""};
+    deferred_equivalence(list_spec, list_words, LEN(list_words), 0x2545f491u);
+    deferred_equivalence(text_spec, text_words, LEN(text_words), 0x1b873593u);
+    deferred_bursts();
+    deferred_messages();
+}
+
 int main(void) {
     behaviour();
     faults();
@@ -1588,7 +1726,8 @@ int main(void) {
     tables();
     table_mutations();
     stored_states();
+    deferred();
     printf("{\"sizeof_module\":%zu,\"sizeof_app\":%zu}\n", sizeof(PureModule), sizeof(PureApp));
-    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration\"}");
+    puts("{\"module_tests\":\"verified-load-atomic-events-faults-signed-zero-if-else-replace-by-name-rejections-mutations-messages-documents-typed-verification-text-cursors-utf8-views-rollback-replace-lists-indices-undo-migration-joins-tables-columns-pick-slide-stored-equivalence-crashes-frames-messages-migration-deferred-bursts\"}");
     return 0;
 }