Write blobs

Signed-off-by: itowlson <[email protected]>

Author: itowlson

Cargo.lock

@@ -147,7 +147,7 @@ sha2 = "0.10"
 tempfile = "3"
 test-environment = { git = "https://github.com/fermyon/conformance-tests", rev = "ecd22a45bcc5c775a56c67689a89aa4005866ac0" }
 thiserror = "1"
-tokio = "1"
+tokio = "1.40"
 toml = "0.8"
 tracing = { version = "0.1", features = ["log"] }
 tracing-opentelemetry = { version = "0.28", default-features = false, features = ["metrics"] }

Cargo.toml

@@ -20,6 +20,7 @@ spin-factor-blobstore = { path = "../factor-blobstore" }
 tokio = { workspace = true }
 tokio-stream = "0.1.16"
 tokio-util = "0.7.12"
+uuid = { version = "1.0", features = ["v4"] }
 wasmtime-wasi = { workspace = true }
 
 [lints]

crates/blobstore-azure/Cargo.toml

@@ -153,6 +153,62 @@ impl Container for AzureBlobContainer {
         Ok(Box::new(AzureBlobIncomingData::new(client, range)))
     }
 
+    async fn connect_stm(&self, name: &str, mut stm: tokio::io::ReadHalf<tokio::io::SimplexStream>, finished_tx: tokio::sync::mpsc::Sender<()>) -> anyhow::Result<()> {
+        use tokio::io::AsyncReadExt;
+
+        // It seems like we can't construct a SeekableStream over a SimplexStream, which
+        // feels unfortunate.  I am not sure that the outgoing-value interface gives
+        // us a way to construct a len-able stream, because we don't know until finish
+        // time how much the guest is going to write to it. (We might be able to do resettable...
+        // but len-able...)  So for now we read it into a buffer and then zoosh that up in
+        // one go.
+        //
+        // We can kind of work around this by doing a series of Put Block calls followed by
+        // a Put Block List.  So we need to buffer only each block. But that still requires
+        // care as you are limited to 50_000 committed / 100_000 uncommitted blocks.
+
+        const APPROX_BLOCK_SIZE: usize = 2 * 1024 * 1024;
+
+        let client = self.client.blob_client(name);
+
+        tokio::spawn(async move {
+            let mut blocks = vec![];
+
+            'put_blocks: loop {
+                let mut bytes = Vec::with_capacity(APPROX_BLOCK_SIZE);  // 2MB buffer x 50k blocks per blob = 100GB.  WHICH SHOULD BE ENOUGH FOR ANYONE.
+                loop {
+                    let read = stm.read_buf(&mut bytes).await.unwrap();
+                    let len = bytes.len();
+
+                    if read == 0 {
+                        // end of stream - send the last block and go
+                        let id_bytes = uuid::Uuid::new_v4().as_bytes().to_vec();
+                        let block_id = azure_storage_blobs::prelude::BlockId::new(id_bytes);
+                        client.put_block(block_id.clone(), bytes).await.unwrap();
+                        blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(block_id));
+                        break 'put_blocks;
+                    }
+                    if len >= APPROX_BLOCK_SIZE {
+                        let id_bytes = uuid::Uuid::new_v4().as_bytes().to_vec();
+                        let block_id = azure_storage_blobs::prelude::BlockId::new(id_bytes);
+                        client.put_block(block_id.clone(), bytes).await.unwrap();
+                        blocks.push(azure_storage_blobs::blob::BlobBlockType::Uncommitted(block_id));
+                        break;
+                    }
+                }
+            }
+
+            let block_list = azure_storage_blobs::blob::BlockList {
+                blocks
+            };
+            client.put_block_list(block_list).await.unwrap();
+
+            finished_tx.send(()).await.expect("should sent finish tx");
+        });
+
+        Ok(())
+    }
+
     async fn list_objects(&self) -> anyhow::Result<Box<dyn spin_factor_blobstore::ObjectNames>> {
         let stm = self.client.list_blobs().into_stream();
         Ok(Box::new(AzureBlobBlobsList::new(stm)))
@@ -179,16 +235,13 @@ impl AzureBlobIncomingData {
         }
     }
 
