sail/

worker.rs

//! Per-sailbox worker-proxy client: the gRPC operations that terminate at a
//! sailbox's own worker proxy (exec wait/cancel, listeners, files), sharing one
//! lazily-dialed, drain-aware channel cache and the API-key credential.
//!
//! Transient-failure retry and drain-aware channel eviction live here; the
//! interrupt UX (Ctrl-C cascades, AbortSignal) stays in the language wrappers.

use std::error::Error as _;
use std::sync::Arc;
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;
use tokio::sync::Mutex as AsyncMutex;
use tokio_stream::wrappers::ReceiverStream;
use tonic::metadata::{AsciiMetadataKey, AsciiMetadataValue};
use tonic::transport::Channel;
use tonic::{Code, Request, Status};

use crate::channels::ChannelCache;
use crate::error::SailError;
use crate::pb::workerproxy::v1 as pb;
use pb::worker_proxy_service_client::WorkerProxyServiceClient;

/// How many file chunks to buffer between the caller and the wire, for
/// backpressure on both read and write.
const FILE_CHANNEL_CAP: usize = 4;

/// Chunk size for streaming a file write: each `FileWriter::write_chunk` call
/// becomes one gRPC message, so this keeps a chunk under the transport's message
/// limit. Bindings stream their source in pieces of this size.
pub const FILE_WRITE_CHUNK_BYTES: usize = 1 << 20;

/// Initial backoff in seconds before the first transient-RPC retry; doubled on
/// each subsequent attempt.
pub(crate) const EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS: f64 = 0.2;
/// Ceiling in seconds for the exponential backoff between transient-RPC retries.
pub(crate) const EXEC_TRANSIENT_RETRY_MAX_DELAY_SECONDS: f64 = 2.0;
/// Per-attempt deadline for a unary exec RPC. Caps a single attempt so a
/// stalled (not dead) connection times out and the retry loop runs again
/// within the overall budget, instead of one await consuming it all.
pub(crate) const EXEC_RPC_ATTEMPT_TIMEOUT_SECONDS: f64 = 10.0;

/// gRPC status-message fragments that mark a mid-flight connection drop (a peer
/// rolling/lameducking during a deploy) rather than a permanent failure, so the
/// RPC is worth retrying. Shared by the exec and imagebuilder retry paths.
pub(crate) const TRANSIENT_TRANSPORT_FRAGMENTS: &[&str] = &[
    "endpoint closing",
    "error reading server preface",
    "connection reset",
    "socket closed",
    "transport is closing",
    "h2 protocol error",
    "keep-alive timed out",
];

/// Whether a gRPC status message names a transient transport drop (see
/// [`TRANSIENT_TRANSPORT_FRAGMENTS`]). Case-insensitive.
pub(crate) fn is_transient_transport_message(message: &str) -> bool {
    let details = message.to_lowercase();
    TRANSIENT_TRANSPORT_FRAGMENTS
        .iter()
        .any(|fragment| details.contains(fragment))
}

/// Authoritative buffered result from polling `WaitSailboxExec`. Callers
/// inspect `status` (e.g. a terminal failure) and shape the public result.
/// Binding plumbing like its producer [`WorkerProxy`], hence doc-hidden: the
/// typed public result is [`crate::exec::ExecResult`].
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct WaitOutcome {
    /// Terminal exec status as the `SailboxExecStatus` proto enum value.
    pub status: i32,
    /// Buffered stdout from the persisted row.
    pub stdout: String,
    /// Buffered stderr from the persisted row.
    pub stderr: String,
    /// The command's exit code.
    pub exit_code: i32,
    /// Whether the command was killed for exceeding its timeout.
    pub timed_out: bool,
    /// Whether stdout overflowed the server output ring and lost its oldest
    /// bytes.
    pub stdout_truncated: bool,
    /// Whether stderr overflowed the server output ring and lost its oldest
    /// bytes.
    pub stderr_truncated: bool,
}

/// A sailbox ingress listener, parsed from the sailbox-API listener JSON. The
/// backend always sends the port, protocol, and route status; the public
/// address fields are absent until the route is active.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Listener {
    /// The in-guest port traffic is forwarded to.
    pub guest_port: u32,
    /// Wire protocol exposed, e.g. `tcp` or `http`.
    pub protocol: crate::sailbox::types::ListenerProtocol,
    /// Status of the listener's ingress route.
    pub route_status: crate::sailbox::types::ListenerRouteStatus,
    /// Publicly reachable URL for the listener.
    #[serde(default)]
    pub public_url: String,
    /// Public hostname the listener is reachable at.
    #[serde(default)]
    pub public_host: String,
    /// Public port the listener is reachable at.
    #[serde(default)]
    pub public_port: u32,
}

