JaimeJunr · cursor · Jun 14, 2026
diff --git a/CHANGELOG.md b/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 de `complete()`, libera o lock `InProgress` via `IdempotencyStore::release`, permitindo que redeliveries do SQS reexecutem o handler dentro do TTL (antes o lock bloqueava reprocessamento por até 24h e a mensagem ia para DLQ sem nova tentativa).
 
 ## [0.3.0] - 2026-05-17
 

diff --git a/serverust-events/src/sqs/layers.rs b/serverust-events/src/sqs/layers.rs
@@ -265,8 +265,18 @@ 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 {
+                                let _ = store.release(&key).await;
+                                return Err(BrokerError::Subscribe(format!(
+                                    "idempotency store: {e}"
+                                )));
+                            }
+                        }
+                        Err(_) => {
+                            let _ = store.release(&key).await;
+                        }
                     }
                     result
                 }

diff --git a/serverust-events/tests/sqs_idempotency.rs b/serverust-events/tests/sqs_idempotency.rs
@@ -154,9 +154,9 @@ async fn expired_record_allows_reprocessing() {
 }
 
 #[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.
+async fn handler_failure_releases_lock_for_sqs_redelivery() {
+    // Após erro, o lock InProgress deve ser liberado para que o SQS redelivery
+    // dentro do TTL possa reexecutar o handler (não ficar preso em InProgress).
     let calls = Arc::new(Mutex::new(0_u32));
     let calls_for_handler = calls.clone();
     let subscriber = SqsSubscriber::new(move |_msg: SqsMessage| {
@@ -173,17 +173,26 @@ async fn handler_failure_does_not_persist_completed_record() {
         .layer(IdempotencyLayer::new(store.clone()).with_ttl(Duration::from_secs(60)))
         .service(subscriber);
 
+    let msg = message_with_id("fail");
     svc.ready().await.unwrap();
-    let err = svc.call(message_with_id("fail")).await.unwrap_err();
+    let err = svc.call(msg.clone()).await.unwrap_err();
     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).
+    // Simula redelivery do SQS: segunda invocação deve rodar o handler de novo.
+    svc.ready().await.unwrap();
+    let err = svc.call(msg).await.unwrap_err();
+    assert!(matches!(err, BrokerError::Subscribe(_)));
+    assert_eq!(
+        *calls.lock().unwrap(),
+        2,
+        "redelivery dentro do TTL deve reexecutar o handler",
+    );
+
     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 não deve deixar lock Completed nem InProgress preso",
     );
 }
 

diff --git a/serverust-telemetry/src/idempotency.rs b/serverust-telemetry/src/idempotency.rs
@@ -18,6 +18,9 @@
 //!   deve recuar) ou [`AcquireOutcome::AlreadyCompleted`] (skip seguro).
 //! - [`IdempotencyStore::complete`] grava o resultado final (estado
 //!   `Completed`) com novo TTL.
+//! - [`IdempotencyStore::release`] remove um lock `InProgress` abandonado
+//!   após falha do handler ou de `complete`, permitindo que o SQS redelivery
+//!   reexecute o processamento dentro do TTL.
 //!
 //! Em DynamoDB, isso é implementado com `condition_expression`
 //! `attribute_not_exists(pk) OR expires_at_ms < :now` no `try_acquire`.
@@ -99,6 +102,11 @@ 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>;
+
+    /// Libera um lock `InProgress` abandonado (ex.: handler falhou após
+    /// `try_acquire`). Registros `Completed` não são alterados. Idempotente
+    /// quando a chave já foi liberada ou expirou.
+    async fn release(&self, key: &str) -> Result<(), IdempotencyError>;
 }
 
 /// Implementação in-memory thread-safe — referência para testes e dev.
@@ -181,6 +189,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()))?;
+        match guard.get(key) {
+            Some(existing) if existing.state == IdempotencyState::InProgress => {
+                guard.remove(key);
+            }
+            _ => {}
+        }
+        Ok(())
+    }
 }
 
 #[cfg(feature = "dynamodb")]
@@ -381,6 +403,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(),
+                        aws_sdk_dynamodb::operation::delete_item::DeleteItemError::ConditionalCheckFailedException(_)
+                    ) =>
+                {
+                    Ok(())
+                }
+                Err(e) => Err(IdempotencyError::Storage(e.to_string())),
+            }
+        }
     }
 }
 

diff --git a/serverust-telemetry/tests/idempotency_lock.rs b/serverust-telemetry/tests/idempotency_lock.rs
@@ -70,15 +70,23 @@ async fn try_acquire_overrides_expired_completed_record() {
 }
 
 #[tokio::test]
-async fn complete_marks_record_as_completed() {
+async fn release_removes_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),
+        "após release, nova aquisição deve ser concedida",
+    );
+}
+
+#[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();
-    let outcome = store.try_acquire("k", 1_600, TTL_MS).await.unwrap();
-    match outcome {
-        AcquireOutcome::AlreadyCompleted(rec) => {
-            assert_eq!(rec.state, IdempotencyState::Completed);
-        }
-        other => panic!("esperava AlreadyCompleted, recebi {other:?}"),
-    }
+    store.release("k").await.unwrap();
+    let outcome = store.try_acquire("k", 2_000, TTL_MS).await.unwrap();
+    assert!(matches!(outcome, AcquireOutcome::AlreadyCompleted(_)));
 }