Client Channels

The channels the client opens on the scope, by tag; the server serves each as its section states and answers a channel request it cannot decode by a channel response finish with nothing before it.

The client opens each of these channels on the scope, as Channels provides. The payload of the channel request begins with one tag byte designating the channel; the remainder is the channel’s request. The server serves each as its section states, and answers a channel request that does not decode by a channel response finish that no channel response precedes.

Tag	Channel
`0`	stop
`1`	filetree
`2`	read
`3`	write
`4`	postgres
`5`	run-loop
`6`	agent-schema
`7`	enqueue
`8`	dequeue

The channel request is defined by diverge-provider-sdk/src/endpoints/containers/agents/run/client/channel_request/frame.rs:

//! What a client's channel request frame carries for an agent container run.

use std::error::Error;
use std::fmt;

use crate::decode::Decode;
use crate::encode::{Encode, Writer};
use crate::shared::containers::{enqueue, run_loop};
use crate::shared::containers::{postgres, read, write_path};

/// What a caller asks a provider for while an agent container runs.
///
/// A payload leads with one byte saying which, and the rest is that
/// variant's own bytes.
///
/// | tag | asks for |
/// |-----|----------|
/// | `0` | [`Stop`](Self::Stop) |
/// | `1` | [`Filetree`](Self::Filetree) |
/// | `2` | [`Read`](Self::Read) |
/// | `3` | [`Write`](Self::Write) |
/// | `4` | [`Postgres`](Self::Postgres) |
/// | `5` | [`RunLoop`](Self::RunLoop) |
/// | `6` | [`AgentSchema`](Self::AgentSchema) |
/// | `7` | [`Enqueue`](Self::Enqueue) |
/// | `8` | [`Dequeue`](Self::Dequeue) |
///
/// The first five are the same in every container scope, in the same
/// order, so a reader of one is a reader of all; what follows is this
/// family's own exchange. All of them but the first reach INTO the
/// container, which is the thing a caller cannot dial: it runs on the
/// provider. That is the whole reason these channels open outward
/// from the client rather than the other way.
#[derive(Debug, Clone, PartialEq)]
pub enum Frame {
    /// Stop the container. Tag `0`.
    ///
    /// # It has no answer, and does not need one
    ///
    /// Nothing comes back on this channel. What comes back is the end
    /// of the SCOPE — a
    /// [`ResponseFinish`](crate::frame::server::ServerFrame::ResponseFinish),
    /// which already means nothing bearing this scope follows on any
    /// channel. Finishing this one first would be a smaller way of
    /// saying the same thing, moments earlier.
    ///
    /// # What it adds over closing the connection
    ///
    /// The scope IS the container's life, so dropping the connection
    /// stops it too. The difference is that a provider cannot tell a
    /// deliberate exit from a network that stopped answering, and has
    /// to wait to find out. This is unambiguous and immediate: a
    /// caller that says so is not gone, it is finished.
    ///
    /// # What it does to everyone else
    ///
    /// Ends them. Connectors hold scopes on a container that no longer
    /// exists, so those scopes finish too — a connection cannot
    /// outlive the thing it joined.
    Stop,
    /// The container's filesystem, watched. Tag `1`.
    ///
    /// Carries nothing — the variant is bare — and the provider answers
    /// with a snapshot and then every change, for as long as the
    /// channel lives. See
    /// [`filetree`](crate::shared::containers::filetree).
    Filetree,
    /// One file, read out of the container. Tag `2`.
    ///
    /// See [`read`](crate::shared::containers::read) for why this is
    /// one file and never a directory.
    Read(read::request::Request),
    /// One file, written into the container. Tag `3`.
    ///
    /// Carries no content. The provider answers by opening a channel
    /// of its own asking for it — see
    /// [`write_path`](crate::shared::containers::write_path) for why
    /// it travels that direction, and
    /// [`write_bytes`](crate::shared::containers::write_bytes) for
    /// what comes back.
    Write(write_path::request::Request),
    /// The caller's half of a database connection. Tag `4`.
    ///
    /// Opened once the caller has taken the provider's half, quoting
    /// the same connection; what comes back is everything the
    /// container wrote. See
    /// [`postgres`](crate::shared::containers::postgres) for the pair.
    Postgres(postgres::request::Postgres),
    /// Run a loop. Tag `5`.
    ///
    /// The family's own exchange. Carries the prompt — the agent was
    /// on the request that made the container, and never changes —
    /// and answers with the loop's chunks. See
    /// [`run_loop`](crate::shared::containers::run_loop).
    RunLoop(run_loop::request::Request),
    /// What the agent may be. Tag `6`.
    ///
    /// Carries nothing — the variant is bare — and the provider answers
    /// with the JSON Schema of the agent value. See
    /// [`agent_schema`](crate::shared::containers::agent_schema).
    AgentSchema,
    /// A message for the running loop's queue. Tag `7`.
    ///
    /// Answered once — by an
    /// [`enqueue::response::Frame`](crate::shared::containers::enqueue::response::Frame)
    /// naming the message's fate, whenever that is known — and then
    /// the finish. See [`enqueue`](crate::shared::containers::enqueue).
    Enqueue(enqueue::request::Request),
    /// Withdraw every message still waiting in the queue. Tag `8`.
    ///
    /// Carries nothing — the variant is bare. Answered once — by a
    /// [`dequeue::response::Frame`](crate::shared::containers::dequeue::response::Frame)
    /// saying whether the queue held anything — and then the finish.
    /// Each message it withdraws is ALSO answered, on its own enqueue
    /// channel. See [`dequeue`](crate::shared::containers::dequeue).
    Dequeue,
}

