fix(box): make CRI ReopenContainerLog synchronous (wait for supervisor reopen)

ReopenContainerLog fired a fire-and-forget Notify and returned immediately,
while the exit supervisor reopened the log writer asynchronously — so output
written after the RPC returned could still land in the old (rotated) file,
making the "reopening container log" conformance spec flaky.

The per-container reopen handle now carries a `done` Notify alongside the
`request`: the RPC signals the request and waits (bounded 5s) for the
supervisor to ack once `CriLogWriter::reopen` has completed, so the new file
is guaranteed in place before the RPC returns. This is the correct CRI
semantic (the call is synchronous).

This substantially reduces the flake but does not fully eliminate it: a
residual race remains in the supervisor's `select!` ordering (pending pipe
output read after the reopen can land in the new file). Tracked as a
follow-up; needs the supervisor to drain buffered output to the old writer
before reopening.

Verified: build + clippy clean, cri tests pass (reopen unit test updated for
the new handle), reopen-log critest now passes ~2/3 (was ~1/2).

Author: Roy Lin

src/cri/src/runtime_service/mod.rs

@@ -71,7 +71,16 @@ type WorkloadStdinSender = StreamingInput;
 type WorkloadStdinMap = Arc<RwLock<HashMap<String, WorkloadStdinSender>>>;
 type WorkloadStopSender = oneshot::Sender<()>;
 type WorkloadStopMap = Arc<RwLock<HashMap<String, WorkloadStopSender>>>;
-type LogReopenMap = Arc<RwLock<HashMap<String, Arc<Notify>>>>;
+/// Per-container handle for CRI ReopenContainerLog. `request` wakes the exit
+/// supervisor to reopen its log writer; `done` is notified once the reopen has
+/// completed, so ReopenContainerLog returns only after the new file is in place
+/// (the RPC is synchronous — output written after it returns must land in the
+/// rotated file).
+pub(super) struct LogReopenHandle {
+    pub(super) request: Arc<Notify>,
+    pub(super) done: Arc<Notify>,
+}
+type LogReopenMap = Arc<RwLock<HashMap<String, LogReopenHandle>>>;
 type ContainerEventSender = broadcast::Sender<ContainerEventResponse>;
 
 const CRI_CONTAINER_ROOTFS_HOST_DIR: &str = "cri-container-rootfs";
