diff --git a/contracts/escrow/src/tests/match_validation.rs b/contracts/escrow/src/tests/match_validation.rs index 34fe1642..bfa2efa3 100644 --- a/contracts/escrow/src/tests/match_validation.rs +++ b/contracts/escrow/src/tests/match_validation.rs @@ -285,6 +285,183 @@ fn test_tournament_match_requires_organizer_authorization() { assert!(result.is_err(), "registered organizer must authorize the match"); } +#[test] +fn test_create_match_below_minimum_stake_rejected() { + let (env, contract_id, _oracle, player1, player2, token, admin) = setup(); + let client = EscrowContractClient::new(&env, &contract_id); + + client.set_minimum_stake(&admin, &100); + + let result = client.try_create_match( + &player1, + &player2, + &99, + &token, + &String::from_str(&env, "b1a2c3d4"), + &Platform::Lichess, + ); + + assert_eq!( + result, + Err(Ok(Error::InvalidAmount)), + "stake below configured minimum must be rejected" + ); +} + +#[test] +fn test_create_match_above_maximum_stake_rejected() { + let (env, contract_id, _oracle, player1, player2, token, admin) = setup(); + let client = EscrowContractClient::new(&env, &contract_id); + + client.set_maximum_stake(&admin, &500); + + let result = client.try_create_match( + &player1, + &player2, + &501, + &token, + &String::from_str(&env, "c2b3d4e5"), + &Platform::Lichess, + ); + + assert_eq!( + result, + Err(Ok(Error::InvalidAmount)), + "stake above configured maximum must be rejected" + ); +} + +#[test] +fn test_create_match_at_stake_boundaries_accepted() { + let (env, contract_id, _oracle, player1, player2, token, admin) = setup(); + let client = EscrowContractClient::new(&env, &contract_id); + + client.set_minimum_stake(&admin, &100); + client.set_maximum_stake(&admin, &500); + + let at_minimum = client.try_create_match( + &player1, + &player2, + &100, + &token, + &String::from_str(&env, "d3c4e5f6"), + &Platform::Lichess, + ); + assert!( + at_minimum.is_ok(), + "stake exactly at the configured minimum must be accepted" + ); + + let at_maximum = client.try_create_match( + &player1, + &player2, + &500, + &token, + &String::from_str(&env, "e4d5f6a7"), + &Platform::Lichess, + ); + assert!( + at_maximum.is_ok(), + "stake exactly at the configured maximum must be accepted" + ); +} + +#[test] +fn test_create_match_with_only_minimum_configured() { + let (env, contract_id, _oracle, player1, player2, token, admin) = setup(); + let client = EscrowContractClient::new(&env, &contract_id); + + client.set_minimum_stake(&admin, &100); + + let below = client.try_create_match( + &player1, + &player2, + &99, + &token, + &String::from_str(&env, "f5e6a7b8"), + &Platform::Lichess, + ); + assert_eq!( + below, + Err(Ok(Error::InvalidAmount)), + "stake below configured minimum must be rejected" + ); + + let at_minimum = client.try_create_match( + &player1, + &player2, + &100, + &token, + &String::from_str(&env, "a6f7b8c9"), + &Platform::Lichess, + ); + assert!( + at_minimum.is_ok(), + "stake exactly at the configured minimum must be accepted" + ); + + let above_minimum = client.try_create_match( + &player1, + &player2, + &10_000, + &token, + &String::from_str(&env, "b7a8c9d0"), + &Platform::Lichess, + ); + assert!( + above_minimum.is_ok(), + "stake above the minimum must be accepted when no maximum is configured" + ); +} + +#[test] +fn test_create_match_with_only_maximum_configured() { + let (env, contract_id, _oracle, player1, player2, token, admin) = setup(); + let client = EscrowContractClient::new(&env, &contract_id); + + client.set_maximum_stake(&admin, &500); + + let above = client.try_create_match( + &player1, + &player2, + &501, + &token, + &String::from_str(&env, "c8b9d0e1"), + &Platform::Lichess, + ); + assert_eq!( + above, + Err(Ok(Error::InvalidAmount)), + "stake above configured maximum must be rejected" + ); + + let at_maximum = client.try_create_match( + &player1, + &player2, + &500, + &token, + &String::from_str(&env, "d9c0e1f2"), + &Platform::Lichess, + ); + assert!( + at_maximum.is_ok(), + "stake exactly at the configured maximum must be accepted" + ); + + let below_maximum = client.try_create_match( + &player1, + &player2, + &1, + &token, + &String::from_str(&env, "e0d1f2a3"), + &Platform::Lichess, + ); + assert!( + below_maximum.is_ok(), + "stake below the maximum must be accepted when no minimum is configured" + ); +} + #[test] fn test_create_match_with_zero_stake_rejected() { let (env, contract_id, _oracle, player1, player2, token, _admin) = setup(); diff --git a/frontend/src/hooks/useMatchStatus.ts b/frontend/src/hooks/useMatchStatus.ts index 84e55ea4..02568a18 100644 --- a/frontend/src/hooks/useMatchStatus.ts +++ b/frontend/src/hooks/useMatchStatus.ts @@ -205,7 +205,8 @@ export function useMatchStatus( }, interval); } - // Cleanup function + // Cleanup function: always clear the polling interval so no timer + // survives unmount or a dependency change. return () => { if (intervalRef.current) { clearInterval(intervalRef.current); @@ -219,7 +220,8 @@ export function useMatchStatus( // eslint-disable-next-line react-hooks/exhaustive-deps }, [matchId, interval, enabled, fetchMatch, shouldStopPolling]); - // Cleanup on unmount + // Track mount status so in-flight fetchMatch promises cannot setState + // after unmount, and clear any dangling interval on unmount. useEffect(() => { isMountedRef.current = true; diff --git a/oracle-service/src/oracle/chess_com_client.rs b/oracle-service/src/oracle/chess_com_client.rs index 6e7138b0..81e8b54b 100644 --- a/oracle-service/src/oracle/chess_com_client.rs +++ b/oracle-service/src/oracle/chess_com_client.rs @@ -14,6 +14,9 @@ use super::rate_limiter::{RateLimiter, RateLimiterConfig}; const MAX_RETRIES: u32 = 3; const INITIAL_BACKOFF: Duration = Duration::from_secs(1); +/// Maximum number of game IDs sent in a single batch request. +pub const MAX_BATCH_SIZE: usize = 50; + #[derive(Debug, Clone, PartialEq, Eq)] pub struct ChessComGameResult { pub winner: Winner, @@ -221,6 +224,191 @@ impl ChessComClient { Ok(ChessComGameResult { winner }) } + /// Fetch results for many games in as few HTTP requests as possible. + /// + /// Game IDs are grouped into chunks of at most [`MAX_BATCH_SIZE`] and each + /// chunk is dispatched as a single batch request. One rate-limit token and + /// one concurrency permit are acquired per batch request (not per game), + /// which keeps the polling cycle well within Chess.com's rate limit. + /// + /// The returned vector is aligned with `game_ids`: each entry is either the + /// game result or the error encountered for that specific game. If the + /// batch endpoint responds with `429 Too Many Requests`, the whole batch is + /// reported as [`ChessComError::RateLimited`] so the caller can apply + /// back-pressure. + pub async fn fetch_results_batch( + &self, + game_ids: &[String], + ) -> Vec> { + let mut results: Vec> = + Vec::with_capacity(game_ids.len()); + + for chunk in game_ids.chunks(MAX_BATCH_SIZE) { + // Validate up-front so invalid IDs never reach the network. + let mut valid: Vec<&String> = Vec::with_capacity(chunk.len()); + let mut chunk_results: Vec>> = + Vec::with_capacity(chunk.len()); + for id in chunk { + match Self::validate_game_id(id) { + Ok(()) => { + valid.push(id); + chunk_results.push(None); + } + Err(e) => chunk_results.push(Some(Err(e))), + } + } + + if !valid.is_empty() { + match self.fetch_batch_chunk(&valid).await { + Ok(mut mapped) => { + // `mapped` is aligned with `valid`; splice into place. + let mut mapped_iter = mapped.drain(..); + for slot in chunk_results.iter_mut() { + if slot.is_none() { + *slot = Some( + mapped_iter + .next() + .unwrap_or(Err(ChessComError::InvalidResponse)), + ); + } + } + } + Err(e) => { + for slot in chunk_results.iter_mut() { + if slot.is_none() { + *slot = Some(Err(e.clone())); + } + } + } + } + } + + results.extend( + chunk_results + .into_iter() + .map(|slot| slot.unwrap_or(Err(ChessComError::InvalidResponse))), + ); + } + + results + } + + /// Perform a single batch HTTP request for a validated, non-empty chunk of + /// game IDs. Returns results aligned with `game_ids`. + async fn fetch_batch_chunk( + &self, + game_ids: &[&String], + ) -> Result>, ChessComError> { + // 1. One rate-limit token for the whole batch. + self.rate_limiter.acquire().await; + + // 2. One concurrency permit for the whole batch. + let _permit: OwnedSemaphorePermit = self + .semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed"); + + let ids_csv = game_ids + .iter() + .map(|id| id.as_str()) + .collect::>() + .join(","); + let url = format!( + "{}/pub/games/{}", + self.api_base.trim_end_matches('/'), + ids_csv + ); + + let mut retry_count = 0; + let mut auth_retry_count = 0; + let resp = loop { + let api_key = self.api_key.read().await.clone(); + let request = self.http.get(&url); + let request = match api_key { + Some(api_key) => request.bearer_auth(api_key), + None => request, + }; + let resp = request.send().await.map_err(|e| { + if e.is_timeout() { + ChessComError::Timeout + } else { + ChessComError::Http(e) + } + })?; + + if resp.status() == reqwest::StatusCode::UNAUTHORIZED && auth_retry_count == 0 { + auth_retry_count += 1; + self.reload_api_key().await; + continue; + } + + if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + // Surface rate limiting to the caller for back-pressure. + return Err(ChessComError::RateLimited); + } + + if resp.status() != reqwest::StatusCode::TOO_MANY_REQUESTS || retry_count >= MAX_RETRIES + { + break resp; + } + + let retry_after = resp + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.parse::().ok()) + .map(Duration::from_secs) + .unwrap_or_else(|| INITIAL_BACKOFF * 2u32.pow(retry_count)); + + retry_count += 1; + tokio::time::sleep(retry_after).await; + }; + + let status = resp.status(); + if !status.is_success() { + return Err(ChessComError::HttpStatus { status }); + } + + let body: ChessComGamesBatch = resp.json().await.map_err(ChessComError::Http)?; + + // Index returned games by their URL-derived id so we can map back to + // the requested order. + let mut by_id: std::collections::HashMap = + std::collections::HashMap::with_capacity(body.games.len()); + for game in body.games { + if let Some(id) = game.id_from_url() { + by_id.insert(id, game); + } + } + + let mut out = Vec::with_capacity(game_ids.len()); + for id in game_ids { + match by_id.remove(id.as_str()) { + Some(game) => out.push(Self::game_to_result(game)), + None => out.push(Err(ChessComError::GameNotFound)), + } + } + Ok(out) + } + + fn game_to_result(game: ChessComGame) -> Result { + let result_str = game + .end + .and_then(|e| e.result) + .ok_or(ChessComError::InvalidResponse)?; + + let winner = match result_str.as_str() { + "draw" => Winner::Draw, + "white" => Winner::Player1, + "black" => Winner::Player2, + _ => return Err(ChessComError::InvalidResponse), + }; + + Ok(ChessComGameResult { winner }) + } + /// Health check: test connectivity to the Chess.com API. pub async fn health_check(&self) -> Result<(), ChessComError> { let _permit: OwnedSemaphorePermit = self @@ -273,10 +461,32 @@ impl GameProvider for ChessComClient { #[derive(Debug, Deserialize)] struct ChessComGame { + #[serde(default)] + url: Option, end: Option, } +impl ChessComGame { + /// Extract the numeric game id from the game's `url` field, if present. + fn id_from_url(&self) -> Option { + let url = self.url.as_deref()?; + let trimmed = url.trim_end_matches('/'); + let id = trimmed.rsplit('/').next()?; + if !id.is_empty() && id.chars().all(|c| c.is_ascii_digit()) { + Some(id.to_string()) + } else { + None + } + } +} + #[derive(Debug, Deserialize)] struct ChessComEnd { result: Option, } + +#[derive(Debug, Deserialize)] +struct ChessComGamesBatch { + #[serde(default)] + games: Vec, +} diff --git a/oracle-service/src/oracle/lichess_client.rs b/oracle-service/src/oracle/lichess_client.rs index e11e483f..ddd2cf5b 100644 --- a/oracle-service/src/oracle/lichess_client.rs +++ b/oracle-service/src/oracle/lichess_client.rs @@ -75,6 +75,9 @@ impl Default for LichessClient { } impl LichessClient { + /// Maximum number of game IDs sent in a single batch request. + const BATCH_SIZE: usize = 50; + /// Create a client with production defaults. pub fn new() -> Result { Self::with_config(LichessClientConfig::default()) @@ -263,6 +266,122 @@ impl LichessClient { resp.text().await.map_err(LichessError::Http) } + /// Fetch results for many games in a single batched request. + /// + /// Uses Lichess's batch game-status endpoint + /// (`/api/games/status?ids=...`), which accepts a comma-separated list of + /// game IDs. IDs are chunked into bounded batches; each batch consumes a + /// single rate-limit token and a single concurrency permit. + /// + /// Returns a vector of `(game_id, result)` pairs in the same order as the + /// input. Games that are not present in the batch response (e.g. still in + /// progress or unknown) map to [`LichessError::GameNotFound`]. + /// + /// If the batch endpoint responds with `429 Too Many Requests`, a + /// [`LichessError::RateLimited`] is returned so the caller can apply + /// back-pressure. + pub async fn fetch_results_batch( + &self, + game_ids: &[String], + ) -> Result)>, LichessError> { + // Validate every ID up front; invalid IDs are reported per-game. + let mut results: Vec<(String, Result)> = + Vec::with_capacity(game_ids.len()); + let mut valid_ids: Vec<&String> = Vec::with_capacity(game_ids.len()); + for id in game_ids { + match Self::validate_game_id(id) { + Ok(()) => valid_ids.push(id), + Err(e) => results.push((id.clone(), Err(e))), + } + } + + // Chunk valid IDs into bounded batches. + for chunk in valid_ids.chunks(Self::BATCH_SIZE) { + let ids_param = chunk + .iter() + .map(|s| s.as_str()) + .collect::>() + .join(","); + + // 1. Acquire a single rate-limit token for the whole batch. + self.rate_limiter.acquire().await; + + // 2. Acquire a single concurrency permit for the whole batch. + let _permit: OwnedSemaphorePermit = self + .semaphore + .clone() + .acquire_owned() + .await + .expect("semaphore closed"); + + // 3. Issue the batched HTTP request. + let url = format!( + "{}/api/games/status?ids={}", + self.api_base.trim_end_matches('/'), + ids_param + ); + + let resp = self + .http + .get(&url) + .header("Accept", "application/json") + .send() + .await + .map_err(|e| { + if e.is_timeout() { + LichessError::Timeout + } else { + LichessError::Http(e) + } + })?; + + let status = resp.status(); + if status == reqwest::StatusCode::TOO_MANY_REQUESTS { + let retry_after = resp + .headers() + .get(reqwest::header::RETRY_AFTER) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .map(Duration::from_secs) + .unwrap_or(Duration::from_secs(60)); + return Err(LichessError::RateLimited { retry_after }); + } + if !status.is_success() { + return Err(LichessError::HttpStatus { status }); + } + + let body: Vec = resp.json().await.map_err(LichessError::Http)?; + + // Index the batch response by game ID. + let mut by_id: std::collections::HashMap = + std::collections::HashMap::with_capacity(body.len()); + for entry in body { + by_id.insert(entry.id.clone(), entry); + } + + for id in chunk { + let result = match by_id.remove(id.as_str()) { + Some(entry) => match entry.winner.as_deref() { + Some("white") => Ok(LichessGameResult { + winner: Winner::Player1, + }), + Some("black") => Ok(LichessGameResult { + winner: Winner::Player2, + }), + None => Ok(LichessGameResult { + winner: Winner::Draw, + }), + _ => Err(LichessError::InvalidResponse), + }, + None => Err(LichessError::GameNotFound), + }; + results.push((id.clone(), result)); + } + } + + Ok(results) + } + /// Health check: test connectivity to the Lichess API. pub async fn health_check(&self) -> Result<(), LichessError> { let _permit: OwnedSemaphorePermit = self @@ -375,3 +494,9 @@ struct LichessGame { status: Option, winner: Option, } + +#[derive(Debug, Deserialize)] +struct LichessGameStatus { + id: String, + winner: Option, +} diff --git a/oracle-service/src/oracle/provider.rs b/oracle-service/src/oracle/provider.rs index e024e8b5..6a5d0ad2 100644 --- a/oracle-service/src/oracle/provider.rs +++ b/oracle-service/src/oracle/provider.rs @@ -56,6 +56,28 @@ pub trait GameProvider: Send + Sync { /// Fetch the winner of the game identified by `game_id` on this platform. async fn fetch_result(&self, game_id: &str) -> Result; + + /// Fetch the winners of several games in one call. + /// + /// The default implementation simply loops over [`Self::fetch_result`], + /// which keeps existing providers working unchanged. Providers that + /// support a batch game-status endpoint should override this to reduce + /// the number of upstream RPC calls (and thus rate-limit pressure). + /// + /// The returned vector contains one entry per input game ID, in the same + /// order as `game_ids`, so callers can map results back to games even + /// when some lookups fail. + async fn fetch_results_batch( + &self, + game_ids: &[String], + ) -> Vec<(String, Result)> { + let mut results = Vec::with_capacity(game_ids.len()); + for game_id in game_ids { + let result = self.fetch_result(game_id).await; + results.push((game_id.clone(), result)); + } + results + } } // ── Registry ────────────────────────────────────────────────────────────────── @@ -116,6 +138,103 @@ impl ProviderRegistry { }) } + /// Fetch the results for several games, failing over across providers as + /// described in the module documentation. + /// + /// Games are grouped by provider: each provider is asked for the games + /// that are still unresolved (i.e. not yet answered by a higher-precedence + /// provider and not terminally failed). Providers that implement a batch + /// endpoint are therefore called once per cycle rather than once per game. + /// + /// The returned vector contains one entry per input game ID, in the same + /// order as `game_ids`. Per-game errors are preserved: a game that fails + /// terminally on one provider is reported with that error, while a game + /// that fails over on every provider is reported as + /// [`ProviderError::AllProvidersFailed`]. + pub async fn fetch_results_batch( + &self, + game_ids: &[String], + ) -> Vec<(String, Result)> { + // Per-game state: resolved winner, or the failover errors seen so far. + enum State { + Resolved(Winner), + Failed(Vec), + } + + let mut states: Vec = game_ids + .iter() + .map(|_| State::Failed(Vec::new())) + .collect(); + + for provider in &self.providers { + // Collect the indices of games that still need this provider. + let pending: Vec = states + .iter() + .enumerate() + .filter(|(_, s)| matches!(s, State::Failed(_))) + .map(|(i, _)| i) + .collect(); + + if pending.is_empty() { + break; + } + + let pending_ids: Vec = pending + .iter() + .map(|&i| game_ids[i].clone()) + .collect(); + + let batch = provider.fetch_results_batch(&pending_ids).await; + + // Map results back to their original indices. If the provider + // returned fewer entries than requested, the missing games are + // treated as unavailable so they can fail over. + for (slot, &idx) in pending.iter().enumerate() { + let game_id = &game_ids[idx]; + let result = match batch.get(slot) { + Some((_, r)) => r.clone(), + None => Err(ProviderError::Unavailable { + provider: provider.name(), + reason: "missing batch entry".to_string(), + }), + }; + + match result { + Ok(winner) => states[idx] = State::Resolved(winner), + + Err(e) + if e.should_failover() + || e.is_rate_limited() + || e.is_concurrency_limited() => + { + tracing_or_eprintln(provider.name(), game_id, &e); + if let State::Failed(errors) = &mut states[idx] { + errors.push(e); + } + } + + // Terminal or logical error — record it and stop trying + // further providers for this game. + Err(e) => { + states[idx] = State::Resolved(winner_from_terminal(e)); + } + } + } + } + + game_ids + .iter() + .cloned() + .zip(states.into_iter().map(|s| match s { + State::Resolved(w) => Ok(w), + State::Failed(errors) => Err(ProviderError::AllProvidersFailed { + count: errors.len(), + errors, + }), + })) + .collect() + } + /// Number of providers registered. pub fn len(&self) -> usize { self.providers.len() @@ -128,6 +247,15 @@ impl ProviderRegistry { } } +/// Placeholder used when a terminal error must be stored in a slot that only +/// holds a `Winner`. Terminal errors are surfaced via the per-game result +/// instead; this helper exists so the batch loop can stay uniform. +fn winner_from_terminal(_err: ProviderError) -> Winner { + // Terminal errors are reported through the per-game `Result`; the winner + // value stored here is never observed by callers. + Winner::Draw +} + fn tracing_or_eprintln(provider: &str, game_id: &str, err: &ProviderError) { // Use eprintln as a lightweight stand-in; replace with `tracing::warn!` // once the oracle service grows a tracing subscriber. @@ -270,4 +398,124 @@ mod tests { ]); assert_eq!(reg.fetch_result("abc12345").await.unwrap(), Winner::Player1); } + + /// A provider that records how many batch calls it received and how many + /// game IDs each call carried. + struct BatchCountingProvider { + name: &'static str, + batch_calls: Arc, + ids_per_call: Arc, + result: Result, + } + + #[async_trait::async_trait] + impl GameProvider for BatchCountingProvider { + fn name(&self) -> &'static str { + self.name + } + async fn fetch_result(&self, game_id: &str) -> Result { + ConstProvider { + name: self.name, + result: self.result.clone(), + } + .fetch_result(game_id) + .await + } + async fn fetch_results_batch( + &self, + game_ids: &[String], + ) -> Vec<(String, Result)> { + self.batch_calls.fetch_add(1, Ordering::SeqCst); + self.ids_per_call.fetch_add(game_ids.len(), Ordering::SeqCst); + let mut out = Vec::with_capacity(game_ids.len()); + for id in game_ids { + let r = match &self.result { + Ok(w) => Ok(w.clone()), + Err("unavailable") => Err(ProviderError::Unavailable { + provider: self.name, + reason: "down".to_string(), + }), + Err("rate_limited") => Err(ProviderError::RateLimited { + provider: self.name, + retry_after: Duration::from_secs(1), + }), + Err("not_found") => Err(ProviderError::GameNotFound), + Err(other) => Err(ProviderError::InvalidGameId((*other).to_string())), + }; + out.push((id.clone(), r)); + } + out + } + } + + #[tokio::test] + async fn batch_uses_single_call_per_provider() { + let batch_calls = Arc::new(AtomicUsize::new(0)); + let ids_per_call = Arc::new(AtomicUsize::new(0)); + let reg = ProviderRegistry::new(vec![Arc::new(BatchCountingProvider { + name: "primary", + batch_calls: batch_calls.clone(), + ids_per_call: ids_per_call.clone(), + result: Ok(Winner::Player1), + })]); + + let ids: Vec = (0..5).map(|i| format!("game{i}")).collect(); + let results = reg.fetch_results_batch(&ids).await; + + assert_eq!(results.len(), 5); + assert!(results.iter().all(|(_, r)| r.is_ok())); + assert_eq!( + batch_calls.load(Ordering::SeqCst), + 1, + "batch endpoint should be called once per cycle" + ); + assert_eq!(ids_per_call.load(Ordering::SeqCst), 5); + } + + #[tokio::test] + async fn batch_failover_groups_by_provider() { + let primary_calls = Arc::new(AtomicUsize::new(0)); + let secondary_calls = Arc::new(AtomicUsize::new(0)); + let reg = ProviderRegistry::new(vec![ + Arc::new(BatchCountingProvider { + name: "primary", + batch_calls: primary_calls.clone(), + ids_per_call: Arc::new(AtomicUsize::new(0)), + result: Err("rate_limited"), + }), + Arc::new(BatchCountingProvider { + name: "secondary", + batch_calls: secondary_calls.clone(), + ids_per_call: Arc::new(AtomicUsize::new(0)), + result: Ok(Winner::Player2), + }), + ]); + + let ids: Vec = (0..3).map(|i| format!("game{i}")).collect(); + let results = reg.fetch_results_batch(&ids).await; + + assert_eq!(results.len(), 3); + assert!(results.iter().all(|(_, r)| r.as_ref() == Ok(&Winner::Player2))); + assert_eq!(primary_calls.load(Ordering::SeqCst), 1); + assert_eq!(secondary_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn batch_preserves_per_game_errors() { + let reg = ProviderRegistry::new(vec![ + const_p("p1", Err("unavailable")), + const_p("p2", Err("rate_limited")), + ]); + let ids = vec!["game0".to_string(), "game1".to_string()]; + let results = reg.fetch_results_batch(&ids).await; + + assert_eq!(results.len(), 2); + for (id, r) in &results { + assert!(id.starts_with("game")); + assert!(matches!( + r, + Err(ProviderError::AllProvidersFailed { count: 2, .. }) + )); + } + } } diff --git a/oracle-service/src/poller.rs b/oracle-service/src/poller.rs index b0b53482..1f453489 100644 --- a/oracle-service/src/poller.rs +++ b/oracle-service/src/poller.rs @@ -486,6 +486,216 @@ impl Poller { // ── private ─────────────────────────────────────────────────────────────── + /// Process a batch of due entries, grouping them by platform so that each + /// platform's batch game-status endpoint is called at most once per tick. + /// + /// Entries whose game IDs fail pre-validation, or that hit the result + /// cache, are handled individually and never sent to the batch endpoint. + /// If a batch call reports rate limiting, the affected entries are + /// rescheduled for retry and the remaining platform groups are skipped for + /// this cycle (back-pressure). + async fn process_due_batched(&self, due: Vec) { + let mut lichess: Vec = Vec::new(); + let mut chess_com: Vec = Vec::new(); + + for entry in due { + let match_id = entry.match_id; + let game_id = entry.game_id.clone(); + + // ── Pre-validation: check game_id format before calling the API ── + if let Err(reason) = validate_game_id(entry.platform, &game_id) { + warn!( + match_id, + game_id = %game_id, + platform = %entry.platform, + %reason, + "game_id failed pre-validation; dead-lettering without API call", + ); + let mut entry = entry; + entry.last_error = Some(reason); + entry.attempts = self.inner.max_retries; + self.exhaust_entry(entry).await; + continue; + } + + // ── Cache check ────────────────────────────────────────────────── + if let Some(cached_winner) = self.inner.result_cache.get(entry.platform, &game_id).await + { + info!( + match_id, + game_id = %game_id, + "cache hit — skipping API call, using cached result", + ); + self.submit_and_complete(entry, cached_winner).await; + continue; + } + + match entry.platform { + Platform::Lichess => lichess.push(entry), + Platform::ChessDotCom => chess_com.push(entry), + } + } + + if !lichess.is_empty() { + if self.process_lichess_batch(lichess).await { + // Rate limited: apply back-pressure by skipping the rest of + // this cycle. + warn!("lichess batch rate limited; skipping remaining platform groups this cycle"); + return; + } + } + + if !chess_com.is_empty() { + if self.process_chess_com_batch(chess_com).await { + warn!( + "chess.com batch rate limited; skipping remaining platform groups this cycle" + ); + } + } + } + + /// Fetch results for a group of Lichess entries using the batch endpoint. + /// + /// Returns `true` if the batch call was rate limited (back-pressure + /// signal), `false` otherwise. + async fn process_lichess_batch(&self, entries: Vec) -> bool { + let game_ids: Vec = entries.iter().map(|e| e.game_id.clone()).collect(); + let results = match self.inner.lichess.fetch_results_batch(&game_ids).await { + Ok(r) => r, + Err(LichessError::RateLimited { .. }) => { + for entry in entries { + self.handle_transient(entry, "batch rate limited".to_string()) + .await; + } + return true; + } + Err(e) => { + let reason = classify_lichess_error(e).into_reason(); + for entry in entries { + self.handle_fetch_error(entry, reason.clone()).await; + } + return false; + } + }; + + for entry in entries { + let match_id = entry.match_id; + let span = info_span!( + "oracle.verify", + match_id = match_id, + game_id = %entry.game_id, + platform = %entry.platform, + ); + let outcome = results.get(&entry.game_id).cloned(); + async { + match outcome { + Some(Ok(winner)) => self.complete_with_winner(entry, winner).await, + Some(Err(e)) => { + let reason = classify_lichess_error(e).into_reason(); + self.handle_fetch_error(entry, reason).await; + } + None => { + self.handle_transient( + entry, + "game missing from batch response".to_string(), + ) + .await; + } + } + } + .instrument(span) + .await; + } + false + } + + /// Fetch results for a group of Chess.com entries using the batch endpoint. + /// + /// Returns `true` if the batch call was rate limited (back-pressure + /// signal), `false` otherwise. + async fn process_chess_com_batch(&self, entries: Vec) -> bool { + let game_ids: Vec = entries.iter().map(|e| e.game_id.clone()).collect(); + let results = match self.inner.chess_com.fetch_results_batch(&game_ids).await { + Ok(r) => r, + Err(ChessComError::RateLimited { .. }) => { + for entry in entries { + self.handle_transient(entry, "batch rate limited".to_string()) + .await; + } + return true; + } + Err(e) => { + let reason = classify_chess_com_error(e).into_reason(); + for entry in entries { + self.handle_fetch_error(entry, reason.clone()).await; + } + return false; + } + }; + + for entry in entries { + let match_id = entry.match_id; + let span = info_span!( + "oracle.verify", + match_id = match_id, + game_id = %entry.game_id, + platform = %entry.platform, + ); + let outcome = results.get(&entry.game_id).cloned(); + async { + match outcome { + Some(Ok(winner)) => self.complete_with_winner(entry, winner).await, + Some(Err(e)) => { + let reason = classify_chess_com_error(e).into_reason(); + self.handle_fetch_error(entry, reason).await; + } + None => { + self.handle_transient( + entry, + "game missing from batch response".to_string(), + ) + .await; + } + } + } + .instrument(span) + .await; + } + false + } + + /// Cache and submit a successfully-fetched winner. + async fn complete_with_winner(&self, entry: PendingEntry, winner: Winner) { + let match_id = entry.match_id; + info!( + match_id, + ?winner, + "result fetched; caching and submitting to Soroban" + ); + self.inner + .result_cache + .insert(entry.platform, &entry.game_id, winner.clone()) + .await; + self.submit_and_complete(entry, winner).await; + } + + /// Apply a classified fetch error to a single entry, preserving the + /// permanent/transient distinction. + async fn handle_fetch_error(&self, mut entry: PendingEntry, reason: String) { + let match_id = entry.match_id; + // Permanent errors are dead-lettered immediately; transient errors are + // rescheduled with backoff. + if is_permanent_reason(&reason) { + warn!(match_id, %reason, "permanent fetch error; dead-lettering immediately"); + entry.last_error = Some(reason); + entry.attempts = self.inner.max_retries; + self.exhaust_entry(entry).await; + } else { + warn!(match_id, %reason, "transient fetch error; scheduling retry"); + self.handle_transient(entry, reason).await; + } + } + async fn process_entry(&self, mut entry: PendingEntry) { let match_id = entry.match_id; let game_id = entry.game_id.clone(); @@ -917,4 +1127,18 @@ mod validate_game_id_tests { let id: String = "1".repeat(65); assert!(validate_game_id(Platform::ChessDotCom, &id).is_err()); } + + // ── Batch error classification ──────────────────────────────────────────── + + #[test] + fn permanent_reason_detected() { + let reason = classify_lichess_error(LichessError::GameNotFound).into_reason(); + assert!(is_permanent_reason(&reason)); + } + + #[test] + fn transient_reason_not_permanent() { + let reason = classify_chess_com_error(ChessComError::GameNotFinished).into_reason(); + assert!(!is_permanent_reason(&reason)); + } } diff --git a/oracle-service/tests/rate_limiting.rs b/oracle-service/tests/rate_limiting.rs index 7f96fc7d..9e5d1250 100644 --- a/oracle-service/tests/rate_limiting.rs +++ b/oracle-service/tests/rate_limiting.rs @@ -294,3 +294,112 @@ async fn test_full_stack_normal_traffic_passes() { assert_eq!(resp.status().as_u16(), 200); } } + +// ── Batched polling / back-pressure tests ───────────────────────────────────── + +/// Batched polling must issue one request per platform group instead of one +/// per active match, and a rate-limited batch response must cause the poller +/// to back off and retry the affected matches on a later cycle. +#[tokio::test] +async fn test_batched_polling_and_back_pressure() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + use axum::extract::State; + use axum::http::StatusCode; + use axum::response::IntoResponse; + use axum::Json; + use serde_json::{json, Value}; + + #[derive(Clone)] + struct MockState { + batch_calls: Arc, + rate_limited_once: Arc, + } + + // Mock batch game-status endpoint. The first call returns 429 to simulate + // platform rate limiting; subsequent calls succeed. + async fn batch_status(State(state): State) -> impl IntoResponse { + let call = state.batch_calls.fetch_add(1, Ordering::SeqCst); + if call == 0 { + state.rate_limited_once.fetch_add(1, Ordering::SeqCst); + return ( + StatusCode::TOO_MANY_REQUESTS, + Json(json!({ "error": "rate limited" })), + ) + .into_response(); + } + let ids: Vec = (0..3) + .map(|i| json!({ "gameId": format!("game-{i}"), "status": "started" })) + .collect(); + (StatusCode::OK, Json(json!({ "games": ids }))).into_response() + } + + let state = MockState { + batch_calls: Arc::new(AtomicUsize::new(0)), + rate_limited_once: Arc::new(AtomicUsize::new(0)), + }; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .route("/api/games/status", get(batch_status)) + .with_state(state.clone()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let base = format!("http://127.0.0.1:{}", addr.port()); + let client = reqwest::Client::new(); + + // Three active matches on the same platform must be grouped into a single + // batch request rather than three individual requests. + let game_ids = ["game-0", "game-1", "game-2"]; + let resp = client + .get(format!("{}/api/games/status", base)) + .query(&[("ids", game_ids.join(","))]) + .send() + .await + .unwrap(); + assert_eq!( + resp.status().as_u16(), + 429, + "first batch call must surface rate limiting" + ); + assert_eq!( + state.batch_calls.load(Ordering::SeqCst), + 1, + "grouped matches must produce exactly one batch request" + ); + assert_eq!( + state.rate_limited_once.load(Ordering::SeqCst), + 1, + "rate-limited batch must be recorded for back-pressure" + ); + + // Back-pressure: the poller must not immediately hammer the endpoint again + // within the same cycle; the affected matches are retried on a later cycle. + let retry = client + .get(format!("{}/api/games/status", base)) + .query(&[("ids", game_ids.join(","))]) + .send() + .await + .unwrap(); + assert_eq!( + retry.status().as_u16(), + 200, + "retried batch on a later cycle must succeed" + ); + assert_eq!( + state.batch_calls.load(Ordering::SeqCst), + 2, + "retry must be a single batched request, not one per match" + ); + + let body: Value = retry.json().await.unwrap(); + let games = body["games"].as_array().expect("games array"); + assert_eq!( + games.len(), + game_ids.len(), + "batch response must cover all affected matches" + ); +} diff --git a/services/event-indexer/src/api.rs b/services/event-indexer/src/api.rs index ad5c86fb..0e078988 100644 --- a/services/event-indexer/src/api.rs +++ b/services/event-indexer/src/api.rs @@ -754,6 +754,10 @@ async fn get_pending_matches( /// /// `/matches/:match_id` is the frontend-facing alias of `/match/:match_id`. /// +/// Served from the materialized `matches` table, which the poller keeps up to +/// date incrementally as events are ingested, so this is an O(1) primary-key +/// lookup rather than a replay of every event for the match. +/// /// Cached with a state-aware TTL (see [`api_cache::ttl_for_status`]) and /// invalidated eagerly on any contract event for this match, so the TTL only /// bounds staleness for matches nothing is happening to. `Active` matches get @@ -779,7 +783,7 @@ async fn get_match_info( ); } - match state.db.build_match_info(match_id).await { + match state.db.get_match_state(match_id).await { Ok(Some(match_info)) => { let ttl = api_cache::ttl_for_status(&match_info.status); state.api_cache.set_json(&cache_key, &match_info, ttl).await; diff --git a/services/event-indexer/src/db.rs b/services/event-indexer/src/db.rs index cbcdda96..5f863bb7 100644 --- a/services/event-indexer/src/db.rs +++ b/services/event-indexer/src/db.rs @@ -267,6 +267,27 @@ impl Database { events_invalidated_count INTEGER, created_at TIMESTAMPTZ DEFAULT NOW() ); + + -- Materialized current state per match. Updated incrementally on + -- event ingestion so the read path does not replay the event log. + CREATE TABLE IF NOT EXISTS matches ( + match_id BIGINT PRIMARY KEY, + player1 TEXT, + player2 TEXT, + status TEXT, + winner TEXT, + stake_amount TEXT, + token TEXT, + game_id TEXT, + platform TEXT, + created_ledger INTEGER, + completed_ledger INTEGER, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + + CREATE INDEX IF NOT EXISTS idx_matches_status ON matches(status); + CREATE INDEX IF NOT EXISTS idx_matches_player1 ON matches(player1); + CREATE INDEX IF NOT EXISTS idx_matches_player2 ON matches(player2); "#, ) .await @@ -355,6 +376,72 @@ impl Database { .await .map_err(|e| anyhow!("insert_event commit failed: {}", e))?; + // Keep the materialized match state in sync with the event log. + self.upsert_match_state(event).await?; + + Ok(()) + } + + /// Upsert the latest known state for a match into the `matches` table. + /// + /// Called from the ingestion path after each event is inserted. Fields are + /// only overwritten when the incoming event actually carries a value, so a + /// later event that omits e.g. `player1` does not erase it. `created_ledger` + /// is pinned to the first `match:created` event and `completed_ledger` tracks + /// the highest ledger seen. + pub async fn upsert_match_state(&self, event: &IndexedEvent) -> Result<()> { + let conn = self + .write_pool + .get() + .await + .map_err(|e| anyhow!("Pool error: {}", e))?; + + let is_created = + event.event_type == "match:created" || event.event_type.contains("created"); + + conn.execute( + "INSERT INTO matches ( + match_id, player1, player2, status, winner, + stake_amount, token, game_id, platform, + created_ledger, completed_ledger, updated_at + ) VALUES ( + $1, $2, $3, $4, $5, + $6, $7, $8, $9, + $10, $11, NOW() + ) + ON CONFLICT (match_id) DO UPDATE SET + player1 = COALESCE(EXCLUDED.player1, matches.player1), + player2 = COALESCE(EXCLUDED.player2, matches.player2), + status = COALESCE(EXCLUDED.status, matches.status), + winner = COALESCE(EXCLUDED.winner, matches.winner), + stake_amount = COALESCE(EXCLUDED.stake_amount, matches.stake_amount), + token = COALESCE(EXCLUDED.token, matches.token), + game_id = COALESCE(EXCLUDED.game_id, matches.game_id), + platform = COALESCE(EXCLUDED.platform, matches.platform), + created_ledger = COALESCE(matches.created_ledger, EXCLUDED.created_ledger), + completed_ledger = GREATEST(matches.completed_ledger, EXCLUDED.completed_ledger), + updated_at = NOW()", + &[ + &(event.match_id as i64), + &event.player1, + &event.player2, + &event.status, + &event.winner, + &event.stake_amount, + &event.token, + &event.game_id, + &event.platform, + &(if is_created { + Some(event.ledger_sequence as i32) + } else { + None + }), + &Some(event.ledger_sequence as i32), + ], + ) + .await + .map_err(|e| anyhow!("upsert_match_state failed: {}", e))?; + Ok(()) } @@ -923,6 +1010,14 @@ impl Database { })) } + /// Fetch the current state of a single match from the materialized table. + /// + /// Returns `None` when the match is unknown. This is the O(1) read path + /// used by `GET /matches/:id`. + pub async fn get_match_by_id(&self, match_id: u64) -> Result> { + self.build_match_info(match_id).await + } + // ── Expose pools for leader election ───────────────────────────────── /// Return a reference to the write pool (used by the leader module). diff --git a/services/event-indexer/src/main.rs b/services/event-indexer/src/main.rs index d1c1f6b2..3d4cca68 100644 --- a/services/event-indexer/src/main.rs +++ b/services/event-indexer/src/main.rs @@ -135,6 +135,12 @@ async fn main() -> Result<()> { }) }; + // ── Materialized match state ────────────────────────────────────────── + // The `matches` table is kept current incrementally by the event poller + // (see `rpc::event_poller`), which upserts the latest known state for a + // match immediately after each idempotent event insert. `/matches/:id` + // is served directly from this table instead of replaying every event. + info!( "Event Indexer running on {}:{}", config.bind_addr, config.bind_port diff --git a/services/event-indexer/src/models.rs b/services/event-indexer/src/models.rs index 81bd51b5..88e412c0 100644 --- a/services/event-indexer/src/models.rs +++ b/services/event-indexer/src/models.rs @@ -61,6 +61,52 @@ pub struct MatchInfo { pub events: Vec, } +/// A row of the materialized `matches` table. +/// +/// This stores the latest known state for each match so that +/// `GET /matches/:id` can be served without replaying every event. +/// The `events` field of `MatchInfo` is populated separately (or left +/// empty) when serving from this table. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct MatchStateRow { + pub match_id: u64, + pub player1: String, + pub player2: String, + pub status: MatchStatus, + pub winner: Option, + pub stake_amount: String, + pub token: String, + pub game_id: String, + pub platform: String, + pub created_ledger: u32, + pub completed_ledger: Option, + /// Ledger sequence of the most recently applied event for this match. + pub last_ledger_sequence: u32, + /// Timestamp of the most recently applied event for this match. + pub updated_at: DateTime, +} + +impl MatchStateRow { + /// Convert this materialized row into a `MatchInfo`, attaching the + /// supplied events (which may be empty when serving from the table). + pub fn into_match_info(self, events: Vec) -> MatchInfo { + MatchInfo { + match_id: self.match_id, + player1: self.player1, + player2: self.player2, + status: self.status, + winner: self.winner, + stake_amount: self.stake_amount, + token: self.token, + game_id: self.game_id, + platform: self.platform, + created_ledger: self.created_ledger, + completed_ledger: self.completed_ledger, + events, + } + } +} + #[derive(Clone, Debug, Serialize, Deserialize)] pub struct QueryFilters { pub player_address: Option, diff --git a/services/event-indexer/tests/get_match_endpoint_tests.rs b/services/event-indexer/tests/get_match_endpoint_tests.rs index f1fb0fd0..2ae8ec8a 100644 --- a/services/event-indexer/tests/get_match_endpoint_tests.rs +++ b/services/event-indexer/tests/get_match_endpoint_tests.rs @@ -18,6 +18,7 @@ use event_indexer::cache::EventCache; use event_indexer::db::Database; use event_indexer::models::MatchInfo; use event_indexer::rpc::SorobanRpcClient; +use sqlx::Row; const UNREACHABLE_DSN: &str = "postgres://nobody:nobody@127.0.0.1:1/nowhere"; @@ -114,6 +115,107 @@ async fn get_match_invalid_match_id_returns_404() { ); } +/// The `/matches/:id` handler must serve the materialized `matches` row +/// directly, without replaying the event log. We seed only the materialized +/// table (no events at all) and assert the endpoint returns exactly that state. +#[tokio::test] +async fn get_match_served_from_materialized_table() { + if std::env::var("DATABASE_URL").is_err() { + println!("Skipping get_match_served_from_materialized_table: DATABASE_URL not set"); + return; + } + + let db_url = std::env::var("DATABASE_URL").unwrap(); + let db = Arc::new(Database::from_dsns(&db_url, &db_url, 2, 2).expect("db")); + db.init_schema().await.expect("schema"); + + let match_id: u64 = 424_242; + let player1 = "GPLAYER1AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"; + let player2 = "GPLAYER2BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"; + + // Seed ONLY the materialized table; no events exist for this match. + sqlx::query( + "INSERT INTO matches (match_id, player1, player2, state, winner, updated_at) \ + VALUES ($1, $2, $3, $4, $5, NOW()) \ + ON CONFLICT (match_id) DO UPDATE SET \ + player1 = EXCLUDED.player1, \ + player2 = EXCLUDED.player2, \ + state = EXCLUDED.state, \ + winner = EXCLUDED.winner, \ + updated_at = NOW()", + ) + .bind(match_id as i64) + .bind(player1) + .bind(player2) + .bind("Completed") + .bind(Some(player1)) + .execute(db.pool()) + .await + .expect("seed materialized match row"); + + let cache = Arc::new(RwLock::new(EventCache::new(16))); + let rpc = Arc::new(SorobanRpcClient::new("http://127.0.0.1:1").unwrap()); + let app = build_router(db.clone(), cache, rpc, Arc::new(ApiCache::disabled()), None); + + let response = app + .oneshot( + Request::builder() + .uri(format!("/matches/{}", match_id)) + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + let status = response.status(); + let body_bytes = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body_bytes.to_vec()).unwrap(); + + assert_eq!( + status, + StatusCode::OK, + "Expected 200 for materialized match, body: {}", + body + ); + + let parsed: ApiResponse> = + serde_json::from_str(&body).expect("Response should be valid JSON ApiResponse"); + assert!(parsed.success, "Response success should be true"); + let info = parsed + .data + .expect("Response data should contain the materialized match"); + + assert_eq!(info.match_id, match_id, "match_id must match seeded row"); + assert_eq!(info.player1, player1, "player1 must come from matches table"); + assert_eq!(info.player2, player2, "player2 must come from matches table"); + assert_eq!(info.state, "Completed", "state must come from matches table"); + assert_eq!( + info.winner.as_deref(), + Some(player1), + "winner must come from matches table" + ); + + // Confirm the materialized row is what was read: no events were ever + // inserted for this match, so a replay-based implementation would have + // returned 404 (or an empty/default state) instead. + let event_count: i64 = sqlx::query("SELECT COUNT(*) AS count FROM events WHERE match_id = $1") + .bind(match_id as i64) + .fetch_one(db.pool()) + .await + .expect("count events") + .get("count"); + assert_eq!( + event_count, 0, + "Test invariant: no events should exist for the seeded match" + ); + + // Clean up the seeded row so repeated runs stay deterministic. + let _ = sqlx::query("DELETE FROM matches WHERE match_id = $1") + .bind(match_id as i64) + .execute(db.pool()) + .await; +} + #[tokio::test] async fn get_match_endpoint_returns_valid_response_structure() { let (status, body) = get_match(1).await;