Line data Source code
1 : #define _GNU_SOURCE
2 :
3 : #include "../../disco/topo/fd_topo.h"
4 : #include "../../disco/fd_clock_tile.h"
5 : #include <linux/futex.h>
6 : #include "generated/fd_repair_tile_seccomp.h"
7 : #include "../../disco/shred/fd_shred_tile.h"
8 : #include "fd_repair.h"
9 : #include "../../choreo/votor/fd_cert.h"
10 : #include "../../discof/genesis/fd_genesi_tile.h"
11 : #include "../../discof/restore/utils/fd_ssmsg.h"
12 : #include "../../discof/votor/fd_votor_tile.h"
13 : #include "../../discof/replay/fd_replay_tile.h"
14 : #include "../../disco/keyguard/fd_keyload.h"
15 : #include "../../disco/keyguard/fd_keyswitch.h"
16 : #include "../../flamenco/gossip/fd_gossip_message.h"
17 : #include "../../flamenco/leaders/fd_leaders_base.h"
18 : #include "../../flamenco/leaders/fd_multi_epoch_leaders.h"
19 : #include "fd_repair_serde.h"
20 : #include "fd_repair_strat.h"
21 : #include "../../ballet/sha256/fd_sha256.h"
22 : #include "../../ballet/chacha/fd_chacha_rng.h"
23 : #include "../../disco/keyguard/fd_keyguard.h"
24 : #include "../../disco/events/generated/fd_event_gen.h"
25 : #include "../../disco/metrics/fd_metrics.h"
26 : #include "../../disco/net/fd_net_tile.h"
27 : #include "../../disco/shred/fd_rnonce_ss.h"
28 : #include "../../util/net/fd_net_headers.h"
29 : #include "../../util/pod/fd_pod_format.h"
30 : #include "fd_repair_tile.h"
31 :
32 : /* timeout: tracks when a repair request has timed out and it's time to
33 : try again. */
34 :
35 : struct timeout {
36 : long timeout; /* ns */
37 : ulong slot;
38 : ulong blk; /* blk_pool idx, checked against slot when due */
39 : uint fec_idx; /* FEC index, WINDOW_INDEX only */
40 : uchar tag; /* FD_REPAIR_SERDE_TAG_*, WINDOW_INDEX for any FEC set */
41 : uchar attempt;
42 : ushort last_peer; /* 1+strat slot of the peer last asked, 0 if none */
43 : uint rcvd; /* the FEC set's data_rcvd at the last attempt, UINT_MAX once complete */
44 : uint stuck; /* attempts in a row that brought no new shred */
45 : };
46 : typedef struct timeout timeout_t;
47 : FD_STATIC_ASSERT( FD_REPAIR_STRAT_PEER_MAX<USHORT_MAX, last_peer );
48 :
49 : #define PRQ_NAME timeout_prq
50 0 : #define PRQ_T timeout_t
51 : #include "../../util/tmpl/fd_prq.c"
52 :
53 : /* pending: an outstanding request we're waiting for to either receive a
54 : response or get timed out. one-to-many relationship with timeout
55 : (there can be many pendings per timeout). */
56 :
57 0 : #define PENDING_TTL ((long)FD_RNONCE_SS_DELTA_MAX<<24) /* a shred reply's nonce stops verifying after this */
58 0 : #define SIGN_NS (6500L) /* fd_ed25519_sign of a request, test_ed25519 --bench */
59 0 : #define FEC_SET_P90 (38UL) /* p90 fec_set_cnt */
60 :
61 : struct pending {
62 : ulong key;
63 : ulong next; /* pool next */
64 : struct {
65 : ulong prev;
66 : ulong next;
67 : } map;
68 : struct {
69 : ulong prev;
70 : ulong next;
71 : } dlist;
72 : long ts; /* sent timestamp, nanos */
73 : ulong slot;
74 : ulong blk; /* blk_pool idx */
75 : fd_hash_t dmr;
76 : fd_pubkey_t peer; /* id_key of the peer picked */
77 : uint fec_idx; /* FEC index */
78 : uint fec_set_cnt; /* the blk's when sent */
79 : uint tag; /* FD_REPAIR_SERDE_TAG_* of the request */
80 : uint mask; /* bit i set while shred i of the FEC set is unanswered, bit 0 for a HIGHEST or ORPHAN; the pending outlives its last answer */
81 : uint answered; /* shreds matched so far */
82 : uint ip4;
83 : ushort port;
84 : uchar attempt;
85 : uchar pick;
86 : uchar role;
87 : };
88 : typedef struct pending pending_t;
89 :
90 : #define POOL_NAME pending_pool
91 0 : #define POOL_T pending_t
92 : #include "../../util/tmpl/fd_pool.c"
93 :
94 : #define MAP_NAME pending_map
95 0 : #define MAP_ELE_T pending_t
96 0 : #define MAP_PREV map.prev
97 0 : #define MAP_NEXT map.next
98 : #define MAP_MULTI 1
99 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
100 : #include "../../util/tmpl/fd_map_chain.c"
101 :
102 : #define DLIST_NAME pending_dlist
103 : #define DLIST_ELE_T pending_t
104 0 : #define DLIST_PREV dlist.prev
105 0 : #define DLIST_NEXT dlist.next
106 : #include "../../util/tmpl/fd_dlist.c"
107 :
108 : #define SIGN_MAX (16UL)
109 0 : #define SIGN_RESERVE (8UL) /* sign slots only pongs may take */
110 0 : #define PING_MAX (4UL) /* pings per after_credit */
111 :
112 : struct sign {
113 : ulong next;
114 : uchar buf[ FD_REPAIR_SER_MAX ];
115 : ulong sz;
116 : ulong sig_off; /* where the signature goes in buf */
117 : uint saddr; /* 0 lets the net tile choose */
118 : uint daddr;
119 : ushort dport; /* net order */
120 : };
121 : typedef struct sign sign_t;
122 :
123 : #define POOL_NAME sign_pool
124 0 : #define POOL_T sign_t
125 : #include "../../util/tmpl/fd_pool.c"
126 :
127 : #define DEQUE_NAME ping_deque
128 0 : #define DEQUE_T fd_pubkey_t
129 : #include "../../util/tmpl/fd_deque_dynamic.c"
130 :
131 0 : #define PING_TTL (500L*1000000L) /* a ping is answered this long after we sent its peer a request, Agave's 2*DELTA */
132 0 : #define BAN_TTL (10L*1000000000L) /* a peer whose response fails to verify is not picked for this long */
133 0 : #define STUCK_MAX (3U) /* attempts with no new shred before a retry leaves the bucket order */
134 0 : #define NO_PEER (10L*1000000L) /* a req with no peer to pick waits this long */
135 :
136 : static fd_ed25519_sig_t const sig_null = {0};
137 :
138 0 : #define IN_KIND_GENESIS (0)
139 0 : #define IN_KIND_SHRED (1)
140 0 : #define IN_KIND_SNAP (2)
141 0 : #define IN_KIND_VOTOR (3)
142 0 : #define IN_KIND_NET (4)
143 0 : #define IN_KIND_GOSSIP (5)
144 0 : #define IN_KIND_REPLAY (6)
145 0 : #define IN_KIND_SIGN (7)
146 0 : #define IN_KIND_EPOCH (8)
147 :
148 : struct fd_repair_tile {
149 :
150 : /* Signing */
151 :
152 : fd_pubkey_t identity_key;
153 : int staked;
154 : fd_keyswitch_t * keyswitch;
155 : int halt_signing;
156 : sign_t * sign_pool;
157 : ulong sign_cnt;
158 : struct {
159 : ulong idx;
160 : fd_wksp_t * mem;
161 : ulong chunk0;
162 : ulong wmark;
163 : ulong chunk;
164 : ulong free;
165 : } sign[SIGN_MAX];
166 : uchar sign_buf[FD_ED25519_SIG_SZ];
167 :
168 : /* Catchup */
169 :
170 : ulong cert_slot0; /* staked: first cert slot, 0 if none yet */
171 : ulong turbine_slot0; /* unstaked: first turbine slot, 0 if none yet */
172 : ulong catchup_slot; /* highest slot past the window with a verified cert (staked) or signed future shred (unstaked), 0 if none yet */
173 : ulong highest_fec_complete_slot; /* highest slot a FEC set completed for off the network, our leader sets excluded: the cluster tip, 0 if none yet */
174 :
175 : /* Snapshot */
176 :
177 : ulong snapshot_slot; /* root of the snapshot being loaded, applied at FD_SSMSG_DONE */
178 : fd_mr32_t snapshot_blk_mr;
179 :
180 : /* Epoch metadata */
181 :
182 : fd_multi_epoch_leaders_t * mleaders;
183 :
184 : /* Repair data structures */
185 :
186 : ulong seed;
187 : fd_repair_t * repair;
188 : fd_repair_strat_t * strat;
189 :
190 : /* Requests */
191 :
192 : fd_clock_tile_t clock[1];
193 : fd_rng_t rng[1];
194 : fd_chacha_rng_t nonce_rng[1];
195 : fd_rnonce_ss_t rnonce_ss[1];
196 : timeout_t * eager_prq;
197 : timeout_t * notar_prq;
198 : timeout_t * final_prq;
199 : pending_t * pending_pool;
200 : pending_map_t * pending_map;
201 : pending_dlist_t * pending_dlist;
202 :
203 : /* Networking */
204 :
205 : int allow_private_address;
206 : int report_repair_requests;
207 : fd_pubkey_t * ping_deque;
208 : fd_ip4_udp_hdrs_t net_hdr[1];
209 : ushort net_id;
210 : uchar net_buf[ FD_NET_MTU ];
211 :
212 : /* Links */
213 :
214 : int in_kind[ FD_TOPO_MAX_TILE_IN_LINKS ];
215 : struct {
216 : fd_wksp_t * mem;
217 : ulong chunk0;
218 : ulong wmark;
219 : ulong mtu;
220 : ulong sign_idx; /* IN_KIND_SIGN, its sign tile in sign */
221 : } in[ FD_TOPO_MAX_TILE_IN_LINKS ];
222 : ulong chunk; /* reliable ins, after_frag */
223 : fd_replay_message_t replay_buf; /* replay_slot is unreliable */
224 :
225 : ulong net_out_idx;
226 : fd_wksp_t * net_out_mem;
227 : ulong net_out_chunk0;
228 : ulong net_out_wmark;
229 : ulong net_out_chunk;
230 :
231 : ulong replay_out_idx;
232 : fd_wksp_t * replay_out_mem;
233 : ulong replay_out_chunk0;
234 : ulong replay_out_wmark;
235 : ulong replay_out_chunk;
236 :
237 : ulong rserve_out_idx; /* ULONG_MAX if rserve is disabled */
238 : fd_wksp_t * rserve_out_mem;
239 : ulong rserve_out_chunk0;
240 : ulong rserve_out_wmark;
241 : ulong rserve_out_chunk;
242 :
243 : /* Events */
244 :
245 : fd_event_block_received_t * event;
246 :
247 : /* Metrics */
248 :
249 : struct {
250 : ulong pkt_tx;
251 : ulong request_tx[ FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID+1U ]; /* by FD_REPAIR_SERDE_TAG_* */
252 : ulong ping_tx;
253 : ulong slot_highest_repaired;
254 : ulong slot_current;
255 : ulong fec_delivered;
256 : ulong shred_old;
257 : ulong shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_CNT ];
258 : ulong meta_rx;
259 : ulong meta_malformed;
260 : ulong meta_unsolicited;
261 : ulong parent_fec_count_ok;
262 : ulong parent_fec_count_failed;
263 : ulong fec_root_ok;
264 : ulong fec_root_failed;
265 : ulong replay_root_advanced;
266 : ulong replay_missing_fec;
267 : ulong ping_malformed;
268 : ulong ping_unknown_peer;
269 : ulong ping_signature_failed;
270 : ulong req_cancelled;
271 : ulong req_no_peer;
272 : ulong req_no_staked_peer;
273 : ulong timeout_deferred[ FD_METRICS_ENUM_TIMEOUT_DEFER_REASON_CNT ];
274 : ulong timeout_full;
275 : ulong timeout_paused;
276 : ulong sign_halted;
277 : ulong ping_sign_unavailable;
278 : ulong ping_address_mismatch;
279 : ulong ping_unsolicited;
280 : ulong req_hedged;
281 : ulong req_expired;
282 : ulong req_stuck_escaped;
283 : ulong req_late_attempt;
284 : ulong peer_byid_banned;
285 : ulong byid_expired;
286 : fd_histf_t response_latency[ 1 ];
287 : fd_histf_t retry_delay[ 1 ];
288 : fd_histf_t escape_hedge[ 1 ];
289 : } metrics[ 1 ];
290 : };
291 : typedef struct fd_repair_tile fd_repair_tile_t;
292 :
293 : /* push_timeout inserts req into its blk's role's timeout_prq. A full
294 : timeout_prq drops it and rewinds the blk, back into its blk_treap,
295 : so the work is discovered again. */
296 :
297 : static void
298 : push_timeout( fd_repair_tile_t * ctx,
299 : fd_repair_blk_t * blk,
300 0 : timeout_t const * req ) {
301 0 : fd_repair_slot_meta_t const * meta = fd_repair_slot_meta( ctx->repair, blk->slot );
302 0 : ulong idx = fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk );
303 0 : timeout_t * prq = meta->final_blk_idx==idx ? ctx->final_prq : meta->notar_blk_idx==idx ? ctx->notar_prq : ctx->eager_prq;
304 0 : if( FD_LIKELY( timeout_prq_cnt( prq )<timeout_prq_max( prq ) ) ) {
305 0 : timeout_prq_insert( prq, req );
306 0 : return;
307 0 : }
308 0 : ctx->metrics->timeout_full++;
309 0 : if( FD_LIKELY( req->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX ) ) blk->wait_fec_cnt = fd_uint_min( blk->wait_fec_cnt, req->fec_idx );
310 0 : blk->meta_req = ( blk->meta_req && req->tag!=FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT );
311 0 : blk->highest_req = ( blk->highest_req && req->tag!=FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX );
312 0 : blk->orphan_req = ( blk->orphan_req && req->tag!=FD_REPAIR_SERDE_TAG_ORPHAN );
313 0 : fd_repair_blk_treap_insert( ctx->repair, blk );
314 0 : }
315 :
316 : /* poll_timeout polls for one or more requests that have timed out. */
317 :
318 : static fd_repair_blk_t *
319 : poll_timeout( fd_repair_tile_t * ctx,
320 : long now,
321 0 : timeout_t * out ) {
322 0 : timeout_t * prqs[3] = { ctx->final_prq, ctx->notar_prq, ctx->eager_prq };
323 0 : for( ulong c=0UL; c<3; c++ ) {
324 0 : timeout_t * prq = prqs[ c ];
325 0 : while( timeout_prq_cnt( prq ) && prq[ 0 ].timeout<=now ) {
326 0 : timeout_t req = prq[ 0 ];
327 0 : timeout_prq_remove_min( prq );
328 :
329 0 : if( FD_UNLIKELY( req.slot <= ctx->repair->root ) ) continue;
330 0 : fd_repair_blk_t * blk = fd_repair_blk_map_ele_query( ctx->repair->blk_map, &req.slot, NULL, ctx->repair->blk_pool );
331 0 : while( blk && fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk )!=req.blk ) blk = (fd_repair_blk_t *)fd_repair_blk_map_ele_next_const( blk, NULL, ctx->repair->blk_pool ); /* the req outlived its blk */
332 0 : if( FD_UNLIKELY( !blk ) ) { ctx->metrics->req_cancelled++; continue; }
333 :
334 0 : fd_repair_slot_meta_t const * meta = fd_repair_slot_meta( ctx->repair, blk->slot );
335 0 : if( FD_UNLIKELY( meta->final_blk_idx!=req.blk && meta->notar_blk_idx!=req.blk && meta->eager_blk_idx==req.blk && !blk->eager ) ) { ctx->metrics->req_cancelled++; continue; } /* the slot's eager blk, never repaired */
336 0 : fd_repair_fec_t const * fec = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, fd_ulong_if( req.tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX, blk->fecs[ req.fec_idx ], fd_repair_fec_pool_idx_null( ctx->repair->fec_pool ) ) );
337 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT && blk->parent_slot!=ULONG_MAX ) ) { ctx->metrics->req_cancelled++; continue; }
338 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX && fec && fec->fec_complete ) ) { ctx->metrics->req_cancelled++; continue; }
339 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX && !blk->eager && req.fec_idx>=blk->cmpl_fec_cnt ) ) { ctx->metrics->req_cancelled++; continue; } /* of a freed blk whose pool idx was reused */
340 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX && blk->cmpl_fec_cnt ) ) { ctx->metrics->req_cancelled++; continue; }
341 0 : fd_repair_blk_t const * parent = fd_repair_blk_pool_ele_const( ctx->repair->blk_pool, blk->parent );
342 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_ORPHAN && ( blk->parent_slot<=ctx->repair->root || ( parent && parent->parent_slot!=ULONG_MAX ) ) ) ) { blk->orphan_req = 0; ctx->metrics->req_cancelled++; continue; } /* its ancestry showed up, died with the root, or a lower blk asks */
343 0 : long eager_delta = fd_repair_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
344 0 : if( FD_UNLIKELY( req.tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX && blk->eager_ts+eager_delta>now ) ) { /* turbine is not quiet yet */
345 0 : ctx->metrics->timeout_deferred[ FD_METRICS_ENUM_TIMEOUT_DEFER_REASON_V_TURBINE_ACTIVE_IDX ]++;
346 0 : req.timeout = blk->eager_ts+eager_delta;
347 0 : push_timeout( ctx, blk, &req );
348 0 : continue;
349 0 : }
350 0 : *out = req;
351 0 : return blk;
352 0 : }
353 0 : }
354 0 : return NULL;
355 0 : }
356 :
357 : /* discover new requests translating repair treaps => repair requests. */
358 :
359 : static void
360 : discover( fd_repair_tile_t * ctx,
361 : fd_repair_blk_t * blk,
362 0 : long now ) {
363 0 : fd_repair_slot_meta_t const * meta = fd_repair_slot_meta( ctx->repair, blk->slot );
364 0 : ulong idx = fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk );
365 0 : if( FD_UNLIKELY( blk->slot<=ctx->repair->root ) ) return;
366 0 : if( FD_UNLIKELY( meta->final_blk_idx!=idx && meta->notar_blk_idx!=idx && meta->eager_blk_idx==idx && !blk->eager ) ) return; /* the slot's eager blk, never repaired */
367 :
368 0 : int eager = blk->eager;
369 0 : long eager_delta = fd_repair_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
370 0 : timeout_t req = { .slot = blk->slot, .blk = fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk ) };
371 :
372 0 : if( FD_UNLIKELY( !eager && blk->parent_slot==ULONG_MAX && blk->meta_req ) ) return;
373 0 : if( FD_UNLIKELY( !eager && blk->parent_slot==ULONG_MAX ) ) {
374 0 : blk->meta_req = 1;
375 0 : req.tag = FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT;
376 0 : req.timeout = now;
377 0 : push_timeout( ctx, blk, &req );
378 0 : return;
379 0 : }
380 :
381 0 : fd_repair_blk_t const * parent = fd_repair_blk_pool_ele_const( ctx->repair->blk_pool, blk->parent );
382 0 : int orphan_root = !parent || parent->parent_slot==ULONG_MAX; /* the root of its disconnected subtree, the blks under it wait on it */
383 0 : int orphaned = orphan_root && !blk->orphan_req && blk->parent_slot!=ULONG_MAX && blk->parent_slot>ctx->repair->root && memcmp( &blk->parent_dmr, &hash_null, sizeof(fd_mr32_t) ) && !( eager && meta->invalidated );
384 0 : if( FD_UNLIKELY( orphaned ) ) {
385 0 : blk->orphan_req = 1;
386 0 : req.tag = FD_REPAIR_SERDE_TAG_ORPHAN; /* its ancestry is unknown, learn up to 11 of it in one reply */
387 0 : req.timeout = fd_long_if( eager, blk->eager_ts+eager_delta, now ); /* an eager blk's parent may still be on turbine */
388 0 : push_timeout( ctx, blk, &req );
389 0 : }
390 :
391 0 : if( FD_UNLIKELY( eager && !blk->cmpl_fec_cnt && !blk->highest_req ) ) {
392 0 : blk->highest_req = 1;
393 0 : req.tag = FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX;
394 0 : req.timeout = blk->eager_ts+eager_delta;
395 0 : push_timeout( ctx, blk, &req );
396 0 : }
397 :
398 0 : timeout_t * prq = meta->final_blk_idx==idx ? ctx->final_prq : meta->notar_blk_idx==idx ? ctx->notar_prq : ctx->eager_prq;
399 0 : uint cmpl_fec_cnt = fd_uint_if( !!blk->cmpl_fec_cnt, blk->cmpl_fec_cnt, fd_uint_if( eager, blk->rcvd_fec_cnt, 0U ) );
400 0 : uint fec_idx = blk->wait_fec_cnt;
401 0 : for( ; fec_idx<cmpl_fec_cnt && timeout_prq_cnt( prq )<timeout_prq_max( prq ); fec_idx++ ) {
402 0 : fd_repair_fec_t const * fec = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, blk->fecs[ fec_idx ] );
403 0 : if( FD_LIKELY( fec && fec->fec_complete ) ) continue;
404 0 : req.tag = FD_REPAIR_SERDE_TAG_WINDOW_INDEX;
405 0 : req.fec_idx = fec_idx;
406 0 : req.timeout = fd_long_if( eager, ( fec ? fec->first_ts : blk->eager_ts )+eager_delta, now ); /* a partial set waits from its first shred, a missing one from the last set turbine showed */
407 0 : push_timeout( ctx, blk, &req );
408 0 : }
409 0 : blk->wait_fec_cnt = fec_idx;
410 0 : if( FD_LIKELY( timeout_prq_cnt( prq )<timeout_prq_max( prq ) ) ) return;
411 0 : int paused = fd_repair_blk_treap_insert( ctx->repair, blk ); /* timeout_prq filled up, resume later */
412 0 : ctx->metrics->timeout_paused += (ulong)paused;
413 0 : }
414 :
415 : FD_STATIC_ASSERT( FD_FEC_BLK_MAX<=FD_EVENT_BLOCK_RECEIVED_FEC_SETS_MAX, block_received_fec_sets );
416 :
417 : static void
418 : report_block_received( void * ctx_,
419 0 : fd_repair_blk_t const * blk ) {
420 0 : if( FD_LIKELY( !fd_event_tl ) ) return;
421 0 : fd_repair_tile_t * ctx = (fd_repair_tile_t *)ctx_;
422 0 : uint fec_cnt = fd_uint_if( !!blk->cmpl_fec_cnt, blk->cmpl_fec_cnt, blk->rcvd_fec_cnt ); /* an incomplete blk reports what it received */
423 0 : fd_repair_fec_t const * last = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, fec_cnt ? blk->fecs[ fec_cnt-1U ] : fd_repair_fec_pool_idx_null( ctx->repair->fec_pool ) );
424 0 : int is_leader = last && last->is_leader;
425 0 : int is_known = fd_repair_slot_meta( ctx->repair, blk->slot )->eager_blk_idx!=fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk );
426 0 : fd_event_block_received_t * ev = ctx->event;
427 0 : fd_memset( ev, 0, FD_EVENT_BLOCK_RECEIVED_PREFIX_SZ );
428 0 : ev->slot = blk->slot;
429 0 : ev->parent_slot = blk->parent_slot;
430 0 : ev->cancelled = !!blk->telemetry.cancelled_reason;
431 0 : ev->cancelled_reason = fd_int_if( !!blk->telemetry.cancelled_reason, blk->telemetry.cancelled_reason, FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_NOT_CANCELLED );
432 0 : ev->cancelled_time = fd_ulong_if( ev->cancelled, (ulong)blk->telemetry.cancelled_ts, 0UL );
433 0 : ev->notarized = is_known && !is_leader;
434 0 : ev->is_leader = is_leader;
435 0 : ev->caught_up = fd_ulong_max( ctx->metrics->slot_current, ctx->catchup_slot )<=ctx->metrics->slot_highest_repaired+4UL;
436 0 : ev->fec_set_count = blk->cmpl_fec_cnt;
437 0 : ev->slot_complete = !!blk->cmpl_fec_cnt;
438 0 : ev->equivocation_detected_shred = blk->telemetry.cancelled_reason==FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_MERKLE_ROOT_MISMATCH;
439 0 : ev->last_completed_fec_set_index = blk->telemetry.last_cmpl_fec_idx;
440 0 : ev->last_shred_received_time = (ulong)blk->telemetry.last_shred_ts;
441 0 : ev->turbine_shred_received = blk->telemetry.turbine_shred_cnt;
442 0 : ev->repair_shred_received = blk->telemetry.repair_shred_cnt;
443 0 : ev->recovered_shred_count = blk->telemetry.recovered_shred_cnt;
444 0 : if( FD_LIKELY( !is_leader ) ) { /* our own blocks are never repaired */
445 0 : ev->repair_request_window_count = blk->telemetry.req_window_cnt;
446 0 : ev->repair_request_highest_window_count = blk->telemetry.req_highest_cnt;
447 0 : ev->repair_request_orphan_count = blk->telemetry.req_orphan_cnt;
448 0 : ev->repair_request_shred_for_block_id_count = blk->telemetry.req_shred_bid_cnt;
449 0 : ev->repair_request_parent_fec_count = blk->telemetry.req_parent_cnt;
450 0 : ev->repair_request_fec_root_count = blk->telemetry.req_fec_root_cnt;
451 0 : ev->first_repair_request_time = (ulong)blk->telemetry.first_req_ts;
452 0 : ev->repair_shred_responses_received = blk->telemetry.shred_res_cnt;
453 0 : ev->parent_fec_count_responses_received = blk->telemetry.parent_res_cnt;
454 0 : ev->fec_root_responses_received = blk->telemetry.fec_root_res_cnt;
455 0 : ev->last_repair_received_time = (ulong)blk->telemetry.last_shred_res_ts;
456 0 : ev->first_meta_received_time = (ulong)blk->telemetry.first_meta_res_ts;
457 0 : }
458 0 : memcpy( ev->block_id, blk->dmr.uc, sizeof(fd_mr32_t) );
459 0 : memcpy( ev->parent_block_id, blk->parent_dmr.uc, sizeof(fd_mr32_t) );
460 0 : ev->fec_sets_cnt = fd_ulong_min( fec_cnt, FD_EVENT_BLOCK_RECEIVED_FEC_SETS_MAX );
461 0 : for( ulong k=0UL; k<ev->fec_sets_cnt; k++ ) {
462 0 : fd_repair_fec_t const * fec = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, blk->fecs[ k ] );
463 0 : fd_event_block_received_fec_sets_t * f = &ev->fec_sets[ k ];
464 0 : fd_memset( f, 0, sizeof(fd_event_block_received_fec_sets_t) );
465 0 : if( FD_UNLIKELY( !fec ) ) continue; /* an incomplete blk can miss FEC sets */
466 0 : if( FD_LIKELY( fec->fec_complete ) ) memcpy( f->merkle_root, fec->mr.mr32.uc, sizeof(fd_mr32_t) ); /* only the 20-byte key is known before */
467 0 : f->index = (uint)k*FD_FEC_SHRED_CNT;
468 0 : f->data_shreds_received_mask = fd_uint_if( fec->fec_complete, fec->data_rcvd, 0U );
469 0 : f->parity_shreds_received_mask = fec->parity_rcvd;
470 0 : f->repair_shreds_received_mask = fec->repair_rcvd;
471 0 : f->first_shred_received_time = (ulong)fec->first_ts;
472 0 : f->completed_time = (ulong)fec->cmpl_ts;
473 0 : f->final_shred_source = fec->is_leader ? FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_LEADER :
474 0 : !fec->fec_complete ? 0 :
475 0 : !fec->data_rcvd ? 0 :
476 0 : fec->repair_rcvd ? FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_REPAIR :
477 0 : FD_EVENT_BLOCK_RECEIVED_FEC_SETS_FINAL_SHRED_SOURCE_TURBINE;
478 0 : ev->first_shred_received_time = fd_ulong_if( fec->first_ts && ( !ev->first_shred_received_time || (ulong)fec->first_ts<ev->first_shred_received_time ), (ulong)fec->first_ts, ev->first_shred_received_time );
479 0 : }
480 0 : fd_event_report_block_received( ev );
481 0 : }
482 :
483 : static void
484 : report_repair_request( fd_repair_tile_t * ctx,
485 : pending_t const * pending,
486 : uint bit_idx,
487 : long res_ts,
488 : int outcome,
489 0 : int ban ) {
490 0 : if( FD_LIKELY( !ctx->report_repair_requests || !fd_event_tl ) ) return;
491 0 : fd_repair_strat_peer_t const * peer = fd_repair_strat_query( ctx->strat, &pending->peer );
492 0 : fd_event_repair_request_t ev = {0};
493 0 : memcpy( ev.peer_identity, pending->peer.uc, sizeof(fd_pubkey_t) );
494 0 : if( FD_LIKELY( pending->ip4 ) ) {
495 0 : FD_STORE( ushort, ev.peer_ip+10, (ushort)0xffff );
496 0 : FD_STORE( uint, ev.peer_ip+12, pending->ip4 );
497 0 : }
498 0 : switch( pending->tag ) {
499 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_WINDOW_INDEX; break;
500 0 : case FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_HIGHEST_WINDOW_INDEX; break;
501 0 : case FD_REPAIR_SERDE_TAG_ORPHAN: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_ORPHAN; break;
502 0 : case FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_PARENT_AND_FEC_SET_COUNT; break;
503 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_FEC_SET_ROOT; break;
504 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: ev.type = FD_EVENT_REPAIR_REQUEST_TYPE_WINDOW_INDEX_FOR_BLOCK_ID; break;
505 0 : default: FD_LOG_CRIT(( "unhandled request tag %u", pending->tag ));
506 0 : }
507 0 : int fec = pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX || pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID || pending->tag==FD_REPAIR_SERDE_TAG_FEC_SET_ROOT || pending->tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX;
508 0 : memcpy( ev.block_id, pending->dmr.uc, sizeof(fd_mr32_t) );
509 0 : ev.peer_port = pending->port;
510 0 : ev.peer_rank = peer && (ulong)( peer-ctx->strat->cur.peers )<ctx->strat->staked_cnt ? (ushort)( peer-ctx->strat->cur.peers ) : USHORT_MAX;
511 0 : ev.slot = pending->slot;
512 0 : ev.shred_idx = fd_uint_if( fec, pending->fec_idx*FD_FEC_SHRED_CNT+bit_idx, 0U );
513 0 : ev.req_time = (ulong)pending->ts;
514 0 : ev.res_time = (ulong)res_ts;
515 0 : ev.outcome = outcome;
516 0 : ev.ban = ban;
517 0 : ev.attempt = pending->attempt;
518 0 : ev.pick = pending->pick;
519 0 : ev.role = pending->role;
520 0 : fd_event_report_repair_request( &ev );
521 0 : }
522 :
523 :
524 : static inline void
525 : handle_shred( fd_repair_tile_t * ctx,
526 : ulong sig,
527 : uchar const * chunk,
528 0 : long rx_ts ) {
529 0 : if( FD_UNLIKELY( fd_shred_sig_src( sig )==SHRED_SIG_FEC_EVICTED ) ) {
530 0 : fd_fec_evicted_t const * evicted = (fd_fec_evicted_t const *)fd_type_pun_const( chunk );
531 0 : if( FD_UNLIKELY( evicted->slot<=ctx->repair->root || evicted->slot>=ctx->repair->root+ctx->repair->slot_max ) ) return; /* outside the window */
532 0 : if( FD_UNLIKELY( evicted->fec_set_idx>=ctx->repair->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
533 0 : fd_mr_t key; memcpy( key.mr32.uc, evicted->merkle_root.uc, sizeof(fd_mr32_t) );
534 0 : fd_repair_fec_evicted( ctx->repair, evicted->slot, evicted->fec_set_idx, &key );
535 0 : return;
536 0 : }
537 :
538 0 : ulong slot = ((fd_shred_base_t const *)fd_type_pun_const( chunk ))->shred.slot; /* complete and shred messages share the header position */
539 0 : ctx->metrics->shred_old += (ulong)( slot<=ctx->repair->root && fd_shred_sig_src( sig )<=SHRED_SIG_SRC_BAD_REPAIR );
540 0 : ctx->metrics->slot_current = fd_ulong_if( slot>ctx->repair->root && fd_shred_sig_src( sig )<=SHRED_SIG_SRC_BAD_REPAIR, fd_ulong_max( ctx->metrics->slot_current, slot ), ctx->metrics->slot_current );
541 0 : int turbine_first = slot>ctx->repair->root && slot<ctx->repair->root+ctx->repair->slot_max && fd_shred_sig_src( sig )==SHRED_SIG_SRC_TURBINE && !ctx->turbine_slot0;
542 0 : ctx->turbine_slot0 = fd_ulong_if( turbine_first, slot, ctx->turbine_slot0 );
543 0 : if( FD_UNLIKELY( slot<=ctx->repair->root || slot>=ctx->repair->root+ctx->repair->slot_max ) ) return; /* outside the window */
544 :
545 0 : if( FD_UNLIKELY( turbine_first && !ctx->staked ) ) fd_repair_slot_catchup( ctx->repair, slot ); /* unstaked has no certs, turbine is untrusted so staked never does this */
546 :
547 0 : switch( fd_shred_sig_src( sig ) ) {
548 0 : case SHRED_SIG_FEC_COMPLETE:
549 0 : case SHRED_SIG_FEC_COMPLETE_LEADER:
550 0 : case SHRED_SIG_FEC_COMPLETE_AGAIN: {
551 0 : fd_fec_complete_t const * complete = (fd_fec_complete_t const *)fd_type_pun_const( chunk );
552 0 : fd_shred_t const * shred = &complete->last_shred_hdr;
553 :
554 0 : ctx->highest_fec_complete_slot = fd_ulong_max( ctx->highest_fec_complete_slot, shred->slot );
555 0 : if( FD_UNLIKELY( shred->idx>=ctx->repair->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
556 0 : if( FD_UNLIKELY( !shred->data.parent_off || shred->data.parent_off>shred->slot-ctx->repair->root ) ) return; /* parent is the slot itself, or below the root */
557 :
558 0 : fd_repair_fec_t const * fec = fd_repair_fec_complete( ctx->repair, shred, fd_type_pun_const( &complete->merkle_root ), sig==SHRED_SIG_FEC_COMPLETE_LEADER, (uint)complete->turbine_shred_cnt, (uint)complete->repair_shred_cnt, (uint)complete->reconstructed_shred_cnt, complete->data_shred_rcvd, complete->parity_shred_rcvd, complete->repair_data_rcvd, rx_ts );
559 0 : if( FD_UNLIKELY( sig!=SHRED_SIG_FEC_COMPLETE || !fec || !fec->first_ts ) ) break; /* a first completion from the network */
560 :
561 0 : fd_repair_strat_turbine_done( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, shred->slot ), rx_ts-fec->first_ts );
562 0 : break;
563 0 : }
564 0 : case SHRED_SIG_SRC_TURBINE:
565 0 : case SHRED_SIG_SRC_LEADER:
566 0 : case SHRED_SIG_SRC_RECONSTRUCTED:
567 0 : case SHRED_SIG_SRC_REPAIR:
568 0 : case SHRED_SIG_SRC_BAD_REPAIR: {
569 0 : fd_shred_base_t const * msg = (fd_shred_base_t const *)fd_type_pun_const( chunk );
570 0 : fd_shred_t const * shred = &msg->shred;
571 0 : long now = fd_clock_tile_now( ctx->clock );
572 :
573 0 : if( FD_UNLIKELY( fd_shred_sig_res( sig )==SHRED_SIG_RESULT_EQVOC ) ) {
574 0 : ctx->metrics->shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_EQVOC_IDX ] += (ulong)( fd_shred_sig_src( sig )==SHRED_SIG_SRC_REPAIR || fd_shred_sig_src( sig )==SHRED_SIG_SRC_BAD_REPAIR );
575 0 : fd_repair_slot_invalidated( ctx->repair, shred->slot, FD_EVENT_BLOCK_RECEIVED_CANCELLED_REASON_MERKLE_ROOT_MISMATCH, now );
576 0 : return;
577 0 : }
578 :
579 0 : if( FD_UNLIKELY( !fd_shred_is_data( fd_shred_type( shred->variant ) ) ) ) return;
580 0 : if( FD_UNLIKELY( shred->idx>=ctx->repair->fec_blk_max*FD_FEC_SHRED_CNT ) ) return;
581 :
582 : /* A repair shred answers the request whose nonce it carries: the
583 : first one ends that request's pick, the last one ends the
584 : request. A HighestWindowIndex answer that does not complete the
585 : slot arrives as BAD_REPAIR. */
586 :
587 0 : int normal = fd_rnonce_ss_normal_repair( msg->rnonce );
588 0 : int reply = fd_shred_sig_src( sig )==SHRED_SIG_SRC_REPAIR || fd_shred_sig_src( sig )==SHRED_SIG_SRC_BAD_REPAIR;
589 0 : ulong key = (ulong)(uint)slot<<32 | fd_uint_if( normal, shred->idx/FD_FEC_SHRED_CNT, (uint)ctx->repair->fec_blk_max );
590 0 : uint bit = fd_uint_if( normal, 1U<<( shred->idx%FD_FEC_SHRED_CNT ), 1U );
591 0 : pending_t * pending = reply ? pending_map_ele_query( ctx->pending_map, &key, NULL, ctx->pending_pool ) : NULL;
592 0 : for( ; pending; pending = (pending_t *)pending_map_ele_next_const( pending, NULL, ctx->pending_pool ) ) {
593 0 : if( FD_UNLIKELY( !( pending->mask & bit ) ) ) continue; /* already answered, or not asked of this request */
594 0 : if( FD_LIKELY( fd_rnonce_ss_compute( ctx->rnonce_ss, normal, slot, shred->idx, pending->ts )!=msg->rnonce ) ) continue;
595 0 : if( FD_LIKELY( !pending->answered ) ) { /* rtt to the shred tile's network rx, not to this tile's turn at the frag */
596 0 : long rtt = fd_long_max( rx_ts-pending->ts, 1L );
597 0 : fd_repair_strat_request_done( ctx->strat, &pending->peer, rtt, pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID );
598 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)rtt );
599 0 : }
600 0 : pending->mask &= ~bit;
601 0 : pending->answered++;
602 0 : report_repair_request( ctx, pending, (uint)fd_uint_find_lsb( bit ), rx_ts, FD_EVENT_REPAIR_REQUEST_OUTCOME_ANSWERED, FD_EVENT_REPAIR_REQUEST_BAN_NONE );
603 0 : ctx->metrics->shred_rx[ fd_ulong_if( pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_FOR_BLOCK_ID_IDX, FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_SLOT_INDEX_IDX ) ]++;
604 0 : fd_repair_blk_t * blk = fd_repair_blk_map_ele_query( ctx->repair->blk_map, &pending->slot, NULL, ctx->repair->blk_pool );
605 0 : while( blk && memcmp( &blk->dmr, &pending->dmr, sizeof(fd_mr32_t) ) ) blk = (fd_repair_blk_t *)fd_repair_blk_map_ele_next_const( blk, NULL, ctx->repair->blk_pool ); /* not by pending->blk, it may be freed or reused, and a notar blk lives on as eager after dedup */
606 0 : if( FD_LIKELY( blk && ( pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX || pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ) ) ) {
607 0 : blk->telemetry.shred_res_cnt++;
608 0 : blk->telemetry.last_shred_res_ts = rx_ts;
609 0 : }
610 0 : if( FD_LIKELY( !pending->mask ) ) { /* every shred answered */
611 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
612 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
613 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
614 0 : }
615 0 : break;
616 0 : }
617 0 : ctx->metrics->shred_rx[ FD_METRICS_ENUM_SHRED_RESPONSE_MATCH_RESULT_V_UNMATCHED_IDX ] += (ulong)( reply && !pending );
618 :
619 0 : if( FD_UNLIKELY( !shred->data.parent_off || shred->data.parent_off>shred->slot-ctx->repair->root ) ) return; /* parent is the slot itself, or below the root */
620 :
621 0 : fd_repair_shred_insert( ctx->repair, shred, fd_type_pun_const( &msg->merkle_root ), rx_ts );
622 0 : break;
623 0 : }
624 0 : default: FD_LOG_ERR(( "unhandled shred sig src %u", fd_shred_sig_src( sig ) ));
625 0 : }
626 0 : }
627 :
628 : static void
629 : handle_snapshot( fd_repair_tile_t * ctx,
630 : ulong sig,
631 0 : uchar const * chunk ) {
632 0 : switch( fd_ssmsg_sig_message( sig ) ) {
633 0 : case FD_SSMSG_MANIFEST_FULL:
634 0 : case FD_SSMSG_MANIFEST_INCREMENTAL: {
635 0 : fd_snapshot_manifest_t const * manifest = (fd_snapshot_manifest_t const *)fd_type_pun_const( chunk );
636 0 : ctx->snapshot_slot = manifest->slot;
637 0 : memcpy( ctx->snapshot_blk_mr.uc, manifest->block_id, sizeof(fd_mr32_t) );
638 0 : break;
639 0 : }
640 0 : case FD_SSMSG_DONE:
641 0 : fd_repair_init( ctx->repair, ctx->snapshot_slot, &ctx->snapshot_blk_mr, report_block_received, ctx );
642 0 : break;
643 0 : default:
644 0 : FD_LOG_ERR(( "unexpected snapshot message %lu", fd_ssmsg_sig_message( sig ) ));
645 0 : }
646 0 : }
647 :
648 : static inline void
649 : handle_votor( fd_repair_tile_t * ctx,
650 : ulong sig,
651 0 : uchar const * chunk ) {
652 0 : fd_votor_msg_t const * msg = (fd_votor_msg_t const *)fd_type_pun_const( chunk );
653 0 : long now = fd_clock_tile_now( ctx->clock );
654 :
655 0 : switch( sig ) {
656 0 : case FD_VOTOR_SIG_CERTED: {
657 0 : fd_votor_certed_t const * certed = &msg->certed;
658 0 : if( FD_UNLIKELY( certed->slot<=ctx->repair->root || certed->slot>=ctx->repair->root+ctx->repair->slot_max ) ) return; /* outside the window */
659 0 : if( FD_UNLIKELY( ctx->staked && !ctx->cert_slot0 ) ) {
660 0 : ctx->cert_slot0 = certed->slot;
661 0 : fd_repair_slot_catchup( ctx->repair, certed->slot );
662 0 : }
663 0 : switch( certed->kind ) {
664 0 : case FD_CERT_KIND_FINAL:
665 0 : case FD_CERT_KIND_FAST_FINAL: fd_repair_blk_finalized( ctx->repair, certed->slot, fd_type_pun_const( &certed->block_id ), now ); break;
666 0 : case FD_CERT_KIND_NOTAR: fd_repair_blk_notarized( ctx->repair, certed->slot, fd_type_pun_const( &certed->block_id ) ); break;
667 0 : case FD_CERT_KIND_NOTAR_FALLBACK: break;
668 0 : case FD_CERT_KIND_SKIP: break; /* a skip cert alone is not final, see fd_repair_slot_skipped */
669 0 : default: FD_LOG_ERR(( "unhandled cert kind %u", certed->kind ));
670 0 : }
671 0 : break;
672 0 : }
673 0 : case FD_VOTOR_SIG_REPAIR: {
674 0 : ulong slot = msg->repair.slot;
675 0 : if( FD_UNLIKELY( slot<=ctx->repair->root ) ) return;
676 0 : if( FD_UNLIKELY( slot>=ctx->repair->root+ctx->repair->slot_max ) ) {
677 0 : if( FD_UNLIKELY( !ctx->staked || slot<=ctx->catchup_slot ) ) return;
678 0 : ctx->catchup_slot = slot;
679 0 : fd_repair_slot_catchup( ctx->repair, ctx->repair->root+ctx->repair->slot_max );
680 0 : return;
681 0 : }
682 0 : fd_repair_blk_notarized( ctx->repair, slot, fd_type_pun_const( &msg->repair.block_id ) );
683 0 : break;
684 0 : }
685 0 : case FD_VOTOR_SIG_QUORUM: {
686 0 : ulong slot = msg->quorum.slot;
687 0 : if( FD_UNLIKELY( slot<=ctx->repair->root || slot>=ctx->repair->root+ctx->repair->slot_max ) ) return; /* outside the window */
688 0 : switch( msg->quorum.kind ) {
689 0 : case FD_VOTOR_QUORUM_KIND_SAFE_TO_NOTAR: break; /* replay already has the block */
690 0 : case FD_VOTOR_QUORUM_KIND_IMPLICITLY_FINALIZED: fd_repair_blk_finalized( ctx->repair, slot, fd_type_pun_const( &msg->quorum.block_id ), now ); break;
691 0 : case FD_VOTOR_QUORUM_KIND_IMPLICITLY_SKIPPED: fd_repair_slot_skipped ( ctx->repair, slot ); break;
692 0 : default: FD_LOG_ERR(( "unhandled quorum kind %u", msg->quorum.kind ));
693 0 : }
694 0 : break;
695 0 : }
696 0 : default: FD_LOG_ERR(( "unhandled votor sig %lu", sig ));
697 0 : }
698 0 : }
699 :
700 : static inline void
701 : handle_replay( fd_repair_tile_t * ctx,
702 : ulong sig,
703 0 : uchar const * chunk ) {
704 0 : switch( sig ) {
705 0 : case REPLAY_SIG_ROOT_ADVANCED: {
706 0 : fd_replay_root_advanced_t const * root = (fd_replay_root_advanced_t const *)fd_type_pun_const( chunk );
707 0 : ctx->metrics->replay_root_advanced++;
708 0 : if( FD_UNLIKELY( root->slot<=ctx->repair->root ) ) return;
709 0 : fd_repair_root_advanced( ctx->repair, root->slot, fd_type_pun_const( &root->block_id ) );
710 0 : if( FD_UNLIKELY( ctx->catchup_slot>ctx->repair->root ) ) fd_repair_slot_catchup( ctx->repair, fd_ulong_min( ctx->catchup_slot+1UL, ctx->repair->root+ctx->repair->slot_max ) );
711 0 : break;
712 0 : }
713 0 : case REPLAY_SIG_SLOT_DEAD: {
714 0 : fd_replay_slot_dead_t const * dead = (fd_replay_slot_dead_t const *)fd_type_pun_const( chunk );
715 0 : if( FD_UNLIKELY( dead->slot<=ctx->repair->root ) ) return;
716 0 : fd_repair_blk_dead( ctx->repair, dead->slot, fd_type_pun_const( &dead->block_id ), dead->latest_fec_idx );
717 0 : break;
718 0 : }
719 0 : case REPLAY_SIG_MISSING_FEC:
720 0 : ctx->metrics->replay_missing_fec++;
721 0 : ctx->repair->reconsume = 1;
722 0 : break;
723 0 : default: FD_LOG_ERR(( "unhandled replay sig %lu", sig ));
724 0 : }
725 0 : }
726 :
727 : static void
728 : handle_gossip( fd_repair_tile_t * ctx,
729 : ulong sig,
730 0 : fd_gossip_update_message_t const * msg ) {
731 :
732 0 : fd_pubkey_t id_key;
733 0 : memcpy( id_key.uc, msg->origin, sizeof(fd_pubkey_t) );
734 :
735 0 : switch( sig ) {
736 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO: {
737 0 : fd_gossip_socket_t const * socket = &msg->contact_info->value->sockets[FD_GOSSIP_CONTACT_INFO_SOCKET_SERVE_REPAIR];
738 0 : uint ip4 = fd_uint_if( !socket->is_ipv6, socket->ip4, 0U );
739 0 : ushort port = fd_ushort_bswap( socket->port );
740 0 : if( FD_UNLIKELY( fd_pubkey_eq( &id_key, &ctx->identity_key ) || !ip4 || !port || fd_ip4_addr_is_mcast( ip4 ) || ( !ctx->allow_private_address && !fd_ip4_addr_is_public( ip4 ) ) ) ) {
741 0 : fd_repair_strat_contact_info_removed( ctx->strat, &id_key );
742 0 : return;
743 0 : }
744 :
745 0 : fd_repair_strat_contact_info_updated( ctx->strat, &id_key, ip4, port );
746 0 : fd_repair_strat_peer_t const * peer = fd_repair_strat_query( ctx->strat, &id_key );
747 0 : if( FD_UNLIKELY( peer && !peer->ping_ts && !ping_deque_full( ctx->ping_deque ) ) ) ping_deque_push_tail( ctx->ping_deque, id_key ); /* new address */
748 0 : break;
749 0 : }
750 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE:
751 0 : fd_repair_strat_contact_info_removed( ctx->strat, &id_key );
752 0 : break;
753 0 : default:
754 0 : FD_LOG_ERR(( "unexpected gossip sig %lu", sig ));
755 0 : }
756 0 : }
757 :
758 : /* Only answer a ping if it comes from the address of a peer we sent a
759 : request to in the last PING_TTL. */
760 :
761 : static void
762 : handle_ping( fd_repair_tile_t * ctx,
763 : fd_stem_context_t * stem,
764 : uchar const * data,
765 : ulong data_sz,
766 : fd_ip4_hdr_t const * ip4,
767 : fd_udp_hdr_t const * udp,
768 0 : long now ) {
769 0 : fd_pubkey_t from;
770 0 : fd_hash_t token;
771 0 : fd_ed25519_sig_t sig;
772 0 : if( FD_UNLIKELY( !fd_repair_ping_de( data, data_sz, &from, &token, sig ) && !fd_repair_block_id_ping_de( data, data_sz, &from, &token, sig ) ) ) { ctx->metrics->ping_malformed++; return; }
773 :
774 0 : fd_repair_strat_peer_t * peer = fd_repair_strat_query( ctx->strat, &from );
775 0 : if( FD_UNLIKELY( !peer ) ) { ctx->metrics->ping_unknown_peer++; return; }
776 0 : if( FD_UNLIKELY( peer->ip4!=ip4->saddr || fd_ushort_bswap( peer->port )!=udp->net_sport ) ) { ctx->metrics->ping_address_mismatch++; return; }
777 0 : if( FD_UNLIKELY( now-peer->ping_ts>=PING_TTL ) ) { ctx->metrics->ping_unsolicited++; return; }
778 0 : if( FD_UNLIKELY( ctx->halt_signing || !sign_pool_free( ctx->sign_pool ) ) ) { ctx->metrics->ping_sign_unavailable++; return; }
779 :
780 0 : peer->ping_ts = now-PING_TTL; /* one verify per request we sent */
781 0 : fd_sha512_t sha[1];
782 0 : if( FD_UNLIKELY( fd_ed25519_verify( token.uc, sizeof(fd_hash_t), sig, from.uc, sha )!=FD_ED25519_SUCCESS ) ) { ctx->metrics->ping_signature_failed++; return; }
783 :
784 0 : ulong s = 0UL;
785 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
786 0 : uchar * preimage = fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk );
787 0 : memcpy( preimage, "SOLANA_PING_PONG", 16UL );
788 0 : memcpy( preimage+16UL, token.uc, sizeof(fd_hash_t) );
789 0 : fd_repair_pong_t pong;
790 0 : fd_sha256_hash( preimage, 16UL+sizeof(fd_hash_t), pong.hash.uc );
791 :
792 0 : sign_t * pending = sign_pool_ele_acquire( ctx->sign_pool );
793 0 : *pending = (sign_t){ .sig_off = FD_REPAIR_PONG_SER_SZ-FD_ED25519_SIG_SZ, .saddr = ip4->daddr, .daddr = ip4->saddr, .dport = udp->net_sport };
794 0 : pending->sz = fd_repair_pong_ser( &pong, sig_null, &ctx->identity_key, pending->buf );
795 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, pending )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519, ctx->sign[ s ].chunk, 16UL+sizeof(fd_hash_t), 0UL, 0UL, 0UL );
796 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, 16UL+sizeof(fd_hash_t), ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
797 0 : ctx->sign[ s ].free--;
798 0 : ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_PONG ]++;
799 0 : }
800 :
801 : static void
802 : handle_parent_and_fec_set_count_res( fd_repair_tile_t * ctx,
803 : uchar const * data,
804 : ulong data_sz,
805 : fd_ip4_hdr_t const * ip4,
806 : fd_udp_hdr_t const * udp,
807 : long now,
808 0 : long rx_ts ) {
809 0 : uint fec_set_cnt;
810 0 : ulong parent_slot;
811 0 : fd_hash_t parent_dmr;
812 0 : uchar proof[ FD_REPAIR_PROOF_MAX ];
813 0 : ulong proof_sz;
814 0 : uint nonce;
815 0 : ctx->metrics->meta_rx++;
816 0 : if( FD_UNLIKELY( fd_repair_res_parent_fec_set_count_de( data, data_sz, &fec_set_cnt, &parent_slot, &parent_dmr, proof, &proof_sz, &nonce )!=data_sz ) ) { ctx->metrics->meta_malformed++; return; }
817 :
818 0 : pending_t * pending = pending_map_ele_query( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool );
819 0 : if( FD_UNLIKELY( !pending || pending->tag!=FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT ) ) { ctx->metrics->meta_unsolicited++; return; }
820 :
821 0 : fd_repair_strat_peer_t const * peer = fd_repair_strat_query( ctx->strat, &pending->peer );
822 0 : int valid = !fd_repair_res_parent_fec_set_count_verify( fec_set_cnt, parent_slot, &parent_dmr, proof, proof_sz, &pending->dmr );
823 0 : int from = peer && peer->ip4==ip4->saddr && fd_ushort_bswap( peer->port )==udp->net_sport;
824 0 : ctx->metrics->parent_fec_count_failed += (ulong)!valid;
825 0 : if( FD_UNLIKELY( !valid && !from ) ) return;
826 :
827 0 : pending_t sent = *pending;
828 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
829 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
830 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
831 0 : report_repair_request( ctx, &sent, 0U, rx_ts, fd_int_if( valid, FD_EVENT_REPAIR_REQUEST_OUTCOME_ANSWERED, FD_EVENT_REPAIR_REQUEST_OUTCOME_BAD ), fd_int_if( valid, FD_EVENT_REPAIR_REQUEST_BAN_NONE, FD_EVENT_REPAIR_REQUEST_BAN_BAD_PROOF ) );
832 0 : if( FD_UNLIKELY( !valid ) ) {
833 0 : fd_repair_strat_request_failed( ctx->strat, &sent.peer, now+BAN_TTL );
834 0 : return;
835 0 : }
836 0 : fd_repair_strat_request_done( ctx->strat, &sent.peer, now-sent.ts, 0 );
837 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)( now-sent.ts ) );
838 0 : ctx->metrics->parent_fec_count_ok++;
839 :
840 0 : if( FD_UNLIKELY( sent.slot<=ctx->repair->root ) ) return; /* rooted while in flight */
841 0 : if( FD_UNLIKELY( parent_slot<ctx->repair->root ) ) return; /* on a fork below the root */
842 0 : if( FD_UNLIKELY( parent_slot>=sent.slot ) ) return; /* the leader signed a bad parent */
843 :
844 0 : fd_repair_blk_parented( ctx->repair, sent.slot, fd_type_pun_const( &sent.dmr ), parent_slot, fd_type_pun_const( &parent_dmr ), fec_set_cnt, now );
845 0 : }
846 :
847 : static void
848 : handle_fec_set_root_res( fd_repair_tile_t * ctx,
849 : uchar const * data,
850 : ulong data_sz,
851 : fd_ip4_hdr_t const * ip4,
852 : fd_udp_hdr_t const * udp,
853 : long now,
854 0 : long rx_ts ) {
855 0 : fd_mr_t fec_mr = {0};
856 0 : uchar proof[ FD_REPAIR_PROOF_MAX ];
857 0 : ulong proof_sz;
858 0 : uint nonce;
859 0 : ctx->metrics->meta_rx++;
860 0 : if( FD_UNLIKELY( fd_repair_res_fec_set_root_de( data, data_sz, fec_mr.mr20.uc, proof, &proof_sz, &nonce )!=data_sz ) ) { ctx->metrics->meta_malformed++; return; }
861 :
862 0 : pending_t * pending = pending_map_ele_query( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool );
863 0 : if( FD_UNLIKELY( !pending || pending->tag!=FD_REPAIR_SERDE_TAG_FEC_SET_ROOT ) ) { ctx->metrics->meta_unsolicited++; return; }
864 :
865 0 : fd_repair_strat_peer_t const * peer = fd_repair_strat_query( ctx->strat, &pending->peer );
866 0 : int valid = !fd_repair_res_fec_set_root_verify( fec_mr.mr20.uc, proof, proof_sz, &pending->dmr, pending->fec_idx*FD_FEC_SHRED_CNT, pending->fec_set_cnt );
867 0 : int from = peer && peer->ip4==ip4->saddr && fd_ushort_bswap( peer->port )==udp->net_sport;
868 0 : ctx->metrics->fec_root_failed += (ulong)!valid;
869 0 : if( FD_UNLIKELY( !valid && !from ) ) return;
870 :
871 0 : pending_t sent = *pending;
872 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
873 0 : pending_dlist_ele_remove ( ctx->pending_dlist, pending, ctx->pending_pool );
874 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
875 0 : report_repair_request( ctx, &sent, 0U, rx_ts, fd_int_if( valid, FD_EVENT_REPAIR_REQUEST_OUTCOME_ANSWERED, FD_EVENT_REPAIR_REQUEST_OUTCOME_BAD ), fd_int_if( valid, FD_EVENT_REPAIR_REQUEST_BAN_NONE, FD_EVENT_REPAIR_REQUEST_BAN_BAD_PROOF ) );
876 0 : if( FD_UNLIKELY( !valid ) ) {
877 0 : fd_repair_strat_request_failed( ctx->strat, &sent.peer, now+BAN_TTL );
878 0 : return;
879 0 : }
880 0 : fd_repair_strat_request_done( ctx->strat, &sent.peer, now-sent.ts, 0 );
881 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)( now-sent.ts ) );
882 0 : ctx->metrics->fec_root_ok++;
883 :
884 0 : if( FD_UNLIKELY( sent.slot<=ctx->repair->root ) ) return; /* rooted while in flight */
885 0 : fd_repair_fec_notarized( ctx->repair, sent.slot, fd_type_pun_const( &sent.dmr ), sent.fec_idx*FD_FEC_SHRED_CNT, &fec_mr, now );
886 0 : }
887 :
888 : static void
889 : handle_net( fd_repair_tile_t * ctx,
890 : fd_stem_context_t * stem,
891 : ulong sz,
892 0 : long rx_ts ) {
893 0 : uchar * data;
894 0 : ulong data_sz;
895 0 : fd_ip4_hdr_t * ip4;
896 0 : fd_udp_hdr_t * udp;
897 0 : if( FD_UNLIKELY( !fd_ip4_udp_hdr_strip( ctx->net_buf, sz, &data, &data_sz, NULL, &ip4, &udp ) ) ) return;
898 0 : if( FD_UNLIKELY( data_sz<sizeof(uint) ) ) return;
899 :
900 0 : long now = fd_clock_tile_now( ctx->clock );
901 0 : if( FD_UNLIKELY( data_sz==FD_REPAIR_PING_DE_SZ ) ) { /* no response has a ping's size */
902 0 : handle_ping( ctx, stem, data, data_sz, ip4, udp, now );
903 0 : return;
904 0 : }
905 0 : switch( FD_LOAD( uint, data ) ) {
906 0 : case FD_REPAIR_SERDE_TAG_PARENT_FEC_SET_COUNT_RES: handle_parent_and_fec_set_count_res( ctx, data, data_sz, ip4, udp, now, rx_ts ); break;
907 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT_RES: handle_fec_set_root_res ( ctx, data, data_sz, ip4, udp, now, rx_ts ); break;
908 0 : default: break;
909 0 : }
910 0 : }
911 :
912 : /* handle_sign sends the message that was waiting for this signature. */
913 :
914 : static void
915 : handle_sign( fd_repair_tile_t * ctx,
916 : fd_stem_context_t * stem,
917 : ulong in_idx,
918 0 : ulong sig ) {
919 0 : ctx->sign[ ctx->in[ in_idx ].sign_idx ].free++;
920 0 : sign_t * pending = sign_pool_ele( ctx->sign_pool, sig>>32 );
921 0 : memcpy( pending->buf+pending->sig_off, ctx->sign_buf, FD_ED25519_SIG_SZ );
922 :
923 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
924 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
925 0 : *hdr = *ctx->net_hdr;
926 0 : hdr->ip4->saddr = pending->saddr;
927 0 : hdr->ip4->daddr = pending->daddr;
928 0 : hdr->ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
929 0 : hdr->ip4->net_tot_len = fd_ushort_bswap( (ushort)( pending->sz+sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t) ) );
930 0 : hdr->ip4->check = fd_ip4_hdr_check_fast( hdr->ip4 );
931 0 : hdr->udp->net_dport = pending->dport;
932 0 : hdr->udp->net_len = fd_ushort_bswap( (ushort)( pending->sz+sizeof(fd_udp_hdr_t) ) );
933 0 : memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), pending->buf, pending->sz );
934 :
935 0 : ulong sz = pending->sz+sizeof(fd_ip4_udp_hdrs_t);
936 0 : ulong ts = fd_frag_meta_ts_comp( fd_tickcount() );
937 0 : fd_stem_publish( stem, ctx->net_out_idx, fd_disco_netmux_sig( pending->daddr, pending->dport, pending->daddr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) ), ctx->net_out_chunk, sz, 0UL, ts, ts );
938 0 : ctx->net_out_chunk = fd_dcache_compact_next( ctx->net_out_chunk, sz, ctx->net_out_chunk0, ctx->net_out_wmark );
939 0 : sign_pool_ele_release( ctx->sign_pool, pending );
940 0 : ctx->metrics->pkt_tx++;
941 0 : }
942 :
943 : static inline void
944 0 : metrics_write( fd_repair_tile_t * ctx ) {
945 0 : long now = fd_clock_tile_now( ctx->clock );
946 0 : long due = now;
947 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->eager_prq ), fd_long_min( due, ctx->eager_prq[ 0 ].timeout ), due );
948 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->notar_prq ), fd_long_min( due, ctx->notar_prq[ 0 ].timeout ), due );
949 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->final_prq ), fd_long_min( due, ctx->final_prq[ 0 ].timeout ), due );
950 0 : ulong queued = fd_repair_treap_ele_cnt( ctx->repair->eager_treap ) + fd_repair_treap_ele_cnt( ctx->repair->notar_treap ) + fd_repair_treap_ele_cnt( ctx->repair->final_treap );
951 0 : ulong waiting = timeout_prq_cnt( ctx->eager_prq ) + timeout_prq_cnt ( ctx->notar_prq ) + timeout_prq_cnt( ctx->final_prq );
952 :
953 0 : FD_MCNT_SET( REPAIR, PKT_TX, ctx->metrics->pkt_tx );
954 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_WINDOW_INDEX, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_WINDOW_INDEX ] );
955 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_HIGHEST_WINDOW_INDEX, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX ] );
956 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_ORPHAN, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_ORPHAN ] );
957 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_PARENT_FEC_SET_COUNT, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT ] );
958 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_FEC_SET_ROOT, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_FEC_SET_ROOT ] );
959 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_WINDOW_INDEX_FOR_BLOCK_ID, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID ] );
960 0 : FD_MCNT_SET( REPAIR, REQUEST_TX_PONG, ctx->metrics->request_tx[ FD_REPAIR_SERDE_TAG_PONG ] );
961 0 : FD_MCNT_SET( REPAIR, PING_PROBE_TX, ctx->metrics->ping_tx );
962 :
963 0 : FD_MGAUGE_SET( REPAIR, HIGHEST_DELIVERED_SLOT, ctx->metrics->slot_highest_repaired );
964 0 : FD_MGAUGE_SET( REPAIR, HIGHEST_RECEIVED_SLOT, ctx->metrics->slot_current );
965 0 : FD_MGAUGE_SET( REPAIR, FIRST_TURBINE_SLOT, ctx->turbine_slot0 );
966 0 : FD_MGAUGE_SET( REPAIR, CATCHUP_SLOT, ctx->catchup_slot );
967 0 : FD_MGAUGE_SET( REPAIR, BLOCK_PENDING, queued );
968 0 : FD_MGAUGE_SET( REPAIR, REQUEST_IN_FLIGHT, pending_pool_used( ctx->pending_pool ) );
969 0 : FD_MGAUGE_SET( REPAIR, TIMEOUT_PENDING, waiting );
970 0 : FD_MGAUGE_SET( REPAIR, EAGER_DELAY_NANOS, (ulong)fd_repair_strat_eager_ns( ctx->strat, NULL ) );
971 0 : FD_MGAUGE_SET( REPAIR, SIGN_REQUEST_PENDING, sign_pool_used( ctx->sign_pool ) );
972 0 : FD_MGAUGE_SET( REPAIR, TIMEOUT_OVERDUE_NANOS, (ulong)( now-due ) );
973 0 : FD_MGAUGE_SET( REPAIR, FIRST_CERT_SLOT, ctx->cert_slot0 );
974 :
975 0 : FD_MCNT_SET( REPAIR, FEC_DELIVERED, ctx->metrics->fec_delivered );
976 0 : FD_MCNT_SET( REPAIR, TIMEOUT_CANCELLED, ctx->metrics->req_cancelled );
977 0 : FD_MCNT_SET( REPAIR, TIMEOUT_NO_PEER_DEFERRED, ctx->metrics->req_no_peer );
978 0 : FD_MCNT_SET( REPAIR, REQUEST_HEDGED, ctx->metrics->req_hedged );
979 0 : FD_MCNT_SET( REPAIR, REQUEST_EXPIRED, ctx->metrics->req_expired );
980 0 : FD_MCNT_SET( REPAIR, REQUEST_BY_ID_EXPIRED, ctx->metrics->byid_expired );
981 0 : FD_MCNT_SET( REPAIR, REQUEST_STUCK_ESCAPED, ctx->metrics->req_stuck_escaped );
982 0 : FD_MCNT_SET( REPAIR, REQUEST_LATE_RETRIED, ctx->metrics->req_late_attempt );
983 0 : FD_MCNT_SET( REPAIR, PEER_BY_ID_BANNED, ctx->metrics->peer_byid_banned );
984 :
985 0 : FD_MCNT_SET ( REPAIR, TIMEOUT_NO_STAKED_PEER_DEFERRED, ctx->metrics->req_no_staked_peer );
986 0 : FD_MCNT_ENUM_COPY( REPAIR, TIMEOUT_DEFERRED, ctx->metrics->timeout_deferred );
987 0 : FD_MCNT_SET ( REPAIR, TIMEOUT_QUEUE_FULL_DROPPED, ctx->metrics->timeout_full );
988 0 : FD_MCNT_SET ( REPAIR, TIMEOUT_QUEUE_FULL_PAUSED, ctx->metrics->timeout_paused );
989 0 : FD_MCNT_SET ( REPAIR, SIGNING_HALTED, ctx->metrics->sign_halted );
990 :
991 0 : FD_MCNT_SET ( REPAIR, SHRED_OLD_IGNORED, ctx->metrics->shred_old );
992 0 : FD_MCNT_ENUM_COPY( REPAIR, SHRED_RX, ctx->metrics->shred_rx );
993 :
994 0 : FD_MCNT_SET( REPAIR, META_RX, ctx->metrics->meta_rx );
995 0 : FD_MCNT_SET( REPAIR, META_MALFORMED_REJECTED, ctx->metrics->meta_malformed );
996 0 : FD_MCNT_SET( REPAIR, META_UNSOLICITED_REJECTED, ctx->metrics->meta_unsolicited );
997 0 : FD_MCNT_SET( REPAIR, PARENT_FEC_SET_COUNT_ACCEPTED, ctx->metrics->parent_fec_count_ok );
998 0 : FD_MCNT_SET( REPAIR, PARENT_FEC_SET_COUNT_FAILED, ctx->metrics->parent_fec_count_failed );
999 0 : FD_MCNT_SET( REPAIR, FEC_SET_ROOT_ACCEPTED, ctx->metrics->fec_root_ok );
1000 0 : FD_MCNT_SET( REPAIR, FEC_SET_ROOT_FAILED, ctx->metrics->fec_root_failed );
1001 :
1002 0 : FD_MCNT_SET( REPAIR, REPLAY_ROOT_ADVANCED_RX, ctx->metrics->replay_root_advanced );
1003 0 : FD_MCNT_SET( REPAIR, REPLAY_MISSING_FEC_RX, ctx->metrics->replay_missing_fec );
1004 :
1005 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_MALFORMED, ctx->metrics->ping_malformed );
1006 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_UNKNOWN_PEER, ctx->metrics->ping_unknown_peer );
1007 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_SIGNATURE_FAILED, ctx->metrics->ping_signature_failed );
1008 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_SIGN_UNAVAILABLE, ctx->metrics->ping_sign_unavailable );
1009 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_ADDRESS_MISMATCH, ctx->metrics->ping_address_mismatch );
1010 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_UNSOLICITED, ctx->metrics->ping_unsolicited );
1011 :
1012 0 : FD_MHIST_COPY( REPAIR, RESPONSE_LATENCY_NANOS, ctx->metrics->response_latency );
1013 0 : FD_MHIST_COPY( REPAIR, RETRY_DELAY_NANOS, ctx->metrics->retry_delay );
1014 0 : FD_MHIST_COPY( REPAIR, ESCAPE_PEER_HEDGE_DELAY_NANOS, ctx->metrics->escape_hedge );
1015 0 : }
1016 :
1017 : FD_FN_CONST static inline ulong
1018 0 : scratch_align( void ) {
1019 0 : return 128UL;
1020 0 : }
1021 :
1022 : FD_FN_PURE static inline ulong
1023 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
1024 0 : ulong sign_max = tile->repair.repair_sign_depth*tile->repair.repair_sign_cnt;
1025 0 : ulong timeout_max = tile->repair.slot_max*(FEC_SET_P90+1UL); /* one blk of the role per slot, each a FEC set req for its p90 FEC sets and a HIGHEST_WINDOW_INDEX or PARENT_AND_FEC_SET_COUNT, a full prq rewinds its blk */
1026 0 : ulong pending_max = tile->repair.repair_sign_cnt*(ulong)( PENDING_TTL/SIGN_NS ); /* what the sign tiles can sign within PENDING_TTL, each pending is at least one */
1027 0 : ulong chain_cnt = pending_map_chain_cnt_est( pending_max );
1028 0 : ulong l = FD_LAYOUT_INIT;
1029 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_repair_tile_t), sizeof(fd_repair_tile_t) );
1030 0 : l = FD_LAYOUT_APPEND( l, fd_repair_align(), fd_repair_footprint( tile->repair.slot_max, tile->repair.fec_max, tile->repair.max_shreds_per_block/FD_FEC_SHRED_CNT ) );
1031 0 : l = FD_LAYOUT_APPEND( l, fd_repair_strat_align(), fd_repair_strat_footprint() );
1032 0 : l = FD_LAYOUT_APPEND( l, fd_multi_epoch_leaders_align(), fd_multi_epoch_leaders_footprint() );
1033 0 : l = FD_LAYOUT_APPEND( l, ping_deque_align(), ping_deque_footprint( FD_REPAIR_STRAT_PEER_MAX ) );
1034 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1035 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1036 0 : l = FD_LAYOUT_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1037 0 : l = FD_LAYOUT_APPEND( l, pending_pool_align(), pending_pool_footprint( pending_max ) );
1038 0 : l = FD_LAYOUT_APPEND( l, pending_map_align(), pending_map_footprint( chain_cnt ) );
1039 0 : l = FD_LAYOUT_APPEND( l, pending_dlist_align(), pending_dlist_footprint() );
1040 0 : l = FD_LAYOUT_APPEND( l, sign_pool_align(), sign_pool_footprint( sign_max ) );
1041 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_event_block_received_t), sizeof(fd_event_block_received_t) );
1042 0 : return FD_LAYOUT_FINI( l, scratch_align() );
1043 0 : }
1044 :
1045 : static inline void
1046 : after_credit( fd_repair_tile_t * ctx,
1047 : fd_stem_context_t * stem,
1048 : int * opt_poll_in,
1049 0 : int * charge_busy ) {
1050 :
1051 : /* Send one reassembled FEC to replay, then repair, which publishes
1052 : only on unreliable links. */
1053 :
1054 0 : if( FD_LIKELY( !fd_repair_deque_empty( ctx->repair->reasm_deque ) ) ) {
1055 0 : *opt_poll_in = 0;
1056 :
1057 0 : fd_repair_deque_t out = fd_repair_deque_pop_head( ctx->repair->reasm_deque );
1058 0 : fd_repair_blk_t const * blk = fd_repair_blk_pool_ele_const( ctx->repair->blk_pool, out.blk_idx );
1059 0 : fd_repair_fec_t const * fec = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, blk->fecs[ out.fec_idx ] );
1060 0 : fd_repair_out_fec_t * out_fec = fd_chunk_to_laddr( ctx->replay_out_mem, ctx->replay_out_chunk );
1061 0 : memset( out_fec, 0, sizeof(fd_repair_out_fec_t) );
1062 0 : out_fec->slot = blk->slot;
1063 0 : out_fec->fec_set_idx = out.fec_idx*FD_FEC_SHRED_CNT;
1064 0 : out_fec->mr = FD_LOAD( fd_hash_t, fec->mr.mr32.uc );
1065 0 : out_fec->parent_slot = blk->parent_slot;
1066 0 : out_fec->parent_block_id = FD_LOAD( fd_hash_t, blk->parent_dmr.uc );
1067 0 : out_fec->update_parent_fec_idx = blk->init_fec_cnt*FD_FEC_SHRED_CNT;
1068 0 : out_fec->slot_complete = out.fec_idx+1U==blk->cmpl_fec_cnt;
1069 0 : out_fec->data_complete = fec->data_complete;
1070 0 : out_fec->is_leader = fec->is_leader;
1071 0 : out_fec->is_known = fd_repair_slot_meta( ctx->repair, blk->slot )->eager_blk_idx!=out.blk_idx;
1072 0 : out_fec->block_id = FD_LOAD( fd_hash_t, blk->dmr.uc );
1073 0 : out_fec->is_finalized = out_fec->slot_complete && fd_repair_slot_meta( ctx->repair, blk->slot )->final_blk_idx==out.blk_idx;
1074 0 : out_fec->highest_fec_complete_slot = fd_ulong_max( ctx->highest_fec_complete_slot, ctx->catchup_slot );
1075 0 : out_fec->fec_completed_ts_nanos = fec->cmpl_ts; /* shred tile rx of the shred that completed it, 0 for our own */
1076 0 : out_fec->votor_repaired = (uchar)out_fec->is_known;
1077 0 : fd_stem_publish( stem, ctx->replay_out_idx, REPAIR_SIG_FEC, ctx->replay_out_chunk, sizeof(fd_repair_out_fec_t), 0UL, 0UL, fd_frag_meta_ts_comp( fd_tickcount() ) );
1078 0 : ctx->replay_out_chunk = fd_dcache_compact_next( ctx->replay_out_chunk, sizeof(fd_repair_out_fec_t), ctx->replay_out_chunk0, ctx->replay_out_wmark );
1079 0 : ctx->metrics->fec_delivered++;
1080 0 : ctx->metrics->slot_highest_repaired = fd_ulong_if( out_fec->slot_complete, fd_ulong_max( ctx->metrics->slot_highest_repaired, blk->slot ), ctx->metrics->slot_highest_repaired );
1081 :
1082 0 : if( FD_UNLIKELY( out_fec->slot_complete && ctx->rserve_out_idx!=ULONG_MAX && blk->cmpl_fec_cnt<=FD_FEC_BLK_MAX ) ) { /* rserve serves protocol sized blocks only */
1083 0 : fd_repair_out_blk_t * out_blk = fd_chunk_to_laddr( ctx->rserve_out_mem, ctx->rserve_out_chunk );
1084 0 : out_blk->slot = blk->slot;
1085 0 : out_blk->block_id = FD_LOAD( fd_hash_t, blk->dmr.uc );
1086 0 : out_blk->parent_slot = blk->parent_slot;
1087 0 : out_blk->parent_block_id = FD_LOAD( fd_hash_t, blk->parent_dmr.uc );
1088 0 : out_blk->fec_set_cnt = blk->cmpl_fec_cnt;
1089 0 : for( uint k=0U; k<blk->cmpl_fec_cnt; k++ ) memcpy( out_blk->merkle_roots[ k ], fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, blk->fecs[ k ] )->mr.mr20.uc, FD_SHRED_MERKLE_NODE_SZ );
1090 0 : ulong sz = offsetof(fd_repair_out_blk_t, merkle_roots) + (ulong)blk->cmpl_fec_cnt*FD_SHRED_MERKLE_NODE_SZ;
1091 0 : fd_stem_publish( stem, ctx->rserve_out_idx, REPAIR_SIG_BLK, ctx->rserve_out_chunk, sz, 0UL, 0UL, fd_frag_meta_ts_comp( fd_tickcount() ) );
1092 0 : ctx->rserve_out_chunk = fd_dcache_compact_next( ctx->rserve_out_chunk, sz, ctx->rserve_out_chunk0, ctx->rserve_out_wmark );
1093 0 : }
1094 0 : *charge_busy = 1;
1095 0 : }
1096 :
1097 : /* Then expire unanswered requests, discover as many blks off the
1098 : blk_treaps as there are free sign slots (a blk yields at least one
1099 : req, a req at least one sign request), and send pings and due
1100 : reqs while sign slots and pending_pool entries last. A req is one
1101 : peer pick. */
1102 :
1103 0 : long now = fd_clock_tile_now( ctx->clock );
1104 :
1105 0 : while( !pending_dlist_is_empty( ctx->pending_dlist, ctx->pending_pool ) ) {
1106 0 : pending_t * pending = pending_dlist_ele_peek_head( ctx->pending_dlist, ctx->pending_pool );
1107 0 : if( FD_LIKELY( now-pending->ts<PENDING_TTL ) ) break;
1108 :
1109 : /* No rtt sample: silence also means a deduped reply, or a highest
1110 : or orphan request no peer can answer yet. Only a run of shred
1111 : and meta requests left unanswered bans the peer. A request
1112 : answered in part settled its pick on the first shred. */
1113 :
1114 0 : int answerable = pending->slot>ctx->repair->root && pending->tag!=FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX && pending->tag!=FD_REPAIR_SERDE_TAG_ORPHAN;
1115 0 : int byid = pending->tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID;
1116 0 : ctx->metrics->byid_expired += (ulong)( answerable && byid && !pending->answered );
1117 0 : int ban = 0;
1118 0 : if( FD_UNLIKELY( pending->answered ) ) {}
1119 0 : else if( FD_LIKELY ( answerable ) ) ban = fd_repair_strat_request_expired( ctx->strat, &pending->peer, now+BAN_TTL, byid );
1120 0 : else fd_repair_strat_request_failed ( ctx->strat, &pending->peer, 0 );
1121 0 : ctx->metrics->peer_byid_banned += (ulong)( ban==1 );
1122 0 : fd_repair_blk_t const * blk = answerable || !ctx->report_repair_requests || !fd_event_tl ? NULL : fd_repair_blk_map_ele_query_const( ctx->repair->blk_map, &pending->slot, NULL, ctx->repair->blk_pool );
1123 0 : while( blk && fd_repair_blk_pool_idx( ctx->repair->blk_pool, blk )!=pending->blk ) blk = fd_repair_blk_map_ele_next_const( blk, NULL, ctx->repair->blk_pool );
1124 0 : fd_repair_blk_t const * parent = blk ? fd_repair_blk_pool_ele_const( ctx->repair->blk_pool, blk->parent ) : NULL;
1125 0 : int superseded = pending->slot>ctx->repair->root && blk && ( ( pending->tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX && blk->cmpl_fec_cnt ) || ( pending->tag==FD_REPAIR_SERDE_TAG_ORPHAN && parent && parent->parent_slot!=ULONG_MAX ) );
1126 0 : int outcome = answerable ? FD_EVENT_REPAIR_REQUEST_OUTCOME_EXPIRED : superseded ? FD_EVENT_REPAIR_REQUEST_OUTCOME_SUPERSEDED : FD_EVENT_REPAIR_REQUEST_OUTCOME_UNANSWERABLE;
1127 0 : int ban_kind = fd_int_if( ban==1, FD_EVENT_REPAIR_REQUEST_BAN_BY_ID, fd_int_if( ban==-1, FD_EVENT_REPAIR_REQUEST_BAN_STREAK, FD_EVENT_REPAIR_REQUEST_BAN_NONE ) );
1128 0 : for( uint m=pending->mask; m; m=fd_uint_pop_lsb( m ) ) report_repair_request( ctx, pending, (uint)fd_uint_find_lsb( m ), 0L, outcome, ban_kind );
1129 0 : ctx->metrics->req_expired+= (ulong)!pending->answered;
1130 0 : pending_dlist_ele_pop_head ( ctx->pending_dlist, ctx->pending_pool );
1131 0 : pending_map_ele_remove_fast( ctx->pending_map, pending, ctx->pending_pool );
1132 0 : pending_pool_ele_release ( ctx->pending_pool, pending );
1133 0 : }
1134 :
1135 0 : for( ulong d=sign_pool_free( ctx->sign_pool ); d; d-- ) {
1136 0 : fd_repair_t * repair = ctx->repair;
1137 0 : fd_repair_treap_t * treap = fd_repair_treap_ele_cnt( repair->final_treap ) ? repair->final_treap :
1138 0 : fd_repair_treap_ele_cnt( repair->notar_treap ) ? repair->notar_treap :
1139 0 : fd_repair_treap_ele_cnt( repair->eager_treap ) ? repair->eager_treap :
1140 0 : NULL;
1141 0 : if( FD_UNLIKELY( !treap ) ) break;
1142 0 : fd_repair_blk_t * blk = fd_repair_treap_fwd_iter_ele( fd_repair_treap_fwd_iter_init( treap, repair->blk_pool ), repair->blk_pool );
1143 0 : fd_repair_treap_ele_remove( treap, blk, repair->blk_pool );
1144 0 : discover( ctx, blk, now );
1145 0 : }
1146 0 : if( FD_UNLIKELY( ctx->halt_signing ) ) return;
1147 :
1148 : /* A ping is any signed request: the peer pings before it answers. No
1149 : one has slot 0 under the null block id, so no one answers it. */
1150 :
1151 0 : ulong sent = 0UL;
1152 0 : ulong ts = (ulong)now/1000000UL;
1153 0 : while( sent<PING_MAX
1154 0 : && !ping_deque_empty( ctx->ping_deque )
1155 0 : && sign_pool_free( ctx->sign_pool )>SIGN_RESERVE ) {
1156 0 : fd_pubkey_t id_key = ping_deque_pop_head( ctx->ping_deque );
1157 0 : fd_repair_strat_peer_t * peer = fd_repair_strat_query( ctx->strat, &id_key );
1158 0 : sent++;
1159 0 : if( FD_UNLIKELY( !peer || !peer->ip4 || peer->ping_ts ) ) continue; /* gone, or sent to since */
1160 0 : peer->ping_ts = now;
1161 :
1162 0 : sign_t * pending = sign_pool_ele_acquire( ctx->sign_pool );
1163 0 : *pending = (sign_t){ .sig_off = sizeof(uint), .daddr = peer->ip4, .dport = fd_ushort_bswap( peer->port ) };
1164 0 : uint nonce = (uint)fd_chacha_rng_ulong( ctx->nonce_rng );
1165 0 : pending->sz = fd_repair_req_parent_and_fec_set_count_ser( &(fd_repair_parent_fec_set_count_t){ .slot = 0UL }, sig_null, &ctx->identity_key, &id_key, ts, nonce, pending->buf );
1166 :
1167 0 : ulong s = 0UL;
1168 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
1169 0 : ulong sz = fd_repair_req_sig_ser( pending->buf, pending->sz, fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk ) );
1170 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, pending )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_ED25519, ctx->sign[ s ].chunk, sz, 0UL, 0UL, 0UL );
1171 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, sz, ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
1172 0 : ctx->sign[ s ].free--;
1173 0 : ctx->metrics->ping_tx++;
1174 0 : *charge_busy = 1;
1175 0 : }
1176 :
1177 0 : while( sign_pool_free( ctx->sign_pool )>=SIGN_RESERVE+FD_FEC_SHRED_CNT ) {
1178 0 : timeout_t req;
1179 0 : fd_repair_blk_t * blk = poll_timeout( ctx, now, &req );
1180 0 : if( FD_UNLIKELY( !blk ) ) break;
1181 :
1182 0 : fd_repair_fec_t const * fec = fd_repair_fec_pool_ele_const( ctx->repair->fec_pool, fd_ulong_if( req.tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX, blk->fecs[ req.fec_idx ], fd_repair_fec_pool_idx_null( ctx->repair->fec_pool ) ) );
1183 0 : int eager = blk->eager;
1184 0 : uint tag = req.tag!=FD_REPAIR_SERDE_TAG_WINDOW_INDEX ? req.tag : /* a FEC set req becomes what its blk still lacks */
1185 0 : eager ? FD_REPAIR_SERDE_TAG_WINDOW_INDEX :
1186 0 : !fec ? FD_REPAIR_SERDE_TAG_FEC_SET_ROOT :
1187 0 : FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID;
1188 0 : int meta = tag==FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT || tag==FD_REPAIR_SERDE_TAG_FEC_SET_ROOT;
1189 0 : uint mask = fd_uint_if( tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX || tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID, fec ? fd_uint_if( fec->fec_complete, 0U, ~fec->data_rcvd ) : UINT_MAX, 1U ); /* bit i is a message for shred i of the FEC set */
1190 0 : if( FD_UNLIKELY( !mask ) ) { /* every shred arrived, the FEC set completes without repair */
1191 0 : ctx->metrics->timeout_deferred[ FD_METRICS_ENUM_TIMEOUT_DEFER_REASON_V_ALL_RECEIVED_IDX ]++;
1192 0 : req.timeout = now+fd_repair_strat_eager_ns( ctx->strat, fd_multi_epoch_leaders_get_leader_for_slot( ctx->mleaders, blk->slot ) );
1193 0 : push_timeout( ctx, blk, &req );
1194 0 : continue;
1195 0 : }
1196 :
1197 : /* A retry asks someone else. After STUCK_MAX attempts that brought
1198 : no new shred, the fastest peers are not answering: pick a staked
1199 : peer at random instead. */
1200 0 : uint rcvd = fec ? fd_uint_if( fec->fec_complete, UINT_MAX, fec->data_rcvd ) : 0U;
1201 0 : uint stuck = fd_uint_if( req.attempt && !( rcvd & ~req.rcvd ), req.stuck+1U, 0U ); /* only new shreds are progress, not an eviction; kept only once sent */
1202 0 : int escape = stuck>=STUCK_MAX && !blk->eager;
1203 0 : fd_repair_strat_peer_t * peer = escape ? fd_repair_strat_pick_staked( ctx->strat, now, (ulong)req.last_peer-1UL, fd_rng_ulong( ctx->rng ) ) : NULL;
1204 0 : int escaped = !!peer;
1205 0 : ctx->metrics->req_stuck_escaped += (ulong)!!peer;
1206 0 : if( FD_UNLIKELY( peer ) ) fd_histf_sample( ctx->metrics->escape_hedge, (ulong)fd_long_if( !!peer->srtt_ns, peer->srtt_ns+4L*peer->rttvar_ns, FD_REPAIR_STRAT_HEDGE_NS ) ); /* uncapped */
1207 0 : if( FD_LIKELY( !peer ) ) peer = fd_repair_strat_pick( ctx->strat, now, !blk->eager && req.attempt, (ulong)req.last_peer-1UL );
1208 0 : if( FD_UNLIKELY( !peer ) ) {
1209 0 : ctx->metrics->req_no_peer++;
1210 0 : ctx->metrics->req_no_staked_peer += (ulong)( !blk->eager && req.attempt );
1211 0 : req.timeout = now+NO_PEER;
1212 0 : push_timeout( ctx, blk, &req );
1213 0 : break;
1214 0 : }
1215 0 : peer->ping_ts = now;
1216 0 : req.stuck = stuck;
1217 0 : req.rcvd = rcvd;
1218 :
1219 0 : uint nonce = (uint)fd_chacha_rng_ulong( ctx->nonce_rng );
1220 0 : while( meta && pending_map_ele_query_const( ctx->pending_map, &(ulong){ nonce }, NULL, ctx->pending_pool ) ) {
1221 0 : nonce = (uint)fd_chacha_rng_ulong( ctx->nonce_rng );
1222 0 : }
1223 :
1224 0 : if( FD_UNLIKELY( !pending_pool_free( ctx->pending_pool ) ) ) { /* evict the oldest, a reply to it no longer matches */
1225 0 : pending_t * oldest = pending_dlist_ele_pop_head( ctx->pending_dlist, ctx->pending_pool );
1226 0 : if( FD_LIKELY( !oldest->answered ) ) fd_repair_strat_request_failed( ctx->strat, &oldest->peer, 0 );
1227 0 : for( uint m=oldest->mask; m; m=fd_uint_pop_lsb( m ) ) report_repair_request( ctx, oldest, (uint)fd_uint_find_lsb( m ), 0L, FD_EVENT_REPAIR_REQUEST_OUTCOME_EVICTED, FD_EVENT_REPAIR_REQUEST_BAN_NONE );
1228 0 : pending_map_ele_remove_fast( ctx->pending_map, oldest, ctx->pending_pool );
1229 0 : pending_pool_ele_release ( ctx->pending_pool, oldest );
1230 0 : }
1231 0 : fd_repair_slot_meta_t const * slot_meta = fd_repair_slot_meta( ctx->repair, blk->slot );
1232 0 : int role = slot_meta->final_blk_idx==req.blk ? FD_EVENT_REPAIR_REQUEST_ROLE_FINAL : slot_meta->notar_blk_idx==req.blk ? FD_EVENT_REPAIR_REQUEST_ROLE_NOTAR : FD_EVENT_REPAIR_REQUEST_ROLE_EAGER;
1233 0 : int pick = escaped ? FD_EVENT_REPAIR_REQUEST_PICK_ESCAPE : !req.attempt ? FD_EVENT_REPAIR_REQUEST_PICK_FIRST : !blk->eager ? FD_EVENT_REPAIR_REQUEST_PICK_STAKED_HEDGE : FD_EVENT_REPAIR_REQUEST_PICK_HEDGE;
1234 0 : pending_t * pending = pending_pool_ele_acquire( ctx->pending_pool );
1235 0 : pending->key = fd_ulong_if( meta, (ulong)nonce, (ulong)(uint)blk->slot<<32 | fd_uint_if( req.tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX || req.tag==FD_REPAIR_SERDE_TAG_ORPHAN, (uint)ctx->repair->fec_blk_max, req.fec_idx ) ); /* an ORPHAN's first reply is of the slot itself, matched like a HIGHEST */
1236 0 : pending->ts = now;
1237 0 : pending->slot = blk->slot;
1238 0 : pending->blk = req.blk;
1239 0 : pending->dmr = FD_LOAD( fd_hash_t, blk->dmr.uc );
1240 0 : pending->peer = peer->id_key;
1241 0 : pending->fec_idx = fd_uint_if( tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX, blk->rcvd_fec_cnt, req.fec_idx );
1242 0 : pending->fec_set_cnt = blk->cmpl_fec_cnt;
1243 0 : pending->tag = tag;
1244 0 : pending->mask = mask;
1245 0 : pending->answered = 0U;
1246 0 : pending->ip4 = peer->ip4;
1247 0 : pending->port = peer->port;
1248 0 : pending->attempt = req.attempt;
1249 0 : pending->pick = (uchar)pick;
1250 0 : pending->role = (uchar)role;
1251 0 : pending_map_ele_insert ( ctx->pending_map, pending, ctx->pending_pool );
1252 0 : pending_dlist_ele_push_tail( ctx->pending_dlist, pending, ctx->pending_pool );
1253 :
1254 0 : for( uint m=mask; m; m=fd_uint_pop_lsb( m ) ) {
1255 0 : uint idx = req.fec_idx*FD_FEC_SHRED_CNT+(uint)fd_uint_find_lsb( m );
1256 0 : uint rnonce = fd_uint_if( meta, nonce, fd_rnonce_ss_compute( ctx->rnonce_ss, tag!=FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX && tag!=FD_REPAIR_SERDE_TAG_ORPHAN, blk->slot, idx, now ) );
1257 0 : sign_t * sign = sign_pool_ele_acquire( ctx->sign_pool );
1258 0 : *sign = (sign_t){ .sig_off = sizeof(uint), .daddr = peer->ip4, .dport = fd_ushort_bswap( peer->port ) };
1259 0 : switch( tag ) {
1260 0 : case FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: sign->sz = fd_repair_req_parent_and_fec_set_count_ser ( &(fd_repair_parent_fec_set_count_t){ .slot = blk->slot, .block_id = FD_LOAD( fd_hash_t, blk->dmr.uc ) }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1261 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT: sign->sz = fd_repair_req_fec_set_root_ser ( &(fd_repair_fec_set_root_t){ .slot = blk->slot, .block_id = FD_LOAD( fd_hash_t, blk->dmr.uc ), .fec_set_idx = idx }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1262 0 : case FD_REPAIR_SERDE_TAG_ORPHAN: sign->sz = fd_repair_req_orphan_ser ( &(fd_repair_orphan_t){ .slot = blk->slot }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1263 0 : case FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX: sign->sz = fd_repair_req_highest_window_index_ser ( &(fd_repair_highest_shred_t){ .slot = blk->slot, .shred_idx = blk->rcvd_fec_cnt*FD_FEC_SHRED_CNT }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1264 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX: sign->sz = fd_repair_req_window_index_ser ( &(fd_repair_shred_t){ .slot = blk->slot, .shred_idx = idx }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1265 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: sign->sz = fd_repair_req_window_index_for_block_id_ser( &(fd_repair_shred_for_block_id_t){ .slot = blk->slot, .shred_idx = idx, .block_id = FD_LOAD( fd_hash_t, blk->dmr.uc ) }, sig_null, &ctx->identity_key, &peer->id_key, ts, rnonce, sign->buf ); break;
1266 0 : }
1267 :
1268 0 : ulong s = 0UL;
1269 0 : for( ulong i=1UL; i<ctx->sign_cnt; i++ ) s = fd_ulong_if( ctx->sign[ i ].free>ctx->sign[ s ].free, i, s );
1270 0 : ulong sz = fd_repair_req_sig_ser( sign->buf, sign->sz, fd_chunk_to_laddr( ctx->sign[ s ].mem, ctx->sign[ s ].chunk ) );
1271 0 : fd_stem_publish( stem, ctx->sign[ s ].idx, (sign_pool_idx( ctx->sign_pool, sign )<<32)|(ulong)FD_KEYGUARD_SIGN_TYPE_ED25519, ctx->sign[ s ].chunk, sz, 0UL, 0UL, 0UL );
1272 0 : ctx->sign[ s ].chunk = fd_dcache_compact_next( ctx->sign[ s ].chunk, sz, ctx->sign[ s ].chunk0, ctx->sign[ s ].wmark );
1273 0 : ctx->sign[ s ].free--;
1274 0 : }
1275 0 : ctx->metrics->request_tx[ tag ] += (ulong)fd_uint_popcnt( mask );
1276 0 : ctx->metrics->req_hedged += (ulong)!!req.attempt;
1277 0 : ctx->metrics->req_late_attempt += (ulong)( req.attempt>=3U );
1278 :
1279 0 : blk->telemetry.first_req_ts = fd_long_if( !blk->telemetry.first_req_ts, now, blk->telemetry.first_req_ts );
1280 0 : uint * req_cnt = NULL;
1281 0 : switch( tag ) {
1282 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX: req_cnt = &blk->telemetry.req_window_cnt; break;
1283 0 : case FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX: req_cnt = &blk->telemetry.req_highest_cnt; break;
1284 0 : case FD_REPAIR_SERDE_TAG_ORPHAN: req_cnt = &blk->telemetry.req_orphan_cnt; break;
1285 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: req_cnt = &blk->telemetry.req_shred_bid_cnt; break;
1286 0 : case FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: req_cnt = &blk->telemetry.req_parent_cnt; break;
1287 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT: req_cnt = &blk->telemetry.req_fec_root_cnt; break;
1288 0 : }
1289 0 : if( FD_LIKELY( req_cnt ) ) *req_cnt += (uint)fd_uint_popcnt( mask );
1290 :
1291 0 : long hedge = fd_long_if( escape, 1L<<24, fd_long_min( fd_long_max( fd_repair_strat_hedge_ns( peer ), 1L<<24 ), FD_REPAIR_STRAT_HEDGE_NS ) );
1292 0 : req.timeout = now+hedge; /* no backoff: a re-pick gets a new shred nonce, and a random escape pick does not slow the next one */
1293 0 : req.attempt = (uchar)fd_uint_min( req.attempt+1U, UCHAR_MAX );
1294 0 : req.last_peer = (ushort)( 1UL+(ulong)( peer-ctx->strat->cur.peers ) );
1295 0 : fd_histf_sample( ctx->metrics->retry_delay, (ulong)( req.timeout-now ) );
1296 0 : push_timeout( ctx, blk, &req );
1297 0 : *charge_busy = 1;
1298 0 : }
1299 0 : }
1300 :
1301 : static int
1302 : before_frag( fd_repair_tile_t * ctx,
1303 : ulong in_idx,
1304 : ulong seq FD_PARAM_UNUSED,
1305 0 : ulong sig ) {
1306 0 : switch( ctx->in_kind[ in_idx ] ) {
1307 0 : case IN_KIND_GENESIS:
1308 0 : return 0;
1309 0 : case IN_KIND_SHRED: {
1310 0 : if( FD_UNLIKELY( ctx->repair->root==ULONG_MAX ) ) return -1; /* hold until the snapshot or genesis gives the root */
1311 0 : if( FD_LIKELY( fd_shred_sig_src( sig )!=SHRED_SIG_FUTURE ) ) return 0;
1312 0 : ulong slot = fd_shred_sig_future_slot( sig );
1313 0 : if( FD_LIKELY( ctx->staked || slot<ctx->repair->root+ctx->repair->slot_max || slot<=ctx->catchup_slot ) ) return 1;
1314 0 : ctx->catchup_slot = slot;
1315 0 : fd_repair_slot_catchup( ctx->repair, ctx->repair->root+ctx->repair->slot_max );
1316 0 : return 1;
1317 0 : }
1318 0 : case IN_KIND_SNAP: {
1319 0 : ulong msg = fd_ssmsg_sig_message( sig );
1320 0 : return msg!=FD_SSMSG_MANIFEST_FULL && msg!=FD_SSMSG_MANIFEST_INCREMENTAL && msg!=FD_SSMSG_DONE;
1321 0 : }
1322 0 : case IN_KIND_VOTOR:
1323 0 : if( FD_UNLIKELY( ctx->repair->root==ULONG_MAX ) ) return -1; /* hold until the snapshot or genesis gives the root */
1324 0 : return sig!=FD_VOTOR_SIG_CERTED && sig!=FD_VOTOR_SIG_REPAIR && sig!=FD_VOTOR_SIG_QUORUM;
1325 0 : case IN_KIND_REPLAY:
1326 0 : return sig!=REPLAY_SIG_ROOT_ADVANCED && sig!=REPLAY_SIG_SLOT_DEAD && sig!=REPLAY_SIG_MISSING_FEC;
1327 0 : case IN_KIND_EPOCH:
1328 0 : return 0;
1329 0 : case IN_KIND_GOSSIP:
1330 0 : return sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO && sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
1331 0 : case IN_KIND_NET:
1332 0 : return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
1333 0 : case IN_KIND_SIGN:
1334 0 : return 0;
1335 0 : default:
1336 0 : FD_LOG_ERR(( "unexpected in_kind %d", ctx->in_kind[ in_idx ] ));
1337 0 : }
1338 0 : }
1339 :
1340 : static void
1341 : during_frag( fd_repair_tile_t * ctx,
1342 : ulong in_idx,
1343 : ulong seq FD_PARAM_UNUSED,
1344 : ulong sig FD_PARAM_UNUSED,
1345 : ulong chunk,
1346 : ulong sz,
1347 0 : ulong ctl FD_PARAM_UNUSED ) {
1348 0 : switch( ctx->in_kind[ in_idx ] ) {
1349 0 : case IN_KIND_NET:
1350 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) {
1351 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1352 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1353 0 : }
1354 0 : memcpy( ctx->net_buf, fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk ), sz );
1355 0 : break;
1356 0 : case IN_KIND_SIGN:
1357 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) {
1358 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1359 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1360 0 : }
1361 0 : memcpy( ctx->sign_buf, fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk ), FD_ED25519_SIG_SZ );
1362 0 : break;
1363 0 : case IN_KIND_REPLAY:
1364 0 : if( FD_UNLIKELY( sz!=0UL && ( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>sizeof(ctx->replay_buf) ) ) ) { /* a missing FEC has no payload */
1365 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1366 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1367 0 : }
1368 0 : memcpy( &ctx->replay_buf, fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk ), sz );
1369 0 : break;
1370 0 : default: /* reliable, read in place by after_frag */
1371 0 : if( FD_UNLIKELY( sz!=0UL && ( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) ) { /* a snapshot DONE has no payload */
1372 0 : FD_LOG_ERR(( "chunk %lu sz %lu from in_kind %d out of bounds, chunk0 %lu wmark %lu",
1373 0 : chunk, sz, ctx->in_kind[ in_idx ], ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
1374 0 : }
1375 0 : ctx->chunk = chunk;
1376 0 : break;
1377 0 : }
1378 0 : }
1379 :
1380 : static void
1381 : after_frag( fd_repair_tile_t * ctx,
1382 : ulong in_idx,
1383 : ulong seq FD_PARAM_UNUSED,
1384 : ulong sig,
1385 : ulong sz,
1386 : ulong tsorig,
1387 : ulong tspub,
1388 0 : fd_stem_context_t * stem ) {
1389 0 : uchar const * src = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, ctx->chunk );
1390 :
1391 0 : switch( ctx->in_kind[ in_idx ] ) {
1392 0 : case IN_KIND_NET: {
1393 0 : long now_tick = fd_tickcount();
1394 0 : long rx_tick = fd_frag_meta_ts_decomp( tspub, now_tick );
1395 0 : if( FD_UNLIKELY( rx_tick>now_tick ) ) rx_tick -= 1L<<32;
1396 0 : handle_net( ctx, stem, sz, fd_clock_tile_tickcount_to_wallclock( ctx->clock, rx_tick ) );
1397 0 : break;
1398 0 : }
1399 0 : case IN_KIND_SIGN:
1400 0 : handle_sign( ctx, stem, in_idx, sig );
1401 0 : break;
1402 0 : case IN_KIND_GENESIS: {
1403 0 : fd_genesis_meta_t const * meta = (fd_genesis_meta_t const *)fd_type_pun_const( src );
1404 0 : if( FD_UNLIKELY( !meta->bootstrap ) ) break; /* booting from a snapshot */
1405 0 : fd_repair_init( ctx->repair, 0UL, fd_type_pun_const( &hash_null ), report_block_received, ctx );
1406 0 : break;
1407 0 : }
1408 0 : case IN_KIND_SHRED: { /* tsorig is the shred tile's network rx of the shred, the completing one for a FEC set */
1409 0 : long now_tick = fd_tickcount();
1410 0 : long rx_tick = fd_frag_meta_ts_decomp( tsorig, now_tick );
1411 0 : if( FD_UNLIKELY( rx_tick>now_tick ) ) rx_tick -= 1L<<32;
1412 0 : handle_shred( ctx, sig, src, fd_clock_tile_tickcount_to_wallclock( ctx->clock, rx_tick ) );
1413 0 : break;
1414 0 : }
1415 0 : case IN_KIND_SNAP:
1416 0 : handle_snapshot( ctx, sig, src );
1417 0 : break;
1418 0 : case IN_KIND_VOTOR:
1419 0 : handle_votor( ctx, sig, src );
1420 0 : break;
1421 0 : case IN_KIND_REPLAY:
1422 0 : handle_replay( ctx, sig, (uchar const *)&ctx->replay_buf );
1423 0 : break;
1424 0 : case IN_KIND_EPOCH: {
1425 0 : fd_epoch_info_msg_t const * msg = (fd_epoch_info_msg_t const *)fd_type_pun_const( src );
1426 0 : fd_repair_strat_epoch_advanced( ctx->strat, fd_epoch_info_msg_id_weights( msg ), msg->staked_id_cnt );
1427 0 : timeout_t * prqs[3] = { ctx->final_prq, ctx->notar_prq, ctx->eager_prq };
1428 0 : for( ulong c=0UL; c<3UL; c++ ) { /* the re-sort moved peers, keep retries avoiding the same one */
1429 0 : for( ulong i=0UL; i<timeout_prq_cnt( prqs[ c ] ); i++ ) {
1430 0 : timeout_t * req = &prqs[ c ][ i ];
1431 0 : req->last_peer = (ushort)( 1UL+fd_repair_strat_slot_remap( ctx->strat, (ulong)req->last_peer-1UL ) );
1432 0 : }
1433 0 : }
1434 0 : fd_repair_strat_peer_t const * self = fd_repair_strat_query( ctx->strat, &ctx->identity_key );
1435 0 : ctx->staked = self && (ulong)( self-ctx->strat->cur.peers )<ctx->strat->staked_cnt;
1436 0 : fd_multi_epoch_leaders_epoch_msg_init( ctx->mleaders, msg );
1437 0 : fd_multi_epoch_leaders_epoch_msg_fini( ctx->mleaders );
1438 0 : break;
1439 0 : }
1440 0 : case IN_KIND_GOSSIP:
1441 0 : handle_gossip( ctx, sig, (fd_gossip_update_message_t const *)fd_type_pun_const( src ) );
1442 0 : break;
1443 0 : default:
1444 0 : FD_LOG_ERR(( "unexpected in_kind %d", ctx->in_kind[ in_idx ] ));
1445 0 : }
1446 0 : }
1447 :
1448 : /* next_deadline is when the tile next has work without a frag: the
1449 : earliest timeout due, or the oldest pending expiring. */
1450 :
1451 : static long
1452 0 : next_deadline( fd_repair_tile_t * ctx ) {
1453 0 : long due = LONG_MAX;
1454 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->eager_prq ), fd_long_min( due, ctx->eager_prq[ 0 ].timeout ), due );
1455 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->notar_prq ), fd_long_min( due, ctx->notar_prq[ 0 ].timeout ), due );
1456 0 : due = fd_long_if( !!timeout_prq_cnt( ctx->final_prq ), fd_long_min( due, ctx->final_prq[ 0 ].timeout ), due );
1457 0 : if( FD_LIKELY( !pending_dlist_is_empty( ctx->pending_dlist, ctx->pending_pool ) ) ) due = fd_long_min( due, pending_dlist_ele_peek_head( ctx->pending_dlist, ctx->pending_pool )->ts+PENDING_TTL );
1458 0 : if( FD_UNLIKELY( due==LONG_MAX ) ) return LONG_MAX;
1459 0 : return fd_clock_tile_wallclock_to_tickcount( ctx->clock, due );
1460 0 : }
1461 :
1462 : static inline void
1463 0 : during_housekeeping( fd_repair_tile_t * ctx ) {
1464 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) fd_clock_tile_recal( ctx->clock );
1465 :
1466 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_UNHALT_PENDING ) ) {
1467 0 : FD_CHECK_CRIT( ctx->halt_signing, "state machine corruption" );
1468 0 : ctx->halt_signing = 0;
1469 :
1470 : /* Peers key their ping-pong cache by our pubkey, so every peer is
1471 : cold for the new identity. Re-warm them all. */
1472 0 : ping_deque_remove_all( ctx->ping_deque );
1473 0 : for( ulong i=0UL; i<FD_REPAIR_STRAT_PEER_MAX && !ping_deque_full( ctx->ping_deque ); i++ ) {
1474 0 : fd_repair_strat_peer_t * peer = &ctx->strat->cur.peers[ i ];
1475 0 : if( FD_LIKELY( !peer->ip4 ) ) continue; /* free slot, or no address */
1476 0 : peer->ping_ts = 0L;
1477 0 : ping_deque_push_tail( ctx->ping_deque, peer->id_key );
1478 0 : }
1479 :
1480 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
1481 0 : }
1482 :
1483 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
1484 0 : ctx->metrics->sign_halted += (ulong)!ctx->halt_signing;
1485 0 : ctx->halt_signing = 1;
1486 0 : memcpy( ctx->identity_key.uc, ctx->keyswitch->bytes, 32UL );
1487 0 : fd_repair_strat_peer_t const * self = fd_repair_strat_query( ctx->strat, &ctx->identity_key );
1488 0 : ctx->staked = self && (ulong)( self-ctx->strat->cur.peers )<ctx->strat->staked_cnt;
1489 0 : if( FD_LIKELY( sign_pool_free( ctx->sign_pool )==sign_pool_max( ctx->sign_pool ) ) ) fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED ); /* every sign request returned */
1490 0 : }
1491 0 : }
1492 :
1493 : static void
1494 : privileged_init( fd_topo_t const * topo,
1495 0 : fd_topo_tile_t const * tile ) {
1496 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1497 :
1498 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1499 0 : fd_repair_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_tile_t), sizeof(fd_repair_tile_t) );
1500 0 : fd_memset( ctx, 0, sizeof(fd_repair_tile_t) );
1501 :
1502 0 : FD_TEST( fd_rng_secure( &ctx->seed, sizeof(ulong) ) );
1503 0 : uchar nonce_seed[ 32 ];
1504 0 : FD_TEST( fd_rng_secure( nonce_seed, sizeof(nonce_seed) ) );
1505 0 : FD_TEST( fd_chacha_rng_join( fd_chacha_rng_new( ctx->nonce_rng, FD_CHACHA_RNG_MODE_SHIFT ) ) );
1506 0 : fd_chacha_rng_init( ctx->nonce_rng, nonce_seed, FD_CHACHA_RNG_ALGO_CHACHA8 );
1507 0 : fd_memzero_explicit( nonce_seed, sizeof(nonce_seed) );
1508 :
1509 0 : uchar const * identity_key = fd_keyload_load( tile->repair.identity_key_path, /* pubkey only: */ 1 );
1510 0 : fd_memcpy( ctx->identity_key.uc, identity_key, sizeof(fd_pubkey_t) );
1511 :
1512 0 : ulong rnonce_ss_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "rnonce_ss" );
1513 0 : FD_TEST( rnonce_ss_id!=ULONG_MAX );
1514 0 : memcpy( ctx->rnonce_ss, fd_topo_obj_laddr( topo, rnonce_ss_id ), sizeof(fd_rnonce_ss_t) );
1515 0 : }
1516 :
1517 : static void
1518 : unprivileged_init( fd_topo_t const * topo,
1519 0 : fd_topo_tile_t const * tile ) {
1520 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1521 0 : ulong sign_max = tile->repair.repair_sign_depth*tile->repair.repair_sign_cnt;
1522 0 : ulong timeout_max = tile->repair.slot_max*(FEC_SET_P90+1UL);
1523 0 : ulong pending_max = tile->repair.repair_sign_cnt*(ulong)( PENDING_TTL/SIGN_NS );
1524 0 : ulong chain_cnt = pending_map_chain_cnt_est( pending_max );
1525 :
1526 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1527 0 : fd_repair_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_tile_t), sizeof(fd_repair_tile_t) );
1528 0 : ctx->repair = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_align(), fd_repair_footprint( tile->repair.slot_max, tile->repair.fec_max, tile->repair.max_shreds_per_block/FD_FEC_SHRED_CNT ) );
1529 0 : void * strat = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_strat_align(), fd_repair_strat_footprint() );
1530 0 : void * mleaders = FD_SCRATCH_ALLOC_APPEND( l, fd_multi_epoch_leaders_align(), fd_multi_epoch_leaders_footprint() );
1531 0 : void * ping_deque = FD_SCRATCH_ALLOC_APPEND( l, ping_deque_align(), ping_deque_footprint( FD_REPAIR_STRAT_PEER_MAX ) );
1532 0 : void * eager_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1533 0 : void * notar_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1534 0 : void * final_prq = FD_SCRATCH_ALLOC_APPEND( l, timeout_prq_align(), timeout_prq_footprint( timeout_max ) );
1535 0 : void * pending_pool = FD_SCRATCH_ALLOC_APPEND( l, pending_pool_align(), pending_pool_footprint( pending_max ) );
1536 0 : void * pending_map = FD_SCRATCH_ALLOC_APPEND( l, pending_map_align(), pending_map_footprint( chain_cnt ) );
1537 0 : void * pending_dlist = FD_SCRATCH_ALLOC_APPEND( l, pending_dlist_align(), pending_dlist_footprint() );
1538 0 : void * sign_pool = FD_SCRATCH_ALLOC_APPEND( l, sign_pool_align(), sign_pool_footprint( sign_max ) );
1539 0 : ctx->event = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_event_block_received_t), sizeof(fd_event_block_received_t) );
1540 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
1541 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
1542 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
1543 :
1544 0 : ctx->repair = fd_repair_join( fd_repair_new( ctx->repair, tile->repair.slot_max, tile->repair.fec_max, tile->repair.max_shreds_per_block/FD_FEC_SHRED_CNT, ctx->seed ) );
1545 0 : FD_TEST( ctx->repair );
1546 0 : ctx->mleaders = fd_multi_epoch_leaders_join( fd_multi_epoch_leaders_new( mleaders ) );
1547 0 : FD_TEST( ctx->mleaders );
1548 :
1549 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
1550 0 : FD_TEST( ctx->keyswitch );
1551 :
1552 0 : ctx->strat = fd_repair_strat_join( fd_repair_strat_new( strat, ctx->seed ) );
1553 0 : ctx->ping_deque = ping_deque_join( ping_deque_new( ping_deque, FD_REPAIR_STRAT_PEER_MAX ) );
1554 0 : ctx->eager_prq = timeout_prq_join( timeout_prq_new( eager_prq, timeout_max ) );
1555 0 : ctx->notar_prq = timeout_prq_join( timeout_prq_new( notar_prq, timeout_max ) );
1556 0 : ctx->final_prq = timeout_prq_join( timeout_prq_new( final_prq, timeout_max ) );
1557 0 : ctx->pending_pool = pending_pool_join( pending_pool_new( pending_pool, pending_max ) );
1558 0 : ctx->pending_map = pending_map_join( pending_map_new( pending_map, chain_cnt, ctx->seed ) );
1559 0 : ctx->pending_dlist = pending_dlist_join( pending_dlist_new( pending_dlist ) );
1560 0 : ctx->sign_pool = sign_pool_join( sign_pool_new( sign_pool, sign_max ) );
1561 0 : FD_TEST( ctx->strat && ctx->ping_deque && ctx->eager_prq && ctx->notar_prq && ctx->final_prq && ctx->pending_pool && ctx->pending_map && ctx->pending_dlist && ctx->sign_pool );
1562 0 : FD_TEST( fd_rng_join( fd_rng_new( ctx->rng, (uint)ctx->seed, ctx->seed>>32 ) ) );
1563 0 : fd_clock_tile_init( ctx->clock );
1564 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->response_latency, FD_MHIST_MIN( REPAIR, RESPONSE_LATENCY_NANOS ), FD_MHIST_MAX( REPAIR, RESPONSE_LATENCY_NANOS ) ) ) );
1565 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->retry_delay, FD_MHIST_MIN( REPAIR, RETRY_DELAY_NANOS ), FD_MHIST_MAX( REPAIR, RETRY_DELAY_NANOS ) ) ) );
1566 0 : FD_TEST( fd_histf_join( fd_histf_new( ctx->metrics->escape_hedge, FD_MHIST_MIN( REPAIR, ESCAPE_PEER_HEDGE_DELAY_NANOS ), FD_MHIST_MAX( REPAIR, ESCAPE_PEER_HEDGE_DELAY_NANOS ) ) ) );
1567 :
1568 0 : FD_TEST( tile->in_cnt<=sizeof(ctx->in_kind)/sizeof(ctx->in_kind[0]) );
1569 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
1570 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
1571 :
1572 0 : if ( FD_LIKELY( !strcmp( link->name, "genesi_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_GENESIS;
1573 0 : else if( FD_LIKELY( !strcmp( link->name, "shred_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_SHRED;
1574 0 : else if( FD_LIKELY( !strcmp( link->name, "snapin_manif" ) ) ) ctx->in_kind[ i ] = IN_KIND_SNAP;
1575 0 : else if( FD_LIKELY( !strcmp( link->name, "votor_out" ) ) ) ctx->in_kind[ i ] = IN_KIND_VOTOR;
1576 0 : else if( FD_LIKELY( !strcmp( link->name, "shred_repair" ) ) ) ctx->in_kind[ i ] = IN_KIND_NET;
1577 0 : else if( FD_LIKELY( !strcmp( link->name, "gossip_ciaddr" ) ) ) ctx->in_kind[ i ] = IN_KIND_GOSSIP;
1578 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_slot" ) ) ) ctx->in_kind[ i ] = IN_KIND_REPLAY;
1579 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_repair" ) ) ) ctx->in_kind[ i ] = IN_KIND_REPLAY;
1580 0 : else if( FD_LIKELY( !strcmp( link->name, "sign_repair" ) ) ) ctx->in_kind[ i ] = IN_KIND_SIGN;
1581 0 : else if( FD_LIKELY( !strcmp( link->name, "replay_epoch" ) ) ) ctx->in_kind[ i ] = IN_KIND_EPOCH;
1582 0 : else FD_LOG_ERR(( "repair tile has unexpected input link %lu %s", i, link->name ));
1583 0 : ctx->in[ i ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1584 0 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].mem, link->dcache );
1585 0 : ctx->in[ i ].wmark = fd_dcache_compact_wmark ( ctx->in[ i ].mem, link->dcache, link->mtu );
1586 0 : ctx->in[ i ].mtu = link->mtu;
1587 0 : }
1588 :
1589 0 : ctx->replay_out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_out", 0UL );
1590 0 : FD_TEST( ctx->replay_out_idx!=ULONG_MAX );
1591 0 : fd_topo_link_t const * replay_out = &topo->links[ tile->out_link_id[ ctx->replay_out_idx ] ];
1592 0 : ctx->replay_out_mem = topo->workspaces[ topo->objs[ replay_out->dcache_obj_id ].wksp_id ].wksp;
1593 0 : ctx->replay_out_chunk0 = fd_dcache_compact_chunk0( ctx->replay_out_mem, replay_out->dcache );
1594 0 : ctx->replay_out_wmark = fd_dcache_compact_wmark ( ctx->replay_out_mem, replay_out->dcache, replay_out->mtu );
1595 0 : ctx->replay_out_chunk = ctx->replay_out_chunk0;
1596 :
1597 : /* sign tile s takes repair_sign s and answers on sign_repair s */
1598 :
1599 0 : ctx->sign_cnt = tile->repair.repair_sign_cnt;
1600 0 : FD_TEST( ctx->sign_cnt<=SIGN_MAX );
1601 0 : for( ulong s=0UL; s<ctx->sign_cnt; s++ ) {
1602 0 : ulong out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_sign", s );
1603 0 : ulong in_idx = fd_topo_find_tile_in_link ( topo, tile, "sign_repair", s );
1604 0 : FD_TEST( out_idx!=ULONG_MAX && in_idx!=ULONG_MAX );
1605 0 : fd_topo_link_t const * sign_out = &topo->links[ tile->out_link_id[ out_idx ] ];
1606 0 : FD_TEST( sign_out->mtu>=FD_REPAIR_SIG_SER_MAX );
1607 0 : ctx->sign[ s ].idx = out_idx;
1608 0 : ctx->sign[ s ].mem = topo->workspaces[ topo->objs[ sign_out->dcache_obj_id ].wksp_id ].wksp;
1609 0 : ctx->sign[ s ].chunk0 = fd_dcache_compact_chunk0( ctx->sign[ s ].mem, sign_out->dcache );
1610 0 : ctx->sign[ s ].wmark = fd_dcache_compact_wmark ( ctx->sign[ s ].mem, sign_out->dcache, sign_out->mtu );
1611 0 : ctx->sign[ s ].chunk = ctx->sign[ s ].chunk0;
1612 0 : ctx->sign[ s ].free = tile->repair.repair_sign_depth;
1613 0 : ctx->in[ in_idx ].sign_idx = s;
1614 0 : }
1615 :
1616 0 : ctx->net_out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_net", 0UL );
1617 0 : FD_TEST( ctx->net_out_idx!=ULONG_MAX );
1618 0 : fd_topo_link_t const * net_out = &topo->links[ tile->out_link_id[ ctx->net_out_idx ] ];
1619 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ net_out->dcache_obj_id ].wksp_id ].wksp;
1620 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, net_out->dcache );
1621 0 : ctx->net_out_wmark = fd_dcache_compact_wmark ( ctx->net_out_mem, net_out->dcache, net_out->mtu );
1622 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
1623 0 : fd_ip4_udp_hdr_init( ctx->net_hdr, 0UL, 0U, tile->repair.repair_client_listen_port );
1624 0 : ctx->allow_private_address = tile->repair.allow_private_address;
1625 0 : ctx->report_repair_requests = tile->repair.report_repair_requests;
1626 :
1627 0 : ctx->rserve_out_idx = fd_topo_find_tile_out_link( topo, tile, "repair_rserve", 0UL );
1628 0 : if( FD_LIKELY( ctx->rserve_out_idx!=ULONG_MAX ) ) {
1629 0 : fd_topo_link_t const * rserve_out = &topo->links[ tile->out_link_id[ ctx->rserve_out_idx ] ];
1630 0 : ctx->rserve_out_mem = topo->workspaces[ topo->objs[ rserve_out->dcache_obj_id ].wksp_id ].wksp;
1631 0 : ctx->rserve_out_chunk0 = fd_dcache_compact_chunk0( ctx->rserve_out_mem, rserve_out->dcache );
1632 0 : ctx->rserve_out_wmark = fd_dcache_compact_wmark ( ctx->rserve_out_mem, rserve_out->dcache, rserve_out->mtu );
1633 0 : ctx->rserve_out_chunk = ctx->rserve_out_chunk0;
1634 0 : }
1635 0 : }
1636 :
1637 : static ulong
1638 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
1639 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1640 : ulong out_cnt,
1641 0 : struct sock_filter * out ) {
1642 0 : populate_sock_filter_policy_fd_repair_tile( out_cnt, out, (uint)fd_log_private_logfile_fd() );
1643 0 : return sock_filter_policy_fd_repair_tile_instr_cnt;
1644 0 : }
1645 :
1646 : static ulong
1647 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
1648 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1649 : ulong out_fds_cnt,
1650 0 : int * out_fds ) {
1651 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
1652 :
1653 0 : ulong out_cnt = 0UL;
1654 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
1655 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
1656 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
1657 0 : return out_cnt;
1658 0 : }
1659 :
1660 0 : #define STEM_BURST (1UL)
1661 0 : #define STEM_LAZY (128L*3000L)
1662 :
1663 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_repair_tile_t
1664 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_repair_tile_t)
1665 :
1666 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1667 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
1668 0 : #define STEM_CALLBACK_NEXT_DEADLINE next_deadline
1669 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
1670 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
1671 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
1672 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
1673 :
1674 : #include "../../disco/stem/fd_stem.c"
1675 :
1676 : static ulong
1677 0 : max_event_sz( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
1678 0 : return sizeof(fd_event_block_received_t);
1679 0 : }
1680 :
1681 : fd_topo_run_tile_t fd_tile_repair = {
1682 : .name = "repair",
1683 : .max_event_sz = max_event_sz,
1684 : .populate_allowed_seccomp = populate_allowed_seccomp,
1685 : .populate_allowed_fds = populate_allowed_fds,
1686 : .scratch_align = scratch_align,
1687 : .scratch_footprint = scratch_footprint,
1688 : .unprivileged_init = unprivileged_init,
1689 : .privileged_init = privileged_init,
1690 : .run = stem_run,
1691 : };
|
