From fd5799aa34eac1c3263f118feda54c8c55086c0a Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Fri, 30 Oct 2026 13:37:28 +0000 Subject: [PATCH 1/2] anchor lock-expiration validation to the first claim Measure the concurrent checks from the first completed claim rather than the startup barrier. Connection acquisition could otherwise consume the expiry margin and make a correct provider fail validation. Add a delayed-fetch regression and compare the recovered token with the original claim's token. Lease durations and timing margins are unchanged. --- CHANGELOG.md | 3 ++ src/provider_validation/instance_locking.rs | 27 ++++++------- tests/common/long_polling.rs | 2 + tests/long_poll_tests.rs | 45 ++++++++++++++++++++- 4 files changed, 61 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91d9615..f2f2733 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 flag. (#74) - Relative SQLite queue delays saturate instead of overflowing their timestamp representation. (#74) +- Provider validation's multi-threaded lock-expiration recovery test measures + its waits from the first completed claim, so connection warm-up cannot consume + the expiry margin. The lease duration and margin are unchanged. ## [0.1.32] - 2026-10-11 diff --git a/src/provider_validation/instance_locking.rs b/src/provider_validation/instance_locking.rs index 72903ef..9ff0c5f 100644 --- a/src/provider_validation/instance_locking.rs +++ b/src/provider_validation/instance_locking.rs @@ -691,21 +691,20 @@ pub async fn test_multi_threaded_lock_expiration_recovery(fa // Capture timeout value before spawning tasks let lock_timeout = factory.lock_timeout(); - // Barrier ensures all threads start at exactly the same time, - // eliminating race conditions from connection pool cold-start latency - let barrier = Arc::new(tokio::sync::Barrier::new(3)); + // Connection acquisition can exceed the expiry margin. Measure from the + // completed claim, not from when the competing tasks were spawned. + let (claimed_tx, claimed_rx) = tokio::sync::watch::channel(None); // Thread 1: Fetch and hold lock (don't ack) - simulates crashed worker let provider1 = provider.clone(); - let barrier1 = barrier.clone(); let handle1 = tokio::spawn(async move { - barrier1.wait().await; let (item, lock_token, _attempt_count) = provider1 .fetch_orchestration_item(lock_timeout, Duration::ZERO, None) .await .unwrap() .unwrap(); assert_eq!(item.instance, "expiration-instance"); + claimed_tx.send(Some(tokio::time::Instant::now())).unwrap(); // Hold lock but don't ack - simulate crashed worker tokio::time::sleep(lock_timeout + Duration::from_millis(200)).await; lock_token @@ -713,11 +712,10 @@ pub async fn test_multi_threaded_lock_expiration_recovery(fa // Thread 2: Try to fetch while lock is held (should fail) let provider2 = provider.clone(); - let barrier2 = barrier.clone(); + let mut claimed2 = claimed_rx.clone(); let handle2 = tokio::spawn(async move { - barrier2.wait().await; - // 200ms delay gives thread 1 time to acquire lock - tokio::time::sleep(Duration::from_millis(200)).await; + let claimed_at = claimed2.wait_for(Option::is_some).await.unwrap().unwrap(); + tokio::time::sleep_until(claimed_at + Duration::from_millis(200)).await; let result = provider2 .fetch_orchestration_item(lock_timeout, Duration::ZERO, None) .await @@ -728,11 +726,10 @@ pub async fn test_multi_threaded_lock_expiration_recovery(fa // Thread 3: Wait for expiration, then fetch (should succeed - lock recovery) let provider3 = provider.clone(); - let barrier3 = barrier.clone(); + let mut claimed3 = claimed_rx; let handle3 = tokio::spawn(async move { - barrier3.wait().await; - // Wait for lock to expire (lock_timeout + margin) - tokio::time::sleep(lock_timeout + Duration::from_millis(100)).await; + let claimed_at = claimed3.wait_for(Option::is_some).await.unwrap().unwrap(); + tokio::time::sleep_until(claimed_at + lock_timeout + Duration::from_millis(100)).await; provider3 .fetch_orchestration_item(lock_timeout, Duration::ZERO, None) .await @@ -741,13 +738,13 @@ pub async fn test_multi_threaded_lock_expiration_recovery(fa // Wait for all threads let (lock_token1, result2, result3) = futures::future::join3(handle1, handle2, handle3).await; - let _lock_token1 = lock_token1.unwrap(); + let lock_token1 = lock_token1.unwrap(); assert!(result2.unwrap().is_none()); let (item3, lock_token3, _attempt_count3) = result3.unwrap().unwrap(); // Thread 3 should have successfully acquired the lock after expiration assert_eq!(item3.instance, "expiration-instance"); - assert_ne!(lock_token3, "expired-token"); // Should be a new token + assert_ne!(lock_token3, lock_token1); tracing::info!("✓ Test passed: multi-threaded lock expiration recovery verified"); } diff --git a/tests/common/long_polling.rs b/tests/common/long_polling.rs index c0193ed..8114762 100644 --- a/tests/common/long_polling.rs +++ b/tests/common/long_polling.rs @@ -23,6 +23,7 @@ use tokio_util::sync::CancellationToken; #[derive(Clone, Copy, PartialEq, Eq)] pub enum ProviderOperation { + BeforeOrchestrationFetch, AfterOrchestrationFetch, AfterWorkFetch, BeforeOrchestrationAck, @@ -136,6 +137,7 @@ impl Provider for LongPollingSqliteProvider { poll_timeout: Duration, _filter: Option<&DispatcherCapabilityFilter>, ) -> Result, ProviderError> { + self.wait_at(ProviderOperation::BeforeOrchestrationFetch).await; self.poll_until(poll_timeout, ProviderOperation::AfterOrchestrationFetch, || { self.inner.fetch_orchestration_item(lock_timeout, Duration::ZERO, None) }) diff --git a/tests/long_poll_tests.rs b/tests/long_poll_tests.rs index 0539453..8fba7a8 100644 --- a/tests/long_poll_tests.rs +++ b/tests/long_poll_tests.rs @@ -14,10 +14,53 @@ use std::time::{Duration, Instant}; mod common; -use common::long_polling::LongPollingSqliteProvider; +use common::long_polling::{LongPollingSqliteProvider, ProviderGate, ProviderOperation}; // --- Tests --- +#[tokio::test] +async fn lock_expiration_validation_waits_for_the_first_claim() { + use duroxide::provider_validation::{ProviderFactory, instance_locking}; + use duroxide::providers::sqlite::SqliteProvider; + + struct GatedFactory(Arc); + + #[async_trait::async_trait] + impl ProviderFactory for GatedFactory { + async fn create_provider(&self) -> Arc { + self.0.clone() + } + + fn lock_timeout(&self) -> Duration { + Duration::from_millis(500) + } + } + + let gate = Arc::new(ProviderGate::default()); + let store = Arc::new(SqliteProvider::new_in_memory().await.unwrap()); + // Isolate claim timing from SQLite's concurrent-writer behavior. + let mut held_connections = Vec::new(); + for _ in 1..store.get_pool().options().get_max_connections() { + held_connections.push(store.get_pool().acquire().await.unwrap()); + } + let provider = Arc::new( + LongPollingSqliteProvider::new(store).with_gate(ProviderOperation::BeforeOrchestrationFetch, gate.clone()), + ); + let validation = tokio::spawn(async move { + instance_locking::test_multi_threaded_lock_expiration_recovery(&GatedFactory(provider)).await; + }); + tokio::time::timeout(Duration::from_secs(2), gate.wait_until_entered()) + .await + .expect("the first fetch should reach the gate"); + // Exceed both the lease and the recovery margin before permitting a claim. + tokio::time::sleep(Duration::from_millis(700)).await; + gate.open(); + tokio::time::timeout(Duration::from_secs(3), validation) + .await + .expect("claim-relative recovery should finish") + .expect("connection acquisition time must not consume the lease-expiry margin"); +} + /// Test 1: Verify fetch waits for the full duration if no work exists #[tokio::test] async fn test_long_poll_waits_for_timeout() { From 5728a64e02dc98a71a644e6678278cc447874da5 Mon Sep 17 00:00:00 2001 From: Thom Chiovoloni Date: Tue, 3 Nov 2026 15:28:49 +0000 Subject: [PATCH 2/2] docs: add PR reference to lock-expiration changelog entry Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f2f2733..8534ea1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -50,7 +50,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 representation. (#74) - Provider validation's multi-threaded lock-expiration recovery test measures its waits from the first completed claim, so connection warm-up cannot consume - the expiry margin. The lease duration and margin are unchanged. + the expiry margin. The lease duration and margin are unchanged. (#76) ## [0.1.32] - 2026-10-11