Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
166 changes: 111 additions & 55 deletions system_test/fixture.py
Original file line number Diff line number Diff line change
Expand Up @@ -369,6 +369,54 @@ def http_sql_query(self, sql_query):
raise QueryError(data['error'])
return data

def _assert_server_alive(self):
"""Raise if the server died during startup.

The readiness probe calls this before each attempt so a crashed
server fails fast instead of looping until the timeout. The base
fixture has no process to inspect; the managed and docker fixtures
override it.
"""

def _check_main_http_up(self):
# Probe the main HTTP server's /ping. The initial start gates on
# this: query_version() and every test SQL query run against the main
# HTTP server, so it must be accepting before start() returns. There
# is no QWP ingest at initial-start time, so /ping can't be starved
# by the shared-network pool that serves it.
return self._probe_http(
self.http_server_port, '/ping', lambda status: status == 204)

def _check_min_http_up(self):
# Probe the dedicated min-HTTP health endpoint. A bounce restart gates
# on this: the min server runs on its own worker pool, so a heavy QWP
# ingest load on the shared-network pool (which serves the main HTTP
# server) can't starve the readiness check. HealthCheckProcessor
# answers any path with 200 "Status: Healthy". The bounce path issues
# no SQL after the probe, so main-HTTP readiness isn't required there.
return self._probe_http(
self.http_min_port, '/status', lambda status: 200 <= status < 300)

def _probe_http(self, port, path, status_ok):
self._assert_server_alive()
req = urllib.request.Request(
f'http://{self.host}:{port}{path}',
headers=self.http_headers(),
method='GET')
try:
resp = urllib.request.urlopen(req, timeout=1)
if status_ok(resp.status):
return True
except OSError:
# The server isn't accepting yet. A not-yet-ready local process
# refuses the connection (urllib.error.URLError); docker
# publishes the host port before QuestDB binds, so an early probe
# can connect and then be reset (RemoteDisconnected /
# ConnectionResetError). URLError and socket.timeout are both
# OSError subclasses, so one handler covers every case.
pass
return False

def query_version(self):
try:
res = self.http_sql_query('select build')
Expand Down Expand Up @@ -504,6 +552,11 @@ def __init__(
self.line_tcp_port = None
self.qwp_udp_port = None
self.pg_port = None
# Dedicated min-HTTP health port. The min server runs on its own
# worker pool, so the readiness probe can't be starved by QWP
# ingest saturating the shared-network pool that serves the main
# HTTP server (where /ping lives).
self.http_min_port = None

self.wrap_tls = wrap_tls
self._tls_proxy = None
Expand All @@ -528,10 +581,28 @@ def print_log(self):
sys.stderr.write(textwrap.indent(log, ' '))
sys.stderr.write('\n\n')

def start(self):
def _assert_server_alive(self):
if self._proc.poll() is not None:
raise RuntimeError('QuestDB died during startup.')

def start(self, start_timeout_sec=300, probe_min_http=False):
# start_timeout_sec bounds the wait for the readiness probe to pass.
# The generous default flags only a genuinely stuck boot; the bounce
# fuzz thread passes a tighter, drain-budget-aware value so a
# pathologically slow restart fails as an infra error rather than
# starving the producers' close_drain.
#
# probe_min_http selects which server gates readiness. The initial
# start probes the main HTTP server (/ping): query_version() and all
# test SQL hit it, so it must be accepting before start() returns,
# and there is no ingest yet to starve it. A bounce restart sets
# probe_min_http=True to gate on the dedicated min-HTTP pool instead,
# which reconnecting producers on the shared-network pool can't
# starve; that path issues no SQL, so main-HTTP readiness isn't
# needed.
if self.http_server_port is None:
ports = discover_avail_ports(3)
self.http_server_port, self.line_tcp_port, self.pg_port = ports
(self.http_server_port, self.line_tcp_port,
self.pg_port, self.http_min_port) = discover_avail_ports(4)
if self.qwp_udp and self.qwp_udp_port is None:
self.qwp_udp_port = discover_avail_udp_port()
auth_config = 'line.tcp.auth.db.path=conf/auth.txt' if self.auth else ''
Expand All @@ -552,7 +623,9 @@ def start(self):
http.bind.to=0.0.0.0:{self.http_server_port}
line.tcp.net.bind.to=0.0.0.0:{self.line_tcp_port}
pg.net.bind.to=0.0.0.0:{self.pg_port}
http.min.enabled=false
http.min.enabled=true
http.min.net.bind.to=0.0.0.0:{self.http_min_port}
http.min.worker.count=1
line.udp.enabled=false
qwp.udp.enabled={qwp_udp_enabled}
{qwp_udp_bind}
Expand Down Expand Up @@ -620,28 +693,15 @@ def start(self):
stderr=subprocess.STDOUT,
creationflags=creationflags)

