try to send frames to S3 even if there were temporary failure

Author: sivukhin

bottomless/src/backup.rs

@@ -5,13 +5,13 @@ use arc_swap::ArcSwapOption;
 use std::ops::{Range, RangeInclusive};
 use std::sync::Arc;
 use tokio::io::AsyncWriteExt;
-use tokio::sync::mpsc::Sender;
+use tokio::sync::mpsc::UnboundedSender;
 use tokio::time::Instant;
 use uuid::Uuid;
 
 #[derive(Debug)]
 pub(crate) struct WalCopier {
-    outbox: Sender<SendReq>,
+    outbox: UnboundedSender<SendReq>,
     use_compression: CompressionKind,
     max_frames_per_batch: usize,
     wal_path: String,
@@ -28,7 +28,7 @@ impl WalCopier {
         db_path: &str,
         max_frames_per_batch: usize,
         use_compression: CompressionKind,
-        outbox: Sender<SendReq>,
+        outbox: UnboundedSender<SendReq>,
     ) -> Self {
         WalCopier {
             bucket,
@@ -76,7 +76,7 @@ impl WalCopier {
             meta_file.write_all(buf.as_ref()).await?;
             meta_file.flush().await?;
             let msg = format!("{}-{}/.meta", self.db_name, generation);
-            if self.outbox.send(SendReq::new(msg)).await.is_err() {
+            if self.outbox.send(SendReq::new(msg)).is_err() {
                 return Err(anyhow!("couldn't initialize local backup dir: {}", dir));
             }
         }
@@ -125,7 +125,6 @@ impl WalCopier {
             if self
                 .outbox
                 .send(SendReq::wal_segment(fdesc, start, end - 1))
-                .await
                 .is_err()
             {
                 tracing::warn!(

bottomless/src/replicator.rs

@@ -350,7 +350,7 @@ impl Replicator {
 
         let mut join_set = JoinSet::new();
 
-        let (frames_outbox, mut frames_inbox) = tokio::sync::mpsc::channel(64);
+        let (frames_outbox, mut frames_inbox) = tokio::sync::mpsc::unbounded_channel();
         let _local_backup = {
             let mut copier = WalCopier::new(
                 bucket.clone(),
@@ -433,24 +433,28 @@ impl Replicator {
 
                     let db_name = db_name.clone();
                     join_set.spawn(async move {
-                        let fpath = format!("{}/{}", bucket, req.path);
-                        let body = ByteStream::from_path(&fpath).await.unwrap();
-                        let start_time = Instant::now();
-                        let response = client
-                            .put_object()
-                            .bucket(bucket)
-                            .key(req.path)
-                            .body(body)
-                            .send()
-                            .await;
-                        Self::record_s3_write_time(&db_name, start_time.elapsed());
-                        if let Err(e) = response {
-                            tracing::error!("Failed to send {} to S3: {}", fpath, e);
-                        } else {
-                            tokio::fs::remove_file(&fpath).await.unwrap();
-                            let elapsed = Instant::now() - start;
-                            tracing::debug!("Uploaded to S3: {} in {:?}", fpath, elapsed);
+                        let fpath = format!("{}/{}", &bucket, &req.path);
+                        loop {
+                            let start_time = Instant::now();
+                            let body = ByteStream::from_path(&fpath).await.unwrap();
+                            let response = client
+                                .put_object()
+                                .bucket(&bucket)
+                                .key(&req.path)
+                                .body(body)
+                                .send()
+                                .await;
+                            Self::record_s3_write_time(&db_name, start_time.elapsed());
+                            if let Err(e) = response {
+                                tracing::error!("Failed to send {} to S3: {}, will retry after 1 second", fpath, e);
+                                tokio::time::sleep(Duration::from_millis(1000)).await;
+                            } else {
+                                break;
+                            }
                         }
+                        tokio::fs::remove_file(&fpath).await.unwrap();
+                        let elapsed = Instant::now() - start;
+                        tracing::debug!("Uploaded to S3: {} in {:?}", fpath, elapsed);
                         if let Some(frames) = req.frames {
                             let mut up = upload_progress.lock().await;
                             up.update(*frames.start(), *frames.end());