impl Listener {
    /// The typed endpoint, or `None` until the listener is routable.
    pub fn endpoint(&self) -> Option<crate::sailbox::types::ListenerEndpoint> {
        use crate::sailbox::types::ListenerEndpoint;
        if !self.public_url.is_empty() {
            return Some(ListenerEndpoint::Http {
                url: self.public_url.clone(),
            });
        }
        if !self.public_host.is_empty() && self.public_port != 0 {
            return Some(ListenerEndpoint::Tcp {
                host: self.public_host.clone(),
                port: self.public_port,
            });
        }
        None
    }

    /// Whether the route is active and ready to carry traffic.
    pub fn is_active(&self) -> bool {
        self.route_status == crate::sailbox::types::ListenerRouteStatus::Active
    }
}

/// Optional settings for [`SailboxFs::write`](crate::SailboxFs::write) and
/// [`SailboxFs::write_stream`](crate::SailboxFs::write_stream).
/// `Default` creates missing parent directories and leaves the file's mode to
/// the guest's default.
#[derive(Debug, Clone)]
pub struct WriteOptions {
    /// Create missing parent directories before writing.
    pub create_parents: bool,
    /// Unix mode bits for the written file; `None` leaves the guest default.
    pub mode: Option<u32>,
}

impl Default for WriteOptions {
    /// Parents are created by default, matching every SDK surface.
    fn default() -> WriteOptions {
        WriteOptions {
            create_parents: true,
            mode: None,
        }
    }
}

/// A streaming reader over a guest-file read response. A background task
/// pumps chunks into a bounded channel (so a slow consumer applies
/// backpressure rather than buffering the whole file); [`FileReader::next`]
/// yields the next chunk, `None` at end of file.
pub struct FileReader {
    rx: AsyncMutex<mpsc::Receiver<Result<Vec<u8>, SailError>>>,
    abort: tokio::task::AbortHandle,
}

impl FileReader {
    /// Yield the next file chunk, or `None` at end of file. A stream error
    /// surfaces as `Some(Err(..))`.
    pub async fn next(&self) -> Option<Result<Vec<u8>, SailError>> {
        self.rx.lock().await.recv().await
    }

    /// Abort the background download, cancelling its gRPC stream even when a
    /// `next()` is stalled awaiting a chunk. Idempotent; a pending or subsequent
    /// `next()` then observes end of stream (`None`).
    pub fn close(&self) {
        self.abort.abort();
    }
}

/// Client for the worker-proxy gRPC operations of a single sailbox, holding the
/// shared lazily-dialed channel cache and the bearer credential applied to every
/// request.
#[doc(hidden)]
pub struct WorkerProxy {
    channels: ChannelCache,
    authorization: AsciiMetadataValue,
}

/// A `retry_timeout <= 0` deadline is "now" so the very first transient
/// failure short-circuits, preserving the no-retry-by-default contract
/// for programmatic cancel callers. A non-finite or overflowing timeout
/// (e.g. `float("inf")` from Python) means "retry forever".
pub(crate) fn retry_deadline(retry_timeout: f64) -> Instant {
    let now = Instant::now();
    if retry_timeout <= 0.0 {
        return now;
    }
    // Clamp before converting: `f64::min` returns the non-NaN argument, so a
    // non-finite or astronomically large timeout (e.g. `float("inf")` from
    // Python, meaning "retry forever") folds to the cap. A finite cap keeps
    // `from_secs_f64` panic-free and `now + dur` within `Instant`'s range.
    now + Duration::from_secs_f64(retry_timeout.min(RETRY_FOREVER_SECS))
}

/// Cap on the retry budget (~a century): large enough to act as "retry until
/// success or cancel", small enough not to overflow `Instant`.
const RETRY_FOREVER_SECS: f64 = 100.0 * 365.0 * 24.0 * 60.0 * 60.0;

