io_http.py (15435B)
1 #!/usr/bin/env python3 2 """Independent real-socket v2 registry framing, streaming, admission and cleanup.""" 3 import argparse 4 import concurrent.futures 5 from contextlib import contextmanager 6 import hashlib 7 import json 8 from pathlib import Path 9 import select 10 import socket 11 import struct 12 import subprocess 13 import threading 14 import time 15 from source_order import Runner 16 from http_headers import number 17 from io_objects import text 18 from process_rss import ChildRSS 19 from rts_metrics import parse_rts_stats 20 21 ROOT = Path(__file__).resolve().parent.parent 22 PHRASE = {200:'OK',201:'Created',400:'Bad Request',404:'Not Found',413:'Content Too Large',431:'Request Header Fields Too Large',500:'Internal Server Error',505:'HTTP Version Not Supported'} 23 def expected(status, body, mime='text/plain; charset=utf-8', head=False): 24 return (f'HTTP/1.1 {status} {PHRASE[status]}\r\nContent-Type: {mime}\r\nContent-Length: {len(body)}\r\nConnection: close\r\n\r\n').encode()+(b'' if head else body) 25 def header(length): 26 return f'POST /_arboricx/bundle HTTP/1.1\r\nHost: local\r\nContent-Length: {length}\r\n\r\n'.encode() 27 def receive(peer): 28 chunks=[] 29 while True: 30 chunk=peer.recv(65536) 31 if not chunk: return b''.join(chunks) 32 chunks.append(chunk) 33 def wait(predicate): 34 until=time.monotonic()+15 35 while time.monotonic()<until: 36 if predicate(): return 37 time.sleep(.02) 38 raise AssertionError('condition timeout') 39 40 def main(): 41 p=argparse.ArgumentParser(description=__doc__) 42 p.add_argument('tricu',type=Path); p.add_argument('--directory',type=Path,required=True) 43 a=p.parse_args(); directory=a.directory.resolve(); directory.mkdir(parents=True,exist_ok=False) 44 runner=Runner(a.tricu,directory) 45 bundle=directory/'registry.arboricx' 46 runner.run('compile',['compile','--file',ROOT/'demos/io/registry.tri','--store',runner.store,'--output',bundle,'--compile-timeout','600'],seconds=660) 47 pin=runner.imported(bundle,'import'); checks=[]; observations=[] 48 @contextmanager 49 def server(label,count,window=4,name='server',root=None): 50 root=root or directory/(label+'-files') 51 argv=[str(runner.exe),'arboricx','run','--module',pin,'--name',name,'--store',str(runner.store),'--io','--no-result'] 52 for i,value in enumerate([text(str(root)),number(count),number(window)]): 53 path=directory/f'{label}-{i}.tree';path.write_bytes(value);argv+=['--arg-tree-file',str(path)] 54 argv+=['+RTS','-N2','-M256m','-T','-s','-RTS'] 55 log=(directory/(label+'.stderr')).open('wb') 56 proc=subprocess.Popen(argv,stdin=subprocess.PIPE if name=='cancelledServer' else subprocess.DEVNULL,stdout=subprocess.PIPE,stderr=log) 57 stopped=threading.Event(); row=dict(label=label,command=argv,peak_kib=0) 58 def monitor(): 59 try: 60 rss=ChildRSS(proc) 61 while not stopped.is_set(): 62 value=rss.sample() 63 if value is not None: 64 row['peak_kib']=max(row['peak_kib'],value) 65 if value>256*1024: raise AssertionError('RSS guard exceeded') 66 stopped.wait(.02) 67 except Exception as error: 68 row['monitor_error']=repr(error);proc.kill() 69 thread=threading.Thread(target=monitor,daemon=True);thread.start();started=time.monotonic() 70 try: 71 assert select.select([proc.stdout],[],[],45)[0], 'startup timeout' 72 line=proc.stdout.readline() 73 assert line.strip().isdigit(), (line,(directory/(label+'.stderr')).read_text()) 74 port=int(line) 75 def connect(): 76 return socket.create_connection(('127.0.0.1',port),timeout=90) 77 def request(wire,fragment=False): 78 with connect() as peer: 79 if fragment: 80 for start in range(0,len(wire),997):peer.sendall(wire[start:start+997]) 81 else:peer.sendall(wire) 82 peer.shutdown(socket.SHUT_WR) 83 return receive(peer) 84 def cancel(): 85 proc.stdin.write(b'x');proc.stdin.flush() 86 request.cancel=cancel 87 yield root,connect,request 88 stdout,_=proc.communicate(timeout=120) 89 assert proc.returncode==0, (directory/(label+'.stderr')).read_text() 90 assert b'private logger detail' not in stdout 91 assert not list(root.rglob('.tricu-stage-*')) 92 row['rts_process']=parse_rts_stats((directory/(label+'.stderr')).read_text()) 93 assert row['rts_process']['maximum_live_bytes']<64*1024*1024,row 94 if label in ['binary','lifecycle','admission']: 95 assert row['peak_kib']<128*1024,row 96 assert row['rts_process']['maximum_live_bytes']<16*1024*1024,row 97 assert 'monitor_error' not in row,row 98 row['seconds']=time.monotonic()-started;observations.append(row) 99 finally: 100 if proc.poll() is None:proc.kill();proc.communicate() 101 stopped.set();thread.join();log.close() 102 def object_path(root,data): 103 digest=hashlib.sha256(data).hexdigest() 104 return digest,root/'objects'/digest[:3]/(digest+'.arboricx') 105 def check(label,actual,wanted): 106 assert actual==wanted,(label,actual[:300],wanted[:300]);checks.append(label) 107 108 cases=[ 109 ('health',b'GET /_arboricx/health HTTP/1.1\r\nHost: local\r\n\r\n',200,b'OK\n',False), 110 ('head',b'HEAD /_arboricx/health HTTP/1.0\r\n\r\n',200,b'OK\n',True), 111 ('duplicate-length',header(0).replace(b'\r\n\r\n',b'\r\nContent-Length: 0\r\n\r\n'),400,b'Bad Request\n',False), 112 ('transfer-encoding',b'POST /_arboricx/bundle HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n',400,b'Bad Request\n',False), 113 ('oversize',header(1048577),413,b'Request body too large\n',False), 114 ('length-junk',header(1048577).replace(b'1048577',b'1048577x'),400,b'Bad Request\n',False), 115 ('bad-version',header(0).replace(b'HTTP/1.1',b'HTTP/9.0'),505,b'HTTP Version Not Supported\n',False), 116 ('syntax-before-version',b'POST /_arboricx/bundle HTTP/9.0\r\nbroken\r\n\r\n',400,b'Bad Request\n',False), 117 ('head-syntax-body',b'HEAD /_arboricx/health HTTP/1.1\r\nbroken\r\n\r\n',400,b'Bad Request\n',False), 118 ('head-length-error',b'HEAD /_arboricx/health HTTP/1.1\r\nContent-Length: x\r\n\r\n',400,b'Bad Request\n',True), 119 ('missing-length',b'POST /_arboricx/bundle HTTP/1.1\r\n\r\nignored',201,hashlib.sha256(b'').hexdigest().encode(),False), 120 ('short-get-body',b'GET /_arboricx/health HTTP/1.1\r\nContent-Length: 5\r\n\r\nabc',400,b'Unexpected end of request body\n',False), 121 ] 122 cases += [('header-limit', b'GET /_arboricx/health HTTP/1.1\r\nX-Long: '+b'x'*65536+b'\r\n\r\n',431,b'Request Header Fields Too Large\n',False)] 123 with server('framing',len(cases)) as (root,connect,request): 124 for label,wire,status,body,head in cases: 125 check(label,request(wire),expected(status,body,head=head)) 126 with server('binary',14) as (root,connect,request): 127 for size in [0,1,4095,4096,4097,65536,1048576]: 128 data=(bytes(range(256))*((size+255)//256))[:size] 129 digest,path=object_path(root,data) 130 for state in ['new','existing']: 131 check(f'binary-{size}-{state}',request(header(size)+data,True),expected(201,digest.encode())) 132 assert path.read_bytes()==data 133 assert list((root/'tmp').iterdir())==[] 134 with server('failures',5) as (root,connect,request): 135 data=b'unpublished-prefix'*200;digest,path=object_path(root,data) 136 with connect() as peer: 137 peer.sendall(header(len(data)+1)+data) 138 wait(lambda:len(list((root/'tmp').iterdir()))==1) 139 assert not path.exists() 140 peer.shutdown(socket.SHUT_WR) 141 check('early-eof',receive(peer),expected(400,b'Upload failed\n')) 142 assert not path.exists();assert list((root/'tmp').iterdir())==[] 143 path.parent.mkdir(exist_ok=True);path.write_bytes(b'X'+data[1:]) 144 check('corrupt-winner',request(header(len(data))+data),expected(400,b'Upload failed\n')) 145 assert path.read_bytes()==b'X'+data[1:] 146 check('corrupt-download',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(404,b'Artifact unavailable\n')) 147 # Peer reset is a connection-local failure, not a server shutdown. 148 with connect() as peer: 149 peer.sendall(header(1048576)+b'partial') 150 peer.setsockopt(socket.SOL_SOCKET,socket.SO_LINGER,struct.pack('ii',1,0)) 151 check('healthy-after-reset',request(cases[0][1]),expected(200,b'OK\n')) 152 with server('downloads',7) as (root,connect,request): 153 data=b'authenticated binary\0\xff';digest,path=object_path(root,data) 154 path.parent.mkdir(exist_ok=True);path.write_bytes(data) 155 wire=f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode() 156 check('download-binary',request(wire),expected(200,data,'application/vnd.arboricx.bundle')) 157 check('download-head',request(wire.replace(b'GET ',b'HEAD ')),expected(200,data,'application/vnd.arboricx.bundle',head=True)) 158 package=root/'packages'/'example';package.mkdir(parents=True) 159 (package/'v1').write_bytes(b'package bytes') 160 check('package-download',request(b'GET /_arboricx/package/example/v1 HTTP/1.1\r\n\r\n'),expected(200,b'package bytes','application/vnd.arboricx.package-release')) 161 for size in [4096,65536,1048576]: 162 data=bytes(range(256))*(size//256);digest,path=object_path(root,data) 163 path.parent.mkdir(exist_ok=True);path.write_bytes(data) 164 check(f'download-{size}',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(200,data,'application/vnd.arboricx.bundle')) 165 # Trailing pipelined bytes never become a second request or uploaded data. 166 data=b'one request';digest,path=object_path(root,data) 167 check('surplus-body',request(header(len(data))+data+cases[0][1]),expected(201,digest.encode())) 168 assert path.read_bytes()==data 169 # Repeated handoff/join windows also exercise completed-ticket retention. 170 with server('lifecycle',128,window=4) as (root,connect,request): 171 for i in range(128): 172 assert request(cases[0][1])==expected(200,b'OK\n') 173 checks.append('128-connections-consumed-tickets') 174 with server('limited',4,name='limited') as (root,connect,request): 175 check('configured-upload-limit',request(header(5)+b'12345'),expected(413,b'Request body too large\n')) 176 assert not list((root/'objects').rglob('*.arboricx')) 177 check('configured-upload-boundary',request(header(4)+b'1234'),expected(201,hashlib.sha256(b'1234').hexdigest().encode())) 178 for data,status in [(b'1234',200),(b'12345',404)]: 179 digest,path=object_path(root,data) 180 path.parent.mkdir(exist_ok=True);path.write_bytes(data) 181 body=data if status==200 else b'Artifact unavailable\n' 182 mime='application/vnd.arboricx.bundle' if status==200 else 'text/plain; charset=utf-8' 183 check(f'configured-download-{len(data)}',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(status,body,mime)) 184 with server('logger',2,name='badLogger') as (root,connect,request): 185 data=b'published before logger failure';digest,path=object_path(root,data) 186 check('logger-failure',request(header(len(data))+data),expected(500,b'Internal Server Error\n')) 187 assert path.read_bytes()==data 188 check('download-logger-failure',request(f'GET /_arboricx/bundle/hash/{digest} HTTP/1.1\r\n\r\n'.encode()),expected(500,b'Internal Server Error\n')) 189 # Two pending streams occupy exactly two admission slots. A third connection 190 # may enter the kernel backlog but cannot create another stage until a ticket 191 # is consumed. FIFO deliberately waits for the oldest, not any completed child. 192 with server('admission',3,window=2) as (root,connect,request): 193 with connect() as first,connect() as second,connect() as third: 194 first.sendall(header(1));second.sendall(header(1));third.sendall(header(1)) 195 wait(lambda:len(list((root/'tmp').iterdir()))==2) 196 second.sendall(b'b');second.shutdown(socket.SHUT_WR) 197 check('second-concurrent',receive(second),expected(201,hashlib.sha256(b'b').hexdigest().encode())) 198 wait(lambda:len(list((root/'tmp').iterdir()))==1) 199 third.settimeout(.3) 200 try: third.recv(1);raise AssertionError('third admitted before oldest completion') 201 except socket.timeout: pass 202 assert len(list((root/'tmp').iterdir()))==1 203 first.sendall(b'a');first.shutdown(socket.SHUT_WR) 204 check('oldest-completes',receive(first),expected(201,hashlib.sha256(b'a').hexdigest().encode())) 205 third.settimeout(90);third.sendall(b'c');third.shutdown(socket.SHUT_WR) 206 check('queued-admitted',receive(third),expected(201,hashlib.sha256(b'c').hexdigest().encode())) 207 with server('cancelled',0,name='cancelledServer') as (root,connect,request): 208 with connect() as first,connect() as second: 209 first.sendall(header(1048576)+b'a'*4096) 210 second.sendall(header(1048576)+b'b'*4096) 211 wait(lambda:len(list((root/'tmp').iterdir()))==2) 212 request.cancel() 213 for peer in [first,second]: 214 try: assert receive(peer)==b'', 'cancellation rendered a response' 215 except ConnectionResetError: pass 216 checks.append('server-cancellation-cleans-active-streams') 217 def startup(label,name,values,flags=()): 218 argv=[str(runner.exe),'arboricx','run','--module',pin,'--name',name,'--store',str(runner.store),'--io','--no-result',*flags] 219 for i,value in enumerate(values): 220 path=directory/f'{label}-{i}.tree';path.write_bytes(value);argv+=['--arg-tree-file',str(path)] 221 result=subprocess.run(argv,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=30) 222 (directory/(label+'.startup.log')).write_bytes(result.stdout+result.stderr) 223 assert result.returncode!=0 and not result.stdout,(label,result.stdout,result.stderr) 224 checks.append(label) 225 for window in [0,257]: 226 root=directory/f'invalid-window-{window}' 227 startup(f'window-{window}','server',[text(str(root)),number(1),number(window)]) 228 assert not root.exists() 229 root=directory/'blocked-root';root.write_bytes(b'unchanged') 230 startup('blocked-root','server',[text(str(root)),number(1),number(1)]) 231 assert root.read_bytes()==b'unchanged' 232 root=directory/'grant-root' 233 startup('reject-v1-grants','server',[text(str(root)),number(1),number(1)],['--unsafe-io']) 234 assert not root.exists() 235 with socket.socket() as occupied: 236 occupied.bind(('127.0.0.1',0));occupied.listen(1) 237 from io_tcp import endpoint 238 root=directory/'bind-failure' 239 startup('bind-failure','listen',[text(str(root)),endpoint(4,occupied.getsockname()[1])]) 240 assert not list(root.rglob('.tricu-stage-*')) 241 (directory/'report.json').write_text(json.dumps(dict(passed=checks,records=runner.records,processes=observations),indent=2)+'\n') 242 print(f'PASS {len(checks)} v2 HTTP controls; deployment unchanged') 243 if __name__=='__main__':main()