liminal-server 0.24.9

//! The tap's own pins: counts, never clocks. Every expected number below is
//! worked out from the sampling rule (keep when `seen % N == 0`; N doubles
//! after a push leaves the ring more than half full; a take halves it).

use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use super::{
    PayloadGrant, PublishFact, Recorded, Tap, TapBatch, TapError, TapFilter, TapIdent, TapOutcome,
    TapSource,
};

fn filter(subject: &str) -> TapFilter {
    TapFilter {
        subject: subject.to_owned(),
    }
}

fn fact<'a>(subject: &'a str, payload: &'a [u8]) -> PublishFact<'a> {
    PublishFact {
        subject,
        payload,
        outcome: TapOutcome::Delivered,
        message_id: None,
        source: TapSource::Connection(7),
        reply_to: None,
        recorded: None,
    }
}

fn publish(tap: &Tap, times: usize) -> Result<(), TapError> {
    for _ in 0..times {
        tap.observe(&fact("orders", b"{}"))?;
    }
    Ok(())
}

fn shape(batches: &[TapBatch]) -> Vec<(u64, u32, u64, usize)> {
    batches
        .iter()
        .map(|batch| {
            (
                batch.seq,
                batch.sampled.get(),
                batch.dropped,
                batch.messages.len(),
            )
        })
        .collect()
}

/// With no subscription the hook returns before counting anything; the
/// observation count moves only for publishes a live tap was shown.
#[test]
fn with_no_tap_the_hook_counts_nothing() -> Result<(), TapError> {
    let tap = Tap::new("local", 8);
    publish(&tap, 1_000)?;
    assert_eq!(tap.observations(), 0);

    let subscription = tap.subscribe(filter("orders"))?;
    publish(&tap, 3)?;
    assert_eq!(tap.observations(), 3, "positive control: a live tap counts");
    drop(subscription);
    publish(&tap, 5)?;
    assert_eq!(tap.observations(), 3);
    assert_eq!(tap.subscriber_count(), 0);
    Ok(())
}

/// A slow subscriber holds at most its bound, and every message it lost is
/// counted once in the tap's lifetime total.
#[test]
fn a_slow_subscriber_never_holds_more_than_its_bound() -> Result<(), TapError> {
    let tap = Tap::new("local", 8);
    let slow = tap.subscribe(filter("orders"))?;
    for _ in 0..1_000 {
        tap.observe(&fact("orders", b"{}"))?;
        assert!(slow.held()? <= 8);
    }
    let batches = slow.take_batches()?;
    let kept: usize = batches.iter().map(|batch| batch.messages.len()).sum();
    let dropped: u64 = batches.iter().map(|batch| batch.dropped).sum();
    assert_eq!(kept, 8);
    assert_eq!(dropped, tap.dropped_total());
    assert!(dropped > 0, "the ring filled, so something was evicted");
    Ok(())
}

/// R12: each batch's `dropped` is the evictions since the previous batch, so
/// the console's running sum equals the tap's lifetime total.
///
/// Capacity 2. p1 and p2 are kept at N=1 (p2 leaves the ring full: N=2);
/// p3 is kept at N=2 and evicts p1 (N=4). The take gives [p2]@1 carrying the
/// one eviction, then [p3]@2, and halves N to 2. p4 is sampled out, p5 kept
/// at 2; take. N=1: p6 kept; take.
#[test]
fn dropped_is_a_delta() -> Result<(), TapError> {
    let tap = Tap::new("local", 2);
    let subscription = tap.subscribe(filter("orders"))?;
    publish(&tap, 3)?;
    let mut published = subscription.take_batches()?;
    publish(&tap, 2)?;
    published.extend(subscription.take_batches()?);
    publish(&tap, 1)?;
    published.extend(subscription.take_batches()?);

    assert_eq!(
        shape(&published),
        vec![(1, 1, 1, 1), (2, 2, 0, 1), (3, 2, 0, 1), (4, 1, 0, 1)]
    );
    let sum: u64 = published.iter().map(|batch| batch.dropped).sum();
    assert_eq!(sum, 1);
    assert_eq!(tap.dropped_total(), 1);

    // An idle take publishes nothing.
    assert!(subscription.take_batches()?.is_empty());
    Ok(())
}

