Function background_reader

crates/eth-engine/src/persistence_meter.rs:301–362

Background task that reads persisted block notifications from the execution layer and updates the shared atomic counter. Reconnects automatically on error. If `initial_connection` is `None`, the task begins by connecting via `reconnect()`.

(
    endpoint: SubscriptionEndpoint,
    shared: Arc<SharedState>,
    initial_connection: Option<(Client, jsonrpsee::core::client::Subscription<BlockNumHash>)>,
)

///
/// If `initial_connection` is `None`, the task begins by connecting via `reconnect()`.
async fn background_reader(
    endpoint: SubscriptionEndpoint,
    shared: Arc<SharedState>,
    initial_connection: Option<(Client, jsonrpsee::core::client::Subscription<BlockNumHash>)>,
) {
    let (mut client, mut subscription) = match initial_connection {
        Some(conn) => conn,
        None => reconnect(&endpoint).await,
    };

    loop {
        match subscription.next().await {
            Some(Ok(block)) => {
                // If this is the first notification received since re-establishing
                // a connection, transition to ACTIVE
                let prev_status = shared
                    .subscription_status
                    .swap(SUBSCRIPTION_STATUS_ACTIVE, Ordering::Release);
                if prev_status != SUBSCRIPTION_STATUS_ACTIVE {
                    info!(
                        persisted_block = block.number,
                        "Persistence meter: subscription now active"
                    );
                }
                let prev = shared
                    .last_persisted_block
                    .fetch_max(block.number, Ordering::Release);
                // Only notify if we've advanced, in case they arrive out-of-order
                if block.number > prev {
                    shared.notify.notify_waiters();
                }
                debug!(
                    persisted_block = block.number,
                    "Persistence meter: received persisted block notification"
                );
            }
            Some(Err(error)) => {
                warn!(
                    %error,
                    "Persistence meter: subscription errored; reconnecting"
                );
                shared
                    .subscription_status
                    .store(SUBSCRIPTION_STATUS_RECONNECTING, Ordering::Release);
                shared.notify.notify_waiters();
                drop(subscription);
                drop(client);
                (client, subscription) = reconnect(&endpoint).await;
            }
            None => {
                warn!("Persistence meter: subscription closed; reconnecting");
                shared
                    .subscription_status
                    .store(SUBSCRIPTION_STATUS_RECONNECTING, Ordering::Release);
                shared.notify.notify_waiters();
                drop(subscription);
                drop(client);
                (client, subscription) = reconnect(&endpoint).await;