Make sure to set max connection to 1

Cargo.toml

@@ -28,7 +28,7 @@ path = "src/lib.rs"
 [dependencies]
 anyhow = "1.0.44"
 time = "0.3.3"
-sqlx = { version = "0.7", features = [ "runtime-tokio-rustls", "sqlite" ] }
+sqlx = { version = "0.7.4", features = [ "runtime-tokio-rustls", "sqlite" ] }
 tokio = { version = "1", features = [ "rt", "rt-multi-thread", "macros"] }
 libc = "0.2"
 futures = "0.3"

flake.nix

@@ -6,22 +6,36 @@
     crane.inputs.nixpkgs.follows = "nixpkgs";
 
     flake-utils.inputs.nixpkgs.follows = "nixpkgs";
+
+    rust-overlay = {
+      url = "github:oxalica/rust-overlay";
+      inputs = {
+        nixpkgs.follows = "nixpkgs";
+        flake-utils.follows = "flake-utils";
+      };
+    };
   };
 
   outputs = {
     self,
     nixpkgs,
     crane,
     flake-utils,
+    rust-overlay,
   }:
     flake-utils.lib.eachSystem
     [
       flake-utils.lib.system.x86_64-linux
       flake-utils.lib.system.aarch64-linux
       flake-utils.lib.system.aarch64-darwin
     ] (system: let
-      pkgs = nixpkgs.legacyPackages.${system};
-      craneLib = crane.lib.${system};
+      pkgs = import nixpkgs {
+        inherit system;
+        overlays = [(import rust-overlay)];
+      };
+
+      customToolchain = pkgs.rust-bin.fromRustupToolchainFile ./rust-toolchain.toml;
+      craneLib = (crane.mkLib pkgs).overrideToolchain customToolchain;
     in {
       devShells.default = craneLib.devShell {
         packages = [

rust-toolchain.toml

@@ -0,0 +1,3 @@
+[toolchain]
+channel = "1.74.0"
+

src/fungi/meta.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use sqlx::{
-    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqliteRow},
+    sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteRow},
     FromRow, Row, SqlitePool,
 };
 
@@ -348,7 +348,10 @@ impl Writer {
             .journal_mode(SqliteJournalMode::Delete)
             .filename(path);
 
-        let pool = SqlitePool::connect_with(opts).await?;
+        let pool = SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect_with(opts)
+            .await?;
 
         sqlx::query(SCHEMA).execute(&pool).await?;
 

src/pack.rs

@@ -1,7 +1,8 @@
 use crate::fungi::meta::{Ino, Inode};
-use crate::fungi::{Result, Writer};
+use crate::fungi::{Error, Result, Writer};
 use crate::store::{BlockStore, Store};
 use anyhow::Context;
+use futures::lock::Mutex;
 use std::collections::LinkedList;
 use std::ffi::OsString;
 use std::fs::Metadata;
@@ -12,6 +13,8 @@ use workers::WorkerPool;
 
 const BLOB_SIZE: usize = 512 * 1024; // 512K
 
+type FailuresList = Arc<Mutex<Vec<(PathBuf, Error)>>>;
+
 #[derive(Debug)]
 struct Item(Ino, PathBuf, OsString, Metadata);
 /// creates an FL from the given root location. It takes ownership of the writer because
@@ -58,7 +61,8 @@ pub async fn pack<P: Into<PathBuf>, S: Store>(
 
     let mut list = LinkedList::default();
 
-    let uploader = Uploader::new(store, writer.clone());
+    let failures = FailuresList::default();
+    let uploader = Uploader::new(store, writer.clone(), Arc::clone(&failures));
     let mut pool = workers::WorkerPool::new(uploader.clone(), super::PARALLEL_UPLOAD);
 
     pack_one(
@@ -75,7 +79,21 @@ pub async fn pack<P: Into<PathBuf>, S: Store>(
     }
 
     pool.close().await;
-    Ok(())
+
+    let failures = failures.lock().await;
+    if failures.is_empty() {
+        return Ok(());
+    }
+
+    log::error!("failed to upload one or more files");
+    for (file, error) in failures.iter() {
+        log::error!("  - failed to upload file {}: {}", file.display(), error);
+    }
+
+    Err(Error::Anyhow(anyhow::anyhow!(
+        "failed to upload ({}) files",
+        failures.len()
+    )))
 }
 
 /// pack_one is called for each dir
@@ -165,6 +183,7 @@ where
     S: Store,
 {
     store: Arc<BlockStore<S>>,
+    failures: FailuresList,
     writer: Writer,
     buffer: [u8; BLOB_SIZE],
 }
@@ -176,6 +195,7 @@ where
     fn clone(&self) -> Self {
         Self {
             store: Arc::clone(&self.store),
+            failures: Arc::clone(&self.failures),
             writer: self.writer.clone(),
             buffer: [0; BLOB_SIZE],
         }
@@ -186,9 +206,10 @@ impl<S> Uploader<S>
 where
     S: Store,
 {
-    fn new(store: BlockStore<S>, writer: Writer) -> Self {
+    fn new(store: BlockStore<S>, writer: Writer, failures: FailuresList) -> Self {
         Self {
             store: Arc::new(store),
+            failures,
             writer,
             buffer: [0; BLOB_SIZE],
         }
@@ -231,7 +252,8 @@ where
     async fn run(&mut self, (ino, path): Self::Input) -> Self::Output {
         log::info!("uploading {:?}", path);
         if let Err(err) = self.upload(ino, &path).await {
-            log::error!("failed to upload file ({:?}): {:#}", path, err);
+            log::error!("failed to upload file {}: {:#}", path.display(), err);
+            self.failures.lock().await.push((path, err));
         }
     }
 }

src/store/s3store.rs

@@ -73,7 +73,7 @@ impl S3Store {
             .with_path_style();
 
         Ok(Self {
-            bucket: bucket,
+            bucket,
             url: url.to_owned(),
         })
     }