From 57131ceeb235789dead831d545aebe2030591848 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Sun, 14 Jun 2026 02:05:33 +0000 Subject: [PATCH] =?UTF-8?q?fix(events):=20liberar=20lock=20InProgress=20ap?= =?UTF-8?q?=C3=B3s=20falha=20do=20IdempotencyLayer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Após falha do handler ou de complete(), o lock InProgress permanecia ativo por até 24h (TTL default). Redeliveries do SQS recebiam InProgress sem executar o handler, acumulando tentativas até DLQ sem reprocessamento real. Adiciona IdempotencyStore::release e chama no IdempotencyLayer em falhas. Testes de regressão cobrem redelivery dentro do TTL. Co-authored-by: Jaime Basso --- CHANGELOG.md | 1 + serverust-events/src/sqs/layers.rs | 14 ++++- serverust-events/tests/sqs_idempotency.rs | 25 ++++++--- serverust-telemetry/src/idempotency.rs | 51 +++++++++++++++++++ serverust-telemetry/tests/idempotency_lock.rs | 24 ++++++--- 5 files changed, 97 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bf86a71..fbbbed8 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 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 index b096a7e..8a8cb43 100644 --- 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 index b39df03..8b0d1da 100644 --- 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 index dd39aef..d21d338 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` 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 index 91e9c32..3330549 100644 --- 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(_))); }