Look for a link's receiver again when its probes go unanswered

A link resolved its receiver's name once, at its start: a receiver whose
store moved to another machine (another address under the same name)
was never found, and a sender whose own network changed kept a socket
that no longer went anywhere. Now, when three probes in a row go
unanswered, and at each probe after (a second apart at most), the link
looks the name up again and sends from a new socket connected there.
It looks before the probe goes, so the probe's answer comes to the new
socket (looking after it lost every answer, one socket after another).
A sender at rest does not look: only probes unanswered count.

make check moves a receiver (H3 of docs/design/011-pure-transport.md):
in namespaces of its own, the sender links to pure://receiver.test at
127.0.0.2; the receiver is killed, its store, key and memo copied, the
old address made unreachable, the name pointed at 127.0.0.3 and the
copy started there. Every order is applied once, and the moved receiver
makes no X25519.

4 files changed, 205 insertions, 26 deletions

diff --git a/link/link-protocol.c b/link/link-protocol.c
index 03c3abc..fe62b5f 100644
--- a/link/link-protocol.c
+++ b/link/link-protocol.c
@@ -1063,6 +1063,8 @@ bool pure_link_sender_shared(PureLinkSender *l, const unsigned char peer[PURE_SE
 
 uint64_t pure_link_sender_queued(const PureLinkSender *l) { return l->queued; }
 
+unsigned pure_link_sender_probes(const PureLinkSender *l) { return l->pto_count; }
+
 void pure_link_sender_epochs(PureLinkSender *l, uint64_t every_ns, uint64_t bytes) {
     l->epoch_every = every_ns;
     l->epoch_most = bytes;
diff --git a/link/link-protocol.h b/link/link-protocol.h
index c6ccc23..16fd683 100644
--- a/link/link-protocol.h
+++ b/link/link-protocol.h
@@ -161,6 +161,9 @@ size_t pure_link_sender_datagram(const PureLinkSender *);
 void pure_link_sender_refused(PureLinkSender *, uint64_t number);
 /* When the next timer is due (ns), or 0 for none. */
 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 *);
 /* 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);
diff --git a/link/link.c b/link/link.c
index 7b1b8d7..7027236 100644
--- a/link/link.c
+++ b/link/link.c
@@ -132,6 +132,7 @@ 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 */
 };
 
 PureLink *pure_link_open(const char *url, const char *source, uint32_t shape, const PureSealKey *own,
@@ -276,6 +277,46 @@ static void send_packets(PureLink *l, uint64_t now) {
     flush(l);
 }
 
+/* Where the receiver's name is now, a socket connected there in place of
+ * the one before: so a link follows a receiver moved to another address,
+ * and goes out a new socket when its own network changed. False (the
+ * socket as it was) when the name or a socket fails. */
+static bool look(PureLink *l, char *error, size_t size) {
+    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found = NULL;
+    int resolved = getaddrinfo(l->host, l->port, &hints, &found);
+    if (resolved) {
+        snprintf(error, size, "%s: %s", l->host, gai_strerror(resolved));
+        return false;
+    }
+    int fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0), family = found->ai_family;
+    bool ok = fd >= 0 && !connect(fd, found->ai_addr, found->ai_addrlen);
+    int failed = errno, buffer = 4 << 20;
+    freeaddrinfo(found);
+    if (!ok) {
+        if (fd >= 0) close(fd);
+        snprintf(error, size, "pure://%s:%s/%s: %s", l->host, l->port, l->input, strerror(failed));
+        return false;
+    }
+    setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
+    setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
+    /* Never cut up on the way (the don't-fragment bit): a packet the path
+     * does not carry is lost, as a probe of a larger size must be. */
+    int probe = IP_PMTUDISC_PROBE, probe6 = IPV6_PMTUDISC_PROBE;
+    if (family == AF_INET6) setsockopt(fd, IPPROTO_IPV6, IPV6_MTU_DISCOVER, &probe6, sizeof(probe6));
+    else setsockopt(fd, IPPROTO_IP, IP_MTU_DISCOVER, &probe, sizeof(probe));
+    /* GSO where the system has it: asked for each burst, never for the socket. */
+    int segment = DATAGRAM, none = 0;
+    l->gso = !l->no_gso && !setsockopt(fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)) &&
+             !setsockopt(fd, SOL_UDP, UDP_SEGMENT, &none, sizeof(none));
+    if (l->fd >= 0) close(l->fd);
+    l->fd = fd;
+    return true;
+}
+
+/* A link whose probes go unanswered three times in a row looks for its
+ * receiver again, and at each probe after (a second apart at most). */
+enum { LOOK_PROBES = 3 };
+
 static void *run_link(void *context) {
     PureLink *l = context;
     for (;;) {
@@ -297,6 +338,13 @@ static void *run_link(void *context) {
         now = now_ns();
         if (waits[0].revents) take_answers(l, now);
         pure_link_sender_tick(l->sender, now);
+        unsigned probes = pure_link_sender_probes(l->sender);
+        if (probes >= LOOK_PROBES && probes != l->looked) { /* before the probe goes, so its answer comes to the new socket */
+            pthread_mutex_unlock(&l->lock); /* a name looked up may take its time: the owner queues meanwhile */
+            look(l, NULL, 0);
+            pthread_mutex_lock(&l->lock);
+        }
+        l->looked = probes;
         send_packets(l, now);
         uint64_t stored = pure_link_sender_stored(l->sender);
         bool told = stored != l->told;
@@ -312,36 +360,14 @@ bool pure_link_start(PureLink *l, void (*notify)(void *), void *context, char *e
         snprintf(error, size, "pure://%s:%s/%s: no receiver's key (--seal)", l->host, l->port, l->input);
         return false;
     }
-    struct addrinfo hints = {.ai_family = AF_UNSPEC, .ai_socktype = SOCK_DGRAM}, *found = NULL;
-    int resolved = getaddrinfo(l->host, l->port, &hints, &found);
-    if (resolved) {
-        snprintf(error, size, "%s: %s", l->host, gai_strerror(resolved));
-        return false;
-    }
-    l->fd = socket(found->ai_family, SOCK_DGRAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0);
-    bool ok = l->fd >= 0 && !connect(l->fd, found->ai_addr, found->ai_addrlen);
-    int buffer = 4 << 20, family = found->ai_family;
-    freeaddrinfo(found);
-    if (ok) {
-        setsockopt(l->fd, SOL_SOCKET, SO_SNDBUF, &buffer, sizeof(buffer));
-        setsockopt(l->fd, SOL_SOCKET, SO_RCVBUF, &buffer, sizeof(buffer));
-        /* Never cut up on the way (the don't-fragment bit): a packet the
-         * path does not carry is lost, as a probe of a larger size must be. */
-        int probe = IP_PMTUDISC_PROBE, probe6 = IPV6_PMTUDISC_PROBE;
-        if (family == AF_INET6) setsockopt(l->fd, IPPROTO_IPV6, IPV6_MTU_DISCOVER, &probe6, sizeof(probe6));
-        else setsockopt(l->fd, IPPROTO_IP, IP_MTU_DISCOVER, &probe, sizeof(probe));
-    }
-    /* GSO where the system has it: asked for each burst, never for the socket. */
-    int segment = DATAGRAM, none = 0;
+    if (!look(l, error, size)) return false;
     l->segment = DATAGRAM;
-    l->gso = ok && !l->no_gso && !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &segment, sizeof(segment)) &&
-             !setsockopt(l->fd, SOL_UDP, UDP_SEGMENT, &none, sizeof(none));
-    ok = ok && l->keyed && (l->burst = malloc(BURST_BYTES + DATAGRAM_MOST)) && (l->taking = malloc(sizeof(*l->taking))) &&
-         (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0;
+    bool ok = l->keyed && (l->burst = malloc(BURST_BYTES + DATAGRAM_MOST)) && (l->taking = malloc(sizeof(*l->taking))) &&
+              (l->wake = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) >= 0;
     l->notify = notify;
     l->context = context;
     if (!ok || pthread_create(&l->thread, NULL, run_link, l)) {
-        snprintf(error, size, "pure://%s:%s/%s: cannot start (a socket, the keys)", l->host, l->port, l->input);
+        snprintf(error, size, "pure://%s:%s/%s: cannot start (the keys, memory)", l->host, l->port, l->input);
         return false;
     }
     pthread_setname_np(l->thread, "pure-link"); /* a link's CPU, seen apart */
diff --git a/tests/check.py b/tests/check.py
index 7be32d7..d567e80 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -946,6 +946,153 @@ print(json.dumps(check.server_links_udp(Path({str(aup)!r}), Path({str(pure)!r}),
     return json.loads(done.stdout.strip().splitlines()[-1])
 
 
+def server_links_moved(aup, pure, work):
+    """A receiver's store moved to another address (H3 of
+    docs/design/011-pure-transport.md), in namespaces of its own (unshare
+    -rmn: its network, and an /etc/hosts of its own): the sender links to
+    pure://receiver.test, which names 127.0.0.2. While clients buy, the
+    receiver is killed (-9), its store, key and memo (KEY.links) copied
+    elsewhere, 127.0.0.2 made unreachable, the name pointed at 127.0.0.3,
+    and the copy started there. The sender, its probes unanswered, looks
+    the name up again and goes on: every order it saved is applied once,
+    and the moved receiver makes no X25519 (its memo came with it). None
+    where there is no unshare or ip."""
+    import shutil
+    unshare = shutil.which('unshare')
+    ip = shutil.which('ip') or next((path for path in ('/usr/sbin/ip', '/sbin/ip', '/bin/ip') if Path(path).exists()), None)
+    if not unshare or not ip or subprocess.run([unshare, '-rmn', 'true'], capture_output=True).returncode:
+        return None
+    hosts = work / 'moved-hosts'
+    hosts.write_text('127.0.0.1 localhost\n127.0.0.2 receiver.test\n')
+    script = f"""
+import json, subprocess, sys
+from pathlib import Path
+sys.path.insert(0, {str(HERE)!r})
+import check
+subprocess.run([{ip!r}, 'link', 'set', 'lo', 'up'], check=True)
+subprocess.run(['mount', '--bind', {str(hosts)!r}, '/etc/hosts'], check=True)
+print(json.dumps(check.server_links_moved_inside(Path({str(aup)!r}), Path({str(pure)!r}), Path({str(work)!r}),
+                                                 Path({str(hosts)!r}), {ip!r})))
+"""
+    done = subprocess.run([unshare, '-rmn', sys.executable, '-c', script], capture_output=True, text=True, timeout=600)
+    assert done.returncode == 0, done.stderr[-3000:]
+    return json.loads(done.stdout.strip().splitlines()[-1])
+
+
+def server_links_moved_inside(aup, pure, work, hosts, ip):
+    import select
+    import shutil
+    import signal
+    import socket
+    import threading
+    import time
+    sender_module, receiver_module = work / 'moved-sender.pmod', work / 'moved-receiver.pmod'
+    for module, text in ((sender_module, LINK_SENDER), (receiver_module, LINK_RECEIVER)):
+        source = module.with_suffix('.aup')
+        source.write_text(text)
+        compile_module(aup, source, module)
+
+    def free_port(kind=socket.SOCK_STREAM):
+        with socket.socket(socket.AF_INET, kind) as probe:
+            probe.bind(('127.0.0.1', 0))
+            return probe.getsockname()[1]
+
+    def start(args, port):
+        process = subprocess.Popen([str(a) for a in args] + ['--port', str(port)], stdout=subprocess.PIPE,
+                                   stderr=subprocess.PIPE)
+        ready, _, _ = select.select([process.stdout], [], [], 10)
+        line = process.stdout.readline() if ready else b''
+        assert line, f'host did not start: {process.stderr.read(2000) if process.poll() is not None else ""}'
+        return process, json.loads(line)
+
+    def post(port, body):
+        import http.client
+        connection = http.client.HTTPConnection('127.0.0.1', port, timeout=10)
+        try:
+            data = json.dumps(body).encode()
+            connection.request('POST', '/inputs/buy', data, {'Content-Length': str(len(data)),
+                                                             'Content-Type': 'application/json'})
+            response = connection.getresponse()
+            response.read()
+            return response.status
+        finally:
+            connection.close()
+
+    def number(port, view):
+        return int(http(port, 'GET', f'/views/{view}')[3])
+
+    keys = {}
+    for name in ('sender', 'receiver'):
+        process, info = start([pure / 'build/pure-serve', '--module', receiver_module, '--store', work / 'moved-key',
+                               '--key', work / f'moved-{name}.key'], 0)
+        keys[name] = info['public_key']
+        process.terminate()
+        process.communicate(timeout=30)
+    port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM)
+    (work / 'moved-senders').write_text(f'moved-sender.ordered {keys["sender"]}\n')
+
+    def receiver(where):
+        return [pure / 'build/pure-serve', '--module', receiver_module, '--store', work / f'moved-{where}',
+                '--key', work / f'moved-{where}.key', '--link-port', link_port, '--senders', work / 'moved-senders']
+
+    shutil.copy(work / 'moved-receiver.key', work / 'moved-first.key')
+    b, _ = start(receiver('first'), port_b)
+    a, _ = start([pure / 'build/pure-serve', '--module', sender_module, '--store', work / 'moved-a',
+                  '--key', work / 'moved-sender.key', '--send', f'ordered=pure://receiver.test:{link_port}/order',
+                  '--seal', f'ordered={keys["receiver"]}'], port_a)
+    acknowledged = [0]
+    stop = threading.Event()
+
+    def client():
+        rng = random.Random(7331)
+        while not stop.is_set():
+            count = rng.randint(1, 20)
+            try:
+                if post(port_a, {'item': 'm', 'count': count}) == 200:
+                    acknowledged[0] += count
+            except OSError:
+                time.sleep(0.02)
+
+    thread = threading.Thread(target=client)
+    thread.start()
+    try:
+        time.sleep(0.6)
+        b.send_signal(signal.SIGKILL)
+        b.wait()
+        # The store, key and memo carried to another machine; the old one gone, the name moved.
+        shutil.copytree(work / 'moved-first', work / 'moved-second')
+        for suffix in ('.key', '.key.links'):
+            shutil.copy(work / f'moved-first{suffix}', work / f'moved-second{suffix}')
+        subprocess.run([ip, 'route', 'add', 'unreachable', '127.0.0.2/32', 'table', 'local'], check=True)
+        hosts.write_text('127.0.0.1 localhost\n127.0.0.3 receiver.test\n')
+        b, _ = start(receiver('second'), port_b)
+        time.sleep(1.5)
+    finally:
+        stop.set()
+        thread.join()
+    try:
+        deadline = time.time() + 60
+        while time.time() < deadline:
+            sent, sent_total = number(port_a, 'sent'), number(port_a, 'sent_total')
+            if (number(port_b, 'count'), number(port_b, 'total')) == (sent, sent_total):
+                break
+            time.sleep(0.1)
+        count, total = number(port_b, 'count'), number(port_b, 'total')
+        assert (count, total) == (sent, sent_total), (count, total, sent, sent_total)
+        assert sent_total >= acknowledged[0] > 0, (sent_total, acknowledged)
+    finally:
+        stats = {}
+        for name, process in (('a', a), ('b', b)):
+            process.terminate()
+            _, error = process.communicate(timeout=30)
+            assert process.returncode == 0, error[-1000:]
+            stats[name] = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1]
+    assert stats['b']['listener_woken'] >= 1 and stats['b']['listener_secrets'] == 0, stats['b']
+    return {'orders': sent, 'acknowledged_total': acknowledged[0], 'packets': stats['a']['link_packets'],
+            'resent': stats['a']['link_resent'], 'received_after_move': stats['b']['listener_received'],
+            'secrets_after_move': stats['b']['listener_secrets']}
+
+
 def server_sealed(aup, pure, work):
     """Sends sealed end to end (--seal to the receiver's --key), in one session
     a wire and authenticated by the sender's own --key: through a relay that
@@ -2045,6 +2192,7 @@ def main():
         report['server_links'] = server_links(aup, pure, work)
         report['server_links_udp'] = server_links_udp(aup, pure, work)
         report['server_links_udp_lossy'] = server_links_udp_lossy(aup, pure, work)
+        report['server_links_moved'] = server_links_moved(aup, pure, work)
         report['server_webhook'] = server_webhook(aup, pure, work)
         report['server_sealed'] = server_sealed(aup, pure, work)
         report['server_surface'] = server_surface(aup, pure, work)