appcore_sync/sync/
transport.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: transport.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/06/02 13:08:16 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/06/04 11:51:31 by dnettoRaw
8//      ###########      S: 0.6.0
9// =============================================================================
10
11//! Transport contracts and the manual HTTP sync transport.
12
13use crate::sync::discovery::SyncPeerScheme;
14use crate::sync::error::{SyncError, SyncResult};
15use crate::sync::types::{HeartbeatMessage, PeerInfo, SyncMessage};
16use appcore_core::CoreIdentity;
17use appcore_transport::{
18    send, CancellationToken, HttpClientConfig, HttpHeader, HttpRequest, HttpTarget, TransportError,
19};
20use std::fmt;
21#[cfg(test)]
22use std::io::Read;
23#[cfg(test)]
24use std::net::TcpStream;
25
26const DEFAULT_TIMEOUT_MS: u64 = 5_000;
27const DEFAULT_MAX_RESPONSE_BYTES: usize = 64 * 1024;
28const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 1024 * 1024;
29
30/// Sync transport contract.
31pub trait SyncTransport {
32    /// Sends an operational heartbeat.
33    fn send_heartbeat(&mut self, heartbeat: HeartbeatMessage) -> SyncResult<()>;
34    /// Sends opaque payload bytes to a compatible peer.
35    fn send_payload(&mut self, peer: &PeerInfo, payload: Vec<u8>) -> SyncResult<()>;
36}
37
38// Plain HTTP does not replace TLS/mTLS. The v1 body binds the source identity,
39// but transport confidentiality and server authentication still require HTTPS
40// or an external secure tunnel.
41//
42// O parsing HTTP é feito de forma manual para manter o runtime minimalista e sem dependências pesadas,
43// mas exige limites rígidos de timeouts e tamanho de payload para evitar DoS por peers maliciosos ou lentos.
44#[derive(Clone, PartialEq, Eq)]
45/// Bounded blocking HTTP client for leader-to-follower sync batches.
46pub struct HttpSyncTransport {
47    host: String,
48    port: u16,
49    scheme: SyncPeerScheme,
50    auth_token: Option<String>,
51    timeout_ms: u64,
52    max_response_bytes: usize,
53    max_request_body_bytes: usize,
54    source_identity: Option<CoreIdentity>,
55    cancellation: CancellationToken,
56}
57
58impl fmt::Debug for HttpSyncTransport {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        f.debug_struct("HttpSyncTransport")
61            .field("host", &self.host)
62            .field("port", &self.port)
63            .field("scheme", &self.scheme)
64            .field("auth_configured", &self.auth_token.is_some())
65            .field("timeout_ms", &self.timeout_ms)
66            .field("max_response_bytes", &self.max_response_bytes)
67            .field("max_request_body_bytes", &self.max_request_body_bytes)
68            .field("source_identity", &self.source_identity)
69            .field("cancelled", &self.cancellation.is_cancelled())
70            .finish()
71    }
72}
73
74impl HttpSyncTransport {
75    /// Creates a plain-HTTP transport for `host` and `port`.
76    pub fn new(host: impl Into<String>, port: u16) -> Self {
77        Self {
78            host: host.into(),
79            port,
80            scheme: SyncPeerScheme::Http,
81            auth_token: None,
82            timeout_ms: DEFAULT_TIMEOUT_MS,
83            max_response_bytes: DEFAULT_MAX_RESPONSE_BYTES,
84            max_request_body_bytes: DEFAULT_MAX_REQUEST_BODY_BYTES,
85            source_identity: None,
86            cancellation: CancellationToken::new(),
87        }
88    }
89
90    /// Adds a bearer token without exposing it through `Debug`.
91    pub fn with_auth_token(mut self, token: impl Into<String>) -> Self {
92        self.auth_token = Some(token.into());
93        self
94    }
95
96    /// Selects the HTTP or HTTPS transport scheme.
97    pub fn with_scheme(mut self, scheme: SyncPeerScheme) -> Self {
98        self.scheme = scheme;
99        self
100    }
101
102    /// Selects HTTPS transport.
103    pub fn with_https(mut self) -> Self {
104        self.scheme = SyncPeerScheme::Https;
105        self
106    }
107
108    /// Sets connect, read, and write deadlines in milliseconds.
109    pub fn with_timeout_ms(mut self, timeout_ms: u64) -> Self {
110        self.timeout_ms = timeout_ms;
111        self
112    }
113
114    /// Sets the maximum number of response bytes read from a peer.
115    pub fn with_max_response_bytes(mut self, max_response_bytes: usize) -> Self {
116        self.max_response_bytes = max_response_bytes;
117        self
118    }
119
120    /// Sets the maximum encoded request-body size.
121    pub fn with_max_request_body_bytes(mut self, max_request_body_bytes: usize) -> Self {
122        self.max_request_body_bytes = max_request_body_bytes;
123        self
124    }
125
126    /// Uses the identity-aware `appcore.sync.v1` envelope for outbound batches.
127    pub fn with_source_identity(mut self, source_identity: CoreIdentity) -> Self {
128        self.source_identity = Some(source_identity);
129        self
130    }
131
132    /// Replaces the cooperative cancellation token used by transport I/O.
133    pub fn with_cancellation_token(mut self, cancellation: CancellationToken) -> Self {
134        self.cancellation = cancellation;
135        self
136    }
137
138    /// Cancels active and future transport operations.
139    pub fn cancel(&self) {
140        self.cancellation.cancel();
141    }
142
143    /// Reports whether cancellation was requested.
144    pub fn is_cancelled(&self) -> bool {
145        self.cancellation.is_cancelled()
146    }
147
148    pub(crate) fn cancellation_token(&self) -> &CancellationToken {
149        &self.cancellation
150    }
151
152    /// Posts a batch to `/v1/sync/events` and requires a 2xx response.
153    pub fn post_sync_events(&self, message: &SyncMessage) -> SyncResult<()> {
154        let identity = self
155            .source_identity
156            .as_ref()
157            .ok_or(SyncError::InvalidSyncMessage(
158                "local sync identity is not configured",
159            ))?;
160        let body = crate::sync::wire::encode_sync_envelope_v1(identity, message)?;
161        if body.len() > self.max_request_body_bytes {
162            return Err(SyncError::RequestBodyTooLarge {
163                size: body.len(),
164                max: self.max_request_body_bytes,
165            });
166        }
167        let target =
168            HttpTarget::parse(&self.base_url(), "/v1/sync/events").map_err(map_transport_error)?;
169        let mut request = HttpRequest::new("POST", body.into_bytes())
170            .map_err(map_transport_error)?
171            .with_header(
172                HttpHeader::new("Content-Type", "application/json").map_err(map_transport_error)?,
173            );
174        if let Some(token) = &self.auth_token {
175            request = request.with_header(
176                HttpHeader::sensitive("Authorization", format!("Bearer {token}"))
177                    .map_err(map_transport_error)?,
178            );
179        }
180        let response = send(
181            &target,
182            &request,
183            HttpClientConfig {
184                timeout_ms: self.timeout_ms,
185                max_request_bytes: self.max_request_body_bytes,
186                max_response_bytes: self.max_response_bytes,
187                max_header_bytes: self.max_response_bytes,
188            },
189            Some(&self.cancellation),
190        )
191        .map_err(map_transport_error)?;
192        if (200..300).contains(&response.status_code) {
193            return Ok(());
194        }
195        Err(SyncError::HttpStatus(response.status_code))
196    }
197
198    fn base_url(&self) -> String {
199        format!("{}://{}:{}", self.scheme.as_str(), self.host, self.port)
200    }
201}
202
203fn map_transport_error(error: TransportError) -> SyncError {
204    match error {
205        TransportError::Timeout => SyncError::TransportTimeout("read".to_string()),
206        TransportError::Dns(reason) => SyncError::DnsResolutionFailed(reason),
207        TransportError::Tls(reason) => SyncError::TlsFailed(reason),
208        TransportError::Cancelled => {
209            SyncError::TransportFailed("sync transport cancelled".to_string())
210        }
211        TransportError::ResponseTooLarge { max } => SyncError::ResponseTooLarge { max },
212        TransportError::RequestTooLarge { max } => SyncError::RequestBodyTooLarge {
213            size: max.saturating_add(1),
214            max,
215        },
216        TransportError::InvalidResponse(reason) if reason == "empty response" => {
217            SyncError::EmptyHttpResponse
218        }
219        other => SyncError::TransportFailed(other.to_string()),
220    }
221}
222
223/// Decodes a versioned v1 envelope and returns its replication message.
224pub fn decode_sync_message(input: &str) -> SyncResult<SyncMessage> {
225    crate::sync::wire::decode_sync_envelope(input).map(|envelope| envelope.message)
226}
227
228#[cfg(test)]
229pub(crate) fn read_http_request_body(stream: &mut TcpStream) -> SyncResult<String> {
230    let mut buffer = Vec::new();
231    let mut headers_end = None;
232    let mut content_length = 0usize;
233    loop {
234        let mut chunk = [0u8; 512];
235        let read = stream
236            .read(&mut chunk)
237            .map_err(|err| SyncError::TransportFailed(err.to_string()))?;
238        if read == 0 {
239            break;
240        }
241        buffer.extend_from_slice(&chunk[..read]);
242        if headers_end.is_none() {
243            headers_end = find_headers_end(&buffer);
244            if let Some(end) = headers_end {
245                content_length = parse_content_length(&buffer[..end])?;
246            }
247        }
248        if let Some(end) = headers_end {
249            let body_len = buffer.len().saturating_sub(end);
250            if body_len >= content_length {
251                let body = &buffer[end..end + content_length];
252                return String::from_utf8(body.to_vec())
253                    .map_err(|_| SyncError::TransportFailed("invalid request body".to_string()));
254            }
255        }
256    }
257    Err(SyncError::TransportFailed(
258        "incomplete HTTP request".to_string(),
259    ))
260}
261
262#[cfg(test)]
263fn find_headers_end(buffer: &[u8]) -> Option<usize> {
264    let marker = b"\r\n\r\n";
265    buffer
266        .windows(marker.len())
267        .position(|window| window == marker)
268        .map(|position| position + marker.len())
269}
270
271#[cfg(test)]
272fn parse_content_length(headers: &[u8]) -> SyncResult<usize> {
273    let as_text = String::from_utf8(headers.to_vec())
274        .map_err(|_| SyncError::TransportFailed("invalid HTTP headers".to_string()))?;
275    for line in as_text.lines() {
276        if let Some(value) = line.strip_prefix("Content-Length:") {
277            return value
278                .trim()
279                .parse::<usize>()
280                .map_err(|_| SyncError::TransportFailed("invalid content length".to_string()));
281        }
282    }
283    Err(SyncError::TransportFailed(
284        "missing content length".to_string(),
285    ))
286}
287
288#[cfg(test)]
289#[path = "transport_tests.rs"]
290mod tests;
appcore_sync/sync/transport.rs

appcore_sync/sync/
transport.rs
