edgestore_repl/
http_client.rs

1//! HTTP replication client implementing the `ReplicationProtocol` trait.
2//!
3//! Uses MessagePack (rmp-serde) for control messages (D07).
4//! Raw bytes for segment data transfer.
5//!
6//! All network errors are mapped to `EdgestoreError::ReplicationError`.
7
8use std::io::Read;
9
10use serde::{Deserialize, Serialize};
11
12use edgestore::replication::{ReplicationProtocol, SegmentRef};
13use edgestore::EdgestoreError;
14
15/// MessagePack wire struct for GET /merkle response.
16#[derive(Serialize, Deserialize)]
17struct MerkleResponse {
18    root: Vec<u8>,
19}
20
21/// MessagePack wire struct for one item in GET /segments response.
22#[derive(Serialize, Deserialize)]
23struct SegmentEntry {
24    segment_id: u64,
25    segment_hash: Vec<u8>,
26}
27
28/// HTTP replication client that implements `ReplicationProtocol` against an `HttpReplicationServer`.
29///
30/// Uses MessagePack (rmp-serde) for control messages and raw bytes for segment transfer.
31pub struct HttpReplicationClient {
32    base_url: String,
33}
34
35impl HttpReplicationClient {
36    /// Create a new client pointing at `base_url` (e.g. `"http://127.0.0.1:8900"`).
37    pub fn new(base_url: impl Into<String>) -> Self {
38        HttpReplicationClient {
39            base_url: base_url.into(),
40        }
41    }
42}
43
44impl ReplicationProtocol for HttpReplicationClient {
45    /// Fetch the remote peer's current Merkle root.
46    ///
47    /// Calls `GET {base_url}/merkle`, deserializes MessagePack body as `{root: Vec<u8>}`,
48    /// and converts to `[u8; 32]`.
49    fn merkle_root(&self) -> Result<[u8; 32], EdgestoreError> {
50        let url = format!("{}/merkle", self.base_url);
51        let response = ureq::get(&url)
52            .call()
53            .map_err(|e| EdgestoreError::ReplicationError(format!("GET /merkle: {}", e)))?;
54
55        let resp: MerkleResponse = rmp_serde::from_read(response.into_reader())
56            .map_err(|e| EdgestoreError::ReplicationError(format!("GET /merkle decode: {}", e)))?;
57
58        if resp.root.len() != 32 {
59            return Err(EdgestoreError::ReplicationError(format!(
60                "GET /merkle: expected 32-byte root, got {} bytes",
61                resp.root.len()
62            )));
63        }
64
65        let mut hash = [0u8; 32];
66        hash.copy_from_slice(&resp.root);
67        Ok(hash)
68    }
69
70    /// Fetch the remote peer's full segment manifest.
71    ///
72    /// Calls `GET {base_url}/segments`, deserializes MessagePack as a list of segment entries,
73    /// and converts each to `SegmentRef`.
74    fn list_segments(&self) -> Result<Vec<SegmentRef>, EdgestoreError> {
75        let url = format!("{}/segments", self.base_url);
76        let response = ureq::get(&url)
77            .call()
78            .map_err(|e| EdgestoreError::ReplicationError(format!("GET /segments: {}", e)))?;
79
80        let entries: Vec<SegmentEntry> =
81            rmp_serde::from_read(response.into_reader()).map_err(|e| {
82                EdgestoreError::ReplicationError(format!("GET /segments decode: {}", e))
83            })?;
84
85        let refs = entries
86            .into_iter()
87            .map(|e| {
88                let mut hash = [0u8; 32];
89                let copy_len = e.segment_hash.len().min(32);
90                hash[..copy_len].copy_from_slice(&e.segment_hash[..copy_len]);
91                SegmentRef {
92                    segment_hash: hash,
93                    segment_id: e.segment_id,
94                }
95            })
96            .collect();
97
98        Ok(refs)
99    }
100
101    /// Fetch one segment's raw bytes by content hash.
102    ///
103    /// Calls `GET {base_url}/segments/{hash_hex}`, reads raw bytes from body.
104    /// Caller MUST verify BLAKE3 before applying (T-04-01).
105    fn fetch_segment(&self, hash: &[u8; 32]) -> Result<Vec<u8>, EdgestoreError> {
106        let hash_hex: String = hash.iter().map(|b| format!("{:02x}", b)).collect();
107        let url = format!("{}/segments/{}", self.base_url, hash_hex);
108
109        let response = ureq::get(&url).call().map_err(|e| {
110            EdgestoreError::ReplicationError(format!("GET /segments/{}: {}", hash_hex, e))
111        })?;
112
113        let mut data = Vec::new();
114        response.into_reader().read_to_end(&mut data).map_err(|e| {
115            EdgestoreError::ReplicationError(format!("GET /segments/{} read body: {}", hash_hex, e))
116        })?;
117
118        Ok(data)
119    }
120}
edgestore_repl/http_client.rs

edgestore_repl/
http_client.rs
