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.

1 files changed, 148 insertions, 0 deletions

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)