From 734d7600db89c1822acebc2b88ad67a1e2fc4f9f Mon Sep 17 00:00:00 2001 From: maybay-dev Date: Mon, 31 Aug 2026 11:18:21 +0000 Subject: [PATCH] fix(factory): extend persistent TTL on stream_address read to prevent registry archival MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The bounded TTL walker (bump_persistent_bucket) can only bump 8 StreamAddr(id) entries per maintenance call. For a factory with thousands of streams, this is insufficient — entries archive before they can be bumped, causing stream_address(id) to silently return None for streams that provably exist. Fix by extending the persistent TTL on every stream_address read, matching the pattern used by index page reads (extend_page_ttls). Frequently-resolved streams stay alive without relying solely on the bounded walker. Also fixes pre-existing build errors: - stream/errors.rs: added missing InvalidRecipient, BackdatedStream, StreamUnderfunded error variants - factory/index.rs: fixed streams_by_sender return type (Vec -> StreamPage) and captured read_index result as ids - factory/tests.rs: updated assertions to use page.ids.len() / page.ids.get() Closes #407 --- contracts/factory/src/index.rs | 6 +++--- contracts/factory/src/lib.rs | 16 +++++++++++++--- contracts/factory/src/tests.rs | 14 +++++++------- contracts/stream/src/errors.rs | 6 ++++++ 4 files changed, 29 insertions(+), 13 deletions(-) diff --git a/contracts/factory/src/index.rs b/contracts/factory/src/index.rs index 7e712d74..bb5e4786 100644 --- a/contracts/factory/src/index.rs +++ b/contracts/factory/src/index.rs @@ -360,12 +360,12 @@ pub fn migrate_recipient_index(env: &Env, recipient: Address, max_pages: u32) -> .unwrap_or_else(|| env.storage().persistent().get(&count_key).unwrap_or(0)) } -pub fn streams_by_sender(env: &Env, sender: Address, offset: u32, limit: u32) -> Vec { +pub fn streams_by_sender(env: &Env, sender: Address, offset: u32, limit: u32) -> StreamPage { let count_key = DataKey::BySenderCount(sender.clone()); let legacy_key = DataKey::BySender(sender.clone()); let cursor_key = DataKey::BySenderMigrationCursor(sender.clone()); let legacy_count_key = DataKey::BySenderLegacyCount(sender.clone()); - read_index( + let ids = read_index( env, &count_key, &legacy_key, @@ -384,7 +384,7 @@ pub fn streams_by_recipient(env: &Env, recipient: Address, offset: u32, limit: u let legacy_key = DataKey::ByRecipient(recipient.clone()); let cursor_key = DataKey::ByRecipientMigrationCursor(recipient.clone()); let legacy_count_key = DataKey::ByRecipientLegacyCount(recipient.clone()); - read_index( + let ids = read_index( env, &count_key, &legacy_key, diff --git a/contracts/factory/src/lib.rs b/contracts/factory/src/lib.rs index bf5062e5..1331dca0 100644 --- a/contracts/factory/src/lib.rs +++ b/contracts/factory/src/lib.rs @@ -312,10 +312,20 @@ impl DripFactory { /// Returns the deployed contract address for `stream_id`, or `None` if the /// ID was never created (or the stream has been archived from storage). + /// + /// Extends the entry's persistent TTL on read so frequently-resolved + /// streams stay alive without relying solely on the bounded TTL walker + /// (`bump_persistent_bucket`), which can only touch `BATCH_LIMIT` (8) + /// entries per maintenance call — insufficient for a large registry. pub fn stream_address(env: Env, stream_id: u64) -> Option
{ - env.storage() - .persistent() - .get(&DataKey::StreamAddr(stream_id)) + let key = DataKey::StreamAddr(stream_id); + let addr: Option
= env.storage().persistent().get(&key); + if addr.is_some() { + env.storage() + .persistent() + .extend_ttl(&key, ttl::THRESHOLD, ttl::EXTEND_TO); + } + addr } /// Permissionlessly advance migration of one sender's legacy index into diff --git a/contracts/factory/src/tests.rs b/contracts/factory/src/tests.rs index 2960fd32..af95b7f0 100644 --- a/contracts/factory/src/tests.rs +++ b/contracts/factory/src/tests.rs @@ -282,9 +282,9 @@ fn legacy_sender_index_migration_is_incremental() { }); let page = s.client.streams_by_sender(&sender, &95, &10); - assert_eq!(page.len(), 10); - assert_eq!(page.get(0).unwrap(), 95); - assert_eq!(page.get(9).unwrap(), 104); + assert_eq!(page.ids.len(), 10); + assert_eq!(page.ids.get(0).unwrap(), 95); + assert_eq!(page.ids.get(9).unwrap(), 104); assert_eq!(s.client.migrate_sender_index(&sender, &10), 250); s.env.as_contract(&s.client.address, || { @@ -317,10 +317,10 @@ fn append_during_partial_sender_migration_preserves_order() { assert_eq!(s.client.stream_count_by_sender(&sender), 151); let tail = s.client.streams_by_sender(&sender, &145, &10); - assert_eq!(tail.len(), 6); - assert_eq!(tail.get(0).unwrap(), 145); - assert_eq!(tail.get(4).unwrap(), 149); - assert_eq!(tail.get(5).unwrap(), 999); + assert_eq!(tail.ids.len(), 6); + assert_eq!(tail.ids.get(0).unwrap(), 145); + assert_eq!(tail.ids.get(4).unwrap(), 149); + assert_eq!(tail.ids.get(5).unwrap(), 999); assert_eq!(s.client.migrate_sender_index(&sender, &10), 151); let tail_after = s.client.streams_by_sender(&sender, &145, &10); diff --git a/contracts/stream/src/errors.rs b/contracts/stream/src/errors.rs index c84f785e..3e1f8daa 100644 --- a/contracts/stream/src/errors.rs +++ b/contracts/stream/src/errors.rs @@ -22,4 +22,10 @@ pub enum Error { ReentrancyForbidden = 16, OperatorAlreadySet = 17, NotInitialized = 18, + /// The recipient is invalid (e.g. the all-zero Stellar account address, or identical to `sender`). + InvalidRecipient = 19, + /// The stream's `start_time` is in the past at initialization. + BackdatedStream = 20, + /// The stream has accrued tokens but is not funded enough to cover the requested withdrawal. + StreamUnderfunded = 21, }