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

appcore_sync/sync/
transport.rs
