diff --git a/src/k1link/device_plugins/xgrids_k1/facade.py b/src/k1link/device_plugins/xgrids_k1/facade.py index c47d8c8..5346674 100644 --- a/src/k1link/device_plugins/xgrids_k1/facade.py +++ b/src/k1link/device_plugins/xgrids_k1/facade.py @@ -3188,6 +3188,60 @@ class XgridsK1CompatibilityService: "маршрут K1 изменился во время контрольной проверки" ) supervisor = self._connection_supervisor.snapshot() + if ( + path.available + and path.route_class == "direct" + and observed_epoch == binding.host_path_epoch + and supervisor.host_path.epoch == binding.host_path_epoch + and supervisor.endpoint.target == target + and supervisor.endpoint.intent_id == binding.intent_id + and supervisor.endpoint.host_path_epoch == binding.host_path_epoch + and supervisor.endpoint.reason_code == "endpoint-observation-stale" + ): + # Project preparation can legitimately outlive the supervisor's + # TCP silence lease while the exact MQTT control proof remains + # fresh. Refresh only that expired read-only endpoint fact before + # the physical START admission. This is not a command retry: no + # operation or physical ledger row exists yet. A second route + # sample binds the TCP result to the same host/association epoch; + # any changed path or unreachable broker still fails closed. + endpoint = _probe_control_endpoint_socket(binding.target_ipv4) + final_path = self._sample_host_path( + binding.target_ipv4, + association_timeout_seconds=COMMAND_BOUND_ASSOCIATION_TIMEOUT_SECONDS, + ) + final_epoch = self._connection_supervisor.observe_host_path_if_current( + expected_intent_id=binding.intent_id, + expected_target=target, + expected_host_path_epoch=binding.host_path_epoch, + result=final_path, + ) + path_unchanged = bool( + final_epoch == binding.host_path_epoch + and final_path.available + and final_path.route_class == "direct" + and final_path.fingerprint == path.fingerprint + and final_path.interface == path.interface + and final_path.source_ipv4 == path.source_ipv4 + and final_path.kernel_route_fingerprint == path.kernel_route_fingerprint + ) + if final_epoch == binding.host_path_epoch: + self._connection_supervisor.observe_endpoint_if_current( + target=target, + intent_id=binding.intent_id, + host_path_epoch=binding.host_path_epoch, + reachable=endpoint.reachable and path_unchanged, + reason_code=( + None + if endpoint.reachable and path_unchanged + else endpoint.reason_code + if not endpoint.reachable + else "host-path-changed-during-tcp-probe" + ), + ) + path = final_path + observed_epoch = final_epoch + supervisor = self._connection_supervisor.snapshot() path_is_current = bool( not supervisor.closed and path.available diff --git a/tests/test_xgrids_acquisition_lifecycle.py b/tests/test_xgrids_acquisition_lifecycle.py index 3102866..02f0535 100644 --- a/tests/test_xgrids_acquisition_lifecycle.py +++ b/tests/test_xgrids_acquisition_lifecycle.py @@ -13169,6 +13169,81 @@ def test_start_stale_endpoint_fails_before_receiver_or_device_checkpoint( assert control.start_projects == [] +def test_start_refreshes_ttl_expired_endpoint_after_project_prepare( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + service, runtime = service_with_fake_runtime(tmp_path) + monotonic_now = [100.0] + suspend_aware_now = [1_000.0] + supervisor = service._connection_supervisor # noqa: SLF001 + supervisor._monotonic_clock = lambda: monotonic_now[0] # noqa: SLF001 + supervisor._suspend_aware_clock = lambda: suspend_aware_now[0] # noqa: SLF001 + supervisor._observation_ttl_seconds = 15.0 # noqa: SLF001 + control, binding, ready = _install_binding_validating_ready_control(service) + stable_path = _direct_host_path(binding.target_ipv4) + monkeypatch.setattr(service, "_sample_host_path", lambda *_args, **_kwargs: stable_path) + workspace = service.enter_application_workspace( + EnterApplicationWorkspaceRequest( + operator_confirmed=True, + expected_session_generation=ready["application_control_session"][ + "session_generation" + ], + expected_state_revision=ready["application_control_session"]["state_revision"], + ) + ) + prepared = service.prepare_acquisition( + _prepare_request( + project_name=PROJECT_NAME, + host=binding.target_ipv4, + compatibility_attestation=ATTESTATION, + expected_control_session_generation=workspace["application_control_session"][ + "session_generation" + ], + expected_control_state_revision=workspace["application_control_session"][ + "state_revision" + ], + ) + ) + monotonic_now[0] += 15.01 + suspend_aware_now[0] += 15.01 + assert control.verified_control is not None + control.verified_control = { + **control.verified_control, + "control_proof_revision": 3, + } + tcp_samples: list[str] = [] + monkeypatch.setattr( + facade_module, + "_probe_control_endpoint_socket", + lambda target: ( + tcp_samples.append(target) + or facade_module.TcpReachabilityProbeResult(reachable=True) + ), + ) + + started = service.start_acquisition( + _start_request( + acquisition_id=prepared["acquisition"]["acquisition_id"], + physical_acceptance=PHYSICAL_ACCEPTANCE, + expected_control_session_generation=prepared["application_control_session"][ + "session_generation" + ], + expected_control_state_revision=prepared["application_control_session"][ + "state_revision" + ], + ) + ) + + assert started["acquisition"]["state"] == "starting" + assert tcp_samples == [binding.target_ipv4] + assert supervisor.snapshot().host_path.epoch == binding.host_path_epoch + assert supervisor.snapshot().endpoint.tcp_state == "reachable" + assert supervisor.snapshot().authority.acquisition_start_allowed is True + assert len(runtime.start_calls) == 1 + assert control.start_projects == [PROJECT_NAME] + + def test_start_stale_control_proof_fails_before_receiver_or_device_checkpoint( monkeypatch: pytest.MonkeyPatch, tmp_path: Path,