fix(k1): refresh stale endpoint before start

2026-08-14 16:27:03 +03:00
parent 9ef115d3b4
commit 9e6b5b403a
2 changed files with 129 additions and 0 deletions
@@ -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
@@ -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,