io_v2_indexed_write.py (9130B)
1 #!/usr/bin/env python3 2 """Independent canonical bytes, winner verification and shared-DAG round trips.""" 3 import argparse 4 import hashlib 5 import json 6 from pathlib import Path 7 import struct 8 from source_order import Runner 9 from process_rss import run_guarded 10 from native_indexed_store import indexed_bytes, digest 11 from executable_object import INDEXED, indexed_manifest 12 from stable_bundle import assembled 13 from http_headers import number, result, tree_list 14 15 ROOT = Path(__file__).resolve().parent.parent 16 17 def text(s): 18 return tree_list([number(ord(c)) for c in s]) 19 20 def info(payload): 21 return tree_list([number(len(payload)), text(digest(INDEXED, payload))]) 22 23 def problem(category, operation, detail, progress): 24 return result(False, tree_list([text(category), text(operation), text(detail), text(progress)])) 25 26 def dag(depth, fork): 27 records = [struct.pack('>IB', 1, 0)] 28 for i in range(1, depth+1): 29 payload = struct.pack('>BII', 2, i-1, i-1) if fork else struct.pack('>BI', 1, i-1) 30 records.append(struct.pack('>I', len(payload))+payload) 31 return assembled(indexed_manifest(depth), struct.pack('>Q', len(records))+b''.join(records)) 32 33 def main(): 34 parser = argparse.ArgumentParser(description=__doc__) 35 parser.add_argument('tricu', type=Path) 36 parser.add_argument('--directory', type=Path, required=True) 37 parser.add_argument('--zig-writer-probe', type=Path, help='private writer-only subset: no authenticated read/round-trip acceptance') 38 parser.add_argument('--zig-probe', type=Path, help='private complete indexed transport controls including authenticated reads') 39 args = parser.parse_args() 40 if args.zig_probe and args.zig_writer_probe: parser.error('select only one private probe mode') 41 private_probe = args.zig_probe or args.zig_writer_probe 42 directory = args.directory.resolve() 43 directory.mkdir(parents=True, exist_ok=False) 44 runner = Runner(args.tricu, directory) 45 source = ROOT/'demos/io-v2/indexed-write.tri' 46 bundle = directory/'writer.arboricx' 47 runner.run('compile', ['compile', '--file', source, '--store', runner.store, '--output', bundle]) 48 pin = runner.imported(bundle, 'import') 49 checks = [] 50 zig_observations = [] 51 def invoke(label, name, values, expected): 52 argv = ['--io'] 53 zig_arguments = [] 54 for i, value in enumerate(values): 55 path = directory/f'{label}-{i}.tree' 56 path.write_bytes(value) 57 argv += ['--arg-tree-file', path] 58 zig_arguments.append('tree-file:'+str(path)) 59 if private_probe: 60 command = [str(private_probe.resolve()),str(bundle),name,*zig_arguments] 61 output,peak = run_guarded(command,memory_mib=256,seconds=180) 62 (directory/(label+'-zig.log')).write_text(output) 63 assert bytes(map(int,output.strip())) == expected,label 64 zig_observations.append(dict(label=label,peak_kib=peak,command=command)) 65 else: 66 assert runner.invoke(label, pin, name, argv) == expected, label 67 checks.append(label) 68 for i, tree in enumerate([b'\0', b'\1\0', b'\2\1\0\0', number(42)]): 69 payload = indexed_bytes(tree) 70 expected = digest(INDEXED, payload) 71 invoke(f'hash-{i}', 'hash', [number(len(payload)), tree], result(True, info(payload))) 72 path = directory/f'write-{i}.bin' 73 invoke(f'write-{i}', 'write', [text(str(path)), number(len(payload)), text(expected), tree], result(True, info(payload))) 74 assert path.read_bytes() == payload 75 root = directory/f'publish-{i}' 76 root.mkdir() 77 for status in ['published', 'exists']: 78 invoke(f'publish-{i}-{status}', 'publish', [text(str(root)), number(len(payload)), tree], 79 result(True, tree_list([text(status),number(len(payload)),text(expected)]))) 80 assert sorted(p.name for p in root.iterdir()) == [expected] 81 assert (root/expected).read_bytes() == payload 82 (root/expected).write_bytes(b'X'+payload[1:]) 83 invoke(f'corrupt-{i}', 'publish', [text(str(root)), number(len(payload)), tree], 84 problem('integrity-failed','indexed-tree/verify','indexed frame differs from canonical tree','unknown')) 85 assert (root/expected).read_bytes() == b'X'+payload[1:] 86 assert sorted(p.name for p in root.iterdir()) == [expected] 87 (root/expected).write_bytes(payload+b'extra') 88 invoke(f'excess-{i}', 'publish', [text(str(root)), number(len(payload)), tree], 89 problem('invalid-data','indexed-file/verify','trailing bytes','unknown')) 90 assert (root/expected).read_bytes() == payload+b'extra' 91 path = directory/'preserve.bin' 92 path.write_bytes(b'untouched') 93 tree = b'\0' 94 payload = indexed_bytes(tree) 95 for label, limit, expected, answer in [ 96 ('wrong-hash',65536,'0'*64,problem('integrity-failed','indexed-tree/write','CAS digest mismatch','none')), 97 ('limit',0,digest(INDEXED,payload),problem('invalid-bound','indexed-tree/write','encoded tree exceeds limit','none')), 98 ('invalid-hash',65536,'A'*64,problem('invalid-argument','indexed-tree/write','','none'))]: 99 invoke(label, 'write', [text(str(path)),number(limit),text(expected),tree], answer) 100 assert path.read_bytes() == b'untouched' 101 # Raw input uses linear-depth stems, not an expanded exponentially shared 102 # tree. Expectations come from the independent wire assembler, not Zig. 103 for depth in [1000,10000]: 104 payload = dag(depth,False) 105 tree = b'\1'*depth+b'\0' 106 invoke(f'deep-hash-{depth}','hash',[number(len(payload)),tree],result(True,info(payload))) 107 output = directory/f'deep-{depth}.bin' 108 invoke(f'deep-write-{depth}','write',[text(str(output)),number(len(payload)),text(digest(INDEXED,payload)),tree],result(True,info(payload))) 109 assert output.read_bytes() == payload 110 original_bundle,original_pin = bundle,pin 111 bundle = directory/'transport.arboricx' 112 runner.run('compile-transport',['compile','--file',ROOT/'test/io-v2/indexed-transport.tri','--store',runner.store,'--output',bundle]) 113 pin = runner.imported(bundle,'import-transport') 114 payload = indexed_bytes(b'\0') 115 framed = directory/'framed.bin' 116 framed.write_bytes(payload+b'X') 117 from http_headers import pair 118 invoke('frame-tail','verifyTail',[text(str(framed)),number(len(payload)),b'\0'],result(True,pair(text('data'),tree_list([number(88)])))) 119 framed.write_bytes(payload[:-1]) 120 invoke('frame-truncated','verifyTail',[text(str(framed)),number(len(payload)),b'\0'],problem('integrity-failed','indexed-tree/verify','indexed frame differs from canonical tree','unknown')) 121 invoke('wrong-kind-inert-tree','wrongKind',[number(0)],problem('wrong-kind','indexed-tree/verify','','none')) 122 bundle,pin = original_bundle,original_pin 123 scratch = directory/'spool' 124 scratch.mkdir() 125 for depth, fork in ([] if args.zig_writer_probe else [(200,True),(1000,False),(10000,False),(100000,False)]): 126 payload = dag(depth,fork) 127 input_path = directory/f'dag-{depth}-{fork}.bin' 128 output = directory/f'dag-{depth}-{fork}-output.bin' 129 input_path.write_bytes(payload) 130 invoke(f'roundtrip-{depth}-{fork}', 'roundTrip', [text(str(input_path)),text(str(output)),text(str(scratch)), 131 number(len(payload)),number(len(payload)),text(digest(INDEXED,payload))], result(True,info(payload))) 132 assert output.read_bytes() == input_path.read_bytes() == payload 133 assert list(scratch.iterdir()) == [] 134 observations = [r for r in runner.records if r['label'].startswith('roundtrip-')] 135 for row in observations: 136 # DAG/table memory is expected to grow; these are regression guards, not 137 # constant-memory claims or a universal safe node-count budget. 138 assert row['peak_kib'] <= 256*1024, row 139 assert row['rts_process']['maximum_live_bytes'] <= 128*1024*1024, row 140 paths = ['test/io-v2/indexed-transport.tri','test/io_v2_indexed_write.py','test/native_indexed_store.py','test/executable_object.py', 141 'test/stable_bundle.py','test/source_order.py','test/rts_metrics.py','lib/io-v2.tri', 142 'demos/io-v2/indexed-write.tri','src/IOV2/Codec.hs','src/IOV2/Resources.hs','src/Wire.hs', 143 'bootstrap/build-stable.arboricx'] 144 if private_probe: paths += ['ext/zig/src/indexed.zig','ext/zig/src/io_v2_file_host.zig'] 145 (directory/'report.json').write_text(json.dumps(dict(checks=checks, observations=observations, zig_observations=zig_observations, 146 scope="private-zig-writer-only" if args.zig_writer_probe else "private-zig-complete" if args.zig_probe else "haskell-complete", 147 zig_probe_sha256=hashlib.sha256(private_probe.read_bytes()).hexdigest() if private_probe else None, 148 source_hashes={p:hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in paths}),indent=2)+'\n') 149 print(f'{len(checks)} independent indexed-writer checks passed'+(' (private Zig writer-only; authenticated read NOT tested)' if args.zig_writer_probe else '')) 150 151 if __name__ == '__main__': 152 main()