Let a receiver whose app is full keep what came, and say so

A link's delivery the app had no room for was handed over again at the
next turn: a receiver over its memory budget would refuse the same
records as fast as the owner could apply them. And its sender, seeing
records received but not stored for a second, sent the whole window
again every second.

Now a host says a delivery found it full (PureLinkDone.full). The
receiver keeps the records and hands them over again after a pause, 100
ms doubling to 500 ms, and meanwhile its STORED frames carry flag 2: the
app is full and what came is kept. A sender hearing it starts its
one-second wait for the save again, and sends none of it again. The
spec (012) says both sides; the pause is a local choice that must stay
under the sender's wait.

pure_link_done takes the PureLinkDone itself. test-link's new case keeps
the app full for 5 s: 12 hand-overs meanwhile, taken within 0.5 s of
having room, and no record sent again (21,640 without the flag).

5 files changed, 76 insertions, 30 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 570ab15..7940d90 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -19,7 +19,10 @@ enum { DATAGRAM_MOST = PURE_LINK_DATAGRAM_MOST, BODY_MOST = DATAGRAM_MOST - HEAD
  * at least this much of it fits. */
 enum { PART_LEAST = 64 };
 enum { F_PADDING = 0, F_STREAM = 1, F_RECORD = 2, F_ACK = 3, F_STORED = 4, F_PING = 5, F_RECORDS = 6, F_CHALLENGE = 7,
-       F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12, STORED_GAP = 1 };
+       F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12 };
+/* STORED's flags: what came after a gap was not kept; the app is full (no
+ * room for more now), and what came is kept for when it has room. */
+enum { STORED_GAP = 1, STORED_FULL = 2 };
 /* A packet's first byte is VERSION with the sending side's epoch (mod 16)
  * in its low four bits; OFFER and ACCEPT are an epoch's number (u32) and a
  * public key made for it, and ACCEPT the first bytes of the key it answers
@@ -702,6 +705,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
         l->stored = heard;
         l->stored_changed = now;
     }
+    if (flags & STORED_FULL) l->stored_changed = now; /* it keeps what came, with no room yet: the wait begins again */
     if ((flags & STORED_GAP) && heard >= l->stored) /* it did not keep what came after a gap: send it again */
         for (size_t k = 0; k < l->count && l->received; ++k) {
             Record *r = record_at(l, k);
@@ -1267,6 +1271,7 @@ typedef struct Stream {
     uint32_t input, shape; /* the host's number for the input; the messages' shape */
     uint64_t heard;              /* the last number heard (stored), once known */
     bool known, out, stored_due, gap, ready;
+    uint64_t pause, waits;       /* while the host is full: the pause, and the time until which it waits */
     /* The line: once the last heard is known, whole records from the one
      * after it on, in order, as a delivery carries them (u32 size, bytes),
      * from line_head: handed over as they are, let go by moving the head.
@@ -1311,6 +1316,9 @@ typedef struct { uint32_t name, next; } Allowed; /* name: where in names (u8 len
  * that often; what they held goes back to the system once TRIM_AFTER have
  * gone, or TRIM_WAIT after any went. */
 static const uint64_t IDLE_FIRST = 30000 * MS, SWEEP_EVERY = 250 * MS, TRIM_WAIT = 10000 * MS;
+/* A full host is handed its records again after these pauses: a STORED
+ * saying it is full comes that often, within the sender's STORED_WAIT. */
+static const uint64_t FULL_FIRST = 100 * MS, FULL_LONGEST = 500 * MS;
 enum { TRIM_AFTER = 1024 };
 
 struct PureLinkReceiver {
@@ -1319,6 +1327,7 @@ struct PureLinkReceiver {
     unsigned char own_public[PURE_SEAL_KEY];
     uint32_t own_id;
     uint64_t idle, swept, trimmed, untrimmed;
+    uint64_t waiting; /* the earliest a stream waits until for a full host (0: none) */
     /* Keys made for a run that then did not open cost a token; tokens come
      * back KEYINGS a second, up to KEYINGS. Under a flood of made-up
      * packets keys are made for that many a second at most, while senders
@@ -2440,7 +2449,7 @@ static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *
             body[at++] = F_STORED;
             at += put_names(body + at, st->source, st->input_name);
             put64(body + at, st->heard);
-            body[at + 8] = st->gap ? STORED_GAP : 0;
+            body[at + 8] = (st->gap ? STORED_GAP : 0) | (st->pause ? STORED_FULL : 0);
             size_t used = st->held_bytes + (st->line_size - st->line_head);
             size_t credit = used >= HELD_MOST ? 1 : st->mean ? (HELD_MOST - used) / st->mean : CREDIT;
             put32(body + at + 9, (uint32_t)(credit < 1 ? 1 : credit > CREDIT_MOST ? CREDIT_MOST : credit));
@@ -2470,10 +2479,12 @@ static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *
 
 /* What the host said of its deliveries: the saved records let go, and how
  * far they are stored said. */
-static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t count) {
+static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t count, uint64_t now) {
     for (size_t k = 0; k < count; ++k) {
         Stream *st = done[k].token;
         st->out = false;
+        st->pause = !done[k].full ? 0 : !st->pause ? FULL_FIRST : 2 * st->pause < FULL_LONGEST ? 2 * st->pause : FULL_LONGEST;
+        st->waits = st->pause ? now + st->pause : 0;
         mark_ready(s, st);
         if (!done[k].stored) continue; /* not saved: handed over again */
         if (!st->known || done[k].heard > st->heard) st->heard = done[k].heard;
@@ -2496,13 +2507,16 @@ static void apply_done(PureLinkReceiver *s, const PureLinkDone *done, size_t cou
 }
 
 /* The streams ready, each handed over; one the host could not take now is
- * tried again at the next turn. */
-static void hand_over_ready(PureLinkReceiver *s) {
+ * tried again at the next turn, and one it was full for after its pause. */
+static void hand_over_ready(PureLinkReceiver *s, uint64_t now) {
     size_t count = s->ready_count;
+    s->waiting = 0;
     for (size_t k = 0; k < count; ++k) {
         Stream *st = s->ready[k];
         st->ready = false;
-        if (!hand_over(s, st)) mark_ready(s, st);
+        bool waits = now < st->waits;
+        if (waits || !hand_over(s, st)) mark_ready(s, st);
+        if (waits && (!s->waiting || st->waits < s->waiting)) s->waiting = st->waits;
     }
     memmove(s->ready, s->ready + count, (s->ready_count - count) * sizeof(*s->ready));
     s->ready_count -= count;
@@ -2515,8 +2529,8 @@ void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size
         explicit_bzero(s->ours_private, sizeof(s->ours_private));
         s->ours_ready = false;
     }
-    apply_done(s, done, count);
-    hand_over_ready(s);
+    apply_done(s, done, count, now);
+    hand_over_ready(s, now);
     for (size_t k = 0; k < s->notice_count; ++k)
         if (send(context, s->notices[k].packet, sizeof(s->notices[k].packet), s->notices[k].to, s->notices[k].to_size))
             ++s->counts.packets;
@@ -2530,7 +2544,8 @@ void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size
 }
 
 uint64_t pure_link_receiver_due(const PureLinkReceiver *s) {
-    return s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0;
+    uint64_t due = s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0;
+    return s->waiting && s->waiting < due ? s->waiting : due;
 }
 
 void pure_link_receiver_counts(const PureLinkReceiver *s, PureLinkCounts *counts) {
diff --git a/link/link-protocol.h b/link/link-protocol.h
index f131e99..96085c6 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -225,8 +225,11 @@ typedef struct {
     void *context;
 } PureLinkHost;
 /* A delivery's end: heard, the last number heard; stored, its save
- * succeeded (else it is handed over again); gap, it stopped at a gap. */
-typedef struct { void *token; uint64_t heard; bool stored, gap; } PureLinkDone;
+ * succeeded (else it is handed over again); gap, it stopped at a gap; full,
+ * the host has no room for more now (its app's memory budget): what it did
+ * not take is kept, and handed over again after a pause, 100 ms doubling to
+ * half a second while it stays full (its sender told so meanwhile). */
+typedef struct { void *token; uint64_t heard; bool stored, gap, full; } PureLinkDone;
 /* A packet to send to an address a packet came from: whether it went. */
 typedef bool PureLinkSend(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
                           size_t to_size);
diff --git a/link/link.c b/link/link.c
index a834991..37a5a2a 100644
--- a/link/link.c
+++ b/link/link.c
@@ -554,7 +554,7 @@ bool pure_link_serve(PureLinkListener *s, char *error, size_t size) {
     return true;
 }
 
-void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool stored, bool gap) {
+void pure_link_done(PureLinkListener *s, PureLinkDone done) {
     pthread_mutex_lock(&s->lock);
     if (s->done_count == s->done_capacity) {
         size_t capacity = s->done_capacity ? 2 * s->done_capacity : 16;
@@ -563,7 +563,7 @@ void pure_link_done(PureLinkListener *s, void *token, uint64_t heard, bool store
         s->done = grown;
         s->done_capacity = capacity;
     }
-    s->done[s->done_count++] = (PureLinkDone){token, heard, stored, gap};
+    s->done[s->done_count++] = done;
     pthread_mutex_unlock(&s->lock);
     wake(s->wake);
 }
diff --git a/link/link.h b/link/link.h
index afe1430..3390a82 100644
--- a/link/link.h
+++ b/link/link.h
@@ -38,7 +38,7 @@ void pure_link_close(PureLink *);
 typedef struct PureLinkListener PureLinkListener;
 PureLinkListener *pure_link_listen(PureLinkReceiver *, unsigned port, char *error, size_t size);
 bool pure_link_serve(PureLinkListener *, char *error, size_t size);
-/* stored: the delivery's save succeeded; gap: it stopped at a gap. */
-void pure_link_done(PureLinkListener *, void *token, uint64_t heard, bool stored, bool gap);
+/* How a delivery went: link-protocol.h's PureLinkDone, with its token. */
+void pure_link_done(PureLinkListener *, PureLinkDone);
 void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts);
 #endif
diff --git a/link/test-link.c b/link/test-link.c
index 7655025..9106b50 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -97,6 +97,10 @@ typedef struct {
     PureLinkDone done[64], told[64];
     size_t done_count;
     unsigned fail_every; /* a delivery's save fails every so many (0: never) */
+    /* The app full (its memory budget) for full_ms from when it heard
+     * full_at: it takes nothing meanwhile, and says so. How often it was
+     * handed records meanwhile, and how long after it had room again. */
+    uint64_t full_at, full_ms, full_until, full_tries, full_late;
     bool later_open;     /* the app has an input "later" (as "order") */
 } World;
 
@@ -154,6 +158,13 @@ static bool deliver(void *context, const PureLinkDelivery *d) {
     CHECK(d->input == 3 && !strcmp(d->source, "test.ordered"), "a delivery to input %u from %s", d->input, d->source);
     if (w->done_count == sizeof(w->done) / sizeof(w->done[0])) return false; /* busy: tried again */
     ++w->deliveries;
+    if (w->full_ms && !w->full_until && w->heard >= w->full_at) w->full_until = w->now + w->full_ms * MS;
+    if (w->now < w->full_until) {
+        ++w->full_tries;
+        w->done[w->done_count++] = (PureLinkDone){d->token, w->heard, true, false, true};
+        return true;
+    }
+    if (w->full_until && !w->full_late) w->full_late = w->now - w->full_until + 1;
     bool fail = w->fail_every && w->deliveries % w->fail_every == 0, gap = false;
     uint64_t heard = w->heard;
     const unsigned char *at = d->records;
@@ -175,7 +186,7 @@ static bool deliver(void *context, const PureLinkDelivery *d) {
         w->applied += heard - w->heard;
         w->heard = heard;
     }
-    w->done[w->done_count++] = (PureLinkDone){d->token, w->heard, !fail, gap};
+    w->done[w->done_count++] = (PureLinkDone){d->token, w->heard, !fail, gap, false};
     return true;
 }
 
@@ -201,6 +212,7 @@ typedef struct {
     bool forgetful;    /* the receiver keeps no memo: it starts again knowing no epoch */
     unsigned cuts;     /* the path cut for 3 s so many times, as an epoch begins */
     bool black_hole;   /* larger packets lost from when the path was found to carry them */
+    uint64_t full_at, full_ms; /* the app full for full_ms once it heard full_at (0: never) */
 } Case;
 
 /* Answers to a packet no one should answer: counted, but a LOST (a packet
@@ -265,6 +277,7 @@ typedef struct {
     uint64_t packets, seconds_ms;
     size_t datagram; /* the sender's at the end */
     bool hole;       /* a black hole opened */
+    uint64_t full_tries, full_late; /* the app's, full */
 } Outcome;
 
 /* One who later has both nodes' private keys: the keys of the sender's
@@ -393,7 +406,8 @@ static Outcome run(const Case *c, const char *dir) {
     unlink(memo);
     World w = {.packets = malloc(PACKETS * sizeof(Packet)), .loss = c->loss, .doubled = c->doubled,
                .spread_ms = c->spread_ms, .fail_every = c->fail_every, .replay = c->replay, .tamper = c->tamper,
-               .cuts = c->cuts, .black_hole = c->black_hole, .history = malloc(HISTORY * sizeof(Packet))};
+               .cuts = c->cuts, .black_hole = c->black_hole, .full_at = c->full_at, .full_ms = c->full_ms,
+               .history = malloc(HISTORY * sizeof(Packet))};
     if (!w.packets || !w.history) exit(1);
     char from[8] = "sender0";
     epoch_ms = c->epoch_ms;
@@ -490,7 +504,8 @@ static Outcome run(const Case *c, const char *dir) {
             snprintf(from, sizeof(from), "sender%u", ++w.gen); /* a new process: another address */
         }
     }
-    Outcome o = {.packets = w.sent, .seconds_ms = w.now / MS, .datagram = pure_link_sender_datagram(s), .hole = w.hole};
+    Outcome o = {.packets = w.sent, .seconds_ms = w.now / MS, .datagram = pure_link_sender_datagram(s), .hole = w.hole,
+                 .full_tries = w.full_tries, .full_late = w.full_late};
     pure_link_sender_counts(s, &o.sent);
     pure_link_receiver_counts(r, &o.heard);
     o.sent.packets += sent_before.packets, o.sent.resent += sent_before.resent, o.sent.epochs += sent_before.epochs;
@@ -514,12 +529,13 @@ static Outcome run(const Case *c, const char *dir) {
     unsigned opened_later = later_with_keys(&w, a, b, public_a, public_b, c->name);
     printf("{\"case\":\"%s\",\"records\":%llu,\"virtual_ms\":%llu,\"packets\":%llu,\"resent\":%llu,\"lost\":%llu,"
            "\"deliveries\":%llu,\"woken\":%llu,\"remembered\":%llu,\"swept\":%llu,\"changed\":%llu,\"replayed\":%llu,"
-           "\"attacker_in\":%llu,\"attacker_out\":%llu,\"epochs\":%llu,\"opened_later\":%u}\n",
+           "\"attacker_in\":%llu,\"attacker_out\":%llu,\"epochs\":%llu,\"opened_later\":%u,\"full_tries\":%llu}\n",
            c->name, (unsigned long long)c->records, (unsigned long long)o.seconds_ms, (unsigned long long)o.sent.packets,
            (unsigned long long)o.sent.resent, (unsigned long long)o.sent.lost, (unsigned long long)w.deliveries,
            (unsigned long long)o.heard.woken, (unsigned long long)o.heard.remembered, (unsigned long long)o.heard.swept,
            (unsigned long long)w.bent, (unsigned long long)w.replayed, (unsigned long long)w.attacker_in,
-           (unsigned long long)w.attacker_out, (unsigned long long)o.sent.epochs, opened_later);
+           (unsigned long long)w.attacker_out, (unsigned long long)o.sent.epochs, opened_later,
+           (unsigned long long)w.full_tries);
     pure_link_sender_free(s);
     pure_link_receiver_free(r);
     pure_seal_key_free(a);
@@ -692,7 +708,7 @@ static bool output_deliver(void *context, const PureLinkDelivery *d) {
         at += 4 + size;
     }
     o->heard[d->input] = heard;
-    o->done[o->done_count++] = (PureLinkDone){d->token, heard, true, gap};
+    o->done[o->done_count++] = (PureLinkDone){d->token, heard, true, gap, false};
     return true;
 }
 static bool output_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
@@ -888,7 +904,7 @@ static bool deliver_largest(void *context, const PureLinkDelivery *d) {
     CHECK(same, "largest: a delivery of %u records from %llu, the first %u bytes", d->count,
           (unsigned long long)d->first, size);
     w->heard = same ? 1 : w->heard;
-    w->done[w->done_count++] = (PureLinkDone){d->token, 1, true, false};
+    w->done[w->done_count++] = (PureLinkDone){d->token, 1, true, false, false};
     return true;
 }
 static bool answer_largest(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
@@ -941,8 +957,9 @@ int main(void) {
         return 1;
     }
     /* A clean network: nothing is sent twice. */
-    Case clean = {"clean", 3000, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, 0, false, 0, false};
+    Case clean = {"clean", 3000, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, 0, false, 0, false, 0, 0};
     Outcome o = run(&clean, dir);
+    uint64_t clean_packets = o.packets;
     CHECK(o.sent.resent == 0 && o.sent.lost == 0, "clean: %llu sent again, %llu lost", (unsigned long long)o.sent.resent,
           (unsigned long long)o.sent.lost);
     /* Loss, packets out of order and doubled, failed saves, a pause long
@@ -950,7 +967,7 @@ int main(void) {
      * starting again: the receiver remembers its secret, so it wakes the
      * sender without an X25519. */
     for (uint64_t s = 1; s <= 4; ++s) {
-        Case rough = {"rough", 3000, s * 7919, 10, 3, 20, 37, 1500, 800, 2200, 300, 0, false, 0, false, 0, false};
+        Case rough = {"rough", 3000, s * 7919, 10, 3, 20, 37, 1500, 800, 2200, 300, 0, false, 0, false, 0, false, 0, 0};
         o = run(&rough, dir);
         CHECK(o.sent.resent > 0 && o.heard.swept >= 1 && o.heard.woken >= 1 && o.heard.remembered >= 1,
               "rough: sent again %llu, swept %llu, woken %llu, remembered %llu", (unsigned long long)o.sent.resent,
@@ -960,36 +977,47 @@ int main(void) {
     /* An attacker on the path: packets changed on the way, and packets of
      * the past sent again, from the sender's address and its own, across the
      * same restarts and pause. */
-    Case changed = {"changed", 1500, 31, 3, 0, 5, 0, 0, 0, 0, 0, 0, true, 0, false, 0, false};
+    Case changed = {"changed", 1500, 31, 3, 0, 5, 0, 0, 0, 0, 0, 0, true, 0, false, 0, false, 0, 0};
     run(&changed, dir);
     for (uint64_t s = 1; s <= 2; ++s) {
-        Case again = {"replayed", 3000, s * 104729, 5, 3, 20, 37, 1500, 800, 2200, 300, 25, false, 0, false, 0, false};
+        Case again = {"replayed", 3000, s * 104729, 5, 3, 20, 37, 1500, 800, 2200, 300, 25, false, 0, false, 0, false, 0, 0};
         run(&again, dir);
     }
     /* Epochs a fifth of a second long (a new one a second, as the receiver
      * accepts), and a receiver that starts again with no memo: it knows no
      * epoch, and the sender begins them again. */
-    Case epochs = {"epochs", 3000, 271, 5, 3, 20, 37, 1500, 800, 2200, 300, 10, false, 200, false, 0, false};
+    Case epochs = {"epochs", 3000, 271, 5, 3, 20, 37, 1500, 800, 2200, 300, 10, false, 200, false, 0, false, 0, 0};
     o = run(&epochs, dir);
     CHECK(o.sent.epochs >= 10, "epochs: %llu begun", (unsigned long long)o.sent.epochs);
-    Case forgetful = {"forgetful", 3000, 409, 5, 3, 20, 37, 0, 800, 0, 0, 0, false, 0, true, 0, false};
+    Case forgetful = {"forgetful", 3000, 409, 5, 3, 20, 37, 0, 800, 0, 0, 0, false, 0, true, 0, false, 0, 0};
     o = run(&forgetful, dir);
     CHECK(o.sent.epochs >= 2, "forgetful: %llu epochs begun (none again)", (unsigned long long)o.sent.epochs);
     /* The path cut for 3 s the moment the sender begins an epoch, three
      * times: the receiver never saw the epoch used, and the sender, hearing
      * nothing, offers another epoch 1 with a new key. The accept of the old
      * offer must not begin it. */
-    Case cut = {"cut", 3000, 919, 0, 0, 20, 37, 0, 0, 0, 0, 0, false, 200, false, 3, false};
+    Case cut = {"cut", 3000, 919, 0, 0, 20, 37, 0, 0, 0, 0, 0, false, 200, false, 3, false, 0, 0};
     o = run(&cut, dir);
     CHECK(o.sent.epochs >= 4 && o.sent.fallbacks == 3, "cut: %llu epochs begun, %llu taken for lost (3 cuts)",
           (unsigned long long)o.sent.epochs, (unsigned long long)o.sent.fallbacks);
     /* A path that carried 1,452-byte packets and then, from the moment the
      * link found it did, carries none larger than 1,200 (a route changed,
      * and no ICMP says so): the link must see it and go back. */
-    Case hole = {"black hole", 3000, 1601, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, 0, false, 0, true};
+    Case hole = {"black hole", 3000, 1601, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, 0, false, 0, true, 0, 0};
     o = run(&hole, dir);
     CHECK(o.hole && o.datagram == PURE_LINK_DATAGRAM, "black hole: %s, a datagram of %zu at the end",
           o.hole ? "opened" : "never opened", o.datagram);
+    /* The app full for 5 s (its memory budget): it is handed the records
+     * again after pauses of 0.1, 0.2, 0.4 and then 0.5 s, not at every
+     * turn, and within half a second of having room it takes them. The
+     * sender, told the receiver keeps what came, sends none of it again. */
+    Case full = {"full", 3000, 2027, 0, 0, 0, 0, 0, 0, 0, 0, 0, false, 0, false, 0, false, 1000, 5000};
+    o = run(&full, dir);
+    CHECK(o.full_tries >= 9 && o.full_tries <= 14 && o.full_late && o.full_late <= 500 * MS + 10 * MS,
+          "full: handed over %llu times while full, taken %llu ms after", (unsigned long long)o.full_tries,
+          (unsigned long long)(o.full_late / MS));
+    CHECK(o.sent.resent == 0 && o.packets < clean_packets + 200, "full: %llu sent again, %llu packets (clean: %llu)",
+          (unsigned long long)o.sent.resent, (unsigned long long)o.packets, (unsigned long long)clean_packets);
     impostors(dir);
     flood(dir);
     outputs(dir);