-    fn consume_async_impl(&mut self) -> wasmtime_wasi::pipe::AsyncReadStream { // Box<dyn futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>>> {
+    fn consume_async_impl(&mut self) -> wasmtime_wasi::pipe::AsyncReadStream {
         use futures::TryStreamExt;
         use tokio_util::compat::FuturesAsyncReadCompatExt;
         let stm = self.consume_as_stream();
         let ar = stm.into_async_read();
         let arr = ar.compat();
         wasmtime_wasi::pipe::AsyncReadStream::new(arr)
-        // Box::new(stm)
-        // let async_read = stm.into_async_read();
-        // todo!()
     }
 
     fn consume_as_stream(&mut self) -> impl futures::stream::Stream<Item = Result<Vec<u8>, std::io::Error>> {

crates/blobstore-azure/src/store.rs

@@ -48,9 +48,7 @@ impl BlobStoreInMemory {
 
 /// The serialized runtime configuration for the in memory blob store.
 #[derive(Deserialize, Serialize)]
-pub struct MemoryBlobStoreRuntimeConfig {
-    ignored: Option<String>,
-}
+pub struct MemoryBlobStoreRuntimeConfig {}
 
 #[async_trait]
 impl spin_factor_blobstore::ContainerManager for BlobStoreInMemory {
@@ -142,6 +140,21 @@ impl spin_factor_blobstore::Container for InMemoryContainer {
 
         Ok(Box::new(InMemoryBlobContent { data }))
     }
+
+    async fn connect_stm(&self, name: &str, mut stm: tokio::io::ReadHalf<tokio::io::SimplexStream>, finished_tx: tokio::sync::mpsc::Sender<()>) -> anyhow::Result<()> {
+        use tokio::io::AsyncReadExt;
+        let name = name.to_owned();
+        let blobs = self.blobs.clone();
+        tokio::spawn(async move {
+            let mut bytes = Vec::with_capacity(1024);
+            stm.read_to_end(&mut bytes).await.unwrap();
+            let mut lock = blobs.write().await;
+            lock.insert(name, bytes);
+            finished_tx.send(()).await.expect("should sent finish tx");
+        });
+        Ok(())
+    }
+
     async fn list_objects(&self) -> anyhow::Result<Box<dyn spin_factor_blobstore::ObjectNames>> {
         let blobs = self.read().await;
         let names = blobs.keys().map(|k| k.to_string()).collect();
@@ -181,21 +194,21 @@ impl spin_factor_blobstore::ObjectNames for InMemoryBlobNames {
     async fn read(&mut self, len: u64) -> anyhow::Result<(Vec<String>, bool)> {
         let len = len.try_into().unwrap();
         if len > self.names.len() {
-            Ok((self.names.drain(..).collect(), false))
+            Ok((self.names.drain(..).collect(), true))
         } else {
             let taken = self.names.drain(0..len).collect();
-            Ok((taken, !self.names.is_empty()))
+            Ok((taken, self.names.is_empty()))
         }
     }
 
     async fn skip(&mut self, num: u64) -> anyhow::Result<(u64,bool)> {
         let len = num.try_into().unwrap();
-        let (count, more) = if len > self.names.len() {
-            (self.names.drain(..).len(), false)
+        let (count, at_end) = if len > self.names.len() {
+            (self.names.drain(..).len(), true)
         } else {
             let taken = self.names.drain(0..len).len();
-            (taken, !self.names.is_empty())
+            (taken, self.names.is_empty())
         };
-        Ok((count.try_into().unwrap(), more))
+        Ok((count.try_into().unwrap(), at_end))
     }
 }

crates/blobstore-memory/src/lib.rs

@@ -6,6 +6,7 @@ edition = { workspace = true }
 
 [dependencies]
 anyhow = { workspace = true }
+bytes = { workspace = true }
 futures = { workspace = true }
 lru = "0.12"
 serde = { workspace = true }
@@ -15,7 +16,7 @@ spin-factors = { path = "../factors" }
 spin-locked-app = { path = "../locked-app" }
 spin-resource-table = { path = "../table" }
 spin-world = { path = "../world" }
-tokio = { workspace = true, features = ["macros", "sync", "rt"] }
+tokio = { workspace = true, features = ["macros", "sync", "rt", "io-util"] }
 toml = { workspace = true }
 tracing = { workspace = true }
 wasmtime-wasi = { workspace = true }

crates/factor-blobstore/Cargo.toml

@@ -0,0 +1 @@
+pub mod write_stream;