Say why a stream is refused (REFUSED), and try it again seldom

A receiver that took no records of a stream (a source the key may not
send as, no input of that name, an input of another shape) said nothing:
its sender heard them acknowledged, never stored, and sent them all
again every second, for ever, and its operator saw nothing. Now the
receiver answers REFUSED (0x0c: the names, and why), and the sender
sends no records for a second, then a packet of them to learn whether
they are taken now, the wait twice as long each time (64 s at most). A
refusal ends when that packet is acknowledged with no REFUSED beside, or
a record is stored. link.c says it once on stderr, as a wire says a
receiver does not take its sends; serve's input says which of the two
the app lacks.

test-link: each of the three is refused seven times in 90 s with nine
packets, and a stream refused until the app has its input goes on then,
all 200 records stored. check.py links a node to an input its receiver
lacks: said once, two packets in 2.5 s.

4 files changed, 213 insertions, 21 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 6120deb..1106ac5 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -19,7 +19,7 @@ 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, STORED_GAP = 1 };
+       F_RESPONSE = 8, F_OFFER = 9, F_ACCEPT = 10, F_LOST = 11, F_REFUSED = 12, STORED_GAP = 1 };
 /* 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
@@ -60,6 +60,9 @@ static const double CUBIC_C = 0.4, CUBIC_BETA = 0.7, RENO_PACE = 3 * (1 - 0.7) /
 static const uint64_t FIRST_RTT = 100 * MS, ACK_DELAY_MOST = 25 * MS, GRANULARITY = 1 * MS, PTO_MOST = 1000 * MS,
                       STORED_WAIT = 1000 * MS, RECORD_MOST = PURE_LINK_RECORD_MOST, HELD_MOST = RECORD_MOST,
                       REACH_AGAIN = 1000 * MS, REACH_RARELY = 600000 * MS;
+/* A stream refused is tried a packet at a time, a second on at first and
+ * twice as long each time it is refused again. */
+static const uint64_t REFUSED_FIRST = 1000 * MS, REFUSED_MOST = 64000 * MS;
 /* An epoch lasts an hour or a GiB sent; a sender that hears nothing for 2 s
  * with packets out, and four probes, takes its epoch for lost. */
 static const uint64_t EPOCH_EVERY = UINT64_C(3600000) * MS, EPOCH_MOST = UINT64_C(1) << 30, FALLBACK_WAIT = 2000 * MS;
