Anchor lock-expiration validation to the first claim

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. (#76)
 
 ## [0.1.32] - 2026-10-11
 

src/provider_validation/instance_locking.rs

@@ -691,33 +691,31 @@ pub async fn test_multi_threaded_lock_expiration_recovery<F: ProviderFactory>(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
     });
 
     // 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<F: ProviderFactory>(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<F: ProviderFactory>(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");
 }

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<Option<(OrchestrationItem, String, u32)>, ProviderError> {
+        self.wait_at(ProviderOperation::BeforeOrchestrationFetch).await;
         self.poll_until(poll_timeout, ProviderOperation::AfterOrchestrationFetch, || {
             self.inner.fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
         })

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<dyn Provider>);
+
+    #[async_trait::async_trait]
+    impl ProviderFactory for GatedFactory {
+        async fn create_provider(&self) -> Arc<dyn Provider> {
+            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() {