/// R13: one divisor per batch. Capacity 8, eleven publishes:
/// p1–p5 kept at N=1 (p5 leaves 5 of 8: N=2); p6 sampled out, p7 kept at 2
/// (N=4); p8 out, p9 kept at 4 (N=8); p10, p11 out. Three batches, each true
/// of all its messages, and the weights sum to the eleven publishes:
/// 5×1 + 1×2 + 1×4.
#[test]
fn each_batch_has_one_divisor() -> Result<(), TapError> {
    let tap = Tap::new("local", 8);
    let subscription = tap.subscribe(filter("orders"))?;
    publish(&tap, 11)?;
    let batches = subscription.take_batches()?;

    assert_eq!(
        shape(&batches),
        vec![(1, 1, 0, 5), (2, 2, 0, 1), (3, 4, 0, 1)]
    );
    let weighted: u64 = batches
        .iter()
        .map(|batch| {
            u64::from(batch.sampled.get()) * u64::try_from(batch.messages.len()).unwrap_or(u64::MAX)
        })
        .sum();
    assert_eq!(weighted, 11);
    assert_eq!(subscription.sample_one_in()?.get(), 4, "a take halves N");
    Ok(())
}

/// R11: one wake per publish however many taps kept it, and none for a
/// publish no tap kept.
#[test]
fn one_wake_per_publish() -> Result<(), TapError> {
    let wakes = Arc::new(AtomicU64::new(0));
    let counter = Arc::clone(&wakes);
    let tap = Tap::with_waker(
        "local",
        64,
        Arc::new(move || {
            counter.fetch_add(1, Ordering::Relaxed);
        }),
    );
    let first = tap.subscribe(filter("orders"))?;
    let second = tap.subscribe(filter("ord*"))?;
    let third = tap.subscribe(filter(""))?;

    assert!(tap.observe(&fact("orders", b"{}"))?);
    assert_eq!(wakes.load(Ordering::Relaxed), 1);
    assert_eq!(
        (first.held()?, second.held()?, third.held()?),
        (1, 1, 1),
        "all three kept it"
    );

    let elsewhere = tap.observe(&fact("audit", b"{}"))?;
    assert!(elsewhere, "the catch-all tap keeps it");
    assert_eq!(wakes.load(Ordering::Relaxed), 2);
    drop(third);
    assert!(!tap.observe(&fact("audit", b"{}"))?);
    assert_eq!(wakes.load(Ordering::Relaxed), 2, "no tap kept it: no wake");
    Ok(())
}

/// R14: a kept message carries its causal parent as reply-to, and a message
/// without one carries none.
#[test]
fn reply_to_rides_the_kept_message() -> Result<(), TapError> {
    let tap = Tap::new("local", 64);
    let subscription = tap.subscribe(filter("orders"))?;
    tap.observe(&PublishFact {
        reply_to: Some("q-1"),
        message_id: Some(41),
        ..fact("orders", b"{}")
    })?;
    tap.observe(&PublishFact {
        outcome: TapOutcome::Refused,
        ..fact("orders", b"{}")
    })?;
    let batches = subscription.take_batches()?;
    let messages: Vec<_> = batches.iter().flat_map(|batch| &batch.messages).collect();
    assert_eq!(messages.len(), 2);
    assert_eq!(messages[0].reply_to.as_deref(), Some("q-1"));
    assert_eq!(messages[0].ident, TapIdent::Message(41));
    assert_eq!(messages[1].reply_to, None);
    assert_eq!(
        messages[1].ident,
        TapIdent::Observation(1),
        "a publish with no message id is named by its observation"
    );
    assert_eq!(messages[1].outcome, TapOutcome::Refused);
    Ok(())
}

