"""Loopback-only HTTP/1.1 connection-affinity illustration. Python stdlib only.""" import csv import http.client import json import select import socket import threading from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path OUT = Path(__file__).resolve().parent / 'run' def make_handler(state): class Handler(BaseHTTPRequestHandler): protocol_version = 'HTTP/1.1' timeout = 3 def do_GET(self): body = json.dumps(state).encode() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Content-Length', str(len(body))) self.end_headers() self.wfile.write(body) self.wfile.flush() def log_message(self, *_): pass return Handler class Proxy: def __init__(self, targets): self.targets = targets self.done = threading.Event() self.workers = [] self.accepts = [] self.errors = [] self.listener = socket.socket() self.listener.bind(('127.0.0.1', 0)) self.listener.listen(32) self.listener.settimeout(.1) self.address = self.listener.getsockname() self.thread = threading.Thread(target=self.accept_loop) self.thread.start() def accept_loop(self): while not self.done.is_set(): try: client, _ = self.listener.accept() except socket.timeout: continue except OSError: break index = len(self.accepts) % 6 self.accepts.append({'connection': len(self.accepts) + 1, 'copy': f'copy-{index + 1}'}) worker = threading.Thread(target=self.relay, args=(client, self.targets[index])) self.workers.append(worker) worker.start() def relay(self, client, target): try: with client, socket.create_connection(target, timeout=3) as upstream: while not self.done.is_set(): ready, _, _ = select.select([client, upstream], [], [], .1) for source in ready: data = source.recv(65536) if not data: return (upstream if source is client else client).sendall(data) except OSError as exc: self.errors.append(repr(exc)) def close(self): self.done.set() self.listener.close() self.thread.join(timeout=4) for worker in self.workers: worker.join(timeout=4) assert not self.thread.is_alive() assert not any(w.is_alive() for w in self.workers) def fetch(connection, phase, mode, sequence, rows): connection.request('GET', '/version') response = connection.getresponse() assert response.status == 200 and response.version == 11 assert not response.will_close payload = json.loads(response.read()) assert payload['health'] == 'healthy' rows.append(dict(phase=phase, mode=mode, sequence=sequence, status=response.status, **payload)) def summarize(rows): return dict(responses=len(rows), old_responses=sum(r['version'] == 'old' for r in rows), fixed_responses=sum(r['version'] == 'fixed' for r in rows), unique_copies=sorted({r['copy'] for r in rows}), all_healthy=True) def chart(results): svg = ['', 'Connection reuse changed which copies the loopback sample saw', 'Before changing the version response, reused connection: zero old responses in 60. Fresh connections: ten in 60. Direct inventory: one old copy in six. After changing the version response: all old counts zero.', '', 'One connection can hide an old copy', 'Controlled local TCP proxy · round-robin per connection · six HTTP/1.1 copies'] for pi, phase in enumerate(('before_change', 'after_change')): y0 = 125 + pi * 210 svg.append(f'{phase.replace("_", " ").capitalize()}') for mi, (mode, label) in enumerate((('reused_connection', 'Reused connection'), ('fresh_connections', 'Fresh connections'), ('direct_inventory', 'Direct inventory'))): r = results[phase][mode] y = y0 + 28 + mi * 46 old, total = r['old_responses'], r['responses'] svg.append(f'{label}') if old: svg.append(f'') svg.append(f'{old}/{total} old; {len(r["unique_copies"])}/6 copies seen') svg.append('OldFixed · response proportions; direct inventory has only six requests') (OUT / 'connection-sampling.svg').write_text('\n'.join(svg), encoding='utf-8') def main(): OUT.mkdir(parents=True, exist_ok=True) states = [dict(copy=f'copy-{i+1}', version='old' if i == 5 else 'fixed', health='healthy') for i in range(6)] servers, threads, rows, results = [], [], [], {} proxy = None try: for state in states: server = ThreadingHTTPServer(('127.0.0.1', 0), make_handler(state)) server.daemon_threads = False thread = threading.Thread(target=server.serve_forever, kwargs={'poll_interval': .05}) thread.start() servers.append(server) threads.append(thread) proxy = Proxy([server.server_address for server in servers]) for phase in ('before_change', 'after_change'): if phase == 'after_change': states[5]['version'] = 'fixed' results[phase] = {} for mode in ('reused_connection', 'fresh_connections', 'direct_inventory'): start, accepts = len(rows), len(proxy.accepts) connection = None try: count = 6 if mode == 'direct_inventory' else 60 for i in range(count): if connection is None or mode != 'reused_connection': address = servers[i].server_address if mode == 'direct_inventory' else proxy.address connection = http.client.HTTPConnection(*address, timeout=3) fetch(connection, phase, mode, i + 1, rows) if mode != 'reused_connection': connection.close() batch = rows[start:] result = summarize(batch) expected_old = (10 if mode == 'fresh_connections' else 1 if mode == 'direct_inventory' else 0) if phase == 'before_change' else 0 assert result['old_responses'] == expected_old, result assert len(result['unique_copies']) == (1 if mode == 'reused_connection' else 6) assert len(proxy.accepts) - accepts == (1 if mode == 'reused_connection' else 60 if mode == 'fresh_connections' else 0) results[phase][mode] = result finally: if connection: connection.close() finally: if proxy: proxy.close() for server in servers: server.shutdown() server.server_close() for thread in threads: thread.join(timeout=4) assert not thread.is_alive() assert len(rows) == 252 and len(proxy.accepts) == 122 assert not proxy.errors, proxy.errors report = dict(run_at_utc=datetime.now(timezone.utc).isoformat(), scope='Loopback-only controlled illustration; not production, Kubernetes, vulnerability or AI benchmark.', topology='Six HTTP/1.1 servers; TCP proxy chooses round-robin at connection acceptance and preserves affinity.', selection='Copies 1–5 start fixed; copy 6 starts old; first accepted connection selects copy 1.', response_rows=len(rows), proxy_connections=len(proxy.accepts), assertions_passed=True, owned_threads_and_sockets_closed=True, results=results, proxy_accept_log=proxy.accepts, proxy_errors=proxy.errors) with (OUT / 'responses.csv').open('w', newline='', encoding='utf-8') as file: writer = csv.DictWriter(file, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) text = json.dumps(report, indent=2) (OUT / 'results.json').write_text(text+'\n', encoding='utf-8') (OUT / 'stdout.txt').write_text(text+'\n', encoding='utf-8') chart(results) print(text) if __name__ == '__main__': main()