1"""Interrupt an owned mixed append workload and require progress after reconnection."""
2import json
3import shlex
4import subprocess
5import time
6
7from native_runner import windows
8import linux_vm
9from verify_smb_overlap import verify
10
11
12def interrupt(output, clients, sequences, processes):
13 config = json.loads((output / 'run.json').read_text())
14 server = config['server']
15 samples = []
16 subprocess.run(linux_vm.ssh_argv(server, 'cat > /tmp/verify_smb_overlap.py'),
17 input=(output / 'harness/verify_smb_overlap.py').read_bytes(), check=True)
18
19 def counts(transport=False):
20 result = {}
21 for actor in processes:
22 data = (output / 'rust' / (actor + '.jsonl')).read_text()
23 events = [json.loads(line) for line in data[:data.rfind('\n') + 1].splitlines()]
24 names = ('transport_read_error', 'transport_commit_error') if transport else ('commit' if actor.startswith('w') else 'read',)
25 result[actor] = sum(event['event'] in names for event in events)
26 return result
27
28 def wait_for(predicate, message):
29 deadline = time.monotonic() + 90
30 while not predicate():
31 assert all(process.poll() is None for process in processes.values()), 'A Rust client exited during the interruption campaign'
32 if time.monotonic() > deadline: raise TimeoutError(message)
33 time.sleep(.1)
34
35 def ssh(command):
36 result = linux_vm.run_ssh(server, command, timeout=15)
37 with (output / 'disconnect-server.jsonl').open('a') as stream:
38 stream.write(json.dumps({'command': command, 'exit': result.returncode, 'stdout': result.stdout, 'stderr': result.stderr}) + '\n')
39 result.check_returncode()
40 return result.stdout
41
42 def phase(control):
43 text = json.dumps(control)
44 ssh("printf '%s' '" + text + "' > /tmp/smb-control.tmp && mv /tmp/smb-control.tmp /tmp/smb-control.json")
45 wait_for(lambda: text in ssh("grep -F '\"control\":' /tmp/smb-trace.jsonl | tail -n 1"), 'Proxy did not acknowledge the disconnect phase')
46
47 previous = {actor: 0 for actor in processes}
48 for cycle in range(2):
49 wait_for(lambda: all(value >= previous[actor] + 3 for actor, value in counts().items()), 'Clients made no progress before the interruption')
50 before = counts()
51 native = []
52 for actor, (client, sequence) in enumerate(zip(clients, sequences)):
53 capture = output / f'disconnect-{cycle}-n{actor}.jsonl'
54 result = windows.do_get(f'C:\\one-tests\\runs\\capture\\outbox\\{sequence}\\events.jsonl', capture, client['name'])
55 assert not result.get('error'), result
56 data = capture.read_text(encoding='utf-8-sig')
57 rows = [json.loads(line) for line in data[:data.rfind('\n') + 1].splitlines()]
58 assert 0 < len(rows) < config['stress_operations'], 'A native writer was inactive before the interruption'
59 native.append(len(rows))
60 before_errors = counts(transport=True)
61 assert all(process.poll() is None for process in processes.values()), 'A Rust client finished before the interruption'
62 try:
63 phase({'phase': f'disconnect-{cycle}', 'cut': 9, 'peer': '10.0.2.2', 'offset': 96,
64 'direction': 'request' if cycle == 0 else 'response'})
65 wait_for(lambda: f'"phase": "disconnect-{cycle}"' in ssh("grep -F '\"control\":' /tmp/smb-trace.jsonl | tail -n 1")
66 and int(ssh("grep -c '\"cut\": {' /tmp/smb-trace.jsonl || true").strip()) == cycle + 1,
67 'The planned write interruption did not occur')
68 time.sleep(3)
69 finally:
70 phase({'phase': f'reconnected-{cycle}'})
71 wait_for(lambda: all(value >= before[actor] + 3 for actor, value in counts().items()), 'A client failed to progress after reconnecting')
72 script = '\n'.join([
73 'import json', 'from verify_smb_overlap import verify, PendingOverlap',
74 "data = open('/tmp/smb-trace.jsonl').read()",
75 "events = [json.loads(line) for line in data[:data.rfind('\\n') + 1].splitlines()]",
76 'try:', f" result = verify(events, phase='reconnected-{cycle}')",
77 "except PendingOverlap: result = None", 'print(json.dumps(result))'])
78 wait_for(lambda: json.loads(ssh('cd /tmp && python3 -c ' + shlex.quote(script))) is not None,
79 'Native and Rust guarded I/O did not overlap after reconnection')
80 previous = counts()
81 samples.append({'cycle': cycle, 'before': before, 'after': previous, 'native_before': native,
82 'before_errors': before_errors, 'after_errors': counts(transport=True)})
83 (output / 'disconnect-progress.json').write_text(json.dumps(samples, indent=2))
84
85
86def verify_disconnect(output):
87 config = json.loads((output / 'run.json').read_text())
88 assert config['stress_clients'] + config['rust_writers'] + config['rust_readers'] >= 12
89 samples = json.loads((output / 'disconnect-progress.json').read_text())
90 assert [sample['cycle'] for sample in samples] == [0, 1]
91 actors = {f'w{i}' for i in range(config['rust_writers'])} | {f'r{i}' for i in range(config['rust_readers'])}
92 events = [json.loads(line) for line in (output / 'smb-trace.jsonl').read_text().splitlines()]
93 assert sum('cut' in event for event in events) == 2, 'Unexpected number of connection interruptions'
94 errors = {}
95 progress = {}
96 progress_limit = 120
97 started = (output / 'rust/start').stat().st_mtime_ns // 1000
98 stopped = (output / 'rust/stop').stat().st_mtime_ns // 1000
99
100 def max_gap(actor, times, units):
101 assert len(times) > 1, 'A client made no progress'
102 gaps = [(end - begin) / units for begin, end in zip(times, times[1:])]
103 assert all(0 <= gap <= progress_limit for gap in gaps), f'{actor}: client progress stalled or went backwards'
104 return max(gaps)
105
106 for i in range(config['stress_clients']):
107 rows = [json.loads(line) for line in (output / f'n{i}/stress-events.jsonl').read_text(encoding='utf-8-sig').splitlines()]
108 assert len(rows) == config['stress_operations'], 'A native writer did not finish'
109 progress[f'n{i}'] = max_gap(f'n{i}', [rows[0]['update_started_ticks'], *[row['updated_ticks'] for row in rows]], 10**7)
110 uncertain = set()
111 for actor in sorted(actors):
112 rows = [json.loads(line) for line in (output / 'rust' / (actor + '.jsonl')).read_text().splitlines()]
113 errors[actor] = sum(row['event'] in ('transport_read_error', 'transport_commit_error') for row in rows)
114 assert errors[actor] and any(row['event'] == 'transport_connected' for row in rows), 'A Rust client did not exercise reconnection'
115 times = [started, *[row['finished_us'] for row in rows if row['event'] == ('commit' if actor.startswith('w') else 'read')]]
116 if actor.startswith('r') and len(times) > 1: times.append(max(times[-1], stopped))
117 progress[actor] = max_gap(actor, times, 10**6)
118 pending = None
119 for row in rows:
120 if row['event'] == 'transport_commit_error': pending = row
121 if row['event'] != 'transport_reconciled': continue
122 assert pending is not None and pending['token'] == row['token']
123 if row['published']: assert row.get('flush_confirmed'), 'Visible recovery lacks a durable acknowledgement'
124 if pending['state'] == 'Unknown': uncertain.add('after' if row['published'] else 'before')
125 pending = None
126 assert uncertain == {'before', 'after'}, 'The mixed workload did not resolve both uncertain outcomes'
127 overlap = []
128 for sample in samples:
129 assert all(set(sample[field]) == actors for field in ('before', 'after', 'before_errors', 'after_errors'))
130 assert all(sample['after'][actor] >= value + 3 for actor, value in sample['before'].items())
131 assert all(sample['after_errors'][actor] > value for actor, value in sample['before_errors'].items()), 'A client did not encounter this interruption'
132 assert len(sample['native_before']) == config['stress_clients']
133 assert all(0 < value < config['stress_operations'] for value in sample['native_before'])
134 end = next((i for i, event in enumerate(events) if event.get('control', {}).get('phase') == f'disconnect-{sample["cycle"] + 1}'), len(events))
135 overlap.append(verify(events[:end], phase=f'reconnected-{sample["cycle"]}'))
136 return {'interruptions': 2, 'transport_errors': errors, 'uncertain_outcomes': sorted(uncertain), 'resumed_overlap': overlap,
137 'max_progress_gap_seconds': progress, 'progress_limit_seconds': progress_limit}