nordisk-pairing 0.5.27

//! **The feed bus** (0.5.15, A7 row 8): a value the installer feeds AFTER a box started
//! takes effect at once, without a restart.
//!
//! Until now each product read its feed where it needed it: Holger read the monetize key,
//! the pay page, the passkey relying party, the estate name, the off-site recipient and the
//! repair bearer ONCE at its start, and its fleet door every 5 s (D13). Gunnar had its own
//! bus (`gunnar-server` `feed::on_published`), in no shared crate. A value the front offers
//! after the boxes serve (the monetize key, the fleet door) was then never applied, or
//! applied only after a restart.
//!
//! The bus is that one place, shared:
//!
//! * [`Bus::publish`] — a new or CHANGED value is handed to every subscriber; the same
//!   value again is not (`false`).
//! * [`Bus::on_published`] — `take` gets the value in force now, if any, and every change
//!   from then on.
//! * [`Bus::on_change`] — the same, for one VIEW of the value (`view(&T) -> V`): `take` runs
//!   only when the view changes (the fleet door, the monetize key), never for a change
//!   elsewhere in the feed.
//! * [`Bus::current`] — the value in force.
//!
//! [`crate::boxside::Installed`] publishes every kept [`JoinReply`] on its own bus
//! ([`crate::boxside::Installed::feed_bus`]): the kept `joined.toml` at open, each feed and
//! re-feed after. [`global`] is a process-wide bus for a product that runs one box per
//! process and wants a free-function shape (Gunnar's `feed::{publish, on_published,
//! current}`); a product with several boxes in one process (tests) uses each box's own.
//!
//! **Ordering.** Deliveries are serialised: every subscriber sees the values in the order
//! they were published, and a subscriber added while a value is delivered sees that value
//! exactly once. A subscriber may subscribe further or publish from inside its own call
//! (that publish is delivered after the current one, on the same thread). A subscriber
//! that panics is said and dropped from that delivery; the others still run.

use std::any::Any;
use std::cell::RefCell;
use std::collections::{BTreeMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock, RwLock};

use crate::join::JoinReply;

/// A feed as the installer carries it: key → value.
pub type Feed = BTreeMap<String, String>;

type Take<T> = Arc<dyn Fn(&T) + Send + Sync>;

/// **A bus of `T`** (a [`JoinReply`] on a box, a product's typed view on its own bus).
pub struct Bus<T: Clone + PartialEq + Send + Sync + 'static> {
    /// The value in force and its generation.
    now: RwLock<Option<(u64, Arc<T>)>>,
    takers: Mutex<Vec<Taker<T>>>,
    /// Held while values are delivered: deliveries never interleave.
    delivering: Mutex<()>,
    /// Values published and not yet delivered (by the thread that delivers on this bus).
    queued: Mutex<VecDeque<(u64, Arc<T>)>>,
}

struct Taker<T> {
    name: &'static str,
    take: Take<T>,
    /// The generation this subscriber saw last: no value reaches it twice.
    seen: Arc<AtomicU64>,
}

impl<T> Clone for Taker<T> {
    fn clone(&self) -> Self {
        Taker { name: self.name, take: Arc::clone(&self.take), seen: Arc::clone(&self.seen) }
    }
}

thread_local! {
    /// The buses this thread is delivering on now (by address): re-entry never takes the
    /// delivery lock twice.
    static DELIVERING: RefCell<Vec<usize>> = const { RefCell::new(Vec::new()) };
}

impl<T: Clone + PartialEq + Send + Sync + 'static> Default for Bus<T> {
    fn default() -> Self {
        Bus { now: RwLock::new(None), takers: Mutex::new(Vec::new()), delivering: Mutex::new(()), queued: Mutex::new(VecDeque::new()) }
    }
}

