Skip to content

Commit 8a6c365

Browse files
ryzhykryzhyk
committed
test(adapters): add fault-tolerant input-replay determinism regression test
Add `tests/platform/test_fault_tolerance.py::test_ft_input_replay_determinism`, an enterprise-only regression test for the intermittent fault-tolerant restart-recovery failures seen in CI. It creates a single-host fault-tolerant pipeline (`fault_tolerance: latest_checkpoint`, datagen input, materialized table + view) and repeatedly force-stops (crashes) and resumes it while data is flowing. Each resume restores the latest checkpoint and replays the input log; replay must be deterministic, so the pipeline must return to Running every cycle. On a buggy build the input is re-sharded across workers non-deterministically on replay and the engine self-terminates with "Logged and replayed step N contained different numbers of records or hashes" (PipelineTerminated), which the test reports as a recovery failure. The bug requires the input sharded across >= 2 workers (a single worker replays deterministically); the test pins 2 workers as the most reliable trigger and crashes for 15 cycles. Reproduces within the first few cycles. Signed-off-by: Leonid Ryzhyk <leonid@feldera.com>
1 parent 9b4c369 commit 8a6c365

1 file changed

Lines changed: 151 additions & 0 deletions

File tree

‎python/tests/platform/test_fault_tolerance.py‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
"""Fault-tolerant restart-recovery tests (enterprise only).
2+
3+
A fault-tolerant pipeline (`fault_tolerance: latest_checkpoint`) must replay its
4+
logged input *deterministically* across restarts: on resume the engine restores
5+
the latest checkpoint and replays the input log, and every replayed step must
6+
contain the same records (count + hash) as were originally logged.
7+
8+
``test_ft_input_replay_determinism`` exercises this by repeatedly force-stopping
9+
(crashing) and restarting a single-host FT pipeline while a datagen connector is
10+
actively producing, so each restart replays a partially-logged step. On a buggy
11+
build the input is re-sharded across workers non-deterministically on replay; the
12+
engine detects
13+
14+
Logged and replayed step N contained different numbers of records or hashes
15+
16+
and self-terminates the pipeline (``PipelineTerminated`` -> ``Stopped``), which
17+
this test reports as a recovery failure. The bug requires >= 2 workers (the input
18+
is sharded across workers); with a single worker replay is deterministic.
19+
20+
Regression test for the intermittent ``fault_tolerance_tests`` CI failures.
21+
"""
22+
23+
import json
24+
import os
25+
import time
26+
from http import HTTPStatus
27+
28+
from tests import enterprise_only
29+
from .helper import (
30+
api_url,
31+
get,
32+
post_json,
33+
wait_for_program_success,
34+
start_pipeline,
35+
stop_pipeline,
36+
wait_for_condition,
37+
gen_pipeline_name,
38+
)
39+
40+
# A large datagen limit keeps input flowing across many restart cycles, so each
41+
# force-stop interrupts an in-progress step whose input is in the log but not yet
42+
# checkpointed -- the precondition for exercising input replay.
43+
_DATAGEN_CONNECTOR = json.dumps(
44+
[
45+
{
46+
"transport": {
47+
"name": "datagen",
48+
"config": {
49+
"seed": 12345,
50+
"plan": [
51+
{
52+
"limit": 200000,
53+
"rate": 500,
54+
"fields": {
55+
"id": {"strategy": "increment", "range": [0, 2000000]},
56+
"name": {"strategy": "word"},
57+
"grp": {"strategy": "uniform", "range": [0, 50]},
58+
},
59+
}
60+
],
61+
},
62+
}
63+
}
64+
]
65+
)
66+
67+
_SQL = f"""
68+
CREATE TABLE example1 (
69+
id BIGINT NOT NULL PRIMARY KEY,
70+
name VARCHAR NOT NULL,
71+
grp INT NOT NULL
72+
) WITH (
73+
'materialized' = 'true',
74+
'connectors' = '{_DATAGEN_CONNECTOR}'
75+
);
76+
77+
CREATE MATERIALIZED VIEW view1 AS (SELECT * FROM example1);
78+
"""
79+
80+
# Number of crash -> resume cycles. The bug reproduces within the first few
81+
# cycles; the extra cycles give margin against its intermittency.
82+
_CYCLES = 15
83+
84+
# The bug needs the input sharded across >= 2 workers (a single worker replays
85+
# deterministically). 2 workers is the most reliable trigger -- it reproduces
86+
# within a few cycles, whereas higher worker counts need more cycles -- so the
87+
# test pins 2 rather than using FELDERA_TEST_NUM_WORKERS.
88+
_WORKERS = 2
89+
90+
91+
def _create_ft_pipeline(name: str, workers: int):
92+
"""Create a single-host fault-tolerant pipeline (mirrors helper.create_pipeline
93+
but with a fault-tolerant runtime config)."""
94+
payload = {
95+
"name": name,
96+
"program_code": _SQL,
97+
"runtime_config": {
98+
"fault_tolerance": "latest_checkpoint",
99+
"storage": True,
100+
"workers": workers,
101+
"logging": "debug",
102+
},
103+
}
104+
runtime_version = os.environ.get("FELDERA_RUNTIME_VERSION")
105+
if runtime_version:
106+
payload["program_config"] = {"runtime_version": runtime_version}
107+
r = post_json(api_url("/pipelines"), payload)
108+
assert r.status_code == HTTPStatus.CREATED, r.text
109+
wait_for_program_success(name, 1)
110+
111+
112+
def _deployment(name: str):
113+
d = get(api_url(f"/pipelines/{name}")).json()
114+
return d.get("deployment_status"), d.get("deployment_error")
115+
116+
117+
@enterprise_only
118+
@gen_pipeline_name
119+
def test_ft_input_replay_determinism(pipeline_name):
120+
_create_ft_pipeline(pipeline_name, _WORKERS)
121+
start_pipeline(pipeline_name)
122+
123+
for cycle in range(1, _CYCLES + 1):
124+
wait_for_condition(
125+
f"running before crash (cycle {cycle})",
126+
lambda: _deployment(pipeline_name)[0] == "Running",
127+
timeout_s=90.0,
128+
poll_interval_s=0.5,
129+
)
130+
# Let datagen feed a partial, not-yet-checkpointed step into the log.
131+
time.sleep(1.0)
132+
133+
stop_pipeline(pipeline_name, force=True) # crash: no final checkpoint
134+
start_pipeline(pipeline_name, wait=False) # resume: restore + replay log
135+
136+
def recovered():
137+
status, error = _deployment(pipeline_name)
138+
assert not (status == "Stopped" and error), (
139+
f"fault-tolerant recovery failed on cycle {cycle}: the pipeline "
140+
f"self-terminated on resume -- most likely non-deterministic input "
141+
f"replay ('Logged and replayed step N contained different numbers "
142+
f"of records or hashes'). deployment_error={error}"
143+
)
144+
return status == "Running"
145+
146+
wait_for_condition(
147+
f"recovered to Running after crash (cycle {cycle})",
148+
recovered,
149+
timeout_s=90.0,
150+
poll_interval_s=0.5,
151+
)

0 commit comments

Comments
 (0)