s3rm-rs 1.3.2

Fast Amazon S3 object deletion tool.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use anyhow::{Context, Result};
use async_channel::{Receiver, Sender};

use crate::config::Config;
use crate::storage::Storage;
use crate::types::token::PipelineCancellationToken;
use crate::types::{DeletionStatistics, S3Object};

/// Result of sending an object to the next stage.
#[derive(Debug, Clone, PartialEq)]
pub enum SendResult {
    Success,
    Closed,
}

/// Shared context passed to each pipeline stage.
///
/// Adapted from s3sync's Stage struct, simplified for deletion-only operations:
/// - Only a single `target` storage (no `source` needed since s3rm-rs doesn't sync)
/// - Channels connect stages: each stage reads from `receiver` and writes to `sender`
/// - The ObjectLister stage has no `receiver` (it's the pipeline entry point)
/// - The Terminator stage has no `sender` (it's the pipeline exit)
///
/// Each stage takes ownership of a `Stage`, consuming it during pipeline construction.
pub struct Stage {
    pub config: Config,
    pub target: Storage,
    pub receiver: Option<Receiver<S3Object>>,
    pub sender: Option<Sender<S3Object>>,
    pub cancellation_token: PipelineCancellationToken,
    pub has_warning: Arc<AtomicBool>,
}

impl Stage {
    pub fn new(
        config: Config,
        target: Storage,
        receiver: Option<Receiver<S3Object>>,
        sender: Option<Sender<S3Object>>,
        cancellation_token: PipelineCancellationToken,
        has_warning: Arc<AtomicBool>,
    ) -> Self {
        Self {
            config,
            target,
            receiver,
            sender,
            cancellation_token,
            has_warning,
        }
    }

    /// Send an object to the next stage via the sender channel.
    ///
    /// Returns `SendResult::Closed` if the downstream channel has been closed
    /// (e.g. due to cancellation), allowing the caller to exit gracefully.
    pub async fn send(&self, object: S3Object) -> Result<SendResult> {
        let result = self
            .sender
            .as_ref()
            .expect("Stage sender not initialized: this stage cannot send objects")
            .send(object)
            .await
            .context("async_channel::Sender::send() failed.");

        if let Err(e) = result {
            return if !self.is_channel_closed() {
                Err(e)
            } else {
                Ok(SendResult::Closed)
            };
        }

        Ok(SendResult::Success)
    }

    /// Check if the sender channel has been closed by the receiver.
    pub fn is_channel_closed(&self) -> bool {
        self.sender
            .as_ref()
            .expect("Stage sender not initialized: this stage cannot send objects")
            .is_closed()
    }

    /// Send a statistics event through the storage stats channel.
    pub async fn send_stats(&self, stats: DeletionStatistics) {
        let _ = self.target.get_stats_sender().send(stats).await;
    }

    /// Set the warning flag to indicate a non-fatal issue occurred.
    pub fn set_warning(&self) {
        self.has_warning.store(true, Ordering::SeqCst);
    }
}