diff --git a/crates/benshi/src/main.rs b/crates/benshi/src/main.rs index 43ab9bc..38dbf17 100644 --- a/crates/benshi/src/main.rs +++ b/crates/benshi/src/main.rs @@ -13,7 +13,7 @@ use std::fmt; use std::io::{self, Write}; use std::process::ExitCode; #[cfg(target_os = "linux")] -use std::sync::Arc; +use std::sync::{Arc, PoisonError}; #[cfg(target_os = "linux")] use std::task::Poll; #[cfg(target_os = "linux")] @@ -51,6 +51,10 @@ const STOPPED: u8 = 0; #[cfg(target_os = "linux")] const FAILED: u8 = 1; +/// Where a token for AniList may be given from the environment. +#[cfg(target_os = "linux")] +const TOKEN: &str = "BENSHI_TOKEN_ANILIST"; + /// What one invocation of this binary is for. /// /// Decided here rather than by the argument parser, because the two halves @@ -293,10 +297,11 @@ async fn serve(voice: &Arc) -> u8 { use benshi_core::clock::SystemClock; use benshi_daemon::accounts::OnAniList; use benshi_daemon::config; + use benshi_daemon::listing::{self, FromAniList, Listing}; use benshi_daemon::paths::{DATABASE, Role, directory}; - use benshi_daemon::recognition::Recogniser; use benshi_detect::mpris::{MprisWatcher, POLL_INTERVAL, SOURCE_DEADLINE}; use benshi_store::Store; + use benshi_sync::anilist::transport::ANILIST; // Taken ahead of the socket and the database, so that a signal sent // while one is bound and the other opened stops the daemon once its @@ -308,9 +313,8 @@ async fn serve(voice: &Arc) -> u8 { Err(code) => return code, }; // Read ahead of the socket and the database, so that a file that cannot - // be used ends the daemon having touched neither. No task reads a - // setting. - let (_settings, origin) = match config::load(directory(Role::Config)) { + // be used ends the daemon having touched neither. + let (settings, origin) = match config::load(directory(Role::Config)) { Ok(loaded) => loaded, Err(unusable) => { voice.say(format_args!("{unusable}")); @@ -354,22 +358,42 @@ async fn serve(voice: &Arc) -> u8 { return FAILED; } }; + // What the store holds of the list is what the first round decides + // against, so a daemon recognises before it reads the list again. + let recogniser = match listing::kept(&store.lock().expect("nothing has used the store yet")) { + Ok(recogniser) => recogniser, + Err(unread) => { + voice.say(format_args!( + "{} cannot be used: {unread}", + database.display() + )); + return FAILED; + } + }; + let given = given(); + let words = Arc::new(UpToTheLast::default()); voice.say(format_args!("{origin}")); voice.say(format_args!("listening on {}", socket.display())); voice.say(format_args!("recording in {}", database.display())); + if given.is_some() { + voice.say(format_args!("AniList token from {TOKEN}")); + } // A factory rather than a watcher: a restart has to reconnect to the // session bus, which is the failure it exists to recover from. let running = benshi_daemon::run( || MprisWatcher::connect(SystemClock::new(), SOURCE_DEADLINE), listener, - // No list to match a name against, so every file is refused and - // `benshi why` says so. - Recogniser::empty(), + recogniser, store, SystemTime::now, POLL_INTERVAL, - Arc::new(OnAniList::new(benshi_sync::anilist::transport::ANILIST)), + Arc::new(OnAniList::new(ANILIST)), + Listing { + source: FromAniList::new(ANILIST, given), + reading: settings.list, + say: saying(voice, &words), + }, // Said as it happens: the socket goes on answering once detection // has stopped, and the process does not end to say so. The // supervisor deals with no task that ended until this returns, so it @@ -393,11 +417,59 @@ async fn serve(voice: &Arc) -> u8 { // task as it is dropped. A task that is aborted ends at an await: the // one it waits at, or the next it reaches. The runtime waits for that // as it is dropped, and for a call into the store that is under way. - voice.say(format_args!("stopped by {signal}")); + words.last(|| voice.say(format_args!("stopped by {signal}"))); STOPPED } +/// The token for AniList given from the environment, where one is: for tests +/// and development, and said in no help text. A token given here is read in +/// place of the one a login keeps, for as long as the daemon runs. +#[cfg(target_os = "linux")] +fn given() -> Option { + std::env::var(TOKEN) + .ok() + .map(|token| token.trim().to_owned()) + .filter(|token| !token.is_empty()) + .map(benshi_core::token::Token::new) +} + +/// Lines said up to the daemon's last words, and none after them. +/// +/// A task that is stopped runs on to the next await it reaches. A call it +/// waits for can fail as the runtime goes, and the task would say so after +/// the line that says why the daemon stopped. +#[cfg(target_os = "linux")] +#[derive(Debug, Default)] +struct UpToTheLast(std::sync::Mutex); + +#[cfg(target_os = "linux")] +impl UpToTheLast { + /// Say a line through `say`, where the last words are not said yet. + fn say(&self, say: impl FnOnce()) { + let said_last = self.0.lock().unwrap_or_else(PoisonError::into_inner); + if !*said_last { + say(); + } + } + + /// Say the last words through `say`. Nothing is said after them. + fn last(&self, say: impl FnOnce()) { + let mut said_last = self.0.lock().unwrap_or_else(PoisonError::into_inner); + *said_last = true; + say(); + } +} + +/// A line a task says, said by `voice` up to the last words of `words`. +#[cfg(target_os = "linux")] +fn saying(voice: &Arc, words: &Arc) -> benshi_daemon::listing::Say { + let voice = Arc::clone(voice); + let words = Arc::clone(words); + + Arc::new(move |line: &str| words.say(|| voice.say(format_args!("{line}")))) +} + /// There is no adapter for this platform, so there is no daemon to run. /// /// Not async, and no runtime is built. A runtime with nothing to watch would @@ -418,6 +490,21 @@ mod tests { use std::io; use super::Invocation; + #[cfg(target_os = "linux")] + use super::UpToTheLast; + + #[cfg(target_os = "linux")] + #[test] + fn nothing_is_said_after_the_last_words() { + let words = UpToTheLast::default(); + let said = RefCell::new(Vec::new()); + + words.say(|| said.borrow_mut().push("a line")); + words.last(|| said.borrow_mut().push("the last words")); + words.say(|| said.borrow_mut().push("a line after them")); + + assert_eq!(said.into_inner(), ["a line", "the last words"]); + } #[cfg(target_os = "linux")] #[test] diff --git a/crates/benshi/tests/the_token_is_given.rs b/crates/benshi/tests/the_token_is_given.rs new file mode 100644 index 0000000..1fd66b8 --- /dev/null +++ b/crates/benshi/tests/the_token_is_given.rs @@ -0,0 +1,105 @@ +//! What the daemon says of a token for AniList given from the environment. +//! +//! The settings of these tests read the list at no time, so that no daemon +//! here asks AniList anything. +//! +//! The binary itself is run, in a session the test chose, because what is +//! under test is what a person reads. + +#![cfg(target_os = "linux")] + +mod common; + +use std::ffi::OsStr; +use std::fs; +use std::path::Path; + +use common::{Daemon, Directory}; + +/// The settings of a daemon that reads the list at no time. +const READING_NOTHING: &str = "[list]\nrefresh = \"off\"\nat_start = false\non_miss = \"off\"\n"; + +/// A session of its own in `temporary`, with settings that read the list at +/// no time. +fn a_session(temporary: &Path) -> Vec<(&'static str, std::ffi::OsString)> { + let config = temporary.join("config"); + fs::create_dir_all(config.join("benshi")).expect("a configuration directory"); + fs::write(config.join("benshi").join("config.toml"), READING_NOTHING) + .expect("a file of settings"); + + vec![ + ("XDG_RUNTIME_DIR", temporary.join("run").into_os_string()), + ("XDG_CONFIG_HOME", config.into_os_string()), + ("BENSHI_DATA_DIR", temporary.join("data").into_os_string()), + ] +} + +/// What a daemon in this session says next once it has said where it +/// records. +fn said_after_starting(session: &[(&str, &OsStr)]) -> Option { + let daemon = Daemon::start(session); + let started = daemon.says(); + assert!( + started + .last() + .is_some_and(|line| line.starts_with("benshi: recording in ")), + "{started:?}" + ); + + daemon.says_next() +} + +#[test] +fn a_daemon_given_a_token_from_the_environment_says_so_as_it_starts() { + let temporary = Directory::of("k-given"); + let mut session = a_session(temporary.path()); + session.push(("BENSHI_TOKEN_ANILIST", "a token".into())); + let session: Vec<(&str, &OsStr)> = session + .iter() + .map(|(name, value)| (*name, value.as_os_str())) + .collect(); + + let next = said_after_starting(&session); + + assert_eq!( + next.as_deref(), + Some("benshi: AniList token from BENSHI_TOKEN_ANILIST") + ); +} + +#[test] +fn a_daemon_given_an_empty_token_says_nothing_of_one_as_it_starts() { + let temporary = Directory::of("k-empty"); + let mut session = a_session(temporary.path()); + session.push(("BENSHI_TOKEN_ANILIST", " ".into())); + let session: Vec<(&str, &OsStr)> = session + .iter() + .map(|(name, value)| (*name, value.as_os_str())) + .collect(); + + let next = said_after_starting(&session); + + assert!( + next.as_deref() + .is_none_or(|line| !line.starts_with("benshi: AniList token")), + "{next:?}" + ); +} + +#[test] +fn a_daemon_given_no_token_says_nothing_of_one_as_it_starts() { + let temporary = Directory::of("k-none"); + let session = a_session(temporary.path()); + let session: Vec<(&str, &OsStr)> = session + .iter() + .map(|(name, value)| (*name, value.as_os_str())) + .collect(); + + let next = said_after_starting(&session); + + assert!( + next.as_deref() + .is_none_or(|line| !line.starts_with("benshi: AniList token")), + "{next:?}" + ); +} diff --git a/crates/cli/src/commands.rs b/crates/cli/src/commands.rs index af088b1..d614c27 100644 --- a/crates/cli/src/commands.rs +++ b/crates/cli/src/commands.rs @@ -529,6 +529,7 @@ mod tests { use benshi_daemon::bus::{BusEvent, EventBus}; use benshi_daemon::detection::Seen; use benshi_daemon::ipc::{Server, bind}; + use benshi_daemon::listing::Wakes; use benshi_daemon::protocol::{Request, Response}; use benshi_daemon::recognition::{Decided, Decision, Recogniser}; use benshi_daemon::secrets; @@ -815,6 +816,7 @@ mod tests { seen, decided.clone(), Arc::clone(&accounts) as Arc, + Wakes::new(), )); Self { diff --git a/crates/cli/tests/the_client_sees_what_the_daemon_does.rs b/crates/cli/tests/the_client_sees_what_the_daemon_does.rs index 426d45d..b36c813 100644 --- a/crates/cli/tests/the_client_sees_what_the_daemon_does.rs +++ b/crates/cli/tests/the_client_sees_what_the_daemon_does.rs @@ -30,7 +30,8 @@ use benshi_daemon::accounts::{Accounts, Done, LoggedIn}; use benshi_daemon::bus::{BusEvent, EventBus}; use benshi_daemon::detection::{Detection, Seen, Wiring}; use benshi_daemon::ipc::{Server, bind}; -use benshi_daemon::recognition::{Decided, Recogniser}; +use benshi_daemon::listing::Wakes; +use benshi_daemon::recognition::{Current, Decided, Recogniser}; use benshi_detect::{PlayerWatcher, PollOutcome, SourceInfo, WatchError}; use benshi_store::Store; use tempfile::TempDir; @@ -153,7 +154,8 @@ impl Daemon { bus: Arc::clone(&bus), policy: Arc::clone(&policy), seen: seen.clone(), - recogniser: Arc::new(Recogniser::empty()), + recogniser: Current::new(Recogniser::empty()), + wakes: Wakes::new(), decided: decided.clone(), store: Arc::new(Mutex::new(store)), now: SystemTime::now, @@ -170,6 +172,7 @@ impl Daemon { seen, decided, Arc::new(NoLogins), + Wakes::new(), )); drop(tokio::spawn(server.listen(Arc::new(listener)))); diff --git a/crates/daemon/src/detection.rs b/crates/daemon/src/detection.rs index ec97d2f..e4a9d48 100644 --- a/crates/daemon/src/detection.rs +++ b/crates/daemon/src/detection.rs @@ -66,7 +66,8 @@ use benshi_store::{Store, Viewing}; use tokio::time::MissedTickBehavior; use crate::bus::{BusEvent, EventBus}; -use crate::recognition::{Decided, Decision, Recogniser}; +use crate::listing::Wakes; +use crate::recognition::{Current, Decided, Decision, Recogniser}; use crate::supervisor::TaskError; /// What is said about a reading whose source the same round did not describe. @@ -83,7 +84,7 @@ const POISONED: &str = "policy table was poisoned by a panic in another task"; const POISONED_SEEN: &str = "what was last seen was poisoned by a panic in another task"; /// What is said when the store cannot be trusted. -const POISONED_STORE: &str = "store was poisoned by a panic in a call into it"; +pub(crate) const POISONED_STORE: &str = "store was poisoned by a panic in a call into it"; /// Every source the last round was able to describe. /// @@ -194,7 +195,9 @@ pub struct Wiring { /// Where each round leaves the sources it described. pub seen: Seen, /// What a reading's file is decided against. - pub recogniser: Arc, + pub recogniser: Current, + /// What a name the list does not hold wakes. + pub wakes: Wakes, /// Where each round leaves its last decision. pub decided: Decided, /// Where what was watched is kept and an episode is recorded. @@ -254,7 +257,8 @@ pub struct Detection { bus: Arc, policy: Arc>, seen: Seen, - recogniser: Arc, + recogniser: Current, + wakes: Wakes, decided: Decided, store: Arc>, now: fn() -> SystemTime, @@ -279,6 +283,7 @@ impl Detection { policy, seen, recogniser, + wakes, decided, store, now, @@ -291,6 +296,7 @@ impl Detection { policy, seen, recogniser, + wakes, decided, store, now, @@ -331,6 +337,7 @@ impl Detection { /// One round: read the platform, apply policy, record and publish what /// was admitted, and let go of whoever has left. async fn round(&mut self) -> Result<(), TaskError> { + let recogniser = self.recogniser.now(); let PollOutcome { sources, snapshots, @@ -375,7 +382,7 @@ impl Detection { if !self.heard.contains(&snapshot.player) { self.heard.insert(snapshot.player.clone()); } - let decision = self.decide(&snapshot); + let decision = self.decide(&recogniser, &snapshot); let effects = self.session.advance(&snapshot, decision.as_ref()); self.carry_out(effects).await?; self.bus.publish(BusEvent::Snapshot(snapshot)); @@ -425,12 +432,18 @@ impl Detection { /// recognition reads filenames, and neither is one, so neither is answered /// for. Every admitted reading of a file is decided, every round, so that /// what a client reads is the decision about the reading the daemon last - /// acted on. - fn decide(&self, snapshot: &PlayerSnapshot) -> Option { + /// acted on. A name that spells a title the list does not hold wakes the + /// list's task, which may read the list again. + fn decide(&self, recogniser: &Recogniser, snapshot: &PlayerSnapshot) -> Option { let MediaRef::LocalFile(path) = &snapshot.media else { return None; }; - let (parsed, answer) = self.recogniser.decide(path); + let (parsed, answer) = recogniser.decide(path); + if let Recognition::Unrecognised(refusal) = &answer + && !refusal.parsed.is_empty() + { + self.wakes.missed(); + } self.decided.record(Decision { player: snapshot.player.clone(), @@ -538,7 +551,8 @@ impl Detection { mod tests { use super::{Detection, Seen, Wiring}; use crate::bus::{BusEvent, EventBus}; - use crate::recognition::{Decided, Recogniser}; + use crate::listing::Wakes; + use crate::recognition::{Current, Decided, Recogniser}; use crate::supervisor::TaskError; use benshi_core::clock::{Clock, TestClock, Timestamp}; use benshi_core::path::RawPath; @@ -796,7 +810,8 @@ mod tests { bus: Arc::new(EventBus::new()), policy: Arc::new(RwLock::new(PolicyTable::allowing_video_players())), seen: Seen::new(), - recogniser: Arc::new(Recogniser::empty()), + recogniser: Current::new(Recogniser::empty()), + wakes: Wakes::new(), decided: Decided::new(), store: Arc::clone(&database.store), now: the_time, @@ -852,7 +867,7 @@ mod tests { ) -> Detection { let wiring = Wiring { policy, - recogniser: Arc::new(a_list()), + recogniser: Current::new(a_list()), ..wired_to(database) }; @@ -1711,7 +1726,7 @@ mod tests { let mpv = a_source("mpv", "mpv"); let watcher = ScriptedWatcher::of(watching(&mpv, THIRD, 0..=HALF_AT)); let wiring = Wiring { - recogniser: Arc::new(a_list()), + recogniser: Current::new(a_list()), now: a_second_on, ..wired_to(&database) }; @@ -2184,7 +2199,7 @@ mod tests { let mpv = a_source("mpv", "mpv"); let watcher = ScriptedWatcher::of(watching(&mpv, THIRD, 0..=1)); let wiring = Wiring { - recogniser: Arc::new(a_list()), + recogniser: Current::new(a_list()), now: before_1970, ..wired_to(&database) }; @@ -2279,4 +2294,187 @@ mod tests { "{failure}" ); } + + /// A platform that puts `with` in place of the list while it is asked + /// for its first round, as the list's task can while a round runs. + struct Replacing { + rounds: ScriptedWatcher, + current: Current, + with: Option, + } + + impl PlayerWatcher for Replacing { + async fn sources(&mut self) -> Result, WatchError> { + self.rounds.sources().await + } + + async fn poll(&mut self) -> Result { + if let Some(with) = self.with.take() { + self.current.replace(with); + } + self.rounds.poll().await + } + } + + /// Detection that decides against `current` and wakes `wakes`, leaving + /// its decisions in `decided`. + fn deciding_against( + watcher: W, + current: Current, + wakes: Wakes, + decided: Decided, + ) -> (Database, Detection) { + let database = Database::new(); + let wiring = Wiring { + recogniser: current, + wakes, + decided, + ..wired_to(&database) + }; + + (database, Detection::new(watcher, wiring)) + } + + /// Rounds that each read this file from mpv. + fn reading(name: &str) -> ScriptedWatcher { + let mpv = a_source("mpv", "mpv"); + ScriptedWatcher::of(vec![ + Ok(a_round_of(&[&mpv], vec![a_reading_of_a_file(&mpv, name)])), + Ok(a_round_of(&[&mpv], vec![a_reading_of_a_file(&mpv, name)])), + ]) + } + + /// Whether the list's task was woken by a name that missed the list. + /// + /// A wake given before anyone waits is kept for the first who does, so + /// a short wait is enough for one given in a round already over. + async fn woken(wakes: &Wakes) -> bool { + tokio::time::timeout(Duration::from_millis(50), wakes.missed.notified()) + .await + .is_ok() + } + + #[tokio::test] + async fn a_round_that_began_before_a_list_was_put_in_place_decides_against_the_one_before() { + let current = Current::new(Recogniser::empty()); + let decided = Decided::new(); + let watcher = Replacing { + rounds: reading(THIRD), + current: current.clone(), + with: Some(a_list()), + }; + let (_database, mut detection) = + deciding_against(watcher, current, Wakes::new(), decided.clone()); + + detection.round().await.expect("a round"); + + let decision = decided.latest().expect("the reading was decided"); + assert!( + matches!(decision.answer, Recognition::Unrecognised(_)), + "got {:?}", + decision.answer + ); + } + + #[tokio::test] + async fn a_list_put_in_place_is_decided_against_from_the_next_round() { + let current = Current::new(Recogniser::empty()); + let decided = Decided::new(); + let (_database, mut detection) = deciding_against( + reading(THIRD), + current.clone(), + Wakes::new(), + decided.clone(), + ); + + detection.round().await.expect("a round"); + current.replace(a_list()); + detection.round().await.expect("a round"); + + let decision = decided.latest().expect("the reading was decided"); + assert!( + matches!(&decision.answer, Recognition::Recognised(found) if found.title == SERIES), + "got {:?}", + decision.answer + ); + } + + #[tokio::test] + async fn a_name_the_list_does_not_hold_wakes_the_lists_task() { + let wakes = Wakes::new(); + let (_database, mut detection) = deciding_against( + reading("/anime/[Group] Another Show Entirely - 03.mkv"), + Current::new(a_list()), + wakes.clone(), + Decided::new(), + ); + + detection.round().await.expect("a round"); + + assert!(woken(&wakes).await, "a miss woke nothing"); + } + + #[tokio::test] + async fn a_name_the_list_holds_wakes_nothing() { + let wakes = Wakes::new(); + let (_database, mut detection) = deciding_against( + reading(THIRD), + Current::new(a_list()), + wakes.clone(), + Decided::new(), + ); + + detection.round().await.expect("a round"); + + assert!(!woken(&wakes).await, "a name the list holds woke the task"); + } + + #[tokio::test] + async fn a_name_that_spells_no_title_wakes_nothing() { + let wakes = Wakes::new(); + let decided = Decided::new(); + let (_database, mut detection) = deciding_against( + reading("/anime/03.mkv"), + Current::new(a_list()), + wakes.clone(), + decided.clone(), + ); + + detection.round().await.expect("a round"); + + let decision = decided.latest().expect("the reading was decided"); + assert_eq!(decision.parsed.title, None, "the name spells a title"); + assert!( + !woken(&wakes).await, + "a name that spells no title woke the task" + ); + } + + #[tokio::test] + async fn a_name_two_entries_of_the_list_hold_wakes_nothing() { + let mut corpus = Corpus::new(); + corpus.file(ShowId(1), SERIES, Vec::<&str>::new()); + corpus.file(ShowId(2), SERIES, Vec::<&str>::new()); + let wakes = Wakes::new(); + let decided = Decided::new(); + let (_database, mut detection) = deciding_against( + reading(THIRD), + Current::new(Recogniser::of(corpus, Altnames::new())), + wakes.clone(), + decided.clone(), + ); + + detection.round().await.expect("a round"); + + let decision = decided.latest().expect("the reading was decided"); + assert!( + matches!(decision.answer, Recognition::Ambiguous(_)), + "got {:?}", + decision.answer + ); + assert!( + !woken(&wakes).await, + "a name the list holds twice woke the task" + ); + } } diff --git a/crates/daemon/src/ipc.rs b/crates/daemon/src/ipc.rs index 2867665..375f50e 100644 --- a/crates/daemon/src/ipc.rs +++ b/crates/daemon/src/ipc.rs @@ -26,6 +26,7 @@ use tokio::task::JoinSet; use crate::accounts::{Accounts, LoggedIn}; use crate::bus::EventBus; use crate::detection::Seen; +use crate::listing::Wakes; use crate::protocol::{Explanation, Request, Response, SourceListing}; use crate::recognition::Decided; use crate::supervisor::TaskError; @@ -269,13 +270,14 @@ pub struct Server { seen: Seen, decided: Decided, accounts: Arc, + wakes: Wakes, } impl Server { /// A server over the bus a `Watch` streams, the table a listing reads and a /// policy is set in, the two records each detection round leaves behind, - /// what it saw and what it last decided, and what a login and a logout - /// are carried out by. + /// what it saw and what it last decided, what a login and a logout are + /// carried out by, and the list's task, which a login wakes. #[must_use] pub fn new( bus: Arc, @@ -283,6 +285,7 @@ impl Server { seen: Seen, decided: Decided, accounts: Arc, + wakes: Wakes, ) -> Self { Self { bus, @@ -290,6 +293,7 @@ impl Server { seen, decided, accounts, + wakes, } } @@ -395,7 +399,10 @@ impl Server { Ok(Request::Login { token }) => { reply(&mut writer, &Response::Ok).await?; match self.accounts.login(token).await { - Ok(LoggedIn { name, expires }) => Response::LoggedIn { name, expires }, + Ok(LoggedIn { name, expires }) => { + self.wakes.logged_in(); + Response::LoggedIn { name, expires } + } Err(failed) => Response::Error { message: failed.to_string(), }, @@ -500,6 +507,7 @@ mod tests { use crate::accounts::{self, Accounts, Done, LoggedIn}; use crate::bus::{BusEvent, EventBus}; use crate::detection::Seen; + use crate::listing::Wakes; use crate::protocol::{Outcome, Request, Response}; use crate::recognition::{Decided, Decision, Recogniser}; use crate::secrets; @@ -657,6 +665,7 @@ mod tests { policy: Arc>, decided: Decided, accounts: Arc, + wakes: Wakes, } impl Daemon { @@ -681,6 +690,7 @@ mod tests { seen.record(sources); let decided = Decided::new(); let accounts = Arc::new(accounts); + let wakes = Wakes::new(); Self { server: Arc::new(Server::new( @@ -689,11 +699,13 @@ mod tests { seen, decided.clone(), Arc::clone(&accounts) as Arc, + wakes.clone(), )), bus, policy, decided, accounts, + wakes, } } @@ -946,6 +958,48 @@ mod tests { ); } + /// Whether the list's task was woken by a login. + async fn woken_by_a_login(wakes: &Wakes) -> bool { + timeout(Duration::from_millis(50), wakes.logged_in.notified()) + .await + .is_ok() + } + + #[tokio::test] + async fn a_login_that_goes_through_wakes_the_lists_task() { + let daemon = Daemon::logging_in(Scripted::going_through()); + let mut client = Client::to(&daemon, 4096); + + client + .ask(&Request::Login { + token: Token::new("a token"), + }) + .await; + assert_eq!(client.answer().await, Response::Ok); + let _logged_in = client.answer().await; + + assert!(woken_by_a_login(&daemon.wakes).await, "nothing was woken"); + } + + #[tokio::test] + async fn a_login_that_fails_wakes_nothing() { + let daemon = Daemon::logging_in(Scripted::failing()); + let mut client = Client::to(&daemon, 4096); + + client + .ask(&Request::Login { + token: Token::new("a token"), + }) + .await; + assert_eq!(client.answer().await, Response::Ok); + let _failed = client.answer().await; + + assert!( + !woken_by_a_login(&daemon.wakes).await, + "a failed login woke the task" + ); + } + #[tokio::test] async fn a_logout_is_answered_once_it_is_read_and_again_once_it_is_done() { let daemon = Daemon::logging_in(Scripted::held()); diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs index 180d1be..4b68320 100644 --- a/crates/daemon/src/lib.rs +++ b/crates/daemon/src/lib.rs @@ -29,7 +29,8 @@ use crate::bus::EventBus; use crate::detection::{Detection, Seen, Wiring}; #[cfg(unix)] use crate::ipc::Server; -use crate::recognition::{Decided, Recogniser}; +use crate::listing::{Listing, Reader, Source, Wakes}; +use crate::recognition::{Current, Decided, Recogniser}; #[cfg(unix)] use crate::supervisor::{Notice, Supervisor, TaskError, TaskRecord}; @@ -44,6 +45,8 @@ pub mod detection; #[cfg(unix)] pub mod ipc; +pub mod listing; + pub mod panics; pub mod paths; @@ -64,12 +67,13 @@ pub mod voice; /// nothing could ask it what it had seen. The platform that needs a named /// pipe instead gains its own. /// -/// Two supervised tasks over state they share: detection, which reads the -/// platform and publishes what it saw, and the socket, which answers clients -/// from it. They are separate tasks on purpose. A session bus that goes away -/// must not take the interface down with it, because the user whose detection -/// has broken is exactly the user who needs to ask the daemon what it thinks -/// is happening. +/// Three supervised tasks over state they share: detection, which reads the +/// platform and publishes what it saw, the socket, which answers clients +/// from it, and the list's task, which reads the list and puts what it read +/// in place of what detection decides against. They are separate tasks on +/// purpose. A session bus that goes away must not take the interface down +/// with it, because the user whose detection has broken is exactly the user +/// who needs to ask the daemon what it thinks is happening. /// /// `start_watcher` builds a watcher rather than being handed one, because the /// supervisor calls a task's body again on every restart. A restart that @@ -85,12 +89,14 @@ pub mod voice; /// The store is opened by the caller for the same reason, and it outlives a /// restart of detection: a detection task that is started again knows nothing /// of the one before it and goes on from what that one kept. `recogniser` is -/// what a file's name is decided against, and `now` reads the time of day the -/// store writes beside what it keeps. `accounts` carries out a login and a -/// logout the socket is asked for. +/// what a file's name is decided against until the list's task puts a list +/// in its place, and `now` reads the time of day the store writes beside what +/// it keeps. `accounts` carries out a login and a +/// logout the socket is asked for, and `list` says what the list's task reads +/// from and when. /// /// `tell` is told of a task as the task is started again and as it stops, -/// whatever the other task is doing. A daemon whose detection has stopped +/// whatever the other tasks are doing. A daemon whose detection has stopped /// goes on answering its socket, and nothing else says that it records /// nothing. The supervisor deals with no task that ended until `tell` has /// returned, so a `tell` that writes hands the line to a @@ -103,7 +109,7 @@ pub mod voice; clippy::too_many_arguments, reason = "each part is built by the caller, so that a test can give its own" )] -pub async fn run( +pub async fn run( mut start_watcher: B, listener: Arc, recogniser: Recogniser, @@ -111,6 +117,7 @@ pub async fn run( now: fn() -> SystemTime, period: Duration, accounts: Arc, + list: Listing, tell: T, ) -> Vec where @@ -118,12 +125,14 @@ where F: Future> + Send + 'static, W: PlayerWatcher + Send + 'static, T: FnMut(&Notice) + Send + 'static, + S: Source + Send + Sync + 'static, { let wiring = Wiring { bus: Arc::new(EventBus::new()), policy: Arc::new(RwLock::new(PolicyTable::allowing_video_players())), seen: Seen::new(), - recogniser: Arc::new(recogniser), + recogniser: Current::new(recogniser), + wakes: Wakes::new(), decided: Decided::new(), store, now, @@ -135,6 +144,15 @@ where wiring.seen.clone(), wiring.decided.clone(), accounts, + wiring.wakes.clone(), + )); + let reader = Arc::new(Reader::new( + list.source, + list.reading, + Arc::clone(&wiring.store), + wiring.recogniser.clone(), + wiring.wakes.clone(), + list.say, )); let mut supervisor = Supervisor::telling(tell); @@ -161,5 +179,11 @@ where async move { server.listen(listener).await } }); + supervisor.supervise("list", move || { + let reader = Arc::clone(&reader); + + async move { reader.run().await } + }); + supervisor.run().await } diff --git a/crates/daemon/src/listing.rs b/crates/daemon/src/listing.rs new file mode 100644 index 0000000..fb10fc1 --- /dev/null +++ b/crates/daemon/src/listing.rs @@ -0,0 +1,956 @@ +//! Reading the list, and what a round decides against. +//! +//! The list's task reads the account's list from AniList, files it in the +//! store, and puts a recogniser of it in place of the one the rounds decide +//! against. It reads at the times `[list]` names: as the daemon starts, every +//! `refresh`, and where a round decided that a name spelling a title is not +//! on the list, at most once in `on_miss`. A login wakes it at once. +//! +//! A reading that fails is tried again after [`FIRST_RETRY`], and after +//! twice as long each time it fails again, up to [`LAST_RETRY`]. A token +//! AniList refuses is not tried again until a login. A token that is not +//! found, or that the secret store cannot give, is looked for again at the +//! next `refresh`. A failure is said once, and said again only where its +//! reason changes, or where a reading went through or a login came in +//! between. A name that misses the list brings a reading only where the last +//! one went through. + +use std::future::Future; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use benshi_core::ShowId; +use benshi_core::list::Entry; +use benshi_core::recognise::altname::Altnames; +use benshi_core::recognise::corpus::Corpus; +use benshi_core::token::Token; +use benshi_store::Store; +#[cfg(target_os = "linux")] +use benshi_sync::anilist::AniList; +#[cfg(target_os = "linux")] +use benshi_sync::anilist::transport::Http; +use humantime::format_duration; +use tokio::sync::Notify; +use tokio::time::Instant; + +use crate::config::{Period, Reading}; +use crate::detection::POISONED_STORE; +use crate::recognition::{Current, Recogniser}; +use crate::secrets::{self, ANILIST}; +#[cfg(target_os = "linux")] +use crate::secrets::{Asking, SecretService, Secrets}; +use crate::supervisor::TaskError; + +/// How long a reading that failed waits before it is tried again. +pub const FIRST_RETRY: Duration = Duration::from_mins(1); + +/// The longest a reading that failed again and again waits. +pub const LAST_RETRY: Duration = Duration::from_hours(1); + +/// What is said where no token is found. +const NO_TOKEN: &str = "no AniList token, so no list is read: benshi login keeps one"; + +/// The backend a list is filed under in the store. +const BACKEND: &str = ANILIST.backend; + +/// What wakes the list's task before its time: a name a round decided the +/// list does not hold, and a token a login kept. Cloning shares them. +#[derive(Debug, Clone, Default)] +pub struct Wakes { + pub(crate) missed: Arc, + pub(crate) logged_in: Arc, +} + +impl Wakes { + /// Nothing has woken the task yet. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// A round decided that a name the list does not hold spelled a title. + pub fn missed(&self) { + self.missed.notify_one(); + } + + /// A login kept a token. + pub fn logged_in(&self) { + self.logged_in.notify_one(); + } +} + +/// Where the token and the list come from. +pub trait Source { + /// The token to read the list with, and none where none is kept. + fn token(&self) -> impl Future, secrets::Error>> + Send; + + /// The list `token` is of. + fn list( + &self, + token: &Token, + ) -> impl Future, benshi_sync::Error>> + Send; +} + +/// The token and the list from AniList. +/// +/// The token is the one given from the environment where there is one, and +/// else the one the Secret Service of the session keeps, which is never asked +/// to unlock: the daemon opens no window of its own. +#[cfg(target_os = "linux")] +#[derive(Debug)] +pub struct FromAniList { + address: String, + given: Option, +} + +#[cfg(target_os = "linux")] +impl FromAniList { + /// Lists read from the service at `address`, with `given` as the token + /// where there is one. + #[must_use] + pub fn new(address: &str, given: Option) -> Self { + Self { + address: address.to_owned(), + given, + } + } +} + +#[cfg(target_os = "linux")] +impl Source for FromAniList { + async fn token(&self) -> Result, secrets::Error> { + if let Some(given) = &self.given { + return Ok(Some(given.clone())); + } + let store = SecretService::connect(secrets::DEADLINE, Asking::Never).await?; + + Ok(store.read(ANILIST).await?.map(Token::new)) + } + + async fn list(&self, token: &Token) -> Result, benshi_sync::Error> { + let http = Http::new(&self.address, token.secret()) + .map_err(|unbuilt| benshi_sync::Error::Unreachable(unbuilt.into()))?; + + AniList::new(http).list().await + } +} + +/// What the task says, a line at a time. +pub type Say = Arc; + +/// What the list's task reads from, when, and how it says what it does. +pub struct Listing { + /// Where the token and the list come from. + pub source: S, + /// When the list is read. + pub reading: Reading, + /// How the task says a line. + pub say: Say, +} + +/// A recogniser of these shows of a list and their entries. +#[must_use] +pub fn recogniser_of(listed: &[(ShowId, Entry)]) -> Recogniser { + let mut corpus = Corpus::new(); + for (show, entry) in listed { + corpus.file(*show, &entry.title, &entry.spellings); + } + + Recogniser::of(corpus, Altnames::new()) +} + +/// A recogniser of the list the store holds. +/// +/// # Errors +/// +/// What [`Store::listed`] fails with. +pub fn kept(store: &Store) -> Result { + store.listed(BACKEND).map(|listed| recogniser_of(&listed)) +} + +/// The list's task: what it reads from, when, and where it puts what it read. +pub struct Reader { + source: S, + reading: Reading, + store: Arc>, + current: Current, + wakes: Wakes, + say: Say, +} + +impl Reader { + /// A task reading from `source` at the times `reading` names, filing + /// into `store` and putting what it read in `current`, woken by `wakes` + /// and saying what it does through `say`. + #[must_use] + pub fn new( + source: S, + reading: Reading, + store: Arc>, + current: Current, + wakes: Wakes, + say: Say, + ) -> Self { + Self { + source, + reading, + store, + current, + wakes, + say, + } + } + + /// Read the list at the times the settings name, for ever. + /// + /// # Errors + /// + /// None: every failure of a reading is said, and the task goes on. The + /// result is the supervisor's. + pub async fn run(&self) -> Result<(), TaskError> { + let mut outcome = Outcome::Read; + let mut wait = FIRST_RETRY; + let mut said = None; + let mut last: Option = None; + let mut due = if self.reading.at_start { + Some(Instant::now()) + } else { + self.refreshed() + }; + + loop { + tokio::select! { + () = self.wakes.logged_in.notified() => said = None, + () = until(due) => {} + // Taken whatever came of the last reading, so that a miss + // heard while readings fail is not kept until one goes + // through. + () = self.wakes.missed.notified() => { + if outcome == Outcome::Read + && let Period::Every(on_miss) = self.reading.on_miss + { + let now = Instant::now(); + let earliest = last.map_or(now, |last| (last + on_miss).max(now)); + due = Some(due.map_or(earliest, |due| due.min(earliest))); + } + continue; + } + } + + last = Some(Instant::now()); + outcome = self.read(&mut said, wait).await; + due = match outcome { + Outcome::Read => { + wait = FIRST_RETRY; + self.refreshed() + } + Outcome::NoToken => self.refreshed(), + Outcome::Refused => None, + Outcome::Failed { asked: Some(after) } => Some(Instant::now() + after), + Outcome::Failed { asked: None } => { + let retried = Instant::now() + wait; + wait = (wait * 2).min(LAST_RETRY); + Some(retried) + } + }; + } + } + + /// When the next reading is due by `refresh`, and none where it is off. + fn refreshed(&self) -> Option { + match self.reading.refresh { + Period::Every(every) => Some(Instant::now() + every), + Period::Off => None, + } + } + + /// Read the token and the list, file the list and put a recogniser of + /// it in place, and say what came of it. `wait` is how long a failure + /// waits before the reading is tried again. + async fn read(&self, said: &mut Option, wait: Duration) -> Outcome { + let token = match self.source.token().await { + Ok(Some(token)) => token, + Ok(None) => { + self.tell(said, NO_TOKEN.to_owned(), || NO_TOKEN.to_owned()); + return Outcome::NoToken; + } + Err(unread) => { + let line = format!("AniList token could not be read: {unread}"); + self.tell_once(said, line); + return Outcome::NoToken; + } + }; + + match self.source.list(&token).await { + Ok(entries) => { + let count = entries.len(); + match self.filed(entries).await { + Ok(recogniser) => { + self.current.replace(recogniser); + *said = None; + let entries = if count == 1 { "entry" } else { "entries" }; + (self.say)(&format!("AniList list read: {count} {entries}")); + Outcome::Read + } + Err(unkept) => { + let line = format!("AniList list could not be kept: {unkept}"); + self.tell_once(said, line); + Outcome::Failed { asked: None } + } + } + } + Err(benshi_sync::Error::Refused(why)) => { + let line = format!("AniList refused the token: {why}; benshi login gives another"); + self.tell_once(said, line); + Outcome::Refused + } + Err(failed) => { + let asked = match failed { + benshi_sync::Error::Wait { after } => Some(after), + _ => None, + }; + let reason = format!("AniList list could not be read: {failed}"); + let retried = format_duration(asked.unwrap_or(wait)); + self.tell(said, reason.clone(), || { + format!("{reason}; tried again in {retried}") + }); + Outcome::Failed { asked } + } + } + } + + /// File `entries` as the list, and answer with a recogniser of what the + /// store then holds, or with why it could not be kept. + async fn filed(&self, entries: Vec) -> Result { + let store = Arc::clone(&self.store); + let filing = tokio::task::spawn_blocking(move || { + let mut store = store.lock().expect(POISONED_STORE); + store.file(BACKEND, &entries)?; + store.listed(BACKEND) + }); + + match filing.await { + Ok(listed) => listed + .map(|listed| recogniser_of(&listed)) + .map_err(|unkept| unkept.to_string()), + // Handed on as it was raised. + Err(panicked) if panicked.is_panic() => { + std::panic::resume_unwind(panicked.into_panic()) + } + Err(given_up) => Err(format!("store could not be asked: {given_up}")), + } + } + + /// Say `line` where it is not what was said last. + fn tell_once(&self, said: &mut Option, line: String) { + self.tell(said, line.clone(), || line); + } + + /// Say `line` where `reason` is not what was said last. + fn tell(&self, said: &mut Option, reason: String, line: impl FnOnce() -> String) { + if said.as_deref() != Some(reason.as_str()) { + (self.say)(&line()); + *said = Some(reason); + } + } +} + +/// What came of a reading. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Outcome { + /// The list was read and filed. + Read, + /// No token was found, or none could be read. + NoToken, + /// AniList refused the token. + Refused, + /// The reading failed, and where AniList asked for a wait, how long. + Failed { asked: Option }, +} + +/// Sleep until `due`, and for ever where nothing is due. +async fn until(due: Option) { + match due { + Some(at) => tokio::time::sleep_until(at).await, + None => std::future::pending().await, + } +} + +#[cfg(test)] +mod tests { + use std::collections::VecDeque; + use std::future::{self, Future}; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + + use benshi_core::list::{Entry, EntryId, Status}; + use benshi_core::path::RawPath; + use benshi_core::recognise::Recognition; + use benshi_core::token::Token; + use benshi_store::Store; + use tempfile::TempDir; + use tokio::task::JoinHandle; + use tokio::time::Instant; + + use super::{BACKEND, FIRST_RETRY, LAST_RETRY, Reader, Source, Wakes, kept, recogniser_of}; + use crate::config::{Period, Reading}; + use crate::recognition::{Current, Recogniser}; + use crate::secrets; + use crate::supervisor::TaskError; + + const HOUR: Duration = Duration::from_hours(1); + const MINUTE: Duration = Duration::from_mins(1); + + /// An entry of a list, under a number of a work no list of the user's + /// holds. + fn an_entry(id: i64, title: &str) -> Entry { + Entry { + id: EntryId(id), + title: title.to_owned(), + spellings: vec![format!("{title} Also")], + format: None, + episodes: Some(12), + episode_length: None, + year: None, + released: true, + status: Status::Watching, + progress: 2, + rewatches: 0, + score: None, + } + } + + fn the_list() -> Vec { + vec![ + an_entry(900_001, "Show Title"), + an_entry(900_002, "Another Show"), + ] + } + + /// Whether `recogniser` recognises a file of this title as it. + fn recognises(recogniser: &Recogniser, title: &str) -> bool { + let path = RawPath::from_bytes(format!("[Group] {title} - 03.mkv").into_bytes()); + matches!( + recogniser.decide(&path).1, + Recognition::Recognised(found) if found.title == title + ) + } + + /// A source that answers as a test scripts it, and a token and the list + /// once the script is spent; every reading of the list is noted. + struct Scripted { + tokens: Mutex, secrets::Error>>>, + lists: Mutex, benshi_sync::Error>>>, + read_at: Mutex>, + began: Instant, + } + + impl Scripted { + fn new() -> Self { + Self { + tokens: Mutex::new(VecDeque::new()), + lists: Mutex::new(VecDeque::new()), + read_at: Mutex::new(Vec::new()), + began: Instant::now(), + } + } + + fn tokens( + self, + tokens: impl IntoIterator, secrets::Error>>, + ) -> Self { + self.tokens.lock().expect("the script").extend(tokens); + self + } + + fn lists( + self, + lists: impl IntoIterator, benshi_sync::Error>>, + ) -> Self { + self.lists.lock().expect("the script").extend(lists); + self + } + } + + impl Source for Arc { + fn token(&self) -> impl Future, secrets::Error>> + Send { + let next = self + .tokens + .lock() + .expect("the script") + .pop_front() + .unwrap_or_else(|| Ok(Some(Token::new("a token")))); + future::ready(next) + } + + fn list( + &self, + _token: &Token, + ) -> impl Future, benshi_sync::Error>> + Send { + self.read_at + .lock() + .expect("the record") + .push(self.began.elapsed()); + let next = self + .lists + .lock() + .expect("the script") + .pop_front() + .unwrap_or_else(|| Ok(the_list())); + future::ready(next) + } + } + + /// A list's task running against a script, with all it touches. + struct Running { + _home: TempDir, + store: Arc>, + current: Current, + wakes: Wakes, + said: Arc>>, + source: Arc, + task: JoinHandle>, + } + + impl Running { + fn reads(&self) -> Vec { + self.source.read_at.lock().expect("the record").clone() + } + + fn said(&self) -> Vec { + self.said.lock().expect("the record").clone() + } + + fn filed(&self) -> Vec { + self.store + .lock() + .expect("the store") + .listed(BACKEND) + .expect("the list reads") + .into_iter() + .map(|(_show, entry)| entry.title) + .collect() + } + } + + impl Drop for Running { + fn drop(&mut self) { + self.task.abort(); + } + } + + /// The list's task with these settings, reading from `source`. + fn started(reading: Reading, source: Scripted) -> Running { + let home = tempfile::tempdir().expect("a directory of our own"); + let store = Store::open(&home.path().join("benshi.db")).expect("the database opens"); + + running(home, store, reading, source) + } + + /// [`started`], over a store that refuses every entry it is given. + fn started_refusing(reading: Reading, source: Scripted) -> Running { + let home = tempfile::tempdir().expect("a directory of our own"); + let path = home.path().join("benshi.db"); + let store = Store::open(&path).expect("the database opens"); + rusqlite::Connection::open(&path) + .expect("the file opens a second time") + .execute_batch( + "CREATE TRIGGER refused BEFORE INSERT ON list_entries \ + BEGIN SELECT RAISE(ABORT, 'refused by the test'); END;", + ) + .expect("the trigger is made"); + + running(home, store, reading, source) + } + + fn running(home: TempDir, store: Store, reading: Reading, source: Scripted) -> Running { + let store = Arc::new(Mutex::new(store)); + let current = Current::new(Recogniser::empty()); + let wakes = Wakes::new(); + let said = Arc::new(Mutex::new(Vec::new())); + let source = Arc::new(source); + let reader = Reader::new( + Arc::clone(&source), + reading, + Arc::clone(&store), + current.clone(), + wakes.clone(), + { + let said = Arc::clone(&said); + Arc::new(move |line: &str| said.lock().expect("the record").push(line.to_owned())) + }, + ); + let task = tokio::spawn(async move { reader.run().await }); + + Running { + _home: home, + store, + current, + wakes, + said, + source, + task, + } + } + + fn reading(refresh: Period, at_start: bool, on_miss: Period) -> Reading { + Reading { + refresh, + at_start, + on_miss, + } + } + + /// Let this much time pass, and the task do what falls in it. + async fn after(passing: Duration) { + tokio::time::sleep(passing).await; + } + + /// A moment, for the task to do what is due now. + const NOW: Duration = Duration::from_millis(1); + + #[test] + fn a_recogniser_of_a_list_recognises_each_entry_under_each_spelling() { + let listed: Vec<_> = the_list() + .into_iter() + .enumerate() + .map(|(place, entry)| { + ( + benshi_core::ShowId(i64::try_from(place).expect("small") + 1), + entry, + ) + }) + .collect(); + let mut spelled = listed.clone(); + // A spelling that shares no word with the title, so the name reaches + // the entry only where its spellings are filed. + spelled[0].1.spellings = vec!["Kotonaru Namae".to_owned()]; + + let recogniser = recogniser_of(&listed); + let spelled = recogniser_of(&spelled); + + assert!(recognises(&recogniser, "Show Title")); + assert!(recognises(&recogniser, "Another Show")); + let (_parsed, answer) = spelled.decide(&RawPath::from_bytes( + b"[Group] Kotonaru Namae - 03.mkv".to_vec(), + )); + assert!( + matches!(&answer, Recognition::Recognised(found) if found.title == "Show Title"), + "{answer:?}" + ); + } + + #[test] + fn a_recogniser_of_the_store_holds_the_list_filed_there() { + let home = tempfile::tempdir().expect("a directory of our own"); + let mut store = Store::open(&home.path().join("benshi.db")).expect("the database opens"); + store.file(BACKEND, &the_list()).expect("the list is filed"); + + let recogniser = kept(&store).expect("the list reads"); + + assert!(recognises(&recogniser, "Show Title")); + } + + #[tokio::test(start_paused = true)] + async fn a_list_read_at_the_start_is_filed_said_and_decided_against() { + let running = started(Reading::default(), Scripted::new()); + + after(NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO]); + assert_eq!(running.filed(), ["Show Title", "Another Show"]); + assert_eq!(running.said(), ["AniList list read: 2 entries"]); + assert!(recognises(&running.current.now(), "Show Title")); + } + + #[tokio::test(start_paused = true)] + async fn a_list_of_one_entry_is_said_as_one() { + let running = started( + Reading::default(), + Scripted::new().lists([Ok(vec![an_entry(900_001, "Show Title")])]), + ); + + after(NOW).await; + + assert_eq!(running.said(), ["AniList list read: 1 entry"]); + } + + #[tokio::test(start_paused = true)] + async fn a_list_is_not_read_at_the_start_where_the_settings_say_so() { + let running = started( + reading(Period::Every(HOUR), false, Period::Off), + Scripted::new(), + ); + + after(MINUTE * 59).await; + assert!(running.reads().is_empty(), "{:?}", running.reads()); + after(MINUTE + NOW).await; + + assert_eq!(running.reads(), [HOUR]); + } + + #[tokio::test(start_paused = true)] + async fn a_list_is_read_again_at_each_refresh() { + let running = started(Reading::default(), Scripted::new()); + + after(HOUR * 2 + NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO, HOUR, HOUR * 2]); + } + + #[tokio::test(start_paused = true)] + async fn a_list_read_once_is_not_read_again_where_refresh_is_off() { + let running = started(reading(Period::Off, true, Period::Off), Scripted::new()); + + after(HOUR * 5).await; + + assert_eq!(running.reads(), [Duration::ZERO]); + } + + #[tokio::test(start_paused = true)] + async fn a_miss_long_after_the_last_reading_brings_one_at_once() { + let running = started(Reading::default(), Scripted::new()); + after(MINUTE * 30).await; + + running.wakes.missed(); + after(NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO, MINUTE * 30]); + } + + #[tokio::test(start_paused = true)] + async fn misses_soon_after_a_reading_bring_one_once_on_miss_has_passed() { + let running = started(Reading::default(), Scripted::new()); + after(MINUTE).await; + + running.wakes.missed(); + after(MINUTE).await; + running.wakes.missed(); + after(MINUTE * 4).await; + + assert_eq!(running.reads(), [Duration::ZERO, MINUTE * 5]); + } + + #[tokio::test(start_paused = true)] + async fn a_miss_brings_nothing_where_on_miss_is_off() { + let running = started( + reading(Period::Every(HOUR), true, Period::Off), + Scripted::new(), + ); + after(MINUTE * 30).await; + + running.wakes.missed(); + after(MINUTE * 10).await; + + assert_eq!(running.reads(), [Duration::ZERO]); + } + + /// A reading that fails as a service that is down fails. + fn down() -> Result, benshi_sync::Error> { + Err(benshi_sync::Error::Unreachable("connection refused".into())) + } + + #[tokio::test(start_paused = true)] + async fn a_failed_reading_is_said_once_and_tried_again_after_a_growing_wait() { + let running = started( + Reading::default(), + Scripted::new().lists((0..8).map(|_| down())), + ); + + after(HOUR * 3 + MINUTE * 3 + NOW).await; + + let minutes: Vec = running.reads().iter().map(|at| at.as_secs() / 60).collect(); + assert_eq!(minutes, [0, 1, 3, 7, 15, 31, 63, 123, 183]); + assert_eq!( + running.said(), + [ + "AniList list could not be read: could not be reached: connection refused; \ + tried again in 1m", + "AniList list read: 2 entries", + ] + ); + assert_eq!(FIRST_RETRY, MINUTE); + assert_eq!(LAST_RETRY, HOUR); + } + + #[tokio::test(start_paused = true)] + async fn a_reading_that_goes_through_starts_the_wait_again() { + let running = started( + Reading::default(), + Scripted::new().lists([down(), down(), Ok(the_list()), down()]), + ); + + after(HOUR + MINUTE * 4 + NOW).await; + + let minutes: Vec = running.reads().iter().map(|at| at.as_secs() / 60).collect(); + assert_eq!(minutes, [0, 1, 3, 63, 64]); + assert_eq!( + running.said(), + [ + "AniList list could not be read: could not be reached: connection refused; \ + tried again in 1m", + "AniList list read: 2 entries", + "AniList list could not be read: could not be reached: connection refused; \ + tried again in 1m", + "AniList list read: 2 entries", + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_request_to_wait_is_waited_out() { + let running = started( + Reading::default(), + Scripted::new().lists([Err(benshi_sync::Error::Wait { + after: Duration::from_secs(90), + })]), + ); + + after(Duration::from_secs(90) + NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO, Duration::from_secs(90)]); + assert_eq!( + running.said()[0], + "AniList list could not be read: asked to wait 90s; tried again in 1m 30s" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_refused_token_is_said_and_not_read_with_again_until_a_login() { + let running = started( + Reading::default(), + Scripted::new().lists([Err(benshi_sync::Error::Refused("Invalid token".to_owned()))]), + ); + after(MINUTE).await; + running.wakes.missed(); + + after(HOUR * 3).await; + assert_eq!(running.reads(), [Duration::ZERO]); + running.wakes.logged_in(); + after(NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO, HOUR * 3 + MINUTE]); + assert_eq!( + running.said(), + [ + "AniList refused the token: Invalid token; benshi login gives another", + "AniList list read: 2 entries", + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_token_refused_again_after_a_login_is_said_again() { + let refused = || Err(benshi_sync::Error::Refused("Invalid token".to_owned())); + let running = started( + Reading::default(), + Scripted::new().lists([refused(), refused()]), + ); + after(MINUTE / 2).await; + + running.wakes.logged_in(); + after(NOW).await; + + assert_eq!( + running.said(), + [ + "AniList refused the token: Invalid token; benshi login gives another", + "AniList refused the token: Invalid token; benshi login gives another", + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn no_token_is_said_once_and_looked_for_again_at_each_refresh() { + let running = started( + Reading::default(), + Scripted::new().tokens([Ok(None), Ok(None), Ok(None)]), + ); + after(MINUTE).await; + running.wakes.missed(); + + after(HOUR * 2 + NOW).await; + assert!(running.reads().is_empty(), "{:?}", running.reads()); + after(HOUR).await; + + assert_eq!(running.reads(), [HOUR * 3]); + assert_eq!( + running.said(), + [ + "no AniList token, so no list is read: benshi login keeps one", + "AniList list read: 2 entries", + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_login_brings_a_reading_at_once() { + let running = started(Reading::default(), Scripted::new().tokens([Ok(None)])); + after(MINUTE * 10).await; + + running.wakes.logged_in(); + after(NOW).await; + + assert_eq!(running.reads(), [MINUTE * 10]); + } + + #[tokio::test(start_paused = true)] + async fn a_token_the_store_cannot_give_is_said_and_read_as_none() { + let running = started( + Reading::default(), + Scripted::new().tokens([Err(secrets::Error::Locked)]), + ); + + after(HOUR + NOW).await; + + assert_eq!(running.reads(), [HOUR]); + assert_eq!( + running.said(), + [ + "AniList token could not be read: secret store is locked", + "AniList list read: 2 entries", + ] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_list_the_store_cannot_keep_is_said_and_tried_again() { + let running = started_refusing(Reading::default(), Scripted::new()); + + after(MINUTE + NOW).await; + + assert_eq!(running.reads(), [Duration::ZERO, MINUTE]); + let said = running.said(); + assert_eq!(said.len(), 1, "{said:?}"); + assert!( + said[0].starts_with("AniList list could not be kept: ") + && said[0].contains("refused by the test"), + "{said:?}" + ); + } + + #[tokio::test(start_paused = true)] + async fn a_miss_while_a_reading_fails_brings_nothing_once_one_goes_through() { + let running = started(Reading::default(), Scripted::new().lists([down()])); + after(MINUTE / 2).await; + + running.wakes.missed(); + after(MINUTE * 10).await; + + assert_eq!(running.reads(), [Duration::ZERO, MINUTE]); + } + + #[tokio::test(start_paused = true)] + async fn misses_bring_nothing_while_readings_fail() { + let running = started( + reading(Period::Every(HOUR), true, Period::Every(Duration::ZERO)), + Scripted::new().lists([down(), down(), down()]), + ); + after(MINUTE + NOW).await; + + running.wakes.missed(); + after(MINUTE).await; + + assert_eq!(running.reads(), [Duration::ZERO, MINUTE]); + } +} diff --git a/crates/daemon/src/recognition.rs b/crates/daemon/src/recognition.rs index 8f7b857..a2f1cce 100644 --- a/crates/daemon/src/recognition.rs +++ b/crates/daemon/src/recognition.rs @@ -11,7 +11,7 @@ //! same reason: an explanation is of what the daemon acted on, and a second //! look would be free to disagree with it. -use std::sync::{Arc, RwLock}; +use std::sync::{Arc, PoisonError, RwLock}; use benshi_core::path::RawPath; use benshi_core::recognise::altname::Altnames; @@ -68,6 +68,38 @@ impl Recogniser { } } +/// The recogniser the rounds decide against, which the list's task replaces +/// when it has read a list. +/// +/// A round takes the one in place as it begins and decides with it to its +/// end, so a round that began before a replacement finishes on the list it +/// began with. Cloning shares the place. +#[derive(Debug, Clone)] +pub struct Current(Arc>>); + +impl Current { + /// A place holding `recogniser`. + #[must_use] + pub fn new(recogniser: Recogniser) -> Self { + Self(Arc::new(RwLock::new(Arc::new(recogniser)))) + } + + /// The recogniser in place. + /// + /// Read through a lock a panic left poisoned: the recogniser is made + /// before the lock is taken and put in whole, so no panic leaves half of + /// one in place. + #[must_use] + pub fn now(&self) -> Arc { + Arc::clone(&self.0.read().unwrap_or_else(PoisonError::into_inner)) + } + + /// Put `recogniser` in place of the one there. + pub fn replace(&self, recogniser: Recogniser) { + *self.0.write().unwrap_or_else(PoisonError::into_inner) = Arc::new(recogniser); + } +} + /// One decision: whose file, which file, what its name spelled, and the answer. #[derive(Debug, Clone, PartialEq)] pub struct Decision { @@ -119,7 +151,7 @@ impl Decided { #[cfg(test)] mod tests { - use super::{Decided, Decision, Recogniser}; + use super::{Current, Decided, Decision, Recogniser}; use benshi_core::path::RawPath; use benshi_core::recognise::altname::Altnames; use benshi_core::recognise::corpus::Corpus; @@ -263,6 +295,34 @@ mod tests { ); } + #[test] + fn a_recogniser_put_in_place_is_the_one_taken_after() { + let current = Current::new(Recogniser::empty()); + let shared = current.clone(); + + shared.replace(Recogniser::of(a_list_of(&["Show Title"]), Altnames::new())); + let (_parsed, answer) = current.now().decide(&a_path("[Group] Show Title - 03.mkv")); + + assert!( + matches!(answer, Recognition::Recognised(_)), + "got {answer:?}" + ); + } + + #[test] + fn a_recogniser_taken_before_a_replacement_decides_as_it_did() { + let current = Current::new(Recogniser::empty()); + let taken = current.now(); + + current.replace(Recogniser::of(a_list_of(&["Show Title"]), Altnames::new())); + let (_parsed, answer) = taken.decide(&a_path("[Group] Show Title - 03.mkv")); + + assert!( + matches!(answer, Recognition::Unrecognised(_)), + "got {answer:?}" + ); + } + #[test] fn a_record_a_panic_left_poisoned_is_not_read() { let decided = Decided::new(); diff --git a/crates/daemon/tests/replay_is_faithful.rs b/crates/daemon/tests/replay_is_faithful.rs index 56b3767..ab5afb9 100644 --- a/crates/daemon/tests/replay_is_faithful.rs +++ b/crates/daemon/tests/replay_is_faithful.rs @@ -21,7 +21,8 @@ use benshi_core::trace::Trace; use benshi_core::{Known, PlayState, PlayerId, PlayerSnapshot}; use benshi_daemon::bus::{BusEvent, EventBus}; use benshi_daemon::detection::{Detection, Seen, Wiring}; -use benshi_daemon::recognition::{Decided, Recogniser}; +use benshi_daemon::listing::Wakes; +use benshi_daemon::recognition::{Current, Decided, Recogniser}; use benshi_detect::replay::ReplayWatcher; use benshi_store::Store; use tokio::time::timeout; @@ -70,7 +71,8 @@ async fn replayed(trace: Trace) -> Vec { bus: Arc::clone(&bus), policy: Arc::new(RwLock::new(PolicyTable::allowing_video_players())), seen: Seen::new(), - recogniser: Arc::new(Recogniser::empty()), + recogniser: Current::new(Recogniser::empty()), + wakes: Wakes::new(), decided: Decided::new(), store: Arc::new(Mutex::new(store)), now: SystemTime::now, diff --git a/crates/daemon/tests/the_daemon_runs.rs b/crates/daemon/tests/the_daemon_runs.rs index 236305d..9c37ba7 100644 --- a/crates/daemon/tests/the_daemon_runs.rs +++ b/crates/daemon/tests/the_daemon_runs.rs @@ -17,12 +17,14 @@ // nothing to await, rather than being spelled out as the future it desugars to. #![allow(clippy::unused_async_trait_impl)] +use std::future::{self, Future}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use benshi_core::clock::{Clock, TestClock, Timestamp}; +use benshi_core::list::{Entry, EntryId, Status}; use benshi_core::path::RawPath; use benshi_core::policy::Policy; use benshi_core::recognise::altname::Altnames; @@ -33,9 +35,12 @@ use benshi_core::{ }; use benshi_daemon::accounts::{Accounts, Done, LoggedIn}; use benshi_daemon::bus::BusEvent; +use benshi_daemon::config::{Period, Reading}; use benshi_daemon::ipc::bind; +use benshi_daemon::listing::{Listing, Source}; use benshi_daemon::protocol::{Request, Response, SourceListing}; use benshi_daemon::recognition::Recogniser; +use benshi_daemon::secrets; use benshi_daemon::supervisor::{BACKOFF_BASE, Notice, RESTART_LIMIT, Stopped, TaskRecord}; use benshi_detect::{PlayerWatcher, PollOutcome, SourceInfo, WatchError}; use benshi_store::Store; @@ -303,8 +308,37 @@ impl Accounts for NoLogins { } } -/// A daemon over this platform with nothing to match a name against, as the -/// binary starts one. +/// A list that answers with these entries, with a token always kept. +struct Answering(Vec); + +impl Source for Answering { + fn token(&self) -> impl Future, secrets::Error>> + Send { + future::ready(Ok(Some(Token::new("a token")))) + } + + fn list( + &self, + _token: &Token, + ) -> impl Future, benshi_sync::Error>> + Send { + future::ready(Ok(self.0.clone())) + } +} + +/// A list's task that never reads: no token, and no time to read at. +fn reading_nothing() -> Listing { + Listing { + source: Answering(Vec::new()), + reading: Reading { + refresh: Period::Off, + at_start: false, + on_miss: Period::Off, + }, + say: Arc::new(|_line: &str| {}), + } +} + +/// A daemon over this platform with nothing to match a name against and a +/// list's task that never reads. fn a_daemon( platform: &Platform, home: &TempDir, @@ -320,6 +354,7 @@ fn a_daemon( SystemTime::now, PERIOD, Arc::new(NoLogins), + reading_nothing(), |_notice| {}, )) } @@ -353,6 +388,7 @@ fn a_daemon_that_tells( the_time, PERIOD, Arc::new(NoLogins), + reading_nothing(), told.listener(), )) } @@ -482,6 +518,152 @@ async fn a_running_daemon_records_what_its_detection_watched() { assert_eq!(seen_at, THE_TIME); } +/// The one entry of the list the platform's file is of. +fn the_entry() -> Entry { + Entry { + id: EntryId(900_001), + title: "Show".to_owned(), + spellings: Vec::new(), + format: None, + episodes: Some(12), + episode_length: None, + year: None, + released: true, + status: Status::Watching, + progress: 2, + rewatches: 0, + score: None, + } +} + +/// Logins that go through. +#[derive(Debug)] +struct Taking; + +impl Accounts for Taking { + fn login(&self, _token: Token) -> Done<'_, LoggedIn> { + Box::pin(future::ready(Ok(LoggedIn { + name: "someone".to_owned(), + expires: None, + }))) + } + + fn logout(&self) -> Done<'_, ()> { + Box::pin(future::ready(Ok(()))) + } +} + +/// A daemon over this platform with nothing to match a name against and an +/// empty store, whose list's task reads [`the_entry`] at the times `reading` +/// names, and whose logins go through. +fn a_daemon_reading( + platform: &Platform, + store: &Arc>, + socket: &Path, + reading: Reading, +) -> tokio::task::JoinHandle> { + let listener = Arc::new(bind(socket).expect("the socket binds")); + + tokio::spawn(benshi_daemon::run( + platform.factory(), + listener, + Recogniser::empty(), + Arc::clone(store), + the_time, + PERIOD, + Arc::new(Taking), + Listing { + source: Answering(vec![the_entry()]), + reading, + say: Arc::new(|_line: &str| {}), + }, + |_notice| {}, + )) +} + +/// An empty store, as a daemon makes one in a directory of its own. +fn an_empty_store(home: &TempDir) -> Arc> { + Arc::new(Mutex::new( + Store::open(&the_database(home)).expect("the database opens"), + )) +} + +/// The one operation a daemon queues once it has recorded the platform's +/// file as an episode of [`the_entry`]. +fn the_episode_queued() -> [Operation; 1] { + [Operation { + id: 1, + show: THE_SHOW, + title: "Show".to_owned(), + kind: Kind::Progress { episode: Some(3) }, + }] +} + +#[tokio::test] +async fn a_running_daemon_records_what_the_list_it_read_holds() { + // A daemon that starts with nothing to match a name against, and an + // empty store: the file it plays is recognised only by a list its own + // task read, filed and put in place for detection. + let (home, socket) = a_socket(); + let store = an_empty_store(&home); + let platform = Platform::default(); + + let running = a_daemon_reading(&platform, &store, &socket, Reading::default()); + + let queued = once_queued(&store, "the daemon recorded an episode of its list").await; + running.abort(); + + assert_eq!(queued, the_episode_queued()); +} + +#[tokio::test] +async fn a_running_daemon_reads_its_list_where_a_name_misses_it() { + // Read at no time but on a miss: the file the platform plays misses + // the empty list, the round wakes the list's task, and what it reads + // recognises the file. + let (home, socket) = a_socket(); + let store = an_empty_store(&home); + let platform = Platform::default(); + let reading = Reading { + refresh: Period::Off, + at_start: false, + on_miss: Period::Every(Duration::ZERO), + }; + + let running = a_daemon_reading(&platform, &store, &socket, reading); + + let queued = once_queued(&store, "a miss brought a reading of the list").await; + running.abort(); + + assert_eq!(queued, the_episode_queued()); +} + +#[tokio::test] +async fn a_running_daemon_reads_its_list_once_a_login_goes_through() { + let (home, socket) = a_socket(); + let store = an_empty_store(&home); + let platform = Platform::default(); + let reading = Reading { + refresh: Period::Off, + at_start: false, + on_miss: Period::Off, + }; + let running = a_daemon_reading(&platform, &store, &socket, reading); + + let taken = ask( + &socket, + &Request::Login { + token: Token::new("a token"), + }, + ) + .await; + assert_eq!(taken, Response::Ok); + let queued = once_queued(&store, "a login brought a reading of the list").await; + running.abort(); + + assert_eq!(queued, the_episode_queued()); +} + #[tokio::test] async fn a_running_daemon_explains_what_its_own_detection_decided() { // The seam of the listing again, for the other record a round leaves. diff --git a/crates/daemon/tests/the_session_bus_is_found.rs b/crates/daemon/tests/the_session_bus_is_found.rs index 111a6cf..54b1c5e 100644 --- a/crates/daemon/tests/the_session_bus_is_found.rs +++ b/crates/daemon/tests/the_session_bus_is_found.rs @@ -20,6 +20,7 @@ use std::time::Duration; use benshi_core::token::Token; use benshi_daemon::accounts::{self, Accounts, OnAniList}; +use benshi_daemon::listing::{FromAniList, Source}; use benshi_daemon::secrets::{Asking, Error, SecretService}; /// Set in a run a test started, and in no other. @@ -67,21 +68,28 @@ fn connected() -> Result { answered(SecretService::connect(SHORT, Asking::Never)) } -/// Logins checked with a service at a port of this machine that nobody -/// listens on, held by the socket handed back, so that no test here -/// reaches AniList. -fn accounts() -> (OnAniList, tokio::net::TcpSocket) { +/// The address of a service at a port of this machine that nobody listens +/// on, held by the socket handed back, so that no test here reaches +/// AniList. +fn nowhere() -> (String, tokio::net::TcpSocket) { let start = u16::try_from(std::process::id() % 20_000).expect("below 20000"); for offset in 0..20_000 { let port = 20_000 + (start + offset) % 20_000; let socket = tokio::net::TcpSocket::new_v4().expect("a socket"); if socket.bind(([127, 0, 0, 1], port).into()).is_ok() { - return (OnAniList::new(&format!("http://127.0.0.1:{port}/")), socket); + return (format!("http://127.0.0.1:{port}/"), socket); } } panic!("no port between 20000 and 40000 is free"); } +/// Logins checked with [`nowhere`]. +fn accounts() -> (OnAniList, tokio::net::TcpSocket) { + let (address, held) = nowhere(); + + (OnAniList::new(&address), held) +} + #[test] fn a_session_with_no_bus_has_no_store_to_reach() { if env::var_os(AGAIN).is_some() { @@ -167,3 +175,52 @@ fn a_logout_in_a_session_with_no_bus_says_the_token_was_not_removed() { "unix:path=/nowhere/bus", ); } + +#[test] +fn a_token_given_from_the_environment_is_read_without_asking_the_store() { + if env::var_os(AGAIN).is_some() { + let (address, _held) = nowhere(); + let source = FromAniList::new(&address, Some(Token::new("given"))); + + let token = answered(source.token()).expect("a token"); + + assert_eq!(token.as_ref().map(Token::secret), Some("given")); + return; + } + + again( + "a_token_given_from_the_environment_is_read_without_asking_the_store", + "unix:path=/nowhere/bus", + ); +} + +#[test] +fn a_token_with_none_given_is_asked_of_the_store() { + if env::var_os(AGAIN).is_some() { + let (address, _held) = nowhere(); + let source = FromAniList::new(&address, None); + + let token = answered(source.token()); + + assert!(matches!(&token, Err(Error::Unreachable(_))), "{token:?}"); + return; + } + + again( + "a_token_with_none_given_is_asked_of_the_store", + "unix:path=/nowhere/bus", + ); +} + +#[test] +fn a_list_is_read_from_the_address_given() { + let (address, _held) = nowhere(); + let source = FromAniList::new(&address, None); + + let list = answered(source.list(&Token::new("a token"))); + + assert!( + matches!(&list, Err(benshi_sync::Error::Unreachable(_))), + "{list:?}" + ); +}