tricu

io_indexed.py (5744B)

---

 #!/usr/bin/env python3
 """Independent framed/authenticated indexed-read and bounded-transfer controls."""
 import argparse
 import hashlib
 import json
 from pathlib import Path
 import struct
 from source_order import Runner
 from process_rss import run_guarded
 from native_indexed_store import indexed_bytes, digest
 from executable_object import INDEXED, indexed_manifest
 from http_headers import number, result, tree_list
 
 ROOT = Path(__file__).resolve().parent.parent
 
 def text(s):
     return tree_list([number(ord(c)) for c in s])
 
 def problem(category, detail, progress):
     return result(False, tree_list([text(category), text('indexed-tree/read'), text(detail), text(progress)]))
 
 def padded(size):
     # Valid noncritical opaque section grows TRANSPORT, not the decoded DAG.
     # This is not a large-DAG memory or compiler-capacity benchmark.
     manifest = indexed_manifest(0)
     nodes = struct.pack('>QIB', 1, 1, 0)
     def section(kind, flags, offset, count):
         return struct.pack('>IHHIQQI', kind, 1, flags, 0, offset, count, 0)
     offset = 128 + len(manifest) + len(nodes)
     return (b'ARBORICX' + struct.pack('>HHIQQ', 1, 0, 3, 0, 32)
             + section(1, 1, 128, len(manifest))
             + section(2, 1, 128 + len(manifest), len(nodes))
             + section(99, 0, offset, size) + manifest + nodes + b'x' * size)
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('tricu', type=Path)
     parser.add_argument('--directory', required=True, type=Path)
     args = parser.parse_args()
     directory = args.directory.resolve()
     directory.mkdir(parents=True, exist_ok=False)
     scratch = directory/'spool-λ'
     scratch.mkdir()
     runner = Runner(args.tricu, directory)
     source = ROOT/'demos/io/indexed-read.tri'
     bundle = directory/'indexed-read.arboricx'
     runner.run('compile', ['compile', '--file', source, '--store', runner.store, '--output', bundle])
     pin = runner.imported(bundle, 'import')
     checks = []
     def check(label, payload, size, limit, expected_digest, expected_reply, position):
         path = directory/(label+'.bin')
         path.write_bytes(payload)
         values = [text(str(path)), text(str(scratch)), number(limit), number(size), text(expected_digest)]
         argv = ['--io']
         for i, value in enumerate(values):
             argument = directory/f'{label}-{i}.tree'
             argument.write_bytes(value)
             argv += ['--arg-tree-file', argument]
         expected = tree_list([expected_reply, number(position)])
         assert runner.invoke(label, pin, 'probe', argv) == expected, label
         assert path.read_bytes() == payload, label
         assert list(scratch.iterdir()) == [], label
         checks.append(dict(label=label, input_bytes=len(payload), frame_bytes=size))
     for i, tree in enumerate([b'\0', b'\1\0', b'\2\1\0\0', number(42)]):
         payload = indexed_bytes(tree)
         check(f'tree-{i}', payload+b'next frame', len(payload), len(payload), digest(INDEXED, payload), result(True, tree), len(payload))
     payload = indexed_bytes(number(7))
     check('wrong-digest', payload, len(payload), len(payload), '0'*64,
           problem('integrity-failed', 'CAS digest mismatch', 'unknown'), len(payload))
     check('truncated', payload[:-1], len(payload), len(payload), digest(INDEXED, payload),
           problem('invalid-data', 'incomplete indexed frame', 'unknown'), len(payload)-1)
     check('limit', payload, len(payload), len(payload)-1, digest(INDEXED, payload),
           problem('invalid-bound', 'invalid snapshot size limit', 'none'), 0)
     check('bad-digest', payload, len(payload), len(payload), 'A'*64,
           problem('invalid-argument', '', 'none'), 0)
     check('bad-bound-wire', payload, 2**63, 2**63, digest(INDEXED, payload),
           problem('invalid-argument', '', 'none'), 0)
     malformed = b'not a bundle'
     check('malformed-authenticated', malformed, len(malformed), len(malformed), digest(INDEXED, malformed),
           problem('invalid-data', 'section extends beyond bundle end', 'unknown'), len(malformed))
     check('malformed-unauthenticated', malformed, len(malformed), len(malformed), '0'*64,
           problem('integrity-failed', 'CAS digest mismatch', 'unknown'), len(malformed))
     for size in [4096, 1048576, 16777216, 67108864]:
         payload = padded(size)
         check(f'transport-{size}', payload, len(payload), len(payload), digest(INDEXED, payload), result(True, b'\0'), len(payload))
     observations = [s for s in runner.records if s['label'].startswith('transport-')]
     for sample in observations:
         assert sample['peak_kib'] <= 48*1024, sample
         assert sample['rts_process']['maximum_live_bytes'] <= 16*1024*1024, sample
     files = ['lib/io.tri', 'demos/io/indexed-read.tri', 'test/io_indexed.py',
              'test/source_order.py', 'test/rts_metrics.py', 'test/native_indexed_store.py',
              'test/executable_object.py', 'test/stable_bundle.py', 'src/ContentStore/Snapshot.hs',
              'src/ContentStore/Indexed.hs', 'src/Wire.hs', 'src/IO/Codec.hs',
              'src/IO/Resources.hs', 'src/IO/Driver.hs', 'src/IO/Protocol.hs',
              'bootstrap/build-stable.arboricx']
     (directory/'report.json').write_text(json.dumps(dict(checks=checks, observations=observations,
 
         measurement_notes='Opaque extension grows transport, not DAG. RSS is sampled: tiny runs can exit before representative samples; use the 64-MiB control for the transfer RSS guard. No concurrency or large-DAG capacity claim.',
         source_hashes={p:hashlib.sha256((ROOT/p).read_bytes()).hexdigest() for p in files}), indent=2)+'\n')
     print(f'{len(checks)} indexed-read controls passed')
 
 if __name__ == '__main__':
     main()