impl<T: Clone + PartialEq + Send + Sync + 'static> Bus<T> {
    pub fn new() -> Arc<Bus<T>> {
        Arc::new(Bus::default())
    }

    fn id(&self) -> usize {
        self as *const Bus<T> as usize
    }

    /// Run `f` as (part of) a delivery: the outermost call on this thread holds the
    /// delivery lock and, after `f`, delivers every value queued meanwhile, in order.
    fn delivery<R>(&self, f: impl FnOnce() -> R) -> R {
        let id = self.id();
        if DELIVERING.with(|d| d.borrow().contains(&id)) {
            return f();
        }
        let _one = self.delivering.lock().unwrap_or_else(|p| p.into_inner());
        DELIVERING.with(|d| d.borrow_mut().push(id));
        let r = f();
        loop {
            let next = self.queued.lock().unwrap_or_else(|p| p.into_inner()).pop_front();
            let Some((generation, v)) = next else { break };
            let takers: Vec<Taker<T>> = self.takers.lock().unwrap_or_else(|p| p.into_inner()).clone();
            for t in &takers {
                give(t, generation, &v);
            }
        }
        DELIVERING.with(|d| {
            let mut d = d.borrow_mut();
            if let Some(i) = d.iter().rposition(|x| *x == id) {
                d.remove(i);
            }
        });
        r
    }

    /// **Publish `value`**: `true` when it is new or changed (every subscriber gets it),
    /// `false` when it equals the value in force (nobody is called).
    pub fn publish(&self, value: T) -> bool {
        self.delivery(|| {
            let mut now = self.now.write().unwrap_or_else(|p| p.into_inner());
            if now.as_ref().map(|(_, v)| v.as_ref()) == Some(&value) {
                return false;
            }
            let generation = now.as_ref().map(|(g, _)| *g).unwrap_or(0) + 1;
            let value = Arc::new(value);
            *now = Some((generation, Arc::clone(&value)));
            drop(now);
            self.queued.lock().unwrap_or_else(|p| p.into_inner()).push_back((generation, value));
            true
        })
    }

    /// **Subscribe**: `take` gets the value in force now (if any) and every change after.
    /// `name` says who it is (in [`Bus::subscribers`] and when it panics).
    pub fn on_published(&self, name: &'static str, take: impl Fn(&T) + Send + Sync + 'static) {
        let t = Taker { name, take: Arc::new(take), seen: Arc::new(AtomicU64::new(0)) };
        self.delivery(|| {
            self.takers.lock().unwrap_or_else(|p| p.into_inner()).push(t.clone());
            let now = self.now.read().unwrap_or_else(|p| p.into_inner()).clone();
            if let Some((generation, v)) = now {
                give(&t, generation, &v);
            }
        })
    }

    /// **Subscribe to one view of the value**: `take` gets the view in force now (if any)
    /// and then only when the view CHANGES. A view of `Option<_>` that turns `None` is a
    /// change too: the subscriber decides what an absent value means.
    pub fn on_change<V>(&self, name: &'static str, view: impl Fn(&T) -> V + Send + Sync + 'static, take: impl Fn(&V) + Send + Sync + 'static)
    where
        V: PartialEq + Send + 'static,
    {
        let last: Mutex<Option<V>> = Mutex::new(None);
        self.on_published(name, move |t| {
            let v = view(t);
            let mut seen = last.lock().unwrap_or_else(|p| p.into_inner());
            if seen.as_ref() == Some(&v) {
                return;
            }
            take(&v);
            *seen = Some(v);
        });
    }

    /// The value in force, if one was published.
    pub fn current(&self) -> Option<Arc<T>> {
        self.now.read().unwrap_or_else(|p| p.into_inner()).as_ref().map(|(_, v)| Arc::clone(v))
    }

    /// How many changes were published (0: none yet).
    pub fn generation(&self) -> u64 {
        self.now.read().unwrap_or_else(|p| p.into_inner()).as_ref().map(|(g, _)| *g).unwrap_or(0)
    }

    /// Who is subscribed, in order.
    pub fn subscribers(&self) -> Vec<&'static str> {
        self.takers.lock().unwrap_or_else(|p| p.into_inner()).iter().map(|t| t.name).collect()
    }
}

/// Hand `v` (generation `generation`) to `t`, unless it saw that generation already.
fn give<T>(t: &Taker<T>, generation: u64, v: &T) {
    if t.seen.fetch_max(generation, Ordering::SeqCst) >= generation {
        return;
    }
    call(t.name, &t.take, v);
}

fn call<T>(name: &str, take: &Take<T>, v: &T) {
    if let Err(p) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| take(v))) {
        let why = panic_words(&*p);
        eprintln!("feed bus: the subscriber `{name}` panicked on a published feed ({why}); the others still run");
    }
}

fn panic_words(p: &(dyn Any + Send)) -> String {
    p.downcast_ref::<&str>().map(|s| s.to_string()).or_else(|| p.downcast_ref::<String>().cloned()).unwrap_or_else(|| "no message".into())
}

/// The bus a box publishes its kept join reply on.
pub type ReplyBus = Bus<JoinReply>;

/// **The process-wide bus of the installer's feed** (one box per process).
pub fn global() -> &'static Bus<Feed> {
    static GLOBAL: OnceLock<Bus<Feed>> = OnceLock::new();
    GLOBAL.get_or_init(Bus::default)
}

/// [`Bus::publish`] on [`global`].
pub fn publish(feed: Feed) -> bool {
    global().publish(feed)
}

/// [`Bus::on_published`] on [`global`].
pub fn on_published(name: &'static str, take: impl Fn(&Feed) + Send + Sync + 'static) {
    global().on_published(name, take)
}

