Fix notification forwarding to avoid duplicate messages and handle lag (#4848)

## Motivation

The notification forwarding system has two bugs:

1. **Duplicate messages to exporters**: When validators have multiple
proxies, each proxy forwards notifications to all exporters. This means
exporters receive duplicate notifications (N copies for N proxies),
causing redundant processing and potential data inconsistencies.

2. **Receiver lag crashes forwarding**: The broadcast receiver for
notifications can lag if the consumer is slower than the producer. When
lag occurs, the code treats `RecvError::Lagged` as a fatal error and
exits the forwarding loop, stopping all notification forwarding
permanently even though the channel is still functional.

Both issues reduce system reliability and create operational problems
that are difficult to diagnose.

## Proposal

**Fix duplicate forwarding:**
- Add `exporter_forwarded` flag to track whether exporters have been
included
- Only the first proxy forwards to exporters, subsequent proxies forward
to an empty exporter list
- Each proxy still forwards to its own validator endpoint

**Fix lag handling:**
- Replace `while let Ok()` pattern with explicit `match` on receiver
result
- Handle `RecvError::Lagged(skipped_count)` by logging a warning and
continuing
- Handle `RecvError::Closed` by logging and breaking (legitimate
shutdown)
- Only actually lagged/skipped messages trigger the warning, not every
message

This ensures exporters receive exactly one copy of each notification and
forwarding continues even when temporary lag occurs.

## Test Plan

1. Deploy network with multiple proxies (2+) and exporters
2. Verify exporters receive exactly one copy of each notification (not N
copies)
3. Generate high notification load to induce receiver lag
4. Verify warning logs appear for lagged messages
5. Confirm notification forwarding continues after lag events
6. Verify no duplicate processing in exporter/indexer

## Release Plan

- These changes should be backported to the latest `testnet` branch,
then
    - be released in a validator hotfix.

Author: ndr-ds

linera-rpc/src/grpc/server.rs

@@ -20,7 +20,7 @@ use linera_core::{
     JoinSetExt as _, TaskHandle,
 };
 use linera_storage::Storage;
-use tokio::sync::oneshot;
+use tokio::sync::{broadcast::error::RecvError, oneshot};
 use tokio_util::sync::CancellationToken;
 use tonic::{transport::Channel, Request, Response, Status};
 use tower::{builder::ServiceBuilder, Layer, Service};
@@ -218,17 +218,24 @@ where
             )
         });
 
+        let mut exporter_forwarded = false;
         for proxy in &internal_network.proxies {
             let receiver = notification_sender.subscribe();
             join_set.spawn_task({
                 info!(
                     nickname = state.nickname(),
                     "spawning notifications thread on {} for shard {}", host, shard_id
                 );
+                let exporter_addresses = if exporter_forwarded {
+                    vec![]
+                } else {
+                    exporter_forwarded = true;
+                    internal_network.exporter_addresses()
+                };
                 Self::forward_notifications(
                     state.nickname().to_string(),
                     proxy.internal_address(&internal_network.protocol),
-                    internal_network.exporter_addresses(),
+                    exporter_addresses,
                     receiver,
                 )
             });
@@ -305,7 +312,25 @@ where
             })
             .collect::<Vec<_>>();
 
-        while let Ok(notification) = receiver.recv().await {
+        loop {
+            let notification = match receiver.recv().await {
+                Ok(notification) => notification,
+                Err(RecvError::Lagged(skipped_count)) => {
+                    warn!(
+                        nickname,
+                        skipped_count, "notification receiver lagged, messages were skipped"
+                    );
+                    continue;
+                }
+                Err(RecvError::Closed) => {
+                    warn!(
+                        nickname,
+                        "notification channel closed, exiting forwarding loop"
+                    );
+                    break;
+                }
+            };
+
             let reason = &notification.reason;
             let notification: api::Notification = match notification.clone().try_into() {
                 Ok(notification) => notification,