Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4baf81e2bf |
+241
@@ -0,0 +1,241 @@
|
||||
//! Per-upstream request budget — a shared token bucket.
|
||||
//!
|
||||
//! Lifted from the card feeder's `tmdb_budget.rs` (its `TmdbBudget` half; the
|
||||
//! per-indexer registry has no counterpart here) and renamed, because the
|
||||
//! mechanism is upstream-agnostic and the music feeder needs three independent
|
||||
//! instances rather than one global.
|
||||
//!
|
||||
//! **Why three, not one.** MusicBrainz's 1 req/s is a hard published ceiling
|
||||
//! enforced per source IP; ListenBrainz and the Internet Archive are far more
|
||||
//! generous. A single shared bucket would either throttle all three to 1/s or
|
||||
//! let MusicBrainz calls burst past a limit that gets the whole box blocked.
|
||||
//!
|
||||
//! Design notes, unchanged from the original:
|
||||
//! - **Lazy refill.** Tokens accrue on each `acquire` from elapsed wall-clock,
|
||||
//! so there is no background timer task and no idle cost.
|
||||
//! - **Cancel-safe.** The token is decremented synchronously under the lock
|
||||
//! right before `acquire` returns [`Lease::Granted`]; there is no `.await`
|
||||
//! between the decision to grant and the return, so a dropped `acquire`
|
||||
//! future (consumer disconnected) never spends a token.
|
||||
//! - **Global pause on throttle.** One upstream 429/503 freezes *all* grants
|
||||
//! on that bucket until the Retry-After window elapses, via
|
||||
//! [`RateBudget::note_throttled`].
|
||||
//!
|
||||
//! The critical sections hold a plain `std::sync::Mutex` (no awaits inside),
|
||||
//! while the waiting happens outside the lock on a `tokio::time::sleep` raced
|
||||
//! against a `tokio::sync::Notify`.
|
||||
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use tokio::sync::Notify;
|
||||
|
||||
/// Outcome of a [`RateBudget::acquire`] call.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Lease {
|
||||
/// A token was granted; the caller may make one upstream call.
|
||||
Granted,
|
||||
/// The wait deadline elapsed before a token became available. The caller
|
||||
/// degrades best-effort — it does **not** call the upstream anyway.
|
||||
DeadlineExceeded,
|
||||
}
|
||||
|
||||
struct BucketState {
|
||||
tokens: f64,
|
||||
capacity: f64,
|
||||
refill_per_sec: f64,
|
||||
last_refill: Instant,
|
||||
/// When set and in the future, all grants are frozen.
|
||||
paused_until: Option<Instant>,
|
||||
}
|
||||
|
||||
impl BucketState {
|
||||
/// Accrue tokens for elapsed time since the last refill. Called under the
|
||||
/// lock before inspecting `tokens`.
|
||||
fn refill(&mut self, now: Instant) {
|
||||
let elapsed = now.saturating_duration_since(self.last_refill).as_secs_f64();
|
||||
if elapsed > 0.0 {
|
||||
self.tokens = (self.tokens + elapsed * self.refill_per_sec).min(self.capacity);
|
||||
self.last_refill = now;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Shared token bucket gating one upstream's API calls across the process.
|
||||
pub struct RateBudget {
|
||||
state: Mutex<BucketState>,
|
||||
/// Woken whenever a grant *might* now be possible (a token accrued, or a
|
||||
/// pause was cleared). Waiters re-check under the lock after waking; a
|
||||
/// missed wake only costs an extra bounded sleep.
|
||||
wake: Notify,
|
||||
}
|
||||
|
||||
impl RateBudget {
|
||||
/// Build a budget sustaining `refill_per_sec` grants/second with a burst
|
||||
/// ceiling of `capacity` tokens. The bucket starts full, so the first
|
||||
/// burst is served immediately.
|
||||
pub fn new(refill_per_sec: f64, capacity: f64) -> Arc<Self> {
|
||||
Arc::new(Self {
|
||||
state: Mutex::new(BucketState {
|
||||
tokens: capacity,
|
||||
capacity,
|
||||
refill_per_sec,
|
||||
last_refill: Instant::now(),
|
||||
paused_until: None,
|
||||
}),
|
||||
wake: Notify::new(),
|
||||
})
|
||||
}
|
||||
|
||||
/// Wait for one token, giving up after `deadline`. Cancel-safe: dropping
|
||||
/// the returned future before it resolves [`Lease::Granted`] spends no
|
||||
/// token.
|
||||
pub async fn acquire(&self, deadline: Duration) -> Lease {
|
||||
let give_up_at = Instant::now() + deadline;
|
||||
loop {
|
||||
// Decide-or-compute-wait, all under the lock (no awaits inside).
|
||||
let sleep_for = {
|
||||
let mut st = self.state.lock().expect("rate budget mutex poisoned");
|
||||
let now = Instant::now();
|
||||
if let Some(until) = st.paused_until {
|
||||
if now >= until {
|
||||
st.paused_until = None;
|
||||
}
|
||||
}
|
||||
let paused = st.paused_until;
|
||||
if paused.is_none() {
|
||||
st.refill(now);
|
||||
if st.tokens >= 1.0 {
|
||||
// Atomic grant: decrement and return with no await between.
|
||||
st.tokens -= 1.0;
|
||||
return Lease::Granted;
|
||||
}
|
||||
}
|
||||
if now >= give_up_at {
|
||||
return Lease::DeadlineExceeded;
|
||||
}
|
||||
let until_deadline = give_up_at.saturating_duration_since(now);
|
||||
let wait = match paused {
|
||||
// Frozen: wake when the pause is scheduled to lift.
|
||||
Some(until) => until.saturating_duration_since(now),
|
||||
// Throttled: wake when the next whole token should accrue.
|
||||
None => {
|
||||
let needed = 1.0 - st.tokens;
|
||||
let secs = if st.refill_per_sec > 0.0 {
|
||||
needed / st.refill_per_sec
|
||||
} else {
|
||||
until_deadline.as_secs_f64()
|
||||
};
|
||||
Duration::from_secs_f64(secs.max(0.0))
|
||||
}
|
||||
};
|
||||
wait.min(until_deadline)
|
||||
};
|
||||
tokio::select! {
|
||||
_ = tokio::time::sleep(sleep_for) => {}
|
||||
_ = self.wake.notified() => {}
|
||||
}
|
||||
if Instant::now() >= give_up_at {
|
||||
return Lease::DeadlineExceeded;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Feed back an upstream throttle response (429, or MusicBrainz's 503):
|
||||
/// pause all grants on this bucket until `now + retry_after`. Extends an
|
||||
/// existing pause but never shortens it.
|
||||
pub fn note_throttled(&self, retry_after: Duration) {
|
||||
let until = Instant::now() + retry_after;
|
||||
{
|
||||
let mut st = self.state.lock().expect("rate budget mutex poisoned");
|
||||
st.paused_until = Some(match st.paused_until {
|
||||
Some(existing) if existing > until => existing,
|
||||
_ => until,
|
||||
});
|
||||
}
|
||||
// Wake waiters so they recompute their sleep against the new pause.
|
||||
self.wake.notify_waiters();
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn grants_up_to_capacity_immediately() {
|
||||
let b = RateBudget::new(3.0, 3.0);
|
||||
for _ in 0..3 {
|
||||
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn next_token_waits_for_refill() {
|
||||
let b = RateBudget::new(2.0, 2.0); // 2/s sustained
|
||||
for _ in 0..2 {
|
||||
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
|
||||
}
|
||||
// Bucket empty; next grant must wait ~0.5 s for one token at 2/s.
|
||||
let start = Instant::now();
|
||||
assert_eq!(b.acquire(Duration::from_secs(10)).await, Lease::Granted);
|
||||
assert!(
|
||||
start.elapsed() >= Duration::from_millis(450),
|
||||
"expected to wait ~0.5s for refill, waited {:?}",
|
||||
start.elapsed()
|
||||
);
|
||||
}
|
||||
|
||||
/// The MusicBrainz shape specifically: 1/s with a burst of 2 must not let
|
||||
/// three calls through inside one second. This is the property that keeps
|
||||
/// the box off their block-list.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn musicbrainz_shape_holds_one_per_second() {
|
||||
let b = RateBudget::new(1.0, 1.0);
|
||||
let start = Instant::now();
|
||||
for _ in 0..4 {
|
||||
assert_eq!(b.acquire(Duration::from_secs(30)).await, Lease::Granted);
|
||||
}
|
||||
// 4 grants from a 2-token bucket refilling at 1/s ⇒ at least 2 s.
|
||||
assert!(
|
||||
start.elapsed() >= Duration::from_millis(1_900),
|
||||
"4 calls at 1/s (burst 2) must take ~2s, took {:?}",
|
||||
start.elapsed()
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn deadline_exceeded_when_drained() {
|
||||
let b = RateBudget::new(0.001, 1.0); // effectively no refill within the test
|
||||
assert_eq!(b.acquire(Duration::from_secs(1)).await, Lease::Granted);
|
||||
assert_eq!(
|
||||
b.acquire(Duration::from_millis(200)).await,
|
||||
Lease::DeadlineExceeded
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn note_throttled_pauses_grants_even_with_tokens_available() {
|
||||
let b = RateBudget::new(100.0, 100.0);
|
||||
b.note_throttled(Duration::from_secs(2));
|
||||
assert_eq!(
|
||||
b.acquire(Duration::from_millis(500)).await,
|
||||
Lease::DeadlineExceeded
|
||||
);
|
||||
// After the pause window, grants resume.
|
||||
assert_eq!(b.acquire(Duration::from_secs(5)).await, Lease::Granted);
|
||||
}
|
||||
|
||||
/// A pause is extended, never shortened — otherwise a late short retry
|
||||
/// would cancel an earlier long one and walk straight back into the block.
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn a_shorter_retry_after_does_not_shorten_an_existing_pause() {
|
||||
let b = RateBudget::new(100.0, 100.0);
|
||||
b.note_throttled(Duration::from_secs(10));
|
||||
b.note_throttled(Duration::from_secs(1));
|
||||
assert_eq!(
|
||||
b.acquire(Duration::from_secs(3)).await,
|
||||
Lease::DeadlineExceeded
|
||||
);
|
||||
}
|
||||
}
|
||||
+154
-1
@@ -13,7 +13,7 @@
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
|
||||
use redb::{Database, ReadableTableMetadata, TableDefinition};
|
||||
use redb::{Database, ReadableTable, ReadableTableMetadata, TableDefinition};
|
||||
|
||||
pub const CACHE_FILENAME: &str = "cache.redb";
|
||||
|
||||
@@ -53,6 +53,44 @@ const SUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("s
|
||||
/// OpenSubtitles search results keyed by `"<tmdb_id>\x01<lang3>"`. Value is the
|
||||
/// resolved subtitle's cid, or the literal `"null"` negative-cache sentinel.
|
||||
const OPENSUBTITLES_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("opensubtitles");
|
||||
|
||||
|
||||
// ── music-family tables (MusicBrainz / Internet Archive / YouTube / Jamendo) ──
|
||||
|
||||
/// Ranked MusicBrainz search results keyed by `"<entity>\x01<normalised text>"`.
|
||||
const MB_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_search");
|
||||
/// Release-group detail keyed by its MBID.
|
||||
const MB_RELEASE_GROUP_TABLE: TableDefinition<'_, &str, &str> =
|
||||
TableDefinition::new("mb_release_group");
|
||||
/// Artist detail keyed by its MBID.
|
||||
const MB_ARTIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_artist");
|
||||
/// A release-group's canonical track list keyed by the release-group MBID.
|
||||
const MB_TRACKLIST_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("mb_tracklist");
|
||||
/// Internet Archive per-item file listing keyed by the item identifier.
|
||||
const IA_ITEM_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("ia_item");
|
||||
/// An artist's YouTube channel keys, keyed by artist MBID. See
|
||||
/// `youtube::ChannelKeys`.
|
||||
///
|
||||
/// ⚠ **Only ever holds a POSITIVE result.** MusicBrainz answers `503` when
|
||||
/// crowded and the client reports every failure as `None`, so caching a
|
||||
/// negative would freeze a transport hiccup into "this artist has no channel"
|
||||
/// — the exact fabrication the match probe hit (study §7.5).
|
||||
const YT_CHANNEL_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_channel");
|
||||
/// Piped `music_songs` results keyed by the normalised search text.
|
||||
const YT_SEARCH_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("yt_search");
|
||||
/// Jamendo per-track detail keyed by the Jamendo track id. Permanent: a track
|
||||
/// id names an immutable recording, and its licence, artist credit and
|
||||
/// download URL do not change under it.
|
||||
const JAMENDO_TRACK_TABLE: TableDefinition<'_, &str, &str> = TableDefinition::new("jamendo_track");
|
||||
/// Jamendo API calls spent, keyed by UTC `YYYY-MM`.
|
||||
///
|
||||
/// ⚠ **This table is not a cache — it is an accounting ledger, and losing it
|
||||
/// loses money.** Every other table here can be deleted to reclaim disk with
|
||||
/// no consequence beyond a slower next request. Deleting this one resets the
|
||||
/// month's spend to zero while Jamendo's own counter keeps climbing, so the
|
||||
/// feeder cheerfully spends a quota it has already exhausted and every call
|
||||
/// fails for the rest of the month. See `consts::JAMENDO_MONTHLY_QUOTA`.
|
||||
const JAMENDO_QUOTA_TABLE: TableDefinition<'_, &str, u64> = TableDefinition::new("jamendo_quota");
|
||||
/// Small durable plugin scratch values that must survive a restart but aren't
|
||||
/// content-derived — today, the torznab plugin's Prowlarr indexer-set
|
||||
/// fingerprint, so a set that changed *while the feeder was down* is still
|
||||
@@ -113,6 +151,15 @@ impl MidhashCache {
|
||||
tx.open_table(SUBTITLES_TABLE)?;
|
||||
tx.open_table(OPENSUBTITLES_TABLE)?;
|
||||
tx.open_table(MISC_TABLE)?;
|
||||
tx.open_table(MB_SEARCH_TABLE)?;
|
||||
tx.open_table(MB_RELEASE_GROUP_TABLE)?;
|
||||
tx.open_table(MB_ARTIST_TABLE)?;
|
||||
tx.open_table(MB_TRACKLIST_TABLE)?;
|
||||
tx.open_table(IA_ITEM_TABLE)?;
|
||||
tx.open_table(YT_CHANNEL_TABLE)?;
|
||||
tx.open_table(YT_SEARCH_TABLE)?;
|
||||
tx.open_table(JAMENDO_TRACK_TABLE)?;
|
||||
tx.open_table(JAMENDO_QUOTA_TABLE)?;
|
||||
tx.commit()?;
|
||||
}
|
||||
Ok(Self { db: Arc::new(db) })
|
||||
@@ -120,6 +167,47 @@ impl MidhashCache {
|
||||
|
||||
str_table_accessors!(get_midhash, put_midhash, MIDHASH_TABLE);
|
||||
|
||||
// ── music-family accessors ──
|
||||
str_table_accessors!(get_mb_search, put_mb_search, MB_SEARCH_TABLE);
|
||||
str_table_accessors!(get_mb_release_group, put_mb_release_group, MB_RELEASE_GROUP_TABLE);
|
||||
str_table_accessors!(get_mb_artist, put_mb_artist, MB_ARTIST_TABLE);
|
||||
str_table_accessors!(get_mb_tracklist, put_mb_tracklist, MB_TRACKLIST_TABLE);
|
||||
str_table_accessors!(get_ia_item, put_ia_item, IA_ITEM_TABLE);
|
||||
str_table_accessors!(get_yt_channel, put_yt_channel, YT_CHANNEL_TABLE);
|
||||
str_table_accessors!(get_yt_search, put_yt_search, YT_SEARCH_TABLE);
|
||||
str_table_accessors!(get_jamendo_track, put_jamendo_track, JAMENDO_TRACK_TABLE);
|
||||
|
||||
pub fn jamendo_quota_spend(
|
||||
&self,
|
||||
month: &str,
|
||||
budget: u64,
|
||||
) -> Result<Option<u64>, redb::Error> {
|
||||
let tx = self.db.begin_write()?;
|
||||
let spent_after;
|
||||
{
|
||||
let mut table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
|
||||
let spent = table.get(month)?.map(|v| v.value()).unwrap_or(0);
|
||||
if spent >= budget {
|
||||
drop(table);
|
||||
// Nothing was written; abort rather than commit an empty txn.
|
||||
tx.abort()?;
|
||||
return Ok(None);
|
||||
}
|
||||
spent_after = spent + 1;
|
||||
table.insert(month, spent_after)?;
|
||||
}
|
||||
tx.commit()?;
|
||||
Ok(Some(spent_after))
|
||||
}
|
||||
|
||||
/// Calls already spent in `month`. Read-only — for health and the config
|
||||
/// surface, never on the request path.
|
||||
pub fn jamendo_quota_used(&self, month: &str) -> Result<u64, redb::Error> {
|
||||
let tx = self.db.begin_read()?;
|
||||
let table = tx.open_table(JAMENDO_QUOTA_TABLE)?;
|
||||
Ok(table.get(month)?.map(|v| v.value()).unwrap_or(0))
|
||||
}
|
||||
|
||||
str_table_accessors!(
|
||||
/// Read a durable plugin scratch value (see [`MISC_TABLE`]).
|
||||
get_misc,
|
||||
@@ -351,3 +439,68 @@ mod tests {
|
||||
assert_eq!(cache.entry_count().unwrap(), 2);
|
||||
}
|
||||
}
|
||||
|
||||
/// Cache key for a search: entity type plus the normalised query text.
|
||||
///
|
||||
/// Normalisation is lowercase + whitespace-collapse only. Deliberately not
|
||||
/// more aggressive: MusicBrainz treats punctuation as significant (`Re:ZERO`,
|
||||
/// `M!LK`, `!!!` are all real artist names), so stripping it would collapse
|
||||
/// distinct queries onto one cache entry and serve the wrong answer.
|
||||
pub fn search_key(entity: &str, text: &str) -> String {
|
||||
let normalised = text.split_whitespace().collect::<Vec<_>>().join(" ").to_lowercase();
|
||||
format!("{entity}\u{1}{normalised}")
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod music_table_tests {
|
||||
use super::*;
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[test]
|
||||
fn round_trips_each_table_independently() {
|
||||
let dir = tempdir().expect("tempdir");
|
||||
let c = MidhashCache::open(dir.path()).expect("open");
|
||||
|
||||
c.put_mb_search("k", "v-search").unwrap();
|
||||
c.put_mb_release_group("k", "v-rg").unwrap();
|
||||
c.put_mb_artist("k", "v-artist").unwrap();
|
||||
c.put_mb_tracklist("k", "v-tracks").unwrap();
|
||||
c.put_ia_item("k", "v-ia").unwrap();
|
||||
|
||||
// Same key in five tables must not collide.
|
||||
assert_eq!(c.get_mb_search("k").unwrap().as_deref(), Some("v-search"));
|
||||
assert_eq!(c.get_mb_release_group("k").unwrap().as_deref(), Some("v-rg"));
|
||||
assert_eq!(c.get_mb_artist("k").unwrap().as_deref(), Some("v-artist"));
|
||||
assert_eq!(c.get_mb_tracklist("k").unwrap().as_deref(), Some("v-tracks"));
|
||||
assert_eq!(c.get_ia_item("k").unwrap().as_deref(), Some("v-ia"));
|
||||
assert!(c.get_mb_search("absent").unwrap().is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn search_key_normalises_case_and_whitespace_only() {
|
||||
assert_eq!(
|
||||
search_key("release-group", " Kind OF Blue "),
|
||||
search_key("release-group", "kind of blue")
|
||||
);
|
||||
// Entity namespacing keeps an artist search off a release-group entry.
|
||||
assert_ne!(
|
||||
search_key("artist", "miles davis"),
|
||||
search_key("release-group", "miles davis")
|
||||
);
|
||||
}
|
||||
|
||||
/// Punctuation is significant to MusicBrainz, so it must survive
|
||||
/// normalisation — otherwise two genuinely different works share a cache
|
||||
/// entry and one of them is served the other's answer.
|
||||
#[test]
|
||||
fn search_key_preserves_punctuation() {
|
||||
assert_ne!(
|
||||
search_key("artist", "m!lk"),
|
||||
search_key("artist", "mlk"),
|
||||
);
|
||||
assert_ne!(
|
||||
search_key("release-group", "re:zero"),
|
||||
search_key("release-group", "rezero"),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -286,3 +286,43 @@ mod tests {
|
||||
assert!(decode_infohash("").is_none());
|
||||
}
|
||||
}
|
||||
|
||||
/// Percent-encode for a URL **path or query value**, escaping space as `%20`.
|
||||
///
|
||||
/// ⚠ **Not interchangeable with [`urlencode`]**, which encodes space as `+`.
|
||||
/// The `+` form is correct for `application/x-www-form-urlencoded` bodies and
|
||||
/// wrong for a Lucene-backed search API — MusicBrainz's `/ws/2` treats `+` as
|
||||
/// the "required term" metacharacter, so a `+`-encoded space silently changes
|
||||
/// the query's meaning rather than failing loudly. Use this one for anything
|
||||
/// going into a query *value*; use `urlencode` for form bodies.
|
||||
pub fn percent_encode(s: &str) -> String {
|
||||
let mut out = String::with_capacity(s.len() * 2);
|
||||
for b in s.bytes() {
|
||||
match b {
|
||||
b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
|
||||
out.push(b as char)
|
||||
}
|
||||
b' ' => out.push_str("%20"),
|
||||
_ => out.push_str(&format!("%{b:02X}")),
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// Normalise a variable-precision date (`YYYY`, `YYYY-MM`, `YYYY-MM-DD`) to a
|
||||
/// full `YYYY-MM-DD`, or `None` if it is not one.
|
||||
///
|
||||
/// ⚠ Validate the shape, never infer it from the length: a first cut matched on
|
||||
/// `len()` alone, which turned the seven-character string "garbage" into the
|
||||
/// `releasedate` "garbage-01" — a malformed value written to the shared hash
|
||||
/// where nothing downstream would question it.
|
||||
pub fn normalise_date(raw: &str) -> Option<String> {
|
||||
let digits = |s: &str, n: usize| s.len() == n && s.chars().all(|c| c.is_ascii_digit());
|
||||
let parts: Vec<&str> = raw.trim().split('-').collect();
|
||||
match parts.as_slice() {
|
||||
[y] if digits(y, 4) => Some(format!("{y}-01-01")),
|
||||
[y, m] if digits(y, 4) && digits(m, 2) => Some(format!("{y}-{m}-01")),
|
||||
[y, m, d] if digits(y, 4) && digits(m, 2) && digits(d, 2) => Some(format!("{y}-{m}-{d}")),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
+225
@@ -423,6 +423,86 @@ fn identity_locator_cid(codec: u64, digest: &[u8]) -> String {
|
||||
/// `beetswap::incoming_stream`). The `.nzb` manifest travels separately, as an
|
||||
/// ordinary sha2-256 IPFS cid in the record's `manifest` field. See
|
||||
/// `meta-gateway/docs/others/self-hosted-usenet-indexer-study.md` §5.
|
||||
/// Custom multicodec for the **url-locator** family.
|
||||
///
|
||||
/// A CID whose *bytes are a URL string* rather than a hash of the file. It lets
|
||||
/// a record reference externally-hosted bytes without storing a raw URL field:
|
||||
/// the URL is wrapped in a CID, so it flows through every cid-shaped code path
|
||||
/// (`cids/*` membership, `/api/file/{cid}/raw`, the reverse index) exactly like
|
||||
/// a content hash — honouring `METADATA_KEYS.md` rule #5, "every byte reference
|
||||
/// is a `cid`, never a URL".
|
||||
///
|
||||
/// Contrast `ext-play` (`0x1009`), which is wire-identical apart from the
|
||||
/// codec: `0x1006` means **"fetch these bytes once and seed them"**, while
|
||||
/// `0x1009` means "open this; there are no bytes for us, ever".
|
||||
///
|
||||
/// ⚠ **Never route a url-CID through a fixed-64-byte multihash type.** An
|
||||
/// archive download URL runs to ~75 bytes, past the 62-ish ceiling of
|
||||
/// meta-share's `CidGeneric<64>` bitswap alias. That is fine — it is never
|
||||
/// bitswap *content*, and the `/raw` dispatch decodes it off the raw string
|
||||
/// before the fixed-size codec dispatch runs — but a consumer that parses it as
|
||||
/// an `MsCid` will reject a perfectly good locator.
|
||||
pub const URL_LOCATOR_CODEC: u64 = 0x1006;
|
||||
|
||||
/// Encode an `http(s)` URL as a `0x1006` identity-multihash CIDv1.
|
||||
///
|
||||
/// The identity multihash (`0x00`) means "the digest *is* the payload", so
|
||||
/// [`decode_url_cid`] recovers the URL verbatim with zero I/O.
|
||||
///
|
||||
/// `None` for anything that is not an absolute http(s) URL — a relative or
|
||||
/// malformed value would encode fine and then fail unresolvably at play time,
|
||||
/// which is a far more expensive way to find out.
|
||||
pub fn compute_url_cid(url: &str) -> Option<String> {
|
||||
let url = url.trim();
|
||||
if !(url.starts_with("http://") || url.starts_with("https://")) {
|
||||
return None;
|
||||
}
|
||||
Some(identity_locator_cid(URL_LOCATOR_CODEC, url.as_bytes()))
|
||||
}
|
||||
|
||||
/// The artwork locator for a `cover` / `photo` field.
|
||||
///
|
||||
/// A thin alias over [`compute_url_cid`] that names the *intent* at the call
|
||||
/// sites: artwork is referenced by a locator cid, never by a raw `*_url` field
|
||||
/// (METADATA_KEYS rule #5). `None` for a non-http(s) URL, in which case the
|
||||
/// caller writes no artwork field at all — better a card the gate hides than
|
||||
/// one carrying an unresolvable reference.
|
||||
pub fn artwork_locator(url: &str) -> Option<String> {
|
||||
compute_url_cid(url)
|
||||
}
|
||||
|
||||
/// Recover the URL from a `url` locator cid — the exact inverse of
|
||||
/// [`compute_url_cid`], with no I/O.
|
||||
///
|
||||
/// `None` if `cid` is not a well-formed `0x1006` identity-multihash CIDv1 over
|
||||
/// an http(s) URL.
|
||||
pub fn decode_url_cid(cid: &str) -> Option<String> {
|
||||
let bytes = base32_lower_decode(cid.strip_prefix('b')?)?;
|
||||
let mut i = 0usize;
|
||||
let mut next = || -> Option<u64> {
|
||||
let mut n = 0u64;
|
||||
let mut shift = 0u32;
|
||||
loop {
|
||||
let b = *bytes.get(i)?;
|
||||
i += 1;
|
||||
n |= u64::from(b & 0x7f) << shift;
|
||||
if b & 0x80 == 0 {
|
||||
return Some(n);
|
||||
}
|
||||
shift += 7;
|
||||
if shift > 63 {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
};
|
||||
if next()? != 1 || next()? != URL_LOCATOR_CODEC || next()? != 0x00 {
|
||||
return None;
|
||||
}
|
||||
let len = next()? as usize;
|
||||
let url = std::str::from_utf8(bytes.get(i..i + len)?).ok()?;
|
||||
(url.starts_with("http://") || url.starts_with("https://")).then(|| url.to_string())
|
||||
}
|
||||
|
||||
pub const NZB_POSTING_CODEC: u64 = 0x1003;
|
||||
|
||||
/// Mint an `nzb-posting` cid (`0x1003`) from a release's article Message-IDs.
|
||||
@@ -1074,3 +1154,148 @@ mod tests {
|
||||
assert_eq!(&decoded[5..], &expect);
|
||||
}
|
||||
}
|
||||
|
||||
/// RFC 4648 base32 (lowercase, no padding) → bytes. Inverse of
|
||||
/// [`base32_lower_no_padding`]; used by the identity-multihash decoders.
|
||||
fn base32_lower_decode(s: &str) -> Option<Vec<u8>> {
|
||||
let mut acc = 0u16;
|
||||
let mut bits = 0u32;
|
||||
let mut out = Vec::with_capacity(s.len() * 5 / 8);
|
||||
for c in s.bytes() {
|
||||
let v = match c {
|
||||
b'a'..=b'z' => c - b'a',
|
||||
b'2'..=b'7' => c - b'2' + 26,
|
||||
_ => return None,
|
||||
};
|
||||
acc = (acc << 5) | u16::from(v);
|
||||
bits += 5;
|
||||
if bits >= 8 {
|
||||
bits -= 8;
|
||||
out.push((acc >> bits) as u8);
|
||||
}
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod url_locator_wire_tests {
|
||||
use super::*;
|
||||
|
||||
/// Decode a base32-lower multibase string back to bytes, so the tests can
|
||||
/// assert the wire layout rather than an opaque string.
|
||||
fn decode(cid: &str) -> Vec<u8> {
|
||||
const A: &str = "abcdefghijklmnopqrstuvwxyz234567";
|
||||
let mut bits = String::new();
|
||||
for c in cid.strip_prefix('b').expect("multibase b").chars() {
|
||||
let i = A.find(c).expect("base32 alphabet");
|
||||
bits.push_str(&format!("{i:05b}"));
|
||||
}
|
||||
bits.as_bytes()
|
||||
.chunks(8)
|
||||
.filter(|c| c.len() == 8)
|
||||
.map(|c| u8::from_str_radix(std::str::from_utf8(c).unwrap(), 2).unwrap())
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn read_varint(data: &[u8], at: &mut usize) -> u64 {
|
||||
let (mut v, mut shift) = (0u64, 0);
|
||||
loop {
|
||||
let b = data[*at];
|
||||
*at += 1;
|
||||
v |= ((b & 0x7f) as u64) << shift;
|
||||
if b & 0x80 == 0 {
|
||||
return v;
|
||||
}
|
||||
shift += 7;
|
||||
}
|
||||
}
|
||||
|
||||
/// Pins the on-the-wire layout `METADATA_KEYS.md` §2.1 specifies. meta-share
|
||||
/// decodes exactly this; a change here is a change to a cross-binary
|
||||
/// contract, not an implementation detail.
|
||||
#[test]
|
||||
fn the_wire_layout_matches_the_specified_encoding() {
|
||||
let url = "https://archive.org/download/item/track.flac";
|
||||
let cid = compute_url_cid(url).expect("cid");
|
||||
let raw = decode(&cid);
|
||||
let mut i = 0;
|
||||
assert_eq!(raw[i], 0x01, "CIDv1");
|
||||
i += 1;
|
||||
assert_eq!(read_varint(&raw, &mut i), URL_LOCATOR_CODEC);
|
||||
assert_eq!(raw[i], 0x00, "identity multihash");
|
||||
i += 1;
|
||||
let len = read_varint(&raw, &mut i) as usize;
|
||||
assert_eq!(len, url.len());
|
||||
assert_eq!(
|
||||
std::str::from_utf8(&raw[i..i + len]).unwrap(),
|
||||
url,
|
||||
"the digest IS the payload — decoding must recover the URL verbatim"
|
||||
);
|
||||
}
|
||||
|
||||
/// ⚠ A real archive URL is ~75 bytes, past the ~62-byte ceiling of a
|
||||
/// fixed-64 multihash type. That is expected and fine (the locator is never
|
||||
/// bitswap content), but it is the thing that trips a consumer parsing it
|
||||
/// as an `MsCid`, so the case is pinned deliberately rather than avoided.
|
||||
#[test]
|
||||
fn a_long_archive_url_still_encodes() {
|
||||
let url = "https://archive.org/download/gd1977-05-08eaton/gd77-05-08eaton-d1t01.ogg";
|
||||
assert!(url.len() > 62, "fixture must exercise the over-64 case");
|
||||
let cid = compute_url_cid(url).expect("cid");
|
||||
let raw = decode(&cid);
|
||||
let mut i = 1;
|
||||
read_varint(&raw, &mut i);
|
||||
i += 1;
|
||||
let len = read_varint(&raw, &mut i) as usize;
|
||||
assert_eq!(std::str::from_utf8(&raw[i..i + len]).unwrap(), url);
|
||||
}
|
||||
|
||||
/// Two URLs are two locators — a url-CID is not byte-derivable, so it is an
|
||||
/// address this one record advertises, never a content identity.
|
||||
#[test]
|
||||
fn distinct_urls_yield_distinct_locators() {
|
||||
assert_ne!(
|
||||
compute_url_cid("https://example.com/a.flac"),
|
||||
compute_url_cid("https://example.com/b.flac")
|
||||
);
|
||||
}
|
||||
|
||||
/// A relative or malformed value would encode fine and then fail
|
||||
/// unresolvably at play time — the expensive place to find out.
|
||||
#[test]
|
||||
fn only_absolute_http_urls_are_accepted() {
|
||||
assert!(compute_url_cid("/download/item/track.flac").is_none());
|
||||
assert!(compute_url_cid("ftp://example.com/x").is_none());
|
||||
assert!(compute_url_cid("").is_none());
|
||||
assert!(compute_url_cid("archive.org/download/x").is_none());
|
||||
assert!(compute_url_cid("http://example.com/x").is_some());
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod roundtrip_tests {
|
||||
use super::*;
|
||||
|
||||
/// Long URLs are the whole reason this locator exists (they overflow the
|
||||
/// fixed-size multihash buffer the `cid` crate uses), so the round trip is
|
||||
/// pinned at a realistic Cover Art Archive length, not a toy one.
|
||||
#[test]
|
||||
fn encode_then_decode_is_identity() {
|
||||
for url in [
|
||||
"https://coverartarchive.org/release/aa997ea0-2936-40bd-884d-3af8a0e064dc/12345-500.jpg",
|
||||
"https://archive.org/services/img/gd1977-05-08",
|
||||
"http://example.test/a.png",
|
||||
] {
|
||||
let cid = compute_url_cid(url).expect("encodes");
|
||||
assert!(cid.starts_with('b'));
|
||||
assert_eq!(decode_url_cid(&cid).as_deref(), Some(url), "round trip: {url}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_non_locator_cid_does_not_decode() {
|
||||
// A card locator (0x1007) must not be mistaken for a url locator.
|
||||
let card = crate::hash::compute_card_cid("musicbrainz", "abc").unwrap();
|
||||
assert_eq!(decode_url_cid(&card), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
/// `metacore-app` service the AppStore feeder apps ship on the shared network.
|
||||
pub const DEFAULT_META_CORE_URL: &str = "http://metacore-app:9000";
|
||||
|
||||
pub mod budget;
|
||||
pub mod cache;
|
||||
pub mod common;
|
||||
pub mod config;
|
||||
@@ -28,6 +29,7 @@ pub mod enrich;
|
||||
pub mod filename_meta;
|
||||
pub mod hash;
|
||||
pub mod lang;
|
||||
pub mod licence;
|
||||
pub mod meta_core;
|
||||
pub mod plugin;
|
||||
pub mod query;
|
||||
@@ -35,7 +37,9 @@ pub mod query_eval;
|
||||
pub mod serve;
|
||||
pub mod types;
|
||||
|
||||
pub use budget::RateBudget;
|
||||
pub use config::{ConfigField, ConfigSchema, FieldKind};
|
||||
pub use licence::licence_from_url;
|
||||
pub use enrich::{EnrichTarget, EnrichmentConfig, Enricher};
|
||||
pub use meta_core::FeederStore;
|
||||
pub use plugin::{
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
//! Licence-URL → SPDX-style identifier.
|
||||
//!
|
||||
//! Generic across upstreams: Internet Archive, Jamendo and Wikimedia Commons
|
||||
//! all publish a Creative Commons *URL*, while `METADATA_KEYS.md` requires an
|
||||
//! SPDX-style identifier — a consumer comparing licences should not be doing
|
||||
//! string surgery on `http://` vs `https://` and a trailing slash.
|
||||
|
||||
/// `METADATA_KEYS.md` requires an SPDX-style identifier rather than a URL,
|
||||
/// because a consumer comparing licences should not be doing string surgery on
|
||||
/// `http://` vs `https://` and a trailing slash.
|
||||
pub fn licence_from_url(url: &str) -> Option<String> {
|
||||
let u = url.trim().trim_end_matches('/').to_ascii_lowercase();
|
||||
if u.is_empty() {
|
||||
return None;
|
||||
}
|
||||
if u.contains("publicdomain/zero") {
|
||||
return Some("CC0-1.0".to_string());
|
||||
}
|
||||
if u.contains("publicdomain/mark") {
|
||||
return Some("PublicDomain".to_string());
|
||||
}
|
||||
// .../licenses/<variant>/<version>[/<jurisdiction>]
|
||||
let rest = u.split("/licenses/").nth(1)?;
|
||||
let mut parts = rest.split('/');
|
||||
let variant = parts.next()?.to_ascii_uppercase();
|
||||
let version = parts.next().unwrap_or("4.0");
|
||||
if variant.is_empty() {
|
||||
return None;
|
||||
}
|
||||
Some(format!("CC-{variant}-{version}"))
|
||||
}
|
||||
Reference in New Issue
Block a user