agpm-cli 0.4.14

AGent Package Manager - A Git-based package manager for coding agents
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
//! File locking utilities for cache operations.
//!
//! This module provides thread-safe and process-safe file locking for cache directories
//! to prevent corruption during concurrent cache operations. The locks are automatically
//! released when the lock object is dropped.
//!
//! # Async Safety
//!
//! All file operations are wrapped in `spawn_blocking` to avoid blocking the tokio
//! runtime. This is critical for preventing worker thread starvation under high
//! parallelism with slow I/O (e.g., network-attached storage).

use crate::constants::{MAX_BACKOFF_DELAY_MS, STARTING_BACKOFF_DELAY_MS, default_lock_timeout};
use anyhow::{Context, Result};
use fs4::fs_std::FileExt;
use std::fs::{File, OpenOptions};
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio_retry::strategy::ExponentialBackoff;
use tracing::debug;

/// A file lock for cache operations.
///
/// The lock is held for the lifetime of this struct and automatically
/// released when dropped. Lock acquisition and release are tracked
/// for deadlock detection.
#[derive(Debug)]
pub struct CacheLock {
    /// The file handle - lock is released when this is dropped
    _file: Arc<File>,
    /// Name of the lock for tracing
    lock_name: String,
}

impl Drop for CacheLock {
    fn drop(&mut self) {
        debug!(lock_name = %self.lock_name, "File lock released");
    }
}

impl CacheLock {
    /// Acquires an exclusive lock for a specific source in the cache directory.
    ///
    /// Creates and acquires an exclusive file lock for the specified source name.
    /// Uses non-blocking lock attempts with exponential backoff and timeout.
    ///
    /// # Lock File Management
    ///
    /// 1. Creates `.locks/` directory if needed
    /// 2. Creates `{source_name}.lock` file
    /// 3. Acquires exclusive access via OS file locking
    /// 4. Keeps file handle open to maintain lock
    ///
    /// # Behavior
    ///
    /// - **Timeout**: 30-second default (configurable via `acquire_with_timeout`)
    /// - **Non-blocking**: `try_lock_exclusive()` in async retry loop
    /// - **Backoff**: 10ms → 20ms → 40ms... up to 500ms max
    /// - **Fair access**: FIFO order typically
    /// - **Interruptible**: Process signals work
    ///
    /// # Lock File Location
    ///
    /// Format: `{cache_dir}/.locks/{source_name}.lock`
    ///
    /// Example: `~/.agpm/cache/.locks/community.lock`
    ///
    /// # Errors
    ///
    /// - Permission denied
    /// - Disk space exhausted
    /// - Timeout acquiring lock
    ///
    /// # Platform Support
    ///
    /// - **Windows**: Win32 `LockFile` API
    /// - **Unix**: POSIX `fcntl()` locking
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use agpm_cli::cache::lock::CacheLock;
    /// use std::path::Path;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let cache_dir = Path::new("/tmp/cache");
    /// let lock = CacheLock::acquire(cache_dir, "my-source").await?;
    /// // Lock released on drop
    /// # Ok(())
    /// # }
    /// ```
    pub async fn acquire(cache_dir: &Path, source_name: &str) -> Result<Self> {
        Self::acquire_with_timeout(cache_dir, source_name, default_lock_timeout()).await
    }

