zenkey-fleet 0.11.1

//! RFC 04 §3.2's seed discipline, proven against real zenoh (issue #42) —
//! self-contained two-peer sessions, no router.

use std::time::Duration;

use zenkey_fleet::{SeedItem, SeedPolicy, seed_subscribe};
use zenoh_ext::AdvancedPublisherBuilderExt;

mod util;
use util::timestamping_pair;

/// Drain a seeded subscriber until the boundary; returns (payloads, coverage).
async fn drain_seed(
    sub: &mut zenkey_fleet::SeededSubscriber,
) -> (Vec<String>, zenkey_fleet::SeedCoverage) {
    let mut values = Vec::new();
    loop {
        match tokio::time::timeout(util::SETTLE, sub.recv())
            .await
            .expect("seed boundary within 5s")
            .expect("stream alive")
        {
            SeedItem::Sample(v) => {
                values.push(String::from_utf8_lossy(&v.payload.to_bytes()).to_string())
            }
            SeedItem::Dropped(n) => {
                panic!("these fixtures never outrun the bounded channel ({n} dropped)")
            }
            SeedItem::SeedComplete(c) => return (values, c),
        }
    }
}

/// The history seed reaches a live publisher's cache — and the boundary
/// arrives strictly AFTER the seed (the race this module exists to close:
/// a boundary that outruns the cache reply turns "loading" into "empty").
/// Live samples keep flowing after the boundary.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn history_seed_lands_before_the_boundary() {
    let (a, b) = timestamping_pair().await;

    let publisher = a
        .declare_publisher("seedtest/state/health")
        .cache(zenoh_ext::CacheConfig::default().max_samples(1))
        .await
        .expect("advanced publisher");
    publisher.put("v1").await.expect("cached put");
    tokio::time::sleep(Duration::from_millis(300)).await;

    let mut sub = seed_subscribe(
        &b,
        "seedtest/state/**",
        SeedPolicy {
            timeout: Duration::from_millis(800),
            ..SeedPolicy::default()
        },
    )
    .await
    .expect("seed subscribe");

    let (seen, coverage) = drain_seed(&mut sub).await;
    assert_eq!(seen, ["v1"], "the cached value seeds — before the boundary");
    assert_eq!(coverage.history_replies, Some(1), "the cache answered once");
    assert_eq!(
        coverage.storage_replies,
        Some(0),
        "no storage on this bus — ran and found nothing, an observation"
    );

    // Live after the boundary.
    publisher.put("v2").await.expect("live put");
    let item = tokio::time::timeout(util::SETTLE, sub.recv())
        .await
        .expect("live within 5s")
        .expect("stream alive");
    match item {
        SeedItem::Sample(v) => assert_eq!(v.payload.to_bytes().as_ref(), b"v2"),
        other => panic!("expected the live sample, got {other:?}"),
    }
}

/// The LWW merge suppresses a stale storage seed: a storage-shaped queryable
/// whose (unstamped) reply arrives after the stamped cache value must not
/// regress the key — and the suppression is counted, never silent (O6).
/// The storage reply is delayed so the order is deterministic: stamped
/// state first, stale echo second.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_stale_storage_seed_cannot_regress_a_key() {
    let (a, b) = timestamping_pair().await;

    // The publisher's cache holds the CURRENT value, stamped now.
    let publisher = a
        .declare_publisher("staletest/state/doc")
        .cache(zenoh_ext::CacheConfig::default().max_samples(1))
        .await
        .expect("advanced publisher");
    publisher.put("current").await.expect("put");

    // A "storage" that answers late and UNstamped — by the time it replies,
    // the stamped cache value has already seeded the key; the merge must
    // refuse the regression (stamped state beats an unstamped echo).
    let _storage = a
        .declare_queryable("staletest/state/doc")
        .callback(move |query| {
            let q = query.clone();
            tokio::spawn(async move {
                tokio::time::sleep(Duration::from_millis(300)).await;
                q.reply("staletest/state/doc", "stale-from-storage")
                    .await
                    .ok();
            });
        })
        .await
        .expect("queryable");
    tokio::time::sleep(Duration::from_millis(300)).await;

    let mut sub = seed_subscribe(
        &b,
        "staletest/state/**",
        SeedPolicy {
            timeout: Duration::from_millis(800),
            ..SeedPolicy::default()
        },
    )
    .await
    .expect("seed subscribe");

    let (values, coverage) = drain_seed(&mut sub).await;
    assert_eq!(
        values,
        ["current"],
        "the stamped cache value seeds once; the late unstamped echo never surfaces"
    );
    assert_eq!(coverage.storage_replies, Some(1), "the storage DID answer");
    assert!(
        coverage.superseded >= 1,
        "…and its suppression is counted, not silent (O6)"
    );
}