@@ -1059,6 +1068,7 @@ impl RuntimeService for BoxRuntimeService {
         let (attach_tx, _) = broadcast::channel(128);
         let (stop_tx, stop_rx) = oneshot::channel();
         let log_reopen = Arc::new(Notify::new());
+        let log_reopen_done = Arc::new(Notify::new());
         if started {
             self.attach_streams
                 .write()
@@ -1074,10 +1084,13 @@ impl RuntimeService for BoxRuntimeService {
                 .write()
                 .await
                 .insert(container_id.clone(), stop_tx);
-            self.log_reopens
-                .write()
-                .await
-                .insert(container_id.clone(), log_reopen.clone());
+            self.log_reopens.write().await.insert(
+                container_id.clone(),
+                LogReopenHandle {
+                    request: log_reopen.clone(),
+                    done: log_reopen_done.clone(),
+                },
+            );
             self.emit_container_event(
                 &container_id,
                 &container.sandbox_id,
@@ -1101,6 +1114,7 @@ impl RuntimeService for BoxRuntimeService {
             attach_tx,
             stop_rx,
             log_reopen,
+            log_reopen_done,
             workload,
         });
 
@@ -2254,11 +2268,34 @@ impl RuntimeService for BoxRuntimeService {
         );
 
         // Signal the container's supervisor to reopen its log writer at
-        // `log_path`. The kubelet rotates by renaming the current file before
-        // calling this, so the supervisor must drop the stale handle and open a
-        // fresh file at the original path (see `CriLogWriter::reopen`).
-        if let Some(reopen) = self.log_reopens.read().await.get(container_id) {
-            reopen.notify_one();
+        // `log_path` and WAIT for it to finish. The kubelet rotates by renaming
+        // the current file before calling this, so the supervisor must drop the
+        // stale handle and open a fresh file at the original path (see
+        // `CriLogWriter::reopen`). Returning synchronously guarantees output
+        // written after this RPC lands in the rotated file rather than racing
+        // the supervisor. Clone the notify handles out of the lock so we never
+        // hold it across the await.
+        let handles = {
+            let reopens = self.log_reopens.read().await;
+            reopens
+                .get(container_id)
+                .map(|handle| (handle.request.clone(), handle.done.clone()))
+        };
+        if let Some((request, done)) = handles {
+            // Register the done future BEFORE signalling so a fast supervisor
+            // cannot complete the reopen before we start waiting (notify_one
+            // also stores a permit, so there is no lost-wakeup either way).
+            let done = done.notified();
+            request.notify_one();
+            if tokio::time::timeout(std::time::Duration::from_secs(5), done)
+                .await
+                .is_err()
+            {
+                tracing::warn!(
+                    container_id = %container_id,
+                    "ReopenContainerLog timed out waiting for the supervisor to reopen the log"
+                );
+            }
         }
 
         Ok(Response::new(ReopenContainerLogResponse {}))

src/cri/src/runtime_service/supervisor.rs

@@ -54,6 +54,7 @@ pub(super) struct ContainerExitSupervisor {
     pub(super) attach_tx: AttachStreamSender,
     pub(super) stop_rx: oneshot::Receiver<()>,
     pub(super) log_reopen: Arc<Notify>,
+    pub(super) log_reopen_done: Arc<Notify>,
     pub(super) workload: SupervisedWorkload,
 }
 
@@ -72,6 +73,7 @@ pub(super) fn spawn_container_exit_supervisor(supervisor: ContainerExitSuperviso
             attach_tx,
             stop_rx,
             log_reopen,
+            log_reopen_done,
             workload,
         } = supervisor;
         let mut workload = workload;
@@ -113,6 +115,9 @@ pub(super) fn spawn_container_exit_supervisor(supervisor: ContainerExitSuperviso
                             log_writer = CriLogWriter::open(&log_path).await.ok().flatten();
                         }
                     }
+                    // Acknowledge so a synchronous ReopenContainerLog can return
+                    // only now that the new log file is in place.
+                    log_reopen_done.notify_one();
                 }
                 stop = &mut stop_rx, if !stop_requested => {
                     stop_requested = true;

src/cri/src/runtime_service/tests.rs

@@ -3191,25 +3191,33 @@ async fn test_reopen_container_log_signals_supervisor() {
     let svc = make_test_service();
     svc.store.containers.add(test_container("c-1", "sb-1")).await;
 
-    // Register a reopen signal as StartContainer does for a running container.
+    // Register a reopen handle as StartContainer does for a running container.
     // The actual file reopen happens in the exit supervisor (integration-tested
-    // via critest); here we verify the RPC fires the per-container signal.
-    let signal = std::sync::Arc::new(tokio::sync::Notify::new());
-    svc.log_reopens
-        .write()
-        .await
-        .insert("c-1".to_string(), signal.clone());
+    // via critest); here we verify the RPC fires the per-container request and
+    // returns once the supervisor acks. Pre-arm `done` so the now-synchronous
+    // RPC does not block waiting for a supervisor that isn't running here.
+    let request = std::sync::Arc::new(tokio::sync::Notify::new());
+    let done = std::sync::Arc::new(tokio::sync::Notify::new());
+    done.notify_one();
+    svc.log_reopens.write().await.insert(
+        "c-1".to_string(),
+        super::LogReopenHandle {
+            request: request.clone(),
+            done: done.clone(),
+        },
+    );
 
     svc.reopen_container_log(Request::new(ReopenContainerLogRequest {
         container_id: "c-1".to_string(),
     }))
     .await
     .unwrap();
 
-    // notify_one() stores a permit, so notified() resolves immediately.
-    tokio::time::timeout(std::time::Duration::from_secs(1), signal.notified())
+    // The RPC fired the per-container reopen request (notify_one stores a
+    // permit, so notified() resolves immediately).
+    tokio::time::timeout(std::time::Duration::from_secs(1), request.notified())
         .await
-        .expect("ReopenContainerLog should signal the container's reopen notify");
+        .expect("ReopenContainerLog should signal the container's reopen request");
 }
 
 // ── Stop/Remove Pod Sandbox (store-only, no VM) ──────────────────