def check_http_up():
if self._proc.poll() is not None:
raise RuntimeError('QuestDB died during startup.')
req = urllib.request.Request(
f'http://127.0.0.1:{self.http_server_port}/ping',
headers=self.http_headers(),
method='GET')
try:
resp = urllib.request.urlopen(req, timeout=1)
if resp.status == 204:
return True
except socket.timeout:
pass
except urllib.error.URLError:
pass
return False

sys.stderr.write('Waiting until HTTP service is up.\n')
check_up = (
self._check_min_http_up if probe_min_http
else self._check_main_http_up)
retry(
check_http_up,
timeout_sec=300,
msg='Timed out waiting for HTTP service to come up.')
check_up,
timeout_sec=start_timeout_sec,
msg=f'Timed out waiting for HTTP service to come up '
f'within {start_timeout_sec}s.')
Comment on lines 700 to +704

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | ⚡ Quick win

Clean up the managed JVM when readiness times out.

The new bounded retry(..., timeout_sec=start_timeout_sec) makes start failures an expected infra signal, but the managed fixture still re-raises without stopping the spawned process or closing the log. Docker already calls self.stop() on the same path; mirror that here so failed initial starts do not leak a JVM/ports.

Proposed fix
         except:
             sys.stderr.write(f'QuestDB log at `{self._log_path}`:\n')
             self.print_log()
