Let a link's receiver find its streams' inputs again (017, stage 3)

A stream finds its input once, when it is made; an app that starts again
with another module may have the input at another number, of another
shape, or none. pure_link_receiver_rebind marks the streams to inputs of
a prefix stale: each finds its input again by name, with the fields it
was laid out with, before its next delivery and at its sender's next
packet, keeping what it holds; one none takes is refused as a new one
would be. pure_link_listener_rebind asks it of the listener's thread.
test-link's rebound case stops its app, starts it with the input at
another number, then of another shape taken by name: all 3,000 records
come once, in order (without the rebind, none after the stop).

5 files changed, 173 insertions, 5 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 169705f..ea38fab 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -1332,6 +1332,7 @@ typedef struct Stream {
     uint64_t run_id;
     unsigned char *fields;       /* the fields its records are laid out with, when its input took them by name */
     size_t fields_size;
+    bool stale;                  /* its input to be found again (pure_link_receiver_rebind) */
 } Stream;
 struct Active {
     uint32_t peer;               /* its place among the peers */
@@ -1897,21 +1898,32 @@ static bool keep_fields(Stream *st, const unsigned char *fields, size_t size) {
  * fields named, which come with FIELDS), or memory runs out (why 0). A
  * stream the sender now sends in another shape (it started again with
  * another module) starts over with what it holds let go: the sender sends
- * those again, laid out anew. */
+ * those again, laid out anew. A stale one (its input may be another now)
+ * finds its input again, with the fields it was laid out with unless these
+ * come, and keeps what it holds. */
 static Stream *stream_of(PureLinkReceiver *s, Active *a, const char *source, const char *input, uint32_t shape,
                          const unsigned char *fields, size_t fields_size, unsigned *why) {
     *why = 0;
     Stream *st = NULL;
     for (unsigned k = 0; k < a->stream_count && !st; ++k)
         if (!strcmp(a->streams[k]->input_name, input) && !strcmp(a->streams[k]->source, source)) st = a->streams[k];
-    if (st && st->shape == shape) return st;
+    if (st && st->shape == shape && !st->stale) return st;
+    bool again = st && st->shape == shape; /* stale */
+    const unsigned char *named = fields ? fields : again ? st->fields : NULL;
+    size_t named_size = fields ? fields_size : again ? st->fields_size : 0;
     int index = !allowed(s, &s->peers[a->peer], source) ? -PURE_LINK_NOT_ALLOWED :
-                s->host.input(s->host.context, input, shape, fields, fields ? fields_size : 0);
+                s->host.input(s->host.context, input, shape, named, named_size);
     if (index < 0) {
         *why = index == -PURE_LINK_OTHER_SHAPE || index == -PURE_LINK_NOT_ALLOWED || index == -PURE_LINK_FIELDS_LACKED
                    ? (unsigned)-index : PURE_LINK_NO_INPUT;
         return NULL;
     }
+    if (again) { /* its input found again; what it holds stays */
+        if (fields && !keep_fields(st, fields, fields_size)) return NULL;
+        st->input = (uint32_t)index;
+        st->stale = false;
+        return st;
+    }
     if (st) { /* another shape now */
         if (st->out) { *why = PURE_LINK_OTHER_SHAPE; return NULL; } /* its last delivery first */
         if (!keep_fields(st, fields, fields_size)) return NULL;
@@ -2077,6 +2089,12 @@ static void hold(Stream *st, uint64_t seq, uint32_t size, uint32_t offset, const
  * False when the host could not take them now. */
 static bool hand_over(PureLinkReceiver *s, Stream *st) {
     if (st->out) return true;
+    if (st->stale) { /* its input found again first: none takes them now, and they wait */
+        int index = s->host.input(s->host.context, st->input_name, st->shape, st->fields, st->fields_size);
+        if (index < 0) return false;
+        st->input = (uint32_t)index;
+        st->stale = false;
+    }
     if (st->known) { /* the line as it is, up to BATCH records */
         line_pull(st);
         if (!st->line_count) return true;
@@ -2641,6 +2659,18 @@ void pure_link_receiver_turn(PureLinkReceiver *s, const PureLinkDone *done, size
     sweep(s, now);
 }
 
+void pure_link_receiver_rebind(PureLinkReceiver *s, const char *prefix) {
+    size_t length = strlen(prefix);
+    for (size_t k = 0; k < s->active_count; ++k) {
+        Active *a = s->actives[k];
+        for (unsigned i = 0; i < a->stream_count; ++i)
+            if (!strncmp(a->streams[i]->input_name, prefix, length)) {
+                a->streams[i]->stale = true;
+                mark_ready(s, a->streams[i]); /* what it holds handed over, its input found again */
+            }
+    }
+}
+
 uint64_t pure_link_receiver_due(const PureLinkReceiver *s) {
     uint64_t due = s->active_count || s->ready_count ? s->swept + SWEEP_EVERY : 0;
     return s->waiting && s->waiting < due ? s->waiting : due;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index feff543..4f263b2 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -278,6 +278,12 @@ void pure_link_receiver_take(PureLinkReceiver *, const unsigned char *packet, si
  * go. */
 void pure_link_receiver_turn(PureLinkReceiver *, const PureLinkDone *done, size_t count, uint64_t now,
                              PureLinkSend *send, void *context);
+/* The host's inputs whose names begin with prefix may be others now (an app
+ * started again with another module): their streams find theirs again, by
+ * name and with the fields they were laid out with, before their next
+ * delivery, a delivery out first done; what they hold stays. One no input
+ * takes now is refused at its sender's next packet, as a new one would be. */
+void pure_link_receiver_rebind(PureLinkReceiver *, const char *prefix);
 /* When a turn is due with no packet (ns), or 0 for none. */
 uint64_t pure_link_receiver_due(const PureLinkReceiver *);
 void pure_link_receiver_counts(const PureLinkReceiver *, PureLinkCounts *);
diff --git a/link/link.c b/link/link.c
index a9270fe..2e1f359 100644
--- a/link/link.c
+++ b/link/link.c
@@ -426,9 +426,11 @@ struct PureLinkListener {
     int fd, wake, family;
     pthread_t thread;
     bool started;
-    pthread_mutex_t lock; /* the done list and stopping */
+    pthread_mutex_t lock; /* the done and rebind lists, and stopping */
     PureLinkDone *done;
     size_t done_count, done_capacity;
+    char **rebinds; /* the prefixes of the inputs to find again */
+    size_t rebind_count, rebind_capacity;
     bool stopping;
 };
 
@@ -503,10 +505,18 @@ static void *run_listener(void *context) {
         }
         pthread_mutex_lock(&s->lock);
         PureLinkDone *done = s->done;
-        size_t count = s->done_count;
+        size_t count = s->done_count, rebinds = s->rebind_count;
+        char **prefixes = s->rebinds;
         s->done = NULL;
         s->done_count = s->done_capacity = 0;
+        s->rebinds = NULL;
+        s->rebind_count = s->rebind_capacity = 0;
         pthread_mutex_unlock(&s->lock);
+        for (size_t k = 0; k < rebinds; ++k) {
+            pure_link_receiver_rebind(s->receiver, prefixes[k]);
+            free(prefixes[k]);
+        }
+        free(prefixes);
         pure_link_receiver_turn(s->receiver, done, count, now, send_answer, s);
         free(done);
     }
@@ -564,6 +574,20 @@ void pure_link_done(PureLinkListener *s, PureLinkDone done) {
     wake(s->wake);
 }
 
+void pure_link_listener_rebind(PureLinkListener *s, const char *prefix) {
+    char *kept = strdup(prefix);
+    pthread_mutex_lock(&s->lock);
+    if (kept && s->rebind_count == s->rebind_capacity) {
+        size_t capacity = s->rebind_capacity ? 2 * s->rebind_capacity : 4;
+        char **grown = realloc(s->rebinds, capacity * sizeof(*grown));
+        if (grown) s->rebinds = grown, s->rebind_capacity = capacity;
+    }
+    if (!kept || s->rebind_count == s->rebind_capacity) abort(); /* an input to find anew cannot be lost */
+    s->rebinds[s->rebind_count++] = kept;
+    pthread_mutex_unlock(&s->lock);
+    wake(s->wake);
+}
+
 void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) {
     if (!s) return;
     if (s->started) {
@@ -576,6 +600,8 @@ void pure_link_listener_close(PureLinkListener *s, PureLinkCounts *counts) {
     if (counts) pure_link_receiver_counts(s->receiver, counts);
     pure_link_receiver_free(s->receiver);
     free(s->done);
+    for (size_t k = 0; k < s->rebind_count; ++k) free(s->rebinds[k]);
+    free(s->rebinds);
     if (s->fd >= 0) close(s->fd);
     if (s->wake >= 0) close(s->wake);
     pthread_mutex_destroy(&s->lock);
diff --git a/link/link.h b/link/link.h
index 9846af4..40fc9f9 100644
--- a/link/link.h
+++ b/link/link.h
@@ -51,5 +51,9 @@ PureLinkListener *pure_link_listen(PureLinkReceiver *, unsigned port, char *erro
 bool pure_link_serve(PureLinkListener *, char *error, size_t size);
 /* How a delivery went: link-protocol.h's PureLinkDone, with its token. */
 void pure_link_done(PureLinkListener *, PureLinkDone);
+/* The host's inputs whose names begin with prefix may be others now: the
+ * receiver's streams to them find theirs again (pure_link_receiver_rebind),
+ * on the listener's thread; any thread. */
+void pure_link_listener_rebind(PureLinkListener *, const char *prefix);
 void pure_link_listener_close(PureLinkListener *, PureLinkCounts *counts);
 #endif
diff --git a/link/test-link.c b/link/test-link.c
index 2e7509a..fb64d52 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -774,6 +774,107 @@ static void outputs_until(Outputs *o, PureLinkReceiver *r, bool epochs) {
         }
     }
 }
+/* The receiving app starts again while records come (docs/design/017's
+ * stage 3): first with no input of that name (the receiver holds what
+ * comes), then with the input at another number, then with one of another
+ * shape that takes the sender's fields by name. Told each time
+ * (pure_link_receiver_rebind), the stream finds its input again and keeps
+ * what it holds: every record is applied once, in order, and a delivery
+ * for the input as it was is never taken. */
+typedef struct {
+    PureLinkSender *sender;
+    uint64_t now, heard, stale, named, to[3];
+    unsigned phase; /* 0: input 3; 1: none; 2: input 4; 3: input 5, by name */
+    PureLinkDone done[64];
+    size_t done_count;
+} Rebound;
+static int rebound_input(void *context, const char *name, uint32_t shape, const unsigned char *fields, size_t size) {
+    const Rebound *r = context;
+    if (strcmp(name, "order") || r->phase == 1) return -PURE_LINK_NO_INPUT;
+    if (r->phase < 3) return shape == 0x5eed ? (int)r->phase + 2 + (r->phase == 0) : -PURE_LINK_OTHER_SHAPE;
+    if (!fields) return -PURE_LINK_OTHER_SHAPE;
+    return size == sizeof(NAMED) && !memcmp(fields, NAMED, size) ? 5 : -PURE_LINK_FIELDS_LACKED;
+}
+static bool rebound_deliver(void *context, const PureLinkDelivery *d) {
+    Rebound *r = context;
+    unsigned now = r->phase == 0 ? 3 : r->phase == 2 ? 4 : r->phase == 3 ? 5 : 0;
+    if (d->input != now) { /* found for the app as it was: not taken, kept */
+        ++r->stale;
+        return false;
+    }
+    if (r->done_count == sizeof(r->done) / sizeof(r->done[0])) return false;
+    CHECK(now != 5 || (d->fields && d->fields_size == sizeof(NAMED)), "rebound: records by name came without fields");
+    r->named += now == 5;
+    ++r->to[now - 3];
+    uint64_t heard = r->heard;
+    bool gap = false;
+    const unsigned char *at = d->records;
+    for (unsigned k = 0; k < d->count; ++k) {
+        uint32_t size = (uint32_t)at[0] | (uint32_t)at[1] << 8 | (uint32_t)at[2] << 16 | (uint32_t)at[3] << 24;
+        uint64_t seq = d->first + k;
+        if (seq > heard + 1) {
+            gap = true;
+            break;
+        }
+        if (seq == heard + 1) {
+            bool same = size == record_size(seq);
+            for (size_t i = 0; same && i < size; ++i) same = at[4 + i] == record_byte(seq, i);
+            CHECK(same, "rebound: record %llu is not as it was sent", (unsigned long long)seq);
+            heard = seq;
+        }
+        at += 4 + size;
+    }
+    r->heard = heard;
+    r->done[r->done_count++] = (PureLinkDone){d->token, heard, true, gap, false};
+    return true;
+}
+static bool rebound_answer(void *context, const unsigned char *packet, size_t size, const unsigned char *to,
+                           size_t to_size) {
+    (void)to, (void)to_size;
+    Rebound *r = context;
+    pure_link_sender_take(r->sender, packet, size, r->now);
+    return true;
+}
+static void rebound(const char *dir) {
+    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST];
+    char memo[512], error[256];
+    PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
+    snprintf(memo, sizeof(memo), "%s/b.key.links", dir);
+    unlink(memo);
+    Rebound r = {0};
+    PureLinkReceiver *receiver = pure_link_receiver(b, public_b, &(PureLinkHost){rebound_input, rebound_deliver, &r},
+                                                    error, sizeof(error));
+    if (!receiver || !pure_link_allow(receiver, "test.ordered", public_a, error, sizeof(error))) exit(1);
+    r.sender = sender_of(a, public_a, public_b, 1, 3000);
+    if (!pure_link_sender_fields(r.sender, NAMED, sizeof(NAMED))) exit(1);
+    uint64_t number, back = 0, end = 60000 * MS;
+    for (; r.now < end && pure_link_sender_stored(r.sender) < 3000; r.now += MS) {
+        unsigned was = r.phase;
+        if (r.phase == 0 && r.heard >= 1000) r.phase = 1, back = r.now + 300 * MS; /* stopped */
+        else if (r.phase == 1 && r.now >= back) r.phase = 2; /* started again, the input another */
+        else if (r.phase == 2 && r.heard >= 2000) r.phase = 3; /* and again, of another shape */
+        if (r.phase != was) pure_link_receiver_rebind(receiver, "ord");
+        for (size_t size; (size = pure_link_sender_packet(r.sender, out, false, r.now, &number));)
+            pure_link_receiver_take(receiver, out, size, "sender0", 7, r.now);
+        PureLinkDone told[64];
+        size_t count = r.done_count;
+        memcpy(told, r.done, count * sizeof(*told));
+        r.done_count = 0;
+        pure_link_receiver_turn(receiver, told, count, r.now, rebound_answer, &r);
+        pure_link_sender_tick(r.sender, r.now);
+    }
+    CHECK(r.heard == 3000 && pure_link_sender_stored(r.sender) == 3000, "rebound: heard %llu, stored %llu of 3000",
+          (unsigned long long)r.heard, (unsigned long long)pure_link_sender_stored(r.sender));
+    CHECK(r.phase == 3 && r.to[0] && r.to[1] && r.to[2], "rebound: deliveries %llu, %llu, %llu (phase %u)",
+          (unsigned long long)r.to[0], (unsigned long long)r.to[1], (unsigned long long)r.to[2], r.phase);
+    printf("{\"case\":\"rebound\",\"records\":3000,\"virtual_ms\":%llu,\"to_3\":%llu,\"to_4\":%llu,\"to_5_by_name\":%llu,"
+           "\"stale_refused\":%llu}\n", (unsigned long long)(r.now / MS), (unsigned long long)r.to[0],
+           (unsigned long long)r.to[1], (unsigned long long)r.to[2], (unsigned long long)r.stale);
+    pure_link_sender_free(r.sender);
+    pure_link_receiver_free(receiver);
+    pure_seal_key_free(a), pure_seal_key_free(b);
+}
+
 static void outputs(const char *dir) {
     unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY];
     char memo[512];
@@ -1162,6 +1263,7 @@ int main(void) {
     impostors(dir);
     flood(dir);
     outputs(dir);
+    rebound(dir);
     refusals(dir);
     refused_restarted(dir);
     named(dir);