sail/sailbox/

object.rs

//! The bound Sailbox object: a Sailbox id paired with the [`Client`] that
//! reaches it, so every operation is a method instead of an id-threading call.
//!
//! Each method is a one-line delegate to the corresponding [`Client`] method,
//! which remains the single implementation (and the surface the language
//! bridges call with plain ids). The delegation means the two surfaces cannot
//! drift: a signature change on either side fails to compile.

use crate::client::Client;
use crate::error::SailError;
use crate::exec::{ExecOptions, ExecProcess, ExecResult, RunOptions};
use crate::sailbox::api::UpgradeResult;
use crate::sailbox::fs::DirEntry;
use crate::sailbox::ssh::{EnableSshOptions, SshEndpoint};
use crate::sailbox::types::{
    CheckpointOptions, ForkOptions, IngressProtocol, SailboxCheckpoint, SailboxHandle, SailboxInfo,
    WaitForListenerOptions,
};
use crate::worker::{FileReader, FileWriter, Listener, WriteOptions};
use std::sync::{Arc, RwLock};
use time::OffsetDateTime;

/// Collect a generic argv parameter into the owned form the transport uses.
fn collect_argv(argv: impl IntoIterator<Item = impl Into<String>>) -> Vec<String> {
    argv.into_iter().map(Into::into).collect()
}

/// A Sailbox bound to the client that reaches it. Obtained from
/// [`Client::create_sailbox`], [`Client::create_from_checkpoint`], or
/// [`Client::sailbox`] (which binds an existing id without a network call).
///
/// Cheap to clone; clones share the underlying client transport.
///
/// ```no_run
/// # async fn demo() -> Result<(), sail::error::SailError> {
/// # let client = sail::Client::from_env()?;
/// let sb = client.sailbox("sb_abc123");
/// let result = sb.exec_shell("echo hello", Default::default()).await?.wait().await?;
/// sb.fs().write("/workspace/input.txt", b"hello\n", Default::default()).await?;
/// sb.terminate().await?;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Sailbox {
    client: Client,
    handle: SailboxHandle,
    exec_endpoint: Arc<RwLock<String>>,
}

impl std::fmt::Debug for Sailbox {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Sailbox")
            .field("sailbox_id", &self.handle.sailbox_id)
            .finish_non_exhaustive()
    }
}

impl Sailbox {
    pub(crate) fn bind(client: Client, handle: SailboxHandle) -> Sailbox {
        let exec_endpoint = Arc::new(RwLock::new(handle.exec_endpoint.clone()));
        Sailbox {
            client,
            handle,
            exec_endpoint,
        }
    }

    pub(crate) fn client(&self) -> &Client {
        &self.client
    }

    /// The Sailbox's stable identifier.
    pub fn sailbox_id(&self) -> &str {
        &self.handle.sailbox_id
    }

    /// The data snapshot from the call that produced this object (create,
    /// from-checkpoint). A Sailbox bound by id via [`Client::sailbox`] carries
    /// only the id; use [`Sailbox::info`] for fresh state either way.
    pub fn handle(&self) -> &SailboxHandle {
        &self.handle
    }

    /// Consume the object, keeping just the data snapshot.
    pub fn into_handle(self) -> SailboxHandle {
        self.handle
    }

    /// Returns the latest worker endpoint learned by this object, or an empty
    /// string when the next operation must resolve placement first.
    fn exec_endpoint_hint(&self) -> String {
        self.exec_endpoint
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }

    /// Replaces the shared routing hint for this object and all its clones.
    fn set_exec_endpoint_hint(&self, endpoint: String) {
        *self
            .exec_endpoint
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner) = endpoint;
    }

    /// Clears routing after a lifecycle operation stops the current VM.
    fn clear_exec_endpoint_hint(&self) {
        self.set_exec_endpoint_hint(String::new());
    }

    /// Fetch this Sailbox's current state.
    pub async fn info(&self) -> Result<SailboxInfo, SailError> {
        self.client.get_sailbox(self.sailbox_id()).await
    }

    // --- lifecycle ---

    /// Terminate the Sailbox (idempotent).
    pub async fn terminate(&self) -> Result<(), SailError> {
        self.client.terminate_sailbox(self.sailbox_id()).await?;
        self.clear_exec_endpoint_hint();
        Ok(())
    }

    /// Pause the Sailbox in memory.
    pub async fn pause(&self) -> Result<(), SailError> {
        self.client.pause_sailbox(self.sailbox_id()).await?;
        self.clear_exec_endpoint_hint();
        Ok(())
    }

    /// Sleep the Sailbox to disk (it wakes on traffic). `wake_at`, when
    /// given, schedules a wall-clock wake before the sleep starts and
    /// returns the effective wake time (the sooner of this request and any
    /// wake already scheduled). If the Sailbox is sleeping when that moment
    /// arrives, Sail restores it; a wake can fire a little late, so treat
    /// the time as approximate. Calling sleep on an already-sleeping Sailbox
    /// succeeds and just updates the scheduled wake.
    pub async fn sleep(
        &self,
        wake_at: Option<OffsetDateTime>,
    ) -> Result<Option<OffsetDateTime>, SailError> {
        let effective = self
            .client
            .sleep_sailbox(self.sailbox_id(), wake_at)
            .await?;
        self.clear_exec_endpoint_hint();
        Ok(effective)
    }

    /// Resume a paused or sleeping Sailbox.
    pub async fn resume(&self) -> Result<(), SailError> {
        let handle = self.client.resume_sailbox(self.sailbox_id()).await?;
        self.set_exec_endpoint_hint(handle.exec_endpoint);
        Ok(())
    }

    /// Checkpoint the Sailbox. `options.name` labels the handle;
    /// `options.ttl`, when given, must be positive and overrides the server's
    /// default retention.
    pub async fn checkpoint(
        &self,
        options: CheckpointOptions,
    ) -> Result<SailboxCheckpoint, SailError> {
        self.client
            .checkpoint_sailbox(
                self.sailbox_id(),
                options.name.as_deref(),
                options.ttl.map(crate::client::duration_to_whole_seconds),
            )
            .await
    }

    /// Upgrade the Sailbox runtime (now if running, else at next wake).
    pub async fn upgrade(&self) -> Result<UpgradeResult, SailError> {
        self.client.upgrade_sailbox(self.sailbox_id()).await
    }

    /// Fork this Sailbox into a new running child in one call. The child copies
    /// this Sailbox's memory and writable disk as they are now, so it branches
    /// from the parent's current state, and the parent is left as it was. The
    /// copy is transient: there is no separate artifact to keep or reuse. To
    /// branch from a saved point in time instead, take a durable
    /// [`Sailbox::checkpoint`] and start children from it with
    /// [`Client::create_from_checkpoint`], which works even after the parent is
    /// gone.
    ///
    /// Because the memory comes across, processes the parent was running carry
    /// on in the child. Commands started with [`Sailbox::exec`] stop in the
    /// child, though their writes up to the fork are kept, and one started with
    /// `background` keeps running there. Start the other execs the child needs.
    /// Sometimes the child comes up cold instead, with the disk intact and
    /// nothing running, and a child that mounts a volume always does. Volumes
    /// are mounted on the child at the same paths as on the parent, and they
    /// are the same volumes, so both Sailboxes read and write the same files.
    pub async fn fork(&self, options: ForkOptions) -> Result<Sailbox, SailError> {
        self.client
            .fork_sailbox(self.sailbox_id(), options.name.as_deref(), options.timeout)
            .await
    }

    // --- exec ---

    /// Run a command from an argv vector (no shell interpretation) and return
    /// a handle to the live process. Resumes (wakes) the Sailbox to reach it.
    /// The returned [`ExecProcess`] streams output, accepts stdin, and
    /// resolves the exit status; dropping it detaches without killing the
    /// command. The output pump spawns on the calling task's tokio runtime.
    pub async fn exec(
        &self,
        argv: impl IntoIterator<Item = impl Into<String>>,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        let exec_endpoint = self.exec_endpoint_hint();
        self.client
            .exec_at_endpoint(
                self.sailbox_id(),
                Some(&exec_endpoint),
                collect_argv(argv),
                options,
            )
            .await
    }

    /// Run a shell command via `/bin/sh -lc` (pipes, globs, and `$VAR`
    /// expansion work), honoring the `cwd`/`background` options. Use
    /// [`Sailbox::exec`] with an argv vector when arguments must reach the
    /// command verbatim. Otherwise behaves like [`Sailbox::exec`].
    pub async fn exec_shell(
        &self,
        command: &str,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        let exec_endpoint = self.exec_endpoint_hint();
        self.client
            .exec_shell_at_endpoint(self.sailbox_id(), Some(&exec_endpoint), command, options)
            .await
    }

    /// Run an argv command to completion and return its buffered
    /// [`ExecResult`]: a one-shot convenience over [`Sailbox::exec`] followed
    /// by [`ExecProcess::wait`]. A nonzero exit code reports through
    /// [`ExecResult::exit_code`], not an error, and an exceeded
    /// `options.timeout` reports through [`ExecResult::timed_out`]. Use
    /// [`Sailbox::exec`] to stream output or feed stdin.
    ///
    /// ```no_run
    /// # async fn demo() -> Result<(), sail::SailError> {
    /// # let client = sail::Client::from_env()?;
    /// let sb = client.sailbox("sb_abc123");
    /// let result = sb.run(["echo", "hello"], Default::default()).await?;
    /// assert_eq!(result.exit_code, 0);
    /// println!("{}", result.stdout);
    /// # Ok(())
    /// # }
    /// ```
    pub async fn run(
        &self,
        argv: impl IntoIterator<Item = impl Into<String>>,
        options: RunOptions,
    ) -> Result<ExecResult, SailError> {
        self.exec(argv, options.into_exec_options())
            .await?
            .wait()
            .await
    }

    /// Run a shell command (`/bin/sh -lc`) to completion and return its
    /// buffered [`ExecResult`], honoring `options.cwd`. Otherwise behaves
    /// like [`Sailbox::run`].
    pub async fn run_shell(
        &self,
        command: &str,
        options: RunOptions,
    ) -> Result<ExecResult, SailError> {
        self.exec_shell(command, options.into_exec_options())
            .await?
            .wait()
            .await
    }

    // --- files ---

    /// Filesystem operations on this Sailbox's guest: read and write files
    /// (buffered or streaming), and directory helpers.
    pub fn fs(&self) -> SailboxFs<'_> {
        SailboxFs { sailbox: self }
    }

    // --- listeners ---

    /// Expose a guest port at runtime. Re-exposing a port under the same
    /// protocol sets its allowlist to what you pass, so pass the whole list
    /// every time; passing an empty one clears the restriction and reopens the
    /// port. The returned [`Listener`] carries the resolved endpoint but
    /// an unknown route status: the expose response does not report
    /// reachability. Confirm with [`Sailbox::wait_for_listener`].
    pub async fn expose(
        &self,
        guest_port: u32,
        protocol: IngressProtocol,
        allowlist: &[String],
    ) -> Result<Listener, SailError> {
        self.client
            .expose_listener(self.sailbox_id(), guest_port, protocol, allowlist)
            .await
    }

    /// Remove a runtime ingress port.
    pub async fn unexpose(&self, guest_port: u32) -> Result<(), SailError> {
        self.client
            .unexpose_listener(self.sailbox_id(), guest_port)
            .await
    }

    /// List this Sailbox's listeners without waking it.
    pub async fn listeners(&self) -> Result<Vec<Listener>, SailError> {
        self.client.list_listeners(self.sailbox_id()).await
    }

    /// Fetch one listener by guest port without waking the Sailbox.
    pub async fn listener(&self, guest_port: u32) -> Result<Listener, SailError> {
        self.client
            .get_listener(self.sailbox_id(), guest_port)
            .await
    }

    /// Block until the listener on `guest_port` is reachable end to end
    /// (route active and its endpoint accepting) and return it. An HTTP
    /// listener is probed by URL, so success means the guest server answered;
    /// a TCP listener is ready once the guest sends bytes or holds the
    /// connection open. This is a connectivity check, not an application
    /// health check. Re-checks every second and fails with a timeout error
    /// after `options.timeout`.
    pub async fn wait_for_listener(
        &self,
        guest_port: u32,
        options: WaitForListenerOptions,
    ) -> Result<Listener, SailError> {
        self.client
            .wait_for_listener(self.sailbox_id(), guest_port, options.timeout)
            .await
    }

    /// Ingress-identity headers for this Sailbox, as name/value pairs.
    pub async fn ingress_auth_headers(&self) -> Result<Vec<(String, String)>, SailError> {
        self.client.ingress_auth_headers(self.sailbox_id()).await
    }

    // --- ssh ---

    /// Make the Sailbox reachable over SSH, returning the endpoint when
    /// `options.wait` is set (else `None`). Installs the org SSH CA as
    /// trusted, (re)starts `sshd`, confirms the CA-only daemon owns guest
    /// port 22, and only then exposes the port as TCP ingress, so a failed
    /// enable never leaves a non-CA daemon reachable. Idempotent. A non-empty
    /// `options.allowlist` restricts port 22 to those source addresses or
    /// ranges; when empty, a first enable is open to any source and a re-enable
    /// keeps an existing restriction.
    pub async fn enable_ssh(
        &self,
        options: EnableSshOptions,
    ) -> Result<Option<SshEndpoint>, SailError> {
        self.client
            .enable_ssh(
                self.sailbox_id(),
                &options.allowlist,
                options.wait,
                options.timeout,
            )
            .await
    }
}

