fix(serverust-events): liberar lock InProgress após falha do IdempotencyLayer

CHANGELOG.md

@@ -35,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `EventRouter` com `RetryPolicy::Exponential`: o atraso `base_delay * 2^n` passa a usar `Duration::saturating_mul` e expoente limitado a 31, evitando panic por overflow de `Duration` em retentativas longas ou `base_delay` grande.
 - `SqsBroker::handle_sqs_event` (Lambda ESM + `ReportBatchItemFailures`): mensagens sem handler para a fila do ARN ou sem `event_source_arn` válido passam a entrar em `batchItemFailures` quando há `messageId`, em vez de serem tratadas como sucesso implícito (a Lambda removia da fila sem processamento).
+- `IdempotencyLayer`: após falha do handler (ou falha ao gravar `Completed`), o lock `InProgress` é liberado via `IdempotencyStore::release`, permitindo redeliveries SQS em vez de bloquear o processamento até o TTL (24 h).
 
 ## [0.3.0] - 2026-05-17
 

serverust-events/src/sqs/layers.rs

@@ -265,8 +265,26 @@ where
             match store.try_acquire(&key, acquired_at, ttl_ms).await {
                 Ok(AcquireOutcome::Acquired) => {
                     let result = inner.call(req).await;
-                    if result.is_ok() {
-                        let _ = store.complete(&key, now_ms(), ttl_ms).await;
+                    match &result {
+                        Ok(()) => {
+                            if let Err(e) = store.complete(&key, now_ms(), ttl_ms).await {
+                                warn!(
+                                    key = %key,
+                                    error = %e,
+                                    "idempotency complete failed after handler success; releasing lock",
+                                );
+                                let _ = store.release(&key).await;
+                            }
+                        }
+                        Err(_) => {
+                            if let Err(e) = store.release(&key).await {
+                                warn!(
+                                    key = %key,
+                                    error = %e,
+                                    "idempotency release failed after handler error",
+                                );
+                            }
+                        }
                     }
                     result
                 }

serverust-events/tests/sqs_idempotency.rs

@@ -153,10 +153,56 @@ async fn expired_record_allows_reprocessing() {
     );
 }
 
+#[tokio::test]
+async fn handler_failure_releases_lock_allowing_sqs_redelivery() {
+    let calls = Arc::new(Mutex::new(0_u32));
+    let calls_for_handler = calls.clone();
+    let subscriber = SqsSubscriber::new(move |_msg: SqsMessage| {
+        let calls = calls_for_handler.clone();
+        async move {
+            let n = {
+                let mut c = calls.lock().unwrap();
+                *c += 1;
+                *c
+            };
+            if n == 1 {
+                Err::<(), _>(BrokerError::Subscribe("transient".into()))
+            } else {
+                Ok(())
+            }
+        }
+    });
+
+    let store: Arc<dyn IdempotencyStore> = Arc::new(InMemoryIdempotencyStore::new());
+
+    let mut svc = ServiceBuilder::new()
+        .layer(IdempotencyLayer::new(store.clone()).with_ttl(Duration::from_secs(60)))
+        .service(subscriber);
+
+    let msg = message_with_id("retry-after-fail");
+    svc.ready().await.unwrap();
+    let err = svc.call(msg.clone()).await.unwrap_err();
+    assert!(matches!(err, BrokerError::Subscribe(_)));
+
+    svc.ready().await.unwrap();
+    svc.call(msg).await.unwrap();
+
+    assert_eq!(
+        *calls.lock().unwrap(),
+        2,
+        "redelivery SQS deve reexecutar o handler após release do lock",
+    );
+
+    let outcome = store.try_acquire("retry-after-fail", now_ms(), 60_000).await.unwrap();
+    assert!(
+        matches!(outcome, AcquireOutcome::AlreadyCompleted(_)),
+        "segunda execução bem-sucedida deve marcar Completed",
+    );
+}
+
 #[tokio::test]
 async fn handler_failure_does_not_persist_completed_record() {
-    // Após erro, o lock InProgress fica gravado e expira; ele NÃO vira Completed,
-    // de modo que uma retentativa após o TTL pode reprocessar.
+    // Após erro, o lock InProgress é liberado para permitir redelivery SQS.
     let calls = Arc::new(Mutex::new(0_u32));
     let calls_for_handler = calls.clone();
     let subscriber = SqsSubscriber::new(move |_msg: SqsMessage| {
@@ -178,12 +224,10 @@ async fn handler_failure_does_not_persist_completed_record() {
     assert!(matches!(err, BrokerError::Subscribe(_)));
     assert_eq!(*calls.lock().unwrap(), 1);
 
-    // Após falha, store está InProgress (não Completed) — try_acquire dentro
-    // do TTL ainda vê InProgress (lock segura até TTL expirar).
     let outcome = store.try_acquire("fail", now_ms(), 60_000).await.unwrap();
     assert!(
-        matches!(outcome, AcquireOutcome::InProgress),
-        "falha não deve marcar Completed",
+        matches!(outcome, AcquireOutcome::Acquired),
+        "falha deve liberar o lock para nova tentativa",
     );
 }
 

serverust-telemetry/src/idempotency.rs

@@ -18,6 +18,8 @@
 //!   deve recuar) ou [`AcquireOutcome::AlreadyCompleted`] (skip seguro).
 //! - [`IdempotencyStore::complete`] grava o resultado final (estado
 //!   `Completed`) com novo TTL.
+//! - [`IdempotencyStore::release`] remove um lock `InProgress` após falha do
+//!   handler, permitindo que redeliveries SQS reexecutem o processamento.
 //!
 //! Em DynamoDB, isso é implementado com `condition_expression`
 //! `attribute_not_exists(pk) OR expires_at_ms < :now` no `try_acquire`.
@@ -99,6 +101,13 @@ pub trait IdempotencyStore: Send + Sync + 'static {
     /// Marca a chave como Completed com novo TTL. Caller deve ter chamado
     /// `try_acquire` antes e obtido `Acquired`.
     async fn complete(&self, key: &str, now_ms: u64, ttl_ms: u64) -> Result<(), IdempotencyError>;
+
+    /// Remove um lock `InProgress` após falha do handler.
+    ///
+    /// Sem isso, redeliveries SQS (visibility timeout) encontram `InProgress`
+    /// e falham até o TTL expirar (default 24 h). Registros `Completed` ou
+    /// chaves ausentes são no-op.
+    async fn release(&self, key: &str) -> Result<(), IdempotencyError>;
 }
 
 /// Implementação in-memory thread-safe — referência para testes e dev.
@@ -181,6 +190,20 @@ impl IdempotencyStore for InMemoryIdempotencyStore {
         );
         Ok(())
     }
+
+    async fn release(&self, key: &str) -> Result<(), IdempotencyError> {
+        let mut guard = self
+            .locks
+            .lock()
+            .map_err(|e| IdempotencyError::Storage(e.to_string()))?;
+        if guard
+            .get(key)
+            .is_some_and(|r| r.state == IdempotencyState::InProgress)
+        {
+            guard.remove(key);
+        }
+        Ok(())
+    }
 }
 
 #[cfg(feature = "dynamodb")]
@@ -192,6 +215,7 @@ mod dynamodb_impl {
     use async_trait::async_trait;
     use aws_sdk_dynamodb::Client;
     use aws_sdk_dynamodb::error::SdkError;
+    use aws_sdk_dynamodb::operation::delete_item::DeleteItemError;
     use aws_sdk_dynamodb::operation::put_item::PutItemError;
     use aws_sdk_dynamodb::primitives::Blob;
     use aws_sdk_dynamodb::types::AttributeValue;
@@ -381,6 +405,35 @@ mod dynamodb_impl {
                 .map_err(|e| IdempotencyError::Storage(e.to_string()))?;
             Ok(())
         }
+
+        async fn release(&self, key: &str) -> Result<(), IdempotencyError> {
+            let result = self
+                .client
+                .delete_item()
+                .table_name(&self.table_name)
+                .key("pk", AttributeValue::S(key.to_string()))
+                .condition_expression("#state = :in_progress")
+                .expression_attribute_names("#state", "state")
+                .expression_attribute_values(
+                    ":in_progress",
+                    AttributeValue::S("InProgress".to_string()),
+                )
+                .send()
+                .await;
+
+            match result {
+                Ok(_) => Ok(()),
+                Err(SdkError::ServiceError(svc))
+                    if matches!(
+                        svc.err(),
+                        DeleteItemError::ConditionalCheckFailedException(_)
+                    ) =>
+                {
+                    Ok(())
+                }
+                Err(e) => Err(IdempotencyError::Storage(e.to_string())),
+            }
+        }
     }
 }
 

