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

appcore_sync/sync/
transport.rs
