Deliver an app's saved sends over HTTP, in order and in batches (083)

pure-serve --send OUTPUT=http://host:port/path wires an output: after
every save the owner queues the new sends as JSON (fields and groups)
and a thread per output posts them in order with X-Pure-Source,
X-Pure-Sequence and Idempotency-Key. A 2xx, or a Pure receiver's
refusal, delivers; the owner then trims. 408, 429, 5xx and no answer
are sent again after a pause growing to 10 s, other answers every 10 s:
nothing is dropped. A Pure receiver applies a delivered message with
pure_app_deliver (now taking placed fields) and advertises
X-Pure-Batch: 256; the sender then posts up to that many numbers in a
row as one JSON array (X-Pure-Count), applied in order in one burst
with one save.

Delivery kept up with orders once batched (391 deliveries/s one at a
time; a backlog of 20,000 went at 60k/s). With four clients ordering,
killing the receiver and then the sender (-9) left the receiver's count
and total equal to what the sender saved, five times and under ASan.

1 files changed, 3 insertions, 2 deletions

diff --git a/prototypes/connections/module.c b/prototypes/connections/module.c
index ee72494..5cb7f11 100644
--- a/prototypes/connections/module.c
+++ b/prototypes/connections/module.c
@@ -1676,7 +1676,8 @@ void pure_app_trim(PureApp *app, uint32_t output, uint64_t through) {
         }
 }
 
-PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload) {
+PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, uint64_t seq, PureSpan payload,
+                            const uint32_t *placed) {
     const PureModule *m = app->module;
     size_t length = source ? strlen(source) : 0;
     app->duplicate = false;
@@ -1700,7 +1701,7 @@ PureResult pure_app_deliver(PureApp *app, uint32_t input, const char *source, ui
     app->delivering.source = source;
     app->delivering.seq = seq;
     app->delivering.pending = true;
-    PureResult result = message(app, input, NULL, -1, payload, NULL);
+    PureResult result = message(app, input, NULL, -1, payload, placed);
     if (!app->delivering.pending) return result;
     /* Not applied. Refused by the app (a fault), the message was delivered
      * all the same: its number is heard, saved like a key event's changes. */