haematite-cli 0.21.8

The haematite command: serve a store, its door and Studio, answer MCP, query with jq, and host the store's assistant seat
Documentation
use super::{Door, Issuer, SignedIn, Started};
use lys_pass::Error;
use std::sync::{Mutex, mpsc};
use std::time::Duration;
type TestResult = Result<(), Box<dyn std::error::Error>>;
struct Slow {
    entered: mpsc::Sender<()>,
    release: Mutex<mpsc::Receiver<()>>,
}
impl Issuer for Slow {
    fn start(&self) -> Result<Started, Error> {
        Ok(Started {
            url: "https://lys.example".into(),
            state: "state".into(),
            browser: "browser".into(),
            verifier: "verifier".into(),
        })
    }
    fn finish(
        &self,
        code: &str,
        _state: &str,
        _browser: &str,
        _started: &Started,
    ) -> Result<SignedIn, Error> {
        Ok(SignedIn {
            token: code.into(),
            holder: code.into(),
            expires_at: if code == "slow" { 110 } else { 900 },
            refresh: Some(code.into()),
        })
    }
    fn refresh(&self, _refresh: &str) -> Result<SignedIn, Error> {
        self.entered
            .send(())
            .map_err(|_| Error::CannotAsk("entered"))?;
        self.release
            .lock()
            .map_err(|_| Error::CannotAsk("poisoned"))?
            .recv()
            .map_err(|_| Error::CannotAsk("release"))?;
        Ok(SignedIn {
            token: "fresh".into(),
            holder: "slow".into(),
            expires_at: 900,
            refresh: Some("slow".into()),
        })
    }
}
#[test]
fn slow_refresh_does_not_block_another_session() -> TestResult {
    let (entered, entering) = mpsc::channel();
    let (release, released) = mpsc::channel();
    let door = Door::new(Slow {
        entered,
        release: Mutex::new(released),
    });
    door.begin(100)?;
    let slow = door.callback("slow", "state", "browser", 100)?;
    door.begin(100)?;
    let fast = door.callback("fast", "state", "browser", 100)?;
    std::thread::scope(|scope| -> TestResult {
        let slow = scope.spawn(|| door.pass(&slow, 100));
        entering.recv_timeout(Duration::from_secs(5))?;
        let (answer, answered) = mpsc::channel();
        let door = &door;
        let fast = scope.spawn(move || answer.send(door.pass(&fast, 100)));
        let independent = answered.recv_timeout(Duration::from_secs(2));
        release.send(())?;
        slow.join().map_err(|_| "refresh panicked")??;
        fast.join().map_err(|_| "request panicked")??;
        assert_eq!(
            independent.map_err(|_| "another session waited for slow refresh")??,
            "fast"
        );
        Ok(())
    })
}

#[test]
fn concurrent_requests_refresh_one_session_once() -> TestResult {
    let (entered, entering) = mpsc::channel();
    let (release, released) = mpsc::channel();
    let door = Door::new(Slow {
        entered,
        release: Mutex::new(released),
    });
    door.begin(100)?;
    let id = door.callback("slow", "state", "browser", 100)?;
    std::thread::scope(|scope| -> TestResult {
        let first = scope.spawn(|| door.pass(&id, 100));
        entering.recv_timeout(Duration::from_secs(5))?;
        let second = scope.spawn(|| door.pass(&id, 100));
        release.send(())?;
        assert_eq!(
            first.join().map_err(|_| "first refresh panicked")??,
            "fresh"
        );
        assert_eq!(
            second.join().map_err(|_| "second refresh panicked")??,
            "fresh"
        );
        assert!(
            entering.try_recv().is_err(),
            "one session refreshed more than once"
        );
        Ok(())
    })
}

struct FailedRefresh(std::sync::atomic::AtomicUsize);
impl Issuer for FailedRefresh {
    fn start(&self) -> Result<Started, Error> {
        Ok(Started {
            url: "https://lys.example".into(),
            state: "state".into(),
            browser: "browser".into(),
            verifier: "verifier".into(),
        })
    }
    fn finish(
        &self,
        _code: &str,
        _state: &str,
        _browser: &str,
        _started: &Started,
    ) -> Result<SignedIn, Error> {
        Ok(SignedIn {
            token: "old".into(),
            holder: "fixture".into(),
            expires_at: 110,
            refresh: Some("refresh".into()),
        })
    }
    fn refresh(&self, _refresh: &str) -> Result<SignedIn, Error> {
        self.0.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        Err(Error::CannotAsk("fixture refresh failed"))
    }
}

#[test]
fn a_queued_request_shares_a_failed_refresh_across_clock_seconds() -> TestResult {
    use std::sync::atomic::{AtomicUsize, Ordering};
    let door = Door::new(FailedRefresh(AtomicUsize::new(0)));
    door.begin(100)?;
    let id = door.callback("code", "state", "browser", 100)?;
    let queued = std::time::Instant::now();
    assert_eq!(door.pass(&id, 100)?, "old");
    assert_eq!(door.pass_started(&id, 101, queued)?, "old");
    assert_eq!(
        door.issuer().0.load(Ordering::SeqCst),
        1,
        "a queued request repeated the failed in-flight refresh"
    );
    assert_eq!(door.pass(&id, 102)?, "old");
    assert_eq!(
        door.issuer().0.load(Ordering::SeqCst),
        2,
        "a later request could not start a new refresh"
    );
    Ok(())
}