The modules an app ran are freed (research 115's next step): every app's module a Loaded with holds (each snapshot of it, each request read with it); one replaced is retired and freed on the apps thread once no app runs it, every connection thread has begun a pass since (woken for it), and nothing holds it; who keeps a module to ask later whether it is the same asks its serial (core: PureModule.serial, a reader's module_serial, pure_snapshot_module, pure_snapshot_release says it freed; serve: routes, groups, links, bindings, queries, the published snapshot), as an address may come back; a link binds under its receiver's lock, a link naming an input under binding_lock; 030's memory.py: 4,000 changes 0 bytes a change (was 19,284), 4,100 freed; stress.py: readers, view workers, push streams and a body across changes, 586 of 600 loads at a freed module's address, every answer right, ASan 1,000 changes clean; a push group that looked while its app resumed missed the new snapshot (published before the app took requests) and waited for its next save: woken again once it takes them (server_view_workers had failed once under load)

1 files changed, 159 insertions, 52 deletions

diff --git a/serve/serve.c b/serve/serve.c
index 1f3a326..7dd81e4 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -84,6 +84,7 @@ typedef struct {
     const char *given;
     App *app;
     const PureModule *module;
+    uint64_t serial; /* the module's: compared, not the address (one freed, another may come to it) */
 } Route;
 
 /* An app's tokens (docs/design/017's 3b): what each grants, by its token's
@@ -242,7 +243,7 @@ typedef struct {
      * itself), and the app's changes and module the last one shows. */
     Published *published;
     uint64_t published_changes, publications;
-    const PureModule *published_module;
+    uint64_t published_serial; /* the module's of the last snapshot published */
     struct PureSurfaceHost *surface; /* told of each snapshot published */
     /* The owner's: outputs wired to HTTP receivers, and how far each was trimmed. */
     PureWire *wires[OUTLETS];
@@ -443,8 +444,32 @@ static App *apps;
 static _Atomic unsigned app_count;
 static unsigned app_slots = 1;
 static unsigned places(void) { return atomic_load_explicit(&app_count, memory_order_acquire); }
-/* The modules apps ran before the one they run, kept as long as the host runs. */
-static struct Retired { PureModule *module; struct Retired *next; } *retired_modules;
+/* A module an app runs or ran (every app's module is one), and what still reads it: its snapshots and the requests
+ * read with it (holds). One an app ran before is retired, then freed (free_retired) once nothing holds it, no app runs
+ * it, and every connection thread has begun a pass since (a request read with it took its hold in resolve, within the
+ * pass it was read in). Who keeps a module to ask later whether it is the same one asks its serial. */
+typedef struct Loaded {
+    PureModule module; /* first: a module's address is its Loaded's */
+    _Atomic uint32_t holds;
+    uint64_t *passes; /* retired: each connection thread's passes then */
+    struct Loaded *next;
+} Loaded;
+static Loaded *retired_modules;
+static uint64_t modules_freed, modules_at_freed; /* (for its stats: freed, and loaded where one freed was) */
+static uintptr_t freed_at[64]; /* the last ones freed: addresses, compared, never read */
+static pthread_mutex_t binding_lock = PTHREAD_MUTEX_INITIALIZER; /* a link naming an input reads an app's module */
+static void hold_module(const PureModule *m) {
+    if (m) atomic_fetch_add_explicit(&((Loaded *)(uintptr_t)m)->holds, 1, memory_order_relaxed);
+}
+static void let_go_module(const PureModule *m) {
+    if (m) atomic_fetch_sub_explicit(&((Loaded *)(uintptr_t)m)->holds, 1, memory_order_release);
+}
+/* A snapshot let go: when it was the last reference, its module's hold too (taken when it was made). */
+static void let_go_snapshot(PureSnapshot *snapshot) {
+    if (!snapshot) return;
+    const PureModule *m = pure_snapshot_module(snapshot);
+    if (pure_snapshot_release(snapshot)) let_go_module(m);
+}
 static bool named_apps; /* --apps: a route's first segment names its app */
 static Host *all_hosts;  /* the connection threads, for moving a connection to its app's home */
 static unsigned host_count;
@@ -458,6 +483,7 @@ struct Host {
     PureApp *readers; /* by app: shows its owner's snapshots here */
     PureSnapshot **snapshots; /* by app: the snapshot a round reads views from, once */
     uint64_t round, *read_in, *handed_in; /* by app: the round it was read in, its requests handed in */
+    _Atomic uint64_t passes; /* its loop's passes begun: a retired module waits for each thread's next */
     unsigned *touched; /* the apps read in this round */
     Applied *views; /* a round's views shown here, answered together */
     size_t views_capacity;
@@ -563,6 +589,8 @@ static void close_connection(Connection *c) {
     if (c->phase == CLOSED) return;
     Host *host = c->host;
     finish_output(c);
+    let_go_module(c->module);
+    c->module = NULL;
     close(c->fd); /* also leaves the epoll set */
     c->phase = CLOSED;
     host->table[c->index] = NULL;
@@ -882,7 +910,10 @@ static const char *resolve(Connection *c, char *query, Route *route, int *status
         return "stopped";
     }
     route->app = app;
+    let_go_module(c->module); /* (a request before it on this connection: answered, so let go already) */
+    hold_module(m);           /* within this pass: no retired module is freed before this thread's next */
     route->module = c->module = m;
+    route->serial = m->serial;
     if (route->kind == ROUTE_NONE || (route->kind == ROUTE_QUERY && !app->owned)) return "route"; /* a query reads states */
     if (route->kind == ROUTE_QUERY) { /* the view is the query module's, found when shown */
         route->from = 0;
@@ -1295,6 +1326,8 @@ static void drive(Connection *c) {
             if (c->out_at < c->out_count) continue;
             /* Answered: the next request may already be here. */
             finish_output(c);
+            let_go_module(c->module);
+            c->module = NULL;
             c->deadline = c->host->now + c->host->timeout_ms;
             if (!c->keep_alive) { close_connection(c); return; }
             c->phase = READING_HEAD;
@@ -1768,16 +1801,17 @@ static void surface_changed(PureSurfaceHost *s) {
  * saved, so a read made after a write's answer sees the write. A failed save
  * returns the app to the last save's states, which the snapshot still shows. */
 static void publish(Executor *ex) {
-    if (!ex->published || (ex->publications && ex->published_changes == ex->app.changes && ex->published_module == ex->module))
+    if (!ex->published || (ex->publications && ex->published_changes == ex->app.changes && ex->published_serial == ex->module->serial))
         return;
     PureSnapshot *made = pure_app_snapshot(&ex->app), *old;
+    if (made) hold_module(ex->module); /* let go with the snapshot's last reference (let_go_snapshot) */
     pthread_mutex_lock(&ex->published->lock);
     old = ex->published->snapshot;
     ex->published->snapshot = made;
     pthread_mutex_unlock(&ex->published->lock);
-    pure_snapshot_release(old);
+    let_go_snapshot(old);
     ex->published_changes = ex->app.changes;
-    ex->published_module = ex->module;
+    ex->published_serial = ex->module->serial;
     ++ex->publications;
     if (ex->surface) surface_changed(ex->surface);
     if (atomic_load_explicit(&ex->published->subscribers, memory_order_relaxed)) wake_pushers();
@@ -1803,6 +1837,7 @@ typedef struct Local {
     Owner *from, *to;
     uint32_t output, input; /* the sender's output, the receiver's input */
     const PureModule *sender, *module; /* the sender's module; the receiver's, which input is in */
+    uint64_t serial; /* the receiver's module's: asked, not its address */
     char input_name[PURE_MODULE_NAME + 1], why[160]; /* the input by name; why it holds its sends, as last said */
     char source[PURE_SOURCE + 1];
     unsigned char fields[PURE_FIELDS_NAMED]; /* the sender's, when the input takes them by name */
@@ -1873,6 +1908,7 @@ static bool local_bind(Local *l, const PureModule *r, char *error, size_t size)
     l->input = (uint32_t)input;
     l->fields_size = fields_size;
     l->module = r;
+    l->serial = r->serial;
     return true;
 }
 
@@ -1889,11 +1925,12 @@ static void local_hand(Executor *ex, Local *l, uint64_t now) {
     pthread_mutex_unlock(&l->lock);
     if (waits) return;
     Owner *to = l->to;
-    pthread_mutex_lock(&to->lock);
+    char error[160] = "its receiver does not run";
+    pthread_mutex_lock(&to->lock); /* bound while it runs the module (a module it ran before may be freed) */
     const PureModule *runs = to->stopping ? NULL : to->module;
+    bool bound = runs && (runs->serial == l->serial || local_bind(l, runs, error, sizeof(error)));
     pthread_mutex_unlock(&to->lock);
-    char error[160] = "its receiver does not run";
-    if (!runs || (runs != l->module && !local_bind(l, runs, error, sizeof(error)))) {
+    if (!bound) {
         if (strcmp(error, l->why)) { /* said once a cause */
             snprintf(l->why, sizeof(l->why), "%s", error);
             fprintf(stderr, "{\"local\":\"%s\",\"input\":\"%s\",\"error\":\"%s; its sends wait\"}\n", l->source,
@@ -1931,10 +1968,11 @@ static void local_hand(Executor *ex, Local *l, uint64_t now) {
     l->count = 0;
     l->retry = 0;
     pthread_mutex_unlock(&l->lock);
-    Job job = {.route = {.kind = ROUTE_INPUT, .target = l->input, .module = l->module}, .link = link,
+    Job job = {.route = {.kind = ROUTE_INPUT, .target = l->input, .module = l->module, .serial = l->serial}, .link = link,
                .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}};
     pthread_mutex_lock(&to->lock);
-    bool queued = !to->stopping && to->module == l->module && grow(&to->jobs, &to->capacity, to->count + 1, sizeof(*to->jobs));
+    bool queued = !to->stopping && to->module && to->module->serial == l->serial &&
+                  grow(&to->jobs, &to->capacity, to->count + 1, sizeof(*to->jobs));
     if (queued) {
         to->jobs[to->count++] = job;
         ++to->arrivals;
@@ -2295,7 +2333,7 @@ static PureSnapshot *read_newest(PureApp *reader, App *app) {
     if (snapshot) pure_snapshot_hold(snapshot);
     pthread_mutex_unlock(&published->lock);
     if (snapshot && !pure_app_read(reader, snapshot)) { /* no room for its states: none shown */
-        pure_snapshot_release(snapshot);
+        let_go_snapshot(snapshot);
         snapshot = NULL;
     }
     reader->work_budget = atomic_load_explicit(&app->work, memory_order_relaxed);
@@ -2370,7 +2408,7 @@ static void *run_view_worker(void *unused) {
         PureApp *reader = &readers[app - apps];
         PureSnapshot *snapshot = read_newest(reader, app);
         Applied a = {.host = job->host, .ticket = job->ticket, .input = job->input, .route = ROUTE_VIEW, .group = job->group};
-        if (!snapshot || reader->module != job->route.module) { /* its app stopped, or runs another module now */
+        if (!snapshot || reader->module_serial != job->route.serial) { /* its app stopped, or runs another module now */
             a.status = PURE_CLOSED;
             a.refused = "stopped";
         } else {
@@ -2378,7 +2416,7 @@ static void *run_view_worker(void *unused) {
             a.changes = reader->changes;
             if (job->group && a.status != PURE_OK && a.fault == PURE_FAULT_NONE) a.fault = pure_app_view_fault(reader);
         }
-        if (snapshot) pure_snapshot_release(snapshot); /* an answer holds what it lends */
+        if (snapshot) let_go_snapshot(snapshot); /* an answer holds what it lends */
         if (!job->group) atomic_fetch_sub_explicit(&app->viewing, 1, memory_order_relaxed);
         give_back(job->host, &a, 1);
         free(job);
@@ -2445,7 +2483,9 @@ typedef struct Group {
     /* A view worker evaluates it (a view with a body): for the module asked
      * of; again: a newer snapshot came meanwhile. It stays until it is back. */
     bool evaluating, again;
-    const PureModule *asked;
+    uint64_t asked; /* the module's serial */
+    uint8_t view_type; /* its view's, and whether it takes arguments: asked again of the module the app runs next */
+    bool arguments;
     Subscriber *first;
     struct Group *next;
 } Group;
@@ -2663,19 +2703,19 @@ static void push_group(Host *host, Group *g) {
     bool quiet = !atomic_load(&app->live);
     PureSnapshot *snapshot = read_published(host, app);
     if (quiet && !snapshot) { close_group(host, g); return; }
-    if (quiet) { pure_snapshot_release(snapshot); return; }
+    if (quiet) { let_go_snapshot(snapshot); return; }
     if (!snapshot) return; /* none yet: the next publication wakes this */
     PureApp *reader = &host->readers[app - apps];
-    if (reader->module != g->route.module) { /* the app changed module: the view of that name, if it is the same kind */
-        const PureModule *was = g->route.module, *now = reader->module;
+    if (reader->module_serial != g->route.serial) { /* the app changed module: the view of that name, if it is the same kind */
+        const PureModule *now = reader->module;
         int view = pure_module_view(now, g->name);
-        if (view < 0 || now->view_type[view] != was->view_type[g->route.target] ||
-            (pure_module_arguments(now, (uint32_t)view) < 0) != (pure_module_arguments(was, g->route.target) < 0)) {
-            pure_snapshot_release(snapshot);
+        if (view < 0 || now->view_type[view] != g->view_type || (pure_module_arguments(now, (uint32_t)view) >= 0) != g->arguments) {
+            let_go_snapshot(snapshot);
             close_group(host, g);
             return;
         }
         g->route.module = now;
+        g->route.serial = reader->module_serial;
         g->route.target = (uint32_t)view;
         g->computed = false;
     }
@@ -2684,7 +2724,7 @@ static void push_group(Host *host, Group *g) {
             if (g->evaluating) g->again = true;
             else {
                 g->evaluating = true;
-                g->asked = g->route.module;
+                g->asked = g->route.serial;
                 hand_view(host, &g->route, (PureTicket){0}, PURE_INPUT_READ, PURE_OBJECT_LIMIT, g);
             }
         } else {
@@ -2695,7 +2735,7 @@ static void push_group(Host *host, Group *g) {
             let_go_applied(&a);
         }
     }
-    pure_snapshot_release(snapshot);
+    let_go_snapshot(snapshot);
     group_send(host, g);
 }
 
@@ -2710,7 +2750,7 @@ static void group_evaluated(Host *host, Applied *a) {
         free_group(g);
         return;
     }
-    if (a->refused || g->asked != g->route.module) g->again = true;
+    if (a->refused || g->asked != g->route.serial) g->again = true;
     else group_value(g, a, a->changes);
     let_go_applied(a);
     if (!group_send(host, g)) return;
@@ -2737,7 +2777,7 @@ static void subscribe(Connection *c, const Route *route, const char *raw, size_t
     if (c->in_used) { refuse(c, 400, "Bad Request", "request"); return; } /* no request follows a stream */
     if (host->subscribers >= subscribers_most) { reject(c, 503, "Service Unavailable", "busy"); return; }
     Group *g = host->groups;
-    while (g && !(g->app == route->app && g->route.module == route->module && g->route.target == route->target &&
+    while (g && !(g->app == route->app && g->route.serial == route->serial && g->route.target == route->target &&
                   g->route.from == route->from && g->route.count == route->count && strlen(g->text) == raw_size &&
                   !memcmp(g->text, raw, raw_size)))
         g = g->next;
@@ -2749,6 +2789,8 @@ static void subscribe(Connection *c, const Route *route, const char *raw, size_t
         g->text[raw_size] = 0;
         g->route.given = g->text;
         snprintf(g->name, sizeof(g->name), "%s", route->module->view_names[route->target]);
+        g->view_type = route->module->view_type[route->target];
+        g->arguments = pure_module_arguments(route->module, route->target) >= 0;
         g->next = host->groups;
         host->groups = g;
     }
@@ -2783,7 +2825,7 @@ static void subscribe(Connection *c, const Route *route, const char *raw, size_t
  * holds up, the owner's writes. */
 enum { QUERIES = 64 };
 typedef struct { PureModule module; unsigned refs; } Held;
-static struct { const App *app; char name[65]; Held *held; const PureModule *checked; } queries[QUERIES];
+static struct { const App *app; char name[65]; Held *held; uint64_t checked; } queries[QUERIES]; /* checked: a serial */
 static pthread_mutex_t queries_lock = PTHREAD_MUTEX_INITIALIZER;
 
 static void let_go_query(Held *held) {
@@ -2838,7 +2880,7 @@ static PureResult register_query(const Job *job) {
         else {
             old = queries[slot].held;
             queries[slot].app = app;
-            queries[slot].checked = job->route.module;
+            queries[slot].checked = job->route.serial;
             snprintf(queries[slot].name, sizeof(queries[slot].name), "%s", job->effect.key);
             queries[slot].held = held;
         }
@@ -2851,7 +2893,7 @@ static PureResult register_query(const Job *job) {
 
 static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a) {
     Held *held = NULL;
-    const PureModule *checked = NULL;
+    uint64_t checked = 0;
     pthread_mutex_lock(&queries_lock);
     for (int i = 0; i < QUERIES && !held; ++i)
         if (queries[i].held && queries[i].app == job->route.app && !strcmp(queries[i].name, job->effect.key)) {
@@ -2870,7 +2912,7 @@ static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a)
         let_go_query(held);
         return PURE_ALLOCATION;
     }
-    if (reader.module != checked && !same_states(&held->module, reader.module)) { /* its app's states are laid out anew */
+    if (reader.module_serial != checked && !same_states(&held->module, reader.module)) { /* its states laid out anew */
         pure_app_stop(&reader);
         let_go_query(held);
         return PURE_CONFLICT;
@@ -2935,7 +2977,7 @@ static void serve_bursts(Host *host) {
             for (size_t k = i; k < count; ++k) {
                 const Job *mine = &host->jobs[k];
                 if (mine->route.app != app || shown_here(host, mine)) continue;
-                if (owner->stopping || mine->route.module != owner->module) { /* its app stopped, or runs another module now */
+                if (owner->stopping || !owner->module || mine->route.serial != owner->module->serial) { /* stopped, or another module now */
                     host->views[answers++] = (Applied){.host = host, .ticket = mine->effect.ticket, .input = mine->effect.input,
                                                        .route = mine->route.kind, .status = PURE_CLOSED, .refused = "stopped"};
                     continue;
@@ -2966,7 +3008,7 @@ static void serve_bursts(Host *host) {
             bool registering = job->route.kind == ROUTE_QUERY && job->effect.input == PURE_INPUT_WRITE;
             *applied = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input,
                                  .route = registering ? ROUTE_QUERY : ROUTE_VIEW};
-            if (job->route.kind == ROUTE_VIEW && host->readers[a].module != job->route.module) { /* another module's view now */
+            if (job->route.kind == ROUTE_VIEW && host->readers[a].module_serial != job->route.serial) { /* another module's view now */
                 applied->status = PURE_CLOSED;
                 applied->refused = "stopped";
                 continue;
@@ -2989,7 +3031,7 @@ static void serve_bursts(Host *host) {
                                                            : run_query(job, host->snapshots[a], applied);
             ++shown;
         }
-        for (size_t i = 0; i < touched; ++i) pure_snapshot_release(host->snapshots[host->touched[i]]); /* answers lend their own */
+        for (size_t i = 0; i < touched; ++i) let_go_snapshot(host->snapshots[host->touched[i]]); /* answers lend their own */
         if (!answers) return;
         for (size_t i = 0; i < answers; ++i) answer_applied(host, &host->views[i]);
         host->shown += shown;
@@ -3201,7 +3243,7 @@ static PureSnapshot *surface_read(PureSurfaceHost *s, PureApp *reader) {
     if (snapshot) pure_snapshot_hold(snapshot);
     pthread_mutex_unlock(&published->lock);
     if (snapshot && !pure_app_read(reader, snapshot)) { /* no room for its states: none shown */
-        pure_snapshot_release(snapshot);
+        let_go_snapshot(snapshot);
         snapshot = NULL;
     }
     return snapshot;
@@ -3209,7 +3251,7 @@ static PureSnapshot *surface_read(PureSurfaceHost *s, PureApp *reader) {
 
 static void surface_release(PureSurfaceHost *s, PureSnapshot *snapshot) {
     (void)s;
-    pure_snapshot_release(snapshot);
+    let_go_snapshot(snapshot);
 }
 
 /* An event queued at the owner as a request is, waited for until its save. */
@@ -3309,9 +3351,10 @@ static bool replies_borrow(const PureModule *m, uint32_t input) {
 /* ---- Main ------------------------------------------------------------- */
 
 /* A module from its file for an app; false, with why, when it does not load
- * or cannot be served. The module the app had is kept (see retired_modules). */
+ * or cannot be served. The module the app had is retired (see Loaded). */
 static bool load_module(App *app, const char *path) {
-    PureModule *m = calloc(1, sizeof(*m));
+    Loaded *loaded = calloc(1, sizeof(*loaded));
+    PureModule *m = loaded ? &loaded->module : NULL;
     char error[160];
     if (!m) {
         snprintf(app->why, sizeof(app->why), "Module: out of memory");
@@ -3319,9 +3362,12 @@ static bool load_module(App *app, const char *path) {
     }
     if (pure_module_read(path, m, error, sizeof(error)) != PURE_OK) {
         snprintf(app->why, sizeof(app->why), "Module: %s", error);
-        free(m);
+        free(loaded);
         return false;
     }
+    bool at_freed = false;
+    for (unsigned i = 0; i < 64; ++i) at_freed = at_freed || freed_at[i] == (uintptr_t)loaded;
+    modules_at_freed += at_freed;
     bool served = true;
     for (uint32_t i = 0; served && i < m->inputs; ++i) {
         if (m->input[i].kind == PURE_KIND_MESSAGE && !replies_borrow(m, i)) {
@@ -3329,24 +3375,60 @@ static bool load_module(App *app, const char *path) {
             served = false;
         }
     }
-    struct Retired *kept = served && app->module ? malloc(sizeof(*kept)) : NULL;
-    if (served && app->module && !kept) {
+    uint64_t *passes = served && app->module ? calloc(host_count ? host_count : 1, sizeof(*passes)) : NULL;
+    if (served && app->module && !passes) {
         snprintf(app->why, sizeof(app->why), "Module: out of memory");
         served = false;
     }
     if (!served) {
         pure_module_release(m);
-        free(m);
+        free(loaded);
         return false;
     }
-    if (kept) { /* requests and snapshots may still point to it */
-        *kept = (struct Retired){app->module, retired_modules};
-        retired_modules = kept;
+    if (passes) { /* requests and snapshots may still read it: freed once none does (free_retired) */
+        Loaded *was = (Loaded *)(uintptr_t)app->module;
+        for (unsigned h = 0; h < host_count; ++h) passes[h] = atomic_load_explicit(&all_hosts[h].passes, memory_order_acquire);
+        was->passes = passes;
+        was->next = retired_modules;
+        retired_modules = was;
+        uint64_t one = 1;
+        for (unsigned h = 0; h < host_count; ++h) /* each begins a pass now, not when its wait ends: freed sooner */
+            if (write(all_hosts[h].wake, &one, sizeof(one)) != sizeof(one)) { /* full: it wakes anyway */ }
     }
     app->module = m;
     return true;
 }
 
+/* The retired modules nothing reads any more, freed, on the thread that changes apps (run_apps): no app runs one (its
+ * line, its owner, its executor, what a request reads), every connection thread has begun a pass since it was retired
+ * (a request resolved with it holds it by then), then (in that order) no snapshot or request holds it. A link naming
+ * an input reads an app's module under binding_lock: taken, none is reading one. */
+static void free_retired(void) {
+    for (Loaded **at = &retired_modules; *at;) {
+        Loaded *l = *at;
+        const PureModule *m = &l->module;
+        bool kept = false;
+        for (unsigned a = 0, count = places(); !kept && a < count; ++a)
+            kept = apps[a].module == m || apps[a].owner.executor.module == m || apps[a].owner.module == m ||
+                   atomic_load_explicit(&apps[a].live, memory_order_acquire) == m;
+        for (unsigned h = 0; !kept && h < host_count; ++h)
+            kept = atomic_load_explicit(&all_hosts[h].passes, memory_order_acquire) <= l->passes[h];
+        kept = kept || atomic_load_explicit(&l->holds, memory_order_acquire) > 0;
+        if (kept) {
+            at = &l->next;
+            continue;
+        }
+        *at = l->next;
+        pthread_mutex_lock(&binding_lock);
+        pthread_mutex_unlock(&binding_lock);
+        pure_module_release(&l->module);
+        free(l->passes);
+        freed_at[modules_freed % 64] = (uintptr_t)l;
+        free(l);
+        ++modules_freed;
+    }
+}
+
 /* A connection on this thread for fd (accepted, or moved here); NULL, fd
  * closed, when it is full (the peer sees the close) or out of memory. */
 static Connection *adopt(Host *host, int fd) {
@@ -3360,6 +3442,7 @@ static Connection *adopt(Host *host, int fd) {
     c->index = index;
     c->fd = fd;
     c->phase = READING_HEAD;
+    c->module = NULL;
     c->readable = c->writable = c->keep_alive = true;
     c->hung_up = c->settled = false;
     c->deadline = host->now + host->timeout_ms;
@@ -3429,6 +3512,8 @@ static bool moved_home(Connection *c, size_t head_size) {
     if (write(home->wake, &one, sizeof(one)) != sizeof(one)) abort();
     Host *host = c->host; /* here it is gone, its descriptor open there */
     epoll_ctl(host->epoll, EPOLL_CTL_DEL, c->fd, NULL);
+    let_go_module(c->module); /* (none yet: a connection moves at its first request) */
+    c->module = NULL;
     c->phase = CLOSED;
     host->table[c->index] = NULL;
     c->next = host->closed;
@@ -3488,7 +3573,7 @@ static void stop_executor(Executor *ex) {
         PureSnapshot *last = ex->published->snapshot;
         ex->published->snapshot = NULL;
         pthread_mutex_unlock(&ex->published->lock);
-        pure_snapshot_release(last);
+        let_go_snapshot(last);
         if (atomic_load_explicit(&ex->published->subscribers, memory_order_relaxed)) wake_pushers(); /* their streams end */
     }
     pure_app_stop(&ex->app);
@@ -3564,6 +3649,7 @@ static void *run_shard(void *context) {
     host->now = now_ms();
     uint64_t next_sweep = host->now + 1000;
     while (!atomic_load(&stopping)) {
+        atomic_fetch_add_explicit(&host->passes, 1, memory_order_acq_rel); /* (a request resolved before is held) */
         /* A thread that runs the app ticks its timers on time. */
         long due = apps[0].owned ? -1 : run_timers(&host->local);
         if (host->push_waiting) { /* a pass waits for --push-every-ms */
@@ -3683,7 +3769,8 @@ static void stop_shard(Host *host) {
  * thread's alone. A delivery whose app runs another module now, or none, is
  * not taken: the receiver keeps it and finds the input again once told
  * (pure_link_listener_rebind, when the app starts). */
-static struct Binding { unsigned app; const PureModule *module; uint32_t input; } *bindings;
+static struct Binding { unsigned app; const PureModule *module; uint64_t serial; uint32_t input; } *bindings; /* serial: the module's */
+static int bind_input(unsigned a, const char *name, uint32_t shape, const unsigned char *fields, size_t size);
 static size_t binding_count, binding_capacity;
 static int link_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
     (void)context;
@@ -3696,6 +3783,15 @@ static int link_input(void *context, const char *name, uint32_t shape, const uns
         if (a == count) return -PURE_LINK_NO_INPUT;
         name = slash + 1;
     }
+    pthread_mutex_lock(&binding_lock); /* the module read stays while it is read (free_retired takes this lock) */
+    int found = bind_input(a, name, shape, fields, size);
+    pthread_mutex_unlock(&binding_lock);
+    return found;
+}
+
+/* link_input's binding of app a's input name in the module it runs, under binding_lock. A binding is of a module (its
+ * shape was checked against that one's): one of a module the app no longer runs stays, refusing what comes to it. */
+static int bind_input(unsigned a, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
     const PureModule *m = atomic_load_explicit(&apps[a].live, memory_order_acquire);
     int index = m ? pure_module_input(m, name) : -1;
     if (index < 0 || m->input[index].kind != PURE_KIND_MESSAGE) return -PURE_LINK_NO_INPUT;
@@ -3703,10 +3799,10 @@ static int link_input(void *context, const char *name, uint32_t shape, const uns
     if (pure_module_shape(m, (uint32_t)index) != shape && !pure_module_takes(m, (uint32_t)index, fields, size))
         return -PURE_LINK_FIELDS_LACKED;
     for (size_t b = 0; b < binding_count; ++b)
-        if (bindings[b].app == a && bindings[b].module == m && bindings[b].input == (uint32_t)index) return (int)b;
+        if (bindings[b].app == a && bindings[b].serial == m->serial && bindings[b].input == (uint32_t)index) return (int)b;
     if (binding_count > INT32_MAX - 1 || !grow(&bindings, &binding_capacity, binding_count + 1, sizeof(*bindings)))
         return -PURE_LINK_NO_INPUT;
-    bindings[binding_count] = (struct Binding){a, m, (uint32_t)index};
+    bindings[binding_count] = (struct Binding){a, m, m->serial, (uint32_t)index};
     return (int)binding_count++;
 }
 
@@ -3734,10 +3830,10 @@ static bool link_deliver(void *context, const PureLinkDelivery *d) {
     memcpy(ingress + 4 + DELIVERY, d->records, d->size);
     if (d->fields_size) memcpy(ingress + 4 + DELIVERY + d->size, d->fields, d->fields_size); /* laid out with those */
     *link = (LinkJob){.listener = link_listener, .token = d->token, .ingress = ingress};
-    Job job = {.route = {.kind = ROUTE_INPUT, .target = bound->input, .module = bound->module}, .link = link,
+    Job job = {.route = {.kind = ROUTE_INPUT, .target = bound->input, .module = bound->module, .serial = bound->serial}, .link = link,
                .effect = {.input = PURE_INPUT_WRITE, .key = "", .expected = -1, .value = {ingress, size}}};
     pthread_mutex_lock(&owner->lock);
-    bool queued = !owner->stopping && owner->module == bound->module &&
+    bool queued = !owner->stopping && owner->module && owner->module->serial == bound->serial &&
                   grow(&owner->jobs, &owner->capacity, owner->count + 1, sizeof(*owner->jobs));
     if (queued) {
         owner->jobs[owner->count++] = job;
@@ -4397,6 +4493,9 @@ static bool resume_app(App *app) {
     pthread_mutex_unlock(&turns.lock);
     atomic_store(&app->gone, false);
     atomic_store_explicit(&app->live, app->module, memory_order_release);
+    /* The snapshot was published before it took requests: a push group that looked meanwhile found it quiet and
+     * waits for the next word. This is it (else a stream waits for the app's next save). */
+    if (ex->published && atomic_load_explicit(&ex->published->subscribers, memory_order_relaxed)) wake_pushers();
     app->running = true;
     if (link_listener) {
         char prefix[APP_NAME + 2];
@@ -4645,7 +4744,12 @@ static void *run_apps(void *context) {
     sigset_t hangup;
     sigemptyset(&hangup);
     sigaddset(&hangup, SIGHUP);
-    for (int signal_number; !sigwait(&hangup, &signal_number) && !atomic_load(&stopping);) read_again(path);
+    for (;;) { /* retired modules waiting: looked at again every 100 ms, until all are freed */
+        free_retired();
+        int got = retired_modules ? sigtimedwait(&hangup, NULL, &(struct timespec){0, 100 * 1000000}) : sigwaitinfo(&hangup, NULL);
+        if (atomic_load(&stopping)) break;
+        if (got == SIGHUP) read_again(path);
+    }
     return NULL;
 }
 
@@ -5002,8 +5106,11 @@ int main(int argc, char **argv) {
             heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused,
             windows_full, unauthorized, forbidden, subscribers, subscribed, pushes, pushed, replaced);
     if (named_apps) { /* what each app did */
-        fprintf(stderr, ",\"workers\":%u,\"moved_home\":%" PRIu64 ",\"handed\":%" PRIu64 ",\"apps\":[",
-                worker_count, moved, handed);
+        size_t still = 0;
+        for (const Loaded *l = retired_modules; l; l = l->next) ++still;
+        fprintf(stderr, ",\"workers\":%u,\"moved_home\":%" PRIu64 ",\"handed\":%" PRIu64 ",\"modules_freed\":%" PRIu64
+                ",\"modules_retired\":%zu,\"modules_at_freed\":%" PRIu64 ",\"apps\":[", worker_count, moved, handed, modules_freed,
+                still, modules_at_freed);
         for (unsigned a = 0, count = places(); a < count; ++a) {
             const App *app = &apps[a];
             fprintf(stderr, "%s{\"name\":", a ? "," : "");