serverust-telemetry/tests/idempotency_lock.rs

@@ -82,3 +82,28 @@ async fn complete_marks_record_as_completed() {
         other => panic!("esperava AlreadyCompleted, recebi {other:?}"),
     }
 }
+
+#[tokio::test]
+async fn release_clears_in_progress_lock() {
+    let store = InMemoryIdempotencyStore::new();
+    store.try_acquire("k", 1_000, TTL_MS).await.unwrap();
+    store.release("k").await.unwrap();
+    let outcome = store.try_acquire("k", 2_000, TTL_MS).await.unwrap();
+    assert!(
+        matches!(outcome, AcquireOutcome::Acquired),
+        "release deve permitir nova aquisição",
+    );
+}
+
+#[tokio::test]
+async fn release_is_noop_for_completed_record() {
+    let store = InMemoryIdempotencyStore::new();
+    store.try_acquire("k", 1_000, TTL_MS).await.unwrap();
+    store.complete("k", 1_500, TTL_MS).await.unwrap();
+    store.release("k").await.unwrap();
+    let outcome = store.try_acquire("k", 2_000, TTL_MS).await.unwrap();
+    assert!(
+        matches!(outcome, AcquireOutcome::AlreadyCompleted(_)),
+        "release não deve remover Completed",
+    );
+}