zakura_network/zakura/block_sync/events.rs
1use super::{request::*, state::*, *};
2
3/// Committed header metadata used by block sync to schedule and validate a body.
4#[derive(Copy, Clone, Debug, Eq, PartialEq)]
5pub struct BlockSyncBlockMeta {
6 /// Header-known block height whose body is missing.
7 pub height: block::Height,
8 /// Committed header hash expected from the downloaded body.
9 pub hash: block::Hash,
10 /// Advisory or confirmed body-size estimate for scheduling.
11 pub size: BlockSizeEstimate,
12}
13
14/// Facts accepted by the block-sync scaffold and later reactor.
15///
16/// The inbound data flow is inverted: a peer's stream-6 frames are decoded and
17/// the download logic runs in the per-peer pipe-routine
18/// ([`PeerRoutine`](super::peer_routine)). Inbound messages no longer flow
19/// through the reactor as a `WireMessage`; the routine forwards only shared
20/// concerns to the reactor over [`RoutineToReactor`].
21#[derive(Clone, Debug)]
22pub enum BlockSyncEvent {
23 /// A peer became available for stream-6 block sync.
24 PeerConnected(BlockSyncPeerSession),
25 /// A peer disconnected; all of its outstanding work is dropped.
26 PeerDisconnected(ZakuraPeerId),
27 /// Header sync advanced the committed header target.
28 HeaderTipChanged {
29 /// Current best header height.
30 height: block::Height,
31 /// Current best header hash.
32 hash: block::Hash,
33 },
34 /// Locally observed finalized or verified-body frontiers changed.
35 StateFrontiersChanged(BlockSyncFrontiers),
36 /// State grew the verified body chain tip.
37 ChainTipGrow(BlockSyncFrontiers),
38 /// State reset the verified body chain tip after a rollback, best-chain switch,
39 /// activation boundary, or coalesced multi-block tip update.
40 ChainTipReset(BlockSyncFrontiers),
41 /// Driver returned the current body-missing, header-known heights with committed hashes.
42 NeededBlocks(Vec<BlockSyncBlockMeta>),
43 /// Node wiring finished applying a submitted block body.
44 BlockApplyFinished {
45 /// Submission token from the matching [`BlockSyncAction::SubmitBlock`].
46 token: BlockApplyToken,
47 /// Submitted block height.
48 height: block::Height,
49 /// Submitted block hash.
50 hash: block::Hash,
51 /// Apply result from the verifier driver.
52 result: BlockApplyResult,
53 /// Locally observed chain frontier after the apply attempt completed.
54 local_frontier: Option<BlockSyncFrontiers>,
55 },
56 /// Node wiring finished or abandoned a `Block` response to an inbound `GetBlocks`.
57 BlockRangeResponseFinished {
58 /// Peer whose served-response slot can be released.
59 peer: ZakuraPeerId,
60 /// First requested height.
61 start_height: block::Height,
62 /// Requested block count.
63 requested_count: u32,
64 /// Number of blocks read from state and sent in the response.
65 returned_count: u32,
66 },
67 /// State returned committed bodies requested by a peer and the reactor should send them.
68 BlockRangeResponseReady {
69 /// Peer whose inbound request is being served.
70 peer: ZakuraPeerId,
71 /// First requested height.
72 start_height: block::Height,
73 /// Requested block count.
74 requested_count: u32,
75 /// Bounded committed blocks returned by state.
76 blocks: Vec<(block::Height, Arc<block::Block>, usize)>,
77 },
78}
79
80/// Result of applying a block-sync body through the verifier driver.
81#[derive(Copy, Clone, Debug, Eq, PartialEq)]
82pub enum BlockApplyResult {
83 /// The block was verified and committed.
84 Committed,
85 /// The verifier reported the block was already committed.
86 Duplicate,
87 /// The verifier rejected the block.
88 Rejected,
89 /// The verifier did not answer before the driver timeout.
90 TimedOut,
91}
92
93/// Monotonic token assigned by the reactor to each verifier submission.
94///
95/// The verifier can return stale duplicate completions after a reset and
96/// resubmission of the same height/hash. Echoing this token lets the reactor
97/// ignore those stale completions instead of releasing a newer in-flight body.
98pub type BlockApplyToken = u64;
99
100/// Actions emitted by the future block-sync reactor for the service seam.
101#[derive(Clone, Debug)]
102pub enum BlockSyncAction {
103 /// Ask node wiring to read `missing_block_bodies`, header hashes, and size hints.
104 QueryNeededBlocks {
105 /// First height to consider for the next local work-buffer refill.
106 from: block::Height,
107 /// Maximum number of heights to scan for this refill.
108 limit: u32,
109 /// Current best header target, used for diagnostics and coalescing.
110 best_header_tip: block::Height,
111 },
112 /// Ask node wiring to read committed bodies for an inbound `GetBlocks`.
113 QueryBlocksByHeightRange {
114 /// Peer that requested the range.
115 peer: ZakuraPeerId,
116 /// First height.
117 start: block::Height,
118 /// Maximum count.
119 count: u32,
120 },
121 /// Parent-first body ready for B3's verifier/commit driver.
122 SubmitBlock {
123 /// Submission token to echo in [`BlockSyncEvent::BlockApplyFinished`].
124 token: BlockApplyToken,
125 /// Block body that is contiguous above `verified_block_tip`.
126 block: Arc<block::Block>,
127 },
128 /// Report peer misbehavior to the supervisor.
129 Misbehavior {
130 /// Misbehaving peer.
131 peer: ZakuraPeerId,
132 /// Reason for reporting.
133 reason: BlockSyncMisbehavior,
134 },
135}
136
137impl BlockSyncAction {
138 /// Stable low-cardinality label for action-channel metrics.
139 pub(super) fn metric_label(&self) -> &'static str {
140 match self {
141 Self::QueryNeededBlocks { .. } => "query_needed_blocks",
142 Self::QueryBlocksByHeightRange { .. } => "query_blocks_by_height_range",
143 Self::SubmitBlock { .. } => "submit_block",
144 Self::Misbehavior { .. } => "misbehavior",
145 }
146 }
147}
148
149/// Block-sync peer-accounting violations.
150#[derive(Copy, Clone, Debug, Eq, PartialEq)]
151pub enum BlockSyncMisbehavior {
152 /// A stream-6 payload was malformed before semantic handling.
153 MalformedMessage,
154 /// A peer sent blocks that were not requested.
155 UnsolicitedBlock,
156 /// A peer requested more blocks than this node advertised it can serve.
157 GetBlocksTooLong,
158 /// A peer exceeded this node's inbound `GetBlocks` serving budget.
159 GetBlocksSpam,
160 /// A peer supplied a body whose hash or size does not match committed metadata.
161 InvalidBlock,
162 /// A peer supplied a body outside the tolerated scheduling-size deviation.
163 SizeMismatch,
164 /// Peer status is internally impossible.
165 InvalidStatus,
166 /// A response terminator arrived without an outstanding range.
167 UnsolicitedDone,
168 /// A peer reported a requested range unavailable.
169 RangeUnavailable,
170 /// A peer sent too many status frames.
171 StatusSpam,
172}
173
174/// The shared routine→reactor channel (per-peer routines inverted data flow).
175///
176/// Each per-peer pipe-routine ([`PeerRoutine`](super::peer_routine)) decodes its
177/// own frames and runs the download logic locally; it forwards only the concerns
178/// that need reactor-global state (serving, status advertisement, the producer,
179/// misbehavior aggregation) over this channel. The sender is `try_send`/bounded
180/// so a busy reactor never backpressures a routine's decode loop into stalling
181/// its transport (the only blocking routine send is the Sequencer `AcceptBody`).
182#[derive(Clone, Debug)]
183pub(super) enum RoutineToReactor {
184 /// A routine received a `Status` and updated its own servable/caps + the
185 /// registry. The reactor advertises our `Status` reply and republishes the
186 /// candidate set. `send_reply` is the routine's rate-meter decision for whether
187 /// a reply is due this time.
188 StatusReceived {
189 /// Peer whose status was applied.
190 peer: ZakuraPeerId,
191 /// Whether the rate meter allows sending a `Status` reply now.
192 send_reply: bool,
193 },
194 /// A peer requested OUR committed blocks (serving). The reactor runs the
195 /// state query + driver path and sends via the peer's session clone.
196 ServeGetBlocks {
197 /// Peer that requested the range.
198 peer: ZakuraPeerId,
199 /// First requested height.
200 start_height: block::Height,
201 /// Requested block count.
202 count: u32,
203 },
204 /// A routine drained its pending work; the producer should re-query (it
205 /// self-gates on low-water, so the ping is idempotent/cheap).
206 RequeryNeeded,
207 /// A routine scored a peer offense that needs the reactor-side
208 /// disconnect/scoring action (serving-side malformed frames report via this
209 /// path; download-side offenses score directly through the `actions`
210 /// channel + the shared registry count).
211 Misbehavior {
212 /// Misbehaving peer.
213 peer: ZakuraPeerId,
214 /// Reason for reporting.
215 reason: BlockSyncMisbehavior,
216 },
217}