diff --git a/contracts/stream/src/events.rs b/contracts/stream/src/events.rs index 91c1d9eb..9e83468c 100644 --- a/contracts/stream/src/events.rs +++ b/contracts/stream/src/events.rs @@ -4,19 +4,31 @@ use crate::{storage::DataKey, Error}; /// Allocate the next event sequence before publishing its payload. /// -/// Event publication and storage writes are part of the same Soroban -/// transaction, so either both commit or both are rolled back. Existing -/// streams that predate this key start at sequence zero. +/// The sequence is now part of the persisted `StreamInfo`/`Config` payload so it +/// survives the consolidated state migration. Legacy streams without `Config` are +/// still readable via the fallback path for as long as they remain on the old +/// storage layout. /// /// Boundary check: `current` is validated to prevent arithmetic overflow /// on the sequence counter (which would silently consume future events). fn next_sequence(env: &Env) -> u64 { let storage = env.storage().instance(); - let current = storage.get::<_, u64>(&DataKey::EventSequence).unwrap_or(0); + let current = if storage.has(&DataKey::Config) { + crate::state::load(env).event_sequence + } else { + storage.get::<_, u64>(&DataKey::EventSequence).unwrap_or(0) + }; let Some(next) = current.checked_add(1) else { panic_with_error!(env, Error::ArithmeticOverflow); }; - storage.set(&DataKey::EventSequence, &next); + + if storage.has(&DataKey::Config) { + let mut info = crate::state::load(env); + info.event_sequence = next; + crate::state::save(env, &info); + } else { + storage.set(&DataKey::EventSequence, &next); + } next } diff --git a/contracts/stream/src/lib.rs b/contracts/stream/src/lib.rs index 81d5d290..b4b7b282 100644 --- a/contracts/stream/src/lib.rs +++ b/contracts/stream/src/lib.rs @@ -122,36 +122,37 @@ impl DripStream { } let s = env.storage().instance(); - s.set(&DataKey::EventSequence, &0_u64); s.set(&DataKey::StorageVersion, &storage::CURRENT_STORAGE_VERSION); - events::created( - &env, - &sender, - &recipient, - &token, - rate_per_second, - start_time, - end_time, - ); - - // Write the entire stream state as a single struct — one storage - // write instead of eleven. All subsequent reads go through - // state::load(), which fetches the whole struct in one call. + // Write the initial state before emitting the creation event so the + // event sequence lives with the consolidated `Config` payload from the + // start. `events::created()` then advances it to sequence 1 and persists + // the updated counter back into `Config`. state::save( &env, &StreamInfo { - sender, - recipient, - token, + sender: sender.clone(), + recipient: recipient.clone(), + token: token.clone(), rate_per_second, start_time, end_time, flags, withdrawn: 0, paused_at: 0, + event_sequence: 0, }, ); + + events::created( + &env, + &sender, + &recipient, + &token, + rate_per_second, + start_time, + end_time, + ); } /// Recipient withdraws `amount` tokens. @@ -762,10 +763,11 @@ impl DripStream { /// processed after reconnecting. A gap means the missing ledger range /// must be replayed before live processing continues. pub fn event_sequence(env: Env) -> u64 { - env.storage() - .instance() - .get(&DataKey::EventSequence) - .unwrap_or(0) + let storage = env.storage().instance(); + if storage.has(&DataKey::Config) { + return state::load(&env).event_sequence; + } + storage.get(&DataKey::EventSequence).unwrap_or(0) } /// Storage layout version this instance was initialized with. diff --git a/contracts/stream/src/state.rs b/contracts/stream/src/state.rs index 8b128a6c..7208f3ce 100644 --- a/contracts/stream/src/state.rs +++ b/contracts/stream/src/state.rs @@ -43,6 +43,7 @@ pub fn try_load(env: &Env) -> Result { withdrawn: s.get(&DataKey::Withdrawn).unwrap_or(0), paused_at: s.get(&DataKey::PausedAt).unwrap_or(0), flags, + event_sequence: s.get(&DataKey::EventSequence).unwrap_or(0), }) } @@ -62,7 +63,7 @@ pub fn load(env: &Env) -> StreamInfo { /// (`withdraw`/`pause`/`resume`/`cancel`/`top_up`/`extend_duration`) for data /// no code path reads. They are removed once, on the first `save()` of a /// pre-consolidation stream, after which `save()` writes only `Config`. -const LEGACY_STATE_KEYS: [DataKey; 11] = [ +const LEGACY_STATE_KEYS: [DataKey; 12] = [ DataKey::Sender, DataKey::Recipient, DataKey::Token, @@ -74,6 +75,7 @@ const LEGACY_STATE_KEYS: [DataKey; 11] = [ DataKey::Flags, DataKey::ClawbackEnabled, DataKey::Cancelled, + DataKey::EventSequence, ]; /// Persist the entire stream state in a single storage write. diff --git a/contracts/stream/src/storage.rs b/contracts/stream/src/storage.rs index 9af9a000..82834dfb 100644 --- a/contracts/stream/src/storage.rs +++ b/contracts/stream/src/storage.rs @@ -35,10 +35,11 @@ pub enum DataKey { /// Replaces the 11 individual keys above for new writes — loaded in one /// storage read instead of eleven. Config, - /// Monotonic identifier attached to every contract event. + /// Legacy standalone copy of the current event sequence value. /// - /// Consumers compare this value with the last sequence they processed - /// after reconnecting so missing ledger events cannot go unnoticed. + /// New writes persist this as part of `StreamInfo`/`Config` so it survives + /// consolidated-key migrations. Older streams may still have this key until + /// the first `save()` migrates them to the single-key layout. EventSequence, /// Lock for re-entrancy protection and concurrency control. Guard, @@ -66,6 +67,7 @@ pub struct StreamInfo { pub withdrawn: i128, pub paused_at: u64, pub flags: u32, + pub event_sequence: u64, } impl StreamInfo { diff --git a/contracts/stream/src/tests.rs b/contracts/stream/src/tests.rs index f31834c5..a7ced974 100644 --- a/contracts/stream/src/tests.rs +++ b/contracts/stream/src/tests.rs @@ -1552,6 +1552,34 @@ fn initialize_writes_only_config_and_not_legacy_keys() { assert!(!info.is_clawback_enabled()); } +#[test] +fn event_sequence_is_persisted_in_config_and_not_left_as_legacy_state() { + let s = Setup::new(100, 3600, false); + s.advance_secs(100); + s.client.pause(&s.sender); + s.client.resume(&s.sender); + + let (has_config, has_event_sequence) = s.env.as_contract(&s.client.address, || { + let storage = s.env.storage().instance(); + ( + storage.has(&DataKey::Config), + storage.has(&DataKey::EventSequence), + ) + }); + + assert!(has_config, "Config must be present after event-driven updates"); + assert!( + !has_event_sequence, + "EventSequence must be stored in Config rather than as a standalone legacy key" + ); + + let info = s + .env + .as_contract(&s.client.address, || crate::state::load(&s.env)); + assert_eq!(info.event_sequence, 3, "pause/resume emits two events after init"); + assert_eq!(s.client.event_sequence(), 3); +} + #[test] fn state_mutation_writes_only_config_not_legacy_keys() { let s = Setup::new(100, 3600, false); @@ -1624,6 +1652,7 @@ fn save_migrates_legacy_keys_to_config_once() { withdrawn: 0, paused_at: 0, flags: 0, + event_sequence: 0, }, ); });