    /// Acquires an exclusive lock with a specified timeout.
    ///
    /// Uses exponential backoff (10ms → 500ms) without blocking the async runtime.
    ///
    /// # Errors
    ///
    /// Returns timeout error if lock cannot be acquired within the specified duration.
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use agpm_cli::cache::lock::CacheLock;
    /// use std::time::Duration;
    /// use std::path::Path;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let cache_dir = Path::new("/tmp/cache");
    /// let lock = CacheLock::acquire_with_timeout(
    ///     cache_dir, "my-source", Duration::from_secs(10)
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn acquire_with_timeout(
        cache_dir: &Path,
        source_name: &str,
        timeout: std::time::Duration,
    ) -> Result<Self> {
        use tokio::fs;

        let lock_name = format!("file:{}", source_name);
        debug!(lock_name = %lock_name, "Waiting for file lock");

        // Create locks directory if it doesn't exist
        let locks_dir = cache_dir.join(".locks");
        fs::create_dir_all(&locks_dir).await.with_context(|| {
            format!("Failed to create locks directory: {}", locks_dir.display())
        })?;

        // Create lock file path
        let lock_path = locks_dir.join(format!("{source_name}.lock"));

        // CRITICAL: Use spawn_blocking for file open to avoid blocking tokio runtime
        // This is essential for preventing worker thread starvation under slow I/O
        let lock_path_clone = lock_path.clone();
        let file = tokio::task::spawn_blocking(move || {
            OpenOptions::new().create(true).write(true).truncate(false).open(&lock_path_clone)
        })
        .await
        .with_context(|| "spawn_blocking panicked")?
        .with_context(|| format!("Failed to open lock file: {}", lock_path.display()))?;

        // Wrap file in Arc for sharing with spawn_blocking
        let file = Arc::new(file);

        // Acquire exclusive lock with timeout and exponential backoff
        let start = std::time::Instant::now();

        // Create exponential backoff strategy with platform-specific tuning:
        // - Windows: 25ms, 50ms, 100ms, 200ms (faster retries for AV delays)
        // - Unix: 10ms, 20ms, 40ms, ... 500ms (standard backoff)
        let backoff = ExponentialBackoff::from_millis(STARTING_BACKOFF_DELAY_MS)
            .max_delay(Duration::from_millis(MAX_BACKOFF_DELAY_MS));

        // Add jitter to prevent thundering herd when multiple processes retry simultaneously
        let mut rng_state: u64 = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(12345);

        for delay in backoff {
            // Simple xorshift for jitter: adds 0-25% random variation
            rng_state ^= rng_state << 13;
            rng_state ^= rng_state >> 7;
            rng_state ^= rng_state << 17;
            let jitter_factor = 1.0 + (rng_state % 25) as f64 / 100.0;
            let jittered_delay =
                Duration::from_millis((delay.as_millis() as f64 * jitter_factor) as u64);
            // CRITICAL: Use spawn_blocking for try_lock_exclusive to avoid blocking tokio runtime
            let file_clone = Arc::clone(&file);
            let lock_result = tokio::task::spawn_blocking(move || file_clone.try_lock_exclusive())
                .await
                .with_context(|| "spawn_blocking panicked")?;

            match lock_result {
                Ok(true) => {
                    debug!(
                        lock_name = %lock_name,
                        wait_ms = start.elapsed().as_millis(),
                        "File lock acquired"
                    );
                    return Ok(Self {
                        _file: file,
                        lock_name,
                    });
                }
                Ok(false) | Err(_) => {
                    // Check remaining time before sleeping to avoid exceeding timeout
                    let remaining = timeout.saturating_sub(start.elapsed());
                    if remaining.is_zero() {
                        return Err(anyhow::anyhow!(
                            "Timeout acquiring lock for '{}' after {:?}",
                            source_name,
                            timeout
                        ));
                    }
                    // Sleep for the shorter of jittered delay or remaining time
                    tokio::time::sleep(jittered_delay.min(remaining)).await;
                }
            }
        }

        // If backoff iterator exhausted without acquiring lock, return timeout error
        Err(anyhow::anyhow!("Timeout acquiring lock for '{}' after {:?}", source_name, timeout))
    }

    /// Acquires a shared (read) lock for a specific source in the cache directory.
    ///
    /// Multiple processes can hold shared locks simultaneously, but a shared lock
    /// blocks exclusive lock acquisition. Use this for operations that can safely
    /// run in parallel, like worktree creation (each SHA writes to a different subdir).
    ///
    /// # Lock Semantics
    ///
    /// - **Shared locks**: Multiple holders allowed simultaneously
    /// - **Exclusive locks**: Blocked while any shared lock is held
    /// - **Shared + Exclusive**: Shared lock blocks until exclusive is released
    ///
    /// # Use Cases
    ///
    /// - Worktree creation: Multiple SHAs can create worktrees in parallel
    /// - Read-only operations on shared state
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// use agpm_cli::cache::lock::CacheLock;
    /// use std::path::Path;
    /// # async fn example() -> anyhow::Result<()> {
    /// # let cache_dir = Path::new("/tmp/cache");
    /// // Multiple processes can hold this simultaneously
    /// let lock = CacheLock::acquire_shared(cache_dir, "bare-worktree-owner_repo").await?;
    /// // Lock released on drop
    /// # Ok(())
    /// # }
    /// ```
    pub async fn acquire_shared(cache_dir: &Path, source_name: &str) -> Result<Self> {
        Self::acquire_shared_with_timeout(cache_dir, source_name, default_lock_timeout()).await
    }

