datum_agent/dcp/
mod.rs

1//! Datum Control Protocol (DCP) server and client.
2//!
3//! DCP is a small prost protocol over length-prefixed TCP or QUIC streams. It is
4//! intentionally local to `datum-agent`: the registry remains the control-plane
5//! owner, while DCP is the management transport used by the CLI/TUI and cluster
6//! work. Unknown major versions are rejected during `Hello`; minor version
7//! bumps are additive request/payload extensions within the same major.
8
9pub mod client;
10pub mod proto;
11pub mod server;
12
13mod frame;
14
15use thiserror::Error;
16
17pub use client::{DcpClient, EventSubscription, MetricSubscription, ShardPipeClient};
18pub use proto::{
19    Auth, ClientKind, ClusterJobList, ClusterJobNode, ClusterJobStart, ClusterNodeError,
20    ClusterNodeInfo, ClusterNodeList, ClusterNodeStatus, ClusterPlacementHistory,
21    CompleteShardingAsk, DCP_PROTOCOL_MAJOR, DCP_PROTOCOL_VERSION, DrainJob, ForwardShardEnvelopes,
22    GetConfig, GetShardAllocations, Hello, JobStatusRequest, ListClusterJobs, ListJobs,
23    OpenShardPipe, PlacementSpec, PlacementStrategy, PutConfig, RememberClusterAssignment,
24    RememberShardAllocations, Request, Response, ResponseStatus, RestartJob, ShardAllocation,
25    ShardAllocationEntry, ShardAllocationRequest, ShardAllocationTable, ShardEnvelopeAck,
26    ShardEnvelopeBatchResult, ShardEnvelopeWire, ShardPipeFrame, StartJob, StopJob,
27    SubmitClusterJob, SubscribeEvents, SubscribeMetrics,
28};
29pub use server::{
30    ClusterViewProvider, DcpJobFactories, DcpQuicServerConfig, DcpServer, DcpServerConfig,
31    DcpServerHandle, DcpTcpServerConfig, ShardingViewProvider,
32};
33
34/// Result type used by DCP APIs.
35pub type DcpResult<T> = Result<T, DcpError>;
36
37/// Errors returned by DCP clients, servers, and frame codecs.
38#[derive(Debug, Error)]
39pub enum DcpError {
40    #[error("DCP connection closed")]
41    Closed,
42    #[error("DCP protocol error: {0}")]
43    Protocol(String),
44    #[error("DCP response {status:?}: {message}")]
45    Response {
46        status: ResponseStatus,
47        message: String,
48    },
49    #[error("DCP IO error: {0}")]
50    Io(#[from] std::io::Error),
51    #[error("DCP encode error: {0}")]
52    Encode(#[from] prost::EncodeError),
53    #[error("DCP decode error: {0}")]
54    Decode(#[from] prost::DecodeError),
55    #[error("DCP agent error: {0}")]
56    Agent(#[from] crate::AgentError),
57    #[error("DCP stream error: {0}")]
58    Stream(#[from] datum::StreamError),
59    #[error("DCP task failed: {0}")]
60    Join(#[from] tokio::task::JoinError),
61}
62
63impl DcpError {
64    pub(crate) fn response(status: ResponseStatus, message: impl Into<String>) -> Self {
65        Self::Response {
66            status,
67            message: message.into(),
68        }
69    }
70}
datum_agent/dcp/mod.rs

datum_agent/dcp/
mod.rs
