mkit_server/admin/
stream.rs

1//! Bounded Connect streaming with acceptance and terminal failure audits.
2use super::{BodyCapture, Engine, Headers, READ_PRESERVED_PATH, Response, auth::Verified};
3use crate::{BoxStream, Code, NamespaceStore, ServerError};
4use base64::{Engine as _, engine::general_purpose::STANDARD};
5use bytes::Bytes;
6use futures::stream;
7use serde_json::{Value, json};
8use std::sync::Arc;
9
10/// One verified canonical slice. No instance is stored in the admin ledger.
11pub struct PreservedPiece {
12    /// Canonical payload; bounded by the preservation piece size.
13    pub data: Bytes,
14    /// Exact requested offset of this slice.
15    pub offset: u64,
16    /// True only after the final piece passes current retention/ownership checks.
17    pub last: bool,
18}
19impl std::fmt::Debug for PreservedPiece {
20    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
21        f.debug_struct("PreservedPiece")
22            .field("length", &self.data.len())
23            .field("offset", &self.offset)
24            .field("last", &self.last)
25            .finish()
26    }
27}
28/// An admin response whose streaming bytes never enter replay storage.
29pub enum Reply {
30    /// Bounded unary JSON or a pre-stream Connect error.
31    Unary(Response),
32    /// Framed Connect JSON, including an end-stream success or error envelope.
33    Stream(BoxStream<'static, Result<Bytes, ServerError>>),
34}
35impl std::fmt::Debug for Reply {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        match self {
38            Self::Unary(response) => response.fmt(f),
39            Self::Stream(_) => f.write_str("AdminStream"),
40        }
41    }
42}
43fn frame(flags: u8, value: &Value) -> Result<Bytes, ServerError> {
44    let encoded = serde_json::to_vec(value)
45        .map_err(|_| ServerError::new(Code::Internal, "admin stream encoding failed"))?;
46    let length = u32::try_from(encoded.len())
47        .map_err(|_| ServerError::new(Code::Internal, "admin stream overflow"))?;
48    let mut out = Vec::with_capacity(encoded.len() + 5);
49    out.push(flags);
50    out.extend_from_slice(&length.to_be_bytes());
51    out.extend(encoded);
52    Ok(Bytes::from(out))
53}
54impl<S: NamespaceStore + 'static> Engine<S> {
55    /// Dispatch signed bytes and return a bounded verified stream for `ReadPreserved`.
56    /// Adapters must send streaming responses with `Cache-Control: no-store`.
57    pub async fn handle_streamed(
58        self: Arc<Self>,
59        path: &str,
60        headers: &Headers,
61        wire: &BodyCapture,
62        decoded: Option<Result<Vec<u8>, ServerError>>,
63        now_ms: i64,
64    ) -> Reply {
65        let verified = match self.config.verify(path, headers, wire, now_ms) {
66            Ok(verified) => verified,
67            Err(error) => return Reply::Unary(Response::error(&error)),
68        };
69        let result = match self
70            .dispatch(path, headers, wire, decoded, now_ms, true)
71            .await
72        {
73            Ok(response) => response,
74            Err(error) => return Reply::Unary(Response::error(&error)),
75        };
76        if path != READ_PRESERVED_PATH || result.status != 200 {
77            return Reply::Unary(result);
78        }
79        let Ok(descriptor) = serde_json::from_slice::<Value>(&result.body) else {
80            let error = ServerError::new(Code::DataLoss, "invalid read descriptor");
81            let result = self
82                .record_result(&verified, now_ms, Response::error(&error))
83                .await
84                .unwrap_or_else(|error| Response::error(&error));
85            return Reply::Unary(result);
86        };
87        // Each poll retains at most one verified piece and one encoded message.
88        Reply::Stream(Box::pin(stream::unfold(
89            (self, verified, descriptor, 0u8),
90            |(engine, verified, mut descriptor, stage)| async move {
91                if stage == 2 {
92                    return None;
93                }
94                if stage == 1 {
95                    return Some((
96                        frame(2, &json!({"metadata":{}})),
97                        (engine, verified, descriptor, 2),
98                    ));
99                }
100                let result = engine.next_piece(&descriptor).await;
101                let (message, next_stage) = match result {
102                    Ok(piece) => match piece.offset.checked_add(piece.data.len() as u64) {
103                        Some(next) => {
104                            descriptor["offset"] = json!(next.to_string());
105                            (
106                                frame(
107                                    0,
108                                    &json!({"data":STANDARD.encode(&piece.data), "offset":piece.offset.to_string(), "last":piece.last}),
109                                ),
110                                u8::from(piece.last),
111                            )
112                        }
113                        None => (
114                            engine
115                                .stream_error(
116                                    &verified,
117                                    &ServerError::new(Code::DataLoss, "invalid preserved offset"),
118                                )
119                                .await,
120                            2,
121                        ),
122                    },
123                    Err(error) => (engine.stream_error(&verified, &error).await, 2),
124                };
125                Some((message, (engine, verified, descriptor, next_stage)))
126            },
127        )))
128    }
129    async fn next_piece(&self, descriptor: &Value) -> Result<PreservedPiece, ServerError> {
130        let service = self
131            .operations
132            .as_ref()
133            .ok_or_else(|| ServerError::unavailable("preservation service unavailable"))?;
134        service.preserved_piece(descriptor).await
135    }
136    async fn stream_error(
137        &self,
138        verified: &Verified,
139        error: &ServerError,
140    ) -> Result<Bytes, ServerError> {
141        self.record_result(
142            verified,
143            self.operations
144                .as_ref()
145                .ok_or_else(|| ServerError::unavailable("preservation service unavailable"))?
146                .preserved_now_ms()?,
147            Response::error(error),
148        )
149        .await?;
150        frame(
151            2,
152            &json!({"metadata":{}, "error":{"code":error.code().as_str(),"message":error.public_message()}}),
153        )
154    }
155}
mkit_server/admin/stream.rs

mkit_server/admin/
stream.rs
