aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--link/link-protocol.c2
-rw-r--r--link/link-protocol.h3
-rw-r--r--link/link.c78
-rw-r--r--tests/check.py148
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)
generated by cgit v1.2.3 (git 2.46.0) at 2026-10-20 11:26:23 +0000