Skip to content

fix(archive): bound archive queries, historical reads, and pagination cursors (#1379) #1440

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
s6pa1rta3n-lab wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

fix(archive): bound archive queries, historical reads, and pagination cursors (#1379) #1440

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions contracts/predictify-hybrid/src/admin.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ use alloc::format;
use soroban_sdk::{contracttype, Address, Env, Map, String, Symbol, Vec};
// use alloc::string::ToString; // Unused import

use crate::config::{ConfigManager, ConfigUtils, ContractConfig, Environment};
use crate::config::{ConfigManager, ConfigUtils, ConfigValidator, ContractConfig, Environment};
use crate::err::Error;
use crate::events::EventEmitter;
use crate::extensions::ExtensionManager;
Expand Down Expand Up @@ -353,7 +353,7 @@ impl AdminInitializer {
Environment::Mainnet => ConfigManager::get_mainnet_config(env),
Environment::Custom => ConfigManager::get_development_config(env),
};
ConfigManager::validate_config(env, &config)?;
ConfigValidator::validate_contract_config(&config)?;

// Initialize basic admin setup
AdminInitializer::initialize(env, admin)?;
Expand Down
77 changes: 55 additions & 22 deletions contracts/predictify-hybrid/src/event_archive.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,12 @@ pub fn derive_archive_key(env: &Env, market_id: &Symbol, suffix: &str) -> (Symbo
/// Maximum events returned per query (gas safety).
pub const MAX_QUERY_LIMIT: u32 = 30;

/// Default events returned per query when limit is 0.
pub const DEFAULT_QUERY_LIMIT: u32 = 10;

/// Maximum entries that can be pruned in a single transaction.
pub const MAX_PRUNE_BATCH: u32 = 30;

/// Hard cap on the number of archived entries stored on-chain.
///
/// Prevents unbounded storage growth. When the archive reaches this limit,
Expand All @@ -124,6 +130,15 @@ pub const MAX_QUERY_LIMIT: u32 = 30;
/// bounding worst-case growth.
pub const MAX_ARCHIVE_SIZE: u32 = 1_000;

/// Sanitize query limit to ensure it never exceeds MAX_QUERY_LIMIT and defaults to DEFAULT_QUERY_LIMIT when 0.
pub fn sanitize_limit(limit: u32) -> u32 {
if limit == 0 {
DEFAULT_QUERY_LIMIT
} else {
core::cmp::min(limit, MAX_QUERY_LIMIT)
}
}

/// Storage key for archived event timestamps (market_id -> archived_at).
const ARCHIVED_TS_KEY: &str = "evt_archived";

Expand Down Expand Up @@ -352,7 +367,7 @@ impl EventArchive {
return Err(Error::Unauthorized);
}

let count = core::cmp::min(count, MAX_QUERY_LIMIT);
let count = core::cmp::min(count, MAX_PRUNE_BATCH);
if count == 0 {
return Ok((0, cursor.unwrap_or_else(|| PruneCursor::new(env))));
}
Expand Down Expand Up @@ -622,7 +637,12 @@ impl EventArchive {
cursor: u32,
limit: u32,
) -> (Vec<EventHistoryEntry>, u32) {
let limit = core::cmp::min(limit, MAX_QUERY_LIMIT);
let total = MarketIdGenerator::get_market_id_registry_len(env);
if cursor >= total {
return (Vec::new(env), total);
}

let limit = sanitize_limit(limit);
let registry_page = MarketIdGenerator::get_market_id_registry(env, cursor, limit);
let mut result = Vec::new(env);
let mut scanned = 0u32;
Expand All @@ -648,7 +668,8 @@ impl EventArchive {
}
}

(result, cursor + scanned)
let next_cursor = core::cmp::min(cursor.saturating_add(scanned), total);
(result, next_cursor)
}

/// Query events by resolution status (paginated, bounded).
Expand All @@ -660,7 +681,12 @@ impl EventArchive {
cursor: u32,
limit: u32,
) -> (Vec<EventHistoryEntry>, u32) {
let limit = core::cmp::min(limit, MAX_QUERY_LIMIT);
let total = MarketIdGenerator::get_market_id_registry_len(env);
if cursor >= total {
return (Vec::new(env), total);
}

let limit = sanitize_limit(limit);
let registry_page = MarketIdGenerator::get_market_id_registry(env, cursor, limit);
let mut result = Vec::new(env);
let mut scanned = 0u32;
Expand All @@ -685,7 +711,8 @@ impl EventArchive {
}
}

(result, cursor + scanned)
let next_cursor = core::cmp::min(cursor.saturating_add(scanned), total);
(result, next_cursor)
}

/// Query archived events directly (paginated, bounded).
Expand All @@ -704,7 +731,6 @@ impl EventArchive {
cursor: u32,
limit: u32,
) -> (Vec<EventHistoryEntry>, u32) {
let limit = core::cmp::min(limit, MAX_QUERY_LIMIT);
let index_key = Symbol::new(env, ARCHIVED_INDEX_KEY);
let index: Vec<(u64, Symbol)> = env
.storage()
Expand All @@ -713,12 +739,16 @@ impl EventArchive {
.unwrap_or_else(|| Vec::new(env));
let total = index.len() as u32;

if cursor >= total {
return (Vec::new(env), total);
}

let limit = sanitize_limit(limit);
let mut result = Vec::new(env);
let mut examined = 0u32;

if !reverse {
// Ascending (oldest archived first): page [cursor, cursor + limit).
let mut idx = core::cmp::min(cursor, total);
let mut idx = cursor;
while examined < limit && idx < total {
if let Some((_, market_id)) = index.get(idx) {
examined += 1;
Expand All @@ -729,9 +759,7 @@ impl EventArchive {
idx += 1;
}
} else {
// Descending (newest archived first): `cursor` counts back from the
// newest entry; cursor 0 starts at the newest (last index).
let mut idx = total.saturating_sub(core::cmp::min(cursor, total));
let mut idx = total.saturating_sub(cursor);
while examined < limit && idx > 0 {
let position = idx - 1;
if let Some((_, market_id)) = index.get(position) {
Expand All @@ -744,9 +772,6 @@ impl EventArchive {
}
}

// Advance the cursor by the number of archive entries examined so the
// caller can page. When a page examined nothing we are at the end (or the
// archive is empty); pin the cursor to signal completion.
let next_cursor = if examined == 0 {
cursor
} else {
Expand All @@ -761,8 +786,6 @@ impl EventArchive {
/// index and market records are consistent).
fn history_entry_for_archived(env: &Env, market_id: &Symbol) -> Option<EventHistoryEntry> {
let market: Market = env.storage().persistent().get(market_id)?;
// `created_at` comes from the market ID registry when available; fall back
// to `end_time` for legacy/synthetic IDs without a registry entry.
let created_at = MarketIdGenerator::get_registry_entry(env, market_id)
.map(|entry| entry.timestamp)
.unwrap_or(market.end_time);
Expand All @@ -779,7 +802,12 @@ impl EventArchive {
cursor: u32,
limit: u32,
) -> (Vec<EventHistoryEntry>, u32) {
let limit = core::cmp::min(limit, MAX_QUERY_LIMIT);
let total = MarketIdGenerator::get_market_id_registry_len(env);
if cursor >= total {
return (Vec::new(env), total);
}

let limit = sanitize_limit(limit);
let registry_page = MarketIdGenerator::get_market_id_registry(env, cursor, limit);
let mut result = Vec::new(env);
let mut scanned = 0u32;
Expand All @@ -792,7 +820,6 @@ impl EventArchive {
.persistent()
.get::<Symbol, Market>(&entry.market_id)
{
// Match against dedicated category field if set, otherwise oracle feed_id
let market_category = market
.category
.clone()
Expand All @@ -809,7 +836,8 @@ impl EventArchive {
}
}

(result, cursor + scanned)
let next_cursor = core::cmp::min(cursor.saturating_add(scanned), total);
(result, next_cursor)
}

/// Query events by tags (paginated, bounded).
Expand All @@ -822,7 +850,12 @@ impl EventArchive {
cursor: u32,
limit: u32,
) -> (Vec<EventHistoryEntry>, u32) {
let limit = core::cmp::min(limit, MAX_QUERY_LIMIT);
let total = MarketIdGenerator::get_market_id_registry_len(env);
if cursor >= total {
return (Vec::new(env), total);
}

let limit = sanitize_limit(limit);
let registry_page = MarketIdGenerator::get_market_id_registry(env, cursor, limit);
let mut result = Vec::new(env);
let mut scanned = 0u32;
Expand All @@ -839,7 +872,6 @@ impl EventArchive {
.persistent()
.get::<Symbol, Market>(&entry.market_id)
{
// Check if any of the market's tags match any of the query tags
let mut matched = false;
for j in 0..market.tags.len() {
if let Some(market_tag) = market.tags.get(j) {
Expand Down Expand Up @@ -868,7 +900,8 @@ impl EventArchive {
}
}

(result, cursor + scanned)
let next_cursor = core::cmp::min(cursor.saturating_add(scanned), total);
(result, next_cursor)
}
}

Expand Down
2 changes: 2 additions & 0 deletions contracts/predictify-hybrid/src/event_topic_compat_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@

#![cfg(test)]

extern crate alloc;
use alloc::format;
use soroban_sdk::{symbol_short, testutils::Events, Env, Symbol, Vec};

use crate::event_topic_compat::{
Expand Down
3 changes: 1 addition & 2 deletions contracts/predictify-hybrid/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -393,8 +393,7 @@ impl PredictifyHybrid {

// Seed permissive-but-valid rate limits so admin entrypoints do not
// fail before a custom policy is configured.
match crate::rate_limiter::RateLimiter::new(env.clone()).init_rate_limiter(
admin.clone(),
match crate::rate_limiter::RateLimiter::new(env.clone()).set_config_internal(
crate::rate_limiter::RateLimitConfig {
voting_limit: 10_000,
dispute_limit: 1_000,
Expand Down
18 changes: 17 additions & 1 deletion contracts/predictify-hybrid/src/market_id_generator.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,6 +190,17 @@ impl MarketIdGenerator {
Err(Error::InvalidInput)
}

/// Return the total number of entries in the market ID registry.
pub fn get_market_id_registry_len(env: &Env) -> u32 {
let key = Symbol::new(env, Self::REGISTRY_KEY);
let registry: Vec<MarketIdRegistryEntry> = env
.storage()
.persistent()
.get(&key)
.unwrap_or(Vec::new(env));
registry.len()
}

/// Return a paginated slice of the market ID registry.
pub fn get_market_id_registry(env: &Env, start: u32, limit: u32) -> Vec<MarketIdRegistryEntry> {
let key = Symbol::new(env, Self::REGISTRY_KEY);
Expand All @@ -199,8 +210,13 @@ impl MarketIdGenerator {
.get(&key)
.unwrap_or(Vec::new(env));

let total = registry.len();
if start >= total {
return Vec::new(env);
}

let mut result = Vec::new(env);
let end = core::cmp::min(start + limit, registry.len());
let end = core::cmp::min(start.saturating_add(limit), total);
for i in start..end {
if let Some(entry) = registry.get(i) {
result.push_back(entry);
Expand Down
Loading
Loading