/// The gap rule: a sample published immediately after `seed_subscribe`
/// returns — racing the seed GETs — arrives exactly once. This is the
/// transition GET-then-subscribe silently drops.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_transition_in_the_seed_window_lands_exactly_once() {
    let (a, b) = timestamping_pair().await;
    let publisher = a
        .declare_publisher("gaptest/state/flag")
        .await
        .expect("publisher");
    let matching = publisher
        .matching_listener()
        .await
        .expect("matching listener");
    // A storage sim that holds the query open (replying nothing) and says
    // when it arrived — the put below waits for both signals, so the seed
    // window is *provably* open when the transition is published, even on a
    // loaded machine.
    let (got_query_tx, mut got_query) = tokio::sync::mpsc::unbounded_channel::<()>();
    let _slow_storage = a
        .declare_queryable("gaptest/state/**")
        .callback(move |query| {
            let _ = got_query_tx.send(());
            tokio::spawn(async move {
                tokio::time::sleep(Duration::from_millis(2000)).await;
                drop(query);
            });
        })
        .await
        .expect("queryable");
    // A ready-probe declared AFTER the storage sim: same-session declarations
    // propagate in order, so once b can query this, b can query the storage —
    // the seed GET below provably reaches it, loaded machine or not.
    let _ready = a
        .declare_queryable("gaptest/ready")
        .callback(|query| {
            let q = query.clone();
            tokio::spawn(async move {
                q.reply("gaptest/ready", "ok").await.ok();
            });
        })
        .await
        .expect("ready queryable");
    let probe_deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        let replies = b
            .get("gaptest/ready")
            .timeout(Duration::from_millis(300))
            .await
            .expect("probe get");
        if replies.recv_async().await.is_ok() {
            break;
        }
        assert!(
            tokio::time::Instant::now() < probe_deadline,
            "routing never converged"
        );
    }

    let mut sub = seed_subscribe(
        &b,
        "gaptest/state/**",
        SeedPolicy {
            history: false, // plain publisher; the subscriber-first order is the point
            timeout: Duration::from_millis(3000),
            ..SeedPolicy::default()
        },
    )
    .await
    .expect("seed subscribe");

    // Wait until (1) the subscriber's interest reached the publishing peer
    // (a network-propagation concern, not part of the seed contract) and
    // (2) the storage GET is in flight — then publish inside the window.
    let ev = tokio::time::timeout(util::SETTLE, matching.recv_async())
        .await
        .expect("matching event within 5s")
        .expect("listener alive");
    assert!(ev.matching(), "the seed subscriber is a real subscriber");
    tokio::time::timeout(util::SETTLE, got_query.recv())
        .await
        .expect("the storage GET reaches the queryable within 5s")
        .expect("channel alive");
    publisher.put("flank").await.expect("put");

    let mut before_boundary = 0;
    let mut after_boundary = 0;
    let mut done = false;
    let mut deadline = tokio::time::Instant::now() + Duration::from_secs(8);
    loop {
        match tokio::time::timeout_at(deadline, sub.recv()).await {
            Ok(Some(SeedItem::Sample(v))) => {
                assert_eq!(v.payload.to_bytes().as_ref(), b"flank");
                if done {
                    after_boundary += 1;
                } else {
                    before_boundary += 1;
                }
            }
            Ok(Some(SeedItem::Dropped(n))) => {
                panic!("this fixture never outruns the bounded channel ({n} dropped)")
            }
            Ok(Some(SeedItem::SeedComplete(c))) => {
                assert_eq!(c.history_replies, None, "history was opted out");
                done = true;
                // Short drain: anything late (a duplicate, a post-boundary
                // copy) has this long to show itself.
                deadline = tokio::time::Instant::now() + Duration::from_millis(700);
            }
            Ok(None) | Err(_) => break,
        }
    }
    assert!(done, "the seed boundary must arrive");
    assert_eq!(
        (before_boundary, after_boundary),
        (1, 0),
        "the in-window transition lands exactly once, inside the seed phase"
    );
}

