Hold a window of each output's sends in its transport, not the outbox

A link's ring and a wire's queue took every saved send of their output
and kept it until delivered: with the receiver down they held the
outbox a second time, 178 B and 313 B more a send (104). Each outlet now
holds a window, --window bytes (4 MiB) and 65,536 sends past what its
receiver has, at least one; the outbox keeps the rest.

The owner remembers each outlet's place in the outbox and looks on from
there after a save and after its transports deliver. Places move back by
the sends that left the outbox since (PureApp.sends_gone), so a trim
before a place skips nothing, and sends queued before are not looked at
again however many wait. pure_app_sent reads a send by its place (the
newest-first visit it replaces is gone); pure_link_held and
pure_wire_held say what a transport holds.

With the receiver down a node now grows by the outbox alone once the
window is full: 219 and 218 B a send over the link and HTTP (399 and
534 B before). check.py server_windows keeps the receivers of a sender's
two outputs, one linked and one over HTTP, down through 300 buys with a
2,000-byte window, then starts them: each applies all 300 once. Without
the refill after deliveries the link stops at 117.

2 files changed, 7 insertions, 13 deletions

diff --git a/core/module.c b/core/module.c
index ac46fb3..b370f4b 100644
--- a/core/module.c
+++ b/core/module.c
@@ -2711,6 +2711,7 @@ static PureResult let_sends_go(PureApp *app, uint32_t output, uint64_t through)
         PureResult removed = pure_list_remove(app->outbox, i, &app->undo);
         if (removed != PURE_OK) return removed;
         --app->sent_saved;
+        ++app->sends_gone;
         app->links_changed = true;
     }
     return PURE_OK;
@@ -2957,7 +2958,7 @@ PureResult pure_app_message_fields(PureApp *app, uint32_t input, const char *key
 }
 
 /* The outbox's item at: its output (the module's, by name), number and payload. */
-static bool sent_at(const PureApp *app, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload) {
+bool pure_app_sent(const PureApp *app, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload) {
     PureCell name = pure_list_get(app->outbox, at, 0), bytes = pure_list_get(app->outbox, at, 2);
     const PureModule *m = app->module;
     uint32_t i = 0;
@@ -2973,16 +2974,7 @@ void pure_app_outbox(const PureApp *app, PureSentVisit visit, void *context) {
         uint32_t output;
         uint64_t seq;
         PureSpan payload;
-        if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
-    }
-}
-
-void pure_app_outbox_newest(const PureApp *app, PureSentVisit visit, void *context) {
-    for (size_t i = app->sent_saved; i-- > 0;) {
-        uint32_t output;
-        uint64_t seq;
-        PureSpan payload;
-        if (sent_at(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
+        if (pure_app_sent(app, i, &output, &seq, &payload) && !visit(context, output, seq, payload)) return;
     }
 }
 
diff --git a/core/module.h b/core/module.h
index 69acd67..7034b0e 100644
--- a/core/module.h
+++ b/core/module.h
@@ -340,6 +340,7 @@ typedef struct {
      * outputs a module dropped. */
     PureList *outbox;
     size_t sent_saved, durable_sent_saved, durable_outbox;
+    uint64_t sends_gone; /* how many ever left it: a host's places in it move back at most that many */
     uint64_t last_sent[PURE_MODULE_INPUTS], saved_sent[PURE_MODULE_INPUTS];
     struct PureCarried { char name[PURE_MODULE_NAME]; uint64_t last; } *carried;
     unsigned carried_count;
@@ -438,8 +439,9 @@ 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);
+/* The saved send at a place in the outbox (0 the oldest, below sent_saved);
+ * false when its output is not the module's. */
+bool pure_app_sent(const PureApp *, size_t at, uint32_t *output, uint64_t *seq, PureSpan *payload);
 /* 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