Skip to content

feat: deterministic archive pruning with paginated PruneCursor #729

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

Merged
greatest0fallt1me merged 1 commit into from
Jun 27, 2026
Merged

feat: deterministic archive pruning with paginated PruneCursor #729

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
239 changes: 190 additions & 49 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 @@ -20,7 +20,7 @@
use crate::errors::Error;
use crate::market_id_generator::MarketIdGenerator;
use crate::types::{EventHistoryEntry, Market, MarketState};
use soroban_sdk::{panic_with_error, Address, Env, String, Symbol, Vec};
use soroban_sdk::{contracttype, panic_with_error, Address, Env, String, Symbol, Vec};

// ---------------------------------------------------------------------------
// Deterministic archive key derivation
Expand Down Expand Up @@ -53,11 +53,7 @@ use soroban_sdk::{panic_with_error, Address, Env, String, Symbol, Vec};
///
/// A 3-tuple `(Symbol, Symbol, Symbol)` suitable for use as a Soroban
/// persistent storage key.
pub fn derive_archive_key(
env: &Env,
market_id: &Symbol,
suffix: &str,
) -> (Symbol, Symbol, Symbol) {
pub fn derive_archive_key(env: &Env, market_id: &Symbol, suffix: &str) -> (Symbol, Symbol, Symbol) {
(
Symbol::new(env, "__archive"),
market_id.clone(),
Expand All @@ -82,6 +78,69 @@ pub const MAX_ARCHIVE_SIZE: u32 = 1_000;
/// Storage key for archived event timestamps (market_id -> archived_at).
const ARCHIVED_TS_KEY: &str = "evt_archived";

/// Storage key for the sorted archive index [(timestamp, market_id)] in ascending order.
const ARCHIVED_INDEX_KEY: &str = "evt_arch_ix";

/// Pagination cursor for archive pruning.
///
/// Returned by `prune_archive` so callers can resume pruning from where they left off.
/// Pass `None` to start from the beginning.
#[derive(Clone, Debug)]
#[contracttype]
pub struct PruneCursor {
/// Timestamp of the last pruned entry; 0 means "start from the beginning".
pub last_timestamp: u64,
/// Market ID of the last pruned entry (tiebreaker for same-timestamp entries).
pub last_market_id: Symbol,
/// Whether there are no more entries to prune.
pub done: bool,
}

impl PruneCursor {
/// Create a new cursor pointing to the beginning (before any entries).
pub fn new(env: &Env) -> Self {
Self {
last_timestamp: 0,
last_market_id: Symbol::new(env, "_"),
done: false,
}
}
}

// ---------------------------------------------------------------------------
// Sorted index helpers
// ---------------------------------------------------------------------------

/// Insert a (timestamp, market_id) pair into the sorted archive index,
/// maintaining ascending order by timestamp (then by market_id as tiebreaker).
fn insert_into_sorted_index(env: &Env, timestamp: u64, market_id: &Symbol) {
let index_key = Symbol::new(env, ARCHIVED_INDEX_KEY);
let index: Vec<(u64, Symbol)> = env
.storage()
.persistent()
.get(&index_key)
.unwrap_or_else(|| Vec::new(env));

let mut new_index = Vec::new(env);
let mut inserted = false;

for i in 0..index.len() {
if let Some(entry) = index.get(i) {
if !inserted && (timestamp < entry.0 || (timestamp == entry.0 && *market_id < entry.1))
{
new_index.push_back((timestamp, market_id.clone()));
inserted = true;
}
new_index.push_back(entry);
}
}
if !inserted {
new_index.push_back((timestamp, market_id.clone()));
}

env.storage().persistent().set(&index_key, &new_index);
}

/// Event archive and historical query manager.
pub struct EventArchive;

Expand Down Expand Up @@ -142,24 +201,37 @@ impl EventArchive {
archived.set(market_id.clone(), now);
env.storage().persistent().set(&key, &archived);

// Maintain the sorted index for deterministic pruning
insert_into_sorted_index(env, now, market_id);

Ok(())
}

/// Remove the oldest `count` entries from the archive (admin only).
///
/// Frees capacity so that new events can be archived after the cap is reached.
/// Entries are removed in insertion order (oldest first). The underlying
/// `Map` does not preserve insertion order, so we iterate the market registry
/// to find the oldest archived IDs.
/// Entries are pruned deterministically by archive timestamp ascending (oldest first).
/// A [`PruneCursor`] enables paginated pruning — pass the returned cursor on the next call
/// to resume where you left off.
///
/// # Arguments
/// * `env` - Soroban environment
/// * `admin` - Caller must be contract admin
/// * `count` - Number of oldest entries to remove (capped at [`MAX_QUERY_LIMIT`])
/// * `cursor` - [`Some(PruneCursor)`] to resume from a previous position, or [`None`] to
/// start from the beginning
///
/// # Returns
/// `(number_pruned, new_cursor)` — the caller should persist `new_cursor` for the next call.
///
/// # Errors
/// * `Unauthorized` - Caller is not admin
pub fn prune_archive(env: &Env, admin: &Address, count: u32) -> Result<u32, Error> {
pub fn prune_archive(
env: &Env,
admin: &Address,
count: u32,
cursor: Option<PruneCursor>,
) -> Result<(u32, PruneCursor), Error> {
admin.require_auth();

let stored_admin: Address = env
Expand All @@ -173,44 +245,109 @@ impl EventArchive {
}

let count = core::cmp::min(count, MAX_QUERY_LIMIT);
let key = Symbol::new(env, ARCHIVED_TS_KEY);
let mut archived: soroban_sdk::Map<Symbol, u64> = env
if count == 0 {
return Ok((0, cursor.unwrap_or_else(|| PruneCursor::new(env))));
}

// Read the sorted index
let index_key = Symbol::new(env, ARCHIVED_INDEX_KEY);
let index: Vec<(u64, Symbol)> = env
.storage()
.persistent()
.get(&key)
.unwrap_or(soroban_sdk::Map::new(env));
.get(&index_key)
.unwrap_or_else(|| Vec::new(env));

if index.is_empty() {
return Ok((
0,
PruneCursor {
last_timestamp: 0,
last_market_id: Symbol::new(env, "_"),
done: true,
},
));
}

if archived.is_empty() || count == 0 {
return Ok(0);
// Determine starting position from cursor
let start_cursor = cursor.unwrap_or_else(|| PruneCursor::new(env));

// If cursor says done, there's nothing more to prune
if start_cursor.done {
return Ok((0, start_cursor));
}

// Walk the registry (chronological order) to find the oldest archived IDs.
let registry_len = MarketIdGenerator::get_market_id_registry(env, 0, u32::MAX).len();
let mut removed = 0u32;
let mut cursor = 0u32;

while removed < count && cursor < registry_len {
let page = MarketIdGenerator::get_market_id_registry(env, cursor, MAX_QUERY_LIMIT);
let page_len = page.len();
for i in 0..page_len {
if removed >= count {
break;
}
if let Some(entry) = page.get(i) {
if archived.get(entry.market_id.clone()).is_some() {
archived.remove(entry.market_id);
removed += 1;
let start_pos = if start_cursor.last_timestamp == 0 {
0u32
} else {
// Find first entry strictly after the cursor
let mut pos = 0u32;
for i in 0..index.len() {
if let Some(entry) = index.get(i) {
if entry.0 > start_cursor.last_timestamp
|| (entry.0 == start_cursor.last_timestamp
&& entry.1 > start_cursor.last_market_id)
{
break;
}
pos = i + 1;
}
}
cursor += page_len;
if page_len == 0 {
break;
pos
};

if start_pos >= index.len() {
return Ok((
0,
PruneCursor {
last_timestamp: start_cursor.last_timestamp,
last_market_id: start_cursor.last_market_id,
done: true,
},
));
}

// Remove up to `count` entries starting from start_pos
let archived_key = Symbol::new(env, ARCHIVED_TS_KEY);
let mut archived: soroban_sdk::Map<Symbol, u64> = env
.storage()
.persistent()
.get(&archived_key)
.unwrap_or_else(|| soroban_sdk::Map::new(env));

let mut removed = 0u32;
let mut last_ts: u64 = 0;
let mut last_id = Symbol::new(env, "_");
let mut new_index = Vec::new(env);

for i in 0..index.len() {
if let Some(entry) = index.get(i) {
if i < start_pos {
// Keep entries before start_pos
new_index.push_back(entry);
} else if removed < count {
// Remove this entry from the archive map
archived.remove(entry.1.clone());
last_ts = entry.0;
last_id = entry.1.clone();
removed += 1;
} else {
// Keep remaining entries
new_index.push_back(entry);
}
}
}

env.storage().persistent().set(&key, &archived);
Ok(removed)
env.storage().persistent().set(&archived_key, &archived);
env.storage().persistent().set(&index_key, &new_index);

let done = removed < count || new_index.is_empty();
let new_cursor = PruneCursor {
last_timestamp: last_ts,
last_market_id: last_id,
done,
};

Ok((removed, new_cursor))
}

/// Check if an event is archived.
Expand Down Expand Up @@ -468,10 +605,10 @@ impl EventArchive {
#[cfg(test)]
mod tests {
use super::*;
use crate::types::{Market, MarketState, OracleConfig};
use alloc::string::ToString;
use soroban_sdk::testutils::Address as _;
use soroban_sdk::Map;
use crate::types::{Market, MarketState, OracleConfig};

// =========================================================================
// derive_archive_key tests
Expand Down Expand Up @@ -921,10 +1058,10 @@ mod tests {
let test = EventArchiveTest::new();
let env = &test.env;
let contract_id = env.register(crate::PredictifyHybrid, ());

env.as_contract(&contract_id, || {
let max_length_id = Symbol::new(env, "a_very_long_market_id_32_chars__");

let market = Market {
admin: test.admin.clone(),
question: String::from_str(env, "Will BTC reach 100k?"),
Expand Down Expand Up @@ -954,7 +1091,8 @@ mod tests {
winnings_swept: false,
};

let res = crate::storage::StorageOptimizer::archive_market_data(env, &max_length_id, &market);
let res =
crate::storage::StorageOptimizer::archive_market_data(env, &max_length_id, &market);
assert!(res.is_ok());
});
}
Expand All @@ -964,10 +1102,10 @@ mod tests {
let test = EventArchiveTest::new();
let env = &test.env;
let contract_id = env.register(crate::PredictifyHybrid, ());

env.as_contract(&contract_id, || {
let market_id = Symbol::new(env, "market_1");

let market = Market {
admin: test.admin.clone(),
question: String::from_str(env, "Will BTC reach 100k?"),
Expand Down Expand Up @@ -997,10 +1135,12 @@ mod tests {
winnings_swept: false,
};

let res1 = crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
let res1 =
crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
assert!(res1.is_ok());

let res2 = crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
let res2 =
crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
assert!(res2.is_ok());
});
}
Expand All @@ -1010,11 +1150,11 @@ mod tests {
let test = EventArchiveTest::new();
let env = &test.env;
let contract_id = env.register(crate::PredictifyHybrid, ());

env.as_contract(&contract_id, || {
let market_id = Symbol::new(env, "retrievable_market");
let timestamp = env.ledger().timestamp();

let market = Market {
admin: test.admin.clone(),
question: String::from_str(env, "Will BTC reach 100k?"),
Expand Down Expand Up @@ -1044,13 +1184,14 @@ mod tests {
winnings_swept: false,
};

let res = crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
let res =
crate::storage::StorageOptimizer::archive_market_data(env, &market_id, &market);
assert!(res.is_ok());

let key = crate::storage::DataKey::ArchivedMarket(market_id.clone(), timestamp);
let retrieved: Option<Market> = env.storage().persistent().get(&key);
assert!(retrieved.is_some());

let retrieved_market = retrieved.unwrap();
assert_eq!(retrieved_market.question, market.question);
assert_eq!(retrieved_market.admin, market.admin);
Expand Down
Loading
Loading