s3 range propagation (#23)

* fix

* fixes

* please our overload, the linter

* fix pr comment

Author: assafvayner

cas_client/src/remote_client.rs

@@ -4,18 +4,20 @@ use anyhow::anyhow;
 use bytes::Buf;
 use cas::key::Key;
 use cas_types::{QueryChunkResponse, QueryReconstructionResponse, UploadXorbResponse};
-use reqwest::{header::{HeaderMap, HeaderValue}, StatusCode, Url};
+use reqwest::{
+    header::{HeaderMap, HeaderValue},
+    StatusCode, Url,
+};
 use serde::{de::DeserializeOwned, Serialize};
 
 use bytes::Bytes;
 use cas_object::CasObject;
 use cas_types::CASReconstructionTerm;
-use tracing::warn;
+use tracing::{debug, warn};
 
 use crate::{error::Result, CasClientError};
 
 use merklehash::MerkleHash;
-use tracing::debug;
 
 use crate::Client;
 pub const CAS_ENDPOINT: &str = "http://localhost:8080";
@@ -83,8 +85,8 @@ impl Client for RemoteClient {
 
 impl RemoteClient {
     pub async fn from_config(endpoint: String, token: Option<String>) -> Self {
-        Self { 
-            client: CASAPIClient::new(&endpoint, token) 
+        Self {
+            client: CASAPIClient::new(&endpoint, token),
         }
     }
 }
@@ -108,7 +110,7 @@ impl CASAPIClient {
         Self {
             client,
             endpoint: endpoint.to_string(),
-            token
+            token,
         }
     }
 
@@ -170,7 +172,7 @@ impl CASAPIClient {
             &key.hash,
             contents,
             &chunk_boundaries.into_iter().map(|x| x as u32).collect(),
-            cas_object::CompressionScheme::LZ4
+            cas_object::CompressionScheme::LZ4,
         )?;
 
         debug!("Upload: POST to {url:?} for {key:?}");
@@ -224,14 +226,17 @@ impl CASAPIClient {
     /// Reconstruct the file
     async fn reconstruct_file(&self, file_id: &MerkleHash) -> Result<QueryReconstructionResponse> {
         let url = Url::parse(&format!(
-            "{}/reconstruction/{}", 
-            self.endpoint, 
+            "{}/reconstruction/{}",
+            self.endpoint,
             file_id.hex()
         ))?;
 
         let mut headers = HeaderMap::new();
         if let Some(tok) = &self.token {
-            headers.insert("Authorization", HeaderValue::from_str(&format!("Bearer {}", tok)).unwrap());
+            headers.insert(
+                "Authorization",
+                HeaderValue::from_str(&format!("Bearer {}", tok)).unwrap(),
+            );
         }
 
         let response = self.client.get(url).headers(headers).send().await?;
@@ -264,9 +269,21 @@ impl CASAPIClient {
 }
 
 async fn get_one(term: &CASReconstructionTerm) -> Result<Bytes> {
+    debug!("term: {term:?}");
+
+    if term.range.end < term.range.start || term.url_range.end < term.url_range.start {
+        return Err(CasClientError::InternalError(anyhow!(
+            "invalid range in reconstruction"
+        )));
+    }
+
     let url = Url::parse(term.url.as_str())?;
     let response = reqwest::Client::new()
         .request(hyper::Method::GET, url)
+        .header(
+            reqwest::header::RANGE,
+            format!("bytes={}-{}", term.url_range.start, term.url_range.end),
+        )
         .send()
         .await?
         .error_for_status()?;
@@ -275,11 +292,13 @@ async fn get_one(term: &CASReconstructionTerm) -> Result<Bytes> {
         .await
         .map_err(CasClientError::ReqwestError)?;
     let mut readseek = Cursor::new(xorb_bytes.to_vec());
+    let data = cas_object::deserialize_chunks(&mut readseek)?;
+    let len = (term.range.end - term.range.start) as usize;
+    let offset = term.range_start_offset as usize;
 
-    let cas_object = CasObject::deserialize(&mut readseek)?;
-    let data = cas_object.get_range(&mut readseek, term.range.start, term.range.end)?;
+    let sliced = data[offset..offset + len].to_vec();
 
-    Ok(Bytes::from(data))
+    Ok(Bytes::from(sliced))
 }
 
 #[cfg(test)]

cas_types/src/lib.rs

@@ -24,6 +24,7 @@ pub struct CASReconstructionTerm {
     pub range: Range,
     pub range_start_offset: u32,
     pub url: String,
+    pub url_range: Range,
 }
 
 #[derive(Debug, Serialize, Deserialize, Clone)]