/// Tag for [`Frame::Stop`].
const STOP: u8 = 0;

/// Tag for [`Frame::Filetree`].
const FILETREE: u8 = 1;

/// Tag for [`Frame::Read`].
const READ: u8 = 2;

/// Tag for [`Frame::Write`].
const WRITE: u8 = 3;

/// Tag for [`Frame::Postgres`].
const POSTGRES: u8 = 4;

/// Tag for [`Frame::RunLoop`].
const RUN_LOOP: u8 = 5;

/// Tag for [`Frame::AgentSchema`].
const AGENT_SCHEMA: u8 = 6;

/// Tag for [`Frame::Enqueue`].
const ENQUEUE: u8 = 7;

/// Tag for [`Frame::Dequeue`].
const DEQUEUE: u8 = 8;

impl Encode for Frame {
    /// The ordinary JSON failure, from whichever half has one.
    type Error = serde_json::Error;

    fn encode(&self, out: &mut Writer<'_>) -> Result<(), Self::Error> {
        match self {
            Frame::Stop => {
                out.extend_from_slice(&[STOP]);
                Ok(())
            }
            Frame::Filetree => {
                out.extend_from_slice(&[FILETREE]);
                Ok(())
            }
            Frame::Read(request) => {
                out.extend_from_slice(&[READ]);
                request.encode(out)
            }
            Frame::Write(request) => {
                out.extend_from_slice(&[WRITE]);
                request.encode(out)
            }
            Frame::Postgres(request) => {
                out.extend_from_slice(&[POSTGRES]);
                request.encode(out).map_err(|error| match error {})
            }
            Frame::RunLoop(request) => {
                out.extend_from_slice(&[RUN_LOOP]);
                request.encode(out)
            }
            Frame::AgentSchema => {
                out.extend_from_slice(&[AGENT_SCHEMA]);
                Ok(())
            }
            Frame::Enqueue(request) => {
                out.extend_from_slice(&[ENQUEUE]);
                request.encode(out)
            }
            Frame::Dequeue => {
                out.extend_from_slice(&[DEQUEUE]);
                Ok(())
            }
        }
    }
}

