Run many apps on one host, by turns on workers (017)

pure-serve --apps FILE hosts an app a line: NAME MODULE STORE and its
budgets (memory=, work=, queue=). Each app's routes start with its name
and reach its own module, states, snapshot and queries; a view is shown
on the connection thread from its own app's snapshot, read once a round.

No app has a thread of its own. Its owner is what used to be a thread's
loop, now a turn: its due timers, what its wires delivered, and the
requests waiting for it applied and saved once, its answers handed back.
Workers (--workers N; one for a host of one app, else one a CPU at most)
take apps from a queue in turn; a worker sleeps until the soonest timer.
A turn takes every waiting request when no other app waits, else 256;
it gathers the clients its last save answered only when alone.

An app's queue budget refuses what would wait past it (503 busy) on the
connection thread, before its owner sees it. An app whose module does
not load, whose store does not open, or whose store is another app's
stops alone: the start line says why, and its routes answer 503
stopped. The stop line counts each app's turns, events and refusals.

A host of one app (--module) runs as before, its owner on one worker
named pure-owner: every check passes unchanged. server_apps runs six
apps of which two stop, floods one past its queue while another
answers, refuses a walk past one app's work budget that another runs,
holds an app's store from another process (that app answers 503
store_busy and applies nothing, the others save), and restarts.

2 files changed, 750 insertions, 241 deletions

diff --git a/serve/serve.c b/serve/serve.c
index b283874..a7a3d6c 100644
--- a/serve/serve.c
+++ b/serve/serve.c
@@ -60,6 +60,7 @@
 #include <sys/epoll.h>
 #include <sys/eventfd.h>
 #include <sys/socket.h>
+#include <sys/stat.h>
 #include <sys/uio.h>
 #include <time.h>
 #include <unistd.h>
@@ -74,7 +75,8 @@ typedef struct Host Host;
 typedef enum { ROUTE_NONE, ROUTE_OBJECT, ROUTE_VIEW, ROUTE_INPUT, ROUTE_QUERY } RouteKind;
 /* A view's (or a query's) arguments come as the request's query: argued
  * says so, and given is its text, kept on the request's work slot. */
-typedef struct { RouteKind kind; uint32_t target; size_t from, count; bool argued; const char *given; } Route;
+typedef struct App App;
+typedef struct { RouteKind kind; uint32_t target; size_t from, count; bool argued; const char *given; App *app; } Route;
 
 /* This node's key, for deliveries sealed to it (--key), and whether it
  * takes only those (--sealed-only), or only those from the senders named,
@@ -114,6 +116,7 @@ typedef struct Connection {
     uint64_t deadline;
     char key[65]; /* the document's key, or the view's or input's name */
     RouteKind route;
+    App *app; /* the app the path names (NULL: none of the host's) */
     /* An app input's message: its fixed fields from the query, which fields
      * were given there, and whether the body is a JSON object of the rest.
      * Its ingress is [u32le payload size][payload]; with JSON it is [u32le
@@ -166,6 +169,7 @@ typedef struct {
     PureFault fault;  /* a failed event's */
     const char *type; /* a view's content type */
     struct LinkJob *link; /* records a link delivered (link/link.h) */
+    const char *refused;  /* refused by the host (503 and this name), not applied */
     uint64_t heard;       /* then the source's last number heard after them */
     bool gap;             /* and whether they stopped at a gap */
 } Applied;
