fix(sidecar_init): log after spawn in start_all to close SIGTERM race
`start_all()` was logging "starting {name}" before appending the
process to `self.procs`, so a SIGTERM arriving between the log write
and the append would call `request_shutdown()` with the new daemon
absent from `self.procs` and therefore never forwarded SIGTERM.
Moving the log after the append eliminates the window.
Signal handlers are also moved before `sup.start_all()` in `main()` so
they are registered before any child is spawned; previously there was a
window where SIGTERM used the default handler (silent process death) if
it arrived between `start_all()` returning and `signal.signal()`.
`TestMainEndToEnd._run` is updated to use a reader thread and a
`wait_for_output` synchronisation barrier instead of a fixed sleep, so
`test_sigterm_clean_shutdown` does not race against slow CI startup time.
This commit is contained in:
@@ -12,6 +12,7 @@ import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import unittest
|
||||
import warnings
|
||||
@@ -476,6 +477,7 @@ class TestMainEndToEnd(unittest.TestCase):
|
||||
|
||||
def _run(self, daemons_csv: str, send_signal: int | None,
|
||||
wait_before_signal: float = 0.4,
|
||||
wait_for_output: tuple[str, ...] = (),
|
||||
overall_timeout: float = 6.0) -> tuple[int, str]:
|
||||
"""Spawn sidecar_init.main() in a child process with the
|
||||
DAEMONS list patched to harmless `sleep 30` commands.
|
||||
@@ -496,19 +498,53 @@ class TestMainEndToEnd(unittest.TestCase):
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||||
env=env,
|
||||
)
|
||||
collected: list[bytes] = []
|
||||
ready = threading.Event()
|
||||
need: set[str] = set(wait_for_output)
|
||||
|
||||
def _drain() -> None:
|
||||
assert proc.stdout is not None
|
||||
for raw in iter(proc.stdout.readline, b""):
|
||||
collected.append(raw)
|
||||
if need:
|
||||
text = raw.decode("utf-8", errors="replace")
|
||||
need.difference_update(p for p in list(need) if p in text)
|
||||
if not need:
|
||||
ready.set()
|
||||
ready.set() # also fires on EOF so the main thread isn't stuck
|
||||
|
||||
reader = threading.Thread(target=_drain, daemon=True)
|
||||
reader.start()
|
||||
|
||||
try:
|
||||
if send_signal is not None:
|
||||
time.sleep(wait_before_signal)
|
||||
if wait_for_output:
|
||||
# Wait until all expected lines have appeared, then
|
||||
# signal. This eliminates the fixed-sleep race where
|
||||
# SIGTERM arrives before the subprocess installs its
|
||||
# handler on a slow CI runner.
|
||||
if not ready.wait(timeout=overall_timeout):
|
||||
proc.kill()
|
||||
reader.join(timeout=2.0)
|
||||
self.fail("timed out waiting for expected output lines")
|
||||
else:
|
||||
time.sleep(wait_before_signal)
|
||||
proc.send_signal(send_signal)
|
||||
out_b, _ = proc.communicate(timeout=overall_timeout)
|
||||
proc.wait(timeout=overall_timeout)
|
||||
except subprocess.TimeoutExpired:
|
||||
proc.kill()
|
||||
out_b, _ = proc.communicate()
|
||||
proc.wait()
|
||||
reader.join(timeout=2.0)
|
||||
self.fail("sidecar_init main() did not exit before timeout")
|
||||
return proc.returncode, out_b.decode("utf-8", errors="replace")
|
||||
|
||||
reader.join(timeout=2.0)
|
||||
return proc.returncode, b"".join(collected).decode("utf-8", errors="replace")
|
||||
|
||||
def test_sigterm_clean_shutdown(self):
|
||||
rc, out = self._run("alpha,beta", signal.SIGTERM)
|
||||
rc, out = self._run(
|
||||
"alpha,beta", signal.SIGTERM,
|
||||
wait_for_output=("starting alpha", "starting beta"),
|
||||
)
|
||||
self.assertIn("starting alpha", out)
|
||||
self.assertIn("starting beta", out)
|
||||
self.assertIn("forwarding SIGTERM", out)
|
||||
|
||||
Reference in New Issue
Block a user