    /// Acquires a shared (read) lock with a specified timeout.
    ///
    /// Uses exponential backoff (10ms → 500ms) without blocking the async runtime.
    ///
    /// # Errors
    ///
    /// Returns timeout error if lock cannot be acquired within the specified duration.
    pub async fn acquire_shared_with_timeout(
        cache_dir: &Path,
        source_name: &str,
        timeout: std::time::Duration,
    ) -> Result<Self> {
        use tokio::fs;

        let lock_name = format!("file-shared:{}", source_name);
        debug!(lock_name = %lock_name, "Waiting for shared file lock");

        // Create locks directory if it doesn't exist
        let locks_dir = cache_dir.join(".locks");
        fs::create_dir_all(&locks_dir).await.with_context(|| {
            format!("Failed to create locks directory: {}", locks_dir.display())
        })?;

        // Create lock file path
        let lock_path = locks_dir.join(format!("{source_name}.lock"));

        // CRITICAL: Use spawn_blocking for file open to avoid blocking tokio runtime
        let lock_path_clone = lock_path.clone();
        let file = tokio::task::spawn_blocking(move || {
            OpenOptions::new().create(true).write(true).truncate(false).open(&lock_path_clone)
        })
        .await
        .with_context(|| "spawn_blocking panicked")?
        .with_context(|| format!("Failed to open lock file: {}", lock_path.display()))?;

        // Wrap file in Arc for sharing with spawn_blocking
        let file = Arc::new(file);

        // Acquire shared lock with timeout and exponential backoff
        let start = std::time::Instant::now();

        let backoff = ExponentialBackoff::from_millis(STARTING_BACKOFF_DELAY_MS)
            .max_delay(Duration::from_millis(MAX_BACKOFF_DELAY_MS));

        // Add jitter to prevent thundering herd
        let mut rng_state: u64 = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_nanos() as u64)
            .unwrap_or(12345);

        for delay in backoff {
            rng_state ^= rng_state << 13;
            rng_state ^= rng_state >> 7;
            rng_state ^= rng_state << 17;
            let jitter_factor = 1.0 + (rng_state % 25) as f64 / 100.0;
            let jittered_delay =
                Duration::from_millis((delay.as_millis() as f64 * jitter_factor) as u64);

            // CRITICAL: Use spawn_blocking for try_lock_shared to avoid blocking tokio runtime
            // Use FileExt trait method explicitly to avoid std::fs::File::try_lock_shared
            let file_clone = Arc::clone(&file);
            let lock_result =
                tokio::task::spawn_blocking(move || FileExt::try_lock_shared(file_clone.as_ref()))
                    .await
                    .with_context(|| "spawn_blocking panicked")?;

            match lock_result {
                Ok(true) => {
                    debug!(
                        lock_name = %lock_name,
                        wait_ms = start.elapsed().as_millis(),
                        "Shared file lock acquired"
                    );
                    return Ok(Self {
                        _file: file,
                        lock_name,
                    });
                }
                Ok(false) | Err(_) => {
                    // Check remaining time before sleeping
                    let remaining = timeout.saturating_sub(start.elapsed());
                    if remaining.is_zero() {
                        return Err(anyhow::anyhow!(
                            "Timeout acquiring shared lock for '{}' after {:?}",
                            source_name,
                            timeout
                        ));
                    }
                    tokio::time::sleep(jittered_delay.min(remaining)).await;
                }
            }
        }

        Err(anyhow::anyhow!(
            "Timeout acquiring shared lock for '{}' after {:?}",
            source_name,
            timeout
        ))
    }
}