/// Filters by exact subject and prefix, and payloads only under a grant.
#[test]
fn filters_and_grants() -> Result<(), TapError> {
    let tap = Tap::new("local", 64);
    let exact = tap.subscribe(filter("orders"))?;
    let prefix = tap.subscribe(filter("ord*"))?;
    let granted = tap.subscribe_with(filter("*"), Some(PayloadGrant::server_issued()))?;

    tap.observe(&fact("orders", b"abc"))?;
    tap.observe(&fact("other", b"z"))?;

    let exact_messages: Vec<_> = exact
        .take_batches()?
        .into_iter()
        .flat_map(|batch| batch.messages)
        .collect();
    assert_eq!(exact_messages.len(), 1);
    assert_eq!(exact_messages[0].subject, "orders");
    assert_eq!(exact_messages[0].size, 3);
    assert_eq!(exact_messages[0].payload, None, "no grant, no payload");
    assert!(
        exact_messages[0].hops.is_empty(),
        "nothing measured it, so it has no hop"
    );

    let prefix_count: usize = prefix
        .take_batches()?
        .iter()
        .map(|batch| batch.messages.len())
        .sum();
    assert_eq!(prefix_count, 1);

    let granted_messages: Vec<_> = granted
        .take_batches()?
        .into_iter()
        .flat_map(|batch| batch.messages)
        .collect();
    assert_eq!(granted_messages.len(), 2);
    assert_eq!(granted_messages[0].payload.as_deref(), Some(&b"abc"[..]));

    assert_eq!(tap.subscriber_count(), 3);
    drop(exact);
    drop(prefix);
    drop(granted);
    assert_eq!(tap.subscriber_count(), 0);
    assert_eq!(tap.unsubscribe_failures(), 0);
    Ok(())
}

/// SRV-012 `to_stations` (Buttercup's finding 1, Hermes 03:05): a fan-out
/// the publish recorded says whom it reached even when that is no registered
/// worker, as an empty list, and one it did not record says nothing, so a
/// reader can tell "reached no worker" from "not recorded".
#[test]
fn a_recorded_fan_out_that_reached_no_worker_is_an_empty_list() -> Result<(), TapError> {
    let tap = Tap::new("local", 64);
    let subscription = tap.subscribe(filter("orders"))?;
    tap.name_station(3, "orders.ledger")?;
    tap.name_receiver(30, 3)?;
    let at = std::time::Instant::now();
    for recipients in [
        Some(&[][..]),
        Some(&[(31, at)][..]),
        Some(&[(30, at), (31, at)][..]),
        None,
    ] {
        tap.observe(&PublishFact {
            recorded: recipients.map(|recipients| Recorded {
                recipients,
                received_at: at,
                appended_at: None,
            }),
            ..fact("orders", b"{}")
        })?;
    }
    let batches = subscription.take_batches()?;
    let stations: Vec<Option<Vec<String>>> = batches
        .iter()
        .flat_map(|batch| &batch.messages)
        .map(|message| {
            message.to_stations.as_deref().map(|list| {
                list.iter()
                    .map(|reached| reached.station.to_string())
                    .collect()
            })
        })
        .collect();
    assert_eq!(
        stations,
        vec![
            Some(vec![]),
            Some(vec![]),
            Some(vec!["orders.ledger".to_owned()]),
            None,
        ]
    );
    Ok(())
}

/// Each kept message's hops, as (node, `at_ms`).
fn hops(batches: &[TapBatch]) -> Vec<Vec<(String, u64)>> {
    batches
        .iter()
        .flat_map(|batch| &batch.messages)
        .map(|message| {
            message
                .hops
                .iter()
                .map(|hop| (hop.node.to_string(), hop.at_ms))
                .collect()
        })
        .collect()
}