/// [`Bus::current`] on [`global`].
pub fn current() -> Option<Arc<Feed>> {
    global().current()
}

/// Pure: the value of `key` in `feed`, trimmed; `None` when absent or blank.
pub fn value(feed: &Feed, key: &str) -> Option<String> {
    feed.get(key).map(|v| v.trim().to_string()).filter(|v| !v.is_empty())
}

/// Pure: the value of the first `offer.<vm>.<key>` in `feed` (what the front offers every
/// box: `offer.front.monetize_pubkey`), in VM name order, trimmed; `None` when no VM offers
/// one. Blank is absent. `<vm>` is one non-empty name without a dot: `offer..<key>` and
/// `offer.a.b.<key>` offer nothing (exactly Gunnar's `feed::offered`, 76e16478).
pub fn offered(feed: &Feed, key: &str) -> Option<String> {
    feed.iter()
        .filter_map(|(k, v)| k.strip_prefix("offer.")?.strip_suffix(key)?.strip_suffix('.').filter(|vm| !vm.is_empty() && !vm.contains('.')).map(|_| v.trim()))
        .find(|v| !v.is_empty())
        .map(str::to_string)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::AtomicUsize;

    fn feed(pairs: &[(&str, &str)]) -> Feed {
        pairs.iter().map(|(k, v)| (k.to_string(), v.to_string())).collect()
    }

    /// ★ Row 8: a subscriber gets the feed in force when it subscribes and every CHANGE
    /// after; the same feed published again reaches nobody.
    #[test]
    fn a_subscriber_gets_the_value_in_force_and_every_change() {
        let bus: Arc<Bus<Feed>> = Bus::new();
        assert!(bus.publish(feed(&[("a", "1")])));
        let seen = Arc::new(Mutex::new(Vec::new()));
        let s = Arc::clone(&seen);
        bus.on_published("t", move |f| s.lock().unwrap().push(f.get("a").cloned().unwrap_or_default()));
        assert!(!bus.publish(feed(&[("a", "1")])), "the same value again is no change");
        assert!(bus.publish(feed(&[("a", "2")])));
        assert!(bus.publish(feed(&[("a", "3")])));
        assert_eq!(*seen.lock().unwrap(), vec!["1", "2", "3"]);
        assert_eq!(bus.generation(), 3);
        assert_eq!(bus.subscribers(), vec!["t"]);
        assert_eq!(bus.current().unwrap().get("a").map(String::as_str), Some("3"));
    }

    /// ★ A view's subscriber runs only when its view changes: a late `fleet` reaches the
    /// fleet subscriber once, a change of another key does not wake it.
    #[test]
    fn a_view_subscriber_runs_only_when_its_view_changes() {
        let bus: Arc<Bus<Feed>> = Bus::new();
        let seen = Arc::new(Mutex::new(Vec::new()));
        let s = Arc::clone(&seen);
        bus.on_change("fleet", |f: &Feed| value(f, "fleet"), move |v: &Option<String>| s.lock().unwrap().push(v.clone()));
        bus.publish(feed(&[("token", "t")]));
        bus.publish(feed(&[("token", "t"), ("clock", "x")]));
        bus.publish(feed(&[("token", "t"), ("clock", "x"), ("fleet", "https://one")]));
        bus.publish(feed(&[("token", "u"), ("clock", "x"), ("fleet", "https://one")]));
        bus.publish(feed(&[("fleet", "https://two")]));
        assert_eq!(*seen.lock().unwrap(), vec![None, Some("https://one".to_string()), Some("https://two".to_string())]);
    }

    /// A subscriber that subscribes or publishes from inside its own call neither
    /// deadlocks nor loses the value: the inner publish is delivered after the outer one.
    #[test]
    fn a_subscriber_may_subscribe_and_publish_from_inside_its_call() {
        let bus: Arc<Bus<Feed>> = Bus::new();
        let seen = Arc::new(Mutex::new(Vec::new()));
        let (b, s) = (Arc::clone(&bus), Arc::clone(&seen));
        bus.on_published("outer", move |f| {
            s.lock().unwrap().push(format!("outer {}", f.get("n").unwrap()));
            if f.get("n").map(String::as_str) == Some("1") {
                let s2 = Arc::clone(&s);
                b.on_published("inner", move |f| s2.lock().unwrap().push(format!("inner {}", f.get("n").unwrap())));
                b.publish(feed(&[("n", "2")]));
            }
        });
        bus.publish(feed(&[("n", "1")]));
        assert_eq!(*seen.lock().unwrap(), vec!["outer 1", "inner 1", "outer 2", "inner 2"]);
    }

    /// A subscriber that panics does not take the others down.
    #[test]
    fn a_panicking_subscriber_is_dropped_from_that_delivery_only() {
        let bus: Arc<Bus<Feed>> = Bus::new();
        let n = Arc::new(AtomicUsize::new(0));
        bus.on_published("panics", |_| panic!("boom"));
        let m = Arc::clone(&n);
        bus.on_published("counts", move |_| {
            m.fetch_add(1, Ordering::SeqCst);
        });
        bus.publish(feed(&[("a", "1")]));
        bus.publish(feed(&[("a", "2")]));
        assert_eq!(n.load(Ordering::SeqCst), 2);
    }

    /// Two threads publishing: every subscriber sees whole values in one order.
    #[test]
    fn concurrent_publishes_are_delivered_one_at_a_time() {
        let bus: Arc<Bus<Feed>> = Bus::new();
        let busy = Arc::new(AtomicUsize::new(0));
        let overlap = Arc::new(AtomicUsize::new(0));
        let (b2, o2) = (Arc::clone(&busy), Arc::clone(&overlap));
        bus.on_published("slow", move |_| {
            if b2.fetch_add(1, Ordering::SeqCst) > 0 {
                o2.fetch_add(1, Ordering::SeqCst);
            }
            std::thread::sleep(std::time::Duration::from_millis(2));
            b2.fetch_sub(1, Ordering::SeqCst);
        });
        let threads: Vec<_> = (0..4)
            .map(|t| {
                let bus = Arc::clone(&bus);
                std::thread::spawn(move || {
                    for i in 0..10 {
                        bus.publish(feed(&[("v", &format!("{t}-{i}"))]));
                    }
                })
            })
            .collect();
        for t in threads {
            t.join().unwrap();
        }
        assert_eq!(overlap.load(Ordering::SeqCst), 0, "two deliveries overlapped");
    }

    #[test]
    fn the_front_offer_is_read_from_any_vm() {
        let f = feed(&[("offer.front.monetize_pubkey", " ab "), ("fleet", " ")]);
        assert_eq!(offered(&f, "monetize_pubkey").as_deref(), Some("ab"));
        assert_eq!(offered(&f, "fleet"), None);
        assert_eq!(value(&f, "fleet"), None);
    }

    /// Ported from Gunnar (gunnar-server `feed::offered`, 76e16478): the front's keys reach
    /// the boxes as a VM's offer; a key that only ends like the wanted one is another key.
    #[test]
    fn a_value_offered_by_a_vm_is_read_and_a_longer_key_is_another_key() {
        let f = feed(&[("offer.front.monetize_pubkey", "A"), ("offer.front.clock_key", "ssh-ed25519 AAAAk"), ("offer.front.xmonetize_pubkey", "B")]);
        assert_eq!(offered(&f, "monetize_pubkey").as_deref(), Some("A"));
        assert_eq!(offered(&f, "clock_key").as_deref(), Some("ssh-ed25519 AAAAk"));
        assert_eq!(offered(&feed(&[("offer.front.xmonetize_pubkey", "B")]), "monetize_pubkey"), None);
    }

    /// Ported from Gunnar: `offer..k` and `offer.a.b.k` name no VM, so they offer nothing.
    #[test]
    fn an_offer_from_no_vm_name_is_not_taken() {
        assert_eq!(offered(&feed(&[("offer..monetize_pubkey", "A"), ("offer.a.b.monetize_pubkey", "A")]), "monetize_pubkey"), None, "not a VM name");
        assert_eq!(offered(&feed(&[("offer..monetize_pubkey", "A")]), "monetize_pubkey"), None);
        assert_eq!(offered(&feed(&[("offer.a.b.monetize_pubkey", "A")]), "monetize_pubkey"), None);
        assert_eq!(offered(&feed(&[("offer.monetize_pubkey", "A")]), "monetize_pubkey"), None, "no VM at all");
        assert_eq!(offered(&feed(&[("offer.a.b.monetize_pubkey", "A"), ("offer.front.monetize_pubkey", "C")]), "monetize_pubkey").as_deref(), Some("C"), "a VM's own offer still counts");
    }

    /// Ported from Gunnar: with several VMs offering, the first by VM name wins, and a blank
    /// offer is absent (the next VM's counts).
    #[test]
    fn the_first_vm_by_name_offers_and_blank_is_absent() {
        let f = feed(&[("offer.vm2.ticket_pubkey", "A"), ("offer.vm3.ticket_pubkey", "B")]);
        assert_eq!(offered(&f, "ticket_pubkey").as_deref(), Some("A"), "the first by name");
        let f = feed(&[("offer.vm1.ticket_pubkey", " "), ("offer.vm3.ticket_pubkey", " B ")]);
        assert_eq!(offered(&f, "ticket_pubkey").as_deref(), Some("B"), "blank is absent");
    }
}