Make error paths explicit

src/error.rs

@@ -25,6 +25,7 @@ impl std::fmt::Display for VolumeError {
 #[derive(Debug)]
 pub enum AppError {
     NotFound,
+    CorruptRecord { key: String },
     Db(rusqlite::Error),
     InsufficientVolumes { need: usize, have: usize },
     PartialWrite,
@@ -43,6 +44,9 @@ impl std::fmt::Display for AppError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             AppError::NotFound => write!(f, "not found"),
+            AppError::CorruptRecord { key } => {
+                write!(f, "corrupt record for key {key}: no volumes")
+            }
             AppError::Db(e) => write!(f, "database error: {e}"),
             AppError::InsufficientVolumes { need, have } => {
                 write!(f, "need {need} volumes but only {have} available")
@@ -56,6 +60,7 @@ impl IntoResponse for AppError {
     fn into_response(self) -> Response {
         let status = match &self {
             AppError::NotFound => StatusCode::NOT_FOUND,
+            AppError::CorruptRecord { .. } => StatusCode::INTERNAL_SERVER_ERROR,
             AppError::Db(_) => StatusCode::INTERNAL_SERVER_ERROR,
             AppError::InsufficientVolumes { .. } => StatusCode::SERVICE_UNAVAILABLE,
             AppError::PartialWrite => StatusCode::BAD_GATEWAY,

src/rebalance.rs

@@ -77,10 +77,18 @@ pub async fn run(args: &Args, dry_run: bool) {
                         }
                     };
                     let dst_url = format!("{dst}/{}", m.key);
-                    if let Err(e) = client.put(&dst_url).body(data).send().await {
-                        eprintln!("  ERROR copy {} to {}: {}", m.key, dst, e);
-                        copy_ok = false;
-                        errors += 1;
+                    match client.put(&dst_url).body(data).send().await {
+                        Ok(resp) if !resp.status().is_success() => {
+                            eprintln!("  ERROR copy {} to {}: status {}", m.key, dst, resp.status());
+                            copy_ok = false;
+                            errors += 1;
+                        }
+                        Err(e) => {
+                            eprintln!("  ERROR copy {} to {}: {}", m.key, dst, e);
+                            copy_ok = false;
+                            errors += 1;
+                        }
+                        Ok(_) => {}
                     }
                 }
                 Ok(resp) => {

src/server.rs

@@ -20,7 +20,10 @@ pub async fn get_key(
     Path(key): Path<String>,
 ) -> Result<Response, AppError> {
     let record = state.db.get(&key).await?;
-    let vol = record.volumes.first().ok_or(AppError::NotFound)?;
+    let vol = record
+        .volumes
+        .first()
+        .ok_or_else(|| AppError::CorruptRecord { key: key.clone() })?;
     let location = format!("{vol}/{key}");
     Ok((StatusCode::FOUND, [(axum::http::header::LOCATION, location)]).into_response())
 }
@@ -82,7 +85,12 @@ pub async fn put_key(
     }
 
     let size = Some(body.len() as i64);
-    state.db.put(key, target_volumes, size).await?;
+    if let Err(e) = state.db.put(key.clone(), target_volumes.clone(), size).await {
+        for vol in &target_volumes {
+            let _ = state.http.delete(format!("{vol}/{key}")).send().await;
+        }
+        return Err(e);
+    }
     Ok(StatusCode::CREATED.into_response())
 }
 
@@ -95,14 +103,25 @@ pub async fn delete_key(
     let mut handles = Vec::new();
     for vol in &record.volumes {
         let url = format!("{vol}/{key}");
-        let client = state.http.clone();
-        handles.push(tokio::spawn(async move { client.delete(&url).send().await }));
+        let handle = tokio::spawn({
+            let client = state.http.clone();
+            async move {
+                let resp = client.delete(&url).send().await.map_err(|e| {
+                    VolumeError::Request { url: url.clone(), source: e }
+                })?;
+                if !resp.status().is_success() {
+                    return Err(VolumeError::BadStatus { url, status: resp.status() });
+                }
+                Ok(())
+            }
+        });
+        handles.push(handle);
     }
     for handle in handles {
         match handle.await {
-            Ok(Err(e)) => tracing::error!("DELETE from volume failed: {e}"),
+            Ok(Err(e)) => tracing::error!("{e}"),
             Err(e) => tracing::error!("volume delete task failed: {e}"),
-            Ok(Ok(_)) => {}
+            Ok(Ok(())) => {}
         }
     }