/// Lane brief rulings 3, 6 and 8 (Hermes 7 Oct 03:51 and 03:54): a kept
/// message is first seen when the server received it, and its hops run from
/// its publishing station through the server to each receiving station, each
/// at its measured time after first sight. A durable message on a server
/// with a persistent store passes the depot as its append completes, and the
/// exchange after it; a message on a passing line, or on a server with no
/// persistent store, has no depot hop.
#[test]
fn a_kept_message_hops_from_its_publisher_through_the_server_to_its_receivers()
-> Result<(), TapError> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let ms = Duration::from_millis;
    let recipients = [(30, received + ms(5)), (40, received + ms(7))];
    let recorded = Recorded {
        recipients: &recipients,
        received_at: received,
        appended_at: Some(received + ms(3)),
    };
    let durable = PublishFact {
        recorded: Some(recorded),
        ..fact("orders", b"{}")
    };
    let passing = PublishFact {
        recorded: Some(Recorded {
            appended_at: None,
            ..recorded
        }),
        ..durable
    };
    let named = |tap: &Tap| -> Result<(), TapError> {
        tap.name_station(7, "orders.worker")?;
        tap.name_station(3, "orders.ledger")?;
        tap.name_receiver(30, 3)?;
        tap.name_station(4, "orders.audit")?;
        tap.name_receiver(40, 4)
    };
    let hop = |node: &str, at_ms: u64| (node.to_owned(), at_ms);

    let tap = Tap::new("local", 64).with_depot(true);
    let subscription = tap.subscribe(filter("orders"))?;
    named(&tap)?;
    tap.observe(&durable)?;
    tap.observe(&passing)?;
    let batches = subscription.take_batches()?;
    assert_eq!(
        hops(&batches),
        [
            vec![
                hop("orders.worker", 0),
                hop("local.depot", 3),
                hop("local", 3),
                hop("orders.ledger", 5),
                hop("orders.audit", 7),
            ],
            vec![
                hop("orders.worker", 0),
                hop("orders.ledger", 5),
                hop("orders.audit", 7),
            ],
        ]
    );
    let first_sight: Vec<Instant> = batches
        .iter()
        .flat_map(|batch| &batch.messages)
        .map(|message| message.seen_at)
        .collect();
    assert_eq!(first_sight, [received, received], "first sight is receipt");

    let memory = Tap::new("local", 64);
    let subscription = memory.subscribe(filter("orders"))?;
    named(&memory)?;
    memory.observe(&durable)?;
    assert_eq!(
        hops(&subscription.take_batches()?),
        [vec![
            hop("orders.worker", 0),
            hop("local", 3),
            hop("orders.ledger", 5),
            hop("orders.audit", 7),
        ]],
        "no persistent store, no depot hop"
    );
    Ok(())
}

/// Hermes 04:09 on lane brief ruling 3: a hop is never clamped. Handed an
/// append that completed after a receiver's inbox queued the message, which
/// the server's own order never gives, the tap keeps every hop as measured
/// and reports the observe as failed, naming the hop.
#[test]
fn a_hop_measured_out_of_order_is_kept_as_measured_and_counted() -> Result<(), TapError> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let ms = Duration::from_millis;
    let recipients = [(30, received + ms(5)), (40, received + ms(7))];
    let tap = Tap::new("local", 64).with_depot(true);
    let subscription = tap.subscribe(filter("orders"))?;
    tap.name_station(7, "orders.worker")?;
    tap.name_station(3, "orders.ledger")?;
    tap.name_receiver(30, 3)?;
    tap.name_station(4, "orders.audit")?;
    tap.name_receiver(40, 4)?;
    let observed = tap.observe(&PublishFact {
        recorded: Some(Recorded {
            recipients: &recipients,
            received_at: received,
            appended_at: Some(received + ms(6)),
        }),
        ..fact("orders", b"{}")
    });
    assert_eq!(
        observed,
        Err(TapError::HopsOutOfOrder {
            hop: 3,
            at_ms: 5,
            before_ms: 6,
        })
    );
    let hop = |node: &str, at_ms: u64| (node.to_owned(), at_ms);
    assert_eq!(
        hops(&subscription.take_batches()?),
        [vec![
            hop("orders.worker", 0),
            hop("local.depot", 6),
            hop("local", 6),
            hop("orders.ledger", 5),
            hop("orders.audit", 7),
        ]],
        "kept, each hop as measured"
    );
    Ok(())
}

/// A tap with its depot, the orders worker publishing, and the ledger and
/// audit stations each receiving.
fn named_tap() -> Result<Tap, TapError> {
    let tap = Tap::new("local", 64).with_depot(true);
    tap.name_station(7, "orders.worker")?;
    tap.name_station(3, "orders.ledger")?;
    tap.name_receiver(30, 3)?;
    tap.name_station(4, "orders.audit")?;
    tap.name_receiver(40, 4)?;
    Ok(tap)
}

