test/harness/exp_fd_flatness.py

1#!/usr/bin/env python3

2"""Do the failure paths leak file descriptors?

4Counts the worker process's open fds before and after exercising the

5paths this change touches: many post-time confirmation waits, then a

6full freeze -> detect -> reconnect cycle.

8Run it against BOTH binaries. A single post-change reading cannot show

9"we did not regress this" -- it can only show a number. The value is in

10the comparison, so the pair is the measurement.

12 APIARY_BIN=/path/to/pre-change python3 test/harness/exp_fd_flatness.py

13 APIARY_BIN=/path/to/post-change python3 test/harness/exp_fd_flatness.py

14"""

15import os

16import re

17import sys

18import time

20sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

21from mcpdrv import (Bot, leader_env, worker_env, load_state, # noqa: E402

22 say, check, unique_nick)

23from proxy import Proxy # noqa: E402

25PROXY_PORT = int(os.environ.get("HARNESS_PROXY_PORT", "49785"))

26POSTS = int(os.environ.get("HARNESS_POSTS", "40"))

29def fd_count(pid):

30 try:

31 return len(os.listdir(f"/proc/{pid}/fd"))

32 except OSError:

33 return None

36def main():

37 st = load_state()

38 port, token = st["PORT"], st["LEADER_TOKEN"]

39 proxy = Proxy(PROXY_PORT, int(port))

40 bots, nick = [], None

41 try:

42 leader = Bot("fd-leader", leader_env(port, token))

43 bots.append(leader)

44 leader.initialize()

45 if not leader.wait_connected(30):

46 raise SystemExit("leader never connected")

48 nick = unique_nick("fd-")

49 out = leader.call("spawn-worker", nick=nick, **{"expires-in": "60m"})

50 m = re.search(r"[a-f0-9]{64}", out)

51 if not m:

52 raise SystemExit(f"spawn failed: {out!r}")

53 w = Bot("fd-worker", worker_env(str(PROXY_PORT), m.group(0), nick))

54 bots.append(w)

55 w.initialize()

56 if not w.wait_connected(30):

57 raise SystemExit("worker never connected")

58 time.sleep(2)

60 pid = w.proc.pid

61 base = fd_count(pid)

62 say(f" worker pid={pid} fds after connect: {base}")

63 if base is None:

64 raise SystemExit("cannot read /proc/<pid>/fd on this host")

66 say(f"\n== {POSTS} healthy posts (confirmation path) ==")

67 for i in range(POSTS):

68 w.call("send-channel", text=f"fd probe healthy {i:03d}")

69 after_posts = fd_count(pid)

70 say(f" fds after {POSTS} confirmed posts: {after_posts}")

72 say("\n== freeze -> detect -> reconnect cycle ==")

73 proxy.freeze_existing()

74 t0 = time.time()

75 i = 0

76 while time.time() - t0 < 45:

77 i += 1

78 try:

79 w.call("send-channel",

80 text=f"fd probe outage {i:03d}", timeout=25)

81 except TimeoutError:

82 pass

83 time.sleep(5)

84 after_outage_posts = fd_count(pid)

85 say(f" fds after {i} posts into a dead link: {after_outage_posts}")

87 proxy.frozen_at = {}

88 deadline = time.time() + 150

89 while (time.time() < deadline

90 and not w.grep_stderr("Apiary reconnected")):

91 time.sleep(1)

92 reconnected = bool(w.grep_stderr("Apiary reconnected"))

93 time.sleep(5)

94 after_reconnect = fd_count(pid)

95 say(f" reconnected: {reconnected}")

96 say(f" fds after reconnect: {after_reconnect}")

98 say("")

99 ok = True

100 ok &= check("fd count flat across the confirmation path",

101 after_posts - base <= 1,

102 f"{base} -> {after_posts} (delta "

103 f"{after_posts - base}) over {POSTS} posts")

104 ok &= check("fd count flat across posts into a dead link",

105 after_outage_posts - base <= 1,

106 f"{base} -> {after_outage_posts} (delta "

107 f"{after_outage_posts - base}) over {i} posts")

108 ok &= check("fd count flat across detect + reconnect",

109 abs(after_reconnect - base) <= 1,

110 f"{base} -> {after_reconnect} (delta "

111 f"{after_reconnect - base})")

112 say(f"\n SUMMARY connect={base} posts={after_posts} "

113 f"outage={after_outage_posts} reconnect={after_reconnect}")

114 return 0 if ok else 1

115 finally:

116 if bots and bots[0].alive() and nick:

117 bots[0].call("revoke-worker", nick=nick)

118 for b in bots:

119 b.close()

120 proxy.close()

121

122

123if __name__ == "__main__":

124 sys.exit(main())