From 9aa123ec5e8827a6101fa7c6db139397b39e04f2 Mon Sep 17 00:00:00 2001 From: eunhhu Date: Wed, 30 Sep 2026 05:33:32 +0100 Subject: Recast waiting sends by field name when an output's fields change Review 004's F3 made a module refuse to start while sends waited on an output whose fields it had changed: their bytes would have meant something else. Safe, but it deadlocks a rolling upgrade. Growing a board shared by two nodes: one upgraded to a card with a new field refuses the other's old sends (another shape), and the other, its old sends waiting, cannot start the new module at all ("Cannot start the module on the store", saying nothing more). Waiting sends now move to the new fields by name, as a list's items do: a value goes over when the new type can hold it (u32 and number are numbers, bytes and text are bytes; a number that is not a u32, or bytes that are not UTF-8 for a text, cannot), the rest start at 0 or empty (pure_send_recast, with the payload layout load uses, pure_fields_lay_out). A start recasts them and a replace does into a new outbox. Their meaning must survive a crash between the recast and the save that names the new fields in the links. So the outbox's image now names the fields its sends were laid out with ("PLO1": the module's outputs after the save it holds), and is written again before any send is logged with other fields; a start reads those names over the links'. Checkpoints of the outbox name them too. A module without an output that sends wait on is still refused, now saying so in PureApp.why, which pure-serve prints ("sends wait on the output 'ordered', which this module does not have"); a failed start keeps it. test-module send_recast (widened, narrowed, UTF-8, a payload not laid out, there and back) and output_meaning (the swapped fields recast on a replace, read right after a restart that never saved, back again, sends logged after the image, the refusal and its reason). check.py server_upgrade: five sends wait on the old output, the sender starts with the new one, the old receiver refuses the new shape, the new one takes all eight. --- tests/check.py | 123 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 123 insertions(+) (limited to 'tests/check.py') diff --git a/tests/check.py b/tests/check.py index ea4f9f6..b377e20 100644 --- a/tests/check.py +++ b/tests/check.py @@ -1317,6 +1317,128 @@ def server_windows(aup, pure, work): return {'windows_full': stats['a']['windows_full'], 'wire_deliveries': stats['a']['delivered'], 'link_records': stats['a']['link_records']} + +UPGRADED_SENDER = """state n = 0 +input buy = message { + item: text + count: u32 +} +output ordered = message { + note: text + item: text + count: u32 +} +on buy { + n = n + 1 + send ordered { note: "second", item: buy.item, count: buy.count } +} +view sent = n +""" + +UPGRADED_RECEIVER = """state orders = 0 +state noted = 0 +state stock = 0 +input order = message { + note: text + item: text + count: u32 +} +on order { + orders = orders + 1 + stock = stock + order.count + if order.note == "second" { noted = noted + 1 } +} +view count = orders +view noted_count = noted +view total = stock +""" + + +def server_upgrade(aup, pure, work): + """A rolling upgrade of an output's fields between two linked nodes. The + sender, its receiver down, holds five sends of ordered {item, count, + to.city}; started again with ordered {note, item, count} it recasts them + by name (note empty, to.city gone) where it was refused before (review + 004's F3), and a module without the output is refused naming it. The old + receiver refuses the new shape, so the sends wait; the upgraded receiver + takes all eight, once each.""" + import select + import socket + import time + modules = {} + for name, text in (('sender-1', LINK_SENDER), ('sender-2', UPGRADED_SENDER), ('receiver-1', LINK_RECEIVER), + ('receiver-2', UPGRADED_RECEIVER), ('sender-0', 'state n = 0\ninput tap = key KeyT\non tap.down { n = 1 }\n')): + source = work / f'upgrade-{name}.aup' + source.write_text(text) + modules[name] = compile_module(aup, source, source.with_suffix('.pmod')) and source.with_suffix('.pmod') + + 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 stop(process): + process.terminate() + _, error = process.communicate(timeout=30) + assert process.returncode == 0, error[-1000:] + return [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + + 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', modules['receiver-1'], '--store', + work / 'upgrade-key', '--key', work / f'upgrade-{name}.key'], 0) + keys[name] = info['public_key'] + stop(process) + port_a, port_b, link_port = free_port(), free_port(), free_port(socket.SOCK_DGRAM) + + def sender_args(module): + return [pure / 'build/pure-serve', '--module', modules[module], '--store', work / 'upgrade-a', '--key', + work / 'upgrade-sender.key', '--name', 'up', '--send', f'ordered=pure://127.0.0.1:{link_port}/order', + '--seal', f'ordered={keys["receiver"]}'] + + def receiver_args(module): + return [pure / 'build/pure-serve', '--module', modules[module], '--store', work / 'upgrade-b', '--key', + work / 'upgrade-receiver.key', '--link-port', link_port, '--sender', f'up.ordered={keys["sender"]}'] + + running, stats, said = {}, {}, '' + try: + running['a'] = start(sender_args('sender-1'), port_a)[0] + for i in range(5): + assert post(port_a, '/inputs/buy', {'item': f'first-{i}', 'count': 1})[0] == 200 + stop(running.pop('a')) + refused = subprocess.run([str(a) for a in sender_args('sender-0')] + ['--port', str(port_a)], + capture_output=True, timeout=30) + assert refused.returncode != 0 and b"sends wait on the output 'ordered'" in refused.stderr, refused.stderr + running['a'] = start(sender_args('sender-2'), port_a)[0] + for i in range(3): + assert post(port_a, '/inputs/buy', {'item': f'second-{i}', 'count': 10})[0] == 200 + running['b'] = start(receiver_args('receiver-1'), port_b)[0] + time.sleep(1.5) + assert number(port_b, 'count') == 0 # the old input takes another shape: the sends wait + stop(running.pop('b')) + running['b'] = start(receiver_args('receiver-2'), port_b)[0] + deadline = time.time() + 20 + while number(port_b, 'count') < 8 and time.time() < deadline: + time.sleep(0.05) + got = (number(port_b, 'count'), number(port_b, 'noted_count'), number(port_b, 'total')) + assert got == (8, 3, 35), got + finally: + for name, process in running.items(): + stats[name] = stop(process) + assert stats['a']['link_refused'] >= 1, stats['a'] + return {'recast_waiting': 5, 'delivered': 8, 'refused_while_old': stats['a']['link_refused']} + def server_links_independent(aup, pure, work): """A receiver written from docs/design/012-link-protocol.md alone, in Python with the cryptography package and no code of link/: a Pure node @@ -3023,6 +3145,7 @@ def main(): report['server_links_refused'] = server_links_refused(aup, pure, work) report['server_memory'] = server_memory(aup, pure, work) report['server_windows'] = server_windows(aup, pure, work) + report['server_upgrade'] = server_upgrade(aup, pure, work) report['server_links_independent'] = server_links_independent(aup, pure, work) report['server_links_moved'] = server_links_moved(aup, pure, work) report['server_webhook'] = server_webhook(aup, pure, work) -- cgit v1.2.3