From 74c6445274794643b9aa8faac58d383273297975 Mon Sep 17 00:00:00 2001 From: Abd-Standard Date: Sat, 27 Jun 2026 12:07:01 +0000 Subject: [PATCH] feat: event schema registry and circuit-breaker resume policy MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #601 – EventSchemaRegistry in events.rs - Add EventSchemaEntry {topic, schema_version} contracttype - Add EventSchemaRegistry::get_schema(env, name) as single source of truth for oracle_result and dispute_created topics (v1) - emit_oracle_result / emit_dispute_created now read topic+version from the registry instead of inline literals; topic tuple gains schema_version - Tests: lookup hit, lookup miss, version mismatch, emit smoke tests #606 – HalfOpen cooldown and admin resume in circuit_breaker.rs - Add half_open_since: u64 to CircuitBreakerState (zero when not HalfOpen) - Add CircuitBreaker::request_resume(env, admin): Open -> HalfOpen with cooldown; errors if breaker is not Open - record_success: probe count only increments after recovery_timeout has elapsed since half_open_since; auto-closes on threshold - record_failure: re-opens from HalfOpen and clears half_open_since - Expose request_resume as a contract entrypoint in lib.rs - Tests: cooldown-not-elapsed, probe-failure-reopens, probe-success-threshold-closes --- .../predictify-hybrid/src/circuit_breaker.rs | 63 ++++++- .../src/circuit_breaker_tests.rs | 147 +++++++++++++++ contracts/predictify-hybrid/src/events.rs | 171 +++++++++++++++++- contracts/predictify-hybrid/src/lib.rs | 17 ++ 4 files changed, 390 insertions(+), 8 deletions(-) diff --git a/contracts/predictify-hybrid/src/circuit_breaker.rs b/contracts/predictify-hybrid/src/circuit_breaker.rs index 6ee379bc..0f2eeffe 100644 --- a/contracts/predictify-hybrid/src/circuit_breaker.rs +++ b/contracts/predictify-hybrid/src/circuit_breaker.rs @@ -65,6 +65,10 @@ pub struct CircuitBreakerState { pub error_count: u32, pub pause_scope: PauseScope, pub allow_withdrawals: bool, + /// Ledger timestamp when the breaker entered HalfOpen state. + /// Zero when the breaker is Closed or Open. Used to enforce the + /// cooldown window before probe requests are counted. + pub half_open_since: u64, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -141,6 +145,7 @@ impl CircuitBreaker { error_count: 0, pause_scope: PauseScope::BettingOnly, allow_withdrawals: false, + half_open_since: 0, }; env.storage() @@ -370,6 +375,7 @@ impl CircuitBreaker { if current_time - state.opened_time >= config.recovery_timeout { state.state = BreakerState::HalfOpen; state.half_open_requests = 0; + state.half_open_since = current_time; Self::update_state(env, &state)?; let _ = Self::emit_circuit_breaker_event( @@ -468,6 +474,7 @@ impl CircuitBreaker { state.state = BreakerState::Closed; state.failure_count = 0; state.half_open_requests = 0; + state.half_open_since = 0; state.last_success_time = env.ledger().timestamp(); // restore safe defaults state.pause_scope = PauseScope::BettingOnly; @@ -492,6 +499,47 @@ impl CircuitBreaker { Ok(()) } + /// Admin-initiated resume: move the breaker from Open → HalfOpen and start + /// the cooldown countdown. + /// + /// # Behaviour + /// + /// - Only an authorised admin may call this. + /// - The breaker must currently be `Open`; calling from `Closed` or + /// `HalfOpen` returns `Err(Error::CBError)`. + /// - Probe requests are not counted until `recovery_timeout` ledger-seconds + /// have elapsed since the `since` timestamp recorded here. This prevents + /// a flapping service from immediately re-tripping the breaker. + /// - After `half_open_max_requests` consecutive successes the breaker + /// auto-closes via [`record_success`]. + /// - A single failure during the probe window re-opens the breaker via + /// [`record_failure`]. + pub fn request_resume(env: &Env, admin: &Address) -> Result<(), Error> { + AdminAccessControl::validate_admin_for_action(env, admin, "emergency_actions")?; + + let mut state = Self::get_state(env)?; + + if state.state != BreakerState::Open { + return Err(Error::CBError); + } + + let current_time = env.ledger().timestamp(); + state.state = BreakerState::HalfOpen; + state.half_open_requests = 0; + state.half_open_since = current_time; + Self::update_state(env, &state)?; + + let _ = Self::emit_circuit_breaker_event( + env, + BreakerAction::Resume, + BreakerCondition::ManualOverride, + &String::from_str(env, "Admin requested resume: entering half-open with cooldown"), + Some(admin.clone()), + ); + + Ok(()) + } + /// Record a successful operation (for half-open state) pub fn record_success(env: &Env) -> Result<(), Error> { let mut state = Self::get_state(env)?; @@ -500,15 +548,23 @@ impl CircuitBreaker { state.total_requests += 1; state.last_success_time = current_time; - // If in half-open state, check if we can close + // If in half-open state, check cooldown then count probe successes if state.state == BreakerState::HalfOpen { + let config = Self::get_config(env)?; + // Enforce cooldown: ignore probe until recovery_timeout has elapsed + // since entering HalfOpen. + if current_time < state.half_open_since + config.recovery_timeout { + Self::update_state(env, &state)?; + return Ok(()); + } + state.half_open_requests += 1; - let config = Self::get_config(env)?; if state.half_open_requests >= config.half_open_max_requests { state.state = BreakerState::Closed; state.failure_count = 0; state.half_open_requests = 0; + state.half_open_since = 0; let _ = Self::emit_circuit_breaker_event( env, @@ -544,6 +600,7 @@ impl CircuitBreaker { state.state = BreakerState::Open; state.opened_time = current_time; state.half_open_requests = 0; + state.half_open_since = 0; let _ = Self::emit_circuit_breaker_event( env, @@ -921,6 +978,7 @@ impl CircuitBreakerTesting { error_count: 0, pause_scope: PauseScope::BettingOnly, allow_withdrawals: false, + half_open_since: 0, } } @@ -1058,6 +1116,7 @@ mod tests { error_count: 0, pause_scope: PauseScope::BettingOnly, allow_withdrawals: false, + half_open_since: 0, }; assert_eq!(state.state, BreakerState::Closed); assert_eq!(state.failure_count, 0); diff --git a/contracts/predictify-hybrid/src/circuit_breaker_tests.rs b/contracts/predictify-hybrid/src/circuit_breaker_tests.rs index 31df5fee..cc0718a5 100644 --- a/contracts/predictify-hybrid/src/circuit_breaker_tests.rs +++ b/contracts/predictify-hybrid/src/circuit_breaker_tests.rs @@ -651,4 +651,151 @@ mod circuit_breaker_tests { assert!(events.len() >= 2); // At least pause and recovery events }); } + + /// Cooldown not elapsed: probe successes before the cooldown window closes + /// must not advance the half_open_requests counter. + #[test] + fn test_half_open_cooldown_not_elapsed() { + let env = Env::default(); + let contract_id = env.register(crate::PredictifyHybrid, ()); + env.mock_all_auths(); + + env.as_contract(&contract_id, || { + CircuitBreaker::initialize(&env).unwrap(); + + let admin = ::generate(&env); + crate::admin::AdminInitializer::initialize(&env, &admin).unwrap(); + AdminRoleManager::assign_role( + &env, + &admin, + crate::admin::AdminRole::SuperAdmin, + &admin, + ) + .unwrap(); + + // Open the breaker + let reason = String::from_str(&env, "pause for cooldown test"); + CircuitBreaker::emergency_pause(&env, &admin, &reason).unwrap(); + + // Admin requests resume → HalfOpen with cooldown timestamp = now (ledger ts = 0) + CircuitBreaker::request_resume(&env, &admin).unwrap(); + let state = CircuitBreaker::get_state(&env).unwrap(); + assert_eq!(state.state, BreakerState::HalfOpen); + + // Record a success immediately (before cooldown has elapsed) + // recovery_timeout = 300 s; ledger time is still 0, so 0 < 0 + 300 + CircuitBreaker::record_success(&env).unwrap(); + + // half_open_requests must still be 0 — cooldown not yet elapsed + let state_after = CircuitBreaker::get_state(&env).unwrap(); + assert_eq!( + state_after.half_open_requests, 0, + "probe should be ignored while cooldown is active" + ); + assert_eq!( + state_after.state, + BreakerState::HalfOpen, + "breaker must remain HalfOpen during cooldown" + ); + }); + } + + /// Probe failure: a single failure while in HalfOpen re-opens the breaker. + #[test] + fn test_half_open_probe_failure_reopens() { + let env = Env::default(); + let contract_id = env.register(crate::PredictifyHybrid, ()); + env.mock_all_auths(); + + env.as_contract(&contract_id, || { + CircuitBreaker::initialize(&env).unwrap(); + + let admin = ::generate(&env); + crate::admin::AdminInitializer::initialize(&env, &admin).unwrap(); + AdminRoleManager::assign_role( + &env, + &admin, + crate::admin::AdminRole::SuperAdmin, + &admin, + ) + .unwrap(); + + // Pause then request resume + let reason = String::from_str(&env, "pause for probe failure test"); + CircuitBreaker::emergency_pause(&env, &admin, &reason).unwrap(); + CircuitBreaker::request_resume(&env, &admin).unwrap(); + + assert_eq!( + CircuitBreaker::get_state(&env).unwrap().state, + BreakerState::HalfOpen + ); + + // A failure during HalfOpen must reopen the breaker + CircuitBreaker::record_failure(&env).unwrap(); + + let state = CircuitBreaker::get_state(&env).unwrap(); + assert_eq!( + state.state, + BreakerState::Open, + "failure during HalfOpen must reopen the breaker" + ); + assert_eq!(state.half_open_since, 0, "half_open_since must be cleared"); + }); + } + + /// Probe success threshold: after the cooldown window passes, + /// `half_open_max_requests` consecutive successes must auto-close the breaker. + #[test] + fn test_half_open_probe_success_threshold_closes() { + let env = Env::default(); + let contract_id = env.register(crate::PredictifyHybrid, ()); + env.mock_all_auths(); + + env.as_contract(&contract_id, || { + CircuitBreaker::initialize(&env).unwrap(); + + let admin = ::generate(&env); + crate::admin::AdminInitializer::initialize(&env, &admin).unwrap(); + AdminRoleManager::assign_role( + &env, + &admin, + crate::admin::AdminRole::SuperAdmin, + &admin, + ) + .unwrap(); + + // Shorten recovery_timeout to 0 so probes are accepted immediately + let mut config = CircuitBreaker::get_config(&env).unwrap(); + config.recovery_timeout = 0; + config.half_open_max_requests = 3; + // bypass admin ACL by writing directly to storage (test only) + env.storage() + .instance() + .set(&soroban_sdk::Symbol::new(&env, "circuit_breaker_config"), &config); + + // Pause then request resume + let reason = String::from_str(&env, "pause for threshold test"); + CircuitBreaker::emergency_pause(&env, &admin, &reason).unwrap(); + CircuitBreaker::request_resume(&env, &admin).unwrap(); + + assert_eq!( + CircuitBreaker::get_state(&env).unwrap().state, + BreakerState::HalfOpen + ); + + // Record half_open_max_requests (3) successes — breaker should close + for _ in 0..3 { + CircuitBreaker::record_success(&env).unwrap(); + } + + let state = CircuitBreaker::get_state(&env).unwrap(); + assert_eq!( + state.state, + BreakerState::Closed, + "breaker must close after reaching the probe success threshold" + ); + assert_eq!(state.half_open_since, 0); + assert_eq!(state.failure_count, 0); + }); + } } diff --git a/contracts/predictify-hybrid/src/events.rs b/contracts/predictify-hybrid/src/events.rs index 81715601..95b7c498 100644 --- a/contracts/predictify-hybrid/src/events.rs +++ b/contracts/predictify-hybrid/src/events.rs @@ -1821,6 +1821,70 @@ pub struct MinPoolSizeNotMetEvent { pub timestamp: u64, } +// ===== EVENT SCHEMA REGISTRY ===== + +/// Describes the canonical topic symbol and schema version for a named event. +#[contracttype] +#[derive(Clone, Debug, Eq, PartialEq)] +pub struct EventSchemaEntry { + /// Short symbol used as the first element of the event topic tuple. + pub topic: Symbol, + /// Monotonically-increasing schema version. Increment whenever the + /// payload struct gains, removes, or renames fields. + pub schema_version: u32, +} + +/// Centralised registry that maps a human-readable event name to its +/// canonical topic symbol and schema version. +/// +/// # Purpose +/// +/// Emit sites **must not** hard-code topic symbols inline. Reading from +/// `EventSchemaRegistry` makes it trivial to grep all consumers when a +/// topic changes and provides a single place to bump `schema_version`. +/// +/// # Usage +/// +/// ```rust +/// # use soroban_sdk::Env; +/// # let env = Env::default(); +/// let schema = predictify_hybrid::events::EventSchemaRegistry::get_schema( +/// &env, "oracle_result", +/// ).unwrap(); +/// assert_eq!(schema.schema_version, 1); +/// ``` +pub struct EventSchemaRegistry; + +impl EventSchemaRegistry { + /// Return the [`EventSchemaEntry`] for a named event. + /// + /// # Errors + /// + /// Returns `None` when `name` is not a registered event. Callers + /// that treat a missing entry as a hard error should `unwrap_or_else` + /// with a panic or propagate an appropriate `Error` variant. + /// + /// # Registered events + /// + /// | name | topic symbol | schema_version | + /// |-------------------|---------------|----------------| + /// | `"oracle_result"` | `oracle_rs` | 1 | + /// | `"dispute_created"` | `dispt_crt` | 1 | + pub fn get_schema(env: &Env, name: &str) -> Option { + match name { + "oracle_result" => Some(EventSchemaEntry { + topic: symbol_short!("oracle_rs"), + schema_version: 1, + }), + "dispute_created" => Some(EventSchemaEntry { + topic: symbol_short!("dispt_crt"), + schema_version: 1, + }), + _ => None, + } + } +} + // ===== EVENT EMISSION UTILITIES ===== /// Event emission utilities @@ -2036,7 +2100,10 @@ impl EventEmitter { .publish((symbol_short!("bet_upd"), market_id.clone()), event); } - /// Emit oracle result event + /// Emit oracle result event. + /// + /// Topic and schema version are resolved from [`EventSchemaRegistry`] so + /// that all emit sites stay in sync with the registry automatically. pub fn emit_oracle_result( env: &Env, market_id: &Symbol, @@ -2047,6 +2114,11 @@ impl EventEmitter { threshold: i128, comparison: &String, ) { + let schema = EventSchemaRegistry::get_schema(env, "oracle_result") + .unwrap_or(EventSchemaEntry { + topic: symbol_short!("oracle_rs"), + schema_version: 1, + }); let event = OracleResultEvent { market_id: market_id.clone(), result: result.clone(), @@ -2058,9 +2130,9 @@ impl EventEmitter { timestamp: env.ledger().timestamp(), }; - Self::store_event(env, &symbol_short!("oracle_rs"), &event); + Self::store_event(env, &schema.topic, &event); env.events() - .publish((symbol_short!("oracle_rs"), market_id.clone()), event); + .publish((schema.topic, market_id.clone(), schema.schema_version), event); } // ===== ORACLE RESULT VERIFICATION EVENT EMISSION METHODS ===== @@ -2336,7 +2408,9 @@ impl EventEmitter { .publish((symbol_short!("pool_lo"), market_id.clone()), event); } - /// Emit dispute created event + /// Emit dispute created event. + /// + /// Topic and schema version are resolved from [`EventSchemaRegistry`]. pub fn emit_dispute_created( env: &Env, market_id: &Symbol, @@ -2344,6 +2418,11 @@ impl EventEmitter { stake: i128, reason: Option, ) { + let schema = EventSchemaRegistry::get_schema(env, "dispute_created") + .unwrap_or(EventSchemaEntry { + topic: symbol_short!("dispt_crt"), + schema_version: 1, + }); let event = DisputeCreatedEvent { market_id: market_id.clone(), disputer: disputer.clone(), @@ -2352,9 +2431,9 @@ impl EventEmitter { timestamp: env.ledger().timestamp(), }; - Self::store_event(env, &symbol_short!("dispt_crt"), &event); + Self::store_event(env, &schema.topic, &event); env.events() - .publish((symbol_short!("dispt_crt"), market_id.clone()), event); + .publish((schema.topic, market_id.clone(), schema.schema_version), event); } /// Emit dispute resolved event @@ -4268,3 +4347,83 @@ pub fn emit_manual_resolution_required(env: &Env, market_id: &Symbol, reason: &S (), ); } + +#[cfg(test)] +mod event_schema_registry_tests { + use super::*; + use soroban_sdk::{testutils::Address as _, Env}; + + #[test] + fn test_registry_lookup_oracle_result() { + let env = Env::default(); + let schema = EventSchemaRegistry::get_schema(&env, "oracle_result").unwrap(); + assert_eq!(schema.topic, symbol_short!("oracle_rs")); + assert_eq!(schema.schema_version, 1); + } + + #[test] + fn test_registry_lookup_dispute_created() { + let env = Env::default(); + let schema = EventSchemaRegistry::get_schema(&env, "dispute_created").unwrap(); + assert_eq!(schema.topic, symbol_short!("dispt_crt")); + assert_eq!(schema.schema_version, 1); + } + + #[test] + fn test_registry_lookup_unknown_event_returns_none() { + let env = Env::default(); + let result = EventSchemaRegistry::get_schema(&env, "nonexistent_event"); + assert!(result.is_none()); + } + + #[test] + fn test_schema_version_matches_expected() { + let env = Env::default(); + // Schema version must equal the pinned baseline; any bump is a breaking change. + const EXPECTED_ORACLE_RESULT_VERSION: u32 = 1; + const EXPECTED_DISPUTE_CREATED_VERSION: u32 = 1; + + let oracle_schema = EventSchemaRegistry::get_schema(&env, "oracle_result").unwrap(); + assert_eq!( + oracle_schema.schema_version, EXPECTED_ORACLE_RESULT_VERSION, + "OracleResultEvent schema_version mismatch: expected {EXPECTED_ORACLE_RESULT_VERSION}" + ); + + let dispute_schema = EventSchemaRegistry::get_schema(&env, "dispute_created").unwrap(); + assert_eq!( + dispute_schema.schema_version, EXPECTED_DISPUTE_CREATED_VERSION, + "DisputeCreatedEvent schema_version mismatch: expected {EXPECTED_DISPUTE_CREATED_VERSION}" + ); + } + + #[test] + fn test_emit_oracle_result_uses_registry_topic() { + let env = Env::default(); + let contract_id = env.register(crate::PredictifyHybrid, ()); + env.as_contract(&contract_id, || { + let market_id = soroban_sdk::symbol_short!("mkt1"); + let result = soroban_sdk::String::from_str(&env, "Yes"); + let provider = soroban_sdk::String::from_str(&env, "Reflector"); + let feed_id = soroban_sdk::String::from_str(&env, "BTC/USD"); + let comparison = soroban_sdk::String::from_str(&env, "gte"); + // Should not panic – registry supplies the topic. + EventEmitter::emit_oracle_result( + &env, &market_id, &result, &provider, &feed_id, 52_000_00000000, + 50_000_00000000, &comparison, + ); + }); + } + + #[test] + fn test_emit_dispute_created_uses_registry_topic() { + let env = Env::default(); + let contract_id = env.register(crate::PredictifyHybrid, ()); + env.as_contract(&contract_id, || { + let market_id = soroban_sdk::symbol_short!("mkt2"); + let disputer = soroban_sdk::Address::generate(&env); + EventEmitter::emit_dispute_created( + &env, &market_id, &disputer, 50_000_000, None, + ); + }); + } +} diff --git a/contracts/predictify-hybrid/src/lib.rs b/contracts/predictify-hybrid/src/lib.rs index 910983ca..519fd98d 100644 --- a/contracts/predictify-hybrid/src/lib.rs +++ b/contracts/predictify-hybrid/src/lib.rs @@ -6595,6 +6595,23 @@ impl PredictifyHybrid { ) -> Result, Error> { queries::QueryManager::get_top_users_by_win_rate(&env, limit, min_bets) } + + /// Admin-initiated circuit-breaker resume: Open → HalfOpen with cooldown. + /// + /// Moves the circuit breaker from `Open` to `HalfOpen` and records the + /// current ledger timestamp as the cooldown start. Probe requests are not + /// counted toward the success threshold until `recovery_timeout` seconds have + /// elapsed. After `half_open_max_requests` consecutive probe successes the + /// breaker auto-closes; any failure during the probe window re-opens it. + /// + /// # Errors + /// + /// - `Error::Unauthorized` — caller is not an authorised admin. + /// - `Error::CBError` — breaker is not currently `Open`. + pub fn request_resume(env: Env, admin: Address) -> Result<(), Error> { + admin.require_auth(); + crate::circuit_breaker::CircuitBreaker::request_resume(&env, &admin) + } } #[cfg(any())]