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::{
18    ClusterEventSubscription, DcpClient, EventSubscription, MetricSubscription, ShardPipeClient,
19};
20pub use proto::{
21    Auth, ClientKind, ClusterEvent, ClusterJobList, ClusterJobNode, ClusterJobStart,
22    ClusterNodeError, ClusterNodeInfo, ClusterNodeList, ClusterNodeStatus, ClusterPlacementHistory,
23    CompleteShardingAsk, DCP_PROTOCOL_MAJOR, DCP_PROTOCOL_VERSION, DrainJob, ForwardShardEnvelopes,
24    GetConfig, GetShardAllocations, Hello, JobStatusRequest, ListClusterJobs, ListJobs,
25    OpenShardPipe, PlacementSpec, PlacementStrategy, PutConfig, RememberClusterAssignment,
26    RememberShardAllocations, Request, Response, ResponseStatus, RestartJob, ShardAllocation,
27    ShardAllocationEntry, ShardAllocationRequest, ShardAllocationTable, ShardEnvelopeAck,
28    ShardEnvelopeBatchResult, ShardEnvelopeWire, ShardPipeFrame, StartJob, StopJob,
29    SubmitClusterJob, SubscribeClusterEvents, SubscribeEvents, SubscribeMetrics,
30    UnsubscribeMetrics,
31};
32pub use server::{
33    ClusterViewProvider, DcpJobFactories, DcpQuicServerConfig, DcpServer, DcpServerConfig,
34    DcpServerHandle, DcpTcpServerConfig, ShardingViewProvider,
35};
36
37/// Result type used by DCP APIs.
38pub type DcpResult<T> = Result<T, DcpError>;
39
40/// Errors returned by DCP clients, servers, and frame codecs.
41#[derive(Debug, Error)]
42pub enum DcpError {
43    #[error("DCP connection closed")]
44    Closed,
45    #[error("DCP protocol error: {0}")]
46    Protocol(String),
47    #[error("DCP response {status:?}: {message}")]
48    Response {
49        status: ResponseStatus,
50        message: String,
51    },
52    #[error("DCP IO error: {0}")]
53    Io(#[from] std::io::Error),
54    #[error("DCP encode error: {0}")]
55    Encode(#[from] prost::EncodeError),
56    #[error("DCP decode error: {0}")]
57    Decode(#[from] prost::DecodeError),
58    #[error("DCP agent error: {0}")]
59    Agent(#[from] crate::AgentError),
60    #[error("DCP stream error: {0}")]
61    Stream(#[from] datum::StreamError),
62    #[error("DCP task failed: {0}")]
63    Join(#[from] tokio::task::JoinError),
64}
65
66impl DcpError {
67    pub(crate) fn response(status: ResponseStatus, message: impl Into<String>) -> Self {
68        Self::Response {
69            status,
70            message: message.into(),
71        }
72    }
73}
datum_agent/dcp/mod.rs

datum_agent/dcp/
mod.rs