/// Issue #92's acceptance: a seeded watch on `…/state/**` against a
/// publisher whose cached value predates the watch shows that value without
/// waiting for a refresh — through the monitor's own bounded broadcast, with
/// the boundary as a typed event carrying this watch's id and coverage.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn a_seeded_watch_shows_pre_existing_state() {
    let (a, b) = timestamping_pair().await;
    let publisher = a
        .declare_publisher("wseed/state/health")
        .cache(zenoh_ext::CacheConfig::default().max_samples(1))
        .await
        .expect("advanced publisher");
    publisher.put("cached-before-watch").await.expect("put");
    tokio::time::sleep(Duration::from_millis(300)).await;

    let monitor = zenkey_fleet::Monitor::start(&b, zenkey_fleet::MonitorSpec::default())
        .await
        .expect("monitor");
    let mut events = monitor.events();
    let id = monitor
        .watch_seeded(
            "wseed/state/**",
            SeedPolicy {
                timeout: Duration::from_millis(800),
                ..SeedPolicy::default()
            },
        )
        .await
        .expect("seeded watch");

    // Drain until the boundary; the cached value must arrive before it.
    let mut seen = Vec::new();
    let coverage = loop {
        let item = tokio::time::timeout(util::SETTLE, events.recv())
            .await
            .expect("boundary within 5s")
            .expect("stream alive");
        match item {
            zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::Sample(s)) => {
                seen.push(String::from_utf8_lossy(&s.payload.to_bytes()).to_string());
            }
            zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::WatchSeeded {
                id: seeded,
                coverage,
            }) => {
                assert_eq!(seeded, id, "the boundary names the watch it closes");
                break coverage;
            }
            _ => {}
        }
    };
    assert_eq!(
        seen,
        ["cached-before-watch"],
        "pre-existing state arrives without waiting for a refresh"
    );
    assert_eq!(coverage.history_replies, Some(1));
    assert_eq!(coverage.storage_replies, Some(0));

    // …and the seeded key is already in the tree at the boundary tick.
    assert_eq!(monitor.tree().keys, 1);

    // Live samples keep flowing after the boundary (the merge is gone).
    publisher.put("live-after").await.expect("live put");
    loop {
        let item = tokio::time::timeout(util::SETTLE, events.recv())
            .await
            .expect("live within 5s")
            .expect("stream alive");
        if let zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::Sample(s)) = item {
            assert_eq!(s.payload.to_bytes().as_ref(), b"live-after");
            break;
        }
    }
}

