1#!/usr/bin/env python3
2"""Compare cold native reads of confirmation snapshots with the recorded editing history."""
3import argparse
4import hashlib
5import json
6from pathlib import Path
7import xml.etree.ElementTree as ET
8
9from native_format import native_characters
10from native_stress import native_history
11from native_xml import ns
12from offline_history import publication_links
13
14
15def verify(output, cold, *, partial=False):
16 config = json.loads((output / 'run.json').read_text())
17 actors = [*(f'w{i}' for i in range(config['rust_writers'])), *(f'r{i}' for i in range(config['rust_readers']))]
18 logs = {actor: [json.loads(line) for line in (output / 'rust' / f'{actor}.jsonl').read_text().splitlines()] for actor in actors}
19 publication_links(logs, config['stress_operations'], partial=partial)
20 documents = {}
21 if config.get('document_operations'):
22 from offline_document_history import document_history
23 assert not partial, 'Document confirmation audit requires a complete workload'
24 documents = document_history(logs, config['stress_operations'])
25 unknown = [row for rows in logs.values() for row in rows if row['event'] == 'remote_attempt' and row['state'] == 'Unknown']
26 assert len(unknown) == 1
27 attempt, = unknown
28 confirmations = {}
29 for actor, rows in logs.items():
30 observed = None
31 for row in rows:
32 if row['event'] == 'read': observed = row
33 if row['event'] != 'remote_confirm': continue
34 name = row['capture']
35 assert Path(name).name == name and name.startswith(str(rows[0]['pid']) + '-')
36 assert name not in confirmations
37 if attempt['revision'] not in row['revisions'].get(attempt['space'], []):
38 assert documents and row.get('current_revisions', {}).get(attempt['space']) in row['revisions'].get(attempt['space'], []), 'Retired attempt lacks a current effect-confirmation revision'
39 if documents:
40 assert observed and observed['text'] == row['text'] and observed['finished_us'] <= row['started_us']
41 assert isinstance(observed.get('documents'), dict)
42 assert all(observed['documents'].get(target) == value for target, value in attempt['document_changes'].items()), 'Confirmation omitted the uncertain document change'
43 confirmations[name] = row, observed['documents'] if documents else {}
44 assert confirmations
45 manifest = json.loads((cold / 'run.json').read_text())['inputs']
46 assert set(manifest) == set(confirmations)
47 assert set(manifest) == {path.name for path in (output / 'rust/confirmations').glob('*.one')}
48 results = json.loads((cold / 'results.json').read_text(encoding='utf-8-sig'))
49 if isinstance(results, dict): results = [results]
50 assert len(results) == len(confirmations) and len({row['name'] for row in results}) == len(results)
51 native = []
52 for i in range(config['stress_clients']):
53 rows = [json.loads(line) for line in (output / f'n{i}/stress-events.jsonl').read_text(encoding='utf-8-sig').splitlines()]
54 assert len(rows) <= config['stress_operations']
55 native_history(rows, i, len(rows) if partial else config['stress_operations'], False)
56 native.append({f'Native {i}:', *(row['before'] + row['token'] for row in rows)})
57 checks = format_checks = 0
58 for result in results:
59 name = result['name'] + '.one'
60 assert result['error'] is None and result['pages'] == 1
61 assert result['source_sha256'] == manifest[name] == hashlib.sha256((output / 'rust/confirmations' / name).read_bytes()).hexdigest()
62 page = ET.parse(cold / 'results' / result['name'] / 'page-0.xml').getroot()
63 formatted = native_characters(page, page.findall('one:Outline', ns))
64 paragraphs = [''.join(c for c, _ in paragraph) for paragraph in formatted]
65 confirmation, observed = confirmations[name]
66 expected = confirmation['text']
67 assert len(paragraphs) == len(native) + 1 + len(observed) and paragraphs.count(expected) == 1
68 paragraphs.remove(expected)
69 if observed:
70 from offline_document_history import characters, verify_native
71 format_checks += verify_native(formatted, (characters(value) for value in observed.values()))
72 for document in observed.values():
73 assert paragraphs.count(document['text']) == 1, 'Confirmation duplicated a document paragraph'
74 paragraphs.remove(document['text'])
75 for index, versions in enumerate(native):
76 selected = [text for text in paragraphs if text.startswith(f'Native {index}:')]
77 assert len(selected) == 1 and selected[0] in versions, 'Native confirmation image contains an unrecorded edit or loses a prefix'
78 checks += len(native) + 1 + len(observed)
79 assert json.loads((cold / 'teardown.json').read_text()) == {'absent': True}
80 confirmed_revision = documents[next(iter(attempt['document_changes']))]['states']['format']['attempt']['receipt_revision'] if documents else attempt['revision']
81 return {'complete_workload': not partial, 'attempted_revision': attempt['revision'], 'confirmed_revision': confirmed_revision, 'native_images': len(results), 'exact_rust_paragraphs': len(results),
82 'validated_paragraphs': checks, 'native_intended_format_checks': format_checks,
83 'maximum_native_export_seconds': max(row['seconds'] for row in results)}
84
85
86if __name__ == '__main__':
87 parser = argparse.ArgumentParser(description=__doc__)
88 parser.add_argument('run', type=Path)
89 parser.add_argument('cold', type=Path)
90 parser.add_argument('--partial', action='store_true', help='Validate preserved confirmations from an interrupted workload')
91 args = parser.parse_args()
92 result = verify(args.run, args.cold, partial=args.partial)
93 (args.run / 'confirmation-cold-verification.json').write_text(json.dumps(result, indent=2))
94 print(json.dumps(result, indent=2))