feat(runtime): add failover replay [GMS 01/18]#11049
Conversation
github-actions
Bot
added
the
frontend
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
| let resp_bytes = serde_json::to_vec(&resp_wrapper) | ||
| .expect("fatal error: invalid response object - this should never happen"); |
There was a problem hiding this comment.
🔴 Empty-stream error frame uses wrong serialization format, corrupting the response when non-JSON codec is active
The empty-stream error frame is serialized with hard-coded JSON (serde_json::to_vec at lib/runtime/src/pipeline/network/ingress/push_handler.rs:234) instead of the negotiated request-plane codec, so the receiving side fails to decode it when MsgPack is configured.
Impact: When the request-plane codec is MsgPack, workers that produce an empty response stream send a JSON-encoded error frame that the client cannot deserialize, replacing a clear "stream ended before first response" error with a confusing deserialization failure.
Codec path inconsistency in pump_response_stream
Every other serialization site in pump_response_stream uses payload_codec.encode(&resp_wrapper) — see the normal response path at line 176 and the complete-final frame at line 260-262. The new empty-stream error injection path at line 234 bypasses this and calls serde_json::to_vec(&resp_wrapper) directly. The payload_codec parameter is available in scope and should be used here instead.
On the egress side (lib/runtime/src/pipeline/network/egress/addressed_router.rs:81-82), the decoder uses payload_codec.decode::<NetworkStreamWrapper<U>>(&res_bytes) for every frame. When the codec is MsgPack, the JSON bytes from the ingress error frame will fail this decode, producing a generic deserialization error instead of the typed Disconnected error that the migration retry manager needs to trigger failover replay.
| let resp_bytes = serde_json::to_vec(&resp_wrapper) | |
| .expect("fatal error: invalid response object - this should never happen"); | |
| let resp_bytes = payload_codec | |
| .encode(&resp_wrapper) | |
| .expect("fatal error: invalid request-plane response object"); |
Was this helpful? React with 👍 or 👎 to provide feedback.
| let mut current: Option<&(dyn StdError + 'static)> = Some(err); | ||
| while let Some(e) = current { | ||
| let message = e.to_string(); | ||
| let message = message.to_ascii_lowercase(); | ||
| if message.contains("no instances found for endpoint") | ||
| || message.contains("no instances in selected device group for endpoint") | ||
| || message.contains("connection refused") | ||
| || message.contains("connection reset by peer") | ||
| || message.contains("connection aborted") | ||
| || message.contains("worker disconnected before response stream was established") | ||
| { | ||
| return true; | ||
| } | ||
| current = e.source(); | ||
| } | ||
|
|
||
| false |
There was a problem hiding this comment.
🚩 String-based error matching in is_migratable is brittle
The new string-matching path in is_migratable (lib/llm/src/migration.rs:89-105) matches error messages by substring (e.g., "no instances found for endpoint", "connection refused", "connection reset by peer"). These patterns are inherently fragile — if the upstream library changes the error wording, migration will silently stop working for those error cases. The code already has a typed DynamoError path via error::match_error_chain; these string matches are a pragmatic workaround for errors that haven't been typed yet. Similarly is_failover_backend_cancel_message (lib/llm/src/migration.rs:108-120) matches "cancellederror", "backendcancelled", and "event loop is closed" by substring. Consider tracking these as tech debt to convert to typed errors.
Was this helpful? React with 👍 or 👎 to provide feedback.
| fn observed_failover_at() -> &'static Mutex<Option<Instant>> { | ||
| static OBSERVED_AT: OnceLock<Mutex<Option<Instant>>> = OnceLock::new(); | ||
| OBSERVED_AT.get_or_init(|| Mutex::new(None)) | ||
| } | ||
|
|
||
| fn mark_failover_observed() { | ||
| if migration_first_chunk_timeout().is_zero() && migration_recent_failover_window().is_zero() { | ||
| return; | ||
| } | ||
| if let Ok(mut observed_at) = observed_failover_at().lock() { | ||
| *observed_at = Some(Instant::now()); | ||
| } | ||
| } | ||
|
|
||
| pub(crate) fn observed_failover_elapsed() -> Option<Duration> { | ||
| let Ok(observed_at) = observed_failover_at().lock() else { | ||
| return None; | ||
| }; | ||
| observed_at.as_ref().map(Instant::elapsed) | ||
| } | ||
|
|
||
| fn recently_observed_failover() -> bool { | ||
| let window = migration_recent_failover_window(); | ||
| if window.is_zero() { | ||
| return false; | ||
| } | ||
| observed_failover_elapsed().is_some_and(|elapsed| elapsed <= window) | ||
| } |
There was a problem hiding this comment.
🚩 Global mutable failover timestamp shared across all request streams
The observed_failover_at() function at lib/llm/src/migration.rs:260 uses a process-global Mutex<Option<Instant>> to track the last failover observation. mark_failover_observed() is called from any RetryManager instance's next() method whenever a migratable or failover-cancel error is seen. This means a failover event from one model/endpoint influences the retry behavior (via recently_observed_failover()) of ALL concurrent requests to ALL models. In a multi-model frontend, a failover on model A could cause model B requests to enter the failover replay path unnecessarily. This is likely acceptable for single-model deployments but could cause unexpected retries in multi-tenant setups.
Was this helpful? React with 👍 or 👎 to provide feedback.
|
|
||
| fn request_timeout_from_value(value: Option<&str>) -> Duration { | ||
| value | ||
| .and_then(|raw| raw.parse::<u64>().ok()) |
There was a problem hiding this comment.
🚩 New NATS request timeout defaults to 2 seconds which may be aggressive
The new NATS ack timeout (lib/runtime/src/pipeline/network/egress/nats_client.rs:29, DEFAULT_NATS_REQUEST_TIMEOUT_MS = 2000) wraps the existing request_with_headers call with a 2-second tokio::time::timeout. The previous code used the async-nats default (10 seconds). Under heavy load or network latency, 2 seconds may be too short for the request-plane ack (which is just the routing acknowledgment, not the full generation). A timeout here produces a CannotConnect error which triggers migration retries — potentially amplifying load during congestion. The value is configurable via DYN_NATS_REQUEST_TIMEOUT_MS, but the 5x reduction in default may surprise existing deployments.
Was this helpful? React with 👍 or 👎 to provide feedback.
| let mut retry_manager = RetryManager::build( | ||
| ctx, | ||
| BTreeMap::new(), | ||
| request, | ||
| next_generate, | ||
| 1, | ||
| None, | ||
| Arc::new(TEST_MODEL.to_string()), | ||
| metrics.clone(), | ||
| ) | ||
| .await | ||
| .expect("Failed to build RetryManager"); |
There was a problem hiding this comment.
🚩 Test calls to RetryManager::build missing session_affinity argument would fail at compile time
Several new test functions (test_retry_manager_empty_first_stream_migration at line 1452, test_retry_manager_empty_stream_retries_after_context_stop at line 1497, test_retry_manager_does_not_replay_when_output_budget_exhausted at line 1606) call RetryManager::build with 8 positional arguments, but the method signature requires 9 (the 9th being session_affinity: Option<SessionAffinityId>). This would be caught by the Rust compiler and is not reported as a bug per review guidelines, but it indicates these tests may not have been compiled in CI yet.
Was this helpful? React with 👍 or 👎 to provide feedback.
hutm
force-pushed
the
review/gms-v2-latehost-06-runtime-failover-replay
branch
from
d52bf50 to
096c478
Compare
hutm
changed the title
|
CI follow-up: moved the three |
hutm
force-pushed
the
review/gms-v2-latehost-06-runtime-failover-replay
branch
from
096c478 to
7eda2c0
Compare
WalkthroughThe PR adds a Bulwark gateway endpoint mode to the LLM frontend (Python config, Rust dynamic backend discovery, readiness monitoring, HTTP failover waiting), overhauls migration/failover retry logic with concurrency controls, unifies discovery logical instance ID resolution across kube and kv-store backends, adds GMS KV placement metadata structs, hardens runtime transport (NATS timeout, empty-stream disconnect propagation, KV watch cleanup), adds a graceful-shutdown fast-exit path, and improves ChangesBulwark Gateway, Failover, and Runtime Reliability
ManagedProcess zombie-aware teardown
Estimated code review effort🎯 5 (Critical) | ⏱️ ~150 minutes 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 11
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/src/dynamo/frontend/main.py (1)
178-195: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winUse the saved
DYN_SYSTEM_PORTvalue for the warning.Line 179 removes the env var before this check, so Line 195 never warns for normal frontend mode. Check
frontend_system_portinstead.Proposed fix
- if os.environ.get("DYN_SYSTEM_PORT") and not config.bulwark_gateway_endpoint: + if frontend_system_port and not config.bulwark_gateway_endpoint:🤖 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 `@components/src/dynamo/frontend/main.py` around lines 178 - 195, The warning check in the frontend startup flow is using os.environ after DYN_SYSTEM_PORT has already been removed, so it will miss the normal-frontend warning path. In the main startup logic around frontend_system_port and parse_args, change the warning condition to read the saved frontend_system_port value instead of re-checking os.environ, while keeping the gateway-endpoint restore behavior unchanged.
🧹 Nitpick comments (1)
tests/utils/managed_process.py (1)
586-598: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNarrow the
terminate()exception handler. KeepProcessLookupError, but change the blanketexcept Exceptiontoexcept OSErrorso non-OS bugs still surface.🤖 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 `@tests/utils/managed_process.py` around lines 586 - 598, The exception handling around managed process termination is too broad in the `terminate()` path. Keep the existing `ProcessLookupError` handling in `ManagedProcess.terminate`, but narrow the generic `except Exception as e` block to `except OSError` so only OS-level failures are swallowed and unrelated bugs still fail visibly. Update the logger warning in that same `self.proc.terminate()` try/except block accordingly.Sources: Coding guidelines, Path instructions, Linters/SAST tools
🤖 Prompt for all review comments with 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.
Inline comments:
In `@components/src/dynamo/common/utils/tests/test_graceful_shutdown.py`:
- Around line 334-351: The shutdown state is leaking between tests because
graceful_shutdown_with_discovery() leaves the module-global
_gs._shutdown_started set, making later assertions in is_shutdown_in_progress()
order-dependent. Add a test fixture in this test module to reset/cleanup the
global shutdown flag before and after each test, and apply it to the
shutdown-related tests (including
test_shutdown_in_progress_reflects_active_shutdown and the other affected cases)
so each test starts with isolated _gs state.
In `@lib/llm/src/entrypoint/input/endpoint.rs`:
- Around line 417-464: The readiness monitor in the private backend watch path
currently exits permanently when the discovery stream ends, leaving the gateway
stuck NotReady after a transient disconnect. Update the loop around
`distributed_runtime.discovery().list_and_watch(...)`,
`discovery_stream.next()`, and `pending_notready_deadline` so it re-establishes
the watch with cancel-aware backoff instead of breaking on end-of-stream,
following the same reconnect pattern used by the existing runtime watcher logic.
Keep `set_gateway_health` transitions and `public_endpoint_id` logging intact
while ensuring the task stays alive and re-subscribes after failures.
In `@lib/llm/src/http/service/service_v2.rs`:
- Around line 126-141: The endpoint_enabled_or_wait logic in
service_v2::endpoint_enabled_or_wait is treating config-disabled routes as
eligible for failover waiting because it falls through to wait_for_failover when
state.flags.get(&endpoint_type) is false. Change the gating so statically
disabled endpoints return false immediately, and only invoke wait_for_failover
for endpoints that are explicitly in a failover/readiness state; use the
existing endpoint_type check and state.flags lookup to distinguish
config-disabled from transiently unavailable routes.
In `@lib/llm/src/local_model/runtime_config.rs`:
- Around line 441-464: The GMS runtime config accessors are swallowing
deserialization failures by turning them into None, which lets the setters
overwrite malformed runtime metadata with defaults. Update gms_runtime_config to
propagate errors instead of using ok().flatten(), and adjust
set_gms_placement_enabled, set_gms_control_enabled, and set_gms_daemon_socket to
use the fallible get_engine_specific path so corrupt GMS runtime_data fails fast
rather than being silently replaced.
In `@lib/llm/src/migration.rs`:
- Around line 846-849: The retry-exhausted early return in
`MigrationModel::handle_stream_output` and the matching path near the later
stream handler are skipping `stop_conditions.max_tokens` accounting, so
`output_budget_exhausted()` never sees the budget hit zero. Update both paths to
keep decrementing or otherwise finalize the output budget even when
`self.retries_left == 0`, while still preventing further retries, and ensure the
stream output bookkeeping stays active after the retry budget is exhausted.
In `@lib/runtime/src/config/environment_names.rs`:
- Around line 591-598: The duplicate-name coverage in
test_no_duplicate_env_var_names is missing the new discovery constants, so
update the test to include DYN_KUBE_DISCOVERY_DEBOUNCE_MS,
DYN_DISCOVERY_LOGICAL_INSTANCE_ID, and DYN_DISCOVERY_LOGICAL_INSTANCE_KEY
alongside the existing discovery env vars. Keep the change in the
environment_names test module and use the same symbol names so future duplicate
public constants are caught automatically.
In `@lib/runtime/src/discovery/mod.rs`:
- Around line 30-34: The discovery ID helper hash_logical_instance_key currently
uses DefaultHasher, which is not a stable contract for persisted or distributed
identities. Replace it with an explicit stable hashing approach inside
hash_logical_instance_key, and make sure the masking with
DISCOVERY_INSTANCE_ID_MASK still applies to the final value. Also add a
fixed-vector test for this helper in the discovery module so the expected ID
output is pinned and won’t drift across Rust versions.
In `@lib/runtime/src/pipeline/network/ingress/push_handler.rs`:
- Around line 230-235: The synthetic error response in the empty-stream path is
serialized with serde_json instead of the negotiated payload codec, so fix the
branch that builds NetworkStreamWrapper in push_handler to encode the error
frame using payload_codec the same way the normal and final frames do. Keep the
response shape the same, but route the U::from_err(err) wrapper through the
codec selected for the stream so non-JSON peers can decode the Disconnected
signal consistently.
In `@tests/utils/managed_process.py`:
- Around line 652-659: The SIGTERM/SIGKILL teardown logic in managed_process.py
is incorrectly treating a PGID as finished when only the snapshotted PIDs are
zombie/dead, which can miss live children forked later in the same group. Update
the PGID checks in the shutdown flow around the existing
_live_process_groups_for_pids usage so a group is skipped only when the process
group itself has no live members, not just because pgid_to_pids contains
zombie-only PIDs. Apply the same helper-based live-group check in both the
initial wait loop and the SIGKILL loop, using the existing sigtermed_pgids,
pgid_to_pids, and _live_process_groups_for_pids symbols.
In `@tests/utils/test_managed_process_teardown.py`:
- Around line 362-457: The teardown tests in TestParentOnlyFirstTeardown rely on
polling and sleep loops, so add an explicit pytest timeout marker to these new
test methods to prevent them from hanging indefinitely. Apply
`@pytest.mark.timeout`(...) consistently to both
test_parent_only_signal_lets_parent_cleanup_child and
test_parent_only_ignores_zombie_child_process_group, using the same style as
other timeout-covered tests in this module.
- Line 405: The two child_pid_file reads currently use open(...).read() without
managing the file handle; update both reads to use context managers so the file
is always closed properly. Locate the reads around child_pid and child_pid_file
and wrap each open call in a with block, keeping the existing int(...) parsing
and behavior unchanged.
---
Outside diff comments:
In `@components/src/dynamo/frontend/main.py`:
- Around line 178-195: The warning check in the frontend startup flow is using
os.environ after DYN_SYSTEM_PORT has already been removed, so it will miss the
normal-frontend warning path. In the main startup logic around
frontend_system_port and parse_args, change the warning condition to read the
saved frontend_system_port value instead of re-checking os.environ, while
keeping the gateway-endpoint restore behavior unchanged.
---
Nitpick comments:
In `@tests/utils/managed_process.py`:
- Around line 586-598: The exception handling around managed process termination
is too broad in the `terminate()` path. Keep the existing `ProcessLookupError`
handling in `ManagedProcess.terminate`, but narrow the generic `except Exception
as e` block to `except OSError` so only OS-level failures are swallowed and
unrelated bugs still fail visibly. Update the logger warning in that same
`self.proc.terminate()` try/except block accordingly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 62bb3112-a31c-4da4-8428-7f9a93741149
📒 Files selected for processing (26)
components/src/dynamo/common/utils/graceful_shutdown.pycomponents/src/dynamo/common/utils/otel_tracing.pycomponents/src/dynamo/common/utils/tests/test_graceful_shutdown.pycomponents/src/dynamo/frontend/frontend_args.pycomponents/src/dynamo/frontend/main.pylib/llm/src/entrypoint/input/endpoint.rslib/llm/src/http/service/openai.rslib/llm/src/http/service/service_v2.rslib/llm/src/local_model/runtime_config.rslib/llm/src/migration.rslib/llm/src/protocols/common/preprocessor.rslib/runtime/src/config/environment_names.rslib/runtime/src/discovery/kube.rslib/runtime/src/discovery/kube/daemon.rslib/runtime/src/discovery/kube/utils.rslib/runtime/src/discovery/kv_store.rslib/runtime/src/discovery/mod.rslib/runtime/src/distributed.rslib/runtime/src/pipeline/network/egress/addressed_router.rslib/runtime/src/pipeline/network/egress/nats_client.rslib/runtime/src/pipeline/network/egress/push_router.rslib/runtime/src/pipeline/network/ingress/push_handler.rslib/runtime/src/storage/kv.rslib/runtime/tests/bulwark_cutover.rstests/utils/managed_process.pytests/utils/test_managed_process_teardown.py
| def test_shutdown_in_progress_reflects_active_shutdown(): | ||
| """The shutdown-in-progress flag is visible before shutdown_event is set.""" | ||
| assert is_shutdown_in_progress() is False | ||
|
|
||
| async def _run(): | ||
| mock_runtime = MagicMock() | ||
| mock_endpoint = AsyncMock() | ||
| mock_endpoint.unregister_endpoint_instance = AsyncMock(return_value=None) | ||
| await graceful_shutdown_with_discovery( | ||
| runtime=mock_runtime, | ||
| endpoints=[mock_endpoint], | ||
| shutdown_event=None, | ||
| grace_period_s=0, | ||
| ) | ||
|
|
||
| asyncio.run(_run()) | ||
|
|
||
| assert is_shutdown_in_progress() is True |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Reset the shutdown flag between tests.
These tests leave _gs._shutdown_started set after calling graceful_shutdown_with_discovery(), so later tests can no-op and Line 336 can fail if an earlier test already initiated shutdown. Add a fixture to isolate this module-global state. As per path instructions, tests should avoid shared mutable module state; based on learnings, module-level mutable test state causes order-dependent failures.
Proposed fix
+@pytest.fixture(autouse=True)
+def _reset_shutdown_started(monkeypatch):
+ monkeypatch.setattr(_gs, "_shutdown_started", asyncio.Event())
+
+
def test_shutdown_in_progress_reflects_active_shutdown():Also applies to: 367-403
🤖 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 `@components/src/dynamo/common/utils/tests/test_graceful_shutdown.py` around
lines 334 - 351, The shutdown state is leaking between tests because
graceful_shutdown_with_discovery() leaves the module-global
_gs._shutdown_started set, making later assertions in is_shutdown_in_progress()
order-dependent. Add a test fixture in this test module to reset/cleanup the
global shutdown flag before and after each test, and apply it to the
shutdown-related tests (including
test_shutdown_in_progress_reflects_active_shutdown and the other affected cases)
so each test starts with isolated _gs state.
Sources: Path instructions, Learnings
| let discovery_stream = match distributed_runtime | ||
| .discovery() | ||
| .list_and_watch(DiscoveryQuery::AllModels, Some(cancel_token.clone())) | ||
| .await | ||
| { | ||
| Ok(stream) => stream, | ||
| Err(err) => { | ||
| tracing::error!( | ||
| %err, | ||
| public_endpoint = %public_endpoint_id, | ||
| "Bulwark gateway failed to start private backend readiness monitor" | ||
| ); | ||
| set_gateway_health(&health_runtime, HealthStatus::NotReady, 0); | ||
| return; | ||
| } | ||
| }; | ||
| tokio::pin!(discovery_stream); | ||
|
|
||
| let notready_grace = gateway_notready_grace(); | ||
| tracing::info!( | ||
| public_endpoint = %public_endpoint_id, | ||
| notready_grace_ms = notready_grace.as_millis(), | ||
| "Bulwark gateway readiness monitor configured" | ||
| ); | ||
|
|
||
| let mut active_backends: HashSet<DiscoveryInstanceId> = HashSet::new(); | ||
| let mut last_ready = false; | ||
| let mut pending_notready_deadline: Option<Instant> = None; | ||
| set_gateway_health(&health_runtime, HealthStatus::NotReady, 0); | ||
|
|
||
| loop { | ||
| let pending_notready = async { | ||
| match pending_notready_deadline { | ||
| Some(deadline) => sleep_until(deadline).await, | ||
| None => pending::<()>().await, | ||
| } | ||
| }; | ||
|
|
||
| tokio::select! { | ||
| event = discovery_stream.next() => { | ||
| let Some(event) = event else { | ||
| tracing::warn!( | ||
| public_endpoint = %public_endpoint_id, | ||
| "Bulwark gateway private backend readiness stream ended" | ||
| ); | ||
| set_gateway_health(&health_runtime, HealthStatus::NotReady, 0); | ||
| break; | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Reconnect the readiness watch instead of exiting permanently.
If list_and_watch succeeds once but the stream later ends, this task sets the gateway NotReady and breaks forever. A transient discovery-watch disconnect would keep the gateway unhealthy even after private backends reappear; wrap the watch in a reconnect loop with cancel-aware backoff, like the existing runtime watcher pattern.
🤖 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 `@lib/llm/src/entrypoint/input/endpoint.rs` around lines 417 - 464, The
readiness monitor in the private backend watch path currently exits permanently
when the discovery stream ends, leaving the gateway stuck NotReady after a
transient disconnect. Update the loop around
`distributed_runtime.discovery().list_and_watch(...)`,
`discovery_stream.next()`, and `pending_notready_deadline` so it re-establishes
the watch with cancel-aware backoff instead of breaking on end-of-stream,
following the same reconnect pattern used by the existing runtime watcher logic.
Keep `set_gateway_health` transitions and `public_endpoint_id` logging intact
while ensuring the task stays alive and re-subscribes after failures.
| async fn endpoint_enabled_or_wait(state: &State, endpoint_type: EndpointType) -> bool { | ||
| if state.flags.get(&endpoint_type) { | ||
| return true; | ||
| } | ||
| if !should_wait_for_endpoint_failover(endpoint_type) { | ||
| return false; | ||
| } | ||
|
|
||
| wait_for_failover( | ||
| "endpoint", | ||
| endpoint_type.as_str(), | ||
| || state.flags.get(&endpoint_type).then_some(()).ok_or(()), | ||
| |_| true, | ||
| ) | ||
| .await | ||
| .is_ok() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Do not wait on statically disabled endpoint flags.
state.flags is initialized from config in build(), so a deliberately disabled Chat/Completion/Responses route now waits up to DYN_HTTP_MODEL_FAILOVER_WAIT_MS before returning 404. Gate waiting on an explicit failover/readiness signal, or keep config-disabled endpoints as immediate 404s.
Also applies to: 1282-1282
🤖 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 `@lib/llm/src/http/service/service_v2.rs` around lines 126 - 141, The
endpoint_enabled_or_wait logic in service_v2::endpoint_enabled_or_wait is
treating config-disabled routes as eligible for failover waiting because it
falls through to wait_for_failover when state.flags.get(&endpoint_type) is
false. Change the gating so statically disabled endpoints return false
immediately, and only invoke wait_for_failover for endpoints that are explicitly
in a failover/readiness state; use the existing endpoint_type check and
state.flags lookup to distinguish config-disabled from transiently unavailable
routes.
| pub fn set_gms_placement_enabled(&mut self) -> anyhow::Result<()> { | ||
| self.set_engine_specific(GMS_RUNTIME_CONFIG_KEY, { | ||
| let mut config = self.gms_runtime_config().unwrap_or_default(); | ||
| config.control_enabled = true; | ||
| config | ||
| }) | ||
| } | ||
|
|
||
| pub fn set_gms_control_enabled(&mut self) -> anyhow::Result<()> { | ||
| self.set_gms_placement_enabled() | ||
| } | ||
|
|
||
| pub fn set_gms_daemon_socket(&mut self, daemon_socket: String) -> anyhow::Result<()> { | ||
| let mut config = self.gms_runtime_config().unwrap_or_default(); | ||
| config.control_enabled = true; | ||
| config.daemon_socket = Some(daemon_socket); | ||
| self.set_engine_specific(GMS_RUNTIME_CONFIG_KEY, config) | ||
| } | ||
|
|
||
| pub fn gms_runtime_config(&self) -> Option<GmsRuntimeConfig> { | ||
| self.get_engine_specific(GMS_RUNTIME_CONFIG_KEY) | ||
| .ok() | ||
| .flatten() | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Propagate malformed GMS runtime metadata instead of treating it as absent.
gms_runtime_config().ok().flatten() hides deserialization errors, and the setters then overwrite bad runtime_data["gms"] with defaults. Return anyhow::Result<Option<GmsRuntimeConfig>> or use get_engine_specific(...)? in setters so corrupt runtime metadata fails fast.
🤖 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 `@lib/llm/src/local_model/runtime_config.rs` around lines 441 - 464, The GMS
runtime config accessors are swallowing deserialization failures by turning them
into None, which lets the setters overwrite malformed runtime metadata with
defaults. Update gms_runtime_config to propagate errors instead of using
ok().flatten(), and adjust set_gms_placement_enabled, set_gms_control_enabled,
and set_gms_daemon_socket to use the fallible get_engine_specific path so
corrupt GMS runtime_data fails fast rather than being silently replaced.
| self.current_stream_has_output = true; | ||
| if self.retries_left == 0 { | ||
| return; | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Keep output-budget accounting active after retry budget is exhausted.
Line 847 returns before decrementing stop_conditions.max_tokens, but output_budget_exhausted() relies on that field. With migration_limit = 0 or on the final attempt, a backend can emit all requested tokens and then send a disconnect error, and this path will still report the error because the budget never reaches Some(0).
Proposed adjustment
self.current_stream_has_output = true;
- if self.retries_left == 0 {
- return;
- }
+ let token_ids = llm_engine_output.token_ids();
+ if let Some(max_tokens) = self.request.stop_conditions.max_tokens {
+ self.request.stop_conditions.max_tokens =
+ Some(max_tokens.saturating_sub(token_ids.len() as u32));
+ }
+ if self.retries_left == 0 {
+ return;
+ }
// Capture the worker's engine.generate span pointer so a future
// migration retry can render an OTel Link back to it. The adapter
// stamps this on the first non-empty chunk; subsequent chunks may
// also carry it. Keep the most-recently-seen value.
if let Some(link) = llm_engine_output.worker_trace_link() {
self.last_worker_link = Some(link.clone());
}
- let token_ids = llm_engine_output.token_ids();
if self.exceed_max_seq_len(token_ids.len() as u32) {
return;
}
- if let Some(max_tokens) = self.request.stop_conditions.max_tokens {
- self.request.stop_conditions.max_tokens =
- Some(max_tokens.saturating_sub(token_ids.len() as u32));
- }Also applies to: 870-872
🤖 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 `@lib/llm/src/migration.rs` around lines 846 - 849, The retry-exhausted early
return in `MigrationModel::handle_stream_output` and the matching path near the
later stream handler are skipping `stop_conditions.max_tokens` accounting, so
`output_budget_exhausted()` never sees the budget hit zero. Update both paths to
keep decrementing or otherwise finalize the output budget even when
`self.retries_left == 0`, while still preventing further retries, and ensure the
stream output bookkeeping stays active after the retry budget is exhausted.
| pub(crate) fn hash_logical_instance_key(key: &str) -> u64 { | ||
| let mut hasher = DefaultHasher::new(); | ||
| key.hash(&mut hasher); | ||
| hasher.finish() & DISCOVERY_INSTANCE_ID_MASK | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Verify whether discovery instance IDs have fixed-vector tests and locate all DefaultHasher identity uses.
rg -n -C3 'DefaultHasher|hash_logical_instance_key|DISCOVERY_INSTANCE_ID_MASK' lib/runtime/srcRepository: ai-dynamo/dynamo
Length of output: 4090
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the discovery module and inspect the hashing/tests around it.
wc -l lib/runtime/src/discovery/mod.rs lib/runtime/src/discovery/kube/utils.rs
ast-grep outline lib/runtime/src/discovery/mod.rs --view expanded
printf '\n--- tests around hash_logical_instance_key ---\n'
sed -n '930,1035p' lib/runtime/src/discovery/mod.rs
printf '\n--- kube utils around instance_id/hash_pod_name ---\n'
sed -n '50,90p' lib/runtime/src/discovery/kube/utils.rs
printf '\n--- all DefaultHasher uses in runtime discovery ---\n'
rg -n -C2 'DefaultHasher|hash_logical_instance_key|DISCOVERY_INSTANCE_ID_MASK' lib/runtime/src/discoveryRepository: ai-dynamo/dynamo
Length of output: 11763
Use a stable hash for discovery IDs.
DefaultHasher is not a stable contract for persisted/distributed identities, and the current test only checks this helper against itself. Switch to an explicit stable hash and add a fixed-vector test so the ID format doesn’t drift across Rust versions.
🤖 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 `@lib/runtime/src/discovery/mod.rs` around lines 30 - 34, The discovery ID
helper hash_logical_instance_key currently uses DefaultHasher, which is not a
stable contract for persisted or distributed identities. Replace it with an
explicit stable hashing approach inside hash_logical_instance_key, and make sure
the masking with DISCOVERY_INSTANCE_ID_MASK still applies to the final value.
Also add a fixed-vector test for this helper in the discovery module so the
expected ID output is pinned and won’t drift across Rust versions.
| let resp_wrapper = NetworkStreamWrapper { | ||
| data: Some(U::from_err(err)), | ||
| complete_final: false, | ||
| }; | ||
| let resp_bytes = serde_json::to_vec(&resp_wrapper) | ||
| .expect("fatal error: invalid response object - this should never happen"); |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Encode the synthetic error with the negotiated payload codec.
Line 234 bypasses payload_codec, unlike the normal and final frames. If this path uses a non-JSON codec, the peer may fail to decode the empty-stream error and miss the intended Disconnected signal.
Proposed fix
- let resp_bytes = serde_json::to_vec(&resp_wrapper)
- .expect("fatal error: invalid response object - this should never happen");
+ let resp_bytes = payload_codec
+ .encode(&resp_wrapper)
+ .expect("fatal error: invalid request-plane empty-stream error object");📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| let resp_wrapper = NetworkStreamWrapper { | |
| data: Some(U::from_err(err)), | |
| complete_final: false, | |
| }; | |
| let resp_bytes = serde_json::to_vec(&resp_wrapper) | |
| .expect("fatal error: invalid response object - this should never happen"); | |
| let resp_wrapper = NetworkStreamWrapper { | |
| data: Some(U::from_err(err)), | |
| complete_final: false, | |
| }; | |
| let resp_bytes = payload_codec | |
| .encode(&resp_wrapper) | |
| .expect("fatal error: invalid request-plane empty-stream error object"); |
🤖 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 `@lib/runtime/src/pipeline/network/ingress/push_handler.rs` around lines 230 -
235, The synthetic error response in the empty-stream path is serialized with
serde_json instead of the negotiated payload codec, so fix the branch that
builds NetworkStreamWrapper in push_handler to encode the error frame using
payload_codec the same way the normal and final frames do. Keep the response
shape the same, but route the U::from_err(err) wrapper through the codec
selected for the stream so non-JSON peers can decode the Disconnected signal
consistently.
| # Check if any signaled group is still alive. Zombie-only | ||
| # child groups (common with MPI launch helpers such as orted) do not | ||
| # hold resources and cannot be killed, so don't burn the timeout. | ||
| still_alive = False | ||
| for pgid in sigtermed_pgids: | ||
| known_pids = pgid_to_pids.get(pgid, set()) | ||
| if known_pids and not self._live_process_groups_for_pids(known_pids): | ||
| continue |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not skip a PGID solely because snapshotted PIDs are zombie-only.
Lines 657-659 and 675-677 treat known zombie/dead PIDs as proof the process group has no live members. If a snapshotted child forks a same-PGID process during shutdown, that live process is unknown to pgid_to_pids, so teardown can skip both waiting and SIGKILL, leaking the child process.
Proposed direction
+ `@staticmethod`
+ def _process_group_has_live_members(pgid: int, known_pids: set[int]) -> bool:
+ if pgid in ManagedProcess._live_process_groups_for_pids(known_pids):
+ return True
+ try:
+ os.killpg(pgid, 0)
+ except (ProcessLookupError, OSError):
+ return False
+
+ # `killpg(..., 0)` can be true for zombie-only groups. Scan before
+ # deciding the group is safe to skip.
+ for process in psutil.process_iter(["status"]):
+ try:
+ if os.getpgid(process.pid) != pgid:
+ continue
+ if process.status() != psutil.STATUS_ZOMBIE:
+ return True
+ except (ProcessLookupError, psutil.NoSuchProcess, OSError):
+ continue
+ except psutil.AccessDenied:
+ return True
+ return False
+
for pgid in sigtermed_pgids:
- known_pids = pgid_to_pids.get(pgid, set())
- if known_pids and not self._live_process_groups_for_pids(known_pids):
+ if not self._process_group_has_live_members(
+ pgid, pgid_to_pids.get(pgid, set())
+ ):
continueApply the same helper in the SIGKILL loop.
Also applies to: 675-677
🤖 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 `@tests/utils/managed_process.py` around lines 652 - 659, The SIGTERM/SIGKILL
teardown logic in managed_process.py is incorrectly treating a PGID as finished
when only the snapshotted PIDs are zombie/dead, which can miss live children
forked later in the same group. Update the PGID checks in the shutdown flow
around the existing _live_process_groups_for_pids usage so a group is skipped
only when the process group itself has no live members, not just because
pgid_to_pids contains zombie-only PIDs. Apply the same helper-based live-group
check in both the initial wait loop and the SIGKILL loop, using the existing
sigtermed_pgids, pgid_to_pids, and _live_process_groups_for_pids symbols.
| class TestParentOnlyFirstTeardown: | ||
| def test_parent_only_signal_lets_parent_cleanup_child(self, tmp_path): | ||
| parent_marker = str(tmp_path / "parent_sigterm") | ||
| child_marker = str(tmp_path / "child_cleaned") | ||
| ready_file = str(tmp_path / "ready") | ||
| child_pid_file = str(tmp_path / "child_pid") | ||
| script = "\n".join( | ||
| [ | ||
| "import pathlib, signal, subprocess, sys, time", | ||
| f"parent_marker = pathlib.Path({parent_marker!r})", | ||
| f"child_marker = pathlib.Path({child_marker!r})", | ||
| f"ready = pathlib.Path({ready_file!r})", | ||
| f"child_pid_file = pathlib.Path({child_pid_file!r})", | ||
| "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(300)'])", | ||
| "child_pid_file.write_text(str(child.pid))", | ||
| "def on_term(*_):", | ||
| " parent_marker.touch()", | ||
| " child.terminate()", | ||
| " child.wait(timeout=5)", | ||
| " child_marker.touch()", | ||
| " raise SystemExit(0)", | ||
| "signal.signal(signal.SIGTERM, on_term)", | ||
| "ready.touch()", | ||
| "while True:", | ||
| " time.sleep(0.1)", | ||
| ] | ||
| ) | ||
| mp = ManagedProcess( | ||
| command=["python3", "-c", script], | ||
| timeout=10, | ||
| display_output=False, | ||
| terminate_all_matching_process_names=False, | ||
| terminate_parent_only_first=True, | ||
| graceful_shutdown_timeout=5.0, | ||
| log_dir=str(tmp_path), | ||
| ) | ||
|
|
||
| child_pid = None | ||
| with mp: | ||
| deadline = time.monotonic() + 5.0 | ||
| while time.monotonic() < deadline and not os.path.exists(ready_file): | ||
| time.sleep(0.05) | ||
| assert os.path.exists(ready_file) | ||
| child_pid = int(open(child_pid_file, encoding="utf-8").read()) | ||
| assert _pid_alive(child_pid) | ||
|
|
||
| assert os.path.exists(parent_marker) | ||
| assert os.path.exists(child_marker) | ||
| assert child_pid is not None | ||
| assert _wait_for_pid_death(child_pid, timeout=5) | ||
|
|
||
| def test_parent_only_ignores_zombie_child_process_group(self, tmp_path, caplog): | ||
| ready_file = str(tmp_path / "ready") | ||
| child_pid_file = str(tmp_path / "child_pid") | ||
| script = "\n".join( | ||
| [ | ||
| "import os, pathlib, time", | ||
| f"ready = pathlib.Path({ready_file!r})", | ||
| f"child_pid_file = pathlib.Path({child_pid_file!r})", | ||
| "pid = os.fork()", | ||
| "if pid == 0:", | ||
| " os.setpgid(0, 0)", | ||
| " os._exit(0)", | ||
| "child_pid_file.write_text(str(pid))", | ||
| "ready.touch()", | ||
| "while True:", | ||
| " time.sleep(0.1)", | ||
| ] | ||
| ) | ||
| mp = ManagedProcess( | ||
| command=["python3", "-c", script], | ||
| timeout=10, | ||
| display_output=False, | ||
| terminate_all_matching_process_names=False, | ||
| terminate_parent_only_first=True, | ||
| graceful_shutdown_timeout=0.5, | ||
| log_dir=str(tmp_path), | ||
| ) | ||
|
|
||
| with caplog.at_level(logging.WARNING, logger="ManagedProcess"): | ||
| with mp: | ||
| deadline = time.monotonic() + 5.0 | ||
| while time.monotonic() < deadline and not os.path.exists(ready_file): | ||
| time.sleep(0.05) | ||
| assert os.path.exists(ready_file) | ||
| child_pid = int(open(child_pid_file, encoding="utf-8").read()) | ||
| deadline = time.monotonic() + 5.0 | ||
| while time.monotonic() < deadline: | ||
| try: | ||
| if psutil.Process(child_pid).status() == psutil.STATUS_ZOMBIE: | ||
| break | ||
| except psutil.NoSuchProcess: | ||
| break | ||
| time.sleep(0.05) | ||
|
|
||
| assert "child process groups are still live" not in caplog.text |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
sed -n '1,90p' tests/utils/test_managed_process_teardown.py
rg -n 'pytestmark|mark\.(timeout|pre_merge|post_merge|nightly|weekly|release|unit|integration|e2e|gpu_[0-9])|time\.sleep|def test_parent_only' tests/utils/test_managed_process_teardown.pyRepository: ai-dynamo/dynamo
Length of output: 3751
Add an explicit timeout to these teardown tests. The module-level pytestmark already covers pre_merge, unit, and gpu_0; the remaining gap is @pytest.mark.timeout(...) for these polling/sleep-heavy tests.
🧰 Tools
🪛 ast-grep (0.44.0)
[warning] 404-404: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(child_pid_file, encoding="utf-8")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(open-filename-from-request)
[warning] 446-446: File path is request-/variable-derived; validate and normalize to prevent path traversal.
Context: open(child_pid_file, encoding="utf-8")
Note: [CWE-22] Improper Limitation of a Pathname to a Restricted Directory ('Path Traversal').
(open-filename-from-request)
🤖 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 `@tests/utils/test_managed_process_teardown.py` around lines 362 - 457, The
teardown tests in TestParentOnlyFirstTeardown rely on polling and sleep loops,
so add an explicit pytest timeout marker to these new test methods to prevent
them from hanging indefinitely. Apply `@pytest.mark.timeout`(...) consistently to
both test_parent_only_signal_lets_parent_cleanup_child and
test_parent_only_ignores_zombie_child_process_group, using the same style as
other timeout-covered tests in this module.
Sources: Coding guidelines, Path instructions, Learnings
| while time.monotonic() < deadline and not os.path.exists(ready_file): | ||
| time.sleep(0.05) | ||
| assert os.path.exists(ready_file) | ||
| child_pid = int(open(child_pid_file, encoding="utf-8").read()) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
file="tests/utils/test_managed_process_teardown.py"
wc -l "$file"
sed -n '380,460p' "$file" | cat -nRepository: ai-dynamo/dynamo
Length of output: 3817
Wrap both reads in with blocks.
open(...).read() leaves the file handle unmanaged; use a context manager for the two child_pid_file reads.
🤖 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 `@tests/utils/test_managed_process_teardown.py` at line 405, The two
child_pid_file reads currently use open(...).read() without managing the file
handle; update both reads to use context managers so the file is always closed
properly. Locate the reads around child_pid and child_pid_file and wrap each
open call in a with block, keeping the existing int(...) parsing and behavior
unchanged.
Sources: Coding guidelines, Path instructions
Signed-off-by: mkhadkevich <mkhadkevich@nvidia.com>
hutm
force-pushed
the
review/gms-v2-latehost-06-runtime-failover-replay
branch
from
7eda2c0 to
faac3c6
Compare
1 participant
Summary
Make in-flight work replayable when an inference worker fails, so a replacement worker can resume service without coupling recovery to engine-owned KV memory.
Scope
Adds the runtime replay and failure-notification path. It does not introduce GMS allocation or engine-specific integration.
Stack
Position 1/18 in the GMS-managed KV review train. This PR is based on
main.Parent: #10450
Tracking: #10454
Review migration
This is the current review entry replacing historical merged PR #10470. GitHub does not allow a merged PR to be retargeted; #10470 and all of its comments and reviews remain intact as the historical record.
The train is rebased on
mainataeda23b483831df2f75b697b8ccf65f4ffd9f3d6. The reordered functionality is preserved while each PR boundary is independently buildable and lintable: compatibility call sites and the engine-facing placement adapter now appear at first use; missing type imports and test markers were added; repository-pinned formatting was applied; one unused constant was removed; and every commit carries a DCO sign-off. The complete range passesgit diff --checkand the full pre-commit suite.Validation
The Bulwark boundary was validated with the Dynamo Go unit tests and command build after positions 1-3 were assembled.
Summary by CodeRabbit
New Features
Bug Fixes