The directory's lib.rs splits into leaf modules (#356)
The crate root held the wire types, Directory with its refresher and reloader, and every /v1 handler. It is now the module list and the re-exports ipx and the directory's main use, so their paths are as before: wire.rs Listing, FeedPage, SnapshotEntry, ... and MAX_SNAPSHOT directory.rs Directory, Shared, the refresher and the reloader api.rs the router, the token check and the /v1 handlers Directory's fields and methods are pub(crate) for the handlers beside it. Its two tests that read those fields went with it, on a scratch Directory of their own; the API tests stay with the API. Every definition and doc comment line of the old lib.rs is in exactly one of the new files. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
218
crates/directory/src/directory.rs
Normal file
218
crates/directory/src/directory.rs
Normal file
@@ -0,0 +1,218 @@
|
||||
//! The running directory: its store, the tokens it accepts, the refresher that fetches its feeds, and
|
||||
//! the reloader that picks up a changed config.
|
||||
|
||||
use crate::config::Token;
|
||||
use crate::wire::{Listing, Me};
|
||||
use crate::{config, store};
|
||||
use reqwest::redirect::Policy;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::time::{Duration, Instant};
|
||||
use store::{Refreshed, Store};
|
||||
use tokio::sync::Notify;
|
||||
use tracing::Instrument;
|
||||
|
||||
/// How often each feed is fetched again. Metadata changes slowly, and every server subscribed
|
||||
/// to a feed fetches it itself anyway: every six hours was 3.9 GB of feeds a day (#243).
|
||||
pub const REFRESH_EVERY: i64 = 12 * 3600;
|
||||
|
||||
/// The running directory: its database, the tokens it accepts and the client its refresher
|
||||
/// fetches feeds with.
|
||||
pub struct Directory {
|
||||
pub store: Store,
|
||||
pub(crate) tokens: RwLock<Vec<Token>>,
|
||||
pub(crate) client: reqwest::Client,
|
||||
/// Wakes the refresher, so a feed a snapshot brings in is fetched now, not in a minute.
|
||||
pub(crate) wake: Notify,
|
||||
/// The public pages' copy of the list, and when it was read.
|
||||
pub(crate) pages_list: tokio::sync::Mutex<Option<(Instant, Arc<Vec<Listing>>)>>,
|
||||
}
|
||||
|
||||
/// How long the public pages show the same copy of the list.
|
||||
pub(crate) const PAGES_LIST_FOR: Duration = Duration::from_secs(60);
|
||||
|
||||
/// The directory as every handler and task holds it.
|
||||
pub type Shared = Arc<Directory>;
|
||||
|
||||
impl Directory {
|
||||
/// A directory on `store`, accepting `tokens`, whose refresher may reach the internal
|
||||
/// addresses `fetch_internal` lists and no others.
|
||||
pub fn new(
|
||||
store: Store,
|
||||
tokens: Vec<Token>,
|
||||
fetch_internal: ipx_feed::guard::Allowed,
|
||||
) -> Result<Shared, reqwest::Error> {
|
||||
// No redirects: ipx_feed::fetch follows them itself, so it sees each hop.
|
||||
let client = reqwest::Client::builder()
|
||||
.connect_timeout(Duration::from_secs(10))
|
||||
.read_timeout(ipx_feed::FEED_SILENCE)
|
||||
.redirect(Policy::none())
|
||||
.user_agent(concat!("ipx-directory/", env!("CARGO_PKG_VERSION")))
|
||||
// The feeds are named by whoever holds a write token, and redirect where their
|
||||
// publishers like: nothing on this network is fetched unless listed (#329).
|
||||
.connector_layer(ipx_feed::guard::Guard::new(fetch_internal))
|
||||
.build()?;
|
||||
Ok(Arc::new(Self {
|
||||
store,
|
||||
tokens: RwLock::new(tokens),
|
||||
client,
|
||||
wake: Notify::new(),
|
||||
pages_list: Default::default(),
|
||||
}))
|
||||
}
|
||||
|
||||
/// The whole list for the public pages, read from the database at most once a minute. Anyone
|
||||
/// may ask for a page, and each asked the database for the whole catalogue on the pool the
|
||||
/// refresher and every server's /v1 calls share; a query string of its own got past any
|
||||
/// cache in front (#341). The lock is held through the read, so a burst waits on one query.
|
||||
pub async fn pages_list(&self) -> anyhow::Result<Arc<Vec<Listing>>> {
|
||||
let mut held = self.pages_list.lock().await;
|
||||
if let Some((at, list)) = held.as_ref()
|
||||
&& at.elapsed() < PAGES_LIST_FOR
|
||||
{
|
||||
return Ok(list.clone());
|
||||
}
|
||||
let list = Arc::new(self.store.list().await?);
|
||||
*held = Some((Instant::now(), list.clone()));
|
||||
Ok(list)
|
||||
}
|
||||
|
||||
/// Takes the tokens of a config file read again, and drops the snapshots of any that went.
|
||||
pub async fn set_tokens(&self, tokens: Vec<Token>) -> anyhow::Result<u64> {
|
||||
let names: Vec<String> = tokens.iter().map(|t| t.name.clone()).collect();
|
||||
*self.tokens.write().expect("a panic elsewhere poisoned this lock") = tokens;
|
||||
self.store.keep_instances(&names).await
|
||||
}
|
||||
|
||||
pub(crate) fn caller(&self, presented: &str) -> Option<Me> {
|
||||
let h = config::hash(presented);
|
||||
// Every token compared, whether or not one matched, so timing says nothing either way.
|
||||
self.tokens.read().expect("a panic elsewhere poisoned this lock").iter().fold(None, |found, t| {
|
||||
if config::same_secret(&t.sha256, &h) { Some(Me { name: t.name.clone(), access: t.access }) } else { found }
|
||||
})
|
||||
}
|
||||
|
||||
/// Fetches one feed and stores what it found.
|
||||
pub async fn refresh(&self, due: &store::Due) -> anyhow::Result<()> {
|
||||
let got = match ipx_feed::fetch(&self.client, &due.url, None, due.etag.as_deref(), due.last_modified.as_deref())
|
||||
.await
|
||||
{
|
||||
Ok((ipx_feed::Fetched::NotModified, _)) => Refreshed::NotModified,
|
||||
Ok((ipx_feed::Fetched::Body { bytes, etag, last_modified }, _)) => {
|
||||
if ipx_feed::is_opml(&bytes) {
|
||||
// ipx sends the feeds inside an OPML, never the OPML.
|
||||
Refreshed::Failed("an OPML list, not a feed".into())
|
||||
} else {
|
||||
match ipx_feed::parse(&bytes) {
|
||||
Ok(parsed) => Refreshed::Feed { parsed, etag, last_modified },
|
||||
Err(e) => Refreshed::Failed(format!("{e:#}")),
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => Refreshed::Failed(format!("{e:#}")),
|
||||
};
|
||||
if let Refreshed::Failed(e) = &got {
|
||||
tracing::info!(feed = %due.id, error = %ipx_feed::redact_text(e), "feed check failed");
|
||||
}
|
||||
// ponytail: a feed that moved for good is still asked for at its old address, which
|
||||
// redirects; follow the move when a feed's id can change without losing its counts.
|
||||
self.store.save(&due.id, got).await
|
||||
}
|
||||
}
|
||||
|
||||
/// Fetches whatever is due, a few at a time, forever.
|
||||
pub async fn refresher(dir: Shared) {
|
||||
loop {
|
||||
match dir.store.due(REFRESH_EVERY, 25).await {
|
||||
Ok(due) => {
|
||||
if !due.is_empty() {
|
||||
// A round is a trace, each feed's check a span in it (#213).
|
||||
let round = tracing::info_span!("refresh", feeds = due.len());
|
||||
Instrument::instrument(
|
||||
async {
|
||||
// A few at a time: in turn, two publishers timing out held a round
|
||||
// for a minute and more (#227).
|
||||
futures_util::StreamExt::for_each_concurrent(futures_util::stream::iter(&due), 5, |d| {
|
||||
let check = tracing::info_span!("feed", feed = %d.id);
|
||||
async {
|
||||
if let Err(e) = Instrument::instrument(dir.refresh(d), check).await {
|
||||
tracing::warn!(feed = %d.id, error = %format!("{e:#}"), "could not store a feed check");
|
||||
}
|
||||
}
|
||||
})
|
||||
.await;
|
||||
// One line a round, for the dashboard's refresher panel (#210); failures have their own.
|
||||
tracing::info!(checked = due.len(), "feeds checked");
|
||||
},
|
||||
round,
|
||||
)
|
||||
.await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
Err(e) => tracing::warn!(error = %format!("{e:#}"), "could not read which feeds are due"),
|
||||
}
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(Duration::from_secs(60)) => {}
|
||||
_ = dir.wake.notified() => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads the config file again whenever it changes, checked each minute. A file that fails to
|
||||
/// load leaves the tokens as they were, so a typo locks nobody out.
|
||||
pub async fn reloader(dir: Shared, path: PathBuf) {
|
||||
let mtime = |p: &std::path::Path| std::fs::metadata(p).and_then(|m| m.modified()).ok();
|
||||
let mut last = mtime(&path);
|
||||
loop {
|
||||
tokio::time::sleep(Duration::from_secs(60)).await;
|
||||
let now = mtime(&path);
|
||||
if now == last {
|
||||
continue;
|
||||
}
|
||||
last = now;
|
||||
match config::Config::load(&path).and_then(|c| c.tokens()) {
|
||||
Ok(tokens) => {
|
||||
let n = tokens.len();
|
||||
match dir.set_tokens(tokens).await {
|
||||
Ok(dropped) => tracing::info!(tokens = n, dropped, "config read again"),
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "config read again, but old snapshots could not be dropped")
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(error = %format!("{e:#}"), "config changed but does not load; keeping the tokens as they were")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
async fn scratch() -> Shared {
|
||||
Directory::new(Store::scratch().await, vec![], Default::default()).unwrap()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_refresher_fetches_nothing_on_this_network() {
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(async move { while listener.accept().await.is_ok() {} });
|
||||
let dir = scratch().await;
|
||||
let err = dir.client.get(format!("http://{addr}/feed.xml")).send().await.unwrap_err();
|
||||
assert!(format!("{err:?}").contains("internal address"), "{err:?}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_public_pages_read_the_list_once_a_minute() {
|
||||
let dir = scratch().await;
|
||||
let first = dir.pages_list().await.unwrap();
|
||||
assert!(Arc::ptr_eq(&first, &dir.pages_list().await.unwrap()), "a second page asks the database nothing");
|
||||
// Once the copy is a minute old, the next page reads the list again.
|
||||
dir.pages_list.lock().await.as_mut().unwrap().0 -= PAGES_LIST_FOR;
|
||||
assert!(!Arc::ptr_eq(&first, &dir.pages_list().await.unwrap()));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user