Instead of a key, use an incrementing index for events. (#3672)

## Motivation

The system needs to enforce that events are never overwritten anyway. It
makes them much harder to use for applications if the applications need
to keep track of that.

## Proposal

Replace the general `key: Vec<u8>` with `index: u32`. Instead of the
application, the system now chooses the index, and always uses the next
available number.

## Test Plan

The tests have been updated, and already use system events.

The user application events will be tested more thoroughly later, when
we have an example (probably `social`) that uses them.

## Release Plan

- Nothing to do / These changes follow the usual release cycle.

## Links

- [reviewer
checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

Author: afck

examples/meta-counter/src/contract.rs

@@ -40,11 +40,8 @@ impl Contract for MetaCounterContract {
         // Send a no-op message to ourselves. This is only for testing contracts that send messages
         // on initialization. Since the value is 0 it does not change the counter value.
         let this_chain = self.runtime.chain_id();
-        self.runtime.emit(
-            StreamName(b"announcements".to_vec()),
-            b"updates",
-            b"instantiated",
-        );
+        self.runtime
+            .emit(StreamName(b"announcements".to_vec()), b"instantiated");
         self.runtime.send_message(this_chain, Message::Increment(0));
     }
 

linera-base/src/data_types.rs

@@ -1198,10 +1198,8 @@ impl BcsHashable<'_> for Blob {}
 pub struct Event {
     /// The ID of the stream this event belongs to.
     pub stream_id: StreamId,
-    /// The event key.
-    #[debug(with = "hex_debug")]
-    #[serde(with = "serde_bytes")]
-    pub key: Vec<u8>,
+    /// The event index, i.e. the number of events in the stream before this one.
+    pub index: u32,
     /// The payload data.
     #[debug(with = "hex_debug")]
     #[serde(with = "serde_bytes")]
@@ -1214,7 +1212,7 @@ impl Event {
         EventId {
             chain_id,
             stream_id: self.stream_id.clone(),
-            key: self.key.clone(),
+            index: self.index,
         }
     }
 }

linera-base/src/identifiers.rs

@@ -523,8 +523,8 @@ pub struct EventId {
     pub chain_id: ChainId,
     /// The ID of the stream this event belongs to.
     pub stream_id: StreamId,
-    /// The event key.
-    pub key: Vec<u8>,
+    /// The event index, i.e. the number of events in the stream before this one.
+    pub index: u32,
 }
 
 /// The destination of a message, relative to a particular application.

linera-core/src/chain_worker/state/temporary_changes.rs

@@ -3,8 +3,6 @@
 
 //! Operations that don't persist any changes to the chain state.
 
-use std::collections::HashMap;
-
 use linera_base::{
     data_types::{ApplicationDescription, ArithmeticError, Blob, Timestamp},
     ensure,
@@ -19,7 +17,7 @@ use linera_chain::{
 };
 use linera_execution::{Query, QueryOutcome};
 use linera_storage::{Clock as _, Storage};
-use linera_views::views::{View, ViewError};
+use linera_views::views::View;
 #[cfg(with_testing)]
 use {
     linera_base::{crypto::CryptoHash, data_types::BlockHeight},
@@ -233,26 +231,6 @@ where
             .await?
         };
 
-        // Verify that no event values are overwritten.
-        let mut new_events = HashMap::new();
-        for event in outcome.events.iter().flatten() {
-            let event_id = event.id(chain.chain_id());
-            if let Some(old_value) = new_events.insert(event_id.clone(), &event.value) {
-                ensure!(
-                    *old_value == event.value,
-                    WorkerError::OverwritingEvent(Box::new(event_id))
-                );
-            }
-            match self.0.storage.read_event(event_id.clone()).await {
-                Ok(old_value) => ensure!(
-                    old_value == event.value,
-                    WorkerError::OverwritingEvent(Box::new(event_id))
-                ),
-                Err(ViewError::EventsNotFound(_)) => {}
-                Err(err) => return Err(err.into()),
-            }
-        }
-
         let executed_block = outcome.with(block.clone());
         ensure!(
             !round.is_fast() || !executed_block.outcome.has_oracle_responses(),

linera-core/src/client/mod.rs

@@ -3108,14 +3108,17 @@ where
             (min_epoch, epoch.try_add_one()?)
         };
         let mut epoch_change_ops = Vec::new();
-        while self.has_admin_event(EPOCH_STREAM_NAME, next_epoch).await? {
+        while self
+            .has_admin_event(EPOCH_STREAM_NAME, next_epoch.0)
+            .await?
+        {
             epoch_change_ops.push(Operation::system(SystemOperation::ProcessNewEpoch(
                 next_epoch,
             )));
             next_epoch.try_add_assign_one()?;
         }
         while self
-            .has_admin_event(REMOVED_EPOCH_STREAM_NAME, min_epoch)
+            .has_admin_event(REMOVED_EPOCH_STREAM_NAME, min_epoch.0)
             .await?
         {
             epoch_change_ops.push(Operation::system(SystemOperation::ProcessRemovedEpoch(
@@ -3131,12 +3134,12 @@ where
     async fn has_admin_event(
         &self,
         stream_name: &[u8],
-        key: impl Serialize,
+        index: u32,
     ) -> Result<bool, ChainClientError> {
         let event_id = EventId {
             chain_id: self.admin_id,
             stream_id: StreamId::system(stream_name),
-            key: bcs::to_bytes(&key).unwrap(),
+            index,
         };
         match self.client.storage.read_event(event_id).await {
             Ok(_) => Ok(true),

linera-core/src/unit_tests/wasm_client_tests.rs

@@ -349,7 +349,7 @@ where
                 application_id: application_id2.forget_abi().into(),
                 stream_name: StreamName(b"announcements".to_vec()),
             },
-            key: b"updates".to_vec(),
+            index: 0,
             value: b"instantiated".to_vec(),
         }]]
     );

linera-core/src/unit_tests/worker_tests.rs

@@ -2450,7 +2450,7 @@ where
     let event_id = EventId {
         chain_id: admin_id,
         stream_id: StreamId::system(NEW_EPOCH_STREAM_NAME),
-        key: bcs::to_bytes(&Epoch::from(1)).unwrap(),
+        index: 1,
     };
     let committee_blob = Blob::new(BlobContent::new_committee(bcs::to_bytes(&committee)?));
     // `PublishCommitteeBlob` is tested e.g. in `client_tests::test_change_voting_rights`, so we
@@ -2470,7 +2470,7 @@ where
                 events: vec![
                     vec![Event {
                         stream_id: event_id.stream_id.clone(),
-                        key: event_id.key.clone(),
+                        index: event_id.index,
                         value: bcs::to_bytes(&blob_hash).unwrap(),
                     }],
                     Vec::new(),
@@ -2560,7 +2560,7 @@ where
                             EventId {
                                 chain_id: admin_id,
                                 stream_id: StreamId::system(NEW_EPOCH_STREAM_NAME),
-                                key: bcs::to_bytes(&Epoch::from(1)).unwrap(),
+                                index: 1,
                             },
                             bcs::to_bytes(&blob_hash).unwrap(),
                         ),
@@ -2698,7 +2698,7 @@ where
                 previous_message_blocks: BTreeMap::new(),
                 events: vec![vec![Event {
                     stream_id: StreamId::system(NEW_EPOCH_STREAM_NAME),
-                    key: bcs::to_bytes(&Epoch::from(1)).unwrap(),
+                    index: 1,
                     value: bcs::to_bytes(&committee_blob.id().hash).unwrap(),
                 }]],
                 blobs: vec![Vec::new()],
@@ -2837,13 +2837,13 @@ where
                 events: vec![
                     vec![Event {
                         stream_id: StreamId::system(NEW_EPOCH_STREAM_NAME),
-                        key: bcs::to_bytes(&Epoch::from(1)).unwrap(),
+                        index: 1,
                         value: bcs::to_bytes(&committee_blob.id().hash).unwrap(),
                     }],
                     vec![Event {
-                        value: Vec::new(),
                         stream_id: StreamId::system(REMOVED_EPOCH_STREAM_NAME),
-                        key: bcs::to_bytes(&Epoch::from(0)).unwrap(),
+                        index: 0,
+                        value: Vec::new(),
                     }],
                 ],
                 blobs: vec![Vec::new(); 2],

linera-core/src/worker.rs

@@ -19,7 +19,7 @@ use linera_base::{
     },
     doc_scalar,
     hashed::Hashed,
-    identifiers::{AccountOwner, ApplicationId, BlobId, ChainId, EventId},
+    identifiers::{AccountOwner, ApplicationId, BlobId, ChainId},
     time::timer::{sleep, timeout},
 };
 use linera_chain::{
@@ -210,8 +210,6 @@ pub enum WorkerError {
     BlobsNotFound(Vec<BlobId>),
     #[error("The block proposal is invalid: {0}")]
     InvalidBlockProposal(String),
-    #[error("Trying to overwrite an event: {0:?}")]
-    OverwritingEvent(Box<EventId>),
     #[error("The worker is too busy to handle new chains")]
     FullChainWorkerCache,
     #[error("Failed to join spawned worker task")]

linera-execution/src/execution.rs

@@ -6,11 +6,12 @@ use std::{mem, vec};
 use futures::{FutureExt, StreamExt};
 use linera_base::{
     data_types::{Amount, BlockHeight},
-    identifiers::{Account, AccountOwner, BlobType, Destination},
+    identifiers::{Account, AccountOwner, BlobType, Destination, StreamId},
 };
 use linera_views::{
     context::Context,
     key_value_store_view::KeyValueStoreView,
+    map_view::MapView,
     reentrant_collection_view::HashedReentrantCollectionView,
     views::{ClonableView, View},
 };
@@ -41,6 +42,8 @@ pub struct ExecutionStateView<C> {
     pub system: SystemExecutionStateView<C>,
     /// User applications.
     pub users: HashedReentrantCollectionView<C, ApplicationId, KeyValueStoreView<C>>,
+    /// The number of events in the streams that this chain is writing to.
+    pub stream_event_counts: MapView<C, StreamId, u32>,
 }
 
 /// How to interact with a long-lived service runtime.

linera-execution/src/execution_state_actor.rs

@@ -15,9 +15,11 @@ use linera_base::prometheus_util::{
     exponential_bucket_latencies, register_histogram_vec, MeasureLatency as _,
 };
 use linera_base::{
-    data_types::{Amount, ApplicationPermissions, BlobContent, BlockHeight, Timestamp},
+    data_types::{
+        Amount, ApplicationPermissions, ArithmeticError, BlobContent, BlockHeight, Timestamp,
+    },
     ensure, hex_debug, hex_vec_debug, http,
-    identifiers::{Account, AccountOwner, BlobId, BlobType, ChainId, MessageId},
+    identifiers::{Account, AccountOwner, BlobId, BlobType, ChainId, MessageId, StreamId},
     ownership::ChainOwnership,
 };
 use linera_views::{batch::Batch, context::Context, views::View};
@@ -433,6 +435,19 @@ where
                 callback.respond(self.system.blob_used(None, blob_id).await?)
             }
 
+            NextEventIndex {
+                stream_id,
+                callback,
+            } => {
+                let count = self
+                    .stream_event_counts
+                    .get_mut_or_default(&stream_id)
+                    .await?;
+                let index = *count;
+                *count = count.checked_add(1).ok_or(ArithmeticError::Overflow)?;
+                callback.respond(index)
+            }
+
             GetApplicationPermissions { callback } => {
                 let app_permissions = self.system.application_permissions.get();
                 callback.respond(app_permissions.clone());
@@ -689,6 +704,12 @@ pub enum ExecutionRequest {
         callback: Sender<bool>,
     },
 
+    NextEventIndex {
+        stream_id: StreamId,
+        #[debug(skip)]
+        callback: Sender<u32>,
+    },
+
     GetApplicationPermissions {
         #[debug(skip)]
         callback: Sender<ApplicationPermissions>,

linera-execution/src/lib.rs

@@ -77,8 +77,6 @@ pub use crate::{
     transaction_tracker::{TransactionOutcome, TransactionTracker},
 };
 
-/// The maximum length of an event key in bytes.
-const MAX_EVENT_KEY_LEN: usize = 64;
 /// The maximum length of a stream name.
 const MAX_STREAM_NAME_LEN: usize = 64;
 
@@ -262,8 +260,6 @@ pub enum ExecutionError {
         local_time: Timestamp,
     },
 
-    #[error("Event keys can be at most {MAX_EVENT_KEY_LEN} bytes.")]
-    EventKeyTooLong,
     #[error("Stream names can be at most {MAX_STREAM_NAME_LEN} bytes.")]
     StreamNameTooLong,
     #[error("Blob exceeds size limit")]
@@ -732,13 +728,8 @@ pub trait ContractRuntime: BaseRuntime {
         argument: Vec<u8>,
     ) -> Result<Vec<u8>, ExecutionError>;
 
-    /// Adds a new item to an event stream.
-    fn emit(
-        &mut self,
-        name: StreamName,
-        key: Vec<u8>,
-        value: Vec<u8>,
-    ) -> Result<(), ExecutionError>;
+    /// Adds a new item to an event stream. Returns the new event's index in the stream.
+    fn emit(&mut self, name: StreamName, value: Vec<u8>) -> Result<u32, ExecutionError>;
 
     /// Queries a service.
     fn query_service(

linera-execution/src/revm.rs

@@ -569,12 +569,9 @@ where
             let mut runtime = self.db.runtime.lock().expect("The lock should be possible");
             let stream_name = bcs::to_bytes("ethereum_event")?;
             let stream_name = StreamName(stream_name);
-            for (log, index) in logs.iter().enumerate() {
-                let mut key = bcs::to_bytes(&contract_address)?;
-                bcs::serialize_into(&mut key, origin)?;
-                bcs::serialize_into(&mut key, index)?;
-                let value = bcs::to_bytes(&log)?;
-                runtime.emit(stream_name.clone(), key, value)?;
+            for log in &logs {
+                let value = bcs::to_bytes(&(origin, contract_address, log))?;
+                runtime.emit(stream_name.clone(), value)?;
             }
         }
         Ok(())

linera-execution/src/runtime.rs

@@ -36,7 +36,7 @@ use crate::{
     FinalizeContext, Message, MessageContext, MessageKind, ModuleId, Operation, OperationContext,
     OutgoingMessage, QueryContext, QueryOutcome, ServiceRuntime, TransactionTracker,
     UserContractCode, UserContractInstance, UserServiceCode, UserServiceInstance,
-    MAX_EVENT_KEY_LEN, MAX_STREAM_NAME_LEN,
+    MAX_STREAM_NAME_LEN,
 };
 
 #[cfg(test)]
@@ -1284,17 +1284,8 @@ impl ContractRuntime for ContractSyncRuntimeHandle {
         Ok(value)
     }
 
-    fn emit(
-        &mut self,
-        stream_name: StreamName,
-        key: Vec<u8>,
-        value: Vec<u8>,
-    ) -> Result<(), ExecutionError> {
+    fn emit(&mut self, stream_name: StreamName, value: Vec<u8>) -> Result<u32, ExecutionError> {
         let mut this = self.inner();
-        ensure!(
-            key.len() <= MAX_EVENT_KEY_LEN,
-            ExecutionError::EventKeyTooLong
-        );
         ensure!(
             stream_name.0.len() <= MAX_STREAM_NAME_LEN,
             ExecutionError::StreamNameTooLong
@@ -1304,8 +1295,15 @@ impl ContractRuntime for ContractSyncRuntimeHandle {
             stream_name,
             application_id,
         };
-        this.transaction_tracker.add_event(stream_id, key, value);
-        Ok(())
+        let index = this
+            .execution_state_sender
+            .send_request(|callback| ExecutionRequest::NextEventIndex {
+                stream_id: stream_id.clone(),
+                callback,
+            })?
+            .recv_response()?;
+        this.transaction_tracker.add_event(stream_id, index, value);
+        Ok(index)
     }
 
     fn query_service(