use serde::{Deserialize, Serialize};
pub const KIND_RANGE: u8 = 1;
pub const HEADER_LEN: usize = 25;
pub const MAX_PAYLOAD: usize = 1024 * 1024;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
#[non_exhaustive]
pub enum FrameError {
Truncated,
BadKind,
LengthMismatch,
EmptyRange,
OffsetOverflow,
TooLarge,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct RangeFrame {
pub file_id: u32,
pub seq: u64,
pub offset: u64,
pub data: Vec<u8>,
}
impl RangeFrame {
pub fn split(file_id: u32, seq: u64, offset: u64, data: &[u8]) -> Vec<RangeFrame> {
data.chunks(MAX_PAYLOAD)
.enumerate()
.map(|(i, chunk)| RangeFrame {
file_id,
seq,
offset: offset + (i * MAX_PAYLOAD) as u64,
data: chunk.to_vec(),
})
.collect()
}
pub fn encode(&self) -> Vec<u8> {
let mut out = Vec::with_capacity(HEADER_LEN + self.data.len());
out.push(KIND_RANGE);
out.extend_from_slice(&self.file_id.to_le_bytes());
out.extend_from_slice(&self.seq.to_le_bytes());
out.extend_from_slice(&self.offset.to_le_bytes());
out.extend_from_slice(&(self.data.len() as u32).to_le_bytes());
out.extend_from_slice(&self.data);
out
}
pub fn decode(bytes: &[u8]) -> Result<RangeFrame, FrameError> {
if bytes.len() < HEADER_LEN {
return Err(FrameError::Truncated);
}
if bytes[0] != KIND_RANGE {
return Err(FrameError::BadKind);
}
let file_id = u32::from_le_bytes(bytes[1..5].try_into().unwrap());
let seq = u64::from_le_bytes(bytes[5..13].try_into().unwrap());
let offset = u64::from_le_bytes(bytes[13..21].try_into().unwrap());
let len = u32::from_le_bytes(bytes[21..25].try_into().unwrap());
if len as usize > MAX_PAYLOAD {
return Err(FrameError::TooLarge);
}
let data = &bytes[HEADER_LEN..];
if data.len() as u64 != len as u64 {
return Err(FrameError::LengthMismatch);
}
if len == 0 {
return Err(FrameError::EmptyRange);
}
if offset.checked_add(len as u64).is_none() {
return Err(FrameError::OffsetOverflow);
}
Ok(RangeFrame {
file_id,
seq,
offset,
data: data.to_vec(),
})
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
#[non_exhaustive]
pub enum ErrorCode {
InvalidRequest,
Unauthenticated,
PermissionDenied,
NotFound,
RateLimited,
UnsupportedVersion,
InternalError,
}
impl ErrorCode {
pub const ALL: [ErrorCode; 7] = [
ErrorCode::InvalidRequest,
ErrorCode::Unauthenticated,
ErrorCode::PermissionDenied,
ErrorCode::NotFound,
ErrorCode::RateLimited,
ErrorCode::UnsupportedVersion,
ErrorCode::InternalError,
];
pub fn as_str(&self) -> &'static str {
match self {
ErrorCode::InvalidRequest => "INVALID_REQUEST",
ErrorCode::Unauthenticated => "UNAUTHENTICATED",
ErrorCode::PermissionDenied => "PERMISSION_DENIED",
ErrorCode::NotFound => "NOT_FOUND",
ErrorCode::RateLimited => "RATE_LIMITED",
ErrorCode::UnsupportedVersion => "UNSUPPORTED_VERSION",
ErrorCode::InternalError => "INTERNAL_ERROR",
}
}
}
pub const PROTOCOL_VERSION: u32 = 1;
pub const MAX_CLIENT_ID_LEN: usize = 128;
pub const MAX_TENANT_ID_LEN: usize = 64;
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "method")]
pub enum Credentials {
#[serde(rename = "api_key")]
ApiKey { key: String },
#[serde(rename = "jwt")]
Jwt { token: String },
#[serde(rename = "basic")]
Basic { username: String, password: String },
#[serde(rename = "custom")]
Custom { data: serde_json::Value },
}
impl std::fmt::Debug for Credentials {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Credentials::ApiKey { .. } => f
.debug_struct("ApiKey")
.field("key", &"<redacted>")
.finish(),
Credentials::Jwt { .. } => f.debug_struct("Jwt").field("token", &"<redacted>").finish(),
Credentials::Basic { username, .. } => f
.debug_struct("Basic")
.field("username", username)
.field("password", &"<redacted>")
.finish(),
Credentials::Custom { .. } => f
.debug_struct("Custom")
.field("data", &"<redacted>")
.finish(),
}
}
}
impl Credentials {
pub fn method(&self) -> &'static str {
match self {
Credentials::ApiKey { .. } => "api_key",
Credentials::Jwt { .. } => "jwt",
Credentials::Basic { .. } => "basic",
Credentials::Custom { .. } => "custom",
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Authenticate {
pub request_id: String,
pub protocol_version: u32,
pub tenant_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub client_version: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub client_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub dev_scopes: Option<Vec<String>>,
#[serde(flatten)]
pub credentials: Credentials,
}
impl Authenticate {
pub fn validate(&self) -> Result<(), &'static str> {
let t = &self.tenant_id;
if t.is_empty()
|| t.len() > MAX_TENANT_ID_LEN
|| !t
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
{
return Err("tenant_id must be 1 to 64 characters of letters, digits, _ and -");
}
if let Some(c) = &self.client_id {
if c.is_empty()
|| c.len() > MAX_CLIENT_ID_LEN
|| !c.chars().all(|ch| {
ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.' | '@' | ':')
})
{
return Err(
"client_id must be 1 to 128 characters of letters, digits and _ - . @ :",
);
}
}
match &self.credentials {
Credentials::ApiKey { key } if key.is_empty() => Err("the api key must not be empty"),
Credentials::Jwt { token } if token.is_empty() => Err("the token must not be empty"),
Credentials::Basic { username, password }
if username.is_empty() || password.is_empty() =>
{
Err("username and password must not be empty")
}
_ => Ok(()),
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AuthenticationResult {
pub request_id: String,
pub success: bool,
pub message: String,
pub server_version: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub client_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub permissions: Option<Vec<String>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub expires_in_seconds: Option<u64>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FlushAck {
pub request_id: String,
pub file_id: u32,
pub seq: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Resync {
pub request_id: String,
pub file_id: u32,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ResyncDone {
pub request_id: String,
pub file_id: u32,
pub seq: u64,
pub size: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ErrorReply {
pub code: ErrorCode,
pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub request_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GetServiceFilePath {
pub request_id: String,
pub service_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ServiceFilePath {
pub request_id: String,
pub service_id: String,
pub file_id: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub path: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub nonce_path: Option<String>,
}
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConfirmLocal {
pub request_id: String,
pub file_id: u32,
pub nonce: String,
}
impl std::fmt::Debug for ConfirmLocal {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ConfirmLocal")
.field("request_id", &self.request_id)
.field("file_id", &self.file_id)
.field("nonce", &"<redacted>")
.finish()
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ConfirmLocalAck {
pub request_id: String,
pub file_id: u32,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
#[non_exhaustive]
pub enum RotateReason {
Resize,
Revocation,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotateRequest {
pub request_id: String,
pub service_id: String,
pub reason: RotateReason,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub new_size: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub revoke_client_id: Option<String>,
}
impl RotateRequest {
pub fn validate(&self) -> Result<(), &'static str> {
match self.reason {
RotateReason::Resize => match (self.new_size, &self.revoke_client_id) {
(Some(n), None) if n > 0 => Ok(()),
(Some(0), _) => Err("new_size must be at least 1"),
(None, _) => Err("RESIZE needs new_size"),
(_, Some(_)) => Err("RESIZE must not name a client"),
_ => Err("invalid RESIZE request"),
},
RotateReason::Revocation => match (self.new_size, &self.revoke_client_id) {
(None, Some(c)) if !c.is_empty() => Ok(()),
(_, None) => Err("REVOCATION needs revoke_client_id"),
(Some(_), _) => Err("REVOCATION must not set new_size"),
_ => Err("invalid REVOCATION request"),
},
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationAccepted {
pub request_id: String,
pub rotation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationPending {
pub rotation_id: String,
pub service_id: String,
pub file_id: u32,
pub deadline_ms: u64,
pub fill_ms: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationPrepared {
pub rotation_id: String,
pub service_id: String,
pub old_file_id: u32,
pub new_file_id: u32,
pub size: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub path: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationFilled {
pub request_id: String,
pub rotation_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationAbort {
pub request_id: String,
pub rotation_id: String,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
#[non_exhaustive]
pub enum AbortReason {
RequesterAbort,
FillTimeout,
ServerError,
#[serde(other)]
Unknown,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationAborted {
pub rotation_id: String,
pub service_id: String,
pub file_id: u32,
pub reason: AbortReason,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationDrained {
pub request_id: String,
pub rotation_id: String,
pub file_id: u32,
pub seq: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RotationCommit {
pub rotation_id: String,
pub service_id: String,
pub reason: RotateReason,
pub old_file_id: u32,
pub new_file_id: u32,
pub generation: u64,
pub size: u64,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
#[non_exhaustive]
pub enum LossKind {
RotationStraggler,
ResyncOverwrote,
Revoked,
#[serde(other)]
Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LossRange {
pub offset: u64,
pub len: u64,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Loss {
pub service_id: String,
pub file_id: u32,
pub kind: LossKind,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub ranges: Vec<LossRange>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ByteRange {
pub offset: u64,
pub len: u64,
}
pub const MAX_RANGES_PER_MESSAGE: usize = 1024;
pub fn validate_ranges(ranges: &[ByteRange]) -> Result<(), &'static str> {
if ranges.is_empty() {
return Err("ranges must not be empty");
}
if ranges.len() > MAX_RANGES_PER_MESSAGE {
return Err("too many ranges in one message");
}
let mut prev_end = 0u64;
for r in ranges {
if r.len == 0 {
return Err("a range must cover at least one byte");
}
let Some(end) = r.offset.checked_add(r.len) else {
return Err("offset plus len overflows");
};
if r.offset < prev_end {
return Err("ranges must be sorted and must not overlap");
}
prev_end = end;
}
Ok(())
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RangesChanged {
pub request_id: String,
pub file_id: u32,
pub ranges: Vec<ByteRange>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RangesUpdated {
pub file_id: u32,
pub seq: u64,
pub ranges: Vec<ByteRange>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct FlushEnd {
pub request_id: String,
pub file_id: u32,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Grant {
pub request_id: String,
pub client_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub service_id: Option<String>,
pub permissions: Vec<String>,
}
impl Grant {
pub fn validate(&self) -> Result<(), &'static str> {
validate_grant_fields(&self.client_id, Some(&self.permissions), true)
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Revoke {
pub request_id: String,
pub client_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub service_id: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub permissions: Vec<String>,
}
impl Revoke {
pub fn validate(&self) -> Result<(), &'static str> {
validate_grant_fields(&self.client_id, Some(&self.permissions), false)
}
}
fn validate_grant_fields(
client_id: &str,
permissions: Option<&[String]>,
need_permissions: bool,
) -> Result<(), &'static str> {
if client_id.trim().is_empty() {
return Err("client_id must not be empty");
}
if let Some(p) = permissions {
if need_permissions && p.is_empty() {
return Err("permissions must not be empty");
}
if p.iter().any(|name| name.trim().is_empty()) {
return Err("a permission name must not be blank");
}
}
Ok(())
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GrantAck {
pub request_id: String,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ListGrants {
pub request_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub client_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub service_id: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct GrantEntry {
pub client_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub service_id: Option<String>,
pub permissions: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Grants {
pub request_id: String,
pub entries: Vec<GrantEntry>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SetLocalMapping {
pub request_id: String,
pub service_id: String,
pub enabled: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LocalMappingAck {
pub request_id: String,
pub service_id: String,
pub enabled: bool,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", content = "data")]
#[non_exhaustive]
pub enum ControlMessage {
Authenticate(Authenticate),
AuthenticationResult(AuthenticationResult),
FlushAck(FlushAck),
Resync(Resync),
ResyncDone(ResyncDone),
Error(ErrorReply),
GetServiceFilePath(GetServiceFilePath),
ServiceFilePath(ServiceFilePath),
ConfirmLocal(ConfirmLocal),
ConfirmLocalAck(ConfirmLocalAck),
RotateRequest(RotateRequest),
RotationAccepted(RotationAccepted),
RotationPending(RotationPending),
RotationPrepared(RotationPrepared),
RotationFilled(RotationFilled),
RotationAbort(RotationAbort),
RotationAborted(RotationAborted),
RotationDrained(RotationDrained),
RotationCommit(RotationCommit),
Loss(Loss),
RangesChanged(RangesChanged),
RangesUpdated(RangesUpdated),
FlushEnd(FlushEnd),
Grant(Grant),
Revoke(Revoke),
GrantAck(GrantAck),
ListGrants(ListGrants),
Grants(Grants),
SetLocalMapping(SetLocalMapping),
LocalMappingAck(LocalMappingAck),
}