From 31c5b441ea9ec8cc901e6ca8947f7c66dccf2544 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 21 Jun 2026 02:04:56 +0000 Subject: [PATCH] =?UTF-8?q?fix(serverust-events):=20liberar=20lock=20InPro?= =?UTF-8?q?gress=20ap=C3=B3s=20falha=20do=20IdempotencyLayer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Após erro do handler, o lock permanecia InProgress pelo TTL (24h). Redeliveries SQS recebiam InProgress e falhavam sem reexecutar o handler, podendo enviar mensagens à DLQ sem processamento bem-sucedido. Adiciona IdempotencyStore::release e chama no IdempotencyLayer em falhas do handler ou do complete. Testes de regressão cobrem redelivery SQS. Co-authored-by: Jaime Basso --- CHANGELOG.md | 1 + serverust-events/src/sqs/layers.rs | 22 +++++++- serverust-events/tests/sqs_idempotency.rs | 56 +++++++++++++++++-- serverust-telemetry/src/idempotency.rs | 53 ++++++++++++++++++ serverust-telemetry/tests/idempotency_lock.rs | 25 +++++++++ 5 files changed, 149 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf86a71..28e963b 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`: 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 diff --git a/serverust-events/src/sqs/layers.rs b/serverust-events/src/sqs/layers.rs index b096a7e..9fbe4ba 100644 --- a/serverust-events/src/sqs/layers.rs +++ b/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 } diff --git a/serverust-events/tests/sqs_idempotency.rs b/serverust-events/tests/sqs_idempotency.rs index b39df03..e3588f9 100644 --- a/serverust-events/tests/sqs_idempotency.rs +++ b/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 = 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", ); } diff --git a/serverust-telemetry/src/idempotency.rs b/serverust-telemetry/src/idempotency.rs index dd39aef..52b8e92 100644 --- a/serverust-telemetry/src/idempotency.rs +++ b/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())), + } + } } } diff --git a/serverust-telemetry/tests/idempotency_lock.rs b/serverust-telemetry/tests/idempotency_lock.rs index 91e9c32..0c4e695 100644 --- a/serverust-telemetry/tests/idempotency_lock.rs +++ b/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", + ); +}