Function start_watch

crates/openshell-driver-podman/src/watcher.rs:72–157

Start a watch stream that emits current state and live events. The stream first emits a snapshot of all currently-running managed sandboxes (initial state sync), then delivers live container events as they arrive from the Podman event stream. # Reconnection contract The returned stream is **single-use**. When the Podman event connection drops (daemon restart, socket error, or clean shutdown),

(client: PodmanClient)

/// would race with `watch_loop`'s retry and produce duplicate initial-sync
/// events that corrupt the server's sandbox index.
pub async fn start_watch(client: PodmanClient) -> Result<WatchStream, PodmanApiError> {
    let (tx, rx) = mpsc::channel::<Result<WatchSandboxesEvent, ComputeDriverError>>(256);

    // 1. Subscribe to events first so we don't miss any during the list.
    let mut event_rx = client.events_stream(LABEL_MANAGED_FILTER).await?;

    // 2. List existing containers for initial state sync.
    let existing = client.list_containers(LABEL_MANAGED_FILTER).await?;

    for entry in &existing {
        // For running containers, use inspect to get full state including
        // health check status — matching the same condition derivation used
        // for live events.
        if entry.state == "running" {
            match client.inspect_container(&entry.id).await {
                Ok(inspect) => {
                    if let Some(sandbox) = driver_sandbox_from_inspect(&inspect) {
                        if tx.send(Ok(sandbox_event(sandbox))).await.is_err() {
                            return Err(PodmanApiError::Connection(
                                "watch receiver dropped during initial sync".into(),
                            ));
                        }
                        continue;
                    }
                }
                Err(e) => {
                    warn!(
                        container_id = %entry.id,
                        error = %e,
                        "Failed to inspect running container during initial sync, falling back to list entry"
                    );
                }
            }
        }
        if let Some(event) = driver_sandbox_from_list_entry(entry).map(sandbox_event)
            && tx.send(Ok(event)).await.is_err()
        {
            return Err(PodmanApiError::Connection(
                "watch receiver dropped during initial sync".into(),
            ));
        }
    }

    // 3. Stream live events (buffered during the list operation above).

    tokio::spawn(async move {
        while let Some(result) = event_rx.recv().await {
            match result {
                Ok(event) => {
                    if let Some(we) = map_podman_event(&event, &client).await
                        && tx.send(Ok(we)).await.is_err()
                    {
                        return;
                    }
                }
                Err(e) => {
                    if tx
                        .send(Err(ComputeDriverError::Message(e.to_string())))