(
policy: &SandboxPolicy,
proxy_bind_ip: Option<IpAddr>,
opa_engine: Option<&Arc<OpaEngine>>,
retained_proto: Option<&ProtoSandboxPolicy>,
entrypoint_pid: Arc<AtomicU32>,
proce
| 73 | /// fails, or if the proxy server fails to start. |
| 74 | #[allow(clippy::too_many_arguments)] |
| 75 | pub async fn run_networking( |
| 76 | policy: &SandboxPolicy, |
| 77 | proxy_bind_ip: Option<IpAddr>, |
| 78 | opa_engine: Option<&Arc<OpaEngine>>, |
| 79 | retained_proto: Option<&ProtoSandboxPolicy>, |
| 80 | entrypoint_pid: Arc<AtomicU32>, |
| 81 | process_enabled: bool, |
| 82 | provider_credentials: &ProviderCredentialState, |
| 83 | sandbox_id: Option<&str>, |
| 84 | sandbox_name: Option<&str>, |
| 85 | openshell_endpoint: Option<&str>, |
| 86 | inference_routes: Option<&str>, |
| 87 | denial_tx: Option<UnboundedSender<DenialEvent>>, |
| 88 | activity_tx: Option<ActivitySender>, |
| 89 | ) -> Result<Networking> { |
| 90 | // Build the policy-local route context. The orchestrator's policy poll |
| 91 | // loop also holds an `Arc` clone (via `Networking::policy_local_ctx`) so |
| 92 | // it can publish updated policy snapshots after a successful reload. |
| 93 | let policy_local_ctx = Arc::new(PolicyLocalContext::new( |
| 94 | retained_proto.cloned(), |
| 95 | openshell_endpoint.map(str::to_string), |
| 96 | sandbox_name |
| 97 | .map(str::to_string) |
| 98 | .or_else(|| sandbox_id.map(str::to_string)), |
| 99 | )); |
| 100 | |
| 101 | // Readiness signal for the proxy accept loop: the proxy binds the TCP |
| 102 | // listener immediately (so the OS backlog queues early SYN packets) but |
| 103 | // defers `accept()` until symlink resolution completes. This eliminates |
| 104 | // the race where an in-flight request observes a generation transition |
| 105 | // during the OPA engine reload. |
| 106 | let (engine_ready_tx, engine_ready_rx) = tokio::sync::watch::channel(false); |
| 107 | |
| 108 | // Spawn a task to resolve policy binary symlinks once the workload's mount |
| 109 | // namespace becomes accessible via /proc/<pid>/root/. The task starts |
| 110 | // before run_process spawns the child, so first wait for the orchestrator |
| 111 | // to publish a non-zero PID, then poll for proc-root readiness. |
| 112 | if let (Some(engine), Some(proto)) = (opa_engine, retained_proto) { |
| 113 | if process_enabled { |
| 114 | let resolve_engine = engine.clone(); |
| 115 | let resolve_proto = proto.clone(); |
| 116 | let resolve_pid = entrypoint_pid.clone(); |
| 117 | tokio::spawn(async move { |
| 118 | // Phase 1: wait for run_process to publish the entrypoint PID. |
| 119 | // 20 attempts * 250ms = 5s window. |
| 120 | let mut pid = 0; |
| 121 | for attempt in 1..=20 { |
| 122 | pid = resolve_pid.load(Ordering::Acquire); |
| 123 | if pid != 0 { |
| 124 | break; |
| 125 | } |
| 126 | debug!( |
| 127 | attempt, |
| 128 | "Entrypoint PID not yet published, waiting before symlink resolution" |
| 129 | ); |
| 130 | tokio::time::sleep(Duration::from_millis(250)).await; |
| 131 | } |
| 132 | if pid == 0 { |
no test coverage detected