1use serde::{Deserialize, Serialize};
8
9pub const KIND_RANGE: u8 = 1;
11
12pub const HEADER_LEN: usize = 25;
14
15pub const MAX_PAYLOAD: usize = 1024 * 1024;
17
18#[derive(Clone, Copy, Debug, PartialEq, Eq)]
20#[non_exhaustive]
21pub enum FrameError {
22 Truncated,
24 BadKind,
26 LengthMismatch,
28 EmptyRange,
30 OffsetOverflow,
32 TooLarge,
34}
35
36#[derive(Clone, Debug, PartialEq, Eq)]
38pub struct RangeFrame {
39 pub file_id: u32,
40 pub seq: u64,
41 pub offset: u64,
42 pub data: Vec<u8>,
43}
44
45impl RangeFrame {
46 pub fn split(file_id: u32, seq: u64, offset: u64, data: &[u8]) -> Vec<RangeFrame> {
48 data.chunks(MAX_PAYLOAD)
49 .enumerate()
50 .map(|(i, chunk)| RangeFrame {
51 file_id,
52 seq,
53 offset: offset + (i * MAX_PAYLOAD) as u64,
54 data: chunk.to_vec(),
55 })
56 .collect()
57 }
58
59 pub fn encode(&self) -> Vec<u8> {
61 let mut out = Vec::with_capacity(HEADER_LEN + self.data.len());
62 out.push(KIND_RANGE);
63 out.extend_from_slice(&self.file_id.to_le_bytes());
64 out.extend_from_slice(&self.seq.to_le_bytes());
65 out.extend_from_slice(&self.offset.to_le_bytes());
66 out.extend_from_slice(&(self.data.len() as u32).to_le_bytes());
67 out.extend_from_slice(&self.data);
68 out
69 }
70
71 pub fn decode(bytes: &[u8]) -> Result<RangeFrame, FrameError> {
72 if bytes.len() < HEADER_LEN {
73 return Err(FrameError::Truncated);
74 }
75 if bytes[0] != KIND_RANGE {
76 return Err(FrameError::BadKind);
77 }
78 let file_id = u32::from_le_bytes(bytes[1..5].try_into().unwrap());
79 let seq = u64::from_le_bytes(bytes[5..13].try_into().unwrap());
80 let offset = u64::from_le_bytes(bytes[13..21].try_into().unwrap());
81 let len = u32::from_le_bytes(bytes[21..25].try_into().unwrap());
82 if len as usize > MAX_PAYLOAD {
83 return Err(FrameError::TooLarge);
84 }
85 let data = &bytes[HEADER_LEN..];
86 if data.len() as u64 != len as u64 {
87 return Err(FrameError::LengthMismatch);
88 }
89 if len == 0 {
90 return Err(FrameError::EmptyRange);
91 }
92 if offset.checked_add(len as u64).is_none() {
93 return Err(FrameError::OffsetOverflow);
94 }
95 Ok(RangeFrame {
96 file_id,
97 seq,
98 offset,
99 data: data.to_vec(),
100 })
101 }
102}
103
104#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
106#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
107#[non_exhaustive]
108pub enum ErrorCode {
109 InvalidRequest,
110 Unauthenticated,
111 PermissionDenied,
112 NotFound,
113 RateLimited,
114 UnsupportedVersion,
115 InternalError,
116}
117
118impl ErrorCode {
119 pub const ALL: [ErrorCode; 7] = [
121 ErrorCode::InvalidRequest,
122 ErrorCode::Unauthenticated,
123 ErrorCode::PermissionDenied,
124 ErrorCode::NotFound,
125 ErrorCode::RateLimited,
126 ErrorCode::UnsupportedVersion,
127 ErrorCode::InternalError,
128 ];
129
130 pub fn as_str(&self) -> &'static str {
131 match self {
132 ErrorCode::InvalidRequest => "INVALID_REQUEST",
133 ErrorCode::Unauthenticated => "UNAUTHENTICATED",
134 ErrorCode::PermissionDenied => "PERMISSION_DENIED",
135 ErrorCode::NotFound => "NOT_FOUND",
136 ErrorCode::RateLimited => "RATE_LIMITED",
137 ErrorCode::UnsupportedVersion => "UNSUPPORTED_VERSION",
138 ErrorCode::InternalError => "INTERNAL_ERROR",
139 }
140 }
141}
142
143pub const PROTOCOL_VERSION: u32 = 1;
145
146pub const MAX_CLIENT_ID_LEN: usize = 128;
148
149pub const MAX_TENANT_ID_LEN: usize = 64;
151
152#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
155#[serde(tag = "method")]
156pub enum Credentials {
157 #[serde(rename = "api_key")]
159 ApiKey { key: String },
160 #[serde(rename = "jwt")]
162 Jwt { token: String },
163 #[serde(rename = "basic")]
165 Basic { username: String, password: String },
166 #[serde(rename = "custom")]
168 Custom { data: serde_json::Value },
169}
170
171impl std::fmt::Debug for Credentials {
172 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
173 match self {
174 Credentials::ApiKey { .. } => f
175 .debug_struct("ApiKey")
176 .field("key", &"<redacted>")
177 .finish(),
178 Credentials::Jwt { .. } => f.debug_struct("Jwt").field("token", &"<redacted>").finish(),
179 Credentials::Basic { username, .. } => f
180 .debug_struct("Basic")
181 .field("username", username)
182 .field("password", &"<redacted>")
183 .finish(),
184 Credentials::Custom { .. } => f
185 .debug_struct("Custom")
186 .field("data", &"<redacted>")
187 .finish(),
188 }
189 }
190}
191
192impl Credentials {
193 pub fn method(&self) -> &'static str {
195 match self {
196 Credentials::ApiKey { .. } => "api_key",
197 Credentials::Jwt { .. } => "jwt",
198 Credentials::Basic { .. } => "basic",
199 Credentials::Custom { .. } => "custom",
200 }
201 }
202}
203
204#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
211pub struct Authenticate {
212 pub request_id: String,
213 pub protocol_version: u32,
214 pub tenant_id: String,
215 #[serde(default, skip_serializing_if = "Option::is_none")]
216 pub client_version: Option<String>,
217 #[serde(default, skip_serializing_if = "Option::is_none")]
218 pub client_id: Option<String>,
219 #[serde(default, skip_serializing_if = "Option::is_none")]
220 pub dev_scopes: Option<Vec<String>>,
221 #[serde(flatten)]
222 pub credentials: Credentials,
223}
224
225impl Authenticate {
226 pub fn validate(&self) -> Result<(), &'static str> {
228 let t = &self.tenant_id;
229 if t.is_empty()
230 || t.len() > MAX_TENANT_ID_LEN
231 || !t
232 .chars()
233 .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
234 {
235 return Err("tenant_id must be 1 to 64 characters of letters, digits, _ and -");
236 }
237 if let Some(c) = &self.client_id {
238 if c.is_empty()
239 || c.len() > MAX_CLIENT_ID_LEN
240 || !c.chars().all(|ch| {
241 ch.is_ascii_alphanumeric() || matches!(ch, '_' | '-' | '.' | '@' | ':')
242 })
243 {
244 return Err(
245 "client_id must be 1 to 128 characters of letters, digits and _ - . @ :",
246 );
247 }
248 }
249 match &self.credentials {
250 Credentials::ApiKey { key } if key.is_empty() => Err("the api key must not be empty"),
251 Credentials::Jwt { token } if token.is_empty() => Err("the token must not be empty"),
252 Credentials::Basic { username, password }
253 if username.is_empty() || password.is_empty() =>
254 {
255 Err("username and password must not be empty")
256 }
257 _ => Ok(()),
258 }
259 }
260}
261
262#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
264#[serde(deny_unknown_fields)]
265pub struct AuthenticationResult {
266 pub request_id: String,
267 pub success: bool,
268 pub message: String,
269 pub server_version: String,
270 #[serde(default, skip_serializing_if = "Option::is_none")]
272 pub client_id: Option<String>,
273 #[serde(default, skip_serializing_if = "Option::is_none")]
275 pub permissions: Option<Vec<String>>,
276 #[serde(default, skip_serializing_if = "Option::is_none")]
277 pub expires_in_seconds: Option<u64>,
278}
279
280#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
282#[serde(deny_unknown_fields)]
283pub struct FlushAck {
284 pub request_id: String,
285 pub file_id: u32,
286 pub seq: u64,
287}
288
289#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
291#[serde(deny_unknown_fields)]
292pub struct Resync {
293 pub request_id: String,
294 pub file_id: u32,
295}
296
297#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
299#[serde(deny_unknown_fields)]
300pub struct ResyncDone {
301 pub request_id: String,
302 pub file_id: u32,
303 pub seq: u64,
304 pub size: u64,
305}
306
307#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
309#[serde(deny_unknown_fields)]
310pub struct ErrorReply {
311 pub code: ErrorCode,
312 pub message: String,
313 #[serde(default, skip_serializing_if = "Option::is_none")]
314 pub request_id: Option<String>,
315}
316
317#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
319#[serde(deny_unknown_fields)]
320pub struct GetServiceFilePath {
321 pub request_id: String,
322 pub service_id: String,
323}
324
325#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
329#[serde(deny_unknown_fields)]
330pub struct ServiceFilePath {
331 pub request_id: String,
332 pub service_id: String,
333 pub file_id: u32,
334 #[serde(default, skip_serializing_if = "Option::is_none")]
335 pub path: Option<String>,
336 #[serde(default, skip_serializing_if = "Option::is_none")]
337 pub nonce_path: Option<String>,
338}
339
340#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
342#[serde(deny_unknown_fields)]
343pub struct ConfirmLocal {
344 pub request_id: String,
345 pub file_id: u32,
346 pub nonce: String,
347}
348
349impl std::fmt::Debug for ConfirmLocal {
350 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
352 f.debug_struct("ConfirmLocal")
353 .field("request_id", &self.request_id)
354 .field("file_id", &self.file_id)
355 .field("nonce", &"<redacted>")
356 .finish()
357 }
358}
359
360#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
363#[serde(deny_unknown_fields)]
364pub struct ConfirmLocalAck {
365 pub request_id: String,
366 pub file_id: u32,
367}
368
369#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
371#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
372#[non_exhaustive]
373pub enum RotateReason {
374 Resize,
376 Revocation,
378}
379
380#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
383#[serde(deny_unknown_fields)]
384pub struct RotateRequest {
385 pub request_id: String,
386 pub service_id: String,
387 pub reason: RotateReason,
388 #[serde(default, skip_serializing_if = "Option::is_none")]
389 pub new_size: Option<u64>,
390 #[serde(default, skip_serializing_if = "Option::is_none")]
391 pub revoke_client_id: Option<String>,
392}
393
394impl RotateRequest {
395 pub fn validate(&self) -> Result<(), &'static str> {
397 match self.reason {
398 RotateReason::Resize => match (self.new_size, &self.revoke_client_id) {
399 (Some(n), None) if n > 0 => Ok(()),
400 (Some(0), _) => Err("new_size must be at least 1"),
401 (None, _) => Err("RESIZE needs new_size"),
402 (_, Some(_)) => Err("RESIZE must not name a client"),
403 _ => Err("invalid RESIZE request"),
404 },
405 RotateReason::Revocation => match (self.new_size, &self.revoke_client_id) {
406 (None, Some(c)) if !c.is_empty() => Ok(()),
407 (_, None) => Err("REVOCATION needs revoke_client_id"),
408 (Some(_), _) => Err("REVOCATION must not set new_size"),
409 _ => Err("invalid REVOCATION request"),
410 },
411 }
412 }
413}
414
415#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
417#[serde(deny_unknown_fields)]
418pub struct RotationAccepted {
419 pub request_id: String,
420 pub rotation_id: String,
421}
422
423#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
429#[serde(deny_unknown_fields)]
430pub struct RotationPending {
431 pub rotation_id: String,
432 pub service_id: String,
433 pub file_id: u32,
434 pub deadline_ms: u64,
435 pub fill_ms: u64,
436}
437
438#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
443#[serde(deny_unknown_fields)]
444pub struct RotationPrepared {
445 pub rotation_id: String,
446 pub service_id: String,
447 pub old_file_id: u32,
448 pub new_file_id: u32,
449 pub size: u64,
450 #[serde(default, skip_serializing_if = "Option::is_none")]
451 pub path: Option<String>,
452}
453
454#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
456#[serde(deny_unknown_fields)]
457pub struct RotationFilled {
458 pub request_id: String,
459 pub rotation_id: String,
460}
461
462#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
464#[serde(deny_unknown_fields)]
465pub struct RotationAbort {
466 pub request_id: String,
467 pub rotation_id: String,
468}
469
470#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
472#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
473#[non_exhaustive]
474pub enum AbortReason {
475 RequesterAbort,
477 FillTimeout,
479 ServerError,
481 #[serde(other)]
483 Unknown,
484}
485
486#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
489#[serde(deny_unknown_fields)]
490pub struct RotationAborted {
491 pub rotation_id: String,
492 pub service_id: String,
493 pub file_id: u32,
494 pub reason: AbortReason,
495}
496
497#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
499#[serde(deny_unknown_fields)]
500pub struct RotationDrained {
501 pub request_id: String,
502 pub rotation_id: String,
503 pub file_id: u32,
504 pub seq: u64,
505}
506
507#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
512#[serde(deny_unknown_fields)]
513pub struct RotationCommit {
514 pub rotation_id: String,
515 pub service_id: String,
516 pub reason: RotateReason,
517 pub old_file_id: u32,
518 pub new_file_id: u32,
519 pub size: u64,
520}
521
522#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
524#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
525#[non_exhaustive]
526pub enum LossKind {
527 RotationStraggler,
529 ResyncOverwrote,
531 Revoked,
533 #[serde(other)]
535 Unknown,
536}
537
538#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
540#[serde(deny_unknown_fields)]
541pub struct LossRange {
542 pub offset: u64,
543 pub len: u64,
544}
545
546#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
549#[serde(deny_unknown_fields)]
550pub struct Loss {
551 pub service_id: String,
552 pub file_id: u32,
553 pub kind: LossKind,
554 #[serde(default, skip_serializing_if = "Vec::is_empty")]
555 pub ranges: Vec<LossRange>,
556}
557
558#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
560#[serde(deny_unknown_fields)]
561pub struct ByteRange {
562 pub offset: u64,
563 pub len: u64,
564}
565
566pub const MAX_RANGES_PER_MESSAGE: usize = 1024;
569
570pub fn validate_ranges(ranges: &[ByteRange]) -> Result<(), &'static str> {
574 if ranges.is_empty() {
575 return Err("ranges must not be empty");
576 }
577 if ranges.len() > MAX_RANGES_PER_MESSAGE {
578 return Err("too many ranges in one message");
579 }
580 let mut prev_end = 0u64;
581 for r in ranges {
582 if r.len == 0 {
583 return Err("a range must cover at least one byte");
584 }
585 let Some(end) = r.offset.checked_add(r.len) else {
586 return Err("offset plus len overflows");
587 };
588 if r.offset < prev_end {
589 return Err("ranges must be sorted and must not overlap");
590 }
591 prev_end = end;
592 }
593 Ok(())
594}
595
596#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
600#[serde(deny_unknown_fields)]
601pub struct RangesChanged {
602 pub request_id: String,
603 pub file_id: u32,
604 pub ranges: Vec<ByteRange>,
605}
606
607#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
610#[serde(deny_unknown_fields)]
611pub struct RangesUpdated {
612 pub file_id: u32,
613 pub seq: u64,
614 pub ranges: Vec<ByteRange>,
615}
616
617#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
620#[serde(deny_unknown_fields)]
621pub struct FlushEnd {
622 pub request_id: String,
623 pub file_id: u32,
624}
625
626#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
630#[serde(deny_unknown_fields)]
631pub struct Grant {
632 pub request_id: String,
633 pub client_id: String,
634 #[serde(default, skip_serializing_if = "Option::is_none")]
635 pub service_id: Option<String>,
636 pub permissions: Vec<String>,
637}
638
639impl Grant {
640 pub fn validate(&self) -> Result<(), &'static str> {
642 validate_grant_fields(&self.client_id, Some(&self.permissions), true)
643 }
644}
645
646#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
650#[serde(deny_unknown_fields)]
651pub struct Revoke {
652 pub request_id: String,
653 pub client_id: String,
654 #[serde(default, skip_serializing_if = "Option::is_none")]
655 pub service_id: Option<String>,
656 #[serde(default, skip_serializing_if = "Vec::is_empty")]
657 pub permissions: Vec<String>,
658}
659
660impl Revoke {
661 pub fn validate(&self) -> Result<(), &'static str> {
663 validate_grant_fields(&self.client_id, Some(&self.permissions), false)
664 }
665}
666
667fn validate_grant_fields(
668 client_id: &str,
669 permissions: Option<&[String]>,
670 need_permissions: bool,
671) -> Result<(), &'static str> {
672 if client_id.trim().is_empty() {
673 return Err("client_id must not be empty");
674 }
675 if let Some(p) = permissions {
676 if need_permissions && p.is_empty() {
677 return Err("permissions must not be empty");
678 }
679 if p.iter().any(|name| name.trim().is_empty()) {
680 return Err("a permission name must not be blank");
681 }
682 }
683 Ok(())
684}
685
686#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
688#[serde(deny_unknown_fields)]
689pub struct GrantAck {
690 pub request_id: String,
691}
692
693#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
695#[serde(deny_unknown_fields)]
696pub struct ListGrants {
697 pub request_id: String,
698 #[serde(default, skip_serializing_if = "Option::is_none")]
699 pub client_id: Option<String>,
700 #[serde(default, skip_serializing_if = "Option::is_none")]
701 pub service_id: Option<String>,
702}
703
704#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
706#[serde(deny_unknown_fields)]
707pub struct GrantEntry {
708 pub client_id: String,
709 #[serde(default, skip_serializing_if = "Option::is_none")]
710 pub service_id: Option<String>,
711 pub permissions: Vec<String>,
712}
713
714#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
716#[serde(deny_unknown_fields)]
717pub struct Grants {
718 pub request_id: String,
719 pub entries: Vec<GrantEntry>,
720}
721
722#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
728#[serde(deny_unknown_fields)]
729pub struct SetLocalMapping {
730 pub request_id: String,
731 pub service_id: String,
732 pub enabled: bool,
733}
734
735#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
737#[serde(deny_unknown_fields)]
738pub struct LocalMappingAck {
739 pub request_id: String,
740 pub service_id: String,
741 pub enabled: bool,
742}
743
744#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
747#[serde(tag = "type", content = "data")]
748#[non_exhaustive]
749pub enum ControlMessage {
750 Authenticate(Authenticate),
751 AuthenticationResult(AuthenticationResult),
752 FlushAck(FlushAck),
753 Resync(Resync),
754 ResyncDone(ResyncDone),
755 Error(ErrorReply),
756 GetServiceFilePath(GetServiceFilePath),
757 ServiceFilePath(ServiceFilePath),
758 ConfirmLocal(ConfirmLocal),
759 ConfirmLocalAck(ConfirmLocalAck),
760 RotateRequest(RotateRequest),
761 RotationAccepted(RotationAccepted),
762 RotationPending(RotationPending),
763 RotationPrepared(RotationPrepared),
764 RotationFilled(RotationFilled),
765 RotationAbort(RotationAbort),
766 RotationAborted(RotationAborted),
767 RotationDrained(RotationDrained),
768 RotationCommit(RotationCommit),
769 Loss(Loss),
770 RangesChanged(RangesChanged),
771 RangesUpdated(RangesUpdated),
772 FlushEnd(FlushEnd),
773 Grant(Grant),
774 Revoke(Revoke),
775 GrantAck(GrantAck),
776 ListGrants(ListGrants),
777 Grants(Grants),
778 SetLocalMapping(SetLocalMapping),
779 LocalMappingAck(LocalMappingAck),
780}