(
endpoint: String,
sandbox_id: String,
mut rx: mpsc::Receiver<SandboxLogLine>,
)
| 111 | const INITIAL_BACKOFF: tokio::time::Duration = tokio::time::Duration::from_secs(1); |
| 112 | |
| 113 | async fn run_push_loop( |
| 114 | endpoint: String, |
| 115 | sandbox_id: String, |
| 116 | mut rx: mpsc::Receiver<SandboxLogLine>, |
| 117 | ) { |
| 118 | let mut batch = Vec::with_capacity(50); |
| 119 | let mut backoff = INITIAL_BACKOFF; |
| 120 | let mut attempt: u64 = 0; |
| 121 | |
| 122 | // Outer reconnect loop — runs for the entire sandbox lifetime. |
| 123 | loop { |
| 124 | attempt += 1; |
| 125 | |
| 126 | // --- Connect --- |
| 127 | let client = match CachedOpenShellClient::connect(&endpoint).await { |
| 128 | Ok(c) => { |
| 129 | if attempt > 1 { |
| 130 | eprintln!("openshell: log push reconnected (attempt {attempt})"); |
| 131 | } |
| 132 | backoff = INITIAL_BACKOFF; |
| 133 | c |
| 134 | } |
| 135 | Err(e) => { |
| 136 | eprintln!("openshell: log push connect failed: {e}"); |
| 137 | // Drain the channel during backoff so the tracing layer doesn't |
| 138 | // block, but discard lines we can't deliver. |
| 139 | drain_during_backoff(&mut rx, &mut batch, backoff).await; |
| 140 | backoff = (backoff * 2).min(MAX_BACKOFF); |
| 141 | continue; |
| 142 | } |
| 143 | }; |
| 144 | |
| 145 | // --- Open the client-streaming RPC --- |
| 146 | let (push_tx, push_rx) = mpsc::channel::<PushSandboxLogsRequest>(32); |
| 147 | let stream = tokio_stream::wrappers::ReceiverStream::new(push_rx); |
| 148 | |
| 149 | // Spawn the gRPC streaming call. When the call ends (success or error), |
| 150 | // `rpc_done_tx` fires so the batch loop below knows whether to retry. |
| 151 | let (rpc_done_tx, mut rpc_done_rx) = mpsc::channel::<bool>(1); |
| 152 | tokio::spawn({ |
| 153 | let mut nav_client = client.raw_client(); |
| 154 | async move { |
| 155 | let fatal_auth = match nav_client.push_sandbox_logs(stream).await { |
| 156 | Ok(_) => false, |
| 157 | Err(e) => { |
| 158 | let fatal_auth = e.code() == tonic::Code::Unauthenticated; |
| 159 | eprintln!("openshell: log push RPC failed: {e}"); |
| 160 | fatal_auth |
| 161 | } |
| 162 | }; |
| 163 | let _ = rpc_done_tx.send(fatal_auth).await; |
| 164 | } |
| 165 | }); |
| 166 | |
| 167 | // --- Flush any lines buffered during reconnect --- |
| 168 | if !batch.is_empty() { |
| 169 | let lines = std::mem::take(&mut batch); |
| 170 | if push_tx |
no test coverage detected