kona-node-service 0.1.3

An implementation of the OP Stack consensus node service
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

//! The RPC server for the sequencer actor.
//! Mostly handles queries from the admin rpc.

use kona_derive::AttributesBuilder;
use kona_protocol::L2BlockInfo;
use kona_rpc::SequencerAdminQuery;
use tokio::sync::watch;

use crate::actors::sequencer::actor::SequencerActorState;

/// Error type for sequencer RPC operations
#[derive(Debug, thiserror::Error)]
pub(super) enum SequencerRpcError {
    /// An error occurred while sending a response to the admin query.
    #[error(
        "Failed to send response to admin query. The response channel was closed, this may mean that the rpc actor was shut down."
    )]
    SendResponse,
}

impl<AB: AttributesBuilder> SequencerActorState<AB> {
    pub(super) async fn handle_admin_query(
        &mut self,
        query: SequencerAdminQuery,
        unsafe_head: &mut watch::Receiver<L2BlockInfo>,
    ) -> Result<(), SequencerRpcError> {
        match query {
            SequencerAdminQuery::SequencerActive(tx) => {
                tx.send(self.is_active).map_err(|_| SequencerRpcError::SendResponse)?;
            }
            SequencerAdminQuery::StartSequencer => {
                info!(target: "sequencer", "Starting sequencer");
                self.is_active = true;
            }
            SequencerAdminQuery::StopSequencer(tx) => {
                info!(target: "sequencer", "Stopping sequencer");
                self.is_active = false;

                tx.send(unsafe_head.borrow().hash())
                    .map_err(|_| SequencerRpcError::SendResponse)?;
            }
            SequencerAdminQuery::ConductorEnabled(tx) => {
                tx.send(self.conductor.is_some()).map_err(|_| SequencerRpcError::SendResponse)?;
            }
            SequencerAdminQuery::SetRecoveryMode(is_active) => {
                self.is_recovery_mode = is_active;
                info!(target: "sequencer", is_active, "Updated recovery mode");
            }
            SequencerAdminQuery::OverrideLeader => {
                if let Some(conductor) = self.conductor.as_mut() {
                    if let Err(e) = conductor.override_leader().await {
                        error!(target: "sequencer::rpc", "Failed to override leader: {}", e);
                    }
                    info!(target: "sequencer", "Overrode leader via the conductor service");
                }
            }
        }

        Ok(())
    }
}