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.

5 files changed, 45 insertions, 13 deletions

diff --git a/CHECKPOINT.md b/CHECKPOINT.md
index 53c306f..62bd853 100644
--- a/CHECKPOINT.md
+++ b/CHECKPOINT.md
@@ -18,7 +18,7 @@
 | HTTP 서비스 | 보기가 GET, 입력이 POST다(JSON 본문과 그 안의 그룹 객체를 제자리에서 읽는다, 라우터 코드 없음). 앱의 `send`는 이벤트와 함께 저장되고, `--send 출력=URL`로 다른 앱이나 HTTP 끝점에 순서대로 묶어 전해지며, 받는 앱은 번호마다 한 번 적용한다(서버를 죽였다 살려도 정확히 한 번). 답은 저장 뒤에 나간다. 상태를 가진 앱은 소유자 스레드 하나가 입력을 적용하고, 연결 스레드는 마지막 저장의 스냅숏으로 보기를 답한다(쓰기가 동기화하는 동안에도 읽기는 기다리지 않는다) | 008 HTTP 호스트, [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [082](docs/research/082-json-in-place.md), [083](docs/research/083-delivery-over-http.md) |
 | 성능(Pi 5) | 문서 서버: 작은 GET 151.6k/s, 1 MiB GET 2,794/s(axum 1,372, drogon 708), 충돌 안전 PUT은 axum-atomic의 5.1–5.8배. 앱 서비스(aup 22줄 대 손으로 쓴 서비스): 내구성 있는 쓰기 약 76배(그 뒤 저장소 커밋을 한 구간으로 바꿔 64연결 21–26k/s, axum·drogon 내구 약 180/s), 20행 JSON GET 145k/s(axum 95k, drogon 144k), 쓰기와 섞인 읽기 149k/s·p99 0.6 ms(axum·drogon 내구 1.1–1.3k/s·p99 111–156 ms), 최대 RSS 3.0–4.1 MiB(axum 6.4, drogon 11.6) | [075](docs/research/075-http-host-against-axum.md), [077](docs/research/077-apps-as-http-services.md), [080](docs/research/080-snapshot-reads.md), [081](docs/research/081-tail-records.md) |
 
-남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 밀린 outbox를 저장마다 다 훑지 않기(083 다음 결정), 앱 라우트의 인증·TLS(문서
+남은 것: macOS 검증, `bool`·`i64`, 꼬리 늘리기·압축을 커밋 밖으로와 열린 부하에서의 저장·적용 겹치기(081 다음 결정), 응답을 기다리는 출력의 묶음 전달(083), 앱 라우트의 인증·TLS(문서
 피어의 TLS는 있다), aup·Vega의 배포 묶음.
 
 ## 공개할 작업물과 완료 증거
diff --git a/docs/research/083-delivery-over-http.md b/docs/research/083-delivery-over-http.md
index 829c666..eedc5f8 100644
--- a/docs/research/083-delivery-over-http.md
+++ b/docs/research/083-delivery-over-http.md
@@ -90,7 +90,7 @@ Raspberry Pi 5. 받는 서버는 CPU 0–1(연결 스레드 2), 보내는 서버
 
 ## 다음 결정
 
-- 저장이 끝날 때마다 outbox 전체를 훑어 새 보냄을 찾는다. 받는 쪽이 오래 꺼져 수만 건이 밀리면 저장마다 그만큼 훑는다.
-  마지막으로 넣은 자리에서 이어 훑게 해야 한다.
+- 저장이 끝날 때마다 outbox를 가장 새것부터 거꾸로 훑어, 배선된 출력마다 이미 줄에 넣은 번호에 닿으면 멈춘다
+  (`pure_app_outbox_newest`). 받는 쪽이 오래 꺼져 수만 건이 밀려도 저장마다 새 보냄만큼만 본다.
 - 덜어 낸 것은 다음 저장에 지워진다. 요청이 없는 앱은 저장하지 않으므로, 다시 시작하면 전한 것을 한 번 더 보낸다(받는
   쪽이 걸러 낸다).
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