@@ -305,7 +309,7 @@ typedef struct LinkJob { PureLinkListener *listener; void *token; unsigned char
 typedef struct { Host *host; PureEffect effect; Route route; Waiter *waiter; LinkJob *link; } Job;
 
 /* The owner of an app with states served by several connection threads:
- * their requests wait here, and it runs them in bursts, one save each
+ * their requests wait here, and a worker runs them in bursts, one save each
  * however many threads they came from, handing each thread back its
  * applied requests to answer (and a surface its events). */
 typedef struct Owner {
@@ -319,14 +323,53 @@ typedef struct Owner {
     bool delivered; /* a wire delivered a send: trim it */
     Host *hosts;
     unsigned host_count;
+    /* Its turns on the workers (docs/design/017-apps-on-one-host.md): queued
+     * with work to do, or running on one; again when work came while it ran;
+     * due when its next timer is (the monotonic clock in ms, 0: none). */
+    struct Owner *next;
+    bool queued, running, again;
+    uint64_t due, turns;
+    size_t queue; /* how many requests may wait for its turn (0: any number) */
+    uint64_t busy; /* requests refused for that */
+    /* Between its turns: whether the last one saved (it then gathers the
+     * clients it answered), the arrivals when it handed them back, and the
+     * jobs' spare array. */
+    bool gather;
+    uint64_t answered_at;
+    Job *spare;
+    size_t spare_capacity;
 } Owner;
 
-struct Host {
-    const PureModule *module;
+/* An app of the host (docs/design/017-apps-on-one-host.md): its module,
+ * store and budgets, and the owner that applies its events on the workers,
+ * taking turns with the host's other apps. The one app of a host started
+ * with --module has no name (its routes start /views/ and so on); each of
+ * --apps has its own, the first segment of its routes. */
+enum { APP_NAME = 32 };
+struct App {
+    char name[APP_NAME + 1];
+    PureModule module;
+    const char *store;
     bool documents; /* the module serves /objects: it has show and edit */
-    Executor local; /* the app, when this thread runs it */
-    Owner *owner;   /* or the owner that runs it */
-    PureApp reader; /* shows the owner's snapshot here */
+    bool owned;     /* an owner runs it (states or outputs, or any of --apps); else the connection threads */
+    uint32_t show, edit;
+    size_t memory;  /* its budgets (0: none): what its states hold, */
+    uint64_t work;  /* what an event or a view's evaluation may run */
+    bool running;   /* started; else its routes answer 503 stopped, and why says why */
+    char why[256];
+    Owner owner;
+};
+static App *apps;
+static unsigned app_count;
+static bool named_apps; /* --apps: a route's first segment names its app */
+static void schedule(Owner *owner);
+
+struct Host {
+    Executor local;   /* the one app of the host, when this thread runs it (documents alone) */
+    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 */
+    unsigned *touched; /* the apps read in this round */
     Applied *views; /* a round's views shown here, answered together */
     size_t views_capacity;
     uint64_t shown; /* views shown from snapshots */
@@ -606,6 +649,15 @@ static bool parse_head(Connection *c, char *head, size_t size, char **query) {
     *version++ = 0;
     if (strcmp(version, "HTTP/1.1")) return false;
     if ((*query = strchr(path, '?'))) *(*query)++ = 0;
+    /* A host of apps (--apps): the path's first segment names one, and the rest routes in it. */
+    c->app = named_apps ? NULL : &apps[0];
+    if (named_apps && *path == '/') {
+        const char *end = strchr(path + 1, '/');
+        size_t length = end ? (size_t)(end - path - 1) : 0;
+        for (unsigned i = 0; end && i < app_count && !c->app; ++i)
+            if (strlen(apps[i].name) == length && !memcmp(apps[i].name, path + 1, length)) c->app = &apps[i];
+        if (end) path += 1 + length;
+    }
     static const struct { const char *prefix; RouteKind kind; } routes[] = {
         {"/objects/", ROUTE_OBJECT}, {"/views/", ROUTE_VIEW}, {"/inputs/", ROUTE_INPUT}, {"/queries/", ROUTE_QUERY}};
     const char *key = "";
@@ -729,13 +781,20 @@ static long percent_decoded(char *text, size_t length) {
  * taken from the query (%XX decoded, as a view's arguments are) into
  * c->fields. NULL, or the error to answer (404 for a name the module lacks,
  * 400 otherwise). */
-static const char *resolve(const Host *host, Connection *c, char *query, Route *route, int *status) {
-    const PureModule *m = host->module;
+static const char *resolve(Connection *c, char *query, Route *route, int *status) {
+    App *app = c->app;
     *status = 404;
     c->field_bytes = 0;
     memset(c->given, 0, sizeof(c->given));
-    if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !host->documents) ||
-        (route->kind == ROUTE_QUERY && !host->owner)) return "route"; /* a query reads an app's states */
+    if (!app) return "no_app";
+    if (!app->running) {
+        *status = 503;
+        return "stopped";
+    }
+    const PureModule *m = &app->module;
+    route->app = app;
+    if (route->kind == ROUTE_NONE || (route->kind == ROUTE_OBJECT && !app->documents) ||
+        (route->kind == ROUTE_QUERY && !app->owned)) return "route"; /* a query reads an app's states */
     if (route->kind == ROUTE_OBJECT) return NULL;
     if (route->kind == ROUTE_QUERY) { /* the view is the query module's, found when shown */
         route->from = 0;
@@ -770,7 +829,7 @@ static const char *resolve(const Host *host, Connection *c, char *query, Route *
     bool message = shape && shape->kind == PURE_KIND_MESSAGE;
     if (message && shape->keyed) return "keyed";
     if (*c->source && !message) return "delivery";
-    if (!host->owner) c->source[0] = 0; /* an app without states or outputs applies it as a message */
+    if (!app->owned) c->source[0] = 0; /* an app without states or outputs applies it as a message */
     if (shape) {
         /* A body is a text input's text, a message's one variable field (rows
          * in their list's stored form), or a JSON object of a message's fields
@@ -841,14 +900,14 @@ enum { DELIVERY = 80 };
  * into the prefix; strings stay in the body, and the table says where.
  * NULL, or the error to answer. */
 static const char *convert(Connection *c) {
-    const PureInputShape *shape = &c->host->module->input[c->input];
+    const PureInputShape *shape = &(&c->app->module)->input[c->input];
     uint32_t delivered = get_u32(c->ingress) & DELIVERED;
     size_t table = 8 * (size_t)shape->field_count, payload = c->batch * c->field_bytes + c->body_size;
     unsigned char *tables = c->ingress + 4 + (delivered ? DELIVERY : 0), *prefixes = tables + c->batch * table;
     unsigned char *json = c->body;
     size_t size = c->body_size, at = pure_json_space(json, 0, size);
     /* A message's rows go after the body: they are encoded, then placed. */
-    PureJsonMessage b = {.module = c->host->module, .shape = shape, .given = c->given, .rows_base = payload};
+    PureJsonMessage b = {.module = (&c->app->module), .shape = shape, .given = c->given, .rows_base = payload};
     const char *why = NULL;
     /* A batch is an array of its messages' objects. */
     if (c->batch > 1) {
@@ -868,7 +927,7 @@ static const char *convert(Connection *c) {
          * member naming no field is refused all the same: it would be dropped. */
         for (uint32_t f = 0; !why && f < shape->field_count; ++f) {
             if (b.seen[f] || c->given[f]) continue;
-            const PureField *field = &c->host->module->field[shape->field_start + f];
+            const PureField *field = &(&c->app->module)->field[shape->field_start + f];
             if (!delivered || field->type == PURE_FIELD_ROWS) why = "missing_field";
             else if (pure_field_fixed(field->type))
                 memset(b.payload + field->offset, 0, pure_field_size(field->type));
@@ -979,12 +1038,15 @@ static void start_request(Connection *c, size_t head_size) {
     ++host->requests;
     Route route = {.kind = c->route};
     int status;
-    const char *why = resolve(host, c, query, &route, &status);
-    if (why) { reject(c, status, status == 404 ? "Not Found" : "Bad Request", why); return; }
+    const char *why = resolve(c, query, &route, &status);
+    if (why) {
+        reject(c, status, status == 404 ? "Not Found" : status == 503 ? "Service Unavailable" : "Bad Request", why);
+        return;
+    }
     /* A document's PUT carries its body; an input's event its payload's size,
      * the payload (the fixed fields, then the body), and room for its JSON. */
     bool write = route.kind == ROUTE_INPUT || c->put;
-    size_t table = route.kind == ROUTE_INPUT && c->json ? 8 * (size_t)host->module->input[c->input].field_count : 0;
+    size_t table = route.kind == ROUTE_INPUT && c->json ? 8 * (size_t)c->app->module.input[c->input].field_count : 0;
     size_t lead = 4 + (route.kind == ROUTE_INPUT && *c->source ? DELIVERY : 0);
     /* A batch has a table and a prefix for each of its messages. */
     size_t input_size = route.kind == ROUTE_INPUT ? lead + c->batch * (table + c->field_bytes) + c->body_size :
@@ -1769,7 +1831,10 @@ static void answer_applied(Host *host, Applied *a) {
         if (result.status == PURE_OK && a->extents)
             respond_typed(c, 200, "OK", "none", result.revision, a->type ? a->type : "application/octet-stream",
                           a->extents, a->extent_count, result.body.size);
-        else if (a->route != ROUTE_OBJECT) answer_app(c, result, a->route == ROUTE_VIEW, a->fault, a->type);
+        else if (a->refused) {
+            respond(c, 503, "Service Unavailable", a->refused, 0, (PureSpan){0});
+            release_ticket(c);
+        } else if (a->route != ROUTE_OBJECT) answer_app(c, result, a->route == ROUTE_VIEW, a->fault, a->type);
         else answer(c, result);
         drive(c); /* most answers fit the socket buffer at once */
     }
@@ -1837,13 +1902,15 @@ static void answer_handed(Host *host) {
 /* The owner's newest snapshot, held, shown by this thread's reader; NULL
  * when there is none. Taken for each round, after the round's answers from
  * the owner: those were handed back after their saves' snapshots. */
-static PureSnapshot *read_published(Host *host) {
-    Executor *ex = &host->owner->executor;
+static PureSnapshot *read_published(Host *host, App *app) {
+    Executor *ex = &app->owner.executor;
     pthread_mutex_lock(&ex->published_lock);
     PureSnapshot *snapshot = ex->published;
     if (snapshot) pure_snapshot_hold(snapshot);
     pthread_mutex_unlock(&ex->published_lock);
-    if (snapshot) pure_app_read(&host->reader, snapshot);
+    PureApp *reader = &host->readers[app - apps];
+    if (snapshot) pure_app_read(reader, snapshot);
+    reader->work_budget = app->work;
     return snapshot;
 }
 
@@ -1855,7 +1922,7 @@ static PureSnapshot *read_published(Host *host) {
  * holds up, the owner's writes. */
 enum { QUERIES = 64 };
 typedef struct { PureModule module; unsigned refs; } Held;
-static struct { char name[65]; Held *held; } queries[QUERIES];
+static struct { const App *app; char name[65]; Held *held; } queries[QUERIES];
 static pthread_mutex_t queries_lock = PTHREAD_MUTEX_INITIALIZER;
 
 static void let_go_query(Held *held) {
@@ -1889,7 +1956,8 @@ static bool same_states(const PureModule *a, const PureModule *b) {
     return true;
 }
 
-static PureResult register_query(const Host *host, const Job *job) {
+static PureResult register_query(const Job *job) {
+    const App *app = job->route.app;
     Held *held = calloc(1, sizeof(*held));
     char error[128];
     if (!held) return PURE_ALLOCATION;
@@ -1897,7 +1965,7 @@ static PureResult register_query(const Host *host, const Job *job) {
         free(held);
         return PURE_INVALID;
     }
-    PureResult status = !same_states(&held->module, host->module)          ? PURE_CONFLICT
+    PureResult status = !same_states(&held->module, &app->module)          ? PURE_CONFLICT
                         : pure_module_view(&held->module, job->effect.key) < 0 ? PURE_NOT_FOUND
                                                                                : PURE_OK;
     Held *old = NULL;
@@ -1906,12 +1974,13 @@ static PureResult register_query(const Host *host, const Job *job) {
         pthread_mutex_lock(&queries_lock);
         int slot = -1;
         for (int i = 0; i < QUERIES && slot < 0; ++i)
-            if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) slot = i;
+            if (queries[i].held && queries[i].app == app && !strcmp(queries[i].name, job->effect.key)) slot = i;
         for (int i = 0; i < QUERIES && slot < 0; ++i)
             if (!queries[i].held) slot = i;
         if (slot < 0) status = PURE_LIMIT;
         else {
             old = queries[slot].held;
+            queries[slot].app = app;
             snprintf(queries[slot].name, sizeof(queries[slot].name), "%s", job->effect.key);
             queries[slot].held = held;
         }
@@ -1926,7 +1995,8 @@ static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a)
     Held *held = NULL;
     pthread_mutex_lock(&queries_lock);
     for (int i = 0; i < QUERIES && !held; ++i)
-        if (queries[i].held && !strcmp(queries[i].name, job->effect.key)) (held = queries[i].held)->refs++;
+        if (queries[i].held && queries[i].app == job->route.app && !strcmp(queries[i].name, job->effect.key))
+            (held = queries[i].held)->refs++;
     pthread_mutex_unlock(&queries_lock);
     if (!held) return PURE_NOT_FOUND;
     if (!snapshot) {
@@ -1935,7 +2005,7 @@ static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a)
     }
     PureApp reader = {0};
     pure_app_read(&reader, snapshot);
-    pure_app_work(&reader, work_budget);
+    pure_app_work(&reader, job->route.app->work);
     reader.module = &held->module; /* the app's states, the query's views */
     Route route = job->route;
     route.target = (uint32_t)pure_module_view(&held->module, job->effect.key);
@@ -1945,6 +2015,12 @@ static PureResult run_query(const Job *job, PureSnapshot *snapshot, Applied *a)
     return status;
 }
 
+/* Whether a job of the round is shown on this thread: a query, or a view
+ * whose app has a snapshot to show it from. */
+static bool shown_here(const Host *host, const Job *job) {
+    return job->route.kind == ROUTE_QUERY || (job->route.kind == ROUTE_VIEW && host->snapshots[job->route.app - apps]);
+}
+
 /* Every request made ready since the last round: run here, or handed to the
  * owner. Views are shown here from the snapshot when there is one, after the
  * rest went to the owner, which meanwhile applies and saves them. Answering
@@ -1956,47 +2032,72 @@ static void serve_bursts(Host *host) {
         while (pure_runtime_take(&host->runtime, &effect) == PURE_TAKE_EFFECT)
             host->jobs[count++] = (Job){host, effect, host->routes[effect.ticket.slot], NULL, NULL};
         if (!count) return;
-        if (!host->owner) {
+        if (!apps[0].owned) { /* the one app of the host, of documents alone: run here */
             execute(&host->local, host->jobs, count, answer_burst, NULL);
             continue;
         }
-        PureSnapshot *snapshot = NULL;
-        size_t views = 0, queried = 0;
+        /* Each app with a view or a query in the round: its snapshot, read
+         * once (NULL: none yet, and its owner shows its views). */
+        ++host->round;
+        size_t touched = 0;
         for (size_t i = 0; i < count; ++i) {
-            views += host->jobs[i].route.kind == ROUTE_VIEW;
-            queried += host->jobs[i].route.kind == ROUTE_QUERY;
+            const Route *route = &host->jobs[i].route;
+            unsigned a = (unsigned)(route->app - apps);
+            if ((route->kind != ROUTE_VIEW && route->kind != ROUTE_QUERY) || host->read_in[a] == host->round) continue;
+            host->read_in[a] = host->round;
+            host->snapshots[a] = read_published(host, route->app);
+            host->touched[touched++] = a;
         }
-        if ((views || queried) && !(snapshot = read_published(host))) views = 0; /* the owner shows them */
-        size_t handed = count - views - queried; /* queries are all this thread's */
-        if (handed) {
-            Owner *owner = host->owner;
+        /* The rest go to their apps' owners, an owner's all at once; past an
+         * app's queue budget they are refused (503 busy). */
+        if (!grow(&host->views, &host->views_capacity, count, sizeof(*host->views))) abort();
+        size_t answers = 0;
+        for (size_t i = 0; i < count; ++i) {
+            const Job *job = &host->jobs[i];
+            App *app = job->route.app;
+            unsigned a = (unsigned)(app - apps);
+            if (shown_here(host, job) || host->handed_in[a] == host->round) continue;
+            host->handed_in[a] = host->round;
+            Owner *owner = &app->owner;
+            size_t handed = 0;
             pthread_mutex_lock(&owner->lock);
-            if (!grow(&owner->jobs, &owner->capacity, owner->count + handed, sizeof(*owner->jobs))) abort();
-            for (size_t i = 0; i < count; ++i)
-                if ((!snapshot || host->jobs[i].route.kind != ROUTE_VIEW) && host->jobs[i].route.kind != ROUTE_QUERY)
-                    owner->jobs[owner->count++] = host->jobs[i];
+            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->queue && owner->count >= owner->queue) {
+                    host->views[answers++] = (Applied){.host = host, .ticket = mine->effect.ticket, .input = mine->effect.input,
+                                                       .route = mine->route.kind, .status = PURE_BUSY, .refused = "busy"};
+                    ++owner->busy;
+                    continue;
+                }
+                if (!grow(&owner->jobs, &owner->capacity, owner->count + 1, sizeof(*owner->jobs))) abort();
+                owner->jobs[owner->count++] = *mine;
+                ++handed;
+            }
             owner->arrivals += handed;
-            pthread_cond_signal(&owner->waiting);
+            pthread_cond_signal(&owner->waiting); /* it may be gathering */
             pthread_mutex_unlock(&owner->lock);
+            if (handed) schedule(owner);
             host->pending += handed;
         }
-        if (!views && !queried) return;
         /* Shown first and answered after, as a burst is. */
-        if (!grow(&host->views, &host->views_capacity, views + queried, sizeof(*host->views))) abort();
         size_t shown = 0;
         for (size_t i = 0; i < count; ++i) {
             const Job *job = &host->jobs[i];
-            if (job->route.kind != ROUTE_QUERY && (job->route.kind != ROUTE_VIEW || !snapshot)) continue;
-            Applied *a = &host->views[shown++];
+            if (!shown_here(host, job)) continue;
+            unsigned a = (unsigned)(job->route.app - apps);
+            Applied *applied = &host->views[answers++];
             bool registering = job->route.kind == ROUTE_QUERY && job->effect.input == PURE_INPUT_WRITE;
-            *a = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input,
-                           .route = registering ? ROUTE_QUERY : ROUTE_VIEW};
-            a->status = job->route.kind == ROUTE_VIEW ? show_view(&host->reader, &job->route, job->effect.available, a)
-                        : registering              ? register_query(host, job)
-                                                   : run_query(job, snapshot, a);
+            *applied = (Applied){.host = host, .ticket = job->effect.ticket, .input = job->effect.input,
+                                 .route = registering ? ROUTE_QUERY : ROUTE_VIEW};
+            applied->status = job->route.kind == ROUTE_VIEW ? show_view(&host->readers[a], &job->route, job->effect.available, applied)
+                              : registering                ? register_query(job)
+                                                           : run_query(job, host->snapshots[a], applied);
+            ++shown;
         }
-        pure_snapshot_release(snapshot); /* what answers lend holds its own mapping */
-        for (size_t i = 0; i < shown; ++i) answer_applied(host, &host->views[i]);
+        for (size_t i = 0; i < touched; ++i) pure_snapshot_release(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;
     }
 }
@@ -2014,84 +2115,161 @@ static bool before(struct timespec a, struct timespec b) {
     return a.tv_sec < b.tv_sec || (a.tv_sec == b.tv_sec && a.tv_nsec < b.tv_nsec);
 }
 
-static void *run_owner(void *context) {
-    Owner *owner = context;
-    Job *spare = NULL;
-    size_t spare_capacity = 0;
-    bool gather = false;
-    uint64_t answered_at = 0; /* arrivals when the last round's answers were handed back */
-    for (;;) {
-        pthread_mutex_lock(&owner->lock);
-        /* Wait for requests, or until the next timer is due; trim what the
-         * wires delivered meanwhile. */
-        for (;;) {
-            bool delivered = owner->delivered;
-            owner->delivered = false;
-            pthread_mutex_unlock(&owner->lock);
-            long due = run_timers(&owner->executor);
-            bool answered = (delivered || owner->executor.answers_wait) && take_deliveries(&owner->executor);
-            if (owner->executor.answers_wait && (due < 0 || due > ANSWER_WAIT_MS)) due = ANSWER_WAIT_MS;
-            pthread_mutex_lock(&owner->lock);
-            /* Answers are saved with the next round, or now if none waits. */
-            if (answered && !owner->count && !owner->stopping) {
-                pthread_mutex_unlock(&owner->lock);
-                if (pure_app_save(&owner->executor.app) == PURE_OK) after_save(&owner->executor);
-                pthread_mutex_lock(&owner->lock);
-            }
-            if (owner->count || owner->stopping) break;
-            if (owner->delivered) continue;
-            if (due < 0) {
-                pthread_cond_wait(&owner->waiting, &owner->lock);
-                continue;
-            }
-            struct timespec wake = pure_wait_until(due * 1000000LL);
-            pthread_cond_timedwait(&owner->waiting, &owner->lock, &wake);
-        }
-        /* After a round that saved, its answers bring requests back from the
-         * connection threads: gather them until none has come for 100 µs (a
-         * millisecond at most), so they share one save as they would on a
-         * single thread, where answering takes that long anyway. The quiet
-         * counts only once a request came after the answers: requests that
-         * waited out the save alone would otherwise start the next round
-         * without the clients just answered, and the clients would settle
-         * into two halves taking turns, each save carrying half of them.
-         * Gathering never outlasts the time the last save waited for its
-         * device, nor does the quiet: waiting pays only while it costs less
-         * than the wait a save of the late requests would take on its own
-         * (a sync of milliseconds on a flash card; in RAM next to none, and
-         * there gathering would only leave the owner idle). */
-        uint64_t waited = owner->executor.save_waited, budget = waited < 1000000 ? waited : 1000000;
-        struct timespec now, end;
-        clock_gettime(PURE_WAIT_CLOCK, &now);
-        end = later(now, (long)budget);
-        while (gather && owner->count && !owner->stopping && before(now, end)) {
-            size_t seen = owner->count;
-            bool returned = owner->arrivals != answered_at;
-            struct timespec quiet = later(now, budget < 100000 ? (long)budget : 100000);
-            pthread_cond_timedwait(&owner->waiting, &owner->lock, returned && before(quiet, end) ? &quiet : &end);
-            if (returned && owner->count == seen) break;
-            clock_gettime(PURE_WAIT_CLOCK, &now);
-        }
-        if (!owner->count) {
-            pthread_mutex_unlock(&owner->lock);
-            break;
-        }
-        Job *jobs = owner->jobs;
-        size_t count = owner->count, capacity = owner->capacity;
-        owner->jobs = spare;
-        owner->capacity = spare_capacity;
-        owner->count = 0;
+/* The workers' queue of apps with work to do (docs/design/017): an app
+ * takes its turns in order, on one worker at a time. */
+static struct {
+    pthread_mutex_t lock;
+    pthread_cond_t wake; /* a worker waits for an app, or for the soonest timer due (on the monotonic clock) */
+    Owner *head, *tail;
+    bool stopping;
+} turns = {.lock = PTHREAD_MUTEX_INITIALIZER};
+
+static void enqueue(Owner *owner) {
+    owner->queued = true;
+    owner->next = NULL;
+    if (turns.tail) turns.tail->next = owner;
+    else turns.head = owner;
+    turns.tail = owner;
+    pthread_cond_signal(&turns.wake);
+}
+
+/* Work came for an app: a turn, unless it waits for one or has one (then
+ * another after it). Any thread, holding none of the owner's locks. */
+static void schedule(Owner *owner) {
+    pthread_mutex_lock(&turns.lock);
+    if (owner->running) owner->again = true;
+    else if (!owner->queued) enqueue(owner);
+    pthread_mutex_unlock(&turns.lock);
+}
+
+/* The most requests a turn takes when other apps wait for a worker (all
+ * of them when none does): a connection thread's round. */
+enum { TURN = BATCH };
+
+/* An app's turn: its timers due, what its wires delivered, and the
+ * requests waiting for it, applied and saved once, their answers handed
+ * back. Alone (no other app waits for a worker), it first gathers the
+ * clients its last save answered. Whether it has requests left at once;
+ * due, its next timer's milliseconds (-1: none). */
+static bool owner_turn(Owner *owner, bool alone, long *due) {
+    Executor *ex = &owner->executor;
+    pthread_mutex_lock(&owner->lock);
+    bool delivered = owner->delivered;
+    owner->delivered = false;
+    pthread_mutex_unlock(&owner->lock);
+    *due = run_timers(ex);
+    bool answered = (delivered || ex->answers_wait) && take_deliveries(ex);
+    if (ex->answers_wait && (*due < 0 || *due > ANSWER_WAIT_MS)) *due = ANSWER_WAIT_MS;
+    pthread_mutex_lock(&owner->lock);
+    /* Answers are saved with the requests waiting, or now if none waits. */
+    if (answered && !owner->count) {
         pthread_mutex_unlock(&owner->lock);
-        owner->executor.wrote = false;
-        execute(&owner->executor, jobs, count, hand_back, owner);
-        gather = owner->executor.wrote;
+        if (pure_app_save(&ex->app) == PURE_OK) after_save(ex);
         pthread_mutex_lock(&owner->lock);
-        answered_at = owner->arrivals;
+    }
+    /* After a turn that saved, its answers bring requests back from the
+     * connection threads: gather them until none has come for 100 µs (a
+     * millisecond at most), so they share one save as they would on a
+     * single thread, where answering takes that long anyway. The quiet
+     * counts only once a request came after the answers: requests that
+     * waited out the save alone would otherwise start the next turn
+     * without the clients just answered, and the clients would settle
+     * into two halves taking turns, each save carrying half of them.
+     * Gathering never outlasts the time the last save waited for its
+     * device, nor does the quiet: waiting pays only while it costs less
+     * than the wait a save of the late requests would take on its own (a
+     * sync of milliseconds on a flash card; in RAM next to none, and there
+     * gathering would only leave the worker idle). It gathers only when no
+     * other app waits for the worker. */
+    uint64_t waited = ex->save_waited, budget = waited < 1000000 ? waited : 1000000;
+    struct timespec now, end;
+    clock_gettime(PURE_WAIT_CLOCK, &now);
+    end = later(now, (long)budget);
+    while (alone && owner->gather && owner->count && !owner->stopping && before(now, end)) {
+        size_t seen = owner->count;
+        bool returned = owner->arrivals != owner->answered_at;
+        struct timespec quiet = later(now, budget < 100000 ? (long)budget : 100000);
+        pthread_cond_timedwait(&owner->waiting, &owner->lock, returned && before(quiet, end) ? &quiet : &end);
+        if (returned && owner->count == seen) break;
+        clock_gettime(PURE_WAIT_CLOCK, &now);
+    }
+    size_t count = owner->count, take = alone || count <= TURN ? count : TURN;
+    if (!take) {
         pthread_mutex_unlock(&owner->lock);
-        spare = jobs;
-        spare_capacity = capacity;
+        return false;
     }
-    free(spare);
+    Job *jobs = owner->jobs;
+    size_t capacity = owner->capacity;
+    if (take == count) { /* all of them: the arrays trade places */
+        owner->jobs = owner->spare;
+        owner->capacity = owner->spare_capacity;
+        owner->count = 0;
+    } else { /* a turn's worth into the spare array; the rest wait for the next */
+        if (!grow(&owner->spare, &owner->spare_capacity, take, sizeof(*owner->spare))) abort();
+        memcpy(owner->spare, jobs, take * sizeof(*jobs));
+        memmove(jobs, jobs + take, (count - take) * sizeof(*jobs));
+        owner->count = count - take;
+        jobs = owner->spare;
+        capacity = owner->spare_capacity;
+    }
+    owner->spare = NULL; /* it is the turn's now, until it is over */
+    owner->spare_capacity = 0;
+    pthread_mutex_unlock(&owner->lock);
+    ex->wrote = false;
+    execute(ex, jobs, take, hand_back, owner);
+    owner->gather = ex->wrote;
+    pthread_mutex_lock(&owner->lock);
+    owner->answered_at = owner->arrivals;
+    owner->spare = jobs;
+    owner->spare_capacity = capacity;
+    bool more = owner->count > 0;
+    pthread_mutex_unlock(&owner->lock);
+    return more;
+}
+
+/* A worker: the next app in the queue takes its turn; an app whose timer
+ * is due joins the queue; with none, it sleeps until the soonest is due.
+ * It stops once the host stops and no app waits. */
+static void *run_worker(void *context) {
+    (void)context;
+    pthread_mutex_lock(&turns.lock);
+    for (;;) {
+        uint64_t now = now_ms(), soonest = UINT64_MAX;
+        for (unsigned i = 0; i < app_count; ++i) {
+            Owner *o = &apps[i].owner;
+            if (!apps[i].running || !apps[i].owned || !o->due || o->queued || o->running) continue;
+            if (o->due <= now) {
+                o->due = 0;
+                enqueue(o);
+            } else if (o->due < soonest) soonest = o->due;
+        }
+        Owner *owner = turns.head;
+        if (!owner) {
+            if (turns.stopping) break;
+            if (soonest == UINT64_MAX) pthread_cond_wait(&turns.wake, &turns.lock);
+            else {
+                struct timespec wake = pure_wait_until((long long)(soonest - now) * 1000000LL);
+                pthread_cond_timedwait(&turns.wake, &turns.lock, &wake);
+            }
+            continue;
+        }
+        turns.head = owner->next;
+        if (!turns.head) turns.tail = NULL;
+        owner->queued = false;
+        owner->running = true;
+        owner->again = false;
+        bool alone = !turns.head;
+        pthread_mutex_unlock(&turns.lock);
+        long due;
+        bool more = owner_turn(owner, alone, &due);
+        pthread_mutex_lock(&turns.lock);
+        owner->running = false;
+        ++owner->turns;
+        owner->due = due < 0 ? 0 : now_ms() + (uint64_t)due;
+        if (more || owner->again) enqueue(owner);
+    }
+    pthread_cond_broadcast(&turns.wake); /* the other workers stop too */
+    pthread_mutex_unlock(&turns.lock);
     return NULL;
 }
 
@@ -2134,6 +2312,7 @@ static PureResult surface_event(PureSurfaceHost *s, uint32_t input, PureSpan tex
         pthread_cond_signal(&owner->waiting);
     }
     pthread_mutex_unlock(&owner->lock);
+    if (queued) schedule(owner);
     pthread_mutex_lock(&s->lock);
     while (queued && !waiter.done) pthread_cond_wait(&s->applied, &s->lock);
     pthread_mutex_unlock(&s->lock);
@@ -2206,10 +2385,14 @@ static bool replies_borrow(const PureModule *m, uint32_t input) {
 
 /* ---- Main ------------------------------------------------------------- */
 
-static bool load_module(PureModule *m, const char *path, bool *documents, uint32_t *show_out, uint32_t *edit_out) {
+/* An app's module from its file, and whether it serves documents; false,
+ * with why, when it does not load or cannot be served. */
+static bool load_module(App *app, const char *path) {
+    PureModule *m = &app->module;
+    bool *documents = &app->documents;
     char error[160];
     if (pure_module_read(path, m, error, sizeof(error)) != PURE_OK) {
-        fprintf(stderr, "Module: %s\n", error);
+        snprintf(app->why, sizeof(app->why), "Module: %s", error);
         return false;
     }
     /* Documents are served when the module keeps a keyed collection: its keyed
@@ -2219,18 +2402,18 @@ static bool load_module(PureModule *m, const char *path, bool *documents, uint32
     *documents = m->collections > 0;
     if (*documents && (show < 0 || edit < 0 || m->input[show].kind != PURE_KIND_MESSAGE || !m->input[show].keyed ||
                        m->input[edit].kind != PURE_KIND_MESSAGE || !m->input[edit].keyed)) {
-        fputs("Module: documents need keyed messages named show and edit\n", stderr);
+        snprintf(app->why, sizeof(app->why), "Module: documents need keyed messages named show and edit");
         return false;
     }
     for (uint32_t i = 0; i < m->inputs; ++i) {
         bool served = (*documents && i == (uint32_t)edit) || (m->input[i].kind == PURE_KIND_MESSAGE && !m->input[i].keyed);
         if (served && !replies_borrow(m, i)) {
-            fprintf(stderr, "Module: %s may reply only a field of its message\n", m->input_names[i]);
+            snprintf(app->why, sizeof(app->why), "Module: %s may reply only a field of its message", m->input_names[i]);
             return false;
         }
     }
-    *show_out = (uint32_t)(show >= 0 ? show : 0);
-    *edit_out = (uint32_t)(edit >= 0 ? edit : 0);
+    app->show = (uint32_t)(show >= 0 ? show : 0);
+    app->edit = (uint32_t)(edit >= 0 ? edit : 0);
     return true;
 }
 
@@ -2274,18 +2457,26 @@ static void accept_all(Host *host) {
 }
 
 
-static bool start_executor(Executor *ex, const PureModule *module, uint32_t show, uint32_t edit, const char *store) {
-    ex->module = module;
-    ex->show = show;
-    ex->edit = edit;
-    if (pure_store_open(store, &ex->store) != PURE_OK) { fputs("Cannot open the store\n", stderr); return false; }
-    if (pure_app_start(&ex->app, module, pure_store_data(ex->store)) != PURE_OK || !pure_app_defer(&ex->app, true)) {
-        fprintf(stderr, "Cannot start the module on the store%s%s\n", ex->app.why[0] ? ": " : "", ex->app.why);
+/* An app's module on its store, with its budgets; false (why says why)
+ * when the store does not open or the module does not start on it. */
+static bool start_executor(Executor *ex, App *app) {
+    ex->module = &app->module;
+    ex->show = app->show;
+    ex->edit = app->edit;
+    if (pure_store_open(app->store, &ex->store) != PURE_OK) {
+        snprintf(app->why, sizeof(app->why), "Cannot open the store");
+        return false;
+    }
+    if (pure_app_start(&ex->app, &app->module, pure_store_data(ex->store)) != PURE_OK || !pure_app_defer(&ex->app, true)) {
+        snprintf(app->why, sizeof(app->why), "Cannot start the module on the store%s%s", ex->app.why[0] ? ": " : "",
+                 ex->app.why);
+        pure_store_close(ex->store);
+        ex->store = NULL;
         return false;
     }
     pure_app_background_checkpoints(&ex->app, true);
-    pure_app_memory(&ex->app, memory_budget);
-    pure_app_work(&ex->app, work_budget);
+    pure_app_memory(&ex->app, app->memory);
+    pure_app_work(&ex->app, app->work);
     return true;
 }
 
@@ -2300,13 +2491,18 @@ static void stop_executor(Executor *ex) {
 /* One connection thread: its own listener on the shared port, epoll set,
  * runtime domain and, unless an owner runs the app for all of them, app and
  * store handle. */
-static bool start_shard(Host *host, const PureModule *module, bool documents, uint32_t show, uint32_t edit,
-                        const char *store, size_t budget, unsigned *port) {
-    host->module = module;
-    host->documents = documents;
+static bool start_shard(Host *host, size_t budget, unsigned *port) {
     host->wake = -1;
-    pure_app_work(&host->reader, work_budget); /* the views it shows from snapshots */
-    if (!host->owner && !start_executor(&host->local, module, show, edit, store)) return false;
+    if (!apps[0].owned && !start_executor(&host->local, &apps[0])) {
+        fprintf(stderr, "%s\n", apps[0].why);
+        return false;
+    }
+    host->readers = calloc(app_count, sizeof(*host->readers));
+    host->snapshots = calloc(app_count, sizeof(*host->snapshots));
+    host->read_in = calloc(app_count, sizeof(*host->read_in));
+    host->handed_in = calloc(app_count, sizeof(*host->handed_in));
+    host->touched = calloc(app_count, sizeof(*host->touched));
+    if (!host->readers || !host->snapshots || !host->read_in || !host->handed_in || !host->touched) return false;
     host->work = calloc(host->capacity, sizeof(*host->work));
     host->table = calloc(host->capacity, sizeof(*host->table));
     host->by_slot = calloc(host->capacity, sizeof(*host->by_slot));
@@ -2339,8 +2535,8 @@ static bool start_shard(Host *host, const PureModule *module, bool documents, ui
         perror("epoll");
         return false;
     }
-    /* The owner says it handed requests back through an eventfd. */
-    if (host->owner) {
+    /* The owners say they handed requests back through an eventfd. */
+    if (apps[0].owned) {
         struct epoll_event wake_event = {.events = EPOLLIN, .data.ptr = &host->wake};
         if (pthread_mutex_init(&host->lock, NULL) || (host->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) < 0 ||
             epoll_ctl(host->epoll, EPOLL_CTL_ADD, host->wake, &wake_event)) {
@@ -2358,7 +2554,7 @@ static void *run_shard(void *context) {
     uint64_t next_sweep = host->now + 1000;
     while (!atomic_load(&stopping)) {
         /* A thread that runs the app ticks its timers on time. */
-        long due = host->owner ? -1 : run_timers(&host->local);
+        long due = apps[0].owned ? -1 : run_timers(&host->local);
         int ready = epoll_wait(host->epoll, events, EVENTS, due >= 0 && due < 1000 ? (int)due : 1000);
         host->now = now_ms();
         for (int e = 0; e < ready; ++e) {
@@ -2401,8 +2597,13 @@ static void *run_shard(void *context) {
 }
 
 static void stop_shard(Host *host) {
-    if (!host->owner) stop_executor(&host->local);
-    pure_app_stop(&host->reader);
+    if (!apps[0].owned) stop_executor(&host->local);
+    for (unsigned i = 0; host->readers && i < app_count; ++i) pure_app_stop(&host->readers[i]);
+    free(host->readers);
+    free(host->snapshots);
+    free(host->read_in);
+    free(host->handed_in);
+    free(host->touched);
     if (host->listener >= 0) close(host->listener);
     if (host->epoll >= 0) close(host->epoll);
     if (host->wake >= 0) close(host->wake);
@@ -2473,7 +2674,7 @@ static bool link_deliver(void *context, const PureLinkDelivery *d) {
     if (!queued) {
         free(ingress);
         free(link);
-    }
+    } else schedule(owner);
     return queued;
 }
 
@@ -2484,6 +2685,7 @@ static void wire_delivered(void *context) {
     owner->delivered = true;
     pthread_cond_signal(&owner->waiting);
     pthread_mutex_unlock(&owner->lock);
+    schedule(owner);
 }
 
 /* --send OUTPUT=URL: that output's saved sends go to url, from NAME.OUTPUT. */
@@ -2649,17 +2851,112 @@ static bool wire_replies(Executor *ex, const PureModule *m, const char *const *r
 
 /* --threads auto: a connection thread for each CPU this process may run on,
  * less the one the owner of an app with states takes. */
-static unsigned auto_threads(bool owner) {
+/* An app's owner, its executor started on the app's store, its snapshot
+ * published; false (why says why) when the app cannot start. */
+static bool start_owner(App *app, Host *hosts, unsigned threads) {
+    Owner *owner = &app->owner;
+    if (!start_executor(&owner->executor, app)) return false;
+    if (pthread_mutex_init(&owner->lock, NULL) || pure_cond_monotonic(&owner->waiting) ||
+        pthread_mutex_init(&owner->executor.published_lock, NULL)) {
+        snprintf(app->why, sizeof(app->why), "Cannot make the owner's locks");
+        return false;
+    }
+    owner->executor.reads_first = true;
+    owner->executor.publishes = true;
+    owner->hosts = hosts;
+    owner->host_count = threads;
+    return true;
+}
+
+/* --apps FILE: an app a line, NAME MODULE STORE and its budgets
+ * (memory=BYTES work=N queue=N), # beginning a comment. An app whose
+ * module does not load is kept, stopped, with why; false (and why on
+ * stderr) when the file cannot be read or a line is not an app. */
+static bool read_apps(const char *path) {
+    FILE *file = fopen(path, "r");
+    if (!file) {
+        fprintf(stderr, "--apps %s: %s\n", path, strerror(errno));
+        return false;
+    }
+    char line[4096];
+    unsigned at = 0;
+    bool ok = true;
+    while (ok && fgets(line, sizeof(line), file)) {
+        ++at;
+        char *hash = strchr(line, '#'), *words[8], *rest = NULL;
+        if (hash) *hash = 0;
+        unsigned count = 0;
+        for (char *word = strtok_r(line, " \t\r\n", &rest); word && count < 8; word = strtok_r(NULL, " \t\r\n", &rest))
+            words[count++] = word;
+        if (!count) continue;
+        App *grown = count >= 3 && count <= 6 ? realloc(apps, (app_count + 1) * sizeof(*apps)) : NULL;
+        if (!grown) {
+            fprintf(stderr, "--apps %s:%u: NAME MODULE STORE [memory=BYTES] [work=N] [queue=N]\n", path, at);
+            ok = false;
+            break;
+        }
+        apps = grown;
+        App *app = &apps[app_count];
+        memset(app, 0, sizeof(*app));
+        size_t length = strlen(words[0]);
+        bool named = length <= APP_NAME;
+        for (size_t i = 0; named && i < length; ++i) {
+            char c = words[0][i];
+            named = (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '-';
+        }
+        for (unsigned i = 0; named && i < app_count; ++i) named = strcmp(apps[i].name, words[0]) != 0;
+        if (!named) {
+            fprintf(stderr, "--apps %s:%u: a name is 1-32 of a-z 0-9 _ -, each app's its own\n", path, at);
+            ok = false;
+            break;
+        }
+        memcpy(app->name, words[0], length + 1);
+        for (unsigned w = 3; ok && w < count; ++w) {
+            char *value = strchr(words[w], '=');
+            uint64_t n;
+            if (value) *value++ = 0;
+            if (value && !strcmp(words[w], "memory") && number(value, SIZE_MAX, &n)) app->memory = (size_t)n;
+            else if (value && !strcmp(words[w], "work") && number(value, UINT64_MAX, &n)) app->work = n;
+            else if (value && !strcmp(words[w], "queue") && number(value, SIZE_MAX, &n)) app->owner.queue = (size_t)n;
+            else {
+                fprintf(stderr, "--apps %s:%u: budgets are memory=BYTES, work=N and queue=N\n", path, at);
+                ok = false;
+            }
+        }
+        app->store = strdup(words[2]);
+        app->owned = true; /* every app of a host of apps takes turns on the workers */
+        if (!app->store) ok = false;
+        else ++app_count;
+        if (ok && !load_module(app, words[1])) app->module = (PureModule){0}; /* stopped, with why */
+        else if (ok) app->running = true; /* until its store says otherwise */
+    }
+    fclose(file);
+    if (ok && !app_count) fprintf(stderr, "--apps %s: no app\n", path);
+    return ok && app_count;
+}
+
+/* A JSON string's quotes and escapes around text. */
+static void print_json(FILE *out, const char *text) {
+    fputc('"', out);
+    for (const unsigned char *c = (const unsigned char *)text; *c; ++c)
+        if (*c == '"' || *c == '\\') fprintf(out, "\\%c", *c);
+        else if (*c < 0x20) fprintf(out, "\\u%04x", *c);
+        else fputc(*c, out);
+    fputc('"', out);
+}
+
+static unsigned online_cpus(void) {
     cpu_set_t set;
     long cpus = sched_getaffinity(0, sizeof(set), &set) ? sysconf(_SC_NPROCESSORS_ONLN) : CPU_COUNT(&set);
-    if (cpus < 1) cpus = 1;
-    if (owner && cpus > 1) --cpus;
-    return cpus > 256 ? 256 : (unsigned)cpus;
+    return cpus < 1 ? 1 : cpus > 256 ? 256 : (unsigned)cpus;
+}
+
+static unsigned auto_threads(bool owner) {
+    unsigned cpus = online_cpus();
+    return owner && cpus > 1 ? cpus - 1 : cpus;
 }
 
 int main(int argc, char **argv) {
-    static PureModule module;
-    static Owner owner;
     static PureSurfaceHost surface;
     /* What follows -- is the surface's: its library first, then its arguments. */
     int options = argc, surface_argc = 0;
@@ -2670,9 +2967,9 @@ int main(int argc, char **argv) {
             break;
         }
     const char *surface_path = NULL, *module_path = NULL, *store = NULL, *name = NULL, *sends[PURE_MODULE_INPUTS], *replies[PURE_MODULE_INPUTS];
-    const char *seals[PURE_MODULE_INPUTS], *named[PURE_MODULE_INPUTS], *key_path = NULL, *senders_path = NULL;
+    const char *seals[PURE_MODULE_INPUTS], *named[PURE_MODULE_INPUTS], *key_path = NULL, *senders_path = NULL, *apps_path = NULL;
     unsigned send_count = 0, reply_count = 0, seal_count = 0, named_count = 0;
-    unsigned port = 8767, threads = 1, link_port = 0;
+    unsigned port = 8767, threads = 1, link_port = 0, workers = 0;
     size_t budget = 64u << 20, capacity = 1024;
     uint64_t timeout_ms = 10000, link_idle_ms = 0;
     for (int i = 1; i < options; i += 2) {
@@ -2680,6 +2977,7 @@ int main(int argc, char **argv) {
         if (!strcmp(argv[i], "--sealed-only")) { sealed_only = true; --i; continue; }
         if (i + 1 == options) goto usage;
         if (!strcmp(argv[i], "--module")) module_path = argv[i + 1];
+        else if (!strcmp(argv[i], "--apps")) apps_path = argv[i + 1];
         else if (!strcmp(argv[i], "--surface")) surface_path = argv[i + 1];
         else if (!strcmp(argv[i], "--key")) key_path = argv[i + 1];
         else if (!strcmp(argv[i], "--seal") && seal_count < PURE_MODULE_INPUTS) seals[seal_count++] = argv[i + 1];
@@ -2700,6 +2998,7 @@ int main(int argc, char **argv) {
         else if (!strcmp(argv[i], "--connections") && number(argv[i + 1], 1u << 20, &value) && value) capacity = (size_t)value;
         else if (!strcmp(argv[i], "--threads") && !strcmp(argv[i + 1], "auto")) threads = 0;
         else if (!strcmp(argv[i], "--threads") && number(argv[i + 1], 256, &value) && value) threads = (unsigned)value;
+        else if (!strcmp(argv[i], "--workers") && number(argv[i + 1], 256, &value) && value) workers = (unsigned)value;
         else if (!strcmp(argv[i], "--timeout-ms") && number(argv[i + 1], 3600000, &value) && value) timeout_ms = value;
         else goto usage;
     }
@@ -2707,10 +3006,33 @@ int main(int argc, char **argv) {
     for (int i = options + 1; i < argc && surface_argc < 63; ++i) surface_argv[surface_argc++] = argv[i];
     surface_argv[surface_argc] = NULL;
     if (options < argc && !surface_path) goto usage;
-    uint32_t show, edit;
-    bool documents;
-    if (!module_path || !store || !load_module(&module, module_path, &documents, &show, &edit)) goto usage;
-    if (!threads) threads = auto_threads(module.states);
+    if (apps_path) { /* a host of apps: each its own module, store and budgets; none sends yet (017, stage 1) */
+        if (module_path || store || name || send_count || reply_count || seal_count || link_port || surface_path ||
+            memory_budget || work_budget) {
+            fputs("--apps gives each app its module, store and budgets; --send, --reply, --seal, --link-port, --name "
+                  "and --surface are for a host of one app (--module)\n", stderr);
+            return 2;
+        }
+        if (!read_apps(apps_path)) return 2;
+        named_apps = true;
+    } else {
+        if (!module_path || !store || !(apps = calloc(1, sizeof(*apps)))) goto usage;
+        app_count = 1;
+        apps[0].store = store;
+        apps[0].memory = memory_budget;
+        apps[0].work = work_budget;
+        if (!load_module(&apps[0], module_path)) {
+            fprintf(stderr, "%s\n", apps[0].why);
+            goto usage;
+        }
+        /* States (and outputs, whose sends are the app's state) have one
+         * owner, whose snapshots the connection threads read. */
+        bool outputs = false;
+        for (uint32_t i = 0; i < apps[0].module.inputs; ++i) outputs = outputs || apps[0].module.input[i].kind == PURE_KIND_OUTPUT;
+        apps[0].owned = apps[0].module.states || outputs;
+        apps[0].running = true;
+    }
+    if (!threads) threads = auto_threads(apps[0].owned);
     char public_key[2 * PURE_SEAL_KEY + 1] = "";
     if (key_path) {
         char error[256];
@@ -2719,68 +3041,92 @@ int main(int argc, char **argv) {
     if ((sealed_only || named_count) && !seal_key) { fputs("--sealed-only and --sender need --key\n", stderr); return 2; }
     if (!name_senders(named, named_count)) return 2;
     if (seal_key && !pure_seal_peer(public_key, own_public)) return 2;
-    if (pure_store_init(store) != PURE_OK) { fputs("Cannot initialize the store directory\n", stderr); return 1; }
     signal(SIGPIPE, SIG_IGN);
     struct sigaction stop_action = {.sa_handler = stop};
     sigemptyset(&stop_action.sa_mask);
     sigaction(SIGINT, &stop_action, NULL);
     sigaction(SIGTERM, &stop_action, NULL);
     Host *hosts = calloc(threads, sizeof(*hosts));
-    pthread_t *thread = calloc(threads, sizeof(*thread)), owner_thread;
-    if (!hosts || !thread) return 1;
-    /* States (and outputs, whose sends are the app's state) have one owner on
-     * a thread of its own, whose snapshots the connection threads read. */
-    bool outputs = false;
-    for (uint32_t i = 0; i < module.inputs; ++i) outputs = outputs || module.input[i].kind == PURE_KIND_OUTPUT;
-    bool owned = module.states || outputs;
+    pthread_t *thread = calloc(threads, sizeof(*thread));
+    if (!hosts || !thread || pure_cond_monotonic(&turns.wake)) return 1;
+    /* Each app's store, and its owner: an app that cannot start stops alone
+     * (in a host of apps; the one app of a host stops the host). Two apps
+     * never share a store. */
+    unsigned owners = 0;
+    for (unsigned a = 0; a < app_count; ++a) {
+        App *app = &apps[a];
+        if (!app->running) continue;
+        struct stat mine, theirs;
+        bool shared = false;
+        if (pure_store_init(app->store) != PURE_OK) snprintf(app->why, sizeof(app->why), "Cannot initialize the store directory");
+        else if (stat(app->store, &mine)) snprintf(app->why, sizeof(app->why), "Cannot read the store directory");
+        else
+            for (unsigned b = 0; b < a && !shared; ++b)
+                if (apps[b].running && !stat(apps[b].store, &theirs) && theirs.st_dev == mine.st_dev && theirs.st_ino == mine.st_ino) {
+                    snprintf(app->why, sizeof(app->why), "The store is %s's", apps[b].name);
+                    shared = true;
+                }
+        app->running = !app->why[0] && (!app->owned || start_owner(app, hosts, threads));
+        if (!app->running && !named_apps) {
+            fprintf(stderr, "%s\n", app->why);
+            return 1;
+        }
+        owners += app->running && app->owned;
+    }
     /* The sender's name: --name, or the module file's name without .pmod. */
     char stem[PURE_SOURCE + 1];
-    if (!name) {
+    if (!named_apps && !name) {
         const char *base = strrchr(module_path, '/') ? strrchr(module_path, '/') + 1 : module_path;
         size_t length = strlen(base);
         if (length > 5 && !strcmp(base + length - 5, ".pmod")) length -= 5;
         snprintf(stem, sizeof(stem), "%.*s", (int)length, base);
         name = stem;
     }
-    if (owned) {
-        if (!start_executor(&owner.executor, &module, show, edit, store) || pthread_mutex_init(&owner.lock, NULL) ||
-            pure_cond_monotonic(&owner.waiting) || pthread_mutex_init(&owner.executor.published_lock, NULL))
-            return 1;
-        owner.executor.reads_first = true;
-        owner.executor.publishes = true;
-        if (!open_wires(&owner.executor, &module, sends, send_count, name) ||
-            !wire_replies(&owner.executor, &module, replies, reply_count) ||
-            !wire_seals(&owner.executor, &module, seals, seal_count, seal_key ? own_public : NULL)) return 2;
-        after_save(&owner.executor); /* the snapshot, and the sends stored before */
-        for (unsigned i = 0; i < owner.executor.wire_count; ++i)
-            if (!pure_wire_start(owner.executor.wires[i], wire_delivered, &owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
-        for (unsigned i = 0; i < owner.executor.link_count; ++i) {
+    Owner *owner = &apps[0].owner; /* the one app's, in a host of one */
+    if (!named_apps && apps[0].owned &&
+        (!open_wires(&owner->executor, &apps[0].module, sends, send_count, name) ||
+         !wire_replies(&owner->executor, &apps[0].module, replies, reply_count) ||
+         !wire_seals(&owner->executor, &apps[0].module, seals, seal_count, seal_key ? own_public : NULL))) return 2;
+    for (unsigned a = 0; a < app_count; ++a)
+        if (apps[a].running && apps[a].owned) after_save(&apps[a].owner.executor); /* the snapshot, and the sends stored before */
+    if (!named_apps && apps[0].owned) {
+        for (unsigned i = 0; i < owner->executor.wire_count; ++i)
+            if (!pure_wire_start(owner->executor.wires[i], wire_delivered, owner)) { fputs("Cannot start a wire\n", stderr); return 1; }
+        for (unsigned i = 0; i < owner->executor.link_count; ++i) {
             char error[256];
-            if (!pure_link_start(owner.executor.links[i], wire_delivered, &owner, error, sizeof(error))) {
+            if (!pure_link_start(owner->executor.links[i], wire_delivered, owner, error, sizeof(error))) {
                 fprintf(stderr, "--send %s\n", error);
                 return 2;
             }
         }
-        owner.hosts = hosts;
-        owner.host_count = threads;
     }
     for (unsigned i = 0; i < threads; ++i) {
         hosts[i].capacity = capacity;
         hosts[i].timeout_ms = timeout_ms;
         hosts[i].listener = hosts[i].epoll = -1;
-        hosts[i].owner = owned ? &owner : NULL;
-        if (!start_shard(&hosts[i], &module, documents, show, edit, store, budget, &port)) return 1;
-    }
-    if (owned && pthread_create(&owner_thread, NULL, run_owner, &owner)) { fputs("Cannot start the owner\n", stderr); return 1; }
-    if (owned) pthread_setname_np(owner_thread, "pure-owner"); /* its CPU, seen apart: one core at most */
+        if (!start_shard(&hosts[i], budget, &port)) return 1;
+    }
+    /* The workers: an app's owner takes its turns on them, none its own
+     * thread. One for the one app of a host; for apps, as many as run, at
+     * most a CPU each. */
+    if (!workers) workers = named_apps ? (owners < online_cpus() ? owners : online_cpus()) : 1;
+    if (!workers) workers = 1;
+    pthread_t *worker = calloc(workers, sizeof(*worker));
+    if (!worker) return 1;
+    for (unsigned i = 0; owners && i < workers; ++i) {
+        if (pthread_create(&worker[i], NULL, run_worker, NULL)) { fputs("Cannot start a worker\n", stderr); return 1; }
+        pthread_setname_np(worker[i], "pure-owner"); /* their CPU, seen apart */
+    }
+    for (unsigned a = 0; a < app_count; ++a)
+        if (apps[a].running && apps[a].owned) schedule(&apps[a].owner); /* a first turn: its timers begin */
     if (link_port) { /* records from the senders named, over UDP */
         char error[256];
-        if (!owned || !seal_key) {
+        if (named_apps || !apps[0].owned || !seal_key) {
             fputs("--link-port needs an app with states and the node's --key\n", stderr);
             return 2;
         }
         PureLinkReceiver *receiver =
-            pure_link_receiver(seal_key, own_public, &(PureLinkHost){link_input, link_deliver, &owner}, error, sizeof(error));
+            pure_link_receiver(seal_key, own_public, &(PureLinkHost){link_input, link_deliver, owner}, error, sizeof(error));
         bool allowed = receiver != NULL && allow_senders(receiver, senders_path, error, sizeof(error));
         for (unsigned i = 0; allowed && i < sender_count; ++i)
             allowed = pure_link_allow(receiver, senders[i].name, senders[i].key, error, sizeof(error));
@@ -2794,15 +3140,29 @@ int main(int argc, char **argv) {
             return 2;
         }
     }
-    /* Its name too: its sends' sources are NAME.OUTPUT, which a receiver's --sender must name. */
-    printf("{\"port\":%u,\"threads\":%u,\"name\":\"", port, threads);
-    for (const char *c = name; *c; ++c) printf(*c == '"' || *c == '\\' ? "\\%c" : "%c", *c);
-    if (*public_key) printf("\",\"public_key\":\"%s\"}\n", public_key);
-    else printf("\"}\n");
+    if (named_apps) { /* each app, and why one stopped */
+        printf("{\"port\":%u,\"threads\":%u,\"workers\":%u,\"apps\":[", port, threads, owners ? workers : 0);
+        for (unsigned a = 0; a < app_count; ++a) {
+            printf("%s{\"name\":", a ? "," : "");
+            print_json(stdout, apps[a].name);
+            printf(",\"running\":%s", apps[a].running ? "true" : "false");
+            if (!apps[a].running) {
+                printf(",\"why\":");
+                print_json(stdout, apps[a].why);
+            }
+            printf("}");
+        }
+        printf("]%s%s%s}\n", *public_key ? ",\"public_key\":\"" : "", public_key, *public_key ? "\"" : "");
+    } else { /* its name too: its sends' sources are NAME.OUTPUT, which a receiver's --sender must name */
+        printf("{\"port\":%u,\"threads\":%u,\"name\":\"", port, threads);
+        for (const char *c = name; *c; ++c) printf(*c == '"' || *c == '\\' ? "\\%c" : "%c", *c);
+        if (*public_key) printf("\",\"public_key\":\"%s\"}\n", public_key);
+        else printf("\"}\n");
+    }
     fflush(stdout);
     /* A surface reads the owner's snapshots and gives it events. */
-    if (surface_path && !owned) { fputs("--surface needs an app with states\n", stderr); return 2; }
-    if (surface_path && !start_surface(&surface, &owner, surface_argc, surface_argv)) return 1;
+    if (surface_path && (named_apps || !apps[0].owned)) { fputs("--surface needs an app with states\n", stderr); return 2; }
+    if (surface_path && !start_surface(&surface, owner, surface_argc, surface_argv)) return 1;
     for (unsigned i = 1; i < threads; ++i)
         if (pthread_create(&thread[i], NULL, run_shard, &hosts[i])) { fputs("Cannot start a thread\n", stderr); return 1; }
     run_shard(&hosts[0]);
@@ -2811,19 +3171,24 @@ int main(int argc, char **argv) {
         surface_changed(&surface);
         pthread_join(surface.thread, NULL);
     }
-    if (owned) {
-        pthread_mutex_lock(&owner.lock);
-        owner.stopping = true;
-        pthread_cond_signal(&owner.waiting);
-        pthread_mutex_unlock(&owner.lock);
-        pthread_join(owner_thread, NULL);
-    }
+    for (unsigned a = 0; a < app_count; ++a) { /* no gathering from now on */
+        Owner *o = &apps[a].owner;
+        if (!apps[a].running || !apps[a].owned) continue;
+        pthread_mutex_lock(&o->lock);
+        o->stopping = true;
+        pthread_cond_signal(&o->waiting);
+        pthread_mutex_unlock(&o->lock);
+    }
+    pthread_mutex_lock(&turns.lock);
+    turns.stopping = true;
+    pthread_cond_broadcast(&turns.wake);
+    pthread_mutex_unlock(&turns.lock);
+    for (unsigned i = 0; owners && i < workers; ++i) pthread_join(worker[i], NULL);
     PureLinkCounts sent = {0}, heard = {0}; /* what links sent, and what the listener took */
     pure_link_listener_close(link_listener, &heard); /* after the owner, which tells it what it stored */
-    size_t active = 0, retained = 0;
-    uint64_t accepted = 0, requests = 0, bursts = owner.executor.bursts, events = owner.executor.burst_events, expired = 0,
-             ticks = owner.executor.ticks, shown = 0, checkpoints = 0, memory_refused = owner.executor.memory_refused,
-             windows_full = owner.executor.windows_full;
+    size_t active = 0, retained = 0, state_bytes = 0;
+    uint64_t accepted = 0, requests = 0, bursts = 0, events = 0, expired = 0, ticks = 0, shown = 0, checkpoints = 0,
+             memory_refused = 0, windows_full = 0, publications = 0, answers = 0, delivered = 0, attempts = 0;
     for (unsigned i = 0; i < threads; ++i) {
         shown += hosts[i].shown;
         active += hosts[i].active;
@@ -2837,28 +3202,36 @@ int main(int argc, char **argv) {
         expired += hosts[i].expired;
         stop_shard(&hosts[i]);
     }
-    uint64_t delivered = 0, attempts = 0;
-    for (unsigned i = 0; i < owner.executor.wire_count; ++i) {
-        uint64_t d, a;
-        pure_wire_counts(owner.executor.wires[i], &d, &a);
-        delivered += d;
-        attempts += a;
-        pure_wire_close(owner.executor.wires[i]); /* what was not delivered stays in the outbox */
-    }
-    for (unsigned i = 0; i < owner.executor.link_count; ++i) {
-        PureLinkCounts c;
-        pure_link_counts(owner.executor.links[i], &c);
-        sent.packets += c.packets, sent.records += c.records, sent.resent += c.resent, sent.lost += c.lost;
-        sent.acknowledged += c.acknowledged, sent.received += c.received, sent.stored += c.stored;
-        sent.refused += c.refused;
-        pure_link_close(owner.executor.links[i]); /* what was not stored stays in the outbox */
-    }
-    size_t state_bytes = owned ? pure_app_state_bytes(&owner.executor.app) : 0;
-    if (owned) {
-        checkpoints = owner.executor.checkpoints;
-        stop_executor(&owner.executor);
-        checkpoints = owner.executor.checkpoints;
-        free(owner.jobs);
+    for (unsigned a = 0; a < app_count; ++a) {
+        Executor *ex = &apps[a].owner.executor;
+        if (!apps[a].running || !apps[a].owned) continue;
+        bursts += ex->bursts;
+        events += ex->burst_events;
+        ticks += ex->ticks;
+        memory_refused += ex->memory_refused;
+        windows_full += ex->windows_full;
+        publications += ex->publications;
+        answers += ex->answers;
+        for (unsigned i = 0; i < ex->wire_count; ++i) {
+            uint64_t d, t;
+            pure_wire_counts(ex->wires[i], &d, &t);
+            delivered += d;
+            attempts += t;
+            pure_wire_close(ex->wires[i]); /* what was not delivered stays in the outbox */
+        }
+        for (unsigned i = 0; i < ex->link_count; ++i) {
+            PureLinkCounts c;
+            pure_link_counts(ex->links[i], &c);
+            sent.packets += c.packets, sent.records += c.records, sent.resent += c.resent, sent.lost += c.lost;
+            sent.acknowledged += c.acknowledged, sent.received += c.received, sent.stored += c.stored;
+            sent.refused += c.refused;
+            pure_link_close(ex->links[i]); /* what was not stored stays in the outbox */
+        }
+        state_bytes += pure_app_state_bytes(&ex->app);
+        stop_executor(ex);
+        checkpoints += ex->checkpoints;
+        free(apps[a].owner.jobs);
+        free(apps[a].owner.spare);
     }
     unsigned long groups, grouped;
     pure_store_group_counts(&groups, &grouped);
@@ -2873,15 +3246,29 @@ int main(int argc, char **argv) {
             ",\"listener_deliveries\":%" PRIu64 ",\"listener_woken\":%" PRIu64 ",\"listener_remembered\":%" PRIu64
             ",\"listener_swept\":%" PRIu64 ",\"listener_active\":%" PRIu64 ",\"listener_keyed\":%" PRIu64
             ",\"listener_secrets\":%" PRIu64 ",\"listener_epochs\":%" PRIu64 ",\"listener_refused\":%" PRIu64
-            ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 ",\"windows_full\":%" PRIu64 "}\n",
-            active, retained, accepted, requests, bursts, events, ticks, expired, threads, owned ? "true" : "false",
-            owner.executor.publications, shown, delivered, attempts, owner.executor.answers, groups, grouped,
+            ",\"state_bytes\":%zu,\"memory_refused\":%" PRIu64 ",\"windows_full\":%" PRIu64,
+            active, retained, accepted, requests, bursts, events, ticks, expired, threads, apps[0].owned ? "true" : "false",
+            publications, shown, delivered, attempts, answers, groups, grouped,
             checkpoints, sent.packets, sent.records, sent.resent, sent.lost, sent.acknowledged, sent.stored,
             sent.refused, heard.packets, heard.received, heard.deliveries, heard.woken, heard.remembered, heard.swept,
             heard.active, heard.keyed, heard.secrets, heard.epochs, heard.refused, state_bytes, memory_refused,
             windows_full);
+    if (named_apps) { /* what each app did */
+        fprintf(stderr, ",\"workers\":%u,\"apps\":[", owners ? workers : 0);
+        for (unsigned a = 0; a < app_count; ++a) {
+            const Owner *o = &apps[a].owner;
+            fprintf(stderr, "%s{\"name\":", a ? "," : "");
+            print_json(stderr, apps[a].name);
+            fprintf(stderr, ",\"running\":%s,\"turns\":%" PRIu64 ",\"bursts\":%" PRIu64 ",\"events\":%" PRIu64
+                    ",\"busy\":%" PRIu64 ",\"memory_refused\":%" PRIu64 "}", apps[a].running ? "true" : "false",
+                    o->turns, o->executor.bursts, o->executor.burst_events, o->busy, o->executor.memory_refused);
+        }
+        fputc(']', stderr);
+    }
+    fputs("}\n", stderr);
     free(hosts);
     free(thread);
+    free(worker);
     return 0;
 usage:
     fputs("pure-serve --module FILE.pmod --store DIRECTORY [--port 8767] [--threads 1|N|auto] [--budget BYTES] "
@@ -2889,6 +3276,11 @@ usage:
           "[--send-ca FILE] [--reply OUTPUT=INPUT ...] [--seal OUTPUT=KEY ...] [--key FILE] [--sealed-only] "
           "[--sender NAME=KEY ...] [--link-port PORT] [--senders FILE] [--link-idle-ms 30000] [--memory BYTES] [--work N] "
           "[--window 4194304] [--surface LIBRARY [-- ARGUMENTS]]\n"
+          "pure-serve --apps FILE [--port 8767] [--threads 1|N|auto] [--workers N] [--budget BYTES] [--connections 1024] "
+          "[--timeout-ms 10000]\n"
+          "  --apps FILE: a host of apps (docs/design/017-apps-on-one-host.md), a line each:\n"
+          "      NAME MODULE.pmod STORE [memory=BYTES] [work=N] [queue=N]; its routes start /NAME/\n"
+          "  --workers N: the threads the apps' owners take turns on (1 for one app; for apps, one a CPU at most)\n"
           "  --send OUTPUT=pure://host:port/INPUT: to a Pure node's --link-port, the way between nodes (UDP, no\n"
           "      connection, docs/design/012-link-protocol.md; its key by --seal OUTPUT=KEY)\n"
           "  --send OUTPUT=http[s]://host:port/path: to anything else, over HTTP\n"
diff --git a/tests/check.py b/tests/check.py
index 47ddf23..fdb4f56 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -3726,6 +3726,122 @@ def server_work(aup, pure, work):
     return first
 
 
+APPS_SOURCE = """record Item {
+  n = 0
+}
+state n = 0
+state items: list of Item
+input add = message { step: number }
+input fill = message { k: number }
+input walk = message { step: number }
+on add {
+  n = n + add.step
+}
+on fill {
+  items.append(Item { n: fill.k })
+}
+on walk {
+  for i in items {
+    i.n = i.n + walk.step
+  }
+}
+view total = n
+view count = count(i in items where i.n > 0)
+"""
+
+
+def server_apps(aup, pure, work):
+    """One host, many apps (docs/design/017): each app's routes start with its
+    name and reach its own states (two apps of one module share nothing); an
+    app whose module does not load, or whose store is another app's, stops
+    alone (503 stopped) while the others run; past an app's queue budget its
+    requests are refused (503 busy) and another app answers meanwhile; past
+    its work budget an event is refused (422 work) where the same event runs
+    in an app without one; an app whose store another process holds cannot
+    save (503 store_busy, nothing applied) while the others save; and after a
+    restart each app has its states back."""
+    import fcntl
+    import socket
+    source = work / 'apps.aup'
+    source.write_text(APPS_SOURCE)
+    compile_module(aup, source, work / 'apps.pmod')
+    (work / 'broken.pmod').write_bytes(b'PUREMOD6 not a module')
+    manifest = work / 'apps.conf'
+    manifest.write_text(f"""# name module store budgets
+a      {work / 'apps.pmod'}   {work / 'store-a'}
+b      {work / 'apps.pmod'}   {work / 'store-b'}
+busy   {work / 'apps.pmod'}   {work / 'store-busy'}   queue=2
+tight  {work / 'apps.pmod'}   {work / 'store-tight'}  work=55
+broken {work / 'broken.pmod'} {work / 'store-broken'}
+twin   {work / 'apps.pmod'}   {work / 'store-a'}
+""")
+    serve = pure / 'build/pure-serve'
+
+    def get(port, path):
+        status, why, _, body = http(port, 'GET', path)
+        return status, why, body
+
+    def post(port, path):
+        status, why, _, _ = http(port, 'POST', path)
+        return status, why
+
+    host = HttpHost([serve, '--apps', manifest, '--workers', 2])
+    try:
+        running = {app['name']: app for app in host.info['apps']}
+        assert [name for name, app in running.items() if app['running']] == ['a', 'b', 'busy', 'tight'], running
+        assert running['broken']['why'].startswith('Module:') and running['twin']['why'] == "The store is a's", running
+        port = host.port
+        assert post(port, '/a/inputs/add?step=2') == (200, 'none')
+        assert get(port, '/a/views/total')[::2] == (200, b'2') and get(port, '/b/views/total')[::2] == (200, b'0')
+        assert get(port, '/broken/views/total')[:2] == (503, 'stopped') and post(port, '/twin/inputs/add?step=1')[0] == 503
+        assert get(port, '/nobody/views/total')[:2] == (404, 'no_app') and get(port, '/views/total')[:2] == (404, 'no_app')
+        # Work: ten items to walk; the tight app refuses the walk and changes nothing, a's runs it.
+        for app in ('a', 'tight'):
+            for k in range(10):
+                assert post(port, f'/{app}/inputs/fill?k={k}') == (200, 'none')
+        assert post(port, '/tight/inputs/walk?step=1') == (422, 'work') and get(port, '/tight/views/count')[::2] == (200, b'9')
+        assert post(port, '/a/inputs/walk?step=1') == (200, 'none') and get(port, '/a/views/count')[::2] == (200, b'10')
+        # Queue: requests at once, from as many connections; b answers meanwhile.
+        flood = []
+        for _ in range(120):
+            connection = socket.create_connection(('127.0.0.1', port))
+            flood.append(connection)
+        for connection in flood:
+            connection.sendall(b'POST /busy/inputs/add?step=1 HTTP/1.1\r\nHost: x\r\nContent-Length: 0\r\n\r\n')
+        assert post(port, '/b/inputs/add?step=5') == (200, 'none')
+        answers = []
+        for connection in flood:
+            head = b''
+            while b'\r\n\r\n' not in head:
+                head += connection.recv(4096)
+            answers.append(head.split(b' ', 2)[1])
+            connection.close()
+        ok, busy = answers.count(b'200'), answers.count(b'503')
+        assert ok + busy == len(flood) and busy > 0 and ok > 0, (ok, busy)
+        assert get(port, '/busy/views/total')[::2] == (200, str(ok).encode())  # the refused changed nothing
+        # The store of b held by another process: b cannot save, a can.
+        with open(work / 'store-b' / '.writer.lock', 'r+b') as held:
+            fcntl.lockf(held, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            assert post(port, '/b/inputs/add?step=1') == (503, 'store_busy')
+            assert post(port, '/a/inputs/add?step=1') == (200, 'none')
+            assert get(port, '/b/views/total')[::2] == (200, b'5')
+            fcntl.lockf(held, fcntl.LOCK_UN)
+        assert post(port, '/b/inputs/add?step=1') == (200, 'none')
+    finally:
+        stats = host.stop()[-1]
+    by_name = {app['name']: app for app in stats['apps']}
+    assert by_name['busy']['busy'] == busy and by_name['a']['events'] > 0 and stats['workers'] == 2, stats['apps']
+    # Restarted: each app's states are back.
+    host = HttpHost([serve, '--apps', manifest])
+    try:
+        assert get(host.port, '/a/views/total')[::2] == (200, b'3') and get(host.port, '/b/views/total')[::2] == (200, b'6')
+        assert get(host.port, '/a/views/count')[::2] == (200, b'10') and get(host.port, '/busy/views/total')[::2] == \
+            (200, str(ok).encode())
+    finally:
+        host.stop()
+    return {'apps': len(running), 'stopped': 2, 'flooded': len(flood), 'busy': busy}
+
+
 def main():
     global EDITOR
     parser = argparse.ArgumentParser()
@@ -3775,6 +3891,7 @@ def main():
         report['server_answers_memory'] = server_answers_memory(aup, pure, work)
         report['server_meaning'] = server_meaning(aup, pure, work)
         report['server_work'] = server_work(aup, pure, work)
+        report['server_apps'] = server_apps(aup, pure, work)
     print(json.dumps(report))