/// Bound a single unary-RPC attempt: at most [`EXEC_RPC_ATTEMPT_TIMEOUT_SECONDS`]
/// and at most half the remaining budget. Halving leaves headroom for at least
/// one retry. Otherwise, on a budget smaller than the ceiling, a single attempt
/// hanging on a half-open connection would consume the whole budget and the loop
/// would give up without ever redialing.
pub(crate) fn rpc_attempt_timeout(deadline: Instant) -> Duration {
    (deadline.saturating_duration_since(Instant::now()) / 2)
        .min(Duration::from_secs_f64(EXEC_RPC_ATTEMPT_TIMEOUT_SECONDS))
        .max(Duration::from_millis(1))
}

pub(crate) fn is_workerproxy_draining(status: &Status) -> bool {
    status.code() == Code::Unavailable && status.message().to_lowercase().contains("draining")
}

/// Whether a retryable failure should also drop the cached channel before the
/// next attempt. The connection itself is suspect when the target is draining,
/// on a server-enforced deadline (`DeadlineExceeded`), or on any client-side
/// transport failure (a half-open socket, keepalive timeout, failed connect, or
/// a fired per-attempt `set_timeout`, all of which tonic surfaces as a status
/// carrying a transport `source`). Reusing such a channel would burn every retry
/// on the dead connection, so dial fresh instead. A server-sent transient (no
/// source, not draining, not a deadline) keeps the connection.
pub(crate) fn should_invalidate_channel(status: &Status) -> bool {
    is_workerproxy_draining(status)
        || status.code() == Code::DeadlineExceeded
        || status.source().is_some()
}

pub(crate) fn should_retry_transient_exec_rpc(status: &Status, deadline: Instant) -> bool {
    if Instant::now() >= deadline {
        return false;
    }
    match status.code() {
        Code::Unavailable | Code::DeadlineExceeded => true,
        // A fired per-attempt `set_timeout` surfaces as `Cancelled` carrying the
        // timeout as a source, as does a client-side transport cancel; both are
        // transient and retryable. A server-sent `CANCELLED` (no source) is a
        // deliberate cancellation and is left alone.
        Code::Cancelled => status.source().is_some(),
        // tonic maps client-side connection failures onto `Internal` (h2
        // protocol errors: a GOAWAY, a stream reset, a dead keepalive) or
        // `Unknown` (transport errors it cannot classify), and always attaches
        // the underlying error as `source`. A status decoded from server
        // response trailers never carries a source, so a source means the
        // connection died, not that the server ruled — retry on a fresh channel
        // (every RPC gated here is idempotent: launches dedupe on the
        // idempotency key, stdin writes carry absolute offsets). The message
        // fragments cover the same failures relayed as text without a source.
        Code::Unknown | Code::Internal => {
            status.source().is_some() || is_transient_transport_message(status.message())
        }
        // The same source rule applies to EVERY code: h2 GOAWAY reasons map
        // outside the set above — grpc-go's keepalive enforcement sends
        // GOAWAY(ENHANCE_YOUR_CALM, "too_many_pings"), which tonic surfaces as
        // RESOURCE_EXHAUSTED with the transport error as source. A genuine
        // server-sent RESOURCE_EXHAUSTED (rate limit, message size) arrives in
        // trailers with no source and is still surfaced, not retried.
        _ => status.source().is_some(),
    }
}

/// Sleep before the next retry attempt, bounded by the max per-retry delay
/// and the remaining budget; returns the doubled delay for the next round.
pub(crate) async fn sleep_before_retry(delay: f64, deadline: Instant) -> f64 {
    let mut sleep_for = delay.min(EXEC_TRANSIENT_RETRY_MAX_DELAY_SECONDS);
    let remaining = deadline
        .saturating_duration_since(Instant::now())
        .as_secs_f64();
    if remaining <= 0.0 {
        return delay;
    }
    sleep_for = sleep_for.min(remaining);
    tokio::time::sleep(Duration::from_secs_f64(sleep_for.max(0.0))).await;
    (delay * 2.0).min(EXEC_TRANSIENT_RETRY_MAX_DELAY_SECONDS)
}

/// Build the `Bearer <key>` gRPC `authorization` metadata value, failing if the
/// key has characters invalid in a metadata value. Shared by the worker-proxy
/// and imagebuilder clients.
pub(crate) fn bearer_metadata(api_key: &str) -> Result<AsciiMetadataValue, SailError> {
    format!("Bearer {api_key}")
        .parse()
        .map_err(|_| SailError::Config {
            message: "SAIL_API_KEY contains characters invalid in a gRPC metadata value"
                .to_string(),
        })
}

