Fix event sequence persistence in stream config state

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
 }
 

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.

contracts/stream/src/state.rs

@@ -43,6 +43,7 @@ pub fn try_load(env: &Env) -> Result<StreamInfo, Error> {
         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.

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 {

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,
             },
         );
     });