node/event_source/

event_source_effects.rs

use p2p::channels::{
    signaling::{
        discovery::P2pChannelsSignalingDiscoveryAction,
        exchange::P2pChannelsSignalingExchangeAction,
    },
    snark::P2pChannelsSnarkAction,
    streaming_rpc::P2pChannelsStreamingRpcAction,
    transaction::P2pChannelsTransactionAction,
};
use snark::user_command_verify::{SnarkUserCommandVerifyAction, SnarkUserCommandVerifyError};

#[cfg(feature = "p2p-libp2p")]
use crate::p2p::{MioEvent, P2pNetworkSchedulerAction};
use crate::{
    action::CheckTimeoutsAction,
    block_producer::{
        vrf_evaluator::BlockProducerVrfEvaluatorAction, BlockProducerEvent,
        BlockProducerVrfEvaluatorEvent,
    },
    external_snark_worker_effectful::ExternalSnarkWorkerEvent,
    ledger::{read::LedgerReadAction, write::LedgerWriteAction},
    p2p::{
        channels::{
            best_tip::P2pChannelsBestTipAction, rpc::P2pChannelsRpcAction,
            snark_job_commitment::P2pChannelsSnarkJobCommitmentAction, ChannelId,
            P2pChannelsMessageReceivedAction,
        },
        connection::{
            incoming::P2pConnectionIncomingAction, outgoing::P2pConnectionOutgoingAction,
            P2pConnectionErrorResponse, P2pConnectionResponse,
        },
        disconnection::{P2pDisconnectionAction, P2pDisconnectionReason},
        P2pChannelEvent,
    },
    rpc::{RpcAction, RpcRequest},
    snark::{block_verify::SnarkBlockVerifyAction, work_verify::SnarkWorkVerifyAction, SnarkEvent},
    transition_frontier::genesis::TransitionFrontierGenesisAction,
    BlockProducerAction, ExternalSnarkWorkerAction, Service, Store,
};

use super::{
    Event, EventSourceAction, EventSourceActionWithMeta, LedgerEvent, P2pConnectionEvent, P2pEvent,
};

