feat(fileTransfer): add streaming to fifo

Signed-off-by: Joshua Chapman <joshua.chapman@secomind.com>

Author: joshuachp

.gitignore

@@ -28,6 +28,8 @@ target/
 
 # RustRover IDE
 .idea/
+# nvim
+.nvim.lua
 
 /telemetry.json
 

Cargo.lock

@@ -114,6 +114,7 @@ bytes.workspace = true
 cfg-if.workspace = true
 clap = { workspace = true, features = ["derive", "env"] }
 color-eyre.workspace = true
+dirs.workspace = true
 displaydoc.workspace = true
 edgehog-containers = { workspace = true, optional = true }
 edgehog-forwarder = { workspace = true, optional = true }
@@ -124,6 +125,7 @@ eyre.workspace = true
 futures.workspace = true
 hex.workspace = true
 pin-project-lite.workspace = true
+rustix = { workspace = true, features = ["fs"] }
 rustls.workspace = true
 serde.workspace = true
 serde_json.workspace = true
@@ -191,6 +193,7 @@ color-eyre = "0.6.5"
 deadpool = { version = "0.12.3", default-features = false }
 diesel = "2.3.5"
 diesel_migrations = "2.3.1"
+dirs = "6.0.0"
 displaydoc = "0.2.5"
 edgehog-containers = { package = "edgehog-device-runtime-containers", path = "./edgehog-device-runtime-containers", version = "=0.10.4" }
 edgehog-device-forwarder-proto = "0.1.0"

Cargo.toml

@@ -0,0 +1,35 @@
+// This file is part of Edgehog.
+//
+// Copyright 2026 SECO Mind Srl
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+use uuid::Uuid;
+
+use super::request::FileDigest;
+#[cfg(unix)]
+use super::request::FilePermissions;
+
+pub(super) mod store;
+pub(super) mod stream;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct FileOptions {
+    pub(super) id: Uuid,
+    pub(super) file_size: u64,
+    pub(super) file_digest: FileDigest,
+    #[cfg(unix)]
+    pub(super) perm: FilePermissions,
+}

src/file_transfer/file_system/mod.rs

@@ -28,19 +28,17 @@ use tokio::io::{AsyncBufReadExt, AsyncWrite, BufReader};
 use tracing::{info, instrument, trace};
 use uuid::Uuid;
 
-use super::request::FileDigest;
-#[cfg(unix)]
-use super::request::FilePermissions;
+use super::FileOptions;
 
 /// Stores files in the storage
 #[derive(Debug)]
-pub(super) struct FileStorage<F> {
-    fs: F,
+pub(crate) struct FileStorage<F> {
     dir: PathBuf,
+    fs: F,
 }
 
 impl FileStorage<Fs> {
-    pub(super) fn new(dir: PathBuf) -> Self {
+    pub(crate) fn new(dir: PathBuf) -> Self {
         Self { fs: Fs {}, dir }
     }
 }
@@ -88,8 +86,6 @@ impl<F> FileStorage<F> {
     where
         F: Limits,
     {
-        trace!("creating write handle");
-
         let file_path = self.partial_file_path(&opt.id);
 
         self.fs
@@ -147,8 +143,6 @@ impl<F> FileStorage<F> {
     where
         F: Limits,
     {
-        trace!("finalizing write handle");
-
         handle
             .file
             .sync_all()
@@ -184,18 +178,9 @@ impl<F> FileStorage<F> {
     }
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub(super) struct FileOptions {
-    pub(super) id: Uuid,
-    pub(super) file_size: u64,
-    pub(super) file_digest: FileDigest,
-    #[cfg(unix)]
-    pub(super) perm: FilePermissions,
-}
-
 pin_project! {
     #[derive(Debug)]
-    pub(super) struct WriteHandle {
+    pub(crate) struct WriteHandle {
         id: Uuid,
         current_size: u64,
         // TODO limit the size of the file
@@ -285,6 +270,10 @@ mod tests {
     use tempdir::TempDir;
     use tokio::io::AsyncWriteExt;
 
+    use crate::file_transfer::request::FileDigest;
+    #[cfg(unix)]
+    use crate::file_transfer::request::FilePermissions;
+
     use super::*;
 
     use pretty_assertions::assert_eq;

src/file_transfer/file_system.rs src/file_transfer/file_system/store.rssrc/file_transfer/file_system.rs renamed to src/file_transfer/file_system/store.rs

@@ -0,0 +1,107 @@
+// This file is part of Edgehog.
+//
+// Copyright 2026 SECO Mind Srl
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+//! Stream the bytes to a named pipe or unix socket.
+
+use tokio::io::{AsyncRead, AsyncWrite};
+use tracing::instrument;
+use uuid::Uuid;
+
+use super::FileOptions;
+
+#[cfg(unix)]
+mod unix;
+#[cfg(windows)]
+mod windows;
+
+cfg_if::cfg_if! {
+    if #[cfg(unix)] {
+        pub(crate) type SysPipe = self::unix::MakeFifo;
+    } else if #[cfg(windows)] {
+        pub(crate) type SysPipe = self::windows::MakeNamedPipe;
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct Streaming<S> {
+    sys: S,
+}
+
+impl<S> Streaming<S> {
+    #[instrument(skip_all, fields(id = %opt.id))]
+    pub(crate) async fn open_writer(
+        &self,
+        opt: &FileOptions,
+    ) -> eyre::Result<(S::Writer, aws_lc_rs::digest::Context)>
+    where
+        S: Pipe,
+    {
+        let writer = self.sys.open_writer(opt).await?;
+        let digest = aws_lc_rs::digest::Context::from(opt.file_digest);
+
+        Ok((writer, digest))
+    }
+
+    #[instrument(skip(self))]
+    pub(crate) async fn create_reader(&self, id: &Uuid) -> eyre::Result<S::Reader>
+    where
+        S: Pipe,
+    {
+        let reader = self.sys.create_reader(id).await?;
+
+        Ok(reader)
+    }
+}
+
+impl Streaming<SysPipe> {
+    pub(crate) fn with_sys() -> Self {
+        cfg_if::cfg_if! {
+            if #[cfg(unix)] {
+                Self {
+                    sys: self::unix::MakeFifo::new(),
+                }
+            } else if #[cfg(windows)] {
+                Self {
+                    sys: self::windows::MakeNamedPipe::new(),
+                }
+            }
+        }
+    }
+}
+
+pub(crate) trait Pipe {
+    type Reader: AsyncRead;
+    type Writer: AsyncWrite;
+
+    fn open_writer(
+        &self,
+        opt: &FileOptions,
+    ) -> impl Future<Output = eyre::Result<Self::Writer>> + Send;
+
+    fn create_reader(&self, id: &Uuid) -> impl Future<Output = eyre::Result<Self::Reader>> + Send;
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn runtime_dir() {
+        Streaming::with_sys();
+    }
+}