Revert "feat(data): allow uploading without known size"

This reverts commit 66af7d0.

Author: kszucs

data/src/bin/example.rs

@@ -91,7 +91,7 @@ async fn clean(mut reader: impl Read, mut writer: impl Write, size: u64) -> Resu
         FileUploadSession::new(TranslatorConfig::local_config(std::env::current_dir()?)?.into(), None).await?;
 
     let mut size_read = 0;
-    let mut handle = translator.start_clean(None, Some(size), None).await;
+    let mut handle = translator.start_clean(None, size, None).await;
 
     loop {
         let bytes = reader.read(&mut read_buf)?;

data/src/data_client.rs

@@ -249,7 +249,7 @@ pub async fn clean_bytes(
     bytes: Vec<u8>,
     name: Option<Arc<str>>,
 ) -> errors::Result<(XetFileInfo, DeduplicationMetrics)> {
-    let mut handle = processor.start_clean(name, Some(bytes.len() as u64), None).await;
+    let mut handle = processor.start_clean(name, bytes.len() as u64, None).await;
     handle.add_data(&bytes).await?;
     handle.finish().await
 }
@@ -270,11 +270,7 @@ pub async fn clean_file(
     let mut buffer = vec![0u8; u64::min(filesize, *xet_config().data.ingestion_block_size) as usize];
 
     let mut handle = processor
-        .start_clean(
-            Some(filename.as_ref().to_string_lossy().into()),
-            Some(filesize),
-            Sha256::from_hex(sha256.as_ref()).ok(),
-        )
+        .start_clean(Some(filename.as_ref().to_string_lossy().into()), filesize, Sha256::from_hex(sha256.as_ref()).ok())
         .await;
 
     loop {

data/src/file_upload_session.rs

@@ -260,14 +260,13 @@ impl FileUploadSession {
     pub async fn start_clean(
         self: &Arc<Self>,
         file_name: Option<Arc<str>>,
-        size: Option<u64>,
+        size: u64,
         sha256: Option<Sha256>,
     ) -> SingleFileCleaner {
-        // Get a new file id for the completion tracking.
-        // When size is unknown (None), register with 0 to disable progress assertions.
+        // Get a new file id for the completion tracking
         let file_id = self
             .completion_tracker
-            .register_new_file(file_name.clone().unwrap_or_default(), size.unwrap_or(0))
+            .register_new_file(file_name.clone().unwrap_or_default(), size)
             .await;
 
         SingleFileCleaner::new(file_name, file_id, sha256, self.clone())
@@ -572,7 +571,7 @@ mod tests {
             .unwrap();
 
         let mut cleaner = upload_session
-            .start_clean(Some("test".into()), Some(read_data.len() as u64), None)
+            .start_clean(Some("test".into()), read_data.len() as u64, None)
             .await;
 
         // Read blocks from the source file and hand them to the cleaning handle

data/src/streaming.rs

@@ -70,7 +70,7 @@ impl XetClient {
         &self,
         progress_updater: Option<Arc<dyn TrackingProgressUpdater>>,
         name: Option<Arc<str>>,
-        size: Option<u64>,
+        size: u64,
     ) -> errors::Result<XetWriter> {
         XetWriter::new(self.config.clone(), progress_updater, name, size).await
     }
@@ -91,12 +91,12 @@ impl XetWriter {
     /// Creates a new writer that will upload a single file.
     ///
     /// `size` is the total number of bytes that will be written and is used for
-    /// progress tracking. Pass `None` if the content length is not known upfront.
+    /// progress tracking. The caller must know the content length before writing.
     pub async fn new(
         config: Arc<TranslatorConfig>,
         progress_updater: Option<Arc<dyn TrackingProgressUpdater>>,
         name: Option<Arc<str>>,
-        size: Option<u64>,
+        size: u64,
     ) -> errors::Result<Self> {
         let session = FileUploadSession::new(config, progress_updater).await?;
         let handle = session.start_clean(name, size, None).await;
@@ -291,8 +291,8 @@ mod tests {
 
     /// Upload `content` via [`XetWriter`] and download it back via [`XetReader`],
     /// asserting the round-tripped bytes match the original.
-    async fn assert_roundtrip(client: &XetClient, content: &[u8], size: Option<u64>) {
-        let mut writer = client.write(None, None, size).await.unwrap();
+    async fn assert_roundtrip(client: &XetClient, content: &[u8]) {
+        let mut writer = client.write(None, None, content.len() as u64).await.unwrap();
         for chunk in content.chunks(4096) {
             writer.write(Bytes::copy_from_slice(chunk)).await.unwrap();
         }
@@ -326,25 +326,12 @@ mod tests {
         let client = XetClient::new(Some(endpoint), None, None, "test".into()).unwrap();
 
         // Empty.
-        assert_roundtrip(&client, &[], Some(0)).await;
+        assert_roundtrip(&client, &[]).await;
         // Small.
-        assert_roundtrip(&client, b"Hello, World!", Some(13)).await;
+        assert_roundtrip(&client, b"Hello, World!").await;
         // Large (1 MB).
         let large: Vec<u8> = (0..1_000_000).map(|i| (i % 256) as u8).collect();
-        assert_roundtrip(&client, &large, Some(large.len() as u64)).await;
-    }
-
-    #[tokio::test]
-    async fn test_roundtrip_unknown_size() {
-        let temp_dir = tempdir().unwrap();
-        let endpoint = format!("local://{}", temp_dir.path().display());
-        let client = XetClient::new(Some(endpoint), None, None, "test".into()).unwrap();
-
-        // Small without explicit size.
-        assert_roundtrip(&client, b"Hello, World!", None).await;
-        // Large (1 MB) without explicit size.
-        let large: Vec<u8> = (0..1_000_000).map(|i| (i % 256) as u8).collect();
-        assert_roundtrip(&client, &large, None).await;
+        assert_roundtrip(&client, &large).await;
     }
 
     #[tokio::test]
@@ -354,7 +341,7 @@ mod tests {
         let client = XetClient::new(Some(endpoint), None, None, "test".into()).unwrap();
 
         let content: Vec<u8> = (0..1_000_000).map(|i| (i % 256) as u8).collect();
-        let mut writer = client.write(None, None, Some(content.len() as u64)).await.unwrap();
+        let mut writer = client.write(None, None, content.len() as u64).await.unwrap();
         for chunk in content.chunks(4096) {
             writer.write(Bytes::copy_from_slice(chunk)).await.unwrap();
         }
@@ -372,7 +359,7 @@ mod tests {
         let endpoint = format!("local://{}", temp_dir.path().display());
         let client = XetClient::new(Some(endpoint), None, None, "test".into()).unwrap();
 
-        let mut writer = client.write(None, None, Some(5)).await.unwrap();
+        let mut writer = client.write(None, None, 5).await.unwrap();
         writer.write(Bytes::from_static(b"hello")).await.unwrap();
         let file_info = writer.close().await.unwrap();
 
@@ -429,7 +416,7 @@ mod tests {
         let endpoint = format!("local://{}", temp_dir.path().display());
         let client = XetClient::new(Some(endpoint), None, None, "test".into()).unwrap();
 
-        let mut writer = client.write(None, None, Some(100)).await.unwrap();
+        let mut writer = client.write(None, None, 100).await.unwrap();
         writer.write(Bytes::from_static(b"some data")).await.unwrap();
         writer.abort().await.unwrap();
 

data/tests/test_session_resume.rs

@@ -67,7 +67,7 @@ mod tests {
 
             // Feed it half the data, and checkpoint.
             let mut cleaner = file_upload_session
-                .start_clean(Some("data".into()), Some(data.len() as u64), None)
+                .start_clean(Some("data".into()), data.len() as u64, None)
                 .await;
             cleaner.add_data(&data[..half_n]).await.unwrap();
             cleaner.checkpoint().await.unwrap();
@@ -85,7 +85,7 @@ mod tests {
 
             // Feed it half the data, and checkpoint.
             let mut cleaner = file_upload_session
-                .start_clean(Some("data".into()), Some(data.len() as u64), None)
+                .start_clean(Some("data".into()), data.len() as u64, None)
                 .await;
 
             // Add all the data.  Roughly the first half should dedup.
@@ -140,7 +140,7 @@ mod tests {
 
             // Feed it half the data, and checkpoint.
             let mut cleaner = file_upload_session
-                .start_clean(Some("data".into()), Some(data.len() as u64), None)
+                .start_clean(Some("data".into()), data.len() as u64, None)
                 .await;
             cleaner.add_data(&data[..rn]).await.unwrap();
             cleaner.checkpoint().await.unwrap();
@@ -172,7 +172,7 @@ mod tests {
 
             // Feed it half the data, and checkpoint.
             let mut cleaner = file_upload_session
-                .start_clean(Some("data".into()), Some(data.len() as u64), None)
+                .start_clean(Some("data".into()), data.len() as u64, None)
                 .await;
 
             // Add all the data.  Roughly the first half should dedup.

progress_tracking/src/upload_tracking.rs

@@ -141,9 +141,7 @@ impl CompletionTrackerImpl {
             if dep.is_external {
                 // This is the freebie case, where we can just increment the progress.
                 file_entry.completed_bytes += dep.n_bytes;
-                if file_entry.total_bytes > 0 {
-                    debug_assert_le!(file_entry.completed_bytes, file_entry.total_bytes);
-                }
+                debug_assert_le!(file_entry.completed_bytes, file_entry.total_bytes);
 
                 let progress_update = ItemProgressUpdate {
                     item_name: file_entry.name.clone(),
@@ -165,9 +163,7 @@ impl CompletionTrackerImpl {
                 // If the entry has already been completed, then just mark this as completed.
                 if entry.is_completed {
                     file_entry.completed_bytes += dep.n_bytes;
-                    if file_entry.total_bytes > 0 {
-                        debug_assert_le!(file_entry.completed_bytes, file_entry.total_bytes);
-                    }
+                    debug_assert_le!(file_entry.completed_bytes, file_entry.total_bytes);
 
                     let progress_update = ItemProgressUpdate {
                         item_name: file_entry.name.clone(),
@@ -190,9 +186,7 @@ impl CompletionTrackerImpl {
         // Register that this much has been completed already
         self.total_bytes_completed += file_bytes_processed;
 
-        if self.total_bytes > 0 {
-            debug_assert_le!(self.total_bytes_completed, self.total_bytes);
-        }
+        debug_assert_le!(self.total_bytes_completed, self.total_bytes);
 
         // There may be a lot of per-file updates, but these don't actually count against the new byte total;
         // this is counted only using xorbs.
@@ -307,9 +301,7 @@ impl CompletionTrackerImpl {
         self.total_upload_bytes_completed += byte_completion_increment;
 
         self.total_bytes_completed += file_bytes_processed;
-        if self.total_bytes > 0 {
-            debug_assert_le!(self.total_bytes_completed, self.total_bytes);
-        }
+        debug_assert_le!(self.total_bytes_completed, self.total_bytes);
 
         ProgressUpdate {
             item_updates,
@@ -421,9 +413,7 @@ impl CompletionTrackerImpl {
         debug_assert_le!(self.total_upload_bytes_completed, self.total_upload_bytes);
 
         self.total_bytes_completed += file_bytes_processed;
-        if self.total_bytes > 0 {
-            debug_assert_le!(self.total_bytes_completed, self.total_bytes);
-        }
+        debug_assert_le!(self.total_bytes_completed, self.total_bytes);
 
         ProgressUpdate {
             item_updates,
@@ -467,14 +457,11 @@ impl CompletionTrackerImpl {
     fn assert_complete(&self) {
         // Check each file for completeness
         for (idx, file) in self.files.iter().enumerate() {
-            // Skip size check for files registered without a known size (total_bytes == 0).
-            if file.total_bytes > 0 {
-                assert_eq!(
-                    file.completed_bytes, file.total_bytes,
-                    "File #{} ({}) is not fully completed: {}/{} bytes",
-                    idx, file.name, file.completed_bytes, file.total_bytes
-                );
-            }
+            assert_eq!(
+                file.completed_bytes, file.total_bytes,
+                "File #{} ({}) is not fully completed: {}/{} bytes",
+                idx, file.name, file.completed_bytes, file.total_bytes
+            );
             assert!(
                 file.remaining_xorbs_parts.is_empty(),
                 "File #{} ({}) still has uncompleted xorb parts: {:?}",

progress_tracking/src/verification_wrapper.rs

@@ -51,14 +51,11 @@ impl ProgressUpdaterVerificationWrapper {
         let tr = self.tr.lock().await;
 
         for (item_name, data) in tr.items.iter() {
-            // Skip for files registered without a known size (total_count == 0).
-            if data.total_count > 0 {
-                assert_eq!(
-                    data.last_completed, data.total_count,
-                    "Item '{}' is not fully complete: {}/{}",
-                    item_name, data.last_completed, data.total_count
-                );
-            }
+            assert_eq!(
+                data.last_completed, data.total_count,
+                "Item '{}' is not fully complete: {}/{}",
+                item_name, data.last_completed, data.total_count
+            );
         }
 
         assert_eq!(tr.total_transfer_bytes_completed, tr.total_transfer_bytes);
@@ -100,16 +97,13 @@ impl TrackingProgressUpdater for ProgressUpdaterVerificationWrapper {
             );
 
             // 2) `completed_count` must not exceed `total_count`
-            // Skip for files registered without a known size (total_bytes == 0).
-            if up.total_bytes > 0 {
-                assert!(
-                    up.bytes_completed <= up.total_bytes,
-                    "Item '{}' completed_count {} exceeds total {}",
-                    up.item_name,
-                    up.bytes_completed,
-                    up.total_bytes
-                );
-            }
+            assert!(
+                up.bytes_completed <= up.total_bytes,
+                "Item '{}' completed_count {} exceeds total {}",
+                up.item_name,
+                up.bytes_completed,
+                up.total_bytes
+            );
 
             // 3) The increment must match the difference
             let expected_new = entry.last_completed + up.bytes_completion_increment;