impl WorkerProxy {
    /// Build a worker proxy that authenticates with `api_key`. Fails if the key
    /// cannot form a valid gRPC `authorization` metadata value.
    pub fn new(api_key: &str) -> Result<WorkerProxy, SailError> {
        let authorization = bearer_metadata(api_key)?;
        Ok(WorkerProxy {
            channels: ChannelCache::new(),
            authorization,
        })
    }

    pub(crate) fn channels(&self) -> &ChannelCache {
        &self.channels
    }

    pub(crate) fn client_for(
        &self,
        endpoint: &str,
    ) -> Result<WorkerProxyServiceClient<Channel>, SailError> {
        let channel = self.channels.get(endpoint)?;
        Ok(WorkerProxyServiceClient::new(channel))
    }

    pub(crate) fn request_for<T>(
        &self,
        message: T,
        extra_metadata: &[(String, String)],
        timeout: Option<Duration>,
    ) -> Result<Request<T>, SailError> {
        let mut request = Request::new(message);
        request
            .metadata_mut()
            .insert("authorization", self.authorization.clone());
        for (key, value) in extra_metadata {
            let key: AsciiMetadataKey = key.parse().map_err(|_| SailError::Config {
                message: format!("invalid gRPC metadata key {key:?}"),
            })?;
            let value: AsciiMetadataValue = value.parse().map_err(|_| SailError::Config {
                message: format!("invalid gRPC metadata value for key {key:?}"),
            })?;
            request.metadata_mut().insert(key, value);
        }
        if let Some(timeout) = timeout {
            request.set_timeout(timeout);
        }
        Ok(request)
    }

    /// Wait for an exec to finish. The retry deadline starts at the FIRST
    /// transient error, not at the call: a wait legitimately blocks for as
    /// long as the guest command runs, so only consecutive failure time is
    /// budgeted.
    pub async fn wait_exec(
        &self,
        endpoint: &str,
        sailbox_id: &str,
        exec_request_id: &str,
        retry_timeout: f64,
    ) -> Result<WaitOutcome, SailError> {
        let mut deadline: Option<Instant> = None;
        let mut delay = EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS;
        loop {
            let message = pb::WaitSailboxExecRequest {
                sailbox_id: sailbox_id.to_string(),
                exec_request_id: exec_request_id.to_string(),
            };
            let request = self.request_for(message, &[], /* timeout */ None)?;
            match self.client_for(endpoint)?.wait_sailbox_exec(request).await {
                Ok(resp) => {
                    let resp = resp.into_inner();
                    return Ok(WaitOutcome {
                        status: resp.status,
                        stdout: resp.stdout,
                        stderr: resp.stderr,
                        exit_code: resp.return_code,
                        timed_out: resp.timed_out,
                        stdout_truncated: resp.stdout_truncated,
                        stderr_truncated: resp.stderr_truncated,
                    });
                }
                Err(status) => {
                    let deadline = *deadline.get_or_insert_with(|| retry_deadline(retry_timeout));
                    if !should_retry_transient_exec_rpc(&status, deadline) {
                        return Err(SailError::from_exec_status(&status));
                    }
                    tracing::warn!(code = ?status.code(), endpoint, "retrying transient worker-proxy RPC");
                    if should_invalidate_channel(&status) {
                        self.channels.invalidate(endpoint);
                    }
                    delay = sleep_before_retry(delay, deadline).await;
                }
            }
        }
    }

