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.

1 files changed, 5 insertions, 1 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 7940d90..12f9a24 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -251,7 +251,7 @@ struct PureLinkSender {
     /* Their bytes, one after another as queued (a record's never cut by the
      * ring's end): taken back in order as they are stored. */
     unsigned char *ring;
-    size_t ring_capacity;
+    size_t ring_capacity, held; /* held: the bytes of the records, together */
     uint64_t queued, stored; /* the last queued; saved by the receiver */
     uint64_t run, next_pn, in_run;
     PureAead *out, *in; /* made once the receiver's key is known */
@@ -461,6 +461,7 @@ bool pure_link_sender_queue(PureLinkSender *l, uint64_t seq, const unsigned char
     if (!ok) return false;
     *record_at(l, l->count++) = r;
     l->queued = seq;
+    l->held += size;
     return true;
 }
 
@@ -699,6 +700,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
             Record *r = record_at(l, 0);
             for (uint32_t i = 0; i < r->parts; ++i) l->received -= part_of(r, i)->state == RECEIVED;
             free(r->more); /* its bytes go back with the ring's head */
+            l->held -= r->size;
             l->head = (l->head + 1) & (l->capacity - 1);
             --l->count;
         }
@@ -1190,6 +1192,8 @@ size_t pure_link_sender_datagram(const PureLinkSender *l) { return l->datagram;
 
 uint64_t pure_link_sender_stored(const PureLinkSender *l) { return l->stored; }
 
+size_t pure_link_sender_held(const PureLinkSender *l) { return l->held; }
+
 void pure_link_sender_counts(const PureLinkSender *l, PureLinkCounts *out) {
     *out = l->counts;
     out->stored = l->stored;