+            try:
+                self.stop()
+            except Exception as cleanup_error:
+                sys.stderr.write(
+                    'Failed to clean up QuestDB after start failure: '
+                    f'{type(cleanup_error).__name__}: {cleanup_error}\n')
             raise
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@system_test/fixture.py` around lines 700 - 704, The readiness timeout path in
the managed JVM startup flow currently re-raises without cleaning up the spawned
process or log handle, which can leak resources on failed starts. Update the
startup logic around the retry call in the managed fixture (the same flow that
uses self.stop() for Docker) so that when startup readiness times out, it
explicitly stops the JVM/process and closes any open log before re-raising. Use
the existing startup/cleanup symbols in this fixture to keep the failure path
consistent with other managed resources.

except:
sys.stderr.write(f'QuestDB log at `{self._log_path}`:\n')
self.print_log()
Expand All @@ -656,8 +716,8 @@ def check_http_up():
# doesn't change across restarts, and a bounce restart must stay
# clear of SQL — right after it the reconnecting fuzz producers
# can keep the network workers busy past the 5s query timeout,
# failing the bounce even though /ping already vouched for
# liveness.
# failing the bounce even though the min-HTTP health endpoint
# already vouched for liveness.
if not self._version_queried:
self.version = self.query_version()
self._version_queried = True
Expand Down Expand Up @@ -837,6 +897,7 @@ def __init__(
self.line_tcp_port = None
self.qwp_udp_port = None
self.pg_port = None
self.http_min_port = None # see QuestDbFixture.__init__

self.wrap_tls = wrap_tls
self._tls_proxy = None
Expand All @@ -857,7 +918,9 @@ def _env_config(self):
'QDB_HTTP_BIND_TO': f'0.0.0.0:{self.http_server_port}',
'QDB_LINE_TCP_NET_BIND_TO': f'0.0.0.0:{self.line_tcp_port}',
'QDB_PG_NET_BIND_TO': f'0.0.0.0:{self.pg_port}',
'QDB_HTTP_MIN_ENABLED': 'false',
'QDB_HTTP_MIN_ENABLED': 'true',
'QDB_HTTP_MIN_NET_BIND_TO': f'0.0.0.0:{self.http_min_port}',
'QDB_HTTP_MIN_WORKER_COUNT': '1',
'QDB_LINE_UDP_ENABLED': 'false',
'QDB_QWP_UDP_ENABLED': 'true' if self.qwp_udp else 'false',
'QDB_LINE_TCP_MAINTENANCE_JOB_INTERVAL': '100',
Expand All @@ -884,18 +947,20 @@ def _env_config(self):
env['QDB_LINE_TCP_AUTH_DB_PATH'] = 'conf/auth.txt'
return env

def start(self):
def start(self, start_timeout_sec=300, probe_min_http=False):
# start_timeout_sec / probe_min_http: see QuestDbFixture.start.
if self.http_server_port is None:
ports = discover_avail_ports(3)
self.http_server_port, self.line_tcp_port, self.pg_port = ports
(self.http_server_port, self.line_tcp_port,
self.pg_port, self.http_min_port) = discover_avail_ports(4)
if self.qwp_udp and self.qwp_udp_port is None:
self.qwp_udp_port = discover_avail_udp_port()

cmd = ['run', '-d', '--name', _unique_container_name('qdb_systest')]
self._container = cmd[3]
for key, value in self._env_config().items():
cmd += ['-e', f'{key}={value}']
for port in (self.http_server_port, self.line_tcp_port, self.pg_port):
for port in (self.http_server_port, self.line_tcp_port,
self.pg_port, self.http_min_port):
cmd += ['-p', f'127.0.0.1:{port}:{port}']
if self.qwp_udp:
cmd += ['-p', f'127.0.0.1:{self.qwp_udp_port}:{self.qwp_udp_port}/udp']
Expand All @@ -922,10 +987,14 @@ def start(self):

try:
sys.stderr.write('Waiting until HTTP service is up.\n')
check_up = (
self._check_min_http_up if probe_min_http
else self._check_main_http_up)
retry(
self._check_http_up,
timeout_sec=300,
msg='Timed out waiting for HTTP service to come up.')
check_up,
timeout_sec=start_timeout_sec,
msg=f'Timed out waiting for HTTP service to come up '
f'within {start_timeout_sec}s.')
except:
sys.stderr.write('QuestDB container log:\n')
self.print_log()
Expand All @@ -950,28 +1019,9 @@ def _is_running(self):
check=False, capture=True)
return res.returncode == 0 and res.stdout.strip() == b'true'

def _check_http_up(self):
def _assert_server_alive(self):
if not self._is_running():
raise RuntimeError('QuestDB container died during startup.')
req = urllib.request.Request(
f'http://{self.host}:{self.http_server_port}/ping',
headers=self.http_headers(),
method='GET')
try:
resp = urllib.request.urlopen(req, timeout=1)
if resp.status == 204:
return True
except urllib.error.URLError:
pass
except OSError:
# Docker publishes the host port before QuestDB is ready, so an
# early probe can connect and then be reset
# (http.client.RemoteDisconnected, a ConnectionResetError /
# OSError). The local-process fixture only sees connection-refused
# (URLError) because its port opens with the app. socket.timeout
# is an OSError too.
pass
return False

def __enter__(self):
self.start()
Expand Down Expand Up @@ -1019,7 +1069,13 @@ def stop(self, wait_timeout_sec=30):
# `docker stop` sends SIGTERM (graceful: the JVM runs shutdown
# hooks), then SIGKILLs after the grace period. Equivalent to
# QuestDbFixture.stop()'s terminate()-then-kill().
self._remove_container(['stop', '-t', str(wait_timeout_sec)])
#
# docker's `-t` takes integer seconds; bounce callers pass a float
# budget (e.g. 120.0), which docker rejects ("invalid argument") —
# coerce to int so the graceful grace period is honoured rather
# than silently skipped, leaving only the immediate `rm -f`.
self._remove_container(
['stop', '-t', str(int(wait_timeout_sec))])

def kill(self):
# `docker kill` sends SIGKILL: an ungraceful crash with no
Expand Down
52 changes: 42 additions & 10 deletions system_test/qwp_ws_fuzz.py
Original file line number Diff line number Diff line change
Expand Up @@ -1419,10 +1419,11 @@ class BounceThread(threading.Thread):
* the bounce budget is exhausted (``bounces_performed >= max_bounces``);
* the producers signal completion via ``writers_done.set()``;
* ``stop_event`` is set; or
* a previous bounce raised, in which case ``failure_counter`` gets
bumped and the thread tries one defensive ``stop()`` + ``start()``
before exiting so the rest of the test still has a server to talk
to (and no half-started instance is left behind).
* a bounce raised — e.g. the server overran ``stop_timeout_s`` on the
way down or ``restart_timeout_s`` on the way back up — in which case
``failure_counter`` gets bumped and the thread tries one defensive
``stop()`` + ``start()`` before exiting so the rest of the test still
has a server to talk to (and no half-started instance is left behind).

Each bounce is atomic from the caller's perspective: once the loop
enters a bounce cycle it completes ``stop()`` + ``start()`` before
Expand All @@ -1439,6 +1440,7 @@ def __init__(
min_interval_s: float,
max_interval_s: float,
stop_timeout_s: float,
restart_timeout_s: float,
writers_done: threading.Event,
stop_event: threading.Event,
record_failure,
Expand All @@ -1452,6 +1454,7 @@ def __init__(
self._min_interval_s = min_interval_s
self._max_interval_s = max(max_interval_s, min_interval_s)
self._stop_timeout_s = stop_timeout_s
self._restart_timeout_s = restart_timeout_s
self._writers_done = writers_done
self._stop_event = stop_event
self._record_failure = record_failure
Expand Down Expand Up @@ -1486,29 +1489,58 @@ def run(self):
# before start() rebinds them.
time.sleep(0.02 + self._rnd.next_int(200) / 1000.0)
self._log(f'fuzz bounce #{idx}: starting QDB')
self._fixture.start()
# Cap the restart wait at restart_timeout_s: a restart that
# overruns it is a stuck boot, and failing here gives the
# trustworthy, correctly-attributed signal (server too slow
# to restart). The producers' reconnect/close_drain budgets
# are sized above a full stop()+start() down-window (see
# TestQwpWsFuzz._producer_loop), so a restart inside this cap
# never strands them — only one that trips the cap does, and
# that surfaces here, not as a downstream client timeout.
#
# Gate on the min-HTTP pool, not main /ping: the producers
# are reconnecting and saturating the shared-network pool
# that serves main HTTP, so a main-HTTP probe here would be
# starved and time out even on a healthy restart. The
# restart issues no SQL, so main-HTTP readiness isn't needed.
self._fixture.start(
start_timeout_sec=self._restart_timeout_s,
probe_min_http=True)
self.bounces_performed += 1
self._log(f'fuzz bounce #{idx}: server back up')
except Exception as e: # noqa: BLE001 — any lifecycle failure fails the run
# A raise here is a real failure, not noise: stop() raises
# when the server won't shut down within its timeout, and
# start() raises when it won't come back up. Either way we
# record it so the end-of-run assertion fails.
# start() raises when it won't come back up within
# restart_timeout_s. Both bounds are sized so only a stuck
# or pathologically slow server trips them, and a server that
# slow is exactly what the producers' down-window budget is
# sized to outlast — so attributing it here, rather than as a
# downstream client timeout, is the correct call.
self._record_failure(
f'fuzz bounce: unexpected failure at attempt '
f'fuzz bounce: server lifecycle failed at attempt '
f'{self.bounces_performed + 1}: '
f'{type(e).__name__}: {e}')
# One defensive recovery attempt so the rest of the test
# has a chance to surface the underlying assertion failure
# rather than a query timeout. stop() first: if start()
# failed partway it may have left a process behind, and we
# must not launch a second one next to it.
# must not launch a second one next to it. Reuse the bounded
# restart timeout and the min-HTTP probe: producers may still
# be reconnecting, so a main-HTTP probe would be starved, and
# the bound keeps this recovery from running unbounded. The
# failure is already recorded, so even if recovery overruns
# the test's bounce wind-down the run still fails with correct
# attribution; the wind-down's is_alive check flags a thread
# still stuck here.
try:
self._fixture.stop(wait_timeout_sec=self._stop_timeout_s)
except Exception:
pass
try:
self._fixture.start()
self._fixture.start(
start_timeout_sec=self._restart_timeout_s,
probe_min_http=True)
except Exception:
pass
return
Expand Down
Loading