fix(sqs): liberar lock InProgress após falha no 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` (SQS): após falha do handler ou falha ao marcar `Completed`, o lock `InProgress` é liberado via `IdempotencyStore::release` — redeliveries do SQS voltam a adquirir a chave em vez de ficarem bloqueadas até o TTL (24h) expirar.
 
 ## [0.3.0] - 2026-05-17
 

serverust-events/src/sqs/layers.rs

@@ -265,8 +265,30 @@ 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 {
+                                if let Err(release_err) = store.release(&key).await {
+                                    warn!(
+                                        idempotency_key = %key,
+                                        error = %release_err,
+                                        "idempotency release after complete failure also failed",
+                                    );
+                                }
+                                return Err(BrokerError::Subscribe(format!(
+                                    "idempotency complete: {e}"
+                                )));
+                            }
+                        }
+                        Err(_) => {
+                            if let Err(release_err) = store.release(&key).await {
+                                warn!(
+                                    idempotency_key = %key,
+                                    error = %release_err,
+                                    "idempotency release after handler failure failed",
+                                );
+                            }
+                        }
                     }
                     result
                 }

serverust-events/tests/sqs_idempotency.rs

@@ -154,9 +154,7 @@ 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() {
     let calls = Arc::new(Mutex::new(0_u32));
     let calls_for_handler = calls.clone();
     let subscriber = SqsSubscriber::new(move |_msg: SqsMessage| {
@@ -178,12 +176,96 @@ 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).
+    // Após falha, o lock InProgress é liberado para que redeliveries do SQS
+    // possam re-adquirir e reexecutar o handler (não ficam bloqueados 24h).
     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, não marcar Completed nem manter InProgress",
+    );
+}
+
+#[tokio::test]
+async fn complete_failure_returns_error_and_releases_lock() {
+    struct FailingCompleteStore {
+        inner: InMemoryIdempotencyStore,
+    }
+
+    #[async_trait::async_trait]
+    impl IdempotencyStore for FailingCompleteStore {
+        async fn get(
+            &self,
+            key: &str,
+        ) -> Result<Option<serverust_telemetry::IdempotencyRecord>, serverust_telemetry::IdempotencyError>
+        {
+            self.inner.get(key).await
+        }
+
+        async fn put(
+            &self,
+            record: serverust_telemetry::IdempotencyRecord,
+        ) -> Result<(), serverust_telemetry::IdempotencyError> {
+            self.inner.put(record).await
+        }
+
+        async fn try_acquire(
+            &self,
+            key: &str,
+            now_ms: u64,
+            ttl_ms: u64,
+        ) -> Result<AcquireOutcome, serverust_telemetry::IdempotencyError> {
+            self.inner.try_acquire(key, now_ms, ttl_ms).await
+        }
+
+        async fn complete(
+            &self,
+            _key: &str,
+            _now_ms: u64,
+            _ttl_ms: u64,
+        ) -> Result<(), serverust_telemetry::IdempotencyError> {
+            Err(serverust_telemetry::IdempotencyError::Storage(
+                "dynamodb throttle".into(),
+            ))
+        }
+
+        async fn release(&self, key: &str) -> Result<(), serverust_telemetry::IdempotencyError> {
+            self.inner.release(key).await
+        }
+    }
+
+    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 {
+            *calls.lock().unwrap() += 1;
+            Ok::<_, BrokerError>(())
+        }
+    });
+
+    let store: Arc<dyn IdempotencyStore> = Arc::new(FailingCompleteStore {
+        inner: InMemoryIdempotencyStore::new(),
+    });
+
+    let mut svc = ServiceBuilder::new()
+        .layer(IdempotencyLayer::new(store.clone()).with_ttl(Duration::from_secs(60)))
+        .service(subscriber);
+
+    svc.ready().await.unwrap();
+    let err = svc
+        .call(message_with_id("complete-fail"))
+        .await
+        .expect_err("complete failure must surface to SQS for retry");
+    assert!(
+        err.to_string().contains("idempotency complete"),
+        "erro deve indicar falha no complete, recebi: {err}",
+    );
+    assert_eq!(*calls.lock().unwrap(), 1);
+
+    let outcome = store.try_acquire("complete-fail", now_ms(), 60_000).await.unwrap();
+    assert!(
+        matches!(outcome, AcquireOutcome::Acquired),
+        "lock liberado após falha no complete para permitir retry",
     );
 }
 

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` após falha do
+//!   handler ou falha ao marcar `Completed`, permitindo que redeliveries do
+//!   SQS re-adquiram a chave.
 //!
 //! Em DynamoDB, isso é implementado com `condition_expression`
 //! `attribute_not_exists(pk) OR expires_at_ms < :now` no `try_acquire`.
@@ -99,6 +102,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 o lock `InProgress` da chave sem marcar `Completed`.
+    ///
+    /// Usado quando o handler falha ou quando `complete` falha após sucesso do
+    /// handler, para que redeliveries do SQS possam chamar `try_acquire` de
+    /// novo em vez de ficarem bloqueados até o TTL expirar.
+    async fn release(&self, key: &str) -> Result<(), IdempotencyError>;
 }
 
 /// Implementação in-memory thread-safe — referência para testes e dev.
@@ -181,6 +191,15 @@ 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()))?;
+        guard.remove(key);
+        Ok(())
+    }
 }
 
 #[cfg(feature = "dynamodb")]
@@ -381,6 +400,17 @@ mod dynamodb_impl {
                 .map_err(|e| IdempotencyError::Storage(e.to_string()))?;
             Ok(())
         }
+
+        async fn release(&self, key: &str) -> Result<(), IdempotencyError> {
+            self.client
+                .delete_item()
+                .table_name(&self.table_name)
+                .key("pk", AttributeValue::S(key.to_string()))
+                .send()
+                .await
+                .map_err(|e| IdempotencyError::Storage(e.to_string()))?;
+            Ok(())
+        }
     }
 }
 

serverust-telemetry/tests/idempotency_lock.rs

@@ -69,6 +69,15 @@ async fn try_acquire_overrides_expired_completed_record() {
     assert!(matches!(outcome, AcquireOutcome::Acquired));
 }
 
+#[tokio::test]
+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", 1_100, TTL_MS).await.unwrap();
+    assert!(matches!(outcome, AcquireOutcome::Acquired));
+}
+
 #[tokio::test]
 async fn complete_marks_record_as_completed() {
     let store = InMemoryIdempotencyStore::new();