    /// Signal the guest command: SIGINT by default, SIGKILL if force.
    /// `retry_timeout > 0` budgets retries across the saild
    /// registration-gap window AND bounds each attempt with a per-call
    /// gRPC deadline so a stuck connection cannot hang past the budget.
    pub async fn cancel_exec(
        &self,
        endpoint: &str,
        sailbox_id: &str,
        exec_request_id: &str,
        force: bool,
        retry_timeout: f64,
    ) -> Result<(), SailError> {
        let deadline = retry_deadline(retry_timeout);
        let mut delay = EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS;
        loop {
            let per_attempt_timeout = if retry_timeout > 0.0 {
                // Cap each attempt so a half-open connection times out and the
                // loop redials, instead of one attempt consuming the whole
                // budget. retry_timeout == 0 keeps a single uncapped attempt
                // (programmatic no-retry cancel).
                Some(rpc_attempt_timeout(deadline))
            } else {
                None
            };
            let message = pb::CancelSailboxExecRequest {
                sailbox_id: sailbox_id.to_string(),
                exec_request_id: exec_request_id.to_string(),
                force,
            };
            let request = self.request_for(message, &[], per_attempt_timeout)?;
            match self
                .client_for(endpoint)?
                .cancel_sailbox_exec(request)
                .await
            {
                Ok(_) => return Ok(()),
                Err(status) => {
                    if !should_retry_transient_exec_rpc(&status, deadline) {
                        return Err(SailError::from_exec_status(&status));
                    }
                    tracing::warn!(code = ?status.code(), endpoint, "retrying transient worker-proxy RPC");
                    if should_invalidate_channel(&status) {
                        self.channels.invalidate(endpoint);
                    }
                    delay = sleep_before_retry(delay, deadline).await;
                }
            }
        }
    }

    /// Open a streaming read of a guest file. The returned [`FileReader`] yields
    /// chunks as they arrive; errors surface from its `next`.
    ///
    /// # Runtime
    ///
    /// Spawns the background pump on the calling task's tokio runtime, so call it
    /// from within one (every binding does, via the shared runtime; an async host
    /// from its own). The dialed channel co-locates on that runtime.
    pub fn read_file(self: &Arc<Self>, endpoint: &str, sailbox_id: &str, path: &str) -> FileReader {
        let (tx, rx) = mpsc::channel(FILE_CHANNEL_CAP);
        let worker = Arc::clone(self);
        let endpoint = endpoint.to_string();
        let message = pb::ReadSailboxFileRequest {
            sailbox_id: sailbox_id.to_string(),
            path: path.to_string(),
        };
        let task = tokio::spawn(async move {
            let request = match worker.request_for(message, &[], /* timeout */ None) {
                Ok(request) => request,
                Err(err) => {
                    let _ = tx.send(Err(err)).await;
                    return;
                }
            };
            let mut client = match worker.client_for(&endpoint) {
                Ok(client) => client,
                Err(err) => {
                    let _ = tx.send(Err(err)).await;
                    return;
                }
            };
            let mut stream = match client.read_sailbox_file(request).await {
                Ok(resp) => resp.into_inner(),
                Err(status) => {
                    let _ = tx.send(Err(SailError::from_file_rpc_status(&status))).await;
                    return;
                }
            };
            loop {
                match stream.message().await {
                    Ok(Some(resp)) => {
                        if !resp.data.is_empty() && tx.send(Ok(resp.data)).await.is_err() {
                            return; // the reader was dropped
                        }
                    }
                    Ok(None) => return,
                    Err(status) => {
                        let _ = tx.send(Err(SailError::from_file_rpc_status(&status))).await;
                        return;
                    }
                }
            }
        });
        FileReader {
            rx: AsyncMutex::new(rx),
            abort: task.abort_handle(),
        }
    }

    /// Open a streaming write to a guest file. The caller feeds chunks via
    /// [`FileWriter::write_chunk`] and ends with [`FileWriter::finish`], so a
    /// large source is never buffered whole. The first chunk carries the
    /// path/flags; the rest carry data only.
    ///
    /// # Runtime
    ///
    /// Spawns the streaming RPC on the calling task's tokio runtime, so call it
    /// from within one. The dialed channel co-locates on that runtime.
    pub fn write_file(
        self: &Arc<Self>,
        endpoint: &str,
        sailbox_id: &str,
        path: &str,
        create_parents: bool,
        mode: Option<u32>,
    ) -> FileWriter {
        let (tx, rx) = mpsc::channel(FILE_CHANNEL_CAP);
        let worker = Arc::clone(self);
        let endpoint = endpoint.to_string();
        // Build the client and run the RPC inside the spawned task so the
        // channel is dialed in the runtime's reactor context.
        let task = tokio::spawn(async move {
            let request =
                worker.request_for(ReceiverStream::new(rx), &[], /* timeout */ None)?;
            worker
                .client_for(&endpoint)?
                .write_sailbox_file(request)
                .await
                .map(|_| ())
                .map_err(|status| SailError::from_file_rpc_status(&status))
        });
        FileWriter {
            tx: Some(tx),
            task: Some(task),
            aborted: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            first: true,
            sailbox_id: sailbox_id.to_string(),
            path: path.to_string(),
            create_parents,
            mode,
        }
    }
}