impl Decode<'_> for Frame {
    /// Several ways to fail, and the parses among them name which.
    type Error = FrameError;

    fn decode(bytes: &[u8]) -> Result<Self, Self::Error> {
        let (tag, rest) = bytes.split_first().ok_or(FrameError::Empty)?;
        match *tag {
            STOP => Ok(Frame::Stop),
            FILETREE => Ok(Frame::Filetree),
            READ => read::request::Request::decode(rest)
                .map(Frame::Read)
                .map_err(FrameError::Read),
            WRITE => write_path::request::Request::decode(rest)
                .map(Frame::Write)
                .map_err(FrameError::Write),
            POSTGRES => postgres::request::Postgres::decode(rest)
                .map(Frame::Postgres)
                .map_err(FrameError::Postgres),
            RUN_LOOP => run_loop::request::Request::decode(rest)
                .map(Frame::RunLoop)
                .map_err(FrameError::RunLoop),
            AGENT_SCHEMA => Ok(Frame::AgentSchema),
            ENQUEUE => enqueue::request::Request::decode(rest)
                .map(Frame::Enqueue)
                .map_err(FrameError::Enqueue),
            DEQUEUE => Ok(Frame::Dequeue),
            tag => Err(FrameError::UnknownTag(tag)),
        }
    }
}

/// An agent container run channel request that could not be read.
#[derive(Debug)]
pub enum FrameError {
    /// No bytes at all, so not even a tag.
    Empty,
    /// A tag that is none of this frame's.
    UnknownTag(u8),
    /// The read request did not parse.
    Read(serde_json::Error),
    /// The write request did not parse.
    Write(serde_json::Error),
    /// The connection id was not four bytes.
    Postgres(postgres::request::PostgresError),
    /// The loop's prompt did not parse as JSON.
    RunLoop(serde_json::Error),
    /// The enqueued message did not parse as JSON.
    Enqueue(serde_json::Error),
}

impl fmt::Display for FrameError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            FrameError::Empty => {
                f.write_str("agents run channel request frame is empty")
            }
            FrameError::UnknownTag(tag) => {
                write!(f, "unknown agents run channel request tag {tag}")
            }
            FrameError::Read(error) => {
                write!(f, "read request did not parse: {error}")
            }
            FrameError::Write(error) => {
                write!(f, "write request did not parse: {error}")
            }
            FrameError::Postgres(error) => write!(f, "{error}"),
            FrameError::RunLoop(error) => {
                write!(f, "run loop request did not parse: {error}")
            }
            FrameError::Enqueue(error) => {
                write!(f, "enqueue request did not parse: {error}")
            }
        }
    }
}

impl Error for FrameError {
    fn source(&self) -> Option<&(dyn Error + 'static)> {
        match self {
            FrameError::Read(error)
            | FrameError::Write(error)
            | FrameError::RunLoop(error)
            | FrameError::Enqueue(error) => Some(error),
            FrameError::Postgres(error) => Some(error),
            FrameError::Empty | FrameError::UnknownTag(_) => None,
        }
    }
}

In this section

Stop — Tag 0: the client ends the scope; the server sends nothing on the channel, stops the container, and sends the response finish of the scope.
Filetree — Tag 1: the client asks for the filesystem of the container, watched; the server opens the proxy’s /filesystem/tree with every mount left out and relays a snapshot and one channel response per change until the proxy closes.
Read — Tag 2: the client names one file of the container; the server opens the proxy’s /filesystem/read and relays the bytes as channel responses, an error last when the read fails, and the channel response finish.
Write — Tag 3: the client names a write by an id and a destination path; the server opens a write-bytes channel for the content, streams it into the proxy’s /filesystem/write, and answers exactly once, written or an error.
Postgres — Tag 4: the client opens its half of a database connection the server announced, quoting the connection id; the server sends everything the container’s driver wrote, and the channel response finish when the driver’s socket ends.
Run loop — Tag 5: the client sends a prompt; the server opens the proxy’s /agent/run and relays the chunks of the loop as channel responses, an error last when there was no loop or it did not finish, and the channel response finish.
Agent schema — Tag 6: the client asks what the agent value may be; the server opens the proxy’s /agent/schema and relays exactly one channel response, the JSON Schema or an error, and the channel response finish.
Enqueue — Tag 7: the client sends a message for the queue of the running loop; the server opens the proxy’s /agent/enqueue and relays exactly one channel response, the fate of the message, when the fate is known.
Dequeue — Tag 8: the client withdraws every message from the queue of the running loop; the server opens the proxy’s /agent/dequeue and relays exactly one channel response stating whether the queue held anything.