Affinity, membership and state continuity

#!/usr/bin/env python3
"""Original bounded loopback fixture. No external services or financial operations."""
import argparse
from collections import Counter
from concurrent.futures import ThreadPoolExecutor
import hashlib
from http.client import HTTPConnection
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
from pathlib import Path
import platform
import re
import signal
import socket
import subprocess
import tempfile
import threading
import time


def digest(path):
 return hashlib.sha256(Path(path).read_bytes).hexdigest


def run(binary):
 version = subprocess.run([binary, '-V'], capture_output=True, text=True, check=True).stderr
 started, release, completed = threading.Event, threading.Event, threading.Event
 events, stores = [], {name: {} for name in 'ABC'}
 lock = threading.Lock
 servers, threads, checks, generations = [], [], {}, []
 master = None
 evidence = dict(scriptSha256=digest(__file__), binarySha256=digest(binary), nginxVersion=version,
 pythonVersion=platform.python_version, platform=platform.platform,
 scope='One NGINX worker, three synthetic HTTP loopback backends, 96 selected keys and explicit in-memory fixture state; not a throughput benchmark, durable session store, identity system, production drain or business cancellation test.')
 tmp = tempfile.TemporaryDirectory(prefix='dr-lb-affinity-')
 root = Path(tmp.name)
 pool = ThreadPoolExecutor(max_workers=1)

 def handler(name):
 class Handler(BaseHTTPRequestHandler):
 protocol_version = 'HTTP/1.1'

 def log_message(self, *args):
 pass

 def do_POST(self):
 self.do_GET

 def do_GET(self):
 key = self.headers.get('X-Lab-Key', '')
 with lock:
 events.append(dict(backend=name, path=self.path, key=key, method=self.command))
 if self.path == '/hold' and name == 'A':
 started.set
 if not release.wait(12):
 return
 # Deliberately independent of socket liveness. No production side effects.
 completed.set
 if self.path == '/session':
 if self.command == 'POST':
 value = self.rfile.read(int(self.headers.get('Content-Length', '0'))).decode
 with lock:
 stores[name][key] = value
 with lock:
 value = stores[name].get(key)
 else:
 value = None
 data = json.dumps(dict(backend=name, value=value)).encode
 try:
 self.send_response(200)
 self.send_header('Content-Length', str(len(data)))
 self.send_header('Connection', 'close')
 self.end_headers
 self.wfile.write(data)
 except (BrokenPipeError, ConnectionResetError):
 pass
 self.close_connection = True
 return Handler

 try:
 for name in 'ABC':
 server = ThreadingHTTPServer(('127.0.0.1', 0), handler(name))
 server.daemon_threads = False
 thread = threading.Thread(target=server.serve_forever, kwargs={'poll_interval':.05})
 thread.start
 servers.append(server)
 threads.append(thread)
 ports = dict(zip('ABC', [s.server_port for s in servers]))
 with socket.socket as reserve:
 reserve.bind(('127.0.0.1', 0))
 port = reserve.getsockname[1]
 conf = root / 'nginx.conf'

 def configure(members, hold, generation):
 lines = '\n'.join(f' server 127.0.0.1:{ports[n]};' for n in members)
 config = f'''daemon off;
worker_processes 1;
worker_shutdown_timeout 1s;
pid {root}/master.pid;
error_log {root}/error.log notice;
events {{ worker_connections 256; }}
http {{
 log_format lab '$request_uri|$status|$upstream_status|$upstream_addr'
 access_log {root}/access.log lab;
 upstream affinity {{ hash $http_x_lab_key consistent; {lines} keepalive 0; }}
 upstream held {{ server 127.0.0.1:{ports[hold]}; keepalive 0; }}
 server {{
 listen 127.0.0.1:{port};
 add_header X-Lab-Generation {generation} always;
 proxy_http_version 1.1;
 proxy_set_header Connection close;
 proxy_connect_timeout 2s;
 proxy_read_timeout 10s;
 proxy_next_upstream off;
 location / {{ proxy_pass http://affinity; }}
 location = /hold {{ proxy_pass http://held; }}
 }}
}}
'''
 conf.write_text(config)
 result = subprocess.run([binary, '-p', str(root) + '/', '-c', str(conf), '-t'], capture_output=True, text=True)
 assert result.returncode == 0, result.stderr
 generations.append(dict(generation=generation, members=list(members), hold=hold, configuration=config))

 def request(path='/', key='probe', method='GET', body=None):
 conn = HTTPConnection('127.0.0.1', port, timeout=6)
 try:
 conn.request(method, path, body=body, headers={'Connection': 'close', 'X-Lab-Key': key})
 res = conn.getresponse
 data = res.read
 return dict(status=res.status, generation=res.getheader('X-Lab-Generation'), **json.loads(data))
 except Exception as exc:
 return dict(error=type(exc).__name__)
 finally:
 conn.close

 def wait_generation(generation):
 deadline = time.monotonic + 5
 while time.monotonic < deadline:
 r = request('/ready')
 if r.get('generation') == generation:
 return r
 time.sleep(.02)
 raise AssertionError('new configuration generation not observed')

 def reload(members, hold, generation, settle=True):
 prior = re.findall(r'start worker process (\d+)', (root / 'error.log').read_text)[-1]
 configure(members, hold, generation)
 master.send_signal(signal.SIGHUP)
 result = wait_generation(generation)
 # Observing a new worker does not prove the previous one has stopped accepting.
 # Settle membership experiments; preserve overlap deliberately for the deadline test.
 if settle:
 deadline = time.monotonic + 5
 marker = f'worker process {prior} exited with code 0'
 while marker not in (root / 'error.log').read_text:
 assert time.monotonic < deadline, 'previous worker did not exit'
 time.sleep(.02)
 return result

 configure('AB', 'A', 'ab')
 master = subprocess.Popen([binary, '-p', str(root) + '/', '-c', str(conf)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 wait_generation('ab')
 keys = [f'key-{i:03}' for i in range(96)]
 before = {k: request(key=k)['backend'] for k in keys}
 repeat = {k: request(key=k)['backend'] for k in keys}
 checks['stableMembership'] = dict(keys=96, identicalMappings=before == repeat)
 checks['sampleCoverage'] = dict(backends=sorted(set(before.values)))
 # Same routing key does not represent separate authenticated identities.
 shared = [request(key=keys[0])['backend'] for _ in range(8)]
 checks['sharedKey'] = dict(requests=8, distinctBackends=len(set(shared)), authenticationImplemented=False)
 reload('ABC', 'A', 'abc')
 after = {k: request(key=k)['backend'] for k in keys}
 moved = [k for k in keys if before[k]!= after[k]]
 checks['membershipRemap'] = dict(someMoved=0 < len(moved) < 96, movedOnlyToAddedBackend=all(after[k] == 'C' for k in moved))
 checks['unmovedOwners'] = dict(retainedOldOwner=all(before[k] == after[k] for k in keys if after[k]!= 'C'), allThreeObserved=set(after.values) == set('ABC'))
 reload('AB', 'A', 'session-ab')
 key = next(k for k in keys if before[k] == 'A')
 written = request('/session', key, 'POST', 'synthetic-session-v1')
 read = request('/session', key)
 checks['localState'] = dict(writeBackend=written['backend'], readBackend=read['backend'], valuePresent=read['value'] == 'synthetic-session-v1')
 reload('B', 'A', 'session-b')
 missing = request('/session', key)
 checks['remappedState'] = dict(backend=missing['backend'], valueMissing=missing['value'] is None, originalStillInA=stores['A'][key] == 'synthetic-session-v1')
 # Explicit test-fixture copy, not a replication capability of NGINX.
 with lock:
 stores['B'][key] = stores['A'][key]
 recovered = request('/session', key)
 checks['explicitFixtureCopy'] = dict(backend=recovered['backend'], valuePresent=recovered['value'] == 'synthetic-session-v1', nginxReplication=False)
 future = pool.submit(request, '/hold', 'held-work')
 assert started.wait(3), 'backend A did not start held request'
 reload('B', 'B', 'deadline-b', settle=False)
 new = request('/hold', 'new-work')
 checks['newTraffic'] = dict(backend=new['backend'], status=new['status'], oldClientPending=not future.done)
 old = future.result(timeout=5)
 checks['shutdownDeadline'] = dict(clientFailed='error' in old, clientError=old.get('error'), backendReleased=release.is_set)
 checks['backendPending'] = dict(started=started.is_set, completedBeforeRelease=completed.is_set)
 release.set
 checks['completionAfterDisconnect'] = dict(completedAfterRelease=completed.wait(3), financialSideEffects=False)
 assert checks['stableMembership']['identicalMappings']
 assert checks['sampleCoverage']['backends'] == ['A', 'B']
 assert checks['sharedKey']['distinctBackends'] == 1
 assert all(checks['membershipRemap'].values) and all(checks['unmovedOwners'].values)
 assert checks['localState'] == dict(writeBackend='A', readBackend='A', valuePresent=True)
 assert checks['remappedState'] == dict(backend='B', valueMissing=True, originalStillInA=True)
 assert checks['explicitFixtureCopy'] == dict(backend='B', valuePresent=True, nginxReplication=False)
 assert checks['newTraffic'] == dict(backend='B', status=200, oldClientPending=True)
 assert checks['shutdownDeadline']['clientFailed']
 assert checks['shutdownDeadline']['backendReleased'] is False
 assert checks['backendPending'] == dict(started=True, completedBeforeRelease=False)
 assert checks['completionAfterDisconnect']['completedAfterRelease']
 evidence.update(checks=checks, passed=len(checks), failed=0, mappingBefore=before, mappingAfter=after,
 remappedCount=len(moved), mappingCountsBefore=dict(Counter(before.values)),
 mappingCountsAfter=dict(Counter(after.values)), generations=generations,
 backendEvents=events, clientAfterDeadline=old)
 finally:
 release.set
 pool.shutdown(wait=True)
 if master is not None and master.poll is None:
 master.send_signal(signal.SIGQUIT)
 try:
 master.wait(timeout=5)
 except subprocess.TimeoutExpired:
 master.terminate
 master.wait(timeout=3)
 for server in servers:
 server.shutdown
 server.server_close
 for thread in threads:
 thread.join(timeout=3)
 evidence['childExited'] = master is not None and master.poll is not None
 evidence['backendThreadsStopped'] = all(not thread.is_alive for thread in threads)
 evidence['errorLog'] = (root / 'error.log').read_text if (root / 'error.log').exists else ''
 evidence['accessLog'] = (root / 'access.log').read_text.splitlines if (root / 'access.log').exists else []
 tmp.cleanup
 evidence['temporaryDirectoryRemoved'] = not root.exists
 return evidence


if __name__ == '__main__':
 parser = argparse.ArgumentParser
 parser.add_argument('--output', required=True)
 args = parser.parse_args
 data = run(os.environ['DR_NGINX_BIN'])
 Path(args.output).write_text(json.dumps(data, indent=2) + '\n')
 print(json.dumps(dict(passed=data['passed'], failed=data['failed'], remappedCount=data['remappedCount'], checks=data['checks'])))

Related topics: Algorithms, affinity, and state · Retries, limits, and client origin · Draining, releases, and operations

Reference: NGINX upstream hashing and membership · BigSavant load balancing 2026-09; selected NGINX, HAProxy 3.2, Kubernetes and AWS ALB behavior