feat: Backend circuit breaker

[msalinas92]

Implement a circuit breaker mechanism for persistent backends to avoid retries on failing systems.

Acceptance Criteria:

• Automatically disables backend after N failures in M seconds.
• Logs circuit open/close events.
• Skips storage until backend is healthy again.

[amartin0]

Hi,

I’ve been looking into this new feature and I think I could contribute, especially regarding the S3 backend.

In the module s3.rs, this function initializes the connection to the S3 backend:

(https://github.com/msalinas92/CacheBolt/blob/master/src/storage/s3.rs#L34-L41)

pub async fn init_s3_client() {
    if S3_CLIENT.get().is_none() {
        let region_provider = RegionProviderChain::default_provider().or_else("us-east-1");
        let config = aws_config::from_env().region(region_provider).load().await;
        let client = Client::new(&config);
        let _ = S3_CLIENT.set(client);
    }
}

And this function writes the key and data to the cache:

(https://github.com/msalinas92/CacheBolt/blob/master/src/storage/s3.rs#L48-L63)

pub async fn store_in_cache(key: String, data: Bytes, headers: Vec<(String, String)>) {
    let client = match S3_CLIENT.get() {
        Some(c) => c,
        None => {
            error!("S3 client not initialized");
            return;
        }
    };

    let bucket = match CONFIG.get() {
        Some(cfg) => cfg.s3_bucket.clone(),
        None => {
            error!("CONFIG not initialized; cannot read s3_bucket");
            return;
        }
    };

I think it would be straightforward to add a check (for example, verifying that the bucket exists) in both functions.
If the check fails, it could log an error message accordingly.

At the same time, I could add a variable in the config file to check the fail number to write in the cache, this number would be control in the proxy.rs module when you call CACHE_WRITER.

If the number of checks was exceeded, a thread to check the storage backend would be launched and the main funcion gives up writing in the storage cache. When the thread checked the storage backend is ok, the main function would write in the cache again and the thread would disappear.

https://github.com/msalinas92/CacheBolt/blob/master/src/proxy.rs#L62-L86)

static CACHE_WRITER: Lazy<mpsc::Sender<(String, Bytes, Vec<(String, String)>)>> = Lazy::new(|| {
    let (tx, mut rx) = mpsc::channel::<(String, Bytes, Vec<(String, String)>)>(100);
    tokio::spawn(async move {
        while let Some((key, data, headers)) = rx.recv().await {
            let backend_label = CONFIG
                .get()
                .map(|c| format!("{:?}", c.storage_backend))
                .unwrap_or("unknown".to_string());
            counter!("cachebolt_persist_attempts_total", "backend" => backend_label.clone())
                .increment(1);
            match CONFIG.get().map(|c| &c.storage_backend) {
                Some(StorageBackend::Azure) => azure::store_in_cache(key, data, headers).await,
                Some(StorageBackend::Gcs) => gcs::store_in_cache(key, data, headers).await,
                Some(StorageBackend::Local) => local::store_in_cache(key, data, headers).await,
                Some(StorageBackend::S3) => s3::store_in_cache(key, data, headers).await,
                None => {
                    tracing::error!("❌ CONFIG not initialized. Unable to persist cache.");
                    counter!("cachebolt_persist_errors_total", "backend" => backend_label)
                        .increment(1);
                }
            }
        }
    });
    tx
});

If you agree with this approach, I’d be happy to implement it and open a pull request with the changes.
With your approval, I could also extend this work to the other backends (GCP, local, and Azure).

[msalinas92]

Hi @amartin0,

Thanks for the detailed suggestion. I agree with your approach. Please open a change request with the S3 changes.

thanks