Remember the epoch of each of a peer's runs that send, not one

A receiver's memo kept one run's epoch a peer. A node whose outputs
send to this one sends in a run each, so after the receiver started
again (or let the peer rest) all runs but the last accepted were told
LOST and began their epochs again: records under the static keys for a
round trip, and an X25519 each. Now a run whose peer's place holds
another run still sending keeps its epoch beside, in one of two of
1,024 places its peer and run find (160 KiB at the file's head: the
memo is PLM3, and a PLM2 one is begun again). A run's place beside goes
once its peer's place has it, so an epoch's secrets stay only until the
next one's are.

test-link's three outputs of one node, each in epoch 1, go on after the
receiver starts again: no LOST, no X25519 (before: two LOSTs, an
accept). 012 says so.

3 files changed, 147 insertions, 51 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 1106ac5..318441e 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -1544,23 +1544,28 @@ static void drop_first(Stream *st) {
     if (!--st->held_count) st->held_first = 0;
 }
 
-/* The memo of secrets (pure_link_remember): a header, then a place a peer
- * in the order allowed, each the first 28 bytes of its key, a check (FNV-1a
- * of its whole key and the secret) and the secret. A place whose check
- * fails (never made, another key, cut short by a crash) is made again.
- * From MEMO_EPOCH on, the epoch last accepted for the peer's run: the run,
- * the epoch, a check (FNV-1a of the rest), its secret, the one before, the
- * key the sender offered and ours; written secrets first, the run last. A
- * node that starts again goes on in the sender's epoch, no key made. */
+/* The memo of secrets (pure_link_remember): a header, MEMO_MORE places
+ * beside (below), then a place a peer in the order allowed, each the first
+ * 28 bytes of its key, a check (FNV-1a of its whole key and the secret) and
+ * the secret. A place whose check fails (never made, another key, cut short
+ * by a crash) is made again. From MEMO_EPOCH on, the epoch last accepted
+ * for the peer's run: the run, the epoch, a check (FNV-1a of the rest), its
+ * secret, the one before, the key the sender offered and ours; written
+ * secrets first, the run last. A node that starts again goes on in the
+ * sender's epoch, no key made. A peer's other runs that send while the one
+ * its place has does (a node's outputs to this one) keep theirs beside:
+ * the peer (+ 1), then an epoch as a place has it, in one of two places
+ * its peer and run find. */
 enum { MEMO_HEAD = 64, MEMO_PLACE = 256, MEMO_EPOCH = 64, MEMO_EPOCH_SIZE = 144 };
-static const unsigned char memo_magic[8] = {'P', 'L', 'M', '2', 0, 0, 0, 0};
+enum { MEMO_MORE = 1024, MEMO_MORE_PLACE = 16 + MEMO_EPOCH_SIZE, MEMO_PLACES = MEMO_HEAD + MEMO_MORE * MEMO_MORE_PLACE };
+static const unsigned char memo_magic[8] = {'P', 'L', 'M', '3', 0, 0, 0, 0};
 
 static uint32_t memo_check(const unsigned char key[PURE_SEAL_KEY], const unsigned char secret[32]) {
     return fnv(fnv(2166136261u, key, PURE_SEAL_KEY), secret, 32);
 }
 
 static unsigned char *memo_place(const PureLinkReceiver *s, uint32_t peer) {
-    size_t at = MEMO_HEAD + (size_t)peer * MEMO_PLACE;
+    size_t at = MEMO_PLACES + (size_t)peer * MEMO_PLACE;
     return s->memo && at + MEMO_PLACE <= s->memo_size ? s->memo + at : NULL;
 }
 
@@ -1568,18 +1573,53 @@ static uint32_t epoch_check(const unsigned char *e) {
     return fnv(fnv(2166136261u, e, 12), e + 16, MEMO_EPOCH_SIZE - 16);
 }
 
-/* The epoch the memo has for the peer's run (its place's epoch part), or NULL. */
+static bool epoch_of(const unsigned char *e, uint64_t run) { return get64(e) == run && get32(e + 12) == epoch_check(e); }
+
+/* The two places beside where a peer's run may be kept. */
+static void memo_beside(const PureLinkReceiver *s, uint32_t peer, uint64_t run, unsigned char *ways[2]) {
+    unsigned char key[12];
+    put32(key, peer), put64(key + 4, run);
+    size_t k = fnv(2166136261u, key, sizeof(key)) & (MEMO_MORE - 1);
+    ways[0] = s->memo + MEMO_HEAD + k * MEMO_MORE_PLACE;
+    ways[1] = s->memo + MEMO_HEAD + (k ^ 1) * MEMO_MORE_PLACE;
+}
+
+/* The epoch the memo has for the peer's run (its place's epoch part, or
+ * beside), or NULL. */
 static const unsigned char *memo_epoch(const PureLinkReceiver *s, uint32_t peer, uint64_t run) {
     const unsigned char *place = memo_place(s, peer), *e = place ? place + MEMO_EPOCH : NULL;
-    return e && get64(e) == run && get32(e + 12) == epoch_check(e) ? e : NULL;
+    if (!e || epoch_of(e, run)) return e;
+    unsigned char *ways[2];
+    memo_beside(s, peer, run, ways);
+    for (unsigned w = 0; w < 2; ++w)
+        if (get32(ways[w]) == peer + 1 && epoch_of(ways[w] + 16, run)) return ways[w] + 16;
+    return NULL;
 }
 