/// A streaming write to a guest file. Chunks feed a bounded channel that backs
/// the client-streaming RPC, so a slow uplink applies backpressure rather than
/// buffering the source. The RPC result surfaces from [`FileWriter::finish`],
/// and only `finish` commits the write: dropping (or [`FileWriter::abort`]ing)
/// an unfinished writer cancels the RPC instead of half-closing into what the
/// server would treat as a completed write. The guest file state after an
/// abort is unspecified (the write was never confirmed).
pub struct FileWriter {
    tx: Option<mpsc::Sender<pb::WriteSailboxFileRequest>>,
    task: Option<tokio::task::JoinHandle<Result<(), SailError>>>,
    aborted: Arc<std::sync::atomic::AtomicBool>,
    first: bool,
    sailbox_id: String,
    path: String,
    create_parents: bool,
    mode: Option<u32>,
}

impl FileWriter {
    fn build(&mut self, data: Vec<u8>) -> pb::WriteSailboxFileRequest {
        let header = self.first;
        self.first = false;
        pb::WriteSailboxFileRequest {
            sailbox_id: if header {
                self.sailbox_id.clone()
            } else {
                String::new()
            },
            path: if header {
                self.path.clone()
            } else {
                String::new()
            },
            data,
            create_parents: header && self.create_parents,
            mode: if header { self.mode } else { None },
        }
    }

    /// Stop the stream and return the RPC's result.
    async fn join(&mut self) -> Result<(), SailError> {
        self.tx = None; // dropping the sender ends the client stream
        match self.task.take() {
            Some(task) => task.await.unwrap_or_else(|join_err| {
                Err(SailError::Internal {
                    message: format!("file write task failed: {join_err}"),
                })
            }),
            None => Ok(()),
        }
    }

    /// Write bytes to the file, splitting them into transport-sized chunks
    /// ([`FILE_WRITE_CHUNK_BYTES`] each). This is the normal write path; use
    /// [`FileWriter::write_chunk`] only to control message framing yourself.
    pub async fn write(&mut self, data: &[u8]) -> Result<(), SailError> {
        for chunk in data.chunks(FILE_WRITE_CHUNK_BYTES) {
            self.write_chunk(chunk.to_vec()).await?;
        }
        Ok(())
    }

    /// Send one chunk of file data as a single transport message. Chunks must
    /// not exceed [`FILE_WRITE_CHUNK_BYTES`]. If the RPC has already ended,
    /// returns its result instead.
    pub async fn write_chunk(&mut self, data: Vec<u8>) -> Result<(), SailError> {
        if self.aborted.load(std::sync::atomic::Ordering::Relaxed) {
            return Err(aborted_write());
        }
        let request = self.build(data);
        match &self.tx {
            // A send error means the RPC task already ended; surface its result.
            Some(tx) if tx.send(request).await.is_ok() => Ok(()),
            _ => self.join().await,
        }
    }

    /// Finish the write and return the RPC's result, creating an empty file when
    /// no chunks were sent.
    pub async fn finish(&mut self) -> Result<(), SailError> {
        if self.aborted.load(std::sync::atomic::Ordering::Relaxed) {
            return Err(aborted_write());
        }
        if self.first {
            // No chunks were written: send a header-only message so an empty
            // file is still created.
            let request = self.build(Vec::new());
            if let Some(tx) = &self.tx {
                let _ = tx.send(request).await;
            }
        }
        self.join().await
    }

    /// Abort the write: cancel the RPC without the clean end-of-stream the
    /// server would commit. Idempotent; a no-op after `finish`. Later `write`
    /// or `finish` calls report the abort instead of succeeding.
    pub fn abort(&mut self) {
        // Cancel the RPC task before dropping the sender: dropping the
        // channel first can wake the task into seeing end-of-stream and
        // half-closing cleanly — exactly the commit an abort must prevent.
        if let Some(task) = self.task.take() {
            self.aborted
                .store(true, std::sync::atomic::Ordering::Relaxed);
            task.abort();
        }
        self.tx = None;
    }

    /// An out-of-band handle that aborts this write without borrowing the
    /// writer, so a concurrent caller (the bindings' abort path) can cancel a
    /// stalled or backpressured `write` instead of queueing behind it.
    pub fn abort_handle(&self) -> WriteAbortHandle {
        WriteAbortHandle {
            aborted: Arc::clone(&self.aborted),
            task: self
                .task
                .as_ref()
                .map(tokio::task::JoinHandle::abort_handle),
        }
    }
}

