Implement event stream subscriptions; port social example. by afck · Pull Request #3784 · linera-io/linera-protocol

afck · 2025-04-16T14:39:26Z

Motivation

We want to replace pub-sub channels with event streams for better scalability (#365).

Events can already be emitted and read; this PR adds subscriptions.

Proposal

Handle the UpdateStreams system operation. Chain owners can inform applications on a chain about new events in streams published by other chains this way. Also add back subscribe_to_events and unsubscribe_from_events.

Keep a note in the TransactionTracker about any changes to event subscriptions or new events in relevant streams. As part of the same transaction, call process_streams in all subscriber contracts.

When synchronizing a chain, and in the chain listener, detect new events and commit UpdateStreams as appropriate. (This is part of process_inbox now.)

Port the social example to use event streams instead of pub-sub channels, without changing any of the business logic (which should be fixed, too, but that is a separate issue: #3782).

Test Plan

CI; the two pub-sub channel tests have been adapted.

Release Plan

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

Links

examples/social/src/contract.rs

examples/social/src/state.rs

linera-client/src/chain_listener.rs

linera-service/tests/linera_net_tests.rs

examples/social/src/state.rs

afck · 2025-04-18T15:53:09Z

I thought this decision - which events to read - will be done externally as well.

No, the current design (see LIP-20) just updates the contracts about the highest index; the rest is up to the contract.

linera-execution/src/lib.rs

linera-execution/src/wasm/wasmtime.rs

deuszx · 2025-04-21T11:03:44Z

linera-execution/src/system.rs

+                    if subscriptions.next_index >= next_index {
+                        continue;
+                    }
+                    subscriptions.next_index = next_index;


Maybe we should update the index after we act on it.

I moved this line to after we put the information into the transaction tracker.

Or do you mean we should update it only after calling process_streams in all the applications? That would complicate the code quite a bit, because we use the same code to call process_streams, regardless of whether we just processed UpdateStreams or any other operation or message that may have added a new subscription.

deuszx

First pass.

linera-client/src/chain_listener.rs

deuszx · 2025-04-21T12:08:20Z

linera-client/src/chain_listener.rs

+        let Some(subscribers) = self.event_subscribers.get(&chain_id).cloned() else {
+            return Ok(());
+        };
+        for subscriber_id in subscribers {


subscriber_id is a chain so I think it actually makes more sense to call maybe_process_inbox(chain_id) than this - we already established the domain language that chains have inboxes - subscribers don't - so let's stick to that.

You mean you'd rename this to subscriber_chain_id? And subscribers to subscriber_chains?

But then event_subscribers should also be event_subscriber_chains?

If I understand correctly you're saying the words "publisher" and "subscriber" should apply to (ChainId, ApplicationId), and therefore we should use "publishing chain" and "subscribing chain" for ChainId? I can try to make those changes across the board, just want to make sure I'm not misunderstanding.

linera-client/src/chain_listener.rs

linera-execution/src/execution.rs

linera-execution/src/system.rs

deuszx · 2025-04-21T13:22:46Z

linera-execution/src/system.rs

+                    self.context()
+                        .extra()
+                        .get_event(EventId {
+                            chain_id,
+                            stream_id: stream_id.clone(),
+                            index,
+                        })


This is just to make sure that we have that event locally?

Yes. Event values are supposed to be small anyway, but I can replace this with something like contains_event.

I was wondering whether this call here is needed at all? The assumption is that we have the events, it's not like we need to sync it here

In the client maybe, but the server needs to verify that the events the block proposer claims exist actually exist.

linera-core/src/client/mod.rs

ma2bd · 2025-04-22T18:55:50Z

linera-base/src/data_types.rs

+    /// The lowest index of a new event.
+    pub previous_index: u32,


Maybe add somewhere that the new events have indices in the range previous_index..next_index (using Rust conventions)?

oh I just saw the function below. Maybe "see new_indices below"

Done in 5484b8c.

…inera-io#3725)" This reverts commit 572b293.

afck · 2025-04-23T09:43:27Z

I mentioned yesterday that I might modify the social example to remove existing received_posts entries when we unsubscribe. However, received_posts is a map view, so if we re-subscribe, there is no danger of duplicating posts, so it's less important.

(I could do it anyway, if we think these posts should disappear on unsubscribing, but I'm not sure. It's also kind of unrelated to this PR now.)

deuszx · 2025-04-23T10:03:56Z

linera-client/src/chain_listener.rs

+                let publishers = self
+                    .update_event_subscriptions(notification.chain_id)
+                    .await?;
+                if !publishers.is_empty() {


This if shouldn't be necessary - if publishers is empty then listen_recursively will return early.

That's true. But maybe we should still avoid needlessly calling process_inbox.

I think we should be allowed to call process_inbox w/o having the apriori knowledge about whether it's necessary or not - i.e. internally the method should quickly (and easily) verify if the call will be idempotent or not.

But process_inbox can't know whether we just started listening to any new publishing chains, so it would always have to look through them again, and load the relevant data from storage.

deuszx

Another batch.

deuszx · 2025-04-23T10:17:15Z

linera-core/src/client/mod.rs

+        if self.chain_id != self.admin_id {
+            publishers.insert(self.admin_id);
+        }


Do we need this special-casing? I think in general a chain should not subscribe to itself so we should always add the admin_id and always make sure that we don't subscribe to our chain.

I don't understand how we could get rid of that: If our chain is the admin chain, we don't need to synchronize the admin chain again here. If it isn't, we do.

I wanted to say that we would filter out our own chain "deeper" in the callstack but in the other comment you wrote that sometimes we do want.

deuszx · 2025-04-23T10:20:35Z

linera-client/src/chain_listener.rs

+                let publishers = self
+                    .update_event_subscriptions(notification.chain_id)
+                    .await?;
+                if !publishers.is_empty() {


I think we should be allowed to call process_inbox w/o having the apriori knowledge about whether it's necessary or not - i.e. internally the method should quickly (and easily) verify if the call will be idempotent or not.

deuszx · 2025-04-23T10:27:51Z

linera-core/src/client/mod.rs

+            .iter()
+            .map(|(chain_id, _)| *chain_id)
+            .chain(iter::once(self.admin_id))
+            .filter(|chain_id| *chain_id != self.chain_id)


Maybe move this check into synchronize_chain_state and return early. That way we don't need to remember about filtering it out everywhere.

But in other call sites we probably do want to synchronize our own chain.

linera-core/src/client/mod.rs

deuszx · 2025-04-23T10:31:11Z

linera-core/src/client/mod.rs


+    /// Synchronizes all chains that any application on this chain subscribes to.
+    /// We always consider the admin chain a relevant publishing chain, for new epochs.
+    async fn synchronize_publisher_chains(&self) -> Result<(), ChainClientError> {


Do we need _publisher_ in the method name? We synchronize the chain state which is more general.

The point I'm trying to make is whether we need to special case the events/publishers here or maybe we want to synchronize "all chains that we track"?

But here we specifically synchronize those chains that our own chain is subscribed to. We don't want to synchronize all tracked chains here.

linera-execution/src/transaction_tracker.rs

deuszx · 2025-04-23T10:45:27Z

linera-execution/src/transaction_tracker.rs

+            .or_insert_with(|| (previous_index, next_index));
+    }
+
+    pub fn remove_stream_to_process(


Can you add a comment here? This method deletes information but it's not explained when it should be used (or expected to be).

Maybe remove_event_stream ?

Same as below: We're not really removing an event stream, just the note that we need to update a particular app about a particular stream. So I'm not sure the changed name would be better.

linera-execution/src/system.rs

linera-execution/src/runtime.rs

deuszx · 2025-04-23T11:21:24Z

linera-execution/src/transaction_tracker.rs

        self.operation_result = result
    }

+    pub fn add_stream_to_process(


Maybe just add_new_event_stream?

But this isn't necessarily a new stream: It can be called either because of a new subscription to a stream, or because of new events in an existing subscription's stream.

## Motivation With #3784, pub-sub channels are no longer needed. ## Proposal Remove them, and the types `ChannelName`, `Origin`, `Target`, `Medium`, `Destination`, etc. ## Test Plan CI ## Release Plan - Nothing to do / These changes follow the usual release cycle. ## Links - Part of #365. - [reviewer checklist](https://github.com/linera-io/linera-protocol/blob/main/CONTRIBUTING.md#reviewer-checklist)

afck mentioned this pull request Apr 16, 2025

Implement process_streams for EVM. #3785

Closed

afck force-pushed the update-streams branch from 606bf25 to 0d968e1 Compare April 16, 2025 15:43

afck marked this pull request as ready for review April 16, 2025 16:11

afck requested review from bart-linera, deuszx and ma2bd April 16, 2025 16:11

afck force-pushed the update-streams branch from 0d968e1 to 8822fad Compare April 17, 2025 14:20

afck mentioned this pull request Apr 17, 2025

Remove pub-sub channels. #3793

Merged

deuszx reviewed Apr 18, 2025

View reviewed changes

examples/social/src/contract.rs Show resolved Hide resolved

deuszx reviewed Apr 18, 2025

View reviewed changes

examples/social/src/state.rs Outdated Show resolved Hide resolved

deuszx reviewed Apr 18, 2025

View reviewed changes

linera-client/src/chain_listener.rs Show resolved Hide resolved

linera-service/tests/linera_net_tests.rs Show resolved Hide resolved

examples/social/src/state.rs Outdated Show resolved Hide resolved

ma2bd reviewed Apr 18, 2025

View reviewed changes

linera-execution/src/lib.rs Outdated Show resolved Hide resolved

deuszx reviewed Apr 21, 2025

View reviewed changes

afck force-pushed the update-streams branch 2 times, most recently from ddcd0da to 8df8f4a Compare April 22, 2025 11:07

ma2bd approved these changes Apr 22, 2025

View reviewed changes

afck added 12 commits April 23, 2025 10:31

UpdateStreams on process_inbox.

d28f466

Register subscriptions in the chain listener.

2acb9c3

Revert "Remove event subscription functions from contract runtime. (l…

488f1e4

…inera-io#3725)" This reverts commit 572b293.

Verify UpdateStreams on execution.

682e2cc

Add process_streams entry point.

1f112dc

Call process_streams on UpdateStreams.

ef83604

Fix SDK.

541fd42

Port the social example to use events instead of pub-sub channels.

37dbbf1

A few fixes; make the test pass.

662def5

More efficiently detect whether to call process_streams.

87e9194

Update tests

8788f2f

Add TODO link.

54ead93

afck added 6 commits April 23, 2025 10:35

Fix social unit test.

5b55acf

Allow creating chain clients for chains that are not in the wallet.

c657bf9

Address review comments.

da3f558

Pass previous index into process_streams.

ab8621d

Simplify synchronize_from_validators.

5f5d1e4

previous_index comment

5484b8c

afck force-pushed the update-streams branch from 50d0eae to 5484b8c Compare April 23, 2025 08:49

deuszx reviewed Apr 23, 2025

View reviewed changes

afck added 2 commits April 23, 2025 12:43

Remove unused context argument.

d93fe7f

Merge branch 'main' into update-streams

7b0d706

deuszx reviewed Apr 23, 2025

View reviewed changes

Address review comments.

fde4ac7

deuszx approved these changes Apr 23, 2025

View reviewed changes

afck merged commit f9f7fc8 into linera-io:main Apr 23, 2025
26 checks passed

afck deleted the update-streams branch April 23, 2025 15:45

		/// The lowest index of a new event.
		pub previous_index: u32,

Conversation

afck commented Apr 16, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Proposal

Test Plan

Release Plan

Links

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

afck commented Apr 18, 2025

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

deuszx left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

afck commented Apr 23, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

deuszx left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

afck commented Apr 16, 2025 •

edited

Loading