/// Buttercup's V2, as Hermes ruled at 04:27: an append measured before the
/// message was first seen leaves its hops out of the record (`at_ms` cannot
/// carry it, and a 0 would read as a measurement), and the observe fails
/// naming the first such hop and how early it was.
#[test]
fn a_hop_before_first_sight_is_left_out_and_counted() -> Result<(), Box<dyn std::error::Error>> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let ms = Duration::from_millis;
    let appended = received.checked_sub(ms(2)).ok_or("no instant 2 ms back")?;
    let recipients = [(30, received + ms(5)), (40, received + ms(7))];
    let tap = named_tap()?;
    let subscription = tap.subscribe(filter("orders"))?;
    let observed = tap.observe(&PublishFact {
        recorded: Some(Recorded {
            recipients: &recipients,
            received_at: received,
            appended_at: Some(appended),
        }),
        ..fact("orders", b"{}")
    });
    assert_eq!(
        observed,
        Err(TapError::HopBeforeFirstSight {
            hop: 1,
            early_ms: 2,
        })
    );
    let hop = |node: &str, at_ms: u64| (node.to_owned(), at_ms);
    assert_eq!(
        hops(&subscription.take_batches()?),
        [vec![
            hop("orders.worker", 0),
            hop("orders.ledger", 5),
            hop("orders.audit", 7),
        ]],
        "no depot hop, and no exchange hop at the append's time"
    );
    Ok(())
}

/// Buttercup's V3, Hermes 04:27: one observe's failures are each reported,
/// and each counts. A hop before first sight and a hop out of order on the
/// same message are both named.
#[test]
fn every_failure_of_one_observe_is_reported() -> Result<(), Box<dyn std::error::Error>> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let ms = Duration::from_millis;
    let appended = received.checked_sub(ms(2)).ok_or("no instant 2 ms back")?;
    let recipients = [(30, received + ms(5)), (40, received + ms(3))];
    let tap = named_tap()?;
    let _subscription = tap.subscribe(filter("orders"))?;
    let observed = tap.observe(&PublishFact {
        recorded: Some(Recorded {
            recipients: &recipients,
            received_at: received,
            appended_at: Some(appended),
        }),
        ..fact("orders", b"{}")
    });
    let expected = TapError::Several(vec![
        TapError::HopBeforeFirstSight {
            hop: 1,
            early_ms: 2,
        },
        TapError::HopsOutOfOrder {
            hop: 4,
            at_ms: 3,
            before_ms: 5,
        },
    ]);
    assert_eq!(observed, Err(expected.clone()));
    assert_eq!(expected.failures(), 2);
    Ok(())
}

/// Jules's F1, as Hermes ruled at 07:22: nothing measured the exchange of a
/// message whose publish appended nothing, so it has no exchange hop, never
/// one written at 0. A recorded publish to a passing line keeps its
/// publisher at 0, first sight by definition, and its receivers as measured;
/// a publish that recorded nothing keeps only its publisher.
#[test]
fn an_exchange_no_clock_measured_has_no_hop() -> Result<(), Box<dyn std::error::Error>> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let ms = Duration::from_millis;
    let recipients = [(30, received + ms(5)), (40, received + ms(7))];
    let tap = named_tap()?;
    let subscription = tap.subscribe(filter("orders"))?;
    tap.observe(&PublishFact {
        recorded: Some(Recorded {
            recipients: &recipients,
            received_at: received,
            appended_at: None,
        }),
        ..fact("orders", b"{}")
    })?;
    tap.observe(&fact("orders", b"{}"))?;
    let hop = |node: &str, at_ms: u64| (node.to_owned(), at_ms);
    assert_eq!(
        hops(&subscription.take_batches()?),
        [
            vec![
                hop("orders.worker", 0),
                hop("orders.ledger", 5),
                hop("orders.audit", 7),
            ],
            vec![hop("orders.worker", 0)],
        ]
    );
    Ok(())
}

/// Hermes 07:22, on Jules's F1: an ephemeral publish takes no exchange
/// instant, which is normal, so its observe reports no failure and counts
/// nothing. `tap_observe_failures` counts only instants that were taken and
/// came out wrong.
#[test]
fn an_ephemeral_publish_counts_no_failure() -> Result<(), Box<dyn std::error::Error>> {
    use std::time::{Duration, Instant};

    let received = Instant::now();
    let recipients = [(30, received + Duration::from_millis(5))];
    let tap = named_tap()?;
    let _subscription = tap.subscribe(filter("orders"))?;
    let observed = tap.observe(&PublishFact {
        recorded: Some(Recorded {
            recipients: &recipients,
            received_at: received,
            appended_at: None,
        }),
        ..fact("orders", b"{}")
    });
    assert_eq!(observed, Ok(true), "kept, with no failure to count");
    Ok(())
}