/// Cancels a [`FileWriter`]'s RPC out of band. See [`FileWriter::abort`] for
/// the semantics: only `finish` commits, and an aborted write reports the
/// abort on later calls.
#[derive(Clone)]
pub struct WriteAbortHandle {
    aborted: Arc<std::sync::atomic::AtomicBool>,
    task: Option<tokio::task::AbortHandle>,
}

impl WriteAbortHandle {
    /// Abort the write: cancel the RPC so the server does not commit it.
    /// Idempotent; a no-op after `finish`.
    pub fn abort(&self) {
        self.aborted
            .store(true, std::sync::atomic::Ordering::Relaxed);
        if let Some(task) = &self.task {
            task.abort();
        }
    }
}

fn aborted_write() -> SailError {
    SailError::InvalidArgument {
        message: "the write was aborted; nothing was committed".to_string(),
    }
}

impl Drop for FileWriter {
    fn drop(&mut self) {
        // An unfinished writer must never half-close into a committed write.
        self.abort();
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn write_splits_at_the_transport_chunk_size() {
        let (tx, mut rx) = mpsc::channel(16);
        let mut writer = FileWriter {
            tx: Some(tx),
            task: Some(tokio::spawn(async { Ok(()) })),
            aborted: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            first: true,
            sailbox_id: "sb_1".to_string(),
            path: "/f".to_string(),
            create_parents: true,
            mode: None,
        };
        writer
            .write(&vec![7u8; FILE_WRITE_CHUNK_BYTES * 2 + 10])
            .await
            .expect("write succeeds");
        writer.finish().await.expect("finish succeeds");
        let mut sizes = Vec::new();
        while let Some(message) = rx.recv().await {
            sizes.push(message.data.len());
        }
        assert_eq!(
            sizes,
            vec![FILE_WRITE_CHUNK_BYTES, FILE_WRITE_CHUNK_BYTES, 10]
        );
    }

    #[test]
    fn transient_codes_retry_within_deadline() {
        let deadline = Instant::now() + Duration::from_secs(5);
        assert!(should_retry_transient_exec_rpc(
            &Status::unavailable("x"),
            deadline
        ));
        assert!(should_retry_transient_exec_rpc(
            &Status::deadline_exceeded("x"),
            deadline
        ));
        assert!(should_retry_transient_exec_rpc(
            &Status::unknown("HTTP/2 connection reset by remote"),
            deadline
        ));
        assert!(!should_retry_transient_exec_rpc(
            &Status::unknown("guest exploded"),
            deadline
        ));
        assert!(!should_retry_transient_exec_rpc(
            &Status::not_found("x"),
            deadline
        ));
    }

    #[test]
    fn h2_connection_failures_retry_within_deadline() {
        let deadline = Instant::now() + Duration::from_secs(5);
        // A dropped HTTP/2 connection under exec-storm concurrency: tonic maps
        // it to Internal/Unknown with the transport error attached as source.
        let io = std::io::Error::new(
            std::io::ErrorKind::ConnectionAborted,
            "connection error: h2 protocol error: http2 error",
        );
        let transport = Status::from_error(Box::new(io));
        assert!(transport.source().is_some());
        assert!(should_retry_transient_exec_rpc(&transport, deadline));
        // The same failure relayed as message text without a source (e.g. a
        // proxy stringifying its upstream error) retries via the fragment list.
        assert!(should_retry_transient_exec_rpc(
            &Status::internal("h2 protocol error: http2 error"),
            deadline
        ));
        assert!(should_retry_transient_exec_rpc(
            &Status::unknown("connection error: keep-alive timed out"),
            deadline
        ));
        // A genuine server-sent INTERNAL (trailer status, no source, no
        // transport fragment) is a server verdict and is not retried.
        assert!(!should_retry_transient_exec_rpc(
            &Status::internal("guest agent panicked"),
            deadline
        ));
    }

    #[test]
    fn goaway_reason_statuses_retry_only_with_transport_source() {
        let deadline = Instant::now() + Duration::from_secs(5);
        // grpc-go keepalive enforcement (the workerproxy's 10s MinTime) sends
        // GOAWAY(ENHANCE_YOUR_CALM, "too_many_pings") when a stalled client's
        // queued ping timers fire in a burst. tonic maps that reason to
        // RESOURCE_EXHAUSTED with the h2 error as source — a dead connection,
        // not a server verdict. Reproduced live against a raw-framer server:
        // code=ResourceExhausted, message="h2 protocol error: http2 error".
        let mut goaway = Status::new(Code::ResourceExhausted, "h2 protocol error: http2 error");
        goaway.set_source(std::sync::Arc::new(std::io::Error::new(
            std::io::ErrorKind::ConnectionReset,
            "transport error",
        )));
        assert!(should_retry_transient_exec_rpc(&goaway, deadline));
        // A genuine server-sent RESOURCE_EXHAUSTED (rate limit, message-size
        // cap) arrives via trailers with no source: surfaced, never retried.
        assert!(!should_retry_transient_exec_rpc(
            &Status::resource_exhausted("org concurrency limit reached"),
            deadline
        ));
        // Same rule for the other GOAWAY-reason escape (INADEQUATE_SECURITY
        // -> PermissionDenied) and any future reason mapping.
        let mut sec = Status::new(Code::PermissionDenied, "h2 protocol error: http2 error");
        sec.set_source(std::sync::Arc::new(std::io::Error::new(
            std::io::ErrorKind::ConnectionReset,
            "transport error",
        )));
        assert!(should_retry_transient_exec_rpc(&sec, deadline));
        assert!(!should_retry_transient_exec_rpc(
            &Status::permission_denied("invalid API key"),
            deadline
        ));
    }

    #[test]
    fn fired_attempt_timeout_retries_but_server_cancel_does_not() {
        let deadline = Instant::now() + Duration::from_secs(5);
        // A fired per-attempt set_timeout reaches us as Cancelled carrying the
        // timeout as a source: retry it.
        let timed_out = Status::from_error(Box::new(tonic::TimeoutExpired(())));
        assert_eq!(timed_out.code(), Code::Cancelled);
        assert!(should_retry_transient_exec_rpc(&timed_out, deadline));
        // A server-sent CANCELLED has no source: do not retry.
        assert!(!should_retry_transient_exec_rpc(
            &Status::cancelled("client went away"),
            deadline
        ));
    }

    #[test]
    fn expired_deadline_never_retries() {
        let deadline = Instant::now();
        assert!(!should_retry_transient_exec_rpc(
            &Status::unavailable("x"),
            deadline
        ));
    }

    #[test]
    fn draining_detection_is_case_insensitive_and_code_scoped() {
        assert!(is_workerproxy_draining(&Status::unavailable(
            "workerproxy DRAINING for deploy"
        )));
        assert!(!is_workerproxy_draining(&Status::internal("draining")));
        assert!(!is_workerproxy_draining(&Status::unavailable("lameduck")));
    }

    #[test]
    fn rpc_attempt_timeout_caps_and_leaves_retry_headroom() {
        let now = Instant::now();
        // Far deadline: capped at the per-attempt ceiling.
        let far = rpc_attempt_timeout(now + Duration::from_mins(1));
        assert!(far <= Duration::from_secs_f64(EXEC_RPC_ATTEMPT_TIMEOUT_SECONDS));
        assert!(far > Duration::from_secs(5));
        // Budget below the ceiling: a single attempt takes less than the whole
        // budget (~half), so a retry can still land if it hangs.
        let small = rpc_attempt_timeout(now + Duration::from_secs(5));
        assert!(small < Duration::from_secs(5));
        assert!(small <= Duration::from_secs(3));
        // Past deadline: never zero, so the attempt still fires and fails fast.
        assert_eq!(
            rpc_attempt_timeout(now.checked_sub(Duration::from_secs(1)).unwrap()),
            Duration::from_millis(1)
        );
    }

    #[test]
    fn invalidate_on_draining_or_transport_failure_only() {
        // Draining: drop the channel even though it is a clean server status.
        assert!(should_invalidate_channel(&Status::unavailable(
            "workerproxy is draining"
        )));
        // A client-side transport failure carries a source: the connection is
        // suspect, so dial fresh.
        let io = std::io::Error::new(std::io::ErrorKind::ConnectionReset, "socket closed");
        assert!(should_invalidate_channel(&Status::from_error(Box::new(io))));
        // A plain server-sent transient keeps the connection.
        assert!(!should_invalidate_channel(&Status::unavailable(
            "try again"
        )));
    }
}