/// Cleans up stale lock files in the cache directory.
///
/// This function removes lock files that are older than the specified TTL.
/// It's useful for cleaning up after crashes or processes that didn't
/// properly release their locks.
///
/// # Parameters
///
/// * `cache_dir` - The cache directory containing the .locks subdirectory
/// * `ttl_seconds` - Time-to-live in seconds for lock files
///
/// # Returns
///
/// Returns the number of lock files that were removed.
///
/// # Errors
///
/// Returns an error if unable to read the locks directory or access lock file metadata
pub async fn cleanup_stale_locks(cache_dir: &Path, ttl_seconds: u64) -> Result<usize> {
    use std::time::{Duration, SystemTime};
    use tokio::fs;

    let locks_dir = cache_dir.join(".locks");
    if !locks_dir.exists() {
        return Ok(0);
    }

    let mut removed_count = 0;
    let now = SystemTime::now();
    let ttl_duration = Duration::from_secs(ttl_seconds);

    let mut entries = fs::read_dir(&locks_dir).await.context("Failed to read locks directory")?;

    while let Some(entry) = entries.next_entry().await? {
        let path = entry.path();

        // Only process .lock files
        if path.extension().and_then(|s| s.to_str()) != Some("lock") {
            continue;
        }

        // Check file age
        let Ok(metadata) = fs::metadata(&path).await else {
            continue; // Skip if we can't read metadata
        };

        let Ok(modified) = metadata.modified() else {
            continue; // Skip if we can't get modification time
        };

        // Remove if older than TTL
        if let Ok(age) = now.duration_since(modified)
            && age > ttl_duration
        {
            // Try to remove the file (it might be locked by another process)
            if fs::remove_file(&path).await.is_ok() {
                removed_count += 1;
            }
        }
    }

    Ok(removed_count)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    #[tokio::test]
    async fn test_cache_lock_acquire_and_release() {
        let temp_dir = TempDir::new().unwrap();
        let cache_dir = temp_dir.path();

        // Acquire lock
        let lock = CacheLock::acquire(cache_dir, "test_source").await.unwrap();

        // Verify lock file was created
        let lock_path = cache_dir.join(".locks").join("test_source.lock");
        assert!(lock_path.exists());

        // Drop the lock
        drop(lock);

        // Lock file should still exist (we don't delete it)
        assert!(lock_path.exists());
    }

    #[tokio::test]
    async fn test_cache_lock_creates_locks_directory() {
        let temp_dir = TempDir::new().unwrap();
        let cache_dir = temp_dir.path();

        // Locks directory shouldn't exist initially
        let locks_dir = cache_dir.join(".locks");
        assert!(!locks_dir.exists());

        // Acquire lock - should create directory
        let lock = CacheLock::acquire(cache_dir, "test").await.unwrap();

        // Verify locks directory was created
        assert!(locks_dir.exists());
        assert!(locks_dir.is_dir());

        // Explicitly drop the lock to release the file handle before TempDir cleanup
        drop(lock);
    }

    #[tokio::test]
    async fn test_cache_lock_exclusive_blocking() {
        use std::sync::Arc;
        use std::time::{Duration, Instant};
        use tokio::sync::Barrier;

        let temp_dir = TempDir::new().unwrap();
        let cache_dir = Arc::new(temp_dir.path().to_path_buf());
        let barrier = Arc::new(Barrier::new(2));

        let cache_dir1 = cache_dir.clone();
        let barrier1 = barrier.clone();

        // Task 1: Acquire lock and hold it
        let handle1 = tokio::spawn(async move {
            let _lock = CacheLock::acquire(&cache_dir1, "exclusive_test").await.unwrap();
            barrier1.wait().await; // Signal that lock is acquired
            tokio::time::sleep(Duration::from_millis(100)).await; // Hold lock
            // Lock released on drop
        });

        let cache_dir2 = cache_dir.clone();

        // Task 2: Try to acquire same lock (should block)
        let handle2 = tokio::spawn(async move {
            barrier.wait().await; // Wait for first task to acquire lock
            let start = Instant::now();
            let _lock = CacheLock::acquire(&cache_dir2, "exclusive_test").await.unwrap();
            let elapsed = start.elapsed();

            // Should have blocked for at least 50ms (less than 100ms due to timing)
            assert!(elapsed >= Duration::from_millis(50));
        });

        handle1.await.unwrap();
        handle2.await.unwrap();
    }

    #[tokio::test]
    async fn test_cache_lock_different_sources_dont_block() {
        use std::sync::Arc;
        use std::time::{Duration, Instant};
        use tokio::sync::Barrier;

        let temp_dir = TempDir::new().unwrap();
        let cache_dir = Arc::new(temp_dir.path().to_path_buf());
        let barrier = Arc::new(Barrier::new(2));

        let cache_dir1 = cache_dir.clone();
        let barrier1 = barrier.clone();

        // Task 1: Lock source1
        let handle1 = tokio::spawn(async move {
            let _lock = CacheLock::acquire(&cache_dir1, "source1").await.unwrap();
            barrier1.wait().await;
            tokio::time::sleep(Duration::from_millis(100)).await;
        });

        let cache_dir2 = cache_dir.clone();

        // Task 2: Lock source2 (different source, shouldn't block)
        let handle2 = tokio::spawn(async move {
            barrier.wait().await;
            let start = Instant::now();
            let _lock = CacheLock::acquire(&cache_dir2, "source2").await.unwrap();
            let elapsed = start.elapsed();

            // Should not block (complete quickly)
            // Increased timeout for slower systems while still ensuring no blocking
            assert!(
                elapsed < Duration::from_millis(200),
                "Lock acquisition took {:?}, expected < 200ms for non-blocking operation",
                elapsed
            );
        });

        handle1.await.unwrap();
        handle2.await.unwrap();
    }

    #[tokio::test]
    async fn test_cache_lock_path_with_special_characters() {
        let temp_dir = TempDir::new().unwrap();
        let cache_dir = temp_dir.path();

        // Test with various special characters in source name
        let special_names = vec![
            "source-with-dash",
            "source_with_underscore",
            "source.with.dots",
            "source@special",
        ];

        for name in special_names {
            let lock = CacheLock::acquire(cache_dir, name).await.unwrap();
            let expected_path = cache_dir.join(".locks").join(format!("{name}.lock"));
            assert!(expected_path.exists());
            drop(lock);
        }
    }

    #[tokio::test]
    async fn test_cache_lock_acquire_timeout() {
        let temp_dir = TempDir::new().unwrap();
        let cache_dir = temp_dir.path();

        // First lock succeeds
        let _lock1 = CacheLock::acquire(cache_dir, "test-source").await.unwrap();

        // Second lock attempt should timeout quickly
        let start = std::time::Instant::now();
        let result =
            CacheLock::acquire_with_timeout(cache_dir, "test-source", Duration::from_millis(100))
                .await;

        let elapsed = start.elapsed();

        // Verify timeout occurred
        assert!(result.is_err(), "Expected timeout error");

        // Verify error message mentions timeout
        match result {
            Ok(_) => panic!("Expected timeout error, but got success"),
            Err(error) => {
                let error_msg = error.to_string();
                assert!(
                    error_msg.contains("Timeout") || error_msg.contains("timeout"),
                    "Error message should mention timeout: {}",
                    error_msg
                );
                assert!(
                    error_msg.contains("test-source"),
                    "Error message should include source name: {}",
                    error_msg
                );
            }
        }

        // Verify timeout happened around the expected time (with some tolerance)
        // Should be ~100ms, allow 50-500ms range to accommodate slow CI runners
        assert!(elapsed >= Duration::from_millis(50), "Timeout too quick: {:?}", elapsed);
        assert!(elapsed < Duration::from_millis(500), "Timeout too slow: {:?}", elapsed);
    }

    #[tokio::test]
    async fn test_cache_lock_acquire_timeout_succeeds_eventually() {
        let temp_dir = TempDir::new().unwrap();
        let cache_dir = temp_dir.path();

        // Acquire lock and release it after 50ms
        let cache_dir_clone = cache_dir.to_path_buf();
        let handle = tokio::spawn(async move {
            let lock = CacheLock::acquire(&cache_dir_clone, "test-source").await.unwrap();
            tokio::time::sleep(Duration::from_millis(50)).await;
            drop(lock); // Release lock
        });

        // Wait a bit for the first lock to be acquired
        tokio::time::sleep(Duration::from_millis(10)).await;

        // This should succeed after the first lock is released
        // Use 500ms timeout to give plenty of time
        let result =
            CacheLock::acquire_with_timeout(cache_dir, "test-source", Duration::from_millis(500))
                .await;

        assert!(result.is_ok(), "Lock should be acquired after first one is released");

        // Clean up spawned task
        handle.await.unwrap();
    }

    #[tokio::test]
    async fn test_shared_locks_dont_block_each_other() {
        use std::sync::Arc;
        use std::time::{Duration, Instant};
        use tokio::sync::Barrier;

        let temp_dir = TempDir::new().unwrap();
        let cache_dir = Arc::new(temp_dir.path().to_path_buf());
        let barrier = Arc::new(Barrier::new(2));

        let cache_dir1 = cache_dir.clone();
        let barrier1 = barrier.clone();

        // Task 1: Acquire shared lock and hold it
        let handle1 = tokio::spawn(async move {
            let _lock = CacheLock::acquire_shared(&cache_dir1, "shared_test").await.unwrap();
            barrier1.wait().await; // Signal that lock is acquired
            tokio::time::sleep(Duration::from_millis(100)).await; // Hold lock
        });

        let cache_dir2 = cache_dir.clone();

        // Task 2: Acquire another shared lock on same resource (should NOT block)
        let handle2 = tokio::spawn(async move {
            barrier.wait().await; // Wait for first task to acquire lock
            let start = Instant::now();
            let _lock = CacheLock::acquire_shared(&cache_dir2, "shared_test").await.unwrap();
            let elapsed = start.elapsed();

            // Should complete quickly since shared locks don't block each other
            assert!(
                elapsed < Duration::from_millis(200),
                "Shared lock took {:?}, expected < 200ms (no blocking)",
                elapsed
            );
        });

        handle1.await.unwrap();
        handle2.await.unwrap();
    }

    #[tokio::test]
    async fn test_exclusive_blocks_shared() {
        use std::sync::Arc;
        use std::time::{Duration, Instant};
        use tokio::sync::Barrier;

        let temp_dir = TempDir::new().unwrap();
        let cache_dir = Arc::new(temp_dir.path().to_path_buf());
        let barrier = Arc::new(Barrier::new(2));

        let cache_dir1 = cache_dir.clone();
        let barrier1 = barrier.clone();

        // Task 1: Acquire EXCLUSIVE lock and hold it
        let handle1 = tokio::spawn(async move {
            let _lock = CacheLock::acquire(&cache_dir1, "exclusive_shared_test").await.unwrap();
            barrier1.wait().await;
            tokio::time::sleep(Duration::from_millis(100)).await;
        });

        let cache_dir2 = cache_dir.clone();

        // Task 2: Try to acquire SHARED lock (should block until exclusive releases)
        let handle2 = tokio::spawn(async move {
            barrier.wait().await;
            let start = Instant::now();
            let _lock =
                CacheLock::acquire_shared(&cache_dir2, "exclusive_shared_test").await.unwrap();
            let elapsed = start.elapsed();

            // Should have blocked for at least 50ms
            assert!(
                elapsed >= Duration::from_millis(50),
                "Shared lock should have blocked: {:?}",
                elapsed
            );
        });

        handle1.await.unwrap();
        handle2.await.unwrap();
    }

    #[tokio::test]
    async fn test_shared_blocks_exclusive() {
        use std::sync::Arc;
        use std::time::{Duration, Instant};
        use tokio::sync::Barrier;

        let temp_dir = TempDir::new().unwrap();
        let cache_dir = Arc::new(temp_dir.path().to_path_buf());
        let barrier = Arc::new(Barrier::new(2));

        let cache_dir1 = cache_dir.clone();
        let barrier1 = barrier.clone();

        // Task 1: Acquire SHARED lock and hold it
        let handle1 = tokio::spawn(async move {
            let _lock =
                CacheLock::acquire_shared(&cache_dir1, "shared_exclusive_test").await.unwrap();
            barrier1.wait().await;
            tokio::time::sleep(Duration::from_millis(100)).await;
        });

        let cache_dir2 = cache_dir.clone();

        // Task 2: Try to acquire EXCLUSIVE lock (should block until shared releases)
        let handle2 = tokio::spawn(async move {
            barrier.wait().await;
            let start = Instant::now();
            let _lock = CacheLock::acquire(&cache_dir2, "shared_exclusive_test").await.unwrap();
            let elapsed = start.elapsed();

            // Should have blocked for at least 50ms
            assert!(
                elapsed >= Duration::from_millis(50),
                "Exclusive lock should have blocked: {:?}",
                elapsed
            );
        });

        handle1.await.unwrap();
        handle2.await.unwrap();
    }
}