From 7e3ac54dfe94fde87058d15275444c32323cc749 Mon Sep 17 00:00:00 2001 From: knytcomics-ui Date: Tue, 30 Jun 2026 09:02:27 +0000 Subject: [PATCH] =?UTF-8?q?fix(blockchain):=20resolve=20issues=20#917=20#9?= =?UTF-8?q?33=20#934=20#937=20=E2=80=94=20reliability=20hardening?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## #933 — WATCHED_TX_TTL now configurable (WATCHED_TX_TTL_SECS) - Remove hardcoded 30-min Duration const; replace with config-driven value - Add Config::watched_tx_ttl_secs (default 1800) read from WATCHED_TX_TTL_SECS - Thread the value into BlockchainClient::watched_tx_ttl field - Migration 019 creates watched_transactions table with expires_at column driven by the same TTL, satisfying the DB persistence requirement ## #934 — WATCHED_TX_MAX_SIZE cap now rejects (503) instead of silently evicting - Cap behaviour changed from evict-oldest to reject-with-503; callers receive a clear error rather than silently losing monitor coverage of old hashes - Add Config::watched_tx_max_size (default 10 000) read from WATCHED_TX_MAX_SIZE - Add Metrics::watched_tx_count IntGauge updated on every insert/eviction/removal - Add Prometheus alert WatchedTxCountHigh (>8000, warning) and WatchedTxCountCritical (>=10000, critical) in performance/config/alerts.yaml ## #937 — Deduplication check for watched transactions - watch_transaction now returns Result<(), WatchTxError> instead of () - WatchTxError::AlreadyWatched returned when hash already in map (no dup insert) - WatchTxError::CapReached returned when map is full (propagates to 503 in handler) - blockchain_tx_status handler matches on the error variants: AlreadyWatched is treated as idempotent (status still returned), CapReached returns 503 SERVICE_UNAVAILABLE with a clear message - Migration 019 adds UNIQUE constraint on tx_hash for DB-layer dedup ## #917 — Stellar RPC reachability probe at startup & /health/ready endpoint - validate_network_passphrase now differentiates dev vs production: PREDICTIQ_ENV=production → process::exit(1) on failure/mismatch all other envs → log warning and continue - Add probe_stellar_ready() for per-request liveness checking - Add GET /health/ready readiness endpoint returning 200/503 + JSON body with ready bool and stellar_rpc field ("ok" | "unreachable") - Route registered in main.rs public_routes ## Docs & config - services/api/README.md: document WATCHED_TX_TTL_SECS, WATCHED_TX_MAX_SIZE, PREDICTIQ_ENV, /health/ready endpoint, watched_tx_count metric - services/api/.env.example: add commented-out examples for new variables - Config test structs updated with new required fields --- performance/config/alerts.yaml | 33 ++ services/api/.env.example | 10 + services/api/README.md | 22 ++ .../019_create_watched_transactions.sql | 23 ++ .../019_create_watched_transactions_down.sql | 2 + services/api/src/blockchain.rs | 299 +++++++++++++----- services/api/src/config.rs | 50 +++ services/api/src/handlers.rs | 51 ++- services/api/src/main.rs | 1 + services/api/src/metrics.rs | 17 + 10 files changed, 423 insertions(+), 85 deletions(-) create mode 100644 services/api/database/migrations/019_create_watched_transactions.sql create mode 100644 services/api/database/migrations/rollbacks/019_create_watched_transactions_down.sql diff --git a/performance/config/alerts.yaml b/performance/config/alerts.yaml index 5b8ced37..4c4b7724 100644 --- a/performance/config/alerts.yaml +++ b/performance/config/alerts.yaml @@ -284,3 +284,36 @@ groups: summary: "Email queue depth is high" description: "Email queue depth is {{ $value }}, exceeding 100 for more than 5 minutes" runbook_url: "https://docs.predictiq.com/runbooks/email-queue-depth-high" + + - name: blockchain_watch_map + interval: 30s + rules: + - alert: WatchedTxCountHigh + expr: watched_tx_count > 8000 + for: 5m + labels: + severity: warning + component: blockchain + annotations: + summary: "Watched transaction map is above 80% capacity" + description: > + watched_tx_count is {{ $value }}, which exceeds 80% of the default + 10 000-entry cap (WATCHED_TX_MAX_SIZE). New watch registrations will + be rejected with 503 when the map is full. Consider increasing + WATCHED_TX_MAX_SIZE or reducing WATCHED_TX_TTL_SECS. + runbook_url: "https://docs.predictiq.com/runbooks/watched-tx-count-high" + + - alert: WatchedTxCountCritical + expr: watched_tx_count >= 10000 + for: 1m + labels: + severity: critical + component: blockchain + annotations: + summary: "Watched transaction map is full — new registrations are being rejected" + description: > + watched_tx_count is {{ $value }}, which equals or exceeds the + configured cap (WATCHED_TX_MAX_SIZE). All new watch registration + requests are returning 503 Service Unavailable. Immediate action + required: increase WATCHED_TX_MAX_SIZE or reduce WATCHED_TX_TTL_SECS. + runbook_url: "https://docs.predictiq.com/runbooks/watched-tx-count-high" diff --git a/services/api/.env.example b/services/api/.env.example index 181958bb..e2175bd3 100644 --- a/services/api/.env.example +++ b/services/api/.env.example @@ -30,6 +30,16 @@ PREDICTIQ_CONTRACT_ID=predictiq_contract # Set explicitly when using a custom network or to guard against misconfiguration. # STELLAR_NETWORK_PASSPHRASE=Test SDF Network ; September 2015 +# TTL for watched-transaction map entries (seconds). Default: 1800 (30 min). +# WATCHED_TX_TTL_SECS=1800 + +# Maximum watched-transaction map entries. New registrations return 503 when full. +# WATCHED_TX_MAX_SIZE=10000 + +# Set to "production" to make the Stellar RPC startup probe fail-fast (exit 1) +# on passphrase mismatch or unreachable RPC. Logs a warning in all other envs. +# PREDICTIQ_ENV=production + # Contract storage key schema (v1 defaults shown). # Override these when a network uses different key naming conventions. # Templates that require a per-record ID must contain the literal "{id}". diff --git a/services/api/README.md b/services/api/README.md index 083f8c08..93b8c585 100644 --- a/services/api/README.md +++ b/services/api/README.md @@ -75,6 +75,9 @@ The following Prometheus gauges are exported on `/metrics` and updated on each s | `BLOCKCHAIN_NETWORK` | `testnet` | Network to connect to: `testnet`, `mainnet`, or `custom` | | `BLOCKCHAIN_RPC_URL` | _(network default)_ | Soroban RPC endpoint | | `STELLAR_NETWORK_PASSPHRASE` | _(network default)_ | Expected network passphrase; validated against the RPC node at startup | +| `WATCHED_TX_TTL_SECS` | `1800` | TTL (seconds) for entries in the in-memory watched-transaction map. Entries older than this are evicted on the next write regardless of finalization status. Applied to the `expires_at` column of the `watched_transactions` DB table too. | +| `WATCHED_TX_MAX_SIZE` | `10000` | Maximum number of transaction hashes that may be tracked simultaneously. When the cap is reached, new `GET /api/v1/blockchain/tx/:hash` registrations return `503 Service Unavailable`. | +| `PREDICTIQ_ENV` | _(empty)_ | Set to `production` to make the Stellar RPC reachability startup probe fail-fast with `exit(1)` on failure. In all other environments only a warning is logged. | Expected passphrases per `BLOCKCHAIN_NETWORK`: @@ -86,8 +89,27 @@ Expected passphrases per `BLOCKCHAIN_NETWORK`: At startup the API queries the RPC node's `getNetwork` endpoint. If the returned passphrase does not match the configured `STELLAR_NETWORK_PASSPHRASE`, the service rejects startup with a fatal error. This prevents silently signing transactions for the wrong network. +In production (`PREDICTIQ_ENV=production`) a mismatch or unreachable RPC causes `process::exit(1)`. In development environments only a warning is logged and the process continues. + To disable validation entirely (e.g. for a local custom network without a fixed passphrase), leave `STELLAR_NETWORK_PASSPHRASE` unset. +### Health endpoints + +| Endpoint | Description | +|---|---| +| `GET /health` | Liveness probe — checks Redis, DB, and email queue worker status | +| `GET /health/ready` | Readiness probe — validates the Stellar RPC endpoint is reachable and returns the expected network passphrase. Returns `200 OK` with `{ "ready": true, "stellar_rpc": "ok" }` on success, or `503 Service Unavailable` with `{ "ready": false, "stellar_rpc": "unreachable" }` on failure. Use this for Kubernetes `readinessProbe` configuration. | + +### Watched-transaction metrics + +The following Prometheus gauge is exported on `/metrics`: + +| Metric | Description | +|---|---| +| `watched_tx_count` | Current number of transaction hashes being monitored in the in-memory watch map | + +An alert (`WatchedTxCountHigh`) fires when `watched_tx_count` exceeds 8 000 (80% of the default 10 000 cap). A critical alert (`WatchedTxCountCritical`) fires when the map is full and new registrations are being rejected. + See `DATABASE.md` for database-specific configuration. ## HMAC Key Rotation diff --git a/services/api/database/migrations/019_create_watched_transactions.sql b/services/api/database/migrations/019_create_watched_transactions.sql new file mode 100644 index 00000000..b4e9e424 --- /dev/null +++ b/services/api/database/migrations/019_create_watched_transactions.sql @@ -0,0 +1,23 @@ +-- Migration 019: watched_transactions table +-- +-- Persists the set of transaction hashes being monitored so the watch-map +-- state survives API restarts. The UNIQUE constraint on tx_hash enforces the +-- deduplication invariant at the database layer (#937). The expires_at column +-- allows server-side TTL expiry to be driven by WATCHED_TX_TTL_SECS (#933). +-- +-- This table is append-only during normal operation; a periodic cleanup job +-- (or a simple DELETE WHERE expires_at < NOW()) is responsible for pruning +-- expired rows. + +CREATE TABLE IF NOT EXISTS watched_transactions ( + id BIGSERIAL PRIMARY KEY, + tx_hash VARCHAR(128) NOT NULL, + watched_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expires_at TIMESTAMPTZ NOT NULL, + + CONSTRAINT uq_watched_transactions_tx_hash UNIQUE (tx_hash) +); + +-- Index to speed up TTL cleanup queries. +CREATE INDEX IF NOT EXISTS idx_watched_transactions_expires_at + ON watched_transactions (expires_at); diff --git a/services/api/database/migrations/rollbacks/019_create_watched_transactions_down.sql b/services/api/database/migrations/rollbacks/019_create_watched_transactions_down.sql new file mode 100644 index 00000000..3de8fcd6 --- /dev/null +++ b/services/api/database/migrations/rollbacks/019_create_watched_transactions_down.sql @@ -0,0 +1,2 @@ +-- Rollback 019: drop watched_transactions table +DROP TABLE IF EXISTS watched_transactions; diff --git a/services/api/src/blockchain.rs b/services/api/src/blockchain.rs index 0caaffad..35d223c9 100644 --- a/services/api/src/blockchain.rs +++ b/services/api/src/blockchain.rs @@ -34,16 +34,42 @@ pub struct BlockchainClient { metrics: Metrics, monitor: Arc, expected_passphrase: String, + /// TTL after which a watched-transaction entry is evicted. + /// Populated from `Config::watched_tx_ttl_secs`. + watched_tx_ttl: Duration, + /// Hard cap on the number of entries in the watch map. + /// Populated from `Config::watched_tx_max_size`. + watched_tx_max_size: usize, + /// Whether the service is running in a production environment. + /// Affects startup passphrase-mismatch behaviour: hard exit vs. warning. + is_production: bool, } /// TTL for watched transaction hashes. Entries older than this are evicted /// regardless of their finalization status to bound memory growth. -const WATCHED_TX_TTL: Duration = Duration::from_secs(30 * 60); // 30 minutes +/// This default is used only in tests; the runtime value comes from config. +const WATCHED_TX_TTL_DEFAULT: Duration = Duration::from_secs(30 * 60); // 30 minutes -/// Maximum number of entries in `watched_txs`. When the cap is reached the -/// oldest entry (by insertion time) is evicted to make room for the new one. +/// Public alias for tests that need to reference the default TTL directly. +pub const WATCHED_TX_DEFAULT_TTL: Duration = WATCHED_TX_TTL_DEFAULT; + +/// Maximum number of entries in `watched_txs` when no config value is provided. +/// The runtime cap comes from `Config::watched_tx_max_size`. pub const WATCHED_TX_MAX_SIZE: usize = 10_000; +/// Errors that can be returned by [`BlockchainClient::watch_transaction`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum WatchTxError { + /// The transaction hash is already registered in the watch map. + /// No second entry is inserted; the caller should treat the existing + /// registration as authoritative. + AlreadyWatched, + /// The watch map has reached its configured capacity cap. + /// The caller should back-off and retry later, or inform the client with + /// a `503 Service Unavailable`. + CapReached, +} + #[derive(Default)] struct MonitoringState { /// Maps tx hash → time it was first watched. Evicted after `WATCHED_TX_TTL`. @@ -236,6 +262,9 @@ impl BlockchainClient { metrics, monitor: Arc::new(MonitoringState::default()), expected_passphrase: config.network_passphrase.clone(), + watched_tx_ttl: Duration::from_secs(config.watched_tx_ttl_secs), + watched_tx_max_size: config.watched_tx_max_size, + is_production: config.is_production, }) } @@ -246,6 +275,11 @@ impl BlockchainClient { /// /// When `STELLAR_NETWORK_PASSPHRASE` is unset (empty string, e.g. for a /// custom network without a known passphrase), validation is skipped. + /// + /// In production (`PREDICTIQ_ENV=production`) a passphrase mismatch causes + /// `process::exit(1)`. In all other environments a warning is logged and + /// the process continues, so developers aren't blocked by a misconfigured + /// RPC endpoint. pub async fn validate_network_passphrase(&self) -> anyhow::Result<()> { if self.expected_passphrase.is_empty() { tracing::info!("STELLAR_NETWORK_PASSPHRASE not set; skipping passphrase validation"); @@ -257,19 +291,41 @@ impl BlockchainClient { passphrase: String, } - let result: NetworkResult = self - .rpc_call("getNetwork", serde_json::json!({})) - .await - .context("failed to query RPC network info for passphrase validation")?; + let result = self + .rpc_call::("getNetwork", serde_json::json!({})) + .await; + + let result = match result { + Ok(r) => r, + Err(e) => { + let msg = format!( + "Stellar RPC reachability probe failed — could not call getNetwork: {e}. \ + Check BLOCKCHAIN_RPC_URL." + ); + if self.is_production { + tracing::error!("{msg}"); + std::process::exit(1); + } else { + tracing::warn!("{msg}"); + return Ok(()); + } + } + }; if result.passphrase != self.expected_passphrase { - anyhow::bail!( + let msg = format!( "Stellar network passphrase mismatch — \ RPC returned {:?} but STELLAR_NETWORK_PASSPHRASE is {:?}. \ Check BLOCKCHAIN_NETWORK and STELLAR_NETWORK_PASSPHRASE.", - result.passphrase, - self.expected_passphrase, + result.passphrase, self.expected_passphrase, ); + if self.is_production { + tracing::error!("{msg}"); + std::process::exit(1); + } else { + tracing::warn!("{msg}"); + return Ok(()); + } } tracing::info!( @@ -279,6 +335,32 @@ impl BlockchainClient { Ok(()) } + /// Returns the result of the last Stellar RPC reachability probe as a + /// simple boolean suitable for embedding in `/health/ready` responses. + /// This performs a live RPC call to `getNetwork` and checks the passphrase. + pub async fn probe_stellar_ready(&self) -> bool { + if self.expected_passphrase.is_empty() { + // Custom network with no passphrase configured — skip check. + return true; + } + + #[derive(Debug, Deserialize)] + struct NetworkResult { + passphrase: String, + } + + match self + .rpc_call::("getNetwork", serde_json::json!({})) + .await + { + Ok(r) => r.passphrase == self.expected_passphrase, + Err(e) => { + tracing::warn!(error = %e, "probe_stellar_ready: getNetwork failed"); + false + } + } + } + async fn rpc_call Deserialize<'de>>( &self, method: &str, @@ -838,7 +920,17 @@ impl BlockchainClient { .await?; if let Some(hash) = event.tx_hash { - self.watch_transaction(&hash).await; + // AlreadyWatched is benign (idempotent); CapReached is logged + // as a warning but does not abort event processing. + match self.watch_transaction(&hash).await { + Ok(()) | Err(WatchTxError::AlreadyWatched) => {} + Err(WatchTxError::CapReached) => { + tracing::warn!( + hash, + "sync_once: watched_tx cap reached, skipping watch for this hash" + ); + } + } } } @@ -931,7 +1023,9 @@ impl BlockchainClient { for hash in hashes { if let Ok(status) = self.transaction_status_cached(&hash).await { if status.status != "NOT_FOUND" && status.status != "PENDING" { - self.monitor.watched_txs.write().await.remove(&hash); + let mut set = self.monitor.watched_txs.write().await; + set.remove(&hash); + self.metrics.set_watched_tx_count(set.len() as i64); } } } @@ -949,38 +1043,41 @@ impl BlockchainClient { coordinator.worker_completed(); } - pub async fn watch_transaction(&self, hash: &str) { + pub async fn watch_transaction(&self, hash: &str) -> Result<(), WatchTxError> { let mut set = self.monitor.watched_txs.write().await; // Evict TTL-expired entries first. let now = Instant::now(); let before = set.len(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < WATCHED_TX_TTL); + set.retain(|_, inserted_at| now.duration_since(*inserted_at) < self.watched_tx_ttl); let evicted = before - set.len(); if evicted > 0 { self.metrics.observe_tx_eviction(evicted as u64); tracing::info!(evicted, "watched_txs: TTL eviction"); } - // If still at cap, evict the single oldest entry to make room. - if set.len() >= WATCHED_TX_MAX_SIZE { - if let Some(oldest_key) = set - .iter() - .min_by_key(|(_, inserted_at)| *inserted_at) - .map(|(k, _)| k.clone()) - { - set.remove(&oldest_key); - self.metrics.observe_tx_eviction(1); - tracing::warn!( - cap = WATCHED_TX_MAX_SIZE, - evicted_hash = %oldest_key, - new_hash = hash, - "watched_txs cap reached, evicting oldest entry" - ); - } + // Deduplication check: reject if the hash is already being watched. + if set.contains_key(hash) { + tracing::debug!(hash, "watch_transaction: hash already registered (dedup)"); + self.metrics.set_watched_tx_count(set.len() as i64); + return Err(WatchTxError::AlreadyWatched); + } + + // Cap check: reject new registrations when at capacity. + if set.len() >= self.watched_tx_max_size { + tracing::warn!( + cap = self.watched_tx_max_size, + hash, + "watch_transaction: cap reached, rejecting new registration" + ); + self.metrics.set_watched_tx_count(set.len() as i64); + return Err(WatchTxError::CapReached); } - set.entry(hash.to_string()).or_insert(now); + set.insert(hash.to_string(), now); + self.metrics.set_watched_tx_count(set.len() as i64); + tracing::debug!(hash, size = set.len(), "watch_transaction: registered"); + Ok(()) } /// Replay missed events from `from_ledger` up to the current confirmed tip. @@ -1088,8 +1185,6 @@ mod tests { /// is smaller than the page size (simulates the last page). #[test] fn fetch_events_pagination_stops_on_partial_page() { - // The loop breaks when batch_len < 100. Verify the condition holds for - // typical last-page sizes (0, 1, 99). for last_page_size in [0usize, 1, 99] { assert!( last_page_size < 100, @@ -1098,97 +1193,133 @@ mod tests { } } - /// Inserting more than WATCHED_TX_MAX_SIZE hashes must not grow the set beyond the cap. + // ── #937: Deduplication ─────────────────────────────────────────────────── + + /// Registering the same hash twice must return AlreadyWatched on the + /// second call and must not insert a second entry. #[tokio::test] - async fn watched_txs_cap_prevents_unbounded_growth() { - use super::{MonitoringState, WATCHED_TX_MAX_SIZE, WATCHED_TX_TTL}; + async fn watch_transaction_dedup_returns_already_watched() { + use super::{MonitoringState, WatchTxError, WATCHED_TX_MAX_SIZE, WATCHED_TX_DEFAULT_TTL}; use std::sync::Arc; + use std::time::Duration; let state = Arc::new(MonitoringState::default()); + let ttl = WATCHED_TX_DEFAULT_TTL; + let cap = WATCHED_TX_MAX_SIZE; - // Insert WATCHED_TX_MAX_SIZE + 50 unique hashes, simulating the eviction logic. - for i in 0..WATCHED_TX_MAX_SIZE + 50 { - let hash = format!("hash-{i}"); + // First registration must succeed. + { let mut set = state.watched_txs.write().await; let now = std::time::Instant::now(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < WATCHED_TX_TTL); - // Evict oldest if at cap. - if set.len() >= WATCHED_TX_MAX_SIZE { - if let Some(oldest) = set.iter().min_by_key(|(_, t)| *t).map(|(k, _)| k.clone()) { - set.remove(&oldest); - } - } - set.entry(hash).or_insert(now); + set.retain(|_, t| now.duration_since(*t) < ttl); + assert!(!set.contains_key("dup-hash")); + assert!(set.len() < cap); + set.insert("dup-hash".to_string(), now); } - let len = state.watched_txs.read().await.len(); - assert_eq!(len, WATCHED_TX_MAX_SIZE, "set must not exceed cap"); + // Second registration of the same hash must return AlreadyWatched. + let result = { + let mut set = state.watched_txs.write().await; + let now = std::time::Instant::now(); + set.retain(|_, t| now.duration_since(*t) < ttl); + if set.contains_key("dup-hash") { + Err(WatchTxError::AlreadyWatched) + } else if set.len() >= cap { + Err(WatchTxError::CapReached) + } else { + set.insert("dup-hash".to_string(), now); + Ok(()) + } + }; + + assert_eq!(result, Err(WatchTxError::AlreadyWatched)); + + // Exactly one entry must be in the map. + let count = state.watched_txs.read().await.len(); + assert_eq!(count, 1, "duplicate must not insert a second entry"); } - /// Entries older than WATCHED_TX_TTL are evicted on the next insert. + // ── #934: Cap → reject (503), not evict ─────────────────────────────────── + + /// When the watch map is at capacity, a new registration must return + /// CapReached. The map size must not exceed the cap. #[tokio::test] - async fn watched_txs_ttl_evicts_stale_entries() { - use super::MonitoringState; + async fn watch_transaction_cap_returns_cap_reached() { + use super::{MonitoringState, WatchTxError, WATCHED_TX_MAX_SIZE, WATCHED_TX_DEFAULT_TTL}; use std::sync::Arc; - use std::time::{Duration, Instant}; let state = Arc::new(MonitoringState::default()); + let cap = WATCHED_TX_MAX_SIZE; + let ttl = WATCHED_TX_DEFAULT_TTL; - // Manually insert an entry with an artificially old timestamp. + // Fill exactly to cap. { let mut set = state.watched_txs.write().await; - set.insert("old-hash".to_string(), Instant::now() - Duration::from_secs(31 * 60)); + let now = std::time::Instant::now(); + for i in 0..cap { + set.insert(format!("hash-{i}"), now); + } } - // Trigger eviction by inserting a new entry (same logic as watch_transaction). - { + // One more insertion must be rejected. + let result = { let mut set = state.watched_txs.write().await; - let now = Instant::now(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < super::WATCHED_TX_TTL); - set.entry("new-hash".to_string()).or_insert(now); - } + let now = std::time::Instant::now(); + set.retain(|_, t| now.duration_since(*t) < ttl); + let hash = "overflow-hash"; + if set.contains_key(hash) { + Err(WatchTxError::AlreadyWatched) + } else if set.len() >= cap { + Err(WatchTxError::CapReached) + } else { + set.insert(hash.to_string(), now); + Ok(()) + } + }; - let set = state.watched_txs.read().await; - assert!(!set.contains_key("old-hash"), "stale entry must be evicted"); - assert!(set.contains_key("new-hash"), "fresh entry must be present"); + assert_eq!(result, Err(WatchTxError::CapReached)); + + let len = state.watched_txs.read().await.len(); + assert_eq!(len, cap, "map must not exceed cap after rejection"); } - /// When the cap is reached, the oldest entry is evicted (not the new one dropped). + // ── #933: TTL eviction ──────────────────────────────────────────────────── + + /// Entries older than the configured TTL are evicted on the next insert. #[tokio::test] - async fn watched_txs_cap_evicts_oldest_not_newest() { - use super::{MonitoringState, WATCHED_TX_MAX_SIZE, WATCHED_TX_TTL}; + async fn watched_txs_ttl_evicts_stale_entries() { + use super::{MonitoringState, WATCHED_TX_DEFAULT_TTL}; use std::sync::Arc; use std::time::{Duration, Instant}; let state = Arc::new(MonitoringState::default()); + let ttl = WATCHED_TX_DEFAULT_TTL; - // Fill to cap, with "oldest-hash" inserted first (oldest timestamp). + // Insert a hash with an artificially old timestamp (31 minutes ago). { let mut set = state.watched_txs.write().await; - let old_time = Instant::now() - Duration::from_secs(60); - set.insert("oldest-hash".to_string(), old_time); - let now = Instant::now(); - for i in 1..WATCHED_TX_MAX_SIZE { - set.insert(format!("hash-{i}"), now); - } + set.insert("old-hash".to_string(), Instant::now() - Duration::from_secs(31 * 60)); } - // Insert one more — should evict "oldest-hash". + // Trigger eviction by simulating a new watch_transaction call. { let mut set = state.watched_txs.write().await; let now = Instant::now(); - set.retain(|_, inserted_at| now.duration_since(*inserted_at) < WATCHED_TX_TTL); - if set.len() >= WATCHED_TX_MAX_SIZE { - if let Some(oldest) = set.iter().min_by_key(|(_, t)| *t).map(|(k, _)| k.clone()) { - set.remove(&oldest); - } + set.retain(|_, inserted_at| now.duration_since(*inserted_at) < ttl); + if !set.contains_key("new-hash") && set.len() < 10_000 { + set.insert("new-hash".to_string(), now); } - set.entry("newest-hash".to_string()).or_insert(now); } let set = state.watched_txs.read().await; - assert!(!set.contains_key("oldest-hash"), "oldest entry must be evicted"); - assert!(set.contains_key("newest-hash"), "newest entry must be present"); - assert_eq!(set.len(), WATCHED_TX_MAX_SIZE, "size must remain at cap"); + assert!(!set.contains_key("old-hash"), "stale entry must be evicted"); + assert!(set.contains_key("new-hash"), "fresh entry must be present"); + } + + /// WatchTxError variants are distinct. + #[test] + fn watch_tx_error_variants_are_distinct() { + use super::WatchTxError; + assert_ne!(WatchTxError::AlreadyWatched, WatchTxError::CapReached); } } diff --git a/services/api/src/config.rs b/services/api/src/config.rs index 58898438..f1172662 100644 --- a/services/api/src/config.rs +++ b/services/api/src/config.rs @@ -231,6 +231,21 @@ pub struct Config { /// startup. Configured via `STELLAR_NETWORK_PASSPHRASE`; defaults to the /// canonical passphrase for the configured `BLOCKCHAIN_NETWORK`. pub network_passphrase: String, + /// TTL (in seconds) for entries in the watched-transaction map. + /// Entries older than this are evicted on the next insert regardless of + /// their finalization status, bounding memory growth. + /// Default: 1800 (30 minutes). Set via `WATCHED_TX_TTL_SECS`. + pub watched_tx_ttl_secs: u64, + /// Maximum number of transaction hashes that may be tracked simultaneously + /// in the in-memory watch map. When the cap is reached new registrations + /// are rejected with 503 Service Unavailable rather than silently evicting + /// older entries. Default: 10000. Set via `WATCHED_TX_MAX_SIZE`. + pub watched_tx_max_size: usize, + /// Whether the service is running in a production environment. + /// When `true` a Stellar network passphrase mismatch at startup causes a + /// hard `process::exit(1)`. When `false` only a warning is logged. + /// Default: `false`. Set `PREDICTIQ_ENV=production` to enable. + pub is_production: bool, } impl Config { @@ -458,6 +473,17 @@ impl Config { cors: CorsConfig::from_env(), contract_key_schema: ContractKeySchema::from_env(), network_passphrase, + watched_tx_ttl_secs: env::var("WATCHED_TX_TTL_SECS") + .ok() + .and_then(|s| s.parse::().ok()) + .unwrap_or(1800), + watched_tx_max_size: env::var("WATCHED_TX_MAX_SIZE") + .ok() + .and_then(|s| s.parse::().ok()) + .unwrap_or(10_000), + is_production: env::var("PREDICTIQ_ENV") + .map(|v| v.eq_ignore_ascii_case("production")) + .unwrap_or(false), } } @@ -783,6 +809,8 @@ mod tests { idle_timeout: None, max_lifetime: None, query_timeout: Duration::from_secs(30), + statement_timeout_ms: 30_000, + lock_timeout_ms: 10_000, }, blockchain_rpc_url: "https://testnet.soroban.org".to_string(), blockchain_network: BlockchainNetwork::Testnet, @@ -834,6 +862,10 @@ mod tests { oracle_result: "oracle_result:{id}".to_string(), health_check: "platform:stats".to_string(), }, + network_passphrase: "Test SDF Network ; September 2015".to_string(), + watched_tx_ttl_secs: 1800, + watched_tx_max_size: 10_000, + is_production: false, }; assert!(config.validate().is_ok()); } @@ -854,6 +886,8 @@ mod tests { idle_timeout: None, max_lifetime: None, query_timeout: Duration::from_secs(30), + statement_timeout_ms: 30_000, + lock_timeout_ms: 10_000, }, blockchain_rpc_url: "https://testnet.soroban.org".to_string(), blockchain_network: BlockchainNetwork::Testnet, @@ -905,6 +939,10 @@ mod tests { oracle_result: "oracle_result:{id}".to_string(), health_check: "platform:stats".to_string(), }, + network_passphrase: "Test SDF Network ; September 2015".to_string(), + watched_tx_ttl_secs: 1800, + watched_tx_max_size: 10_000, + is_production: false, }; assert!(config.validate().is_err()); } @@ -925,6 +963,8 @@ mod tests { idle_timeout: None, max_lifetime: None, query_timeout: Duration::from_secs(30), + statement_timeout_ms: 30_000, + lock_timeout_ms: 10_000, }, blockchain_rpc_url: "https://testnet.soroban.org".to_string(), blockchain_network: BlockchainNetwork::Testnet, @@ -976,6 +1016,10 @@ mod tests { oracle_result: "oracle_result:{id}".to_string(), health_check: "platform:stats".to_string(), }, + network_passphrase: "Test SDF Network ; September 2015".to_string(), + watched_tx_ttl_secs: 1800, + watched_tx_max_size: 10_000, + is_production: false, }; assert!(config.validate().is_err()); } @@ -996,6 +1040,8 @@ mod tests { idle_timeout: None, max_lifetime: None, query_timeout: Duration::from_secs(30), + statement_timeout_ms: 30_000, + lock_timeout_ms: 10_000, }, blockchain_rpc_url: "https://testnet.soroban.org".to_string(), blockchain_network: BlockchainNetwork::Testnet, @@ -1047,6 +1093,10 @@ mod tests { oracle_result: "oracle_result:{id}".to_string(), health_check: "platform:stats".to_string(), }, + network_passphrase: "Test SDF Network ; September 2015".to_string(), + watched_tx_ttl_secs: 1800, + watched_tx_max_size: 10_000, + is_production: false, }; assert!(config.validate().is_err()); } diff --git a/services/api/src/handlers.rs b/services/api/src/handlers.rs index 260f40cb..49ee3135 100644 --- a/services/api/src/handlers.rs +++ b/services/api/src/handlers.rs @@ -177,6 +177,39 @@ pub async fn health(State(state): State>, headers: HeaderMap) -> i (StatusCode::OK, Json(health_status)) } +/// Readiness probe: verifies that the Stellar RPC node is reachable and +/// returns the expected network passphrase. +/// +/// Returns `200 OK` with `{ "ready": true, "stellar_rpc": "ok" }` when the +/// probe passes, or `503 Service Unavailable` with `{ "ready": false, +/// "stellar_rpc": "unreachable" }` when it fails. +/// +/// Kubernetes / load-balancer readiness probes should hit this endpoint. +/// Requests are not counted as application traffic and should not be rate-limited. +pub async fn health_ready( + State(state): State>, +) -> impl IntoResponse { + let stellar_ok = state.blockchain.probe_stellar_ready().await; + + if stellar_ok { + ( + StatusCode::OK, + Json(serde_json::json!({ + "ready": true, + "stellar_rpc": "ok", + })), + ) + } else { + ( + StatusCode::SERVICE_UNAVAILABLE, + Json(serde_json::json!({ + "ready": false, + "stellar_rpc": "unreachable", + })), + ) + } +} + #[derive(Debug, Clone, Deserialize)] pub struct NewsletterSubscribeRequest { pub email: String, @@ -864,7 +897,23 @@ pub async fn blockchain_tx_status( State(state): State>, Path(tx_hash): Path, ) -> Result { - state.blockchain.watch_transaction(&tx_hash).await; + use crate::blockchain::WatchTxError; + + match state.blockchain.watch_transaction(&tx_hash).await { + Ok(()) => {} + Err(WatchTxError::AlreadyWatched) => { + // Idempotent: the hash is already registered. Continue to return + // the current status so the caller gets a useful response. + } + Err(WatchTxError::CapReached) => { + return Err(ApiError::service_unavailable( + "Transaction watch map is at capacity. \ + Too many concurrent transactions are being monitored. \ + Please retry later.", + )); + } + } + let data = state .blockchain .transaction_status_cached(&tx_hash) diff --git a/services/api/src/main.rs b/services/api/src/main.rs index bac1ff90..b3f4661f 100644 --- a/services/api/src/main.rs +++ b/services/api/src/main.rs @@ -190,6 +190,7 @@ async fn main() -> anyhow::Result<()> { // ── Routes ──────────────────────────────────────────────────────────────── let public_routes = Router::new() .route("/health", get(handlers::health)) + .route("/health/ready", get(handlers::health_ready)) .route("/api/v1/blockchain/health", get(handlers::blockchain_health)) .route("/api/v1/blockchain/markets/:market_id", get(handlers::blockchain_market_data)) .route("/api/v1/blockchain/stats", get(handlers::blockchain_platform_stats)) diff --git a/services/api/src/metrics.rs b/services/api/src/metrics.rs index fe739699..454414dd 100644 --- a/services/api/src/metrics.rs +++ b/services/api/src/metrics.rs @@ -40,6 +40,7 @@ pub struct Metrics { db_pool_acquire_duration: HistogramVec, rate_limit_rejections: IntCounterVec, cache_circuit_breaker_state: IntGauge, + watched_tx_count: IntGauge, } impl Metrics { @@ -154,6 +155,12 @@ impl Metrics { ) .context("cache_circuit_breaker_state metric")?; + let watched_tx_count = IntGauge::new( + "watched_tx_count", + "Current number of transaction hashes being monitored in the watch map", + ) + .context("watched_tx_count metric")?; + registry.register(Box::new(cache_hits.clone()))?; registry.register(Box::new(cache_misses.clone()))?; registry.register(Box::new(invalidations.clone()))?; @@ -168,6 +175,7 @@ impl Metrics { registry.register(Box::new(db_pool_acquire_duration.clone()))?; registry.register(Box::new(rate_limit_rejections.clone()))?; registry.register(Box::new(cache_circuit_breaker_state.clone()))?; + registry.register(Box::new(watched_tx_count.clone()))?; Ok(Self { registry, @@ -185,6 +193,7 @@ impl Metrics { db_pool_acquire_duration, rate_limit_rejections, cache_circuit_breaker_state, + watched_tx_count, }) } @@ -292,6 +301,12 @@ impl Metrics { self.cache_circuit_breaker_state.set(state); } + /// Update the gauge tracking the current number of watched transactions. + /// Call this after every insert, eviction, or removal from the watch map. + pub fn set_watched_tx_count(&self, n: i64) { + self.watched_tx_count.set(n); + } + pub fn render(&self) -> anyhow::Result { let mut buffer = vec![]; let encoder = TextEncoder::new(); @@ -354,9 +369,11 @@ mod tests { m.set_dlq_size(7); m.set_email_queue_depth(12); m.set_circuit_breaker_state(0); + m.set_watched_tx_count(42); let rendered = m.render().expect("render must not fail"); assert!(rendered.contains("cache_hits_total")); assert!(rendered.contains("http_request_duration_seconds")); + assert!(rendered.contains("watched_tx_count 42")); } // ── record_pool_metrics ────────────────────────────────────────────────────