@@ -197,6 +200,7 @@ static size_t get_names(const unsigned char *p, size_t size, char source[PURE_LI
 /* ---- The sending side */
 
 enum { UNSENT, FLYING, RECEIVED };
+enum { TRYING = 1, TRIED = 2 }; /* a packet sent to learn whether a stream refused is taken now */
 /* Part of a record's bytes, gone whole in one packet (numbered pn): a
  * record is one part until a packet has room for only some of it. */
 typedef struct { uint64_t pn; uint32_t offset, length; uint8_t state; } Part;
@@ -286,6 +290,14 @@ struct PureLinkSender {
     uint64_t received;    /* parts received and not yet stored */
     uint64_t unsent_from; /* no record before this number has a part unsent */
     unsigned pto_count, probes;
+    /* The receiver refuses the stream (why; 0: it does not): records go
+     * again from refused_until (0: now), a packet of them (numbered
+     * tried_pn: TRYING, then TRIED once acknowledged), the wait twice as
+     * long each time it is refused (and back to the first once a record is
+     * stored); the refusal is over when that packet is acknowledged in an
+     * answer with no REFUSED. */
+    unsigned refusal, tried;
+    uint64_t refused_until, refused_wait, tried_pn;
     PureLinkCounts counts;
     /* Epochs (forward secrecy). Keys come from the secret of the epoch it
      * sends in (epoch 0's is the two nodes' static secret), each made from
@@ -635,6 +647,7 @@ static void on_ack(PureLinkSender *l, const unsigned char *p, size_t size, uint6
         uint64_t oldest = l->flights[l->flight_head].pn, newest = oldest + l->flight_count - 1;
         for (uint64_t pn = most(first, oldest); pn <= least(last, newest); ++pn) {
             Flight *f = flight_of(l, pn);
+            if (f && l->tried == TRYING && pn == l->tried_pn) l->tried = TRIED;
             if (f && f->lost) came_late(l, f, now);
             if (!f || f->done) continue;
             settle(l, f);
@@ -671,6 +684,7 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
     uint64_t heard = get64(p + names);
     unsigned flags = p[names + 8];
     uint32_t credit = get32(p + names + 9);
+    l->refusal = l->tried = 0, l->refused_until = l->refused_wait = 0; /* the stream is taken, and stored */
     if (heard > l->stored) {
         while (l->count && record_at(l, 0)->seq <= heard) {
             Record *r = record_at(l, 0);
@@ -691,6 +705,24 @@ static void on_stored(PureLinkSender *l, const unsigned char *p, size_t size, ui
     l->limit = most(l->limit, heard + (credit ? least(credit, 65536) : 1));
 }
 
+/* The receiver refuses the stream: no records go for a while, then a
+ * packet of them, to learn whether it takes them now. Said of packets that
+ * went before that one (it is not acknowledged yet: the ACK comes first),
+ * it changes nothing. What went goes again: none of it was taken. */
+static void refused(PureLinkSender *l, unsigned why, uint64_t now) {
+    if (l->refusal && l->tried != TRIED) return;
+    l->refusal = why ? why : PURE_LINK_NO_INPUT;
+    l->refused_wait = l->refused_wait ? least(2 * l->refused_wait, REFUSED_MOST) : REFUSED_FIRST;
+    l->refused_until = now + l->refused_wait;
+    l->tried = 0;
+    ++l->counts.refused;
+    for (size_t k = 0; k < l->count; ++k) {
+        Record *r = record_at(l, k);
+        for (uint32_t i = 0; i < r->parts; ++i)
+            if (part_of(r, i)->state != UNSENT) set_part(l, r, part_of(r, i), UNSENT);
+    }
+}
+
 void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_t n, uint64_t now) {
     unsigned char plain[DATAGRAM_MOST], nonce[PURE_AEAD_NONCE];
     if (!l->out || n < HEADER + PURE_AEAD_TAG || n > DATAGRAM_MOST || (packet[0] & ~EPOCH_MASK) != VERSION ||
@@ -716,6 +748,7 @@ void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_
     if (!pure_aead_open(l->in, NULL, nonce, packet, HEADER, packet + HEADER, n - HEADER, plain)) return;
     ++l->counts.received;
     l->heard_at = now;
+    bool refusing = false;
     for (size_t at = 0; at < size;) {
         unsigned type = plain[at++];
         if (type == F_ACK && at + 13 <= size && at + 13 + 16u * plain[at + 12] <= size) {
@@ -749,9 +782,16 @@ void pure_link_sender_take(PureLinkSender *l, const unsigned char *packet, size_
             Flight *f = flight_of(l, get64(plain + at));
             if (f && !f->done && (l->epoch || l->renegotiating)) epoch_lost(l);
             at += 8;
+        } else if (type == F_REFUSED) {
+            char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
+            size_t names = get_names(plain + at, size - at, source, input);
+            if (!names || at + names + 1 > size) break;
+            if (!strcmp(source, l->source) && !strcmp(input, l->input)) refused(l, plain[at + names], now), refusing = true;
+            at += names + 1;
         } else if (type != F_PING)
             break; /* padding, or what this side does not know: the rest */
     }
+    if (l->refusal && l->tried == TRIED && !refusing) l->refusal = l->tried = 0, l->refused_until = 0; /* taken now */
 }
 
 /* A packet of what is not sent yet, in order, within the receiver's credit
@@ -916,6 +956,7 @@ size_t pure_link_sender_packet(PureLinkSender *l, unsigned char out[PURE_LINK_DA
     l->pace_due = 0; /* set again only when pacing holds a packet back */
     if (!probe && l->in_flight + l->datagram > l->cc.window) l->unused = false;
     if (!l->out || (!probe && l->in_flight + l->datagram > l->cc.window) || !flight_room(l)) return 0;
+    if (l->refusal && !probe && (l->refused_until || l->tried)) return 0; /* refused: a packet a wait */
     if (!probe && !paced(l, now)) return 0;
     if (!l->offering && !l->renegotiating && (now - l->epoch_at >= l->epoch_every || l->epoch_bytes >= l->epoch_most))
         offer(l, l->epoch + 1); /* on to the next epoch: a key made for it goes in the packets until accepted */
@@ -946,6 +987,7 @@ size_t pure_link_sender_packet(PureLinkSender *l, unsigned char out[PURE_LINK_DA
     size_t total = transmit(l, &f, body, size, out, now);
     if (!total) return 0;
     if (probe) --l->probes;
+    if (l->refusal && f.count) l->tried = TRYING, l->tried_pn = f.pn;
     *number = f.pn;
     return total;
 }
@@ -970,10 +1012,12 @@ uint64_t pure_link_sender_due(const PureLinkSender *l) {
         uint64_t stored = l->stored_changed + STORED_WAIT;
         if (!due || stored < due) due = stored;
     }
+    if (l->refusal && l->refused_until && l->count && (!due || l->refused_until < due)) due = l->refused_until;
     return due;
 }
 
 void pure_link_sender_tick(PureLinkSender *l, uint64_t now) {
+    if (l->refusal && l->refused_until && l->refused_until <= now) l->refused_until = 0; /* a packet may try */
     if (l->epoch && !l->renegotiating && l->eliciting && l->heard_at && now - l->heard_at > FALLBACK_WAIT &&
         l->pto_count >= FALLBACK_PROBES) {
         /* Nothing heard for long with packets out: the receiver lost the
@@ -1125,6 +1169,8 @@ uint64_t pure_link_sender_queued(const PureLinkSender *l) { return l->queued; }
 
 unsigned pure_link_sender_probes(const PureLinkSender *l) { return l->pto_count; }
 
+unsigned pure_link_sender_refusal(const PureLinkSender *l) { return l->refusal; }
+
 void pure_link_sender_epochs(PureLinkSender *l, uint64_t every_ns, uint64_t bytes) {
     l->epoch_every = every_ns;
     l->epoch_most = bytes;
@@ -1296,6 +1342,10 @@ struct PureLinkReceiver {
      * LOST, to a sender whose packet came in an epoch it does not know. */
     struct { unsigned char packet[HEADER + 9 + PURE_AEAD_TAG], to[PURE_LINK_ADDRESS]; size_t to_size; } *notices;
     size_t notice_count, notice_capacity;
+    /* Streams refused since the last answers: REFUSED frames, each in the
+     * answer to the run that named it (a slot may hold another run by then). */
+    struct { Run *run; uint64_t id; size_t size; unsigned char frame[3 + 2 * PURE_LINK_NAME]; } *refusals;
+    size_t refusal_count, refusal_capacity;
     PureAead *aead; /* keyed by each packet's run */
     unsigned char open_key[PURE_AEAD_KEY], seal_key[PURE_AEAD_KEY]; /* the keys it has now */
     bool open_keyed, seal_keyed;
@@ -1733,16 +1783,24 @@ static bool allowed(const PureLinkReceiver *s, const Peer *peer, const char *sou
     return false;
 }
 
-/* The stream of a peer's source to input, made the first time: NULL when the
- * source is not allowed from the peer, or the host has no input that takes
- * such messages. */
-static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape) {
+/* The stream of a peer's source to input, made the first time: NULL when it
+ * is refused (why: the source is not allowed from the peer, or the host has
+ * no input that takes such messages), or memory runs out (why 0). */
+static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape,
+                         unsigned *why) {
+    *why = 0;
     for (unsigned k = 0; k < a->stream_count; ++k) {
         Stream *st = a->streams[k];
-        if (!strcmp(st->input_name, input) && !strcmp(st->source, source)) return st->shape == shape ? st : NULL;
+        if (strcmp(st->input_name, input) || strcmp(st->source, source)) continue;
+        if (st->shape != shape) *why = PURE_LINK_OTHER_SHAPE;
+        return *why ? NULL : st;
+    }
+    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape)
+                                                       : -PURE_LINK_NOT_ALLOWED;
+    if (index < 0) {
+        *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED ? (unsigned)-index : PURE_LINK_NO_INPUT;
+        return NULL;
     }
-    int index = allowed(s, &s->peers[a->peer], source) ? s->host.input(s->host.context, input, shape) : -1;
-    if (index < 0) return NULL;
     size_t capacity = a->stream_capacity;
     Stream **streams = room(a->streams, &capacity, a->stream_count, sizeof(*streams), 2), *st;
     if (!streams) return NULL;
@@ -2091,6 +2149,25 @@ static void lost(PureLinkReceiver *s, const Peer *peer, const Active *a, uint64_
     explicit_bzero(key, sizeof(key));
 }
 
+/* A stream refused (its names as the packet had them), said in the answer
+ * to its run: once a turn. */
+static void refuse(PureLinkReceiver *s, Run *r, const unsigned char *names, size_t size, unsigned why) {
+    for (size_t k = 0; k < s->refusal_count; ++k)
+        if (s->refusals[k].run == r && s->refusals[k].size == size + 2 && !memcmp(s->refusals[k].frame + 1, names, size))
+            return;
+    void *more = room(s->refusals, &s->refusal_capacity, s->refusal_count, sizeof(*s->refusals), 4);
+    if (!more) return;
+    s->refusals = more;
+    unsigned char *frame = s->refusals[s->refusal_count].frame;
+    frame[0] = F_REFUSED;
+    memcpy(frame + 1, names, size);
+    frame[1 + size] = (unsigned char)why;
+    s->refusals[s->refusal_count].run = r;
+    s->refusals[s->refusal_count].id = r->run;
+    s->refusals[s->refusal_count++].size = size + 2;
+    ++s->counts.refused;
+}
+
 /* Whether keys may be made for a run not seen (a token is there). */
 static bool may_key(PureLinkReceiver *s, uint64_t now) {
     return has_token(&s->key_tokens, &s->key_at, KEYINGS, KEYINGS, now);
@@ -2185,11 +2262,13 @@ static bool take_from(PureLinkReceiver *s, Peer *peer, const unsigned char *pack
             char source[PURE_LINK_NAME + 1], input[PURE_LINK_NAME + 1];
             size_t names = get_names(plain + at, size - at, source, input);
             if (!names || at + names + 4 > size) break;
-            st = stream_of(s, a, source, input, get32(plain + at + names));
+            unsigned why;
+            st = stream_of(s, a, source, input, get32(plain + at + names), &why);
             if (st) {
                 st->run = r;
                 st->run_id = r->run;
-            }
+            } else if (why)
+                refuse(s, r, plain + at, names, why);
             at += names + 4;
             eliciting = true;
         } else if (type == F_RECORD) {
@@ -2320,6 +2399,11 @@ static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *
             at += 13;
             st->stored_due = st->gap = false;
         }
+        for (size_t i = 0; i < s->refusal_count; ++i)
+            if (s->refusals[i].run == r && s->refusals[i].id == r->run && at + s->refusals[i].size <= BODY) {
+                memcpy(body + at, s->refusals[i].frame, s->refusals[i].size);
+                at += s->refusals[i].size;
+            }
         r->ack_due = false;
         if (!at) continue;
         if (!r->checked && r->bytes_out + HEADER + at + PURE_AEAD_TAG > 3 * r->bytes_in) continue; /* no amplifier */
@@ -2333,7 +2417,7 @@ static void answer(PureLinkReceiver *s, uint64_t now, PureLinkSend *send, void *
             if (!r->checked) r->bytes_out += total;
         }
     }
-    s->due_count = 0;
+    s->due_count = s->refusal_count = 0;
 }
 
 /* What the host said of its deliveries: the saved records let go, and how
@@ -2412,6 +2496,7 @@ void pure_link_receiver_free(PureLinkReceiver *s) {
     explicit_bzero(s->ours_private, sizeof(s->ours_private));
     free(s->offers);
     free(s->notices);
+    free(s->refusals);
     if (s->memo) munmap(s->memo, s->memo_size);
     free(s->actives);
     free(s->peers);
diff --git a/link/link-protocol.h b/link/link-protocol.h
index b9a3eeb..df15246 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -106,6 +106,11 @@
  *                its epoch for lost offers epoch 1 again with another)
  *   0x0b LOST    u64 a packet's number: a receiver knows no epoch of its run
  *                (sealed in epoch 0, in a run of answers of its own)
+ *   0x0c REFUSED source, input (as STREAM), u8 why (PURE_LINK_NO_INPUT and
+ *                on): the receiver takes no records of that stream; its
+ *                sender sends one packet of them a wait from then (a
+ *                second, twice as long each time, 64 s at most) until one
+ *                is stored
  *   0x00 PADDING to the end of the packet: a full packet sent in a burst is
  *                a whole datagram
  * Losses are detected as RFC 9002 says, with RACK's patience for packets
@@ -123,18 +128,23 @@
  * stream not yet stored). */
 enum { PURE_LINK_DATAGRAM = 1200, PURE_LINK_DATAGRAM_MOST = 1452, PURE_LINK_NAME = 64, PURE_LINK_ADDRESS = 48,
        PURE_LINK_RECORD_MOST = 16 << 20 };
+/* Why a receiver refuses a stream (REFUSED): no input of that name takes
+ * messages, the input takes messages of another shape, or the sender's key
+ * may not send as that source. */
+enum { PURE_LINK_NO_INPUT = 1, PURE_LINK_OTHER_SHAPE = 2, PURE_LINK_NOT_ALLOWED = 3 };
 
 /* Counts for a host's report: packets and records sent, records sent again,
  * packets lost and acknowledged, packets received, deliveries handed over,
  * the last number stored; packets called lost that came late, and
  * congestion events undone for it; runs keyed (keys made from a secret),
  * secrets made (X25519), epochs begun, and a sender's epochs taken for lost
- * (begun again from the nodes' keys); a receiver's senders woken (their
- * active state made), of them with a secret remembered (no X25519), let go
- * when idle, and active now. */
+ * (begun again from the nodes' keys); streams refused (a receiver's said, a
+ * sender's heard); a receiver's senders woken (their active state made), of
+ * them with a secret remembered (no X25519), let go when idle, and active
+ * now. */
 typedef struct {
     uint64_t packets, records, resent, lost, acknowledged, received, deliveries, stored, late, undone, keyed, secrets,
-        epochs, fallbacks, woken, remembered, swept, active;
+        epochs, fallbacks, refused, woken, remembered, swept, active;
 } PureLinkCounts;
 
 /* ---- The sending side: one output's saved sends to a receiver's input, as
@@ -177,6 +187,9 @@ uint64_t pure_link_sender_due(const PureLinkSender *);
 /* Probes in a row that nothing answered (0 once anything is): the path, or
  * the receiver where it was, may be gone. */
 unsigned pure_link_sender_probes(const PureLinkSender *);
+/* Why the receiver refuses the stream (PURE_LINK_NO_INPUT and on), or 0:
+ * cleared once it stores one of its records. */
+unsigned pure_link_sender_refusal(const PureLinkSender *);
 /* How often it moves on to a new epoch (ns), and after how many bytes sent
  * (an hour and a GiB at first). */
 void pure_link_sender_epochs(PureLinkSender *, uint64_t every_ns, uint64_t bytes);
@@ -189,7 +202,9 @@ void pure_link_sender_free(PureLinkSender *);
 /* ---- The receiving side: packets from the senders allowed (a source name
  * and the key it seals with, a few bytes each however many there are: see
  * pure_link_idle). A stream (a source to an input) goes to the host's input
- * its name and shape find (its number, or -1: the stream is refused). Whole
+ * its name and shape find: its number, or -PURE_LINK_OTHER_SHAPE when it
+ * takes another shape, or -1 (PURE_LINK_NO_INPUT) when none takes messages
+ * by that name; a stream refused is said to be (REFUSED). Whole
  * records of a stream, in order, go to deliver as one delivery; the host
  * applies them from the one after the source's last heard number, stops at
  * a gap, saves, and then says so with a done (pure_link_receiver_turn).
diff --git a/link/link.c b/link/link.c
index a020c2d..a834991 100644
--- a/link/link.c
+++ b/link/link.c
@@ -112,7 +112,7 @@ static int take_datagrams(int fd, Taking *t) {
 
 struct PureLink {
     PureLinkSender *sender; /* under the lock */
-    char host[256], port[8], input[PURE_LINK_NAME + 1];
+    char host[256], port[8], input[PURE_LINK_NAME + 1], source[PURE_LINK_NAME + 1];
     bool peered, keyed; /* given the receiver's key; the keys made from it */
     pthread_mutex_t lock;
     bool stopping, started;
@@ -132,8 +132,9 @@ struct PureLink {
     uint64_t burst_pn[BURST];
     size_t segment; /* the datagram's size the bursts are cut at */
     bool gso, no_gso;
-    unsigned looked; /* the probes unanswered when it last looked for the receiver */
-    bool nowhere;    /* a send failed as the system saw no way there (no route, nobody at the port) */
+    unsigned looked;  /* the probes unanswered when it last looked for the receiver */
+    bool nowhere;     /* a send failed as the system saw no way there (no route, nobody at the port) */
+    unsigned refusal; /* why the receiver refused the stream when it was last said (0: it takes it) */
 };
 
 PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
@@ -159,6 +160,7 @@ PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, co
     memcpy(l->host, host, (size_t)(colon - host));
     memcpy(l->port, colon + 1, (size_t)(slash - colon - 1));
     strcpy(l->input, slash + 1);
+    memcpy(l->source, source, length);
     l->fd = l->wake = -1;
     if (!(l->sender = pure_link_sender(source, l->input, shape, own, own_public))) {
         pthread_mutex_destroy(&l->lock);
@@ -362,8 +364,16 @@ static void *run_link(void *context) {
         uint64_t stored = pure_link_sender_stored(l->sender);
         bool told = stored != l->told;
         l->told = stored;
+        unsigned refusal = pure_link_sender_refusal(l->sender), before = l->refusal;
+        l->refusal = refusal;
         pthread_mutex_unlock(&l->lock);
         if (told && l->notify) l->notify(l->context);
+        if (refusal && refusal != before) /* as a wire says the receiver does not take its sends */
+            fprintf(stderr, "{\"link\":\"pure://%s:%s/%s\",\"source\":\"%s\",\"error\":\"the receiver refuses the stream: %s; "
+                    "trying again\"}\n", l->host, l->port, l->input, l->source,
+                    refusal == PURE_LINK_NOT_ALLOWED ? "this node's key may not send as that source"
+                    : refusal == PURE_LINK_OTHER_SHAPE ? "its input takes messages of another shape"
+                                                       : "it has no input of that name taking messages");
     }
     return NULL;
 }
diff --git a/link/test-link.c b/link/test-link.c
index 7024260..f609fe5 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -97,6 +97,7 @@ typedef struct {
     PureLinkDone done[64], told[64];
     size_t done_count;
     unsigned fail_every; /* a delivery's save fails every so many (0: never) */
+    bool later_open;     /* the app has an input "later" (as "order") */
 } World;
 
 static void put_packet(World *w, bool to_receiver, const unsigned char *bytes, size_t size) {
@@ -142,8 +143,9 @@ static bool answer(void *context, const unsigned char *packet, size_t size, cons
 }
 
 static int input_of(void *context, const char *name, uint32_t shape) {
-    (void)context;
-    return !strcmp(name, "order") && shape == 0x5eed ? 3 : -1;
+    const World *w = context;
+    if (strcmp(name, "order") && !(w && w->later_open && !strcmp(name, "later"))) return -PURE_LINK_NO_INPUT;
+    return shape == 0x5eed ? 3 : -PURE_LINK_OTHER_SHAPE;
 }
 
 /* Applied from the one after the last heard, in order; a gap stops it. */
@@ -741,6 +743,85 @@ static void outputs(const char *dir) {
     pure_seal_key_free(a), pure_seal_key_free(b);
 }
 
+/* Streams a receiver refuses: a source its key may not send as, an input it
+ * does not have, one of another shape. Each sender hears why, and then
+ * tries a packet a wait (1, 2, 4 s ...) instead of sending its records
+ * again every second; and one refused until the receiver's app has the
+ * input goes on then, all its records delivered. */
+static void refusals(const char *dir) {
+    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST];
+    PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
+    char memo[512];
+    snprintf(memo, sizeof(memo), "%s/b.key.links", dir);
+    unlink(memo);
+    World w = {.packets = malloc(PACKETS * sizeof(Packet))};
+    if (!w.packets) exit(1);
+    /* Its senders let go only after two minutes quiet: else one of the runs
+     * the memo does not keep is answered LOST at a try (a minute apart). */
+    PureLinkReceiver *r = receiver_of(&w, b, public_b, public_a, memo, 120000);
+    struct { const char *source, *input; uint32_t shape; unsigned why; PureLinkSender *s; uint64_t packets; } links[] = {
+        {"other.source", "order", 0x5eed, PURE_LINK_NOT_ALLOWED, NULL, 0},
+        {"test.ordered", "nothing", 0x5eed, PURE_LINK_NO_INPUT, NULL, 0},
+        {"test.ordered", "order", 0x5eee, PURE_LINK_OTHER_SHAPE, NULL, 0},
+        {"test.ordered", "later", 0x5eed, 0, NULL, 0}, /* no such input until 10 s on */
+    };
+    enum { LINKS = sizeof(links) / sizeof(links[0]) };
+    static unsigned char bytes[20200];
+    for (unsigned i = 0; i < LINKS; ++i) {
+        PureLinkSender *s = links[i].s = pure_link_sender(links[i].source, links[i].input, links[i].shape, a, public_a);
+        if (!s || !pure_link_sender_peer(s, public_b)) exit(1);
+        for (uint64_t seq = 1; seq <= 200; ++seq) {
+            size_t size = record_size(seq);
+            for (size_t k = 0; k < size; ++k) bytes[k] = record_byte(seq, k);
+            if (!pure_link_sender_queue(s, seq, bytes, size)) exit(1);
+        }
+    }
+    uint64_t number;
+    for (; w.now < 90000 * MS; w.now += MS) {
+        if (w.now == 10000 * MS) w.later_open = true;
+        for (unsigned i = 0; i < LINKS; ++i)
+            for (size_t size; (size = pure_link_sender_packet(links[i].s, out, false, w.now, &number)); ++links[i].packets)
+                pure_link_receiver_take(r, out, size, "sender0", 7, w.now);
+        size_t told = w.done_count;
+        memcpy(w.told, w.done, told * sizeof(*w.done));
+        w.done_count = 0;
+        pure_link_receiver_turn(r, w.told, told, w.now, answer, &w);
+        for (size_t k = 0; k < w.count;) { /* each answer to all: only its sender's run opens it */
+            if (w.packets[k].at > w.now) {
+                ++k;
+                continue;
+            }
+            for (unsigned i = 0; i < LINKS; ++i) pure_link_sender_take(links[i].s, w.packets[k].bytes, w.packets[k].size, w.now);
+            w.packets[k] = w.packets[--w.count];
+        }
+        for (unsigned i = 0; i < LINKS; ++i) pure_link_sender_tick(links[i].s, w.now);
+    }
+    PureLinkCounts heard;
+    pure_link_receiver_counts(r, &heard);
+    for (unsigned i = 0; i < LINKS; ++i) {
+        PureLinkCounts c;
+        pure_link_sender_counts(links[i].s, &c);
+        unsigned why = pure_link_sender_refusal(links[i].s);
+        printf("{\"case\":\"refused\",\"source\":\"%s\",\"input\":\"%s\",\"why\":%u,\"refused\":%llu,\"packets\":%llu,"
+               "\"stored\":%llu}\n", links[i].source, links[i].input, why, (unsigned long long)c.refused,
+               (unsigned long long)links[i].packets, (unsigned long long)pure_link_sender_stored(links[i].s));
+        if (links[i].why) /* refused at once, then at 1, 3, 7, 15, 31 and 63 s: a packet each */
+            CHECK(why == links[i].why && c.refused == 7 && links[i].packets <= 30,
+                  "refused %s to %s: why %u (not %u), refused %llu times, %llu packets", links[i].source,
+                  links[i].input, why, links[i].why, (unsigned long long)c.refused, (unsigned long long)links[i].packets);
+        else /* taken at the try after 10 s (15 s on) */
+            CHECK(!why && c.refused == 4 && pure_link_sender_stored(links[i].s) == 200 && w.heard == 200,
+                  "refused until taken: why %u, refused %llu times, %llu stored, %llu heard", why,
+                  (unsigned long long)c.refused, (unsigned long long)pure_link_sender_stored(links[i].s),
+                  (unsigned long long)w.heard);
+        pure_link_sender_free(links[i].s);
+    }
+    CHECK(heard.refused >= 3 * 7 + 4, "refused: the receiver said so %llu times", (unsigned long long)heard.refused);
+    pure_link_receiver_free(r);
+    pure_seal_key_free(a), pure_seal_key_free(b);
+    free(w.packets);
+}
+
 /* The largest send (PURE_LINK_RECORD_MOST) goes whole, over a receiver that
  * never heard of the stream before; one byte more is not taken to send. */
 typedef struct { World *w; PureLinkSender *s; } Largest;
@@ -858,6 +939,7 @@ int main(void) {
     impostors(dir);
     flood(dir);
     outputs(dir);
+    refusals(dir);
     largest(dir);
     char command[600];
     snprintf(command, sizeof(command), "rm -rf %s", dir);