/// Filesystem operations on a Sailbox's guest, reached via [`Sailbox::fs`]:
/// buffered and streaming reads and writes, plus directory helpers with
/// coreutils semantics (`mkdir -p`, `rm -rf`, `test -e`), documented per
/// method.
pub struct SailboxFs<'a> {
    sailbox: &'a Sailbox,
}

impl std::fmt::Debug for SailboxFs<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SailboxFs")
            .field("sailbox_id", &self.sailbox.sailbox_id())
            .finish()
    }
}

impl SailboxFs<'_> {
    /// Read a guest file into memory in one call.
    pub async fn read(&self, path: &str) -> Result<Vec<u8>, SailError> {
        self.sailbox
            .client
            .read_file(self.sailbox.sailbox_id(), path)
            .await
    }

    /// Write `data` to a guest file in one call.
    pub async fn write(
        &self,
        path: &str,
        data: &[u8],
        options: WriteOptions,
    ) -> Result<(), SailError> {
        self.sailbox
            .client
            .write_file(self.sailbox.sailbox_id(), path, data, options)
            .await
    }

    /// Open a streaming read of a guest file.
    pub async fn read_stream(&self, path: &str) -> Result<FileReader, SailError> {
        self.sailbox
            .client
            .read_stream(self.sailbox.sailbox_id(), path)
            .await
    }

    /// Open a streaming write to a guest file.
    pub async fn write_stream(
        &self,
        path: &str,
        options: WriteOptions,
    ) -> Result<FileWriter, SailError> {
        self.sailbox
            .client
            .write_stream(self.sailbox.sailbox_id(), path, options)
            .await
    }

    /// Create a directory and any missing parents (like `mkdir -p`); a no-op if
    /// it already exists.
    pub async fn mkdir(&self, path: &str) -> Result<(), SailError> {
        self.sailbox
            .client
            .make_dir(self.sailbox.sailbox_id(), path)
            .await
    }

    /// Remove a file or directory tree (like `rm -rf`); a no-op if it is already
    /// absent.
    pub async fn remove(&self, path: &str) -> Result<(), SailError> {
        self.sailbox
            .client
            .remove_path(self.sailbox.sailbox_id(), path)
            .await
    }

    /// Whether `path` exists in the guest. Follows symlinks (like `test -e`), so
    /// a dangling symlink reports `false` even though [`ls`](Self::ls) lists it.
    pub async fn exists(&self, path: &str) -> Result<bool, SailError> {
        self.sailbox
            .client
            .path_exists(self.sailbox.sailbox_id(), path)
            .await
    }

    /// List a directory's immediate entries as [`DirEntry`] records (no
    /// recursion). Runs GNU `find` in the guest, which the default Debian image
    /// ships. A missing path errors, as does a path that is not a directory and
    /// a listing too large for the exec output cap. An entry whose name is not
    /// valid UTF-8 fails the listing, since the path API cannot address it.
    pub async fn ls(&self, path: &str) -> Result<Vec<DirEntry>, SailError> {
        self.sailbox
            .client
            .list_dir(self.sailbox.sailbox_id(), path)
            .await
    }
}

impl Client {
    /// Bind an existing Sailbox id to this client without a network call,
    /// giving the method-style surface over it.
    pub fn sailbox(&self, sailbox_id: impl Into<String>) -> Sailbox {
        Sailbox::bind(
            self.clone(),
            SailboxHandle {
                sailbox_id: sailbox_id.into(),
                ..Default::default()
            },
        )
    }
}