/// #342: dropping a monitor aborts its seed tasks too.
///
/// `Drop` aborted only `self.tasks`; a seeded watch's task lives in `watches`
/// and was merely dropped, which detaches. It holds a cloned `Session` and
/// goes on ingesting until its own seed timeout — one per dropped monitor, in
/// a GUI that rebuilds its monitor on every re-scope.
///
/// The boundary event is the tell: an aborted seed task never reaches the
/// `WatchSeeded` send, a detached one does. The `EventStream` outlives the
/// monitor (it holds the core), so it is still listening either way.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn dropping_a_monitor_aborts_its_seed_tasks() {
    let (a, b) = timestamping_pair().await;

    // **The seed has to be genuinely pending**, and nothing else here makes
    // it so. `seed_get` loops until the reply channel closes, and with no
    // queryable at all zenoh closes it as soon as the query resolves — so
    // the boundary fired in microseconds and `SEED_TIMEOUT` never entered
    // into it. The test then passed only when the machine was loaded enough
    // to hold the query past the premise window, which is not a test.
    //
    // A queryable that takes the query and never answers holds it open until
    // the *querier's* timeout, so the seed reliably runs the full
    // `SEED_TIMEOUT` and the abort below is what decides the outcome.
    let _blocker = a
        .declare_queryable("wdrop/state/**")
        .callback(move |query| {
            tokio::spawn(async move {
                tokio::time::sleep(Duration::from_secs(30)).await;
                drop(query);
            });
        })
        .await
        .expect("blocking queryable");
    // A ready-probe declared AFTER the blocker, the same device
    // `a_transition_in_the_seed_window_lands_exactly_once` uses: same-session
    // declarations propagate in order, so once `b` can query this, `b` can
    // reach the blocker. Without it the seed GET can go out before the
    // blocker is visible, resolve against nobody, and void the premise —
    // which is the second way this test was environmental.
    let _ready = a
        .declare_queryable("wdrop/ready")
        .callback(|query| {
            let q = query.clone();
            tokio::spawn(async move {
                q.reply("wdrop/ready", "ok").await.ok();
            });
        })
        .await
        .expect("ready queryable");
    let probe_deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    loop {
        let replies = b
            .get("wdrop/ready")
            .timeout(Duration::from_millis(300))
            .await
            .expect("probe get");
        if replies.recv_async().await.is_ok() {
            break;
        }
        assert!(
            tokio::time::Instant::now() < probe_deadline,
            "routing never converged"
        );
    }

    let monitor = zenkey_fleet::Monitor::start(&b, zenkey_fleet::MonitorSpec::default())
        .await
        .expect("monitor");
    let mut events = monitor.events();

    // What the *querier* waits before giving up on the blocker above — so it
    // is now what the seed's duration actually is, rather than an upper bound
    // nothing approached.
    const SEED_TIMEOUT: Duration = Duration::from_secs(2);
    monitor
        .watch_seeded(
            "wdrop/state/**",
            SeedPolicy {
                timeout: SEED_TIMEOUT,
                ..SeedPolicy::default()
            },
        )
        .await
        .expect("seeded watch");

    // The premise, asserted rather than assumed: the boundary has not fired
    // yet, so the abort below is what decides the outcome. If this ever trips,
    // the machine was slow enough to void the test — which is a legible
    // failure, unlike the silent one it replaces.
    //
    // **Everything queued, not one item.** The stream also carries
    // `StatsTick` (every 250 ms) and `WatchChanged`, so reading a single
    // item could consume one of those, leave a `WatchSeeded` queued behind
    // it, and pass a premise that is false — after which the loop below
    // pulls that boundary and reports it as having "outlived the monitor"
    // when it fired *before* the drop. That is this check's own failure
    // wearing the other check's message, and it is what made this test flake
    // under a loaded `--workspace` run while passing 12 times in isolation.
    while let Ok(Some(item)) = tokio::time::timeout(Duration::from_millis(50), events.recv()).await
    {
        assert!(
            !matches!(
                item,
                zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::WatchSeeded { .. })
            ),
            "the seed announced before the monitor was dropped — the premise is void, \
             not the code (raise SEED_TIMEOUT)"
        );
    }

    // The blocker holds the seed GET, so the task would otherwise run its
    // timeout out and then announce the boundary.
    drop(monitor);

    // Listen past the timeout by a clear margin: a *detached* task announces
    // at ~SEED_TIMEOUT, so a window shorter than that would pass either way
    // and prove nothing.
    let listen = tokio::time::Instant::now() + SEED_TIMEOUT * 2;
    while let Ok(Some(item)) = tokio::time::timeout_at(listen, events.recv()).await {
        assert!(
            !matches!(
                item,
                zenkey_fleet::StreamItem::Event(zenkey_fleet::FleetEvent::WatchSeeded { .. })
            ),
            "the seed task outlived the monitor that owned it"
        );
    }
}