pub fn event_source_effects<S: Service>(store: &mut Store<S>, action: EventSourceActionWithMeta) {
    let (action, meta) = action.split();
    match action {
        EventSourceAction::ProcessEvents => {
            // This action gets continously called until there are no more
            // events available.
            //
            // Retrieve and process max 1024 events at a time and dispatch
            // `CheckTimeoutsAction` in between `EventSourceProcessEventsAction`
            // calls so that we make sure, that action gets called even
            // if we are continously flooded with events.
            for _ in 0..1024 {
                match store.service.next_event() {
                    Some(event) => {
                        store.dispatch(EventSourceAction::NewEvent { event });
                    }
                    None => break,
                }
            }
            store.dispatch(CheckTimeoutsAction {});
        }
        // "Translate" event into the corresponding action and dispatch it.
        EventSourceAction::NewEvent { event } => match event {
            Event::P2p(e) => match e {
                #[cfg(not(feature = "p2p-libp2p"))]
                P2pEvent::MioEvent(_) => {}
                #[cfg(feature = "p2p-libp2p")]
                P2pEvent::MioEvent(e) => match e {
                    MioEvent::InterfaceDetected(ip) => {
                        store.dispatch(P2pNetworkSchedulerAction::InterfaceDetected { ip });
                    }
                    MioEvent::InterfaceExpired(ip) => {
                        store.dispatch(P2pNetworkSchedulerAction::InterfaceExpired { ip });
                    }
                    MioEvent::ListenerReady { listener } => {
                        store.dispatch(P2pNetworkSchedulerAction::ListenerReady { listener });
                    }
                    MioEvent::ListenerError { listener, error } => {
                        store
                            .dispatch(P2pNetworkSchedulerAction::ListenerError { listener, error });
                    }
                    MioEvent::IncomingConnectionIsReady { listener } => {
                        store.dispatch(P2pNetworkSchedulerAction::IncomingConnectionIsReady {
                            listener,
                        });
                    }
                    MioEvent::IncomingConnectionDidAccept(addr, result) => {
                        store.dispatch(P2pNetworkSchedulerAction::IncomingDidAccept {
                            addr,
                            result,
                        });
                    }
                    MioEvent::OutgoingConnectionDidConnect(addr, result) => {
                        store.dispatch(P2pNetworkSchedulerAction::OutgoingDidConnect {
                            addr,
                            result,
                        });
                    }
                    MioEvent::IncomingDataIsReady(addr) => {
                        store.dispatch(P2pNetworkSchedulerAction::IncomingDataIsReady { addr });
                    }
                    MioEvent::IncomingDataDidReceive(addr, result) => {
                        store.dispatch(P2pNetworkSchedulerAction::IncomingDataDidReceive {
                            addr,
                            result,
                        });
                    }
                    MioEvent::OutgoingDataDidSend(_, _result) => {}
                    MioEvent::ConnectionDidClose(addr, result) => {
                        if let Err(e) = result {
                            store.dispatch(P2pNetworkSchedulerAction::Error {
                                addr,
                                error: p2p::P2pNetworkConnectionError::MioError(e),
                            });
                        } else {
                            store.dispatch(P2pNetworkSchedulerAction::Error {
                                addr,
                                error: p2p::P2pNetworkConnectionError::RemoteClosed,
                            });
                        }
                    }
                    MioEvent::ConnectionDidCloseOnDemand(addr) => {
                        store.dispatch(P2pNetworkSchedulerAction::Prune { addr });
                    }
                },
                P2pEvent::Connection(e) => match e {
                    P2pConnectionEvent::OfferSdpReady(peer_id, res) => match res {
                        Err(error) => {
                            store.dispatch(P2pConnectionOutgoingAction::OfferSdpCreateError {
                                peer_id,
                                error,
                            });
                        }
                        Ok(sdp) => {
                            store.dispatch(P2pConnectionOutgoingAction::OfferSdpCreateSuccess {
                                peer_id,
                                sdp,
                            });
                        }
                    },
                    P2pConnectionEvent::AnswerSdpReady(peer_id, res) => match res {
                        Err(error) => {
                            store.dispatch(P2pConnectionIncomingAction::AnswerSdpCreateError {
                                peer_id,
                                error,
                            });
                        }
                        Ok(sdp) => {
                            store.dispatch(P2pConnectionIncomingAction::AnswerSdpCreateSuccess {
                                peer_id,
                                sdp,
                            });
                        }
                    },
                    P2pConnectionEvent::AnswerReceived(peer_id, res) => match res {
                        P2pConnectionResponse::Accepted(answer) => {
                            store.dispatch(P2pConnectionOutgoingAction::AnswerRecvSuccess {
                                peer_id,
                                answer,
                            });
                        }
                        P2pConnectionResponse::Rejected(reason) => {
                            store.dispatch(P2pConnectionOutgoingAction::AnswerRecvError {
                                peer_id,
                                error: P2pConnectionErrorResponse::Rejected(reason),
                            });
                        }
                        P2pConnectionResponse::SignalDecryptionFailed => {
                            store.dispatch(P2pConnectionOutgoingAction::AnswerRecvError {
                                peer_id,
                                error: P2pConnectionErrorResponse::SignalDecryptionFailed,
                            });
                        }
                        P2pConnectionResponse::InternalError => {
                            store.dispatch(P2pConnectionOutgoingAction::AnswerRecvError {
                                peer_id,
                                error: P2pConnectionErrorResponse::InternalError,
                            });
                        }
                    },
                    P2pConnectionEvent::Finalized(peer_id, res) => match res {
                        Err(error) => {
                            store.dispatch(P2pConnectionOutgoingAction::FinalizeError {
                                peer_id,
                                error: error.clone(),
                            });
                            store.dispatch(P2pConnectionIncomingAction::FinalizeError {
                                peer_id,
                                error,
                            });
                        }
                        Ok(auth) => {
                            let _ = store.dispatch(P2pConnectionOutgoingAction::FinalizeSuccess {
                                peer_id,
                                remote_auth: Some(auth.clone()),
                            }) || store.dispatch(
                                P2pConnectionIncomingAction::FinalizeSuccess {
                                    peer_id,
                                    remote_auth: auth.clone(),
                                },
                            );
                        }
                    },
                    P2pConnectionEvent::Closed(peer_id) => {
                        store.dispatch(P2pDisconnectionAction::PeerClosed { peer_id });
                        store.dispatch(P2pDisconnectionAction::Finish { peer_id });
                    }
                },
                P2pEvent::Channel(e) => match e {
                    P2pChannelEvent::Opened(peer_id, chan_id, res) => match res {
                        Err(err) => {
                            openmina_core::log::warn!(meta.time(); kind = "P2pChannelEvent::Opened", peer_id = peer_id.to_string(), error = err);
                            // TODO(binier): dispatch error action.
                        }
                        Ok(_) => match chan_id {
                            ChannelId::SignalingDiscovery => {
                                store.dispatch(P2pChannelsSignalingDiscoveryAction::Ready {
                                    peer_id,
                                });
                            }
                            ChannelId::SignalingExchange => {
                                store.dispatch(P2pChannelsSignalingExchangeAction::Ready {
                                    peer_id,
                                });
                            }
                            ChannelId::BestTipPropagation => {
                                store.dispatch(P2pChannelsBestTipAction::Ready { peer_id });
                            }
                            ChannelId::TransactionPropagation => {
                                store.dispatch(P2pChannelsTransactionAction::Ready { peer_id });
                            }
                            ChannelId::SnarkPropagation => {
                                store.dispatch(P2pChannelsSnarkAction::Ready { peer_id });
                            }
                            ChannelId::SnarkJobCommitmentPropagation => {
                                store.dispatch(P2pChannelsSnarkJobCommitmentAction::Ready {
                                    peer_id,
                                });
                            }
                            ChannelId::Rpc => {
                                store.dispatch(P2pChannelsRpcAction::Ready { peer_id });
                            }
                            ChannelId::StreamingRpc => {
                                store.dispatch(P2pChannelsStreamingRpcAction::Ready { peer_id });
                            }
                        },
                    },
                    P2pChannelEvent::Sent(peer_id, _, _, res) => {
                        if let Err(err) = res {
                            let reason = P2pDisconnectionReason::P2pChannelSendFailed(err);
                            store.dispatch(P2pDisconnectionAction::Init { peer_id, reason });
                        }
                    }
                    P2pChannelEvent::Received(peer_id, res) => match res {
                        Err(err) => {
                            let reason = P2pDisconnectionReason::P2pChannelReceiveFailed(err);
                            store.dispatch(P2pDisconnectionAction::Init { peer_id, reason });
                        }
                        Ok(message) => {
                            store.dispatch(P2pChannelsMessageReceivedAction {
                                peer_id,
                                message: Box::new(message),
                            });
                        }
                    },
                    P2pChannelEvent::Closed(peer_id, chan_id) => {
                        let reason = P2pDisconnectionReason::P2pChannelClosed(chan_id);
                        store.dispatch(P2pDisconnectionAction::Init { peer_id, reason });
                    }
                },
            },
            Event::Ledger(event) => match event {
                LedgerEvent::Write(response) => {
                    store.dispatch(LedgerWriteAction::Success { response });
                }
                LedgerEvent::Read(id, response) => {
                    store.dispatch(LedgerReadAction::Success { id, response });
                }
            },
            Event::Snark(event) => match event {
                SnarkEvent::BlockVerify(req_id, result) => match result {
                    Err(error) => {
                        store.dispatch(SnarkBlockVerifyAction::Error { req_id, error });
                    }
                    Ok(()) => {
                        store.dispatch(SnarkBlockVerifyAction::Success { req_id });
                    }
                },
                SnarkEvent::WorkVerify(req_id, result) => match result {
                    Err(error) => {
                        store.dispatch(SnarkWorkVerifyAction::Error { req_id, error });
                    }
                    Ok(()) => {
                        store.dispatch(SnarkWorkVerifyAction::Success { req_id });
                    }
                },
                SnarkEvent::UserCommandVerify(req_id, result) => {
                    if let Ok(commands) = result {
                        store.dispatch(SnarkUserCommandVerifyAction::Success { req_id, commands });
                    } else {
                        store.dispatch(SnarkUserCommandVerifyAction::Error {
                            req_id,
                            error: SnarkUserCommandVerifyError::VerificationFailed,
                        });
                    }
                }
            },
            Event::Rpc(rpc_id, e) => match *e {
                RpcRequest::StateGet(filter) => {
                    store.dispatch(RpcAction::GlobalStateGet { rpc_id, filter });
                }
                RpcRequest::StatusGet => {
                    store.dispatch(RpcAction::StatusGet { rpc_id });
                }
                RpcRequest::HeartbeatGet => {
                    store.dispatch(RpcAction::HeartbeatGet { rpc_id });
                }
                RpcRequest::ActionStatsGet(query) => {
                    store.dispatch(RpcAction::ActionStatsGet { rpc_id, query });
                }
                RpcRequest::SyncStatsGet(query) => {
                    store.dispatch(RpcAction::SyncStatsGet { rpc_id, query });
                }
                RpcRequest::BlockProducerStatsGet => {
                    store.dispatch(RpcAction::BlockProducerStatsGet { rpc_id });
                }
                RpcRequest::PeersGet => {
                    store.dispatch(RpcAction::PeersGet { rpc_id });
                }
                RpcRequest::MessageProgressGet => {
                    store.dispatch(RpcAction::MessageProgressGet { rpc_id });
                }
                RpcRequest::P2pConnectionOutgoing(opts) => {
                    store.dispatch(RpcAction::P2pConnectionOutgoingInit { rpc_id, opts });
                }
                RpcRequest::P2pConnectionIncoming(opts) => {
                    store.dispatch(RpcAction::P2pConnectionIncomingInit { rpc_id, opts });
                }
                RpcRequest::ScanStateSummaryGet(query) => {
                    store.dispatch(RpcAction::ScanStateSummaryGetInit { rpc_id, query });
                }
                RpcRequest::SnarkPoolGet => {
                    store.dispatch(RpcAction::SnarkPoolAvailableJobsGet { rpc_id });
                }
                RpcRequest::SnarkPoolJobGet { job_id } => {
                    store.dispatch(RpcAction::SnarkPoolJobGet { rpc_id, job_id });
                }
                RpcRequest::SnarkPoolCompletedJobsGet => {
                    store.dispatch(RpcAction::SnarkPoolCompletedJobsGet { rpc_id });
                }
                RpcRequest::SnarkPoolPendingJobsGet => {
                    store.dispatch(RpcAction::SnarkPoolPendingJobsGet { rpc_id });
                }
                RpcRequest::SnarkerConfig => {
                    store.dispatch(RpcAction::SnarkerConfigGet { rpc_id });
                }
                RpcRequest::SnarkerJobCommit { job_id } => {
                    store.dispatch(RpcAction::SnarkerJobCommit { rpc_id, job_id });
                }
                RpcRequest::SnarkerJobSpec { job_id } => {
                    store.dispatch(RpcAction::SnarkerJobSpec { rpc_id, job_id });
                }
                RpcRequest::SnarkerWorkers => {
                    store.dispatch(RpcAction::SnarkerWorkersGet { rpc_id });
                }
                RpcRequest::HealthCheck => {
                    store.dispatch(RpcAction::HealthCheck { rpc_id });
                }
                RpcRequest::ReadinessCheck => {
                    store.dispatch(RpcAction::ReadinessCheck { rpc_id });
                }
                RpcRequest::DiscoveryRoutingTable => {
                    store.dispatch(RpcAction::DiscoveryRoutingTable { rpc_id });
                }
                RpcRequest::DiscoveryBoostrapStats => {
                    store.dispatch(RpcAction::DiscoveryBoostrapStats { rpc_id });
                }
                RpcRequest::TransactionPoolGet => {
                    store.dispatch(RpcAction::TransactionPool { rpc_id });
                }
                RpcRequest::LedgerAccountsGet(account_query) => {
                    store.dispatch(RpcAction::LedgerAccountsGetInit {
                        rpc_id,
                        account_query,
                    });
                }
                RpcRequest::TransactionInject(commands) => {
                    store.dispatch(RpcAction::TransactionInjectInit { rpc_id, commands });
                }
                RpcRequest::TransitionFrontierUserCommandsGet => {
                    store.dispatch(RpcAction::TransitionFrontierUserCommandsGet { rpc_id });
                }
                RpcRequest::BestChain(max_length) => {
                    store.dispatch(RpcAction::BestChain { rpc_id, max_length });
                }
                RpcRequest::ConsensusConstantsGet => {
                    store.dispatch(RpcAction::ConsensusConstantsGet { rpc_id });
                }
                RpcRequest::TransactionStatusGet(tx) => {
                    store.dispatch(RpcAction::TransactionStatusGet { rpc_id, tx });
                }
                RpcRequest::GetBlock(query) => {
                    store.dispatch(RpcAction::BlockGet { rpc_id, query });
                }
                RpcRequest::PooledUserCommands(query) => {
                    store.dispatch(RpcAction::PooledUserCommands { rpc_id, query });
                }
                RpcRequest::PooledZkappCommands(query) => {
                    store.dispatch(RpcAction::PooledZkappCommands { rpc_id, query });
                }
                RpcRequest::ConsensusTimeGet(query) => {
                    store.dispatch(RpcAction::ConsensusTimeGet { rpc_id, query });
                }
                RpcRequest::GenesisBlockGet => {
                    store.dispatch(RpcAction::GenesisBlock { rpc_id });
                }
                RpcRequest::LedgerStatusGet(ledger_hash) => {
                    store.dispatch(RpcAction::LedgerStatusGetInit {
                        rpc_id,
                        ledger_hash,
                    });
                }
                RpcRequest::LedgerAccountDelegatorsGet(ledger_hash, account_id) => {
                    store.dispatch(RpcAction::LedgerAccountDelegatorsGetInit {
                        rpc_id,
                        ledger_hash,
                        account_id,
                    });
                }
            },
            Event::ExternalSnarkWorker(e) => match e {
                ExternalSnarkWorkerEvent::Started => {
                    store.dispatch(ExternalSnarkWorkerAction::Started);
                }
                ExternalSnarkWorkerEvent::Killed => {
                    store.dispatch(ExternalSnarkWorkerAction::Killed);
                }
                ExternalSnarkWorkerEvent::WorkResult(result) => {
                    store.dispatch(ExternalSnarkWorkerAction::WorkResult { result });
                }
                ExternalSnarkWorkerEvent::WorkError(error) => {
                    store.dispatch(ExternalSnarkWorkerAction::WorkError { error });
                }
                ExternalSnarkWorkerEvent::WorkCancelled => {
                    store.dispatch(ExternalSnarkWorkerAction::WorkCancelled);
                }
                ExternalSnarkWorkerEvent::Error(error) => {
                    store.dispatch(ExternalSnarkWorkerAction::Error {
                        error,
                        permanent: false,
                    });
                }
            },
            Event::BlockProducerEvent(e) => match e {
                BlockProducerEvent::VrfEvaluator(vrf_e) => match vrf_e {
                    BlockProducerVrfEvaluatorEvent::Evaluated(vrf_output_with_hash) => {
                        store.dispatch(
                            BlockProducerVrfEvaluatorAction::ProcessSlotEvaluationSuccess {
                                vrf_output: vrf_output_with_hash.evaluation_result,
                                staking_ledger_hash: vrf_output_with_hash.staking_ledger_hash,
                            },
                        );
                    }
                },
                BlockProducerEvent::BlockProve(block_hash, res) => match res {
                    Err(err) => todo!(
                        "error while trying to produce block proof for block {block_hash} - {err}"
                    ),
                    Ok(proof) => {
                        if store
                            .state()
                            .transition_frontier
                            .genesis
                            .prove_pending_block_hash()
                            .is_some_and(|hash| hash == block_hash)
                        {
                            // TODO(refactor): before this is dispatched, genesis inject must be dispatched
                            store.dispatch(TransitionFrontierGenesisAction::ProveSuccess { proof });
                        } else {
                            store.dispatch(BlockProducerAction::BlockProveSuccess { proof });
                        }
                    }
                },
            },
            Event::GenesisLoad(res) => match res {
                Err(err) => todo!("error while trying to load genesis config/ledger. - {err}"),
                Ok(data) => {
                    store.dispatch(TransitionFrontierGenesisAction::LedgerLoadSuccess { data });
                }
            },
        },
        EventSourceAction::WaitTimeout => {
            store.dispatch(CheckTimeoutsAction {});
        }
        EventSourceAction::WaitForEvents => {}
    }
}