Request
The payload of the channel request the server sends: the tag byte 1 and the ask.
[1][request JSON …]digest. The blob digest,<algorithm>:<hex>.
The channel request is defined by diverge-sdk-rs/src/provider/endpoints/containers/tools/run/server/channel_request/frame.rs:
//! What a server's channel request frame carries for a tool container
//! run.
use std::fmt;
use crate::wire::decode::Decode;
use crate::wire::encode::{Encode, Writer};
use crate::shared::containers::{daemon, fuse, oci, postgres, registration, write_bytes};
/// What a provider asks a caller for while a tool container runs.
///
/// A payload leads with one byte saying which, and the rest is that
/// variant's own bytes.
///
/// | tag | asks for |
/// |-----|----------|
/// | `0` | [`OciManifest`](Self::OciManifest) |
/// | `1` | [`OciBlob`](Self::OciBlob) |
/// | `2` | [`OciHas`](Self::OciHas) |
/// | `3` | [`Write`](Self::Write) |
/// | `4` | [`Postgres`](Self::Postgres) |
/// | `5` | [`Daemon`](Self::Daemon) |
/// | `6` | [`FuseRead`](Self::FuseRead) |
/// | `7` | [`FuseWrite`](Self::FuseWrite) |
/// | `8` | [`FuseList`](Self::FuseList) |
/// | `9` | [`FuseRemove`](Self::FuseRemove) |
/// | `10` | [`FuseRename`](Self::FuseRename) |
/// | `11` | [`FuseMkdir`](Self::FuseMkdir) |
/// | `12` | [`FuseStat`](Self::FuseStat) |
/// | `13` | [`FuseTruncate`](Self::FuseTruncate) |
/// | `14` | [`FuseSetattr`](Self::FuseSetattr) |
/// | `15` | [`Registration`](Self::Registration) |
///
/// Sixteen. Five are the provider's own asks — whether the caller
/// holds an image, its manifest and blobs, a write's content, the
/// container's registration, the last of them numbered after the rest
/// because it came last — and the rest are the CONTAINER's, relayed:
/// its database connections, its daemon connection, and the files the
/// caller mounted live. No tool call outward: a tool container makes
/// none, and only an agent container asks the caller's MCP servers;
/// and no dependencies in the registration, which a tool container
/// never declares.
#[derive(Debug, Clone, PartialEq)]
pub enum Frame<'a> {
/// A manifest of an image the caller holds, by digest. Tag `0`.
///
/// Opened when the provider takes the image from the caller, and
/// opened by the container RUNTIME's appetite rather
/// than the provider's: the provider's registry serves the pull,
/// and a manifest its store does not hold becomes one of these.
/// See [`oci`](crate::shared::containers::oci).
OciManifest(oci::manifest::request::Request),
/// A blob of an image the caller holds, by digest. Tag `1`.
///
/// The other half of a pull; see
/// [`oci`](crate::shared::containers::oci).
OciBlob(oci::blob::request::Request),
/// Whether the caller holds an image, by digest, and under what
/// name. Tag `2`.
///
/// Opened by a provider that would take the image from the caller,
/// before it asks for anything of it; see
/// [`oci`](crate::shared::containers::oci).
OciHas(oci::has::request::Request),
/// Send the content for a write. Tag `3`.
///
/// Opened in answer to a write the caller started. A write cannot
/// carry its own content — only a responder can finish a channel —
/// so the bytes travel as responses on this one. See
/// [`write_bytes`](crate::shared::containers::write_bytes).
Write(write_bytes::request::Request),
/// The provider's half of a database connection the container
/// opened. Tag `4`.
///
/// What comes back is everything the database says; the caller
/// opens the other half, or declines. See
/// [`postgres`](crate::shared::containers::postgres).
Postgres(postgres::request::Postgres),
/// A daemon connection the program opened, by the id the provider
/// minted. Tag `5`.
///
/// This half is the proxy's, relayed under an id of the provider's.
/// What comes back on it is everything the DAEMON says, as
/// [`daemon::server::Frame`](crate::shared::containers::daemon::server::Frame)s;
/// its finish is the connection closed, and an empty finish is the
/// connection declined. See
/// [`daemon`](crate::shared::containers::daemon) for the pair.
Daemon(daemon::request::Daemon),
/// Read a piece of a file the caller mounted live, by the mount's
/// id, the file's path in it, an offset and a length. Tag `6`.
///
/// The container's proxy asking on behalf of a FUSE mount: every
/// `read(2)` of the file. See
/// [`fuse`](crate::shared::containers::fuse).
FuseRead(fuse::read::request::Request<'a>),
/// Write a piece of a file the caller mounted live, in place at
/// an offset. Tag `7`.
///
/// Every `write(2)` of the file.
FuseWrite(fuse::write::request::Request<'a>),
/// List a directory of a tree the caller mounted live. Tag `8`.
///
/// Every listing in a directory mount: names and kinds.
FuseList(fuse::list::request::Request<'a>),
/// Remove a file or an empty directory of such a tree. Tag `9`.
FuseRemove(fuse::remove::request::Request<'a>),
/// Rename an entry within such a tree. Tag `10`.
FuseRename(fuse::rename::request::Request<'a>),
/// Make a directory in such a tree. Tag `11`.
FuseMkdir(fuse::mkdir::request::Request<'a>),
/// What an entry the caller mounted live is: kind, size, mode,
/// owner, group and times. Tag `12`.
///
/// Every attribute of a file mount, and every lookup and
/// attribute of an entry in a directory mount: a `stat` costs
/// fifty-seven bytes back, not the file.
FuseStat(fuse::stat::request::Request<'a>),
/// Set a file the caller mounted live to a length. Tag `13`.
///
/// Every `truncate(2)`, `ftruncate(2)` and `O_TRUNC` open.
FuseTruncate(fuse::truncate::request::Request<'a>),
/// Set some attributes of an entry the caller mounted live. Tag
/// `14`.
///
/// Every `chmod(2)`, `chown(2)` and `utimensat(2)`.
FuseSetattr(fuse::setattr::request::Request<'a>),
/// Register the container: resolve the secrets it takes. Tag
/// `15`.
///
/// Opened once, after the proxy's `Begun` carried a non-empty
/// declaration, and before the id; its dependencies are empty,
/// since a tool container declares none. Answered with the
/// secrets' first values, which the provider feeds to the
/// container before it sends the id. See
/// [`registration`](crate::shared::containers::registration).
Registration(registration::request::Request<'a>),
}
/// Tag for [`Frame::OciManifest`].
const OCI_MANIFEST: u8 = 0;
/// Tag for [`Frame::OciBlob`].
const OCI_BLOB: u8 = 1;
/// Tag for [`Frame::OciHas`].
const OCI_HAS: u8 = 2;
/// Tag for [`Frame::Write`].
const WRITE: u8 = 3;
/// Tag for [`Frame::Postgres`].
const POSTGRES: u8 = 4;
/// Tag for [`Frame::Daemon`].
const DAEMON: u8 = 5;
/// Tag for [`Frame::FuseRead`].
const FUSE_READ: u8 = 6;
/// Tag for [`Frame::FuseWrite`].
const FUSE_WRITE: u8 = 7;
/// Tag for [`Frame::FuseList`].
const FUSE_LIST: u8 = 8;
/// Tag for [`Frame::FuseRemove`].
const FUSE_REMOVE: u8 = 9;
/// Tag for [`Frame::FuseRename`].
const FUSE_RENAME: u8 = 10;
/// Tag for [`Frame::FuseMkdir`].
const FUSE_MKDIR: u8 = 11;
/// Tag for [`Frame::FuseStat`].
const FUSE_STAT: u8 = 12;
/// Tag for [`Frame::FuseTruncate`].
const FUSE_TRUNCATE: u8 = 13;
/// Tag for [`Frame::FuseSetattr`].
const FUSE_SETATTR: u8 = 14;
/// Tag for [`Frame::Registration`].
const REGISTRATION: u8 = 15;
impl Encode for Frame<'_> {
/// The JSON failure from the asks that are JSON, or a fuse id or
/// path too long for its prefix; everything else is bytes copied.
type Error = FrameEncodeError;
fn encode(&self, out: &mut Writer<'_>) -> Result<(), FrameEncodeError> {
match self {
Frame::OciManifest(request) => {
out.extend_from_slice(&[OCI_MANIFEST]);
request.encode(out).map_err(FrameEncodeError::Json)
}
Frame::OciBlob(request) => {
out.extend_from_slice(&[OCI_BLOB]);
request.encode(out).map_err(FrameEncodeError::Json)
}
Frame::OciHas(request) => {
out.extend_from_slice(&[OCI_HAS]);
request.encode(out).map_err(FrameEncodeError::Json)
}
Frame::Write(request) => {
out.extend_from_slice(&[WRITE]);
// Its error is `Infallible`, and an empty match on one
// is how you say so: there is no value to handle.
request.encode(out).map_err(|error| match error {})
}
Frame::Postgres(request) => {
out.extend_from_slice(&[POSTGRES]);
request.encode(out).map_err(|error| match error {})
}
Frame::Daemon(request) => {
out.extend_from_slice(&[DAEMON]);
request.encode(out).map_err(|error| match error {})
}
Frame::FuseRead(request) => {
out.extend_from_slice(&[FUSE_READ]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseWrite(request) => {
out.extend_from_slice(&[FUSE_WRITE]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseList(request) => {
out.extend_from_slice(&[FUSE_LIST]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseRemove(request) => {
out.extend_from_slice(&[FUSE_REMOVE]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseRename(request) => {
out.extend_from_slice(&[FUSE_RENAME]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseMkdir(request) => {
out.extend_from_slice(&[FUSE_MKDIR]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseStat(request) => {
out.extend_from_slice(&[FUSE_STAT]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseTruncate(request) => {
out.extend_from_slice(&[FUSE_TRUNCATE]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::FuseSetattr(request) => {
out.extend_from_slice(&[FUSE_SETATTR]);
request.encode(out).map_err(FrameEncodeError::Fuse)
}
Frame::Registration(request) => {
out.extend_from_slice(&[REGISTRATION]);
request.encode(out).map_err(FrameEncodeError::Json)
}
}
}
}
/// A tool container run channel request that could not be written.
#[derive(Debug)]
pub enum FrameEncodeError {
/// A JSON ask did not serialize.
Json(serde_json::Error),
/// A fuse ask would not encode: an id or a path too long for its
/// prefix.
Fuse(fuse::RequestEncodeError),
}
impl fmt::Display for FrameEncodeError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
FrameEncodeError::Json(error) => {
write!(f, "channel request did not serialize: {error}")
}
FrameEncodeError::Fuse(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for FrameEncodeError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
FrameEncodeError::Json(error) => Some(error),
FrameEncodeError::Fuse(error) => Some(error),
}
}
}
impl<'a> Decode<'a> for Frame<'a> {
/// Several ways to fail, and each names which ask it was.
type Error = FrameError;
fn decode(bytes: &'a [u8]) -> Result<Self, Self::Error> {
let (tag, rest) = bytes.split_first().ok_or(FrameError::Empty)?;
match *tag {
OCI_MANIFEST => oci::manifest::request::Request::decode(rest)
.map(Frame::OciManifest)
.map_err(FrameError::Oci),
OCI_BLOB => oci::blob::request::Request::decode(rest)
.map(Frame::OciBlob)
.map_err(FrameError::Oci),
OCI_HAS => oci::has::request::Request::decode(rest)
.map(Frame::OciHas)
.map_err(FrameError::Oci),
WRITE => write_bytes::request::Request::decode(rest)
.map(Frame::Write)
.map_err(FrameError::Write),
POSTGRES => postgres::request::Postgres::decode(rest)
.map(Frame::Postgres)
.map_err(FrameError::Postgres),
DAEMON => daemon::request::Daemon::decode(rest)
.map(Frame::Daemon)
.map_err(FrameError::Daemon),
FUSE_READ => fuse::read::request::Request::decode(rest)
.map(Frame::FuseRead)
.map_err(FrameError::Fuse),
FUSE_WRITE => fuse::write::request::Request::decode(rest)
.map(Frame::FuseWrite)
.map_err(FrameError::Fuse),
FUSE_LIST => fuse::list::request::Request::decode(rest)
.map(Frame::FuseList)
.map_err(FrameError::Fuse),
FUSE_REMOVE => fuse::remove::request::Request::decode(rest)
.map(Frame::FuseRemove)
.map_err(FrameError::Fuse),
FUSE_RENAME => fuse::rename::request::Request::decode(rest)
.map(Frame::FuseRename)
.map_err(FrameError::Fuse),
FUSE_MKDIR => fuse::mkdir::request::Request::decode(rest)
.map(Frame::FuseMkdir)
.map_err(FrameError::Fuse),
FUSE_STAT => fuse::stat::request::Request::decode(rest)
.map(Frame::FuseStat)
.map_err(FrameError::Fuse),
FUSE_TRUNCATE => fuse::truncate::request::Request::decode(rest)
.map(Frame::FuseTruncate)
.map_err(FrameError::Fuse),
FUSE_SETATTR => fuse::setattr::request::Request::decode(rest)
.map(Frame::FuseSetattr)
.map_err(FrameError::Fuse),
REGISTRATION => registration::request::Request::decode(rest)
.map(Frame::Registration)
.map_err(FrameError::Registration),
tag => Err(FrameError::UnknownTag(tag)),
}
}
}
/// A tool 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 sixteen.
UnknownTag(u8),
/// An image ask did not parse.
Oci(serde_json::Error),
/// The registration did not parse.
Registration(serde_json::Error),
/// The write content request did not decode.
Write(write_bytes::request::RequestError),
/// The connection id was not four bytes.
Postgres(postgres::request::PostgresError),
/// A fuse ask did not decode.
Fuse(fuse::RequestError),
/// The daemon connection's id was not four bytes.
Daemon(daemon::request::DaemonError),
}
impl fmt::Display for FrameError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
FrameError::Empty => {
f.write_str("tools run channel request frame is empty")
}
FrameError::UnknownTag(tag) => {
write!(f, "unknown tools run channel request tag {tag}")
}
FrameError::Oci(error) => {
write!(f, "image ask did not parse: {error}")
}
FrameError::Registration(error) => {
write!(f, "registration did not parse: {error}")
}
FrameError::Write(error) => {
write!(f, "write content request did not decode: {error}")
}
FrameError::Postgres(error) => write!(f, "{error}"),
FrameError::Fuse(error) => write!(f, "{error}"),
FrameError::Daemon(error) => write!(f, "{error}"),
}
}
}
impl std::error::Error for FrameError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
FrameError::Daemon(error) => Some(error),
FrameError::Oci(error) | FrameError::Registration(error) => Some(error),
FrameError::Write(error) => Some(error),
FrameError::Postgres(error) => Some(error),
FrameError::Fuse(error) => Some(error),
FrameError::Empty | FrameError::UnknownTag(_) => None,
}
}
}and by diverge-sdk-rs/src/shared/containers/oci/blob/request/request.rs:
//! Asking for a blob by its digest.
use serde::{Deserialize, Serialize};
use serde_json::Error;
use crate::wire::decode::Decode;
use crate::wire::encode::{Encode, Writer};
/// The digest, and deliberately nothing else.
///
/// No offset and no length: the provider asks for the blob, and
/// what it does with the bytes — holds them, streams them through —
/// is its own, so long as what it serves is the digest's. No name,
/// because a blob is identified by its digest and a blob under one
/// name is the same blob under another.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct Request {
/// The blob's digest, `<algorithm>:<hex>`.
pub digest: String,
}
/// Its JSON, and nothing in front of it. The tag that says which
/// request this is belongs to whichever frame carries it.
impl Encode for Request {
/// The ordinary JSON failure.
type Error = Error;
fn encode(&self, out: &mut Writer<'_>) -> Result<(), Error> {
serde_json::to_writer(out, self)
}
}
impl Decode<'_> for Request {
/// The ordinary JSON failure.
type Error = Error;
fn decode(bytes: &[u8]) -> Result<Self, Error> {
serde_json::from_slice(bytes)
}
}