lazily 0.22.0

Lazy reactive signals with dependency tracking and cache invalidation
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

//! Deterministic regression coverage for the `#k03k` `get_async` resolve-loop
//! windows in `src/async_context.rs`.
//!
//! Loom cannot model this code path — `AsyncContext` is built on tokio's async
//! executor and `tokio::sync::watch`, while Loom only shims synchronous
//! `loom::sync` primitives (which is why `thread_safe.rs` has a Loom model and
//! this module does not). Instead we drive the two race windows directly:
//!
//!   * **Window 1** — the slot transitions `Computing/Empty -> Resolved` between
//!     the lock-free fast-path `get()` check and the re-lock, so `get_async`
//!     must return through the `Resolved`-after-re-lock arm. The gap is purely
//!     synchronous (no `.await`), so cooperative scheduling cannot force it; we
//!     use the `instrumentation`-gated one-shot seam to resolve the slot inside
//!     the gap deterministically and assert the race arm was taken.
//!   * **Window 2** — the notifier's `watch` senders all drop without a final
//!     `Resolved` send when an in-flight compute is superseded; `get_async` must
//!     re-resolve rather than panic. This is already exercised by
//!     `async_stress.rs::get_async_waiter_cancellation_and_stale_completion_keep_latest`
//!     (oneshot-gated supersede); the test below pins the same guarantee with a
//!     focused, named assertion.

#![cfg(all(feature = "async", feature = "instrumentation"))]

use lazily::AsyncContext;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, oneshot};

/// Window 1: resolve the slot inside `get_async`'s fast-path/re-lock gap and
/// assert the reader returns through the `Resolved`-after-re-lock arm.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn window1_resolved_between_fastpath_and_relock() {
    let ctx = Arc::new(AsyncContext::new());
    let slot = ctx.computed_async(|_| async { 42i32 });

    // One-shot seam: when reader A parks in the window-1 gap, resolve the slot
    // from a separate reader (the hook has already disarmed itself, so reader B
    // takes the normal compute path), then wait until the value is published so
    // reader A re-locks straight into the `Resolved` arm.
    let ctx_hook = ctx.clone();
    ctx.__install_window1_hook(Arc::new(move || {
        let ctx_hook = ctx_hook.clone();
        Box::pin(async move {
            let resolver = tokio::spawn({
                let ctx_hook = ctx_hook.clone();
                async move { ctx_hook.get_async(&slot).await }
            });
            assert_eq!(resolver.await.unwrap(), 42, "resolver should compute slot");

            let deadline = Instant::now() + Duration::from_secs(2);
            while ctx_hook.get(&slot).is_none() {
                assert!(Instant::now() < deadline, "slot never resolved in gap");
                tokio::task::yield_now().await;
            }
        })
    }));

    let value = ctx.get_async(&slot).await;
    assert_eq!(value, 42);
    assert_eq!(
        ctx.__window1_resolved_hits(),
        1,
        "get_async must return through the window-1 Resolved-after-re-lock arm"
    );
}

/// Window 2: a superseded compute drops its notifier without a final `Resolved`
/// send; the waiting `get_async` must re-resolve to the latest value instead of
/// panicking on the dropped `watch` sender.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn window2_superseded_notifier_drop_reresolves_to_latest() {
    let ctx = Arc::new(AsyncContext::new());
    let cell = ctx.cell(1i32);

    // Gate the first compute so it is still in flight when we supersede it.
    let (release_first, first_gate) = oneshot::channel::<()>();
    let first_gate = Arc::new(Mutex::new(Some(first_gate)));

    let slot = ctx.computed_async({
        let first_gate = first_gate.clone();
        move |ctx| {
            let observed = ctx.get_cell(&cell);
            let first_gate = first_gate.clone();
            async move {
                // Only the first (observed == 1) compute parks on the gate; the
                // superseding compute (observed == 2) returns immediately.
                if observed == 1 {
                    let gate = first_gate.lock().await.take();
                    if let Some(gate) = gate {
                        let _ = gate.await;
                    }
                }
                observed * 10
            }
        }
    });

    // Reader parks while the first compute is gated in flight.
    let reader = tokio::spawn({
        let ctx = ctx.clone();
        async move { ctx.get_async(&slot).await }
    });
    tokio::task::yield_now().await;

    // Supersede: a new revision drops the first compute's notifier without a
    // final `Resolved` send, forcing the reader through window 2.
    ctx.set_cell(&cell, 2);
    let _ = release_first.send(());

    assert_eq!(
        reader.await.unwrap(),
        20,
        "reader must re-resolve to the latest superseding value, not panic"
    );
    assert_eq!(ctx.get(&slot), Some(20));
}