sausage-vfs: review fixes (fallible growth, symlink flags, stream cancel, sized reads, timestamps, error kinds, linker shadowing)

API: Layer::stat(path, follow), Layer::readlink, Open::follow,
Stat::symlink and Entry::symlink. Process::idle documents the shutdown
order: abort, await the task, await idle, release locks.

vfs/src/streams.rs

@@ -26,18 +26,20 @@ impl Reader {
             error: None,
         }
     }
-    fn start(&mut self) {
+    fn start(&mut self, size: usize) {
         if self.pending.is_none() && self.data.is_empty() && !self.eof {
             let f = self.file.clone();
             let offset = self.offset;
-            self.pending = Some(Box::pin(async move { f.read(offset, 65536).await }));
+            self.pending = Some(Box::pin(
+                async move { f.read(offset, size.min(65536)).await },
+            ));
         }
     }
 }
 #[async_trait]
 impl Pollable for Reader {
     async fn ready(&mut self) {
-        self.start();
+        self.start(4096);
         if let Some(task) = &mut self.pending {
             match task.await {
                 Ok(data) => {
@@ -53,6 +55,14 @@ impl Pollable for Reader {
 }
 #[async_trait]
 impl InputStream for Reader {
+    async fn blocking_read(&mut self, size: usize) -> StreamResult<Bytes> {
+        if size == 0 {
+            return Ok(Bytes::new());
+        }
+        self.start(size);
+        self.ready().await;
+        self.read(size)
+    }
     fn read(&mut self, size: usize) -> StreamResult<Bytes> {
         if let Some(e) = self.error.take() {
             return Err(failure(e));
@@ -62,7 +72,9 @@ impl InputStream for Reader {
         }
         let n = size.min(self.data.len());
         let data = self.data.split_to(n);
-        self.start();
+        if data.is_empty() && size != 0 {
+            self.start(size);
+        }
         Ok(data)
     }
 }
@@ -100,6 +112,9 @@ impl Pollable for Writer {
 }
 #[async_trait]
 impl OutputStream for Writer {
+    async fn cancel(&mut self) {
+        self.ready().await;
+    }
     fn write(&mut self, bytes: Bytes) -> StreamResult<()> {
         if self.check_write()? == 0 {
             return Err(StreamError::trap("write without a permit"));
@@ -132,3 +147,38 @@ impl OutputStream for Writer {
         Ok(if self.pending.is_some() { 0 } else { 65536 })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    /// Both read APIs request only the bytes needed instead of caching later mutations.
+    #[tokio::test]
+    async fn small_reads() {
+        for blocking in [true, false] {
+            let memory = Memory::default();
+            let file = memory
+                .open(
+                    "/file",
+                    Open {
+                        write: true,
+                        create: true,
+                        ..Open::default()
+                    },
+                )
+                .await
+                .unwrap();
+            file.write(0, b"abcd".to_vec()).await.unwrap();
+            let mut reader = Reader::new(file.clone(), 0);
+            let first = if blocking {
+                reader.blocking_read(1).await.unwrap()
+            } else {
+                assert!(reader.read(1).unwrap().is_empty());
+                reader.ready().await;
+                reader.read(1).unwrap()
+            };
+            assert_eq!(first.as_ref(), b"a");
+            file.write(1, b"Z".to_vec()).await.unwrap();
+            assert_eq!(reader.blocking_read(1).await.unwrap().as_ref(), b"Z");
+        }
+    }
+}