diff --git a/tracking/src/link_tests.rs b/tracking/src/link_tests.rs new file mode 100644 index 000000000..b927f33b5 --- /dev/null +++ b/tracking/src/link_tests.rs @@ -0,0 +1,260 @@ +use super::*; +use axum::{extract::State, http::StatusCode, response::IntoResponse, routing::get, Router}; +use std::sync::atomic::AtomicUsize; +use tokio::sync::Barrier; + +struct Backend { + calls: AtomicUsize, + status: AtomicU32, + delay: Duration, + body: String, +} + +async fn lookup( + State(backend): State>, + headers: axum::http::HeaderMap, +) -> impl IntoResponse { + assert_eq!( + headers[axum::http::header::AUTHORIZATION], + "Bearer fixture-token" + ); + backend.calls.fetch_add(1, Ordering::Relaxed); + tokio::time::sleep(backend.delay).await; + ( + StatusCode::from_u16(backend.status.load(Ordering::Relaxed) as u16).unwrap(), + backend.body.clone(), + ) +} + +struct Fixture { + resolver: Arc, + backend: Arc, + server: tokio::task::JoinHandle<()>, +} + +impl Fixture { + async fn new(status: u16, body: &str, delay: Duration) -> Self { + let _ = rustls::crypto::ring::default_provider().install_default(); + let backend = Arc::new(Backend { + calls: AtomicUsize::new(0), + status: AtomicU32::new(status.into()), + delay, + body: body.into(), + }); + let app = Router::new() + .route("/api/v1/internal/tracked-links/:id", get(lookup)) + .with_state(backend.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let resolver = Arc::new(LinkResolver::new( + format!("http://{}", listener.local_addr().unwrap()), + "fixture-token".into(), + )); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + Self { + resolver, + backend, + server, + } + } + + async fn wave(&self, ticket: &str, callers: usize, same_source: bool) -> Vec { + let gate = Arc::new(Barrier::new(callers)); + let mut tasks = Vec::new(); + for i in 0..callers { + let (gate, resolver) = (gate.clone(), self.resolver.clone()); + let ticket = ticket.to_string(); + tasks.push(tokio::spawn(async move { + gate.wait().await; + resolver + .resolve( + &ticket, + &format!("source-{}", if same_source { 0 } else { i }), + ) + .await + })); + } + let mut results = Vec::new(); + for task in tasks { + results.push(task.await.unwrap()); + } + results + } + + fn calls(&self) -> usize { + self.backend.calls.load(Ordering::Relaxed) + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.server.abort(); + } +} + +const LINK: &str = r#"{"destination":"https://example.com/a?q=1#top","task_id":"11111111-2222-3333-4444-555555555555","identify":true}"#; + +fn assert_link(result: Resolution, identify: bool) { + match result { + Resolution::Found(link) => { + assert_eq!(link.destination, "https://example.com/a?q=1#top"); + assert_eq!(link.task_id, "11111111-2222-3333-4444-555555555555"); + assert_eq!(link.identify, identify); + } + _ => panic!("expected a verified destination"), + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn concurrent_ticket_lookups_share_one_backend_request() { + let fixture = Fixture::new(200, LINK, Duration::from_millis(40)).await; + for result in fixture.wave("ticket", 32, false).await { + assert_link(result, true); + } + assert_eq!(fixture.calls(), 1); + assert_link(fixture.resolver.resolve("ticket", "source-0").await, true); + assert_eq!( + fixture.calls(), + 1, + "repeat clicks use the same positive cache" + ); + assert_link( + fixture.resolver.resolve("another-ticket", "source-0").await, + true, + ); + assert_eq!(fixture.calls(), 2, "different tickets are not merged"); +} + +#[tokio::test] +async fn old_ticket_payloads_still_default_to_no_identification() { + let fixture = Fixture::new(200, &LINK.replace(",\"identify\":true", ""), Duration::ZERO).await; + assert_link(fixture.resolver.resolve("ticket", "source").await, false); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn coalesced_unknown_tickets_charge_each_source_and_keep_the_miss_budget() { + let fixture = Fixture::new(404, "unknown", Duration::from_millis(40)).await; + for result in fixture.wave("unknown", 8, false).await { + assert!(matches!(result, Resolution::NotFound)); + } + assert_eq!(fixture.calls(), 1); + for i in 0..8 { + let source = format!("source-{i}"); + assert_eq!( + fixture + .resolver + .miss_budget + .get(&source) + .await + .unwrap() + .load(Ordering::Relaxed), + 1 + ); + } + for result in fixture.wave("unknown", 12, true).await { + assert!(matches!(result, Resolution::NotFound)); + } + assert_eq!(fixture.calls(), 1, "negative hits never look up again"); + assert_eq!( + fixture + .resolver + .miss_budget + .get("source-0") + .await + .unwrap() + .load(Ordering::Relaxed), + 13 + ); + assert!(matches!( + fixture.resolver.resolve("fresh", "source-0").await, + Resolution::NotFound + )); + assert_eq!( + fixture.calls(), + 1, + "miss budget must still stop backend requests" + ); + fixture.backend.status.store(200, Ordering::Relaxed); + // A confirmed positive ticket remains available even to a source that spent its miss budget. + fixture + .resolver + .found + .insert( + "positive".into(), + ResolvedLink { + destination: "https://example.com/a?q=1#top".into(), + task_id: "11111111-2222-3333-4444-555555555555".into(), + identify: true, + }, + ) + .await; + assert_link(fixture.resolver.resolve("positive", "source-0").await, true); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn backend_errors_are_not_cached_or_charged_as_unknown_tickets() { + let fixture = Fixture::new(503, LINK, Duration::from_millis(40)).await; + for result in fixture.wave("ticket", 8, false).await { + assert!(matches!(result, Resolution::Unavailable)); + } + assert_eq!(fixture.calls(), 1); + assert_eq!(fixture.resolver.breaker_failures.load(Ordering::Relaxed), 1); + assert!(!fixture.resolver.not_found.contains_key("ticket")); + for i in 0..8 { + assert!(fixture.resolver.miss_allowed(&format!("source-{i}")).await); + } + fixture.backend.status.store(200, Ordering::Relaxed); + assert_link(fixture.resolver.resolve("ticket", "source-0").await, true); + assert_eq!(fixture.calls(), 2, "a backend error does not prevent retry"); + assert_eq!(fixture.resolver.breaker_failures.load(Ordering::Relaxed), 0); +} + +#[tokio::test] +async fn decode_errors_fail_closed_and_actual_failures_still_trip_the_breaker() { + let fixture = Fixture::new(200, "not json", Duration::ZERO).await; + for _ in 0..BREAKER_TRIP { + assert!(matches!( + fixture.resolver.resolve("ticket", "source").await, + Resolution::Unavailable + )); + } + assert_eq!(fixture.calls(), BREAKER_TRIP as usize); + assert!(fixture.resolver.breaker_is_open()); + assert!(matches!( + fixture.resolver.resolve("other-ticket", "source").await, + Resolution::Unavailable + )); + assert_eq!(fixture.calls(), BREAKER_TRIP as usize); + assert!(!fixture.resolver.not_found.contains_key("ticket")); + assert!(fixture.resolver.miss_allowed("source").await); +} + +#[tokio::test] +async fn denied_internal_requests_fail_closed_without_becoming_cache_entries() { + for status in [401, 403] { + let fixture = Fixture::new(status, LINK, Duration::ZERO).await; + assert!(matches!( + fixture.resolver.resolve("ticket", "source").await, + Resolution::Unavailable + )); + assert!(!fixture.resolver.not_found.contains_key("ticket")); + assert!(fixture.resolver.found.get("ticket").await.is_none()); + fixture.backend.status.store(200, Ordering::Relaxed); + assert_link(fixture.resolver.resolve("ticket", "source").await, true); + assert_eq!(fixture.calls(), 2); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "synthetic localhost call-count benchmark; run with --ignored --nocapture"] +async fn benchmark_concurrent_ticket_resolution() { + let fixture = Fixture::new(200, LINK, Duration::from_millis(40)).await; + let started = Instant::now(); + const TICKETS: usize = 25; + const CALLERS: usize = 32; + for i in 0..TICKETS { + for result in fixture.wave(&format!("ticket-{i}"), CALLERS, false).await { + assert_link(result, true); + } + } + println!("synthetic cold-cache tickets={TICKETS}, callers_per_ticket={CALLERS}, backend_calls={}, elapsed_ms={}", fixture.calls(), started.elapsed().as_millis()); +} diff --git a/tracking/src/links.rs b/tracking/src/links.rs index a850a1205..c6f2ba4ca 100644 --- a/tracking/src/links.rs +++ b/tracking/src/links.rs @@ -44,6 +44,12 @@ pub enum Resolution { Unavailable, } +#[derive(Debug)] +enum LookupError { + NotFound, + Unavailable, +} + /// Consecutive backend failures before the breaker opens. const BREAKER_TRIP: u32 = 5; /// How long the breaker stays open once tripped. @@ -114,6 +120,31 @@ impl LinkResolver { return Resolution::Unavailable; } + // Coalesce immutable-ticket cache fills, but charge every caller's own misses. + match self + .found + .try_get_with(link_id.to_string(), self.fetch(link_id)) + .await + { + Ok(link) => Resolution::Found(link), + Err(error) => match error.as_ref() { + LookupError::NotFound => { + self.count_miss(source).await; + Resolution::NotFound + } + LookupError::Unavailable => Resolution::Unavailable, + }, + } + } + + async fn fetch(&self, link_id: &str) -> Result { + if self.not_found.contains_key(link_id) { + return Err(LookupError::NotFound); + } + if self.breaker_is_open() { + return Err(LookupError::Unavailable); + } + let url = format!( "{}/api/v1/internal/tracked-links/{}", self.backend_url, link_id @@ -129,35 +160,32 @@ impl LinkResolver { Ok(resp) if resp.status().is_success() => match resp.json::().await { Ok(body) => { self.breaker_failures.store(0, Ordering::Relaxed); - let link = ResolvedLink { + Ok(ResolvedLink { destination: body.destination, task_id: body.task_id, identify: body.identify, - }; - self.found.insert(link_id.to_string(), link.clone()).await; - Resolution::Found(link) + }) } Err(e) => { warn!("tracked-link decode failed: {}", e); self.record_failure(); - Resolution::Unavailable + Err(LookupError::Unavailable) } }, Ok(resp) if resp.status() == reqwest::StatusCode::NOT_FOUND => { self.breaker_failures.store(0, Ordering::Relaxed); self.not_found.insert(link_id.to_string(), ()).await; - self.count_miss(source).await; - Resolution::NotFound + Err(LookupError::NotFound) } Ok(resp) => { warn!("tracked-link lookup unexpected status: {}", resp.status()); self.record_failure(); - Resolution::Unavailable + Err(LookupError::Unavailable) } Err(e) => { warn!("tracked-link lookup failed: {}", e); self.record_failure(); - Resolution::Unavailable + Err(LookupError::Unavailable) } } } @@ -198,3 +226,7 @@ impl LinkResolver { } } } + +#[cfg(test)] +#[path = "link_tests.rs"] +mod tests;