test/test-leader-multi-relay-stress.py

1#!/usr/bin/env python3

2"""Leader-courier multi-relay burst repro.

4Reproduces the GC heap-corruption SIGSEGV tracked in task

5`fix-courier-leader-multi-relay-crash`: connects N worker sockets

6simultaneously via a threading.Barrier, each sends K messages then

7closes, and we close the leader's stdin. The buggy pre-fix leader

8segfaults inside Sigil's GC mark phase (shadow_stack parent=PAIR with

9garbage pointer) 3-15 seconds after stdin close.

11Usage:

12 COURIER_BIN=/path/to/courier python3 test-leader-multi-relay-stress.py

14Exit 0 = pass (leader shut down cleanly); exit 1 = crash reproduced.

16Regression contract: a fixed leader must pass this test 20 runs in a row

17with zero crashes. SIGIL_GC_CRASH_DEBUG=1 on the binary prints root/

18parent info to stderr on the bad mark.

19"""

20import json, os, socket, subprocess, sys, tempfile, threading, time, shutil, select

22COURIER = os.environ.get("COURIER_BIN",

23 "/home/daviwil/Projects/Code/sigil/courier/build/release/bin/courier")

24N_RELAYS = int(os.environ.get("N_RELAYS", "5"))

25N_MSGS = int(os.environ.get("N_MSGS", "50"))

26RUNS = int(os.environ.get("RUNS", "1"))

27SHUTDOWN_WAIT = float(os.environ.get("SHUTDOWN_WAIT", "20"))

29def send(p, msg):

30 try: p.stdin.write((json.dumps(msg) + "\n").encode()); p.stdin.flush(); return True

31 except (BrokenPipeError, OSError): return False

33def read_line(p, timeout=3.0):

34 end = time.time() + timeout

35 while time.time() < end:

36 r, _, _ = select.select([p.stdout], [], [], max(0.01, end - time.time()))

37 if r:

38 line = p.stdout.readline()

39 return line if line else None

40 return None

42def one_run(idx):

43 relay_dir = tempfile.mkdtemp(prefix=f"courier-stress-{idx}-")

44 core_dir = tempfile.mkdtemp(prefix=f"courier-core-{idx}-")

45 env = os.environ.copy()

46 env["COURIER_RELAY_DIR"] = relay_dir

47 wrapper = f"cd {core_dir}; ulimit -c unlimited; exec {COURIER}"

48 p = subprocess.Popen(["bash", "-c", wrapper],

49 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,

50 env=env)

52 send(p, {"jsonrpc":"2.0","id":1,"method":"initialize",

53 "params":{"protocolVersion":"2024-11-05","capabilities":{},

54 "clientInfo":{"name":"stress","version":"1"}}})

55 if not read_line(p, 5):

56 p.kill(); p.wait(); return "no-init"

57 send(p, {"jsonrpc":"2.0","method":"notifications/initialized"})

59 names = [f"stress-{idx}-{i}" for i in range(N_RELAYS)]

60 for i, name in enumerate(names):

61 send(p, {"jsonrpc":"2.0","id":10+i,"method":"tools/call",

62 "params":{"name":"create-relay","arguments":{"name":name}}})

63 read_line(p, 3)

65 barrier = threading.Barrier(N_RELAYS)

66 def wfn(name):

67 s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)

68 try: s.connect(os.path.join(relay_dir, name + ".sock"))

69 except FileNotFoundError: return

70 barrier.wait()

71 for k in range(N_MSGS):

72 try:

73 s.sendall((json.dumps({"type":"message","sender":"worker",

74 "text":f"burst {name} {k}"}) + "\n").encode())

75 except: break

76 time.sleep(0.2)

77 s.close()

79 ts = [threading.Thread(target=wfn, args=(n,)) for n in names]

80 for t in ts: t.start()

81 for t in ts: t.join(10)

82 time.sleep(1.0)

84 try: p.stdin.close()

85 except: pass

86 try: p.wait(timeout=SHUTDOWN_WAIT)

87 except subprocess.TimeoutExpired:

88 p.kill(); p.wait()

89 result = "hang"

90 else:

91 result = "crash" if p.returncode < 0 else "ok"

93 cores = []

94 try: cores = [f for f in os.listdir(core_dir) if f.startswith("core")]

95 except: pass

96 print(f"[run {idx}] result={result} exit={p.returncode} cores={len(cores)}",

97 flush=True)

98 if not cores: shutil.rmtree(core_dir, ignore_errors=True)

99 shutil.rmtree(relay_dir, ignore_errors=True)

100 return result

101

102if __name__ == "__main__":

103 crashes = 0

104 for i in range(RUNS):

105 if one_run(i) == "crash":

106 crashes += 1

107 print(f"=== {crashes}/{RUNS} crashes ===")

108 sys.exit(1 if crashes > 0 else 0)