-static void remember_epoch(PureLinkReceiver *s, uint32_t peer, const Run *r) {
-    unsigned char *place = memo_place(s, peer);
+/* The epoch accepted for r, in the peer's place, or beside when the run
+ * the place has still sends: in the way that has r, else an empty one,
+ * else the first. What a run had beside goes once its place has it (an
+ * epoch's secrets are kept only until the next one's are). */
+static Run *run_found(Active *a, uint64_t run);
+static void remember_epoch(PureLinkReceiver *s, uint32_t peer, const Run *r, uint64_t now) {
+    unsigned char *place = memo_place(s, peer), *ways[2], *beside = NULL;
     if (!place) return;
     unsigned char *e = place + MEMO_EPOCH;
+    Active *a = s->peers[peer].active;
+    const Run *other = a && get64(e) != r->run && epoch_of(e, get64(e)) ? run_found(a, get64(e)) : NULL;
+    memo_beside(s, peer, r->run, ways);
+    if (other && now - other->seen < s->idle) {
+        for (unsigned w = 0; w < 2 && !beside; ++w)
+            if (get32(ways[w]) == peer + 1 && get64(ways[w] + 16) == r->run) beside = ways[w];
+        for (unsigned w = 0; w < 2 && !beside; ++w)
+            if (!epoch_of(ways[w] + 16, get64(ways[w] + 16))) beside = ways[w];
+        e = (beside = beside ? beside : ways[0]) + 16;
+    } else
+        for (unsigned w = 0; w < 2; ++w)
+            if (get32(ways[w]) == peer + 1 && get64(ways[w] + 16) == r->run) explicit_bzero(ways[w], MEMO_MORE_PLACE);
     memset(e, 0, 16);
     __atomic_thread_fence(__ATOMIC_RELEASE);
+    if (beside) put32(beside, peer + 1);
     memcpy(e + 16, r->next.secret, 32);
     memcpy(e + 48, r->now.secret, 32);
     memcpy(e + 80, r->offered, PURE_SEAL_KEY);
@@ -1594,7 +1634,7 @@ bool pure_link_remember(PureLinkReceiver *s, const char *path, char *error, size
     int fd = open(path, O_RDWR | O_CREAT | O_NOFOLLOW | O_CLOEXEC, 0600);
     struct stat info;
     unsigned char head[MEMO_HEAD];
-    size_t needed = MEMO_HEAD + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
+    size_t needed = MEMO_PLACES + (s->peer_count ? s->peer_count : 1) * MEMO_PLACE;
     bool ok = fd >= 0 && !fstat(fd, &info) && S_ISREG(info.st_mode) && !(info.st_mode & 077);
     /* Another node's key, or none yet: begun again. */
     bool ours = ok && pread(fd, head, MEMO_HEAD, 0) == MEMO_HEAD && !memcmp(head, memo_magic, 8) &&
@@ -2115,7 +2155,7 @@ static void accept_waiting(PureLinkReceiver *s, uint64_t now) {
         if (base == a->secret && r->now.n && epoch_made(s, peer, r, &zero, 0, a->secret)) /* begun again from 0: so are its answers */
             epoch_now(s, peer, r, &zero);
         explicit_bzero(&zero, sizeof(zero));
-        remember_epoch(s, a->peer, r);
+        remember_epoch(s, a->peer, r, now);
         mark_due(s, r);
     }
     s->offer_count = 0;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index df15246..f131e99 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -244,7 +244,9 @@ void pure_link_idle(PureLinkReceiver *, uint64_t ms);
  * (path, 0600; begun again for another node's key): a place a sender in
  * the order allowed, mapped, so a sender at rest takes no memory for it.
  * A node that starts again then makes no X25519 for a sender it heard from
- * before. After the senders are allowed, before the first packet. */
+ * before, and goes on in the epoch of each of its runs (a key's runs that
+ * send beside its first, as a node's outputs to this one do, in 1,024
+ * places beside). After the senders are allowed, before the first packet. */
 bool pure_link_remember(PureLinkReceiver *, const char *path, char *error, size_t size);
 /* A packet that came from an address (from_size bytes, at most
  * PURE_LINK_ADDRESS kept): answers to its sender go there. */
diff --git a/link/test-link.c b/link/test-link.c
index f609fe5..7655025 100644
--- a/link/test-link.c
+++ b/link/test-link.c
@@ -657,8 +657,15 @@ static void flood(const char *dir) {
 /* Three outputs of one node to this one: three senders with one key, each a
  * run of its own, sending at once. Each run keeps a place of its own (none
  * pushes another out, which sent its answers to the wrong run), each keyed
- * once, and every stream's records are delivered once, in order. */
-typedef struct { uint64_t now, heard[3]; PureLinkSender *senders[3]; PureLinkDone done[64]; size_t done_count; } Outputs;
+ * once, and every stream's records are delivered once, in order. Then,
+ * each in an epoch past 0, the receiver starts again (kill -9) with its
+ * memo: every run goes on in its epoch, none told LOST, no X25519 made. */
+typedef struct {
+    uint64_t now, heard[3], queued;
+    PureLinkSender *senders[3];
+    PureLinkDone done[64];
+    size_t done_count;
+} Outputs;
 static int output_input(void *context, const char *name, uint32_t shape) {
     (void)context;
     return shape != 0x5eed ? -1 : !strcmp(name, "one") ? 0 : !strcmp(name, "two") ? 1 : !strcmp(name, "three") ? 2 : -1;
@@ -695,49 +702,96 @@ static bool output_answer(void *context, const unsigned char *packet, size_t siz
     for (unsigned i = 0; i < 3; ++i) pure_link_sender_take(o->senders[i], packet, size, o->now); /* its run's opens */
     return true;
 }
-static void outputs(const char *dir) {
-    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY], out[PURE_LINK_DATAGRAM_MOST];
+static PureLinkReceiver *outputs_receiver(Outputs *o, const PureSealKey *b, const unsigned char *public_b,
+                                          const unsigned char *public_a, const char *memo) {
     char error[256];
-    static unsigned char bytes[20200];
-    PureSealKey *a = key(dir, "a.key", public_a), *b = key(dir, "b.key", public_b);
-    Outputs o = {0};
-    PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){output_input, output_deliver, &o}, error,
+    PureLinkReceiver *r = pure_link_receiver(b, public_b, &(PureLinkHost){output_input, output_deliver, o}, error,
                                              sizeof(error));
-    if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error))) exit(1);
-    const char *inputs[3] = {"one", "two", "three"};
-    for (unsigned i = 0; i < 3; ++i) {
-        PureLinkSender *s = o.senders[i] = pure_link_sender("test.ordered", inputs[i], 0x5eed, a, public_a);
-        if (!s || !pure_link_sender_peer(s, public_b)) exit(1);
-        for (uint64_t seq = 1; seq <= 1000; ++seq) {
+    if (!r || !pure_link_allow(r, "test.ordered", public_a, error, sizeof(error)) ||
+        !pure_link_remember(r, memo, error, sizeof(error))) {
+        fprintf(stderr, "test-link: %s\n", error);
+        exit(1);
+    }
+    return r;
+}
+/* Each sender's records to last queued. */
+static void outputs_queue(Outputs *o, uint64_t last) {
+    static unsigned char bytes[20200];
+    for (unsigned i = 0; i < 3; ++i)
+        for (uint64_t seq = o->queued + 1; seq <= last; ++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);
+            if (!pure_link_sender_queue(o->senders[i], seq, bytes, size)) exit(1);
         }
-    }
-    uint64_t number;
-    bool all = false;
-    for (; o.now < 60000 * MS && !all; o.now += MS) {
+    o->queued = last;
+}
+/* Until each sender's records are stored, or a minute; when epochs, a record
+ * more each 50 ms until each is in an epoch past 0 (an offer goes in a
+ * packet of records). */
+static void outputs_until(Outputs *o, PureLinkReceiver *r, bool epochs) {
+    unsigned char out[PURE_LINK_DATAGRAM_MOST];
+    uint64_t number, end = o->now + 60000 * MS;
+    for (bool all = false; o->now < end && !all; o->now += MS) {
+        if (epochs && o->now % (50 * MS) == 0) outputs_queue(o, o->queued + 1);
         for (unsigned i = 0; i < 3; ++i)
-            for (size_t size; (size = pure_link_sender_packet(o.senders[i], out, false, o.now, &number));)
-                pure_link_receiver_take(r, out, size, "sender0", 7, o.now);
+            for (size_t size; (size = pure_link_sender_packet(o->senders[i], out, false, o->now, &number));)
+                pure_link_receiver_take(r, out, size, "sender0", 7, o->now);
         PureLinkDone told[64];
-        size_t count = o.done_count;
-        memcpy(told, o.done, count * sizeof(*told));
-        o.done_count = 0;
-        pure_link_receiver_turn(r, told, count, o.now, output_answer, &o);
+        size_t count = o->done_count;
+        memcpy(told, o->done, count * sizeof(*told));
+        o->done_count = 0;
+        pure_link_receiver_turn(r, told, count, o->now, output_answer, o);
         all = true;
         for (unsigned i = 0; i < 3; ++i) {
-            pure_link_sender_tick(o.senders[i], o.now);
-            all = all && pure_link_sender_stored(o.senders[i]) == 1000;
+            PureLinkCounts c;
+            pure_link_sender_tick(o->senders[i], o->now);
+            pure_link_sender_counts(o->senders[i], &c);
+            all = all && pure_link_sender_stored(o->senders[i]) == o->queued && (!epochs || c.epochs >= 1);
         }
     }
-    PureLinkCounts heard;
-    pure_link_receiver_counts(r, &heard);
-    CHECK(all && o.heard[0] == 1000 && o.heard[1] == 1000 && o.heard[2] == 1000 && heard.keyed == 3,
-          "outputs: heard %llu, %llu, %llu of 1000 each, %llu runs keyed (3 runs)", (unsigned long long)o.heard[0],
-          (unsigned long long)o.heard[1], (unsigned long long)o.heard[2], (unsigned long long)heard.keyed);
-    printf("{\"case\":\"outputs\",\"runs\":3,\"records\":3000,\"virtual_ms\":%llu,\"keyed\":%llu,\"packets\":%llu}\n",
-           (unsigned long long)(o.now / MS), (unsigned long long)heard.keyed, (unsigned long long)heard.received);
+}
+static void outputs(const char *dir) {
+    unsigned char public_a[PURE_SEAL_KEY], public_b[PURE_SEAL_KEY];
+    char memo[512];
+    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);
+    Outputs o = {0};
+    PureLinkReceiver *r = outputs_receiver(&o, b, public_b, public_a, memo);
+    const char *inputs[3] = {"one", "two", "three"};
+    for (unsigned i = 0; i < 3; ++i)
+        if (!(o.senders[i] = pure_link_sender("test.ordered", inputs[i], 0x5eed, a, public_a)) ||
+            !pure_link_sender_peer(o.senders[i], public_b))
+            exit(1);
+    outputs_queue(&o, 500);
+    outputs_until(&o, r, false);
+    PureLinkCounts first;
+    pure_link_receiver_counts(r, &first);
+    uint64_t took = o.now / MS;
+    outputs_until(&o, r, true); /* a new epoch a second for a peer: the third's after two */
+    pure_link_receiver_free(r); /* kill -9: what it held is gone, the memo stays */
+    o.done_count = 0;
+    r = outputs_receiver(&o, b, public_b, public_a, memo);
+    outputs_queue(&o, 1000);
+    outputs_until(&o, r, false);
+    PureLinkCounts again, sent = {0};
+    pure_link_receiver_counts(r, &again);
+    for (unsigned i = 0; i < 3; ++i) {
+        PureLinkCounts c;
+        pure_link_sender_counts(o.senders[i], &c);
+        sent.fallbacks += c.fallbacks;
+        CHECK(pure_link_sender_stored(o.senders[i]) == 1000 && o.heard[i] == 1000, "outputs: %u stored %llu, heard %llu",
+              i, (unsigned long long)pure_link_sender_stored(o.senders[i]), (unsigned long long)o.heard[i]);
+    }
+    CHECK(first.keyed == 3 && took < 1000, "outputs: %llu runs keyed (3 runs), stored in %llu ms",
+          (unsigned long long)first.keyed, (unsigned long long)took);
+    CHECK(!sent.fallbacks && !again.secrets && again.keyed == 3,
+          "outputs, started again: %llu epochs taken for lost, %llu secrets made, %llu runs keyed",
+          (unsigned long long)sent.fallbacks, (unsigned long long)again.secrets, (unsigned long long)again.keyed);
+    printf("{\"case\":\"outputs\",\"runs\":3,\"records\":3000,\"stored_ms\":%llu,\"keyed\":%llu,"
+           "\"keyed_again\":%llu,\"secrets_again\":%llu,\"fallbacks\":%llu}\n",
+           (unsigned long long)took, (unsigned long long)first.keyed, (unsigned long long)again.keyed,
+           (unsigned long long)again.secrets, (unsigned long long)sent.fallbacks);
     for (unsigned i = 0; i < 3; ++i) pure_link_sender_free(o.senders[i]);
     pure_link_receiver_free(r);
     pure_seal_key_free(a), pure_seal_key_free(b);