Keep the numbers heard as a list: a save logs only the ones it changed

The last number heard from each source was written whole into the links
value at every save, and counted in two bytes: a node that heard from
more than 65,535 sources stored a wrong count, and each save cost as many
bytes as there were sources. The numbers are now a runtime-owned list
{source, seq} in order of source, stored as a list state is (its image
"0heard" and its changes in the log), changed with the event that
delivers, so a failed save takes them back with it. Links stored before
(PLK1) move their numbers to the list at a start and are stored again
without them (PLK2).

5 files changed, 290 insertions, 123 deletions

diff --git a/core/module-save.c b/core/module-save.c
index ddc2797..968e65b 100644
--- a/core/module-save.c
+++ b/core/module-save.c
@@ -42,12 +42,21 @@ unsigned long pure_save_test_checkpoints;
 enum { JOURNAL_MIN = 1 << 20 };
 #define LOG_KEPT (UINT64_C(64) << 20)
 #endif
-/* links: "PLK1"; u8 outputs, each {u8 name length, name, u8 field count, the
+/* links: "PLK2"; u8 outputs, each {u8 name length, name, u8 field count, the
  * fields' types, u64le last number} (the module's outputs, then those it
- * dropped, without fields); u16le sources, each {u8 length, the name, u64le
- * last number heard}; then the sent messages to the end, each {u8 output (in
- * the table), u64le number, u32le size, payload}. */
-static const unsigned char links_magic[4] = {'P', 'L', 'K', '1'};
+ * dropped, without fields); then the sent messages to the end, each {u8
+ * output (in the table), u64le number, u32le size, payload}. Before the
+ * heard list, "PLK1" kept the numbers heard between the two: u16le sources,
+ * each {u8 length, the name, u64le last number heard}; a start moves them to
+ * the heard list.
+ *
+ * The heard list (module.h, PureApp) is stored as a list state is, by a name
+ * no state has (a state's is an identifier), with the fields source (text)
+ * and seq (number): the numbers of every source, which only grow with the
+ * sources a node hears from, cost a save only the records of their changes. */
+static const unsigned char links_magic[4] = {'P', 'L', 'K', '2'}, links_magic_1[4] = {'P', 'L', 'K', '1'};
+static const char heard_name[] = "0heard", heard_names[2][PURE_MODULE_NAME] = {"source", "seq"};
+static const uint8_t heard_text[2] = {1, 0};
 static const PureVersion unsaved = {UINT64_MAX, UINT64_MAX};
 
 static unsigned char *put_le(unsigned char *at, uint64_t value, unsigned width) {
@@ -135,22 +144,38 @@ PURE_COLD PureResult pure_save_open_text(PureApp *app, const PureModule *m, uint
 
 enum { IMAGE_HEADER = 5 + PURE_RECORD_FIELDS * (2 + PURE_MODULE_NAME) + 8 };
 
-static size_t image_header(const PureModule *m, uint32_t record, uint64_t seq, unsigned char *out) {
-    const PureRecord *r = &m->record[record];
+static size_t header_of(unsigned count, const uint8_t *text, const char (*names)[PURE_MODULE_NAME], uint64_t seq,
+                        unsigned char *out) {
     size_t at = 0;
     memcpy(out, image_magic, 4);
     at = 4;
-    out[at++] = (unsigned char)r->count;
-    for (unsigned f = 0; f < r->count; ++f) {
-        const char *name = m->record_field_names[r->first + f];
-        out[at++] = m->record_field_text[r->first + f];
-        out[at++] = (unsigned char)strlen(name);
-        memcpy(out + at, name, strlen(name));
-        at += strlen(name);
+    out[at++] = (unsigned char)count;
+    for (unsigned f = 0; f < count; ++f) {
+        out[at++] = text[f];
+        out[at++] = (unsigned char)strlen(names[f]);
+        memcpy(out + at, names[f], strlen(names[f]));
+        at += strlen(names[f]);
     }
     return (size_t)(put_le(out + at, seq, 8) - out);
 }
 
+static size_t image_header(const PureModule *m, uint32_t record, uint64_t seq, unsigned char *out) {
+    const PureRecord *r = &m->record[record];
+    return header_of(r->count, m->record_field_text + r->first, m->record_field_names + r->first, seq, out);
+}
+
+PureList *pure_heard_new(void) { return pure_list_new(2, heard_text); }
+
+/* A stored list by its place: a list state, or the heard list (PURE_LIST_HEARD). */
+static PureList *stored_list(const PureApp *app, uint32_t i) {
+    if (i == PURE_LIST_HEARD) return app->heard;
+    return i < app->module->states && app->module->state_type[i] == PURE_STATE_LIST ? app->lists[i] : NULL;
+}
+
+static const char *stored_name(const PureApp *app, uint32_t i) {
+    return i == PURE_LIST_HEARD ? heard_name : app->module->state_names[i];
+}
+
 static const char *directory_of(const PureApp *app) {
     return app->data.directory ? app->data.directory(app->data.context) : NULL;
 }
@@ -257,7 +282,7 @@ PURE_COLD PureResult pure_save_open_log(PureApp *app) {
 void pure_save_release(PureApp *app) {
     if (!app->log) return;
     uint64_t kept = app->seq + 1;
-    for (uint32_t i = 0; i < PURE_MODULE_STATES; ++i)
+    for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i)
         if (app->logged_since[i] && app->logged_since[i] < kept) kept = app->logged_since[i];
     pure_log_release(app->log, kept - 1);
 }
@@ -408,6 +433,49 @@ PURE_COLD PureResult pure_save_open_list(PureApp *app, const PureModule *m, uint
     return result;
 }
 
+PURE_COLD PureResult pure_save_open_heard(PureApp *app, PureList **out, PureSaved *saved) {
+    *saved = (PureSaved){0};
+    *out = NULL;
+    if (!directory_of(app)) return PURE_STORE_ERROR;
+    unsigned char *bytes = NULL, header[IMAGE_HEADER];
+    size_t size = 0, length = header_of(2, heard_text, heard_names, 0, header) - 8;
+    PureResult result = pure_image_read(directory_of(app), heard_name, &bytes, &size);
+    bool image = result == PURE_OK;
+    uint64_t holds = image && size >= length + 8 && !memcmp(bytes, header, length) ? get_le(bytes + length, 8) : 0;
+    if (image && (size < length + 8 || memcmp(bytes, header, length))) result = PURE_STORE_ERROR;
+    else if (image || result == PURE_NOT_FOUND) {
+        result = (*out = pure_heard_new()) ? PURE_OK : PURE_ALLOCATION;
+        if (result == PURE_OK && image) result = pure_list_read(*out, bytes + length + 8, size - length - 8);
+    }
+    free(bytes);
+    /* Then the changes logged after the save it holds (an image is written
+     * before any change of the list is logged). */
+    struct Named *logged = app->start_log ? named_list(app->start_log, heard_name, strlen(heard_name), false) : NULL;
+    for (size_t k = 0; result == PURE_OK && logged && k < logged->count; ++k) {
+        const Logged *change = &logged->changes[k];
+        if (change->seq <= holds) continue;
+        if (!image) result = PURE_STORE_ERROR;
+        else if ((result = pure_list_replay(*out, change->records.data, change->records.size)) == PURE_OK) {
+            if (!saved->since) saved->since = change->seq;
+            saved->logged += change->records.size;
+        }
+    }
+    if (result != PURE_OK) {
+        pure_list_free(*out);
+        *out = NULL;
+        return result;
+    }
+    if (image) saved->image = holds;
+    else saved->rewrite = true;
+    return PURE_OK;
+}
+
+PureResult pure_save_heard_image(PureApp *app) {
+    unsigned char header[IMAGE_HEADER];
+    size_t header_size = header_of(2, heard_text, heard_names, app->seq, header);
+    return write_image(directory_of(app), heard_name, header, header_size, app->heard);
+}
+
 /* ---- Checkpoints */
 
 struct PureCheckpoint {
@@ -425,32 +493,33 @@ PureCheckpoint *pure_app_checkpoint(PureApp *app) {
     const PureModule *m = app->module;
     if (!app->log || app->checkpoint || app->open || app->uncertain || app->reading) return NULL;
     /* A list whose logged changes outgrew its items, or, when the log grew
-     * past LOG_KEPT, the one that holds it back most. */
+     * past LOG_KEPT, the one that holds it back most: the heard list is one. */
     int due = -1, oldest = -1;
-    for (uint32_t i = 0; i < m->states && due < 0; ++i) {
-        if (m->state_type[i] != PURE_STATE_LIST || !app->logged_since[i]) continue;
-        size_t items = pure_list_size(app->lists[i]);
-        if (app->logged[i] > (items > JOURNAL_MIN ? items : JOURNAL_MIN)) due = (int)i;
+    uint64_t items = 0;
+    for (uint32_t i = 0; i <= PURE_LIST_HEARD && due < 0; ++i) {
+        const PureList *list = stored_list(app, i);
+        size_t size = list ? pure_list_size(list) : 0;
+        items += size;
+        if (!list || !app->logged_since[i]) continue;
+        if (app->logged[i] > (size > JOURNAL_MIN ? size : JOURNAL_MIN)) due = (int)i;
         else if (oldest < 0 || app->logged_since[i] < app->logged_since[oldest]) oldest = (int)i;
     }
-    uint64_t items = 0;
-    for (uint32_t i = 0; due < 0 && i < m->states; ++i)
-        if (m->state_type[i] == PURE_STATE_LIST && app->lists[i]) items += pure_list_size(app->lists[i]);
     int pick = due >= 0 ? due : pure_log_bytes(app->log) > (items > LOG_KEPT ? items : LOG_KEPT) ? oldest : -1;
     if (pick < 0) return NULL;
     PureCheckpoint *c = calloc(1, sizeof(*c));
     const char *directory = directory_of(app);
     size_t length = strlen(directory) + 1;
     if (c && (c->directory = malloc(length))) memcpy(c->directory, directory, length);
-    if (!c || !c->directory || !(c->snapshot = pure_app_snapshot(app))) {
+    if (!c || !c->directory || !(c->snapshot = pure_save_snapshot(app, pick == PURE_LIST_HEARD))) {
         if (c) free(c->directory);
         free(c);
         return NULL;
     }
     c->state = (uint32_t)pick;
-    snprintf(c->name, sizeof(c->name), "%s", m->state_names[pick]);
+    snprintf(c->name, sizeof(c->name), "%s", stored_name(app, (uint32_t)pick));
     c->seq = app->seq;
-    c->header_size = image_header(m, m->state_ref[pick], c->seq, c->header);
+    c->header_size = pick == PURE_LIST_HEARD ? header_of(2, heard_text, heard_names, c->seq, c->header) :
+                                               image_header(m, m->state_ref[pick], c->seq, c->header);
     c->logged = app->logged[pick];
     c->since = app->logged_since[pick];
     app->checkpoint = c;
@@ -563,14 +632,13 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
     const PureModule *m = app->module;
     unsigned outputs = 0;
     uint8_t table[PURE_MODULE_INPUTS];
-    size_t total = 4 + 1 + 2;
+    size_t total = 4 + 1;
     for (uint32_t i = 0; i < m->inputs; ++i)
         if (m->input[i].kind == PURE_KIND_OUTPUT) {
             table[i] = (uint8_t)outputs++;
             total += 2 + strlen(m->input_names[i]) + m->input[i].field_count + 8;
         }
     for (unsigned c = 0; c < app->carried_count; ++c) total += 2 + strlen(app->carried[c].name) + 8;
-    for (unsigned h = 0; h < app->heard_count; ++h) total += 1 + strlen(app->heard[h].source) + 8;
     for (size_t s = 0; s < app->sent_count; ++s)
         if (!app->sent[s].trimmed) total += 1 + 8 + 4 + app->sent[s].size;
     unsigned char *bytes = malloc(total), *at = bytes;
@@ -596,13 +664,6 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
         *at++ = 0;
         at = put_le(at, app->carried[c].last, 8);
     }
-    at = put_le(at, app->heard_count, 2);
-    for (unsigned h = 0; h < app->heard_count; ++h) {
-        size_t length = strlen(app->heard[h].source);
-        *at++ = (unsigned char)length;
-        memcpy(at, app->heard[h].source, length);
-        at = put_le(at + length, app->heard[h].seq, 8);
-    }
     for (size_t s = 0; s < app->sent_count; ++s) {
         const struct PureSent *sent = &app->sent[s];
         if (sent->trimmed) continue;
@@ -616,11 +677,22 @@ static unsigned char *encode_links(const PureApp *app, size_t *size) {
     return bytes;
 }
 
+/* A number heard, as links kept it before the heard list. */
+PURE_COLD static PureResult heard_before(PureApp *app, const unsigned char *source, size_t length, uint64_t seq) {
+    PureCell key = {.text = (unsigned char *)source, .size = length};
+    bool found;
+    size_t at = pure_list_seek(app->heard, 0, key, &found);
+    if (!found) return pure_list_insert(app->heard, at, (PureCell[2]){key, {.num = (double)seq}}, NULL);
+    if (pure_list_get(app->heard, at, 1).num >= (double)seq) return PURE_OK;
+    return pure_list_set(app->heard, at, 1, (PureCell){.num = (double)seq}, NULL);
+}
+
 /* The stored links of a starting app: outputs by name (an output whose
  * fields changed or that the module lacks refuses the start while it has
  * sends left: CONFLICT; one without sends keeps its number for later),
- * sources and sent messages. */
-PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
+ * sources kept before the heard list, and sent messages. */
+PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m, bool *moved) {
+    *moved = false;
     PureResult result = app->stored_links ? PURE_OK : open_state(app, "", "links", &app->stored_links);
     if (result != PURE_OK || !pure_draft_revision(app->stored_links)) return result;
     unsigned char *bytes;
@@ -629,8 +701,9 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
     int output[256];
     bool fits[256];
     size_t at = 5;
-    unsigned outputs = size >= 5 && !memcmp(bytes, links_magic, 4) ? bytes[4] : 0;
-    result = size >= 5 && !memcmp(bytes, links_magic, 4) ? PURE_OK : PURE_STORE_ERROR;
+    bool first = size >= 5 && !memcmp(bytes, links_magic_1, 4), known = first || (size >= 5 && !memcmp(bytes, links_magic, 4));
+    unsigned outputs = known ? bytes[4] : 0;
+    result = known ? PURE_OK : PURE_STORE_ERROR;
     for (unsigned t = 0; t < outputs && result == PURE_OK; ++t) {
         char name[PURE_MODULE_NAME];
         size_t length = at < size ? bytes[at] : 0;
@@ -655,18 +728,18 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
             grown[app->carried_count++].last = last;
         }
     }
-    unsigned sources = result == PURE_OK && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
-    if (result == PURE_OK && size - at < 2) result = PURE_STORE_ERROR;
-    at += 2;
-    if (result == PURE_OK && sources && !(app->heard = calloc(sources, sizeof(*app->heard)))) result = PURE_ALLOCATION;
-    app->heard_capacity = sources;
+    unsigned sources = result == PURE_OK && first && size - at >= 2 ? (unsigned)get_le(bytes + at, 2) : 0;
+    if (result == PURE_OK && first && size - at < 2) result = PURE_STORE_ERROR;
+    at += first ? 2 : 0;
     for (unsigned h = 0; h < sources && result == PURE_OK; ++h) {
         size_t length = at < size ? bytes[at] : 0;
         if (!length || length > PURE_SOURCE || size - at < 1 + length + 8) { result = PURE_STORE_ERROR; break; }
-        struct PureHeard *heard = &app->heard[app->heard_count++];
-        memcpy(heard->source, bytes + at + 1, length);
-        heard->seq = heard->saved = get_le(bytes + at + 1 + length, 8);
+        uint64_t seq = get_le(bytes + at + 1 + length, 8);
+        if (seq > PURE_SEQ_MOST) { result = PURE_STORE_ERROR; break; }
+        result = heard_before(app, bytes + at + 1, length, seq);
         at += 1 + length + 8;
+        *moved = true;
+        app->links_changed = true; /* stored again without them */
     }
     while (result == PURE_OK && at < size) {
         if (size - at < 13 || bytes[at] >= outputs) { result = PURE_STORE_ERROR; break; }
@@ -698,18 +771,19 @@ PURE_COLD PureResult pure_save_open_links(PureApp *app, const PureModule *m) {
 /* The lists' changes since the last save, as one log entry: per list changed,
  * its name and its records. NULL with size 0 when no list changed. */
 static unsigned char *list_changes(const PureApp *app, size_t *size) {
-    const PureModule *m = app->module;
     *size = 0;
-    for (uint32_t i = 0; i < m->states; ++i)
-        if (m->state_type[i] == PURE_STATE_LIST && app->lists[i] && pure_list_changes(app->lists[i]).size)
-            *size += 1 + strlen(m->state_names[i]) + 4 + pure_list_changes(app->lists[i]).size;
+    for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i) {
+        const PureList *list = stored_list(app, i);
+        if (list && pure_list_changes(list).size) *size += 1 + strlen(stored_name(app, i)) + 4 + pure_list_changes(list).size;
+    }
     unsigned char *payload = *size ? malloc(*size) : NULL, *at = payload;
-    for (uint32_t i = 0; payload && i < m->states; ++i) {
-        PureSpan changes = m->state_type[i] == PURE_STATE_LIST && app->lists[i] ? pure_list_changes(app->lists[i]) : (PureSpan){0};
+    for (uint32_t i = 0; payload && i <= PURE_LIST_HEARD; ++i) {
+        const PureList *list = stored_list(app, i);
+        PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0};
         if (!changes.size) continue;
-        size_t length = strlen(m->state_names[i]);
+        size_t length = strlen(stored_name(app, i));
         *at++ = (unsigned char)length;
-        memcpy(at, m->state_names[i], length);
+        memcpy(at, stored_name(app, i), length);
         at = put_le(at + length, changes.size, 4);
         memcpy(at, changes.data, changes.size);
         at += changes.size;
@@ -814,12 +888,13 @@ PureResult pure_save_commit(PureApp *app, const double *values, void *const *doc
     if (result == PURE_OK) {
         for (size_t k = 0; k < count; ++k)
             if (states[k] != UINT32_MAX) app->saved[states[k]] = pure_draft_version(drafts[k]);
-        for (uint32_t i = 0; i < m->states; ++i) {
-            PureSpan changes = m->state_type[i] == PURE_STATE_LIST && app->lists[i] ? pure_list_changes(app->lists[i]) : (PureSpan){0};
+        for (uint32_t i = 0; i <= PURE_LIST_HEARD; ++i) {
+            PureList *list = stored_list(app, i);
+            PureSpan changes = list ? pure_list_changes(list) : (PureSpan){0};
             if (!changes.size) continue;
             app->logged[i] += changes.size;
             if (!app->logged_since[i]) app->logged_since[i] = seq;
-            pure_list_forget(app->lists[i], 0);
+            pure_list_forget(list, 0);
         }
         if (tree || payload_size) app->seq = seq;
     }
diff --git a/core/module-save.h b/core/module-save.h
index 819f948..32e68c7 100644
--- a/core/module-save.h
+++ b/core/module-save.h
@@ -24,16 +24,27 @@ PureResult pure_save_image(PureApp *, const PureModule *m, uint32_t state, const
 void pure_save_close_start(PureApp *);
 /* The log keeps only the entries some list's image does not hold yet. */
 void pure_save_release(PureApp *);
-/* A snapshot's list (module.c). */
+/* A snapshot's list (module.c): a state's, or the heard list's
+ * (PURE_LIST_HEARD) in a snapshot made with it (pure_save_snapshot). */
 const PureList *pure_snapshot_list(const PureSnapshot *, uint32_t state);
+PureSnapshot *pure_save_snapshot(PureApp *, bool heard);
+/* The heard list of a starting app (module.h, PureApp): from its image with
+ * its logged changes made again, or new (saved.rewrite: it needs an image,
+ * which pure_save_heard_image writes). */
+PureResult pure_save_open_heard(PureApp *, PureList **out, PureSaved *saved);
+PureResult pure_save_heard_image(PureApp *);
+/* An empty heard list. */
+PureList *pure_heard_new(void);
 /* Open the stored numbers, cursors and indices, and put each stored one of the
  * same name and kind that still fits into values, for the states of m that
  * kept marks as new (-1); texts and lists are m's. */
 PureResult pure_save_open_values(PureApp *, const PureModule *m, double *values, const int *kept,
                                  PureDraft *const *texts, PureList *const *lists);
 /* The stored links of a starting app (module.h, PureApp); CONFLICT when an
- * output with sends left is missing or has other fields. */
-PureResult pure_save_open_links(PureApp *, const PureModule *m);
+ * output with sends left is missing or has other fields. Links stored before
+ * the heard list kept the numbers heard: they go to it (moved: it needs an
+ * image). */
+PureResult pure_save_open_links(PureApp *, const PureModule *m, bool *moved);
 /* 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,
diff --git a/core/module.c b/core/module.c
index 45f67f7..3a694a1 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2442,6 +2442,7 @@ static void open_burst(PureApp *app) {
         }
         if (app->lists[i]) app->durable_lists[i] = pure_list_changes(app->lists[i]).size;
     }
+    if (app->heard) app->durable_heard = pure_list_changes(app->heard).size;
     app->open = true;
 }
 
@@ -2462,6 +2463,7 @@ static void close_burst(PureApp *app, bool keep) {
         }
         if (app->lists[i] && !keep) pure_list_forget(app->lists[i], app->durable_lists[i]);
     }
+    if (app->heard && !keep) pure_list_forget(app->heard, app->durable_heard);
     if (keep) pure_undo_commit(&app->undo);
     else {
         pure_undo_rollback(&app->undo, 0);
@@ -2478,7 +2480,8 @@ static void drop_sent(PureApp *app, size_t count) {
 }
 
 /* After a save: what it stored is the links' saved state (trimmed sends go),
- * or, when it failed, the links return to the last saved state. */
+ * or, when it failed, the links return to the last saved state (the numbers
+ * heard go back with the burst's list changes). */
 static void settle_links(PureApp *app, bool saved) {
     if (saved) {
         size_t kept = 0;
@@ -2488,18 +2491,10 @@ static void settle_links(PureApp *app, bool saved) {
         }
         app->sent_count = app->sent_durable = kept;
         memcpy(app->saved_sent, app->last_sent, sizeof(app->saved_sent));
-        for (unsigned i = 0; i < app->heard_count; ++i) app->heard[i].saved = app->heard[i].seq;
     } else {
         drop_sent(app, app->sent_durable);
         for (size_t i = 0; i < app->sent_count; ++i) app->sent[i].trimmed = false;
         memcpy(app->last_sent, app->saved_sent, sizeof(app->last_sent));
-        unsigned kept = 0;
-        for (unsigned i = 0; i < app->heard_count; ++i)
-            if (app->heard[i].saved) {
-                app->heard[kept] = app->heard[i];
-                app->heard[kept++].seq = app->heard[i].saved;
-            }
-        app->heard_count = kept;
     }
     app->links_changed = false;
 }
@@ -2536,13 +2531,16 @@ struct PureSnapshot {
     double values[PURE_MODULE_STATES];
     PureDraft *texts[PURE_MODULE_STATES];
     PureList *lists[PURE_MODULE_STATES];
+    PureList *heard; /* only in a checkpoint's of the heard list */
     PureRetired retired;
     struct PureSnapshot *newer;
 };
 
 void pure_snapshot_hold(PureSnapshot *snapshot) { __atomic_add_fetch(&snapshot->refs, 1, __ATOMIC_RELAXED); }
 
-const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state) { return snapshot->lists[state]; }
+const PureList *pure_snapshot_list(const PureSnapshot *snapshot, uint32_t state) {
+    return state == PURE_LIST_HEARD ? snapshot->heard : snapshot->lists[state];
+}
 
 void pure_snapshot_release(PureSnapshot *snapshot) {
     while (snapshot && !__atomic_sub_fetch(&snapshot->refs, 1, __ATOMIC_ACQ_REL)) {
@@ -2551,6 +2549,7 @@ void pure_snapshot_release(PureSnapshot *snapshot) {
             pure_draft_close(snapshot->texts[i]);
             pure_list_free(snapshot->lists[i]);
         }
+        pure_list_free(snapshot->heard);
         pure_retired_free(&snapshot->retired);
         free(snapshot);
         snapshot = newer;
@@ -2574,7 +2573,7 @@ void pure_app_stop(PureApp *app) {
     pure_draft_close(app->stored_values);
     drop_sent(app, 0);
     free(app->sent);
-    free(app->heard);
+    pure_list_free(app->heard);
     free(app->carried);
     pure_draft_close(app->stored_links);
     pure_draft_close(app->stored_seq);
@@ -2593,7 +2592,7 @@ void pure_app_stop(PureApp *app) {
 /* Where one event began inside the burst: a failed event returns there. */
 typedef struct {
     PureVersion texts[PURE_MODULE_STATES];
-    size_t lists[PURE_MODULE_STATES]; /* stored lists' journal lengths */
+    size_t lists[PURE_MODULE_STATES], heard; /* stored lists' journal lengths */
     size_t undo, sent;
     uint64_t last_sent[PURE_MODULE_INPUTS];
     bool opened;
@@ -2606,6 +2605,7 @@ static void mark_event(PureApp *app, Mark *mark) {
         if (app->texts[i]) mark->texts[i] = pure_draft_version(app->texts[i]);
         if (app->lists[i]) mark->lists[i] = pure_list_changes(app->lists[i]).size;
     }
+    mark->heard = app->heard ? pure_list_changes(app->heard).size : 0;
     mark->undo = app->undo.count;
     mark->sent = app->sent_count;
     memcpy(mark->last_sent, app->last_sent, sizeof(mark->last_sent));
@@ -2616,29 +2616,32 @@ static void undo_event(PureApp *app, const Mark *mark) {
         if (app->texts[i]) pure_draft_restore(app->texts[i], mark->texts[i]);
         if (app->lists[i]) pure_list_forget(app->lists[i], mark->lists[i]);
     }
+    if (app->heard) pure_list_forget(app->heard, mark->heard);
     pure_undo_rollback(&app->undo, mark->undo);
     drop_sent(app, mark->sent);
     memcpy(app->last_sent, mark->last_sent, sizeof(app->last_sent));
     if (mark->opened) close_burst(app, true); /* nothing happened: release the holds */
 }
 
-/* The last number heard from source, or NULL. */
-static struct PureHeard *heard_from(PureApp *app, const char *source) {
-    for (unsigned i = 0; i < app->heard_count; ++i)
-        if (!strcmp(app->heard[i].source, source)) return &app->heard[i];
-    return NULL;
+/* Where source's item is in the heard list, or would go; found says which. */
+static size_t heard_at(const PureApp *app, const char *source, bool *found) {
+    *found = false;
+    return app->heard ? pure_list_seek(app->heard, 0, (PureCell){.text = (unsigned char *)source, .size = strlen(source)}, found) : 0;
 }
 
-/* Hear seq from source (room for a new source was made before). */
-static void hear(PureApp *app, const char *source, uint64_t seq) {
-    struct PureHeard *heard = heard_from(app, source);
-    if (!heard) {
-        heard = &app->heard[app->heard_count++];
-        *heard = (struct PureHeard){.saved = 0};
-        memcpy(heard->source, source, strlen(source) + 1);
-    }
-    heard->seq = seq;
-    app->links_changed = true;
+uint64_t pure_app_heard(const PureApp *app, const char *source) {
+    bool found;
+    size_t at = heard_at(app, source, &found);
+    return found ? (uint64_t)pure_list_get(app->heard, at, 1).num : 0;
+}
+
+/* Hear seq from source: a change of the heard list, undone with the burst. */
+static PureResult hear(PureApp *app, const char *source, uint64_t seq) {
+    bool found;
+    size_t at = heard_at(app, source, &found);
+    if (found) return pure_list_set(app->heard, at, 1, (PureCell){.num = (double)seq}, &app->undo);
+    PureCell item[2] = {{.text = (unsigned char *)source, .size = strlen(source)}, {.num = (double)seq}};
+    return pure_list_insert(app->heard, at, item, &app->undo);
 }
 
 static PureResult keep_reply(PureApp *app);
@@ -2673,6 +2676,8 @@ 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. */
+    if (result == PURE_OK && app->delivering.pending) result = hear(app, app->delivering.source, app->delivering.seq);
     if (result != PURE_OK) {
         undo_event(app, &mark);
         clear_rows(app);
@@ -2691,10 +2696,7 @@ static PureResult dispatch(PureApp *app, uint32_t event, Event *ev) {
         app->document_pending = true;
         app->revision += ev->modified; /* a save moves an edited document one revision */
     }
-    if (app->delivering.pending) {
-        hear(app, app->delivering.source, app->delivering.seq);
-        app->delivering.pending = false;
-    }
+    app->delivering.pending = false;
     return event && !app->deferred ? pure_app_save(app) : PURE_OK;
 }
 
@@ -2889,23 +2891,17 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     const PureModule *m = app->module;
     size_t length = source ? strlen(source) : 0;
     app->duplicate = false;
+    /* Kept states without a directory would not keep the numbers heard. */
     if (input >= m->inputs || m->input[input].kind != PURE_KIND_MESSAGE || m->input[input].keyed || !length ||
-        length > PURE_SOURCE || !seq) return PURE_INVALID;
+        length > PURE_SOURCE || !seq || seq > PURE_SEQ_MOST || !app->heard || (pure_save_enabled(app) && !app->log))
+        return PURE_INVALID;
     for (size_t i = 0; i < length; ++i)
         if (source[i] <= ' ' || source[i] > '~') return PURE_INVALID;
-    struct PureHeard *heard = heard_from(app, source);
-    if (heard && seq <= heard->seq) {
+    if (seq <= pure_app_heard(app, source)) {
         pure_app_finish(app);
         app->duplicate = true;
         return PURE_OK;
     }
-    if (!heard && app->heard_count == app->heard_capacity) {
-        unsigned capacity = app->heard_capacity ? 2 * app->heard_capacity : 4;
-        struct PureHeard *grown = realloc(app->heard, capacity * sizeof(*grown));
-        if (!grown) return PURE_ALLOCATION;
-        app->heard = grown;
-        app->heard_capacity = capacity;
-    }
     app->delivering.source = source;
     app->delivering.seq = seq;
     app->delivering.pending = true;
@@ -2917,17 +2913,15 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     if (app->fault == PURE_FAULT_NONE) return result;
     Mark mark;
     mark_event(app, &mark);
-    hear(app, source, seq);
+    PureResult heard = hear(app, source, seq);
+    if (heard != PURE_OK) {
+        undo_event(app, &mark);
+        return heard;
+    }
     PureResult saved = app->deferred ? PURE_OK : pure_app_save(app);
     return saved == PURE_OK ? result : saved;
 }
 
-uint64_t pure_app_heard(const PureApp *app, const char *source) {
-    for (unsigned i = 0; i < app->heard_count; ++i)
-        if (!strcmp(app->heard[i].source, source)) return app->heard[i].seq;
-    return 0;
-}
-
 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);
@@ -3099,7 +3093,9 @@ PureResult pure_app_cell_text(PureApp *app, uint32_t column, size_t row, bool (*
     return result == PURE_OK ? visit_value(app, &app->view_arena, slot.bytes, visit, context) : result;
 }
 
-PureSnapshot *pure_app_snapshot(PureApp *app) {
+PureSnapshot *pure_app_snapshot(PureApp *app) { return pure_save_snapshot(app, false); }
+
+PureSnapshot *pure_save_snapshot(PureApp *app, bool heard) {
     if (app->open || app->reading) return NULL;
     const PureModule *m = app->module;
     PureSnapshot *made = calloc(1, sizeof(*made)), *last = app->snapshot;
@@ -3118,6 +3114,7 @@ PureSnapshot *pure_app_snapshot(PureApp *app) {
         } else
             ok = (made->lists[i] = pure_list_share(app->lists[i], &app->retired)) != NULL;
     }
+    if (ok && heard) ok = (made->heard = pure_list_share(app->heard, &app->retired)) != NULL;
     if (!ok) {
         pure_snapshot_release(made);
         return NULL;
@@ -3256,6 +3253,11 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
     bool lists_stored = stored && app->data.directory && app->data.directory(app->data.context);
     if (old && app->checkpoint) return PURE_CONFLICT; /* the host has a checkpoint of the running module out */
     PureResult result = !old && lists_stored ? pure_save_open_log(app) : PURE_OK;
+    /* The numbers heard belong to the app, whatever its module. */
+    PureSaved heard = {0};
+    if (!old && result == PURE_OK)
+        result = lists_stored ? pure_save_open_heard(app, &app->heard, &heard) :
+                 (app->heard = pure_heard_new()) ? PURE_OK : PURE_ALLOCATION;
     for (uint32_t i = 0; i < module->states; ++i) {
         int previous = old ? pure_module_state(old, module->state_names[i]) : -1;
         kept[i] = previous >= 0 && old->state_type[previous] == module->state_type[i] ? previous : -1;
@@ -3310,7 +3312,9 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
     }
     if (result == PURE_OK && stored) result = pure_save_open_values(app, module, values, kept, texts, lists);
     /* Outputs keep their sends and numbers by name. */
-    if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module) : PURE_OK;
+    bool heard_moved = false;
+    if (result == PURE_OK) result = old ? remap_outputs(app, old, module, false) : stored ? pure_save_open_links(app, module, &heard_moved) : PURE_OK;
+    heard.rewrite = heard.rewrite || (heard_moved && lists_stored);
     for (uint32_t i = 0; i < module->states; ++i) {
         uint32_t list = module->state_ref[i];
         if (module->state_type[i] == PURE_STATE_INDEX && moved[list] && values[i] < (double)pure_list_count(lists[list]))
@@ -3332,6 +3336,8 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
     for (uint32_t i = 0; i < module->states && result == PURE_OK; ++i)
         if (rewrite[i] && (result = pure_save_image(app, module, i, lists[i])) == PURE_OK)
             listed[i] = (PureSaved){.image = app->seq};
+    if (result == PURE_OK && heard.rewrite && (result = pure_save_heard_image(app)) == PURE_OK)
+        heard = (PureSaved){.image = app->seq};
     if (imaged && result != PURE_OK && old) app->uncertain = true;
     if (result != PURE_OK) {
         for (uint32_t i = 0; i < module->states; ++i) {
@@ -3367,6 +3373,12 @@ PURE_COLD static PureResult open_module(PureApp *app, const PureModule *module)
         app->logged_since[i] = listed[i].since;
         app->logged[i] = listed[i].logged;
     }
+    if (!old) {
+        app->image_seq[PURE_LIST_HEARD] = heard.image;
+        app->logged_since[PURE_LIST_HEARD] = heard.since;
+        app->logged[PURE_LIST_HEARD] = heard.logged;
+        if (lists_stored) pure_list_journal(app->heard, true);
+    }
     pure_save_release(app);
     for (uint32_t i = 0; i < module->states; ++i) {
         /* A stored list's changes are recorded from now on. */
diff --git a/core/module.h b/core/module.h
index e9e5dd3..8176bea 100644
--- a/core/module.h
+++ b/core/module.h
@@ -80,6 +80,8 @@ typedef enum { PURE_LESS, PURE_LESS_EQUAL, PURE_GREATER, PURE_GREATER_EQUAL, PUR
 typedef enum { PURE_KIND_KEY = 1, PURE_KIND_MESSAGE = 2, PURE_KIND_TEXT = 3, PURE_KIND_TIMER = 4,
                PURE_KIND_OUTPUT = 5, PURE_KIND_ARGUMENTS = 6 } PureInputKind;
 enum { PURE_TIMER_LONGEST = 86400000, PURE_SOURCE = 64 };
+/* The largest number a delivery may have: it is kept as a binary64. */
+#define PURE_SEQ_MOST (UINT64_C(1) << 53)
 /* A message field: a u32le or a binary64 (fixed, at its offset in the
  * payload's prefix), or bytes, text or rows of any length. The prefix also
  * holds each variable field's u32le length but the last one's (its offset is
@@ -261,6 +263,8 @@ typedef struct { unsigned char *bytes; size_t used, capacity; } PureArena;
  * 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, or one past PURE_APP_PENDING keys, saves the burst first). */
+/* The heard list's place among the stored lists' records (PureApp). */
+enum { PURE_LIST_HEARD = PURE_MODULE_STATES };
 typedef struct {
     const PureModule *module;
     PureData data;
@@ -269,12 +273,13 @@ typedef struct {
     PureList *lists[PURE_MODULE_STATES];
     PureUndo undo; /* list changes of the current event */
     PureDraft *list_drafts[PURE_MODULE_STATES]; /* lists the tree kept before images, until emptied */
-    /* Stored lists (module-save.c): the save each one's image holds, the
-     * first save logged since that changed it (0: none), and the bytes of its
-     * changes logged since; the log, the tree's number of its last save, the
-     * last save's number, and a checkpoint being written. */
-    uint64_t image_seq[PURE_MODULE_STATES], logged_since[PURE_MODULE_STATES];
-    size_t logged[PURE_MODULE_STATES];
+    /* Stored lists (module-save.c), by state, and the heard list's last
+     * (PURE_LIST_HEARD): the save each one's image holds, the first save
+     * logged since that changed it (0: none), and the bytes of its changes
+     * logged since; the log, the tree's number of its last save, the last
+     * save's number, and a checkpoint being written. */
+    uint64_t image_seq[PURE_MODULE_STATES + 1], logged_since[PURE_MODULE_STATES + 1];
+    size_t logged[PURE_MODULE_STATES + 1];
     struct PureLog *log;
     PureDraft *stored_seq;
     uint64_t seq;
@@ -318,15 +323,18 @@ typedef struct {
     /* Links (runtime-owned, saved with the states): messages sent on outputs
      * and not trimmed, in the order sent (the first sent_durable are saved;
      * a trimmed one goes at the next save), each output's last number (and
-     * as saved), those of outputs a module dropped, and the last number
-     * delivered from each source (and as saved). */
+     * as saved), and those of outputs a module dropped. */
     struct PureSent { uint32_t output; uint64_t seq; unsigned char *payload; size_t size; bool trimmed; } *sent;
     size_t sent_count, sent_capacity, sent_durable;
     uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS];
     struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried;
     unsigned carried_count;
-    struct PureHeard { char source[PURE_SOURCE + 1]; uint64_t seq, saved; } *heard;
-    unsigned heard_count, heard_capacity;
+    /* The last number delivered from each source: a list of {source (text),
+     * number}, in order of source, stored as a list state is (module-save.c)
+     * and changed with the events that deliver, so a failed save takes its
+     * numbers back with them; its journal's length when the burst began. */
+    PureList *heard;
+    size_t durable_heard;
     PureDraft *stored_links;
     bool links_changed; /* since the last save */
     bool duplicate;     /* the last delivery was heard before and not applied */
@@ -421,7 +429,9 @@ void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context)
 /* Forget output's saved sends numbered up to through (saved with the next save). */
 void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
 /* Deliver a message event from source (a sender's name, 1 to 64 printable
- * ASCII bytes) numbered seq (from 1, growing per source). A number heard
+ * ASCII bytes) numbered seq (from 1 to PURE_SEQ_MOST, growing per source).
+ * An app whose states are kept takes deliveries only where its lists are
+ * (a directory: the numbers heard are a list). A number heard
  * 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
diff --git a/core/test-module.c b/core/test-module.c
index b2024b5..9a1f8a4 100644
--- a/core/test-module.c
+++ b/core/test-module.c
@@ -727,6 +727,15 @@ static const char *state_text(const PureApp *app, uint32_t state) {
     return t.text;
 }
 
+/* A stored value's first four bytes (or fewer). */
+static const char *read_state(const PureDraft *draft) {
+    static Text t;
+    t.size = 0;
+    size_t size = pure_draft_size(draft) < 4 ? pure_draft_size(draft) : 4;
+    CHECK(pure_draft_read(draft, 0, size, text_span, &t) == PURE_OK);
+    return t.text;
+}
+
 static double number_view(PureApp *app, uint32_t view) {
     double value;
     CHECK(pure_app_view(app, view, &value));
@@ -3918,6 +3927,56 @@ static void links(void) {
     CHECK(deliver(&b, "orders.ordered", 2, "ink", 3) == PURE_OK && !b.duplicate && b.values[0] == 5);
     CHECK(pure_app_save(&b) == PURE_OK);
     pure_app_stop(&b);
+    /* Links stored before the heard list kept its numbers: a start moves them
+     * there (the larger number of a source wins) and stores the links without. */
+    PureDraft *stored_links;
+    int64_t revision;
+    unsigned char old_links[64] = "PLK1";
+    size_t at = 5;
+    old_links[4] = 0; /* no outputs */
+    old_links[at++] = 2, old_links[at++] = 0;
+    const char *old_sources[] = {"legacy", "orders.ordered"};
+    uint64_t old_numbers[] = {5, 1};
+    for (unsigned k = 0; k < 2; ++k) {
+        old_links[at++] = (unsigned char)strlen(old_sources[k]);
+        memcpy(old_links + at, old_sources[k], strlen(old_sources[k]));
+        at += strlen(old_sources[k]);
+        for (unsigned i = 0; i < 8; ++i) old_links[at++] = (unsigned char)(old_numbers[k] >> (8 * i));
+    }
+    CHECK(pure_state_open(path_b, "links", &stored_links) == PURE_OK);
+    CHECK(pure_draft_splice(stored_links, 0, pure_draft_size(stored_links), (PureSpan){old_links, at}) == PURE_OK);
+    CHECK(pure_draft_commit(stored_links, &revision) == PURE_OK);
+    pure_draft_close(stored_links);
+    CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
+    CHECK(pure_app_heard(&b, "legacy") == 5 && pure_app_heard(&b, "orders.ordered") == 2 && pure_app_heard(&b, "x") == 1);
+    CHECK(deliver(&b, "legacy", 5, "old", 1) == PURE_OK && b.duplicate && deliver(&b, "legacy", 6, "new", 1) == PURE_OK && !b.duplicate);
+    pure_app_stop(&b);
+    CHECK(pure_state_open(path_b, "links", &stored_links) == PURE_OK && !memcmp(read_state(stored_links), "PLK2", 4));
+    pure_draft_close(stored_links);
+    /* Numbers heard from more sources than two bytes count stay heard across
+     * a start; a save logs only the numbers it changed, and the heard list's
+     * image is written again once they outgrow it (from a snapshot: through a
+     * store handle). */
+    PureStore *store;
+    CHECK(pure_store_open(path_b, &store) == PURE_OK);
+    CHECK(pure_app_start(&b, &module_b, pure_store_data(store)) == PURE_OK && pure_app_heard(&b, "legacy") == 6);
+    unsigned long checkpoints = pure_save_test_checkpoints;
+    for (unsigned k = 0; k < 400; ++k) CHECK(deliver(&b, "s1", 1 + k, "more", 1) == PURE_OK);
+    CHECK(pure_save_test_checkpoints > checkpoints && pure_app_heard(&b, "s1") == 400);
+    CHECK(pure_app_defer(&b, true));
+    char many[16];
+    for (unsigned k = 0; k < 70000; ++k) {
+        snprintf(many, sizeof(many), "s%u", k);
+        CHECK(deliver(&b, many, 400 + k % 3, "one", 1) == PURE_OK && b.duplicate == (k == 1 && k % 3 == 0));
+        if (k % 5000 == 4999) CHECK(pure_app_save(&b) == PURE_OK);
+    }
+    CHECK(pure_app_save(&b) == PURE_OK);
+    pure_app_stop(&b);
+    CHECK(pure_app_start(&b, &module_b, pure_store_data(store)) == PURE_OK && pure_list_count(b.heard) == 70003);
+    CHECK(pure_app_heard(&b, "s0") == 400 && pure_app_heard(&b, "s1") == 401 && pure_app_heard(&b, "s65537") == 402);
+    CHECK(deliver(&b, "s65537", 402, "again", 1) == PURE_OK && b.duplicate);
+    pure_app_stop(&b);
+    pure_store_close(store);
 
     /* Replacing: a module without the output is refused while sends are left
      * on it; once they are trimmed it replaces, and the output's number is
@@ -3970,7 +4029,7 @@ static void links(void) {
         CHECK(pure_app_start(&b, &module_b, pure_tree_data(path_b)) == PURE_OK);
         o = outbox_of(&a);
         CHECK(o.count == (size_t)a.values[0] && (!o.count || o.items[o.count - 1].seq == o.count));
-        heard = b.heard_count ? b.heard[0].seq : 0;
+        heard = pure_app_heard(&b, "orders.ordered");
         CHECK(b.values[0] == (double)heard);
         kept[0] = (size_t)heard;
         kept[1] = o.count;