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.

4 files changed, 158 insertions, 55 deletions

diff --git a/docs/design/012-link-protocol.md b/docs/design/012-link-protocol.md
index 359dfbe..c99bdb0 100644
--- a/docs/design/012-link-protocol.md
+++ b/docs/design/012-link-protocol.md
@@ -254,8 +254,13 @@ LOST는 그 패킷의 번호를 담아 에폭 0의 키로 봉인한다. 답 run�
 
 **쉬는 피어.** 30초 동안 조용하고 쥔 것이 없는 피어는 run, 주소, 스트림을 놓는다(지역 선택). 다음 패킷이 다시 만든다.
 
-**메모.** 선 위의 일은 아니지만 재시작에 관계된다. 받는 쪽은 피어마다 S와, 그 피어의 run 하나에 대해 받아들인 에폭을 파일에
-적어 둘 수 있다. 다시 떠도 그 run은 비대칭 연산 없이 이어진다. 메모에 없는 run은 LOST를 받고 에폭을 다시 시작한다.
+**메모.** 선 위의 일은 아니지만 재시작에 관계된다. 받는 쪽은 피어마다 S와, 그 피어의 run마다 받아들인 에폭을 파일에 적어 둘
+수 있다.
+
+- 피어의 자리에는 run 하나를 둔다.
+- 그 run이 아직 보내는 동안 같은 키의 다른 run(한 노드의 출력 여럿)은 곁의 자리 1,024개에 둔다.
+- 다시 떠도, 쉬었다 돌아와도 이 run들은 비대칭 연산 없이 이어진다.
+- 메모에 없는 run은 LOST를 받고 에폭을 다시 시작한다.
 
 ## 수와 지역 선택
 
@@ -308,8 +313,10 @@ LOST는 그 패킷의 번호를 담아 에폭 0의 키로 봉인한다. 답 run�
 
 - 받는 쪽의 개인 키를 가진 이는 그 받는 쪽에게 어느 보내는 쪽으로든 봉인할 수 있다(Noise KK의 첫 메시지와 같은 KCI). 에폭은
   이것을 바꾸지 않는다. 다만 다른 run의 에폭 안으로는 들어오지 못한다.
-- 메모는 피어마다 run 하나의 에폭만 든다. 한 노드의 출력 여럿이 같은 노드로 가면, 받는 쪽이 다시 뜬 뒤나 30초 넘게 쉰 뒤에
-  나머지 run은 LOST를 받고 에폭 0에서 다시 시작한다.
+- 메모의 곁 자리는 1,024개이고 한 run이 두 자리 가운데 하나에 든다. 한 받는 쪽에 함께 보내는 "피어의 둘째 이후 run"이
+  그보다 많아 자리가 겹치면, 밀려난 run은 받는 쪽이 다시 뜬 뒤 LOST를 받고 에폭 0에서 다시 시작한다.
+- 메모는 받아들인 에폭의 비밀을 그 run의 다음 에폭이 덮을 때까지 디스크에 둔다. 다시 떠도 잇기 위해서다. 그 사이 메모
+  파일을 가져간 이는 그 에폭과 앞 에폭을 열 수 있다.
 - NAT 뒤에서 쉬는 기기에 먼저 보내는 일은 풀지 않는다.
 
 ## 이 명세와 구현
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);