Look for new sends newest first, only as far back as they go

After each save the owner walked the whole outbox to find the sends
not yet queued, so a receiver long down made every save walk the
backlog. pure_app_outbox_newest visits saved sends newest first, and
the owner stops once every wired output reaches the number it queued
last, then queues what it found in order.

3 files changed, 42 insertions, 10 deletions

diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index 5cb7f11..697e918 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -1668,6 +1668,13 @@ void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) {
     }
 }
 
+void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) {
+    for (size_t i = app->sent_durable; i-- > 0;) {
+        const struct PureSent *sent = &app->sent[i];
+        if (!sent->trimmed && !visit(context, sent->output, sent->seq, (PureSpan){sent->payload, sent->size})) return;
+    }
+}
+
 void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) {
     for (size_t i = 0; i < app->sent_durable; ++i)
         if (app->sent[i].output == output && app->sent[i].seq <= through && !app->sent[i].trimmed) {
diff --git a/prototypes/connections/module.h b/prototypes/connections/module.h
index 3441c03..f62326b 100644
--- a/prototypes/connections/module.h
+++ b/prototypes/connections/module.h
@@ -309,6 +309,8 @@ PureResult pure_app_replace(PureApp *, const PureModule *);
  * and trims what its receiver has saved; a receiver applies each once. */
 typedef bool (*PureSentVisit)(void *context, uint32_t output, uint64_t seq, PureSpan payload);
 void pure_app_outbox(const PureApp *, PureSentVisit visit, void *context);
+/* The same, newest first (a host looking only for what is new stops early). */
+void pure_app_outbox_newest(const PureApp *, PureSentVisit visit, void *context);
 /* Forget output's saved sends numbered up to through (saved with the next save). */
 void pure_app_trim(PureApp *, uint32_t output, uint64_t through);
 /* Deliver a message event from source (a sender's name, 1 to 64 printable
diff --git a/prototypes/connections/serve.c b/prototypes/connections/serve.c
index 60771e1..2ea7adb 100644
--- a/prototypes/connections/serve.c
+++ b/prototypes/connections/serve.c
@@ -1033,23 +1033,46 @@ static void publish(Executor *ex) {
     ++ex->publications;
 }
 
-/* The saved sends of wired outputs not queued yet go to their wires, in order;
- * a wire that cannot take one takes the rest at a later save. */
-typedef struct { Executor *executor; bool blocked[PURE_MODULE_INPUTS]; } Queueing;
-
-static bool queue_send(void *context, uint32_t output, uint64_t seq, PureSpan payload) {
+/* The saved sends of wired outputs not queued yet go to their wires, in
+ * order; a wire that cannot take one takes the rest at a later save. The
+ * outbox is looked at newest first, only as far back as the new sends go,
+ * however many older ones wait for their receivers. */
+typedef struct { uint32_t wire; uint64_t seq; PureSpan payload; } Fresh;
+typedef struct { Executor *executor; Fresh *fresh; size_t count, capacity; unsigned done; bool failed; } Queueing;
+
+static bool find_fresh(void *context, uint32_t output, uint64_t seq, PureSpan payload) {
     Queueing *q = context;
     Executor *ex = q->executor;
-    for (unsigned i = 0; i < ex->wire_count; ++i)
-        if (pure_wire_output(ex->wires[i]) == output && !q->blocked[i] && seq > pure_wire_queued(ex->wires[i]))
-            q->blocked[i] = !pure_wire_queue(ex->wires[i], seq, payload);
-    return true;
+    for (unsigned i = 0; i < ex->wire_count; ++i) {
+        if (pure_wire_output(ex->wires[i]) != output || (q->done >> i & 1)) continue;
+        if (seq <= pure_wire_queued(ex->wires[i])) {
+            q->done |= 1u << i; /* older ones were queued before */
+            break;
+        }
+        if (q->count == q->capacity) {
+            size_t capacity = q->capacity ? 2 * q->capacity : 64;
+            Fresh *grown = realloc(q->fresh, capacity * sizeof(*grown));
+            if (!grown) { q->failed = true; return false; }
+            q->fresh = grown;
+            q->capacity = capacity;
+        }
+        q->fresh[q->count++] = (Fresh){i, seq, payload};
+        break;
+    }
+    return q->done != (ex->wire_count >= 32 ? UINT32_MAX : (1u << ex->wire_count) - 1);
 }
 
 static void after_save(Executor *ex) {
     publish(ex);
+    if (!ex->wire_count) return;
     Queueing q = {.executor = ex};
-    if (ex->wire_count) pure_app_outbox(&ex->app, queue_send, &q);
+    pure_app_outbox_newest(&ex->app, find_fresh, &q);
+    bool blocked[PURE_MODULE_INPUTS] = {false};
+    for (size_t k = q.failed ? 0 : q.count; k-- > 0;) {
+        const Fresh *f = &q.fresh[k];
+        if (!blocked[f->wire]) blocked[f->wire] = !pure_wire_queue(ex->wires[f->wire], f->seq, f->payload);
+    }
+    free(q.fresh);
 }
 
 /* The answers the wires made become events, in order; then what the wires