diff --git a/CHANGELOG.md b/CHANGELOG.md index bf86a71..30d567e 100644 --- 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` (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 diff --git a/serverust-events/src/sqs/layers.rs b/serverust-events/src/sqs/layers.rs index b096a7e..21f88eb 100644 --- a/serverust-events/src/sqs/layers.rs +++ b/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 } diff --git a/serverust-events/tests/sqs_idempotency.rs b/serverust-events/tests/sqs_idempotency.rs index b39df03..e6305cd 100644 --- a/serverust-events/tests/sqs_idempotency.rs +++ b/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, 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 { + 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 = 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", ); } diff --git a/serverust-telemetry/src/idempotency.rs b/serverust-telemetry/src/idempotency.rs index dd39aef..37d5117 100644 --- 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` 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(()) + } } } diff --git a/serverust-telemetry/tests/idempotency_lock.rs b/serverust-telemetry/tests/idempotency_lock.rs index 91e9c32..4e1f024 100644 --- a/serverust-telemetry/tests/idempotency_lock.rs +++ b/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();