Merge AccumulatorEvents in temporary_store so that transaction effects are accurate

Author: alex-mysten

crates/sui-core/src/accumulators/mod.rs

@@ -162,7 +162,7 @@ impl MergedValueIntermediate {
         match value {
             AccumulatorValue::Integer(_) => Self::SumU128(0),
             AccumulatorValue::IntegerTuple(_, _) => Self::SumU128U128(0, 0),
-            AccumulatorValue::EventDigest(_, _) => Self::Events(vec![]),
+            AccumulatorValue::EventDigest(_) => Self::Events(vec![]),
         }
     }
 
@@ -178,13 +178,15 @@ impl MergedValueIntermediate {
                 *v1 += w1 as u128;
                 *v2 += w2 as u128;
             }
-            (Self::Events(commitments), AccumulatorValue::EventDigest(event_idx, digest)) => {
-                commitments.push(EventCommitment::new(
-                    checkpoint_seq,
-                    transaction_idx,
-                    event_idx,
-                    digest,
-                ));
+            (Self::Events(commitments), AccumulatorValue::EventDigest(event_digests)) => {
+                for (event_idx, digest) in event_digests {
+                    commitments.push(EventCommitment::new(
+                        checkpoint_seq,
+                        transaction_idx,
+                        event_idx,
+                        digest,
+                    ));
+                }
             }
             _ => {
                 fatal!("invalid merge");

crates/sui-core/src/rpc_index.rs

@@ -702,15 +702,17 @@ impl IndexStoreTables {
         for acc in acc_events {
             if let Some(stream_id) =
                 sui_types::accumulator_root::stream_id_from_accumulator_event(&acc)
-                && let AccumulatorValue::EventDigest(idx, _d) = acc.write.value
+                && let AccumulatorValue::EventDigest(event_digests) = &acc.write.value
             {
-                let key = EventIndexKey {
-                    stream_id,
-                    checkpoint_seq,
-                    transaction_idx: tx_idx,
-                    event_index: idx as u32,
-                };
-                entries.push((key, ()));
+                for (idx, _d) in event_digests {
+                    let key = EventIndexKey {
+                        stream_id,
+                        checkpoint_seq,
+                        transaction_idx: tx_idx,
+                        event_index: *idx as u32,
+                    };
+                    entries.push((key, ()));
+                }
             }
         }
 

crates/sui-e2e-tests/tests/address_balance_tests.rs

@@ -2301,3 +2301,143 @@ async fn test_soft_bundle_different_gas_payers() {
 
     test_cluster.trigger_reconfiguration().await;
 }
+
+#[sim_test]
+async fn test_multiple_deposits_merged_in_effects() {
+    let _guard = ProtocolConfig::apply_overrides_for_testing(|_, mut cfg| {
+        cfg.create_root_accumulator_object_for_testing();
+        cfg.enable_accumulators_for_testing();
+        cfg
+    });
+
+    let mut test_cluster = TestClusterBuilder::new().build().await;
+    let rgp = test_cluster.get_reference_gas_price().await;
+    let context = &mut test_cluster.wallet;
+
+    let (sender, gas) = get_sender_and_gas(context).await;
+    let recipient = sender;
+
+    let initial_deposit = make_send_to_account_tx(10000, recipient, sender, gas, rgp);
+    let res = test_cluster
+        .sign_and_execute_transaction(&initial_deposit)
+        .await;
+    let gas = res.effects.unwrap().gas_object().reference.to_object_ref();
+
+    test_cluster.fullnode_handle.sui_node.with(|node| {
+        let state = node.state();
+        let child_object_resolver = state.get_child_object_resolver().as_ref();
+        verify_accumulator_exists(child_object_resolver, recipient, 10000);
+    });
+
+    let mut builder = ProgrammableTransactionBuilder::new();
+
+    let deposit_amounts = vec![1000u64, 2000u64, 3000u64];
+    for amount in &deposit_amounts {
+        let amount_arg = builder.pure(*amount).unwrap();
+        let recipient_arg = builder.pure(recipient).unwrap();
+        let coin = builder.command(Command::SplitCoins(Argument::GasCoin, vec![amount_arg]));
+
+        let Argument::Result(coin_idx) = coin else {
+            panic!("coin is not a result");
+        };
+
+        let coin = Argument::NestedResult(coin_idx, 0);
+
+        builder.programmable_move_call(
+            SUI_FRAMEWORK_PACKAGE_ID,
+            Identifier::new("coin").unwrap(),
+            Identifier::new("send_funds").unwrap(),
+            vec!["0x2::sui::SUI".parse().unwrap()],
+            vec![coin, recipient_arg],
+        );
+    }
+
+    let withdraw_amounts = vec![500u64, 1500u64];
+    for amount in &withdraw_amounts {
+        let withdraw_arg = sui_types::transaction::FundsWithdrawalArg::balance_from_sender(
+            *amount,
+            sui_types::type_input::TypeInput::from(sui_types::gas_coin::GAS::type_tag()),
+        );
+        let withdraw_arg = builder.funds_withdrawal(withdraw_arg).unwrap();
+
+        let amount_arg = builder
+            .pure(move_core_types::u256::U256::from(*amount))
+            .unwrap();
+
+        let split_withdraw_arg = builder.programmable_move_call(
+            SUI_FRAMEWORK_PACKAGE_ID,
+            Identifier::new("funds_accumulator").unwrap(),
+            Identifier::new("withdrawal_split").unwrap(),
+            vec!["0x2::balance::Balance<0x2::sui::SUI>".parse().unwrap()],
+            vec![withdraw_arg, amount_arg],
+        );
+
+        let coin = builder.programmable_move_call(
+            SUI_FRAMEWORK_PACKAGE_ID,
+            Identifier::new("coin").unwrap(),
+            Identifier::new("redeem_funds").unwrap(),
+            vec!["0x2::sui::SUI".parse().unwrap()],
+            vec![split_withdraw_arg],
+        );
+
+        builder.transfer_arg(sender, coin);
+    }
+
+    let tx = TransactionKind::ProgrammableTransaction(builder.finish());
+    let tx_data = TransactionData::new(tx, sender, gas, 10000000, rgp);
+
+    let signed_tx = test_cluster.wallet.sign_transaction(&tx_data).await;
+    let (effects, _) = test_cluster
+        .execute_transaction_return_raw_effects(signed_tx)
+        .await
+        .unwrap();
+
+    assert!(
+        effects.status().is_ok(),
+        "Transaction should succeed, got: {:?}",
+        effects.status()
+    );
+
+    let acc_events = effects.accumulator_events();
+    assert_eq!(
+        acc_events.len(),
+        1,
+        "Should have exactly 1 accumulator event (merged), got: {}",
+        acc_events.len()
+    );
+
+    let event = &acc_events[0];
+    assert_eq!(
+        event.write.address.address, recipient,
+        "Accumulator event should be for the recipient"
+    );
+
+    let deposit_total: u64 = deposit_amounts.iter().sum();
+    let withdraw_total: u64 = withdraw_amounts.iter().sum();
+    let expected_net = deposit_total - withdraw_total;
+
+    match &event.write.value {
+        sui_types::effects::AccumulatorValue::Integer(value) => {
+            assert_eq!(
+                *value, expected_net,
+                "Merged accumulator value should be {} (deposits {} - withdrawals {}), got {}",
+                expected_net, deposit_total, withdraw_total, value
+            );
+        }
+        _ => panic!("Expected Integer accumulator value"),
+    }
+
+    match &event.write.operation {
+        sui_types::effects::AccumulatorOperation::Merge => {}
+        _ => panic!("Expected Merge operation since deposits > withdrawals"),
+    }
+
+    let expected_final_balance = 10000 + expected_net;
+    test_cluster.fullnode_handle.sui_node.with(|node| {
+        let state = node.state();
+        let child_object_resolver = state.get_child_object_resolver().as_ref();
+        verify_accumulator_exists(child_object_resolver, recipient, expected_final_balance);
+    });
+
+    test_cluster.trigger_reconfiguration().await;
+}

crates/sui-e2e-tests/tests/authenticated_events_api_tests.rs

@@ -100,6 +100,38 @@ async fn emit_test_event(
     test_cluster.sign_and_execute_transaction(&tx_data).await;
 }
 
+async fn emit_multiple_test_events(
+    test_cluster: &TestCluster,
+    package_id: ObjectID,
+    sender: SuiAddress,
+    values: Vec<u64>,
+) -> sui_json_rpc_types::SuiTransactionBlockResponse {
+    let rgp = test_cluster.get_reference_gas_price().await;
+    let mut ptb = ProgrammableTransactionBuilder::new();
+    let vals = ptb.pure(values).unwrap();
+    ptb.programmable_move_call(
+        package_id,
+        move_core_types::identifier::Identifier::new("events").unwrap(),
+        move_core_types::identifier::Identifier::new("emit_multiple").unwrap(),
+        vec![],
+        vec![vals],
+    );
+    let gas_object = test_cluster
+        .wallet
+        .get_one_gas_object_owned_by_address(sender)
+        .await
+        .unwrap()
+        .unwrap();
+    let tx_data = TransactionData::new(
+        sui_types::transaction::TransactionKind::ProgrammableTransaction(ptb.finish()),
+        sender,
+        gas_object,
+        50_000_000_000,
+        rgp,
+    );
+    test_cluster.sign_and_execute_transaction(&tx_data).await
+}
+
 async fn query_authenticated_events(
     rpc_url: &str,
     stream_id: &str,
@@ -810,3 +842,83 @@ async fn authenticated_events_backfill_test() {
         "expected 5 authenticated events after backfill, got {count}"
     );
 }
+
+#[sim_test]
+async fn authenticated_events_multiple_events_per_transaction() {
+    let _guard: sui_protocol_config::OverrideGuard =
+        ProtocolConfig::apply_overrides_for_testing(|_, mut cfg| {
+            cfg.enable_authenticated_event_streams_for_testing();
+            cfg
+        });
+
+    let rpc_config = create_rpc_config_with_authenticated_events();
+
+    let test_cluster = TestClusterBuilder::new()
+        .disable_fullnode_pruning()
+        .with_rpc_config(rpc_config)
+        .build()
+        .await;
+
+    let package_id = publish_test_package(&test_cluster).await;
+    let sender = test_cluster.wallet.config.keystore.addresses()[0];
+
+    let _response =
+        emit_multiple_test_events(&test_cluster, package_id, sender, vec![100, 200, 300]).await;
+
+    let mut event_client = EventServiceClient::connect(test_cluster.rpc_url().to_owned())
+        .await
+        .unwrap();
+
+    let mut req = ListAuthenticatedEventsRequest::default();
+    req.stream_id = Some(package_id.to_string());
+    req.start_checkpoint = Some(0);
+    req.page_size = None;
+    req.page_token = None;
+    let response = event_client
+        .list_authenticated_events(req)
+        .await
+        .unwrap()
+        .into_inner();
+
+    let count = response.events.len();
+    assert_eq!(
+        count, 3,
+        "expected 3 authenticated events (all from one transaction), got {count}"
+    );
+
+    #[derive(serde::Deserialize)]
+    struct E {
+        value: u64,
+    }
+
+    let values: Vec<u64> = response
+        .events
+        .iter()
+        .filter_map(|event| {
+            event.event.as_ref().and_then(|e| {
+                e.contents.as_ref().and_then(|c| {
+                    c.value
+                        .as_ref()
+                        .and_then(|v| bcs::from_bytes::<E>(v).ok().map(|e| e.value))
+                })
+            })
+        })
+        .collect();
+
+    assert_eq!(values.len(), 3, "should extract 3 event values");
+    assert!(values.contains(&100), "should contain event with value 100");
+    assert!(values.contains(&200), "should contain event with value 200");
+    assert!(values.contains(&300), "should contain event with value 300");
+
+    let tx_indices: std::collections::HashSet<u32> = response
+        .events
+        .iter()
+        .filter_map(|event| event.transaction_idx)
+        .collect();
+
+    assert_eq!(
+        tx_indices.len(),
+        1,
+        "all events should be from the same transaction"
+    );
+}

crates/sui-e2e-tests/tests/authenticated_events_end_to_end_test.rs

@@ -101,10 +101,12 @@ fn build_event_commitments_from_checkpoint(
 
         let accumulator_events = effects.accumulator_events();
         for acc_event in accumulator_events.iter() {
-            if let AccumulatorValue::EventDigest(event_idx, digest) = &acc_event.write.value {
-                let event_commitment =
-                    EventCommitment::new(checkpoint_seq, idx as u64, *event_idx, *digest);
-                event_commitments.push(event_commitment);
+            if let AccumulatorValue::EventDigest(event_digests) = &acc_event.write.value {
+                for (event_idx, digest) in event_digests {
+                    let event_commitment =
+                        EventCommitment::new(checkpoint_seq, idx as u64, *event_idx, *digest);
+                    event_commitments.push(event_commitment);
+                }
             }
         }
     }

crates/sui-e2e-tests/tests/data/auth_event/sources/auth_event.move

@@ -1,12 +1,20 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-module auth_event::events {
-    use sui::event;
+module auth_event::events;
 
-    public struct E has copy, drop { value: u64 }
+use sui::event;
 
-    public entry fun emit(value: u64) {
-        event::emit_authenticated(E { value });
-    }
-}
+public struct E has copy, drop { value: u64 }
+
+public entry fun emit(value: u64) {
+    event::emit_authenticated(E { value });
+}
+
+public entry fun emit_multiple(values: vector<u64>) {
+    let mut i = 0;
+    while (i < values.length()) {
+        event::emit_authenticated(E { value: values[i] });
+        i = i + 1;
+    };
+}

crates/sui-json-rpc-types/src/sui_transaction.rs

@@ -822,15 +822,15 @@ impl From<AccumulatorOperation> for SuiAccumulatorOperation {
 pub enum SuiAccumulatorValue {
     Integer(u64),
     IntegerTuple(u64, u64),
-    EventDigest(u64 /* event index in the transaction */, Digest),
+    EventDigest(Vec<(u64 /* event index in the transaction */, Digest)>),
 }
 
 impl From<AccumulatorValue> for SuiAccumulatorValue {
     fn from(value: AccumulatorValue) -> Self {
         match value {
             AccumulatorValue::Integer(value) => Self::Integer(value),
             AccumulatorValue::IntegerTuple(value1, value2) => Self::IntegerTuple(value1, value2),
-            AccumulatorValue::EventDigest(idx, value) => Self::EventDigest(idx, value),
+            AccumulatorValue::EventDigest(digests) => Self::EventDigest(digests),
         }
     }
 }

crates/sui-types/src/accumulator_event.rs

@@ -97,7 +97,7 @@ impl AccumulatorEvent {
                     match value {
                         AccumulatorValue::Integer(v) => *v,
                         AccumulatorValue::IntegerTuple(_, _) => fatal!("invalid accumulator value"),
-                        AccumulatorValue::EventDigest(_, _) => fatal!("invalid accumulator value"),
+                        AccumulatorValue::EventDigest(_) => fatal!("invalid accumulator value"),
                     }
                 }
             }