Line data Source code
1 : /* The repair tile is responsible for repairing missing shreds that were
2 : not received via Turbine. The goal is to ensure that slots we "care"
3 : about have their FEC sets inserted into store.
4 :
5 : Generally there are two distinct traffic patterns:
6 :
7 : a. Firedancer boots up and fires off a large number of repairs to
8 : recover all the blocks between the snapshot on which it is booting
9 : and the head of the chain. In this mode, repair tile utilization
10 : is very high along with net and sign utilization.
11 :
12 : b. Firedancer catches up to the head of the chain and enters steady
13 : state where most shred traffic is delivered over turbine. In this
14 : state, repairs are only occasionally needed to recover shreds lost
15 : due to anomalies like packet loss, transmitter (leader) never sent
16 : them or even a malicious leader etc. On rare occasion, repair
17 : will also need to recover a different version of a block that
18 : equivocated.
19 :
20 : To accomplish the above, repair mainly processes 4 kinds of frags:
21 :
22 : 1. Shred data (from shred tile)
23 :
24 : Any shred (coding or data) that passes validation and filtering in
25 : the shred tile is forwarded to repair. Repair uses these to track
26 : which shreds have been received in `fd_forest`, a tree data
27 : structure that mirrors the block/slot ancestry chain. It also
28 : uses these shreds to discover slots or ancestries that were not
29 : known. fd_forest tracks metadata for each slot, including slot
30 : completion status, merkle roots, and metrics.
31 :
32 : Any shred that we can correlate with a repair request we made is
33 : used to update peer response latency metrics in fd_policy (See
34 : fd_policy.h for more details).
35 :
36 : 2. FEC status messages (from shred tile)
37 :
38 : These fall under two categories: FEC completion and FEC eviction.
39 : When all shreds in a FEC set have been recovered, the shred tile
40 : sends a completion message. This may trigger chained merkle
41 : verification if the slot has a confirmed block_id. The completed
42 : FEC message is always forwarded to replay via repair_out.
43 :
44 : When an incomplete FEC set is evicted from the shred tile's FEC
45 : resolver (e.g. due to capacity), it also notifies repair. Repair
46 : clears the corresponding FEC set entries from the forest so those
47 : shred indices can be re-requested if they are necessary. As
48 : mentioned in fd_forest.h, forest needs to maintain a strict subset
49 : of shreds that are known by fec_resolver, store, and reasm in
50 : order to guarantee forward progress always.
51 :
52 : 3. Pings (from net tile)
53 :
54 : Repair peers use a ping-pong protocol to verify liveness before
55 : serving repair requests. When a ping arrives over the network,
56 : repair validates the message and constructs a pong response. To
57 : prevent spam attacks, repair has stopgaps like tracking how many
58 : pongs per peer are currently in the sign queue and dropping pings
59 : from unknown peers. These are the only untrusted inputs to
60 : repair.
61 :
62 : 4. Sign task responses (from sign tile)
63 :
64 : Repair requests are signed asynchronously; the repair tile
65 : constructs a repair request, dispatches it to a sign tile via the
66 : repair_sign output link. The repair-sign communication is
67 : manually managed via credit tracking in the repair tile (see
68 : comment in out_ctx_t struct definition).
69 :
70 : After receiving the signature back from sign tile, repair injects
71 : the signature into the pending request and dispatches it to the
72 : net tile via the repair_net output link. The behavior depends on
73 : the request type:
74 : - Pong: the signed pong is sent to the peer that pinged us.
75 : - Warmup: a proactive request sent when a new peer's contact info
76 : first arrives, prepaying the ping-pong RTT cost. The signed
77 : request is sent to the peer if it is still active.
78 : - Regular shred request: the request is recorded in an inflight
79 : table (for tracking response latency and timeouts) and the
80 : signed packet is sent to the selected peer.
81 :
82 : Secondary "other" frags that are processed but not part of core
83 : repair logic:
84 :
85 : 5. Confirmation messages from tower
86 :
87 : Tower sends two kinds of messages relevant to repair:
88 : - slot_done: indicates a slot has finished replay and may advance
89 : the root. Repair publishes (roots) the forest up to that slot,
90 : pruning old ancestry.
91 : - slot_confirmed: indicates a slot has reached a confirmation
92 : level (e.g. duplicate-confirmed). If the slot is not yet in the
93 : forest, repair creates a sentinel block so it can be repaired.
94 : It also stores the confirmed block_id and may trigger chained
95 : merkle verification. See fd_forest.h on more details about
96 : chained merkle verification.
97 :
98 : 6. Eviction messages from replay (reasm)
99 :
100 : When the replay tile's reassembly buffer evicts a FEC set (e.g.
101 : due to pool capacity) from itself and from store, it notifies
102 : repair with the slot and fec_set_idx. Repair clears those FEC
103 : entries from the forest so the shreds can be re-requested.
104 :
105 : 7. Contact info messages from gossip
106 :
107 : Gossip forwards contact info updates and removals for other
108 : validators. Repair uses these to maintain a list of peers to
109 : make requests to.
110 :
111 : If fd_forest tracks what we know about each shred, fd_policy and
112 : fd_inflights is responsible for deciding what next repair request to
113 : make. fd_policy and fd_inflights split responsibility: fd_policy
114 : makes any new requests, orphan requests, and requests directly off
115 : the forest iterator, while fd_inflights re-requests anything that has
116 : been requested but not received yet within a timeout window. */
117 :
118 : #define _GNU_SOURCE
119 :
120 : #include "../../disco/topo/fd_topo.h"
121 : #include "../../disco/fd_clock_tile.h"
122 : #include <linux/futex.h>
123 : #include "generated/fd_repair_tile_seccomp.h"
124 : #include "../../disco/keyguard/fd_keyload.h"
125 : #include "../../disco/keyguard/fd_keyguard.h"
126 : #include "../../disco/keyguard/fd_keyswitch.h"
127 : #include "../../disco/metrics/fd_metrics.h"
128 : #include "../../disco/net/fd_net_tile.h"
129 : #include "../../disco/shred/fd_rnonce_ss.h"
130 : #include "../../disco/shred/fd_shred_tile.h"
131 : #include "fd_repair_tile.h"
132 : #include "../../flamenco/gossip/fd_gossip_message.h"
133 : #include "../replay/fd_replay_tile.h"
134 : #include "../tower/fd_tower_tile.h"
135 : #include "../../discof/restore/utils/fd_ssmsg.h"
136 : #include "../../util/net/fd_net_headers.h"
137 : #include "../../util/pod/fd_pod_format.h"
138 : #include "../../tango/fd_tango_base.h"
139 :
140 : #include "../forest/fd_forest.h"
141 : #include "fd_repair_metrics.h"
142 : #include "fd_inflight.h"
143 : #include "fd_repair.h"
144 : #include "fd_policy.h"
145 : #include "fd_repair_tile_private.h"
146 :
147 : #define DEBUG_LOGGING 0
148 :
149 : #define IN_KIND_CONTACT (0)
150 0 : #define IN_KIND_NET (1)
151 0 : #define IN_KIND_TOWER (2)
152 0 : #define IN_KIND_SHRED (3)
153 0 : #define IN_KIND_SIGN (4)
154 0 : #define IN_KIND_SNAP (5)
155 0 : #define IN_KIND_GOSSIP (6)
156 0 : #define IN_KIND_REPLAY (7)
157 :
158 : /* static map from request type to metric array index */
159 : static uint metric_index[FD_REPAIR_KIND_ORPHAN + 1] = {
160 : [FD_REPAIR_KIND_PONG] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_PONG_IDX,
161 : [FD_REPAIR_KIND_SHRED] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_WINDOW_INDEX_IDX,
162 : [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_HIGHEST_WINDOW_INDEX_IDX,
163 : [FD_REPAIR_KIND_ORPHAN] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_ORPHAN_IDX,
164 : };
165 :
166 : FD_FN_CONST static inline ulong
167 0 : scratch_align( void ) {
168 0 : return 128UL;
169 0 : }
170 :
171 : FD_FN_PURE static inline ulong
172 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
173 0 : ulong total_sign_depth = tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt;
174 0 : int lg_sign_depth = fd_ulong_find_msb( fd_ulong_pow2_up(total_sign_depth) ) + 1;
175 :
176 0 : ulong l = FD_LAYOUT_INIT;
177 0 : l = FD_LAYOUT_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
178 0 : l = FD_LAYOUT_APPEND( l, fd_repair_align(), fd_repair_footprint () );
179 0 : l = FD_LAYOUT_APPEND( l, fd_forest_align(), fd_forest_footprint ( tile->repair.slot_max, tile->repair.max_shreds_per_block ) );
180 0 : l = FD_LAYOUT_APPEND( l, fd_policy_align(), fd_policy_footprint ( FD_REPAIR_PEER_MAX ) );
181 0 : l = FD_LAYOUT_APPEND( l, fd_reqlim_align(), fd_reqlim_footprint ( FD_REQLIM_CACHE_MAX ) );
182 0 : l = FD_LAYOUT_APPEND( l, fd_inflights_align(), fd_inflights_footprint () );
183 0 : l = FD_LAYOUT_APPEND( l, fd_signs_map_align(), fd_signs_map_footprint ( lg_sign_depth ) );
184 0 : l = FD_LAYOUT_APPEND( l, fd_signs_queue_align(), fd_signs_queue_footprint() );
185 0 : l = FD_LAYOUT_APPEND( l, fd_repair_metrics_align(), fd_repair_metrics_footprint() );
186 0 : return FD_LAYOUT_FINI( l, scratch_align() );
187 0 : }
188 :
189 : /* Below functions manage the current pending sign requests. */
190 :
191 : static sign_req_t *
192 : sign_map_insert( ctx_t * ctx,
193 : fd_repair_msg_t const * msg,
194 0 : pong_data_t const * opt_pong_data ) {
195 :
196 : /* Check if there is any space for a new pending sign request. Should never fail as long as credit management is working. */
197 0 : if( FD_UNLIKELY( fd_signs_map_key_cnt( ctx->signs_map )==fd_signs_map_key_max( ctx->signs_map ) ) ) return NULL;
198 :
199 0 : sign_req_t * pending = fd_signs_map_insert( ctx->signs_map, ctx->pending_key_next++ );
200 0 : if( FD_UNLIKELY( !pending ) ) return NULL; /* Not possible, unless the same key is used twice. */
201 0 : pending->msg = *msg;
202 0 : pending->buflen = fd_repair_sz( msg );
203 0 : if( FD_UNLIKELY( opt_pong_data ) ) pending->pong_data = *opt_pong_data;
204 0 : return pending;
205 0 : }
206 :
207 : static int
208 : sign_map_remove( ctx_t * ctx,
209 0 : ulong key ) {
210 0 : sign_req_t * pending = fd_signs_map_query( ctx->signs_map, key, NULL );
211 0 : if( FD_UNLIKELY( !pending ) ) return -1;
212 0 : fd_signs_map_remove( ctx->signs_map, pending );
213 0 : return 0;
214 0 : }
215 :
216 : static void
217 : send_packet( ctx_t * ctx,
218 : fd_stem_context_t * stem,
219 : uint dst_ip_addr,
220 : ushort dst_port,
221 : uint src_ip_addr,
222 : uchar const * payload,
223 : ulong payload_sz,
224 0 : ulong tsorig ) {
225 0 : ctx->metrics->send_pkt_cnt++;
226 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_ctx->mem, ctx->net_out_ctx->chunk );
227 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
228 0 : *hdr = *ctx->intake_hdr;
229 :
230 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
231 0 : ip4->saddr = src_ip_addr;
232 0 : ip4->daddr = dst_ip_addr;
233 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
234 0 : ip4->check = 0U;
235 0 : ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
236 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
237 :
238 0 : fd_udp_hdr_t * udp = hdr->udp;
239 0 : udp->net_dport = dst_port;
240 0 : udp->net_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
241 0 : fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
242 0 : hdr->udp->check = 0U;
243 :
244 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
245 0 : ulong sig = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
246 0 : ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
247 0 : ulong chunk = ctx->net_out_ctx->chunk;
248 0 : fd_stem_publish( stem, ctx->net_out_ctx->idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
249 0 : ctx->net_out_ctx->chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_ctx->chunk0, ctx->net_out_ctx->wmark );
250 0 : }
251 :
252 : /* Returns a sign_out context with max available credits.
253 : If no sign_out context has available credits, returns NULL. */
254 : static out_ctx_t *
255 0 : sign_avail_credits( ctx_t * ctx ) {
256 0 : out_ctx_t * sign_out = NULL;
257 0 : ulong max_credits = 0;
258 0 : for( uint i = 0; i < ctx->repair_sign_cnt; i++ ) {
259 0 : if( ctx->repair_sign_out_ctx[i].credits > max_credits ) {
260 0 : max_credits = ctx->repair_sign_out_ctx[i].credits;
261 0 : sign_out = &ctx->repair_sign_out_ctx[i];
262 0 : }
263 0 : }
264 0 : return sign_out;
265 0 : }
266 :
267 : /* Prepares the signing preimage and publishes a signing request that
268 : will be signed asynchronously by the sign tile. The signed data will
269 : be returned via dcache as a frag. */
270 : static void
271 : fd_repair_send_sign_request( ctx_t * ctx,
272 : out_ctx_t * sign_out,
273 : fd_repair_msg_t const * msg,
274 0 : pong_data_t const * opt_pong_data ) {
275 :
276 0 : if( FD_UNLIKELY( ctx->halt_signing ) ) FD_LOG_CRIT(( "can't dispatch sign requests while halting signing" ));
277 :
278 : /* New sign request */
279 0 : sign_req_t * pending = sign_map_insert( ctx, msg, opt_pong_data );
280 0 : if( FD_UNLIKELY( !pending ) ) return;
281 :
282 0 : ulong sig = 0;
283 0 : ulong preimage_sz = 0;
284 0 : uchar * dst = fd_chunk_to_laddr( sign_out->mem, sign_out->chunk );
285 :
286 0 : if( FD_UNLIKELY( msg->kind == FD_REPAIR_KIND_PONG ) ) {
287 0 : uchar pre_image[FD_REPAIR_PONG_PREIMAGE_SZ];
288 0 : preimage_pong( &opt_pong_data->hash, pre_image );
289 0 : preimage_sz = FD_REPAIR_PONG_PREIMAGE_SZ;
290 0 : fd_memcpy( dst, pre_image, preimage_sz );
291 0 : sig = ((ulong)pending->key << 32) | (uint)FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519;
292 0 : } else {
293 : /* Sign and prepare the message directly into the pending buffer */
294 0 : uchar * preimage = preimage_req( &pending->msg, &preimage_sz );
295 0 : fd_memcpy( dst, preimage, preimage_sz );
296 0 : sig = ((ulong)pending->key << 32) | (uint)FD_KEYGUARD_SIGN_TYPE_ED25519;
297 0 : }
298 :
299 0 : fd_stem_publish( ctx->stem, sign_out->idx, sig, sign_out->chunk, preimage_sz, 0UL, 0UL, 0UL );
300 0 : sign_out->chunk = fd_dcache_compact_next( sign_out->chunk, preimage_sz, sign_out->chunk0, sign_out->wmark );
301 :
302 0 : ctx->metrics->sent_pkt_types[metric_index[msg->kind]]++;
303 0 : sign_out->credits--;
304 0 : }
305 :
306 : static inline int
307 : before_frag( ctx_t * ctx,
308 : ulong in_idx,
309 : ulong seq FD_PARAM_UNUSED,
310 0 : ulong sig ) {
311 0 : uint in_kind = ctx->in_kind[ in_idx ];
312 0 : if( FD_LIKELY ( in_kind==IN_KIND_NET ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
313 0 : if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) return fd_int_if( fd_forest_root_slot( ctx->forest )==ULONG_MAX, -1, 0 ); /* not ready to read frag */
314 0 : if( FD_UNLIKELY( in_kind==IN_KIND_GOSSIP ) ) {
315 0 : return sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO &&
316 0 : sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
317 0 : }
318 0 : if( FD_UNLIKELY( in_kind==IN_KIND_REPLAY ) ) return sig!=REPLAY_SIG_REASM_EVICTED;
319 0 : return 0;
320 0 : }
321 :
322 : static void
323 : during_frag( ctx_t * ctx,
324 : ulong in_idx,
325 : ulong seq FD_PARAM_UNUSED,
326 : ulong sig,
327 : ulong chunk,
328 : ulong sz,
329 0 : ulong ctl ) {
330 0 : ctx->skip_frag = 0;
331 :
332 0 : uint in_kind = ctx->in_kind[ in_idx ];
333 0 : in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
334 0 : ctx->chunk = chunk;
335 :
336 0 : if( FD_UNLIKELY( in_kind==IN_KIND_NET ) ) {
337 0 : ulong hdr_sz = fd_disco_netmux_sig_hdr_sz( sig );
338 0 : FD_TEST( hdr_sz <= sz ); /* Should be ensured by the net tile */
339 0 : uchar const * dcache_entry = fd_net_rx_translate_frag( &in_ctx->net_rx, chunk, ctl, sz );
340 0 : fd_memcpy( ctx->net_buf, dcache_entry, sz );
341 0 : return;
342 0 : }
343 :
344 0 : if( FD_UNLIKELY( sz!=0UL && ( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) ) )
345 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu] in kind %u", chunk, sz, in_ctx->chunk0, in_ctx->wmark, in_kind ));
346 :
347 0 : if( FD_UNLIKELY( in_kind==IN_KIND_SNAP ) ) {
348 0 : if( FD_UNLIKELY( fd_ssmsg_sig_message( sig )!=FD_SSMSG_DONE ) ) ctx->snap_out_chunk = chunk;
349 0 : return;
350 0 : }
351 :
352 0 : if( FD_UNLIKELY( in_kind==IN_KIND_SIGN ) ) {
353 : /* sign_repair is unreliable, so we copy the frag for convention.
354 : Theoretically impossible to overrun. */
355 0 : uchar const * dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
356 0 : fd_memcpy( ctx->sign_buf, dcache_entry, sz );
357 0 : return;
358 0 : }
359 0 : }
360 :
361 : static inline void
362 : after_snap( ctx_t * ctx,
363 : ulong sig,
364 0 : uchar const * chunk ) {
365 0 : if( FD_UNLIKELY( fd_ssmsg_sig_message( sig )!=FD_SSMSG_DONE ) ) return;
366 0 : fd_snapshot_manifest_t * manifest = (fd_snapshot_manifest_t *)chunk;
367 :
368 0 : fd_forest_init( ctx->forest, manifest->slot );
369 0 : }
370 :
371 : static inline void
372 0 : after_gossip( ctx_t * ctx, fd_gossip_update_message_t const * msg, ulong sig ) {
373 0 : switch( sig ) {
374 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE: {
375 0 : fd_policy_peer_remove( ctx->policy, fd_type_pun_const( msg->origin ) );
376 0 : break;
377 0 : }
378 0 : case FD_GOSSIP_UPDATE_TAG_CONTACT_INFO: {
379 0 : fd_gossip_contact_info_t const * contact_info = msg->contact_info->value;
380 0 : fd_ip4_port_t repair_peer;
381 0 : repair_peer.addr = contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_SERVE_REPAIR ].is_ipv6 ? 0U : contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_SERVE_REPAIR ].ip4;
382 0 : repair_peer.port = contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_SERVE_REPAIR ].port;
383 0 : if( FD_UNLIKELY( !repair_peer.addr || !repair_peer.port ) ) return;
384 0 : fd_policy_peer_t const * peer = fd_policy_peer_upsert( ctx->policy, fd_type_pun_const( msg->origin ), &repair_peer );
385 0 : if( FD_LIKELY( peer && !fd_signs_queue_full( ctx->pong_queue ) ) ) {
386 : /* The repair process uses a Ping-Pong protocol that incurs one
387 : round-trip time (RTT) for the initmial repair request. To
388 : optimize this, we proactively send a placeholder repair request
389 : as soon as we receive a peer's contact information for the first
390 : time, effectively prepaying the RTT cost. */
391 0 : fd_repair_msg_t * init = fd_repair_shred( ctx->protocol, fd_type_pun_const( msg->origin ), (ulong)fd_clock_tile_now( ctx->clock )/1000000UL, 0, 0, 0 );
392 0 : fd_signs_queue_push( ctx->pong_queue, (sign_pending_t){ .msg = *init } );
393 0 : }
394 0 : break;
395 0 : }
396 0 : default: FD_LOG_ERR(( "bad gossip sig %lu", sig ));
397 0 : }
398 0 : }
399 :
400 : static inline void
401 : after_sign( ctx_t * ctx,
402 : ulong in_idx,
403 : ulong sig,
404 0 : fd_stem_context_t * stem ) {
405 0 : ulong pending_key = sig >> 32;
406 : /* Look up the pending request. Since the repair_sign links are
407 : reliable, the incoming sign_repair fragments represent a complete
408 : set of the previously sent outgoing messages. However, with
409 : multiple sign tiles, the responses may arrive interleaved. */
410 :
411 : /* Find which sign tile sent this response and increment its credits */
412 0 : for( uint i = 0; i < ctx->repair_sign_cnt; i++ ) {
413 0 : if( ctx->repair_sign_out_ctx[i].in_idx == in_idx ) {
414 0 : if( FD_LIKELY( ctx->repair_sign_out_ctx[i].credits < ctx->repair_sign_out_ctx[i].max_credits ) ) ctx->repair_sign_out_ctx[i].credits++;
415 0 : break;
416 0 : }
417 0 : }
418 :
419 0 : sign_req_t * pending_ = fd_signs_map_query( ctx->signs_map, pending_key, NULL );
420 0 : if( FD_UNLIKELY( !pending_ ) ) FD_LOG_CRIT(( "No pending request found for key %lu", pending_key )); /* implies either bad programmer error or something happened with sign tile */
421 :
422 0 : sign_req_t pending[1] = { *pending_ }; /* Make a copy of the pending request so we can sign_map_remove immediately. */
423 0 : sign_map_remove( ctx, pending_key );
424 :
425 : /* This is a pong message */
426 0 : if( FD_UNLIKELY( pending->msg.kind == FD_REPAIR_KIND_PONG ) ) {
427 0 : fd_policy_peer_t * peer = fd_policy_peer_query( ctx->policy, &pending->pong_data.key );
428 0 : if( FD_LIKELY( peer && peer->ping ) ) peer->ping--; /* prevent underflow if the peer was removed/readded */
429 :
430 0 : fd_memcpy( pending->msg.pong.sig, ctx->sign_buf, 64UL );
431 0 : send_packet( ctx, stem, pending->pong_data.peer_addr.addr, pending->pong_data.peer_addr.port, pending->pong_data.daddr, pending->buf, fd_repair_sz( &pending->msg ), fd_frag_meta_ts_comp( fd_tickcount() ) );
432 0 : return;
433 0 : }
434 :
435 : /* Inject the signature into the pending request */
436 0 : fd_memcpy( pending->buf + 4, ctx->sign_buf, 64UL );
437 0 : uint src_ip4 = 0U;
438 :
439 : /* This is a warmup message */
440 0 : if( FD_UNLIKELY( pending->msg.kind == FD_REPAIR_KIND_SHRED && pending->msg.shred.slot == 0 ) ) {
441 0 : fd_policy_peer_t * peer = fd_policy_peer_query( ctx->policy, &pending->msg.shred.to );
442 0 : if( FD_UNLIKELY( peer ) ) send_packet( ctx, stem, peer->ip4, peer->port, src_ip4, pending->buf, pending->buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
443 0 : else { /* This is a warmup request for a peer that is no longer active. There's no reason to pick another peer for a warmup rq, so just drop it. */ }
444 0 : return;
445 0 : }
446 :
447 : /* This is a regular repair shred request
448 :
449 : We need to ensure we always send out any shred requests we have,
450 : because policy_next has no way to revisit a shred. But the fact
451 : that peers can drop out of the peer list makes this complicated.
452 : If the peer is still there (common), it's fine. If the peer is not
453 : there, we can add this request to the inflights table, pretend
454 : we've sent it and let the inflight timeout request it down the
455 : line. */
456 :
457 0 : fd_policy_peer_t * active = fd_policy_peer_query( ctx->policy, &pending->msg.shred.to );
458 0 : if( FD_UNLIKELY( !active ) ) {
459 : /* Already added to the inflights table, pretend we've sent it
460 : and let the inflight timeout request it down the line. */
461 0 : return;
462 0 : }
463 : /* Happy path - all is well, our peer didn't drop out from beneath us. */
464 0 : if( FD_UNLIKELY( pending->msg.kind == FD_REPAIR_KIND_ORPHAN ) ) ctx->metrics->last_requested_orphan = pending->msg.orphan.slot;
465 0 : else ctx->metrics->last_requested_slot = pending->msg.shred.slot;
466 :
467 0 : send_packet( ctx, stem, active->ip4, active->port, src_ip4, pending->buf, pending->buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
468 0 : }
469 :
470 : static int
471 0 : blk_insert_check( ctx_t * ctx, fd_forest_blk_t * new_blk, ulong new_slot, ulong evicted ) {
472 0 : if( FD_UNLIKELY( !new_blk ) ) {
473 0 : ctx->metrics->blk_failed_insert++;
474 0 : ctx->metrics->slot_failed_insert = new_slot;
475 0 : return 0;
476 0 : } else {
477 0 : if( FD_UNLIKELY( evicted != ULONG_MAX ) ) {
478 0 : ctx->metrics->blk_evicted++;
479 0 : ctx->metrics->slot_evicted = evicted;
480 0 : ctx->metrics->slot_evicted_by = new_slot;
481 0 : }
482 0 : return 1;
483 0 : }
484 0 : }
485 :
486 0 : static inline int shred_src( ulong sig ) {
487 0 : uint sig_src = fd_shred_sig_src( sig );
488 0 : switch( sig_src ) {
489 0 : case SHRED_SIG_SRC_TURBINE:
490 0 : return SHRED_SRC_TURBINE;
491 0 : case SHRED_SIG_SRC_RECONSTRUCTED:
492 0 : return SHRED_SRC_RECOVERED;
493 0 : case SHRED_SIG_SRC_REPAIR:
494 0 : return SHRED_SRC_REPAIR;
495 0 : case SHRED_SIG_SRC_BAD_REPAIR:
496 0 : return SHRED_SRC_TURBINE; /* like fec_resolver, treat bad repair shreds as turbine shreds */
497 0 : case SHRED_SIG_SRC_LEADER:
498 0 : return SHRED_SRC_LEADER;
499 0 : default: FD_LOG_CRIT(( "bad shred sig src %u", sig_src ));
500 0 : }
501 0 : }
502 :
503 : static inline void
504 : after_shred( ctx_t * ctx,
505 : ulong sig,
506 : fd_shred_t * shred,
507 : ulong nonce,
508 : fd_hash_t * mr,
509 : fd_hash_t * cmr,
510 0 : long rx_tick ) {
511 :
512 0 : int is_code = fd_shred_is_code( fd_shred_type( shred->variant ) );
513 :
514 : /* we don't want to add a slot to the forest that chains to a slot
515 : older than root, to avoid filling forest up with junk.
516 : Especially if we are close to full and we are having trouble
517 : rooting, we can't rely on publishing to prune these useless
518 : subtrees. TODO: do the same with reasm/store/shred? */
519 :
520 0 : if( FD_UNLIKELY( shred->slot <= fd_forest_root_slot( ctx->forest ) ||
521 0 : ( !is_code && shred->slot - shred->data.parent_off < fd_forest_root_slot( ctx->forest ) ) ) ) {
522 0 : ctx->metrics->old_shred++;
523 0 : return;
524 0 : }
525 :
526 : /* Insert the shred sig (shared by all shred members in the FEC set)
527 : into the map. */
528 0 : int src = shred_src( sig );
529 :
530 0 : if( FD_LIKELY( !is_code ) ) {
531 0 : long rtt = 0;
532 0 : fd_pubkey_t peer;
533 :
534 0 : int slot_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE);
535 0 : int ref_tick = shred->data.flags & FD_SHRED_DATA_REF_TICK_MASK;
536 0 : ulong evicted = ULONG_MAX;
537 0 : fd_forest_blk_t * blk = fd_forest_blk_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, &evicted );
538 0 : if( FD_UNLIKELY( !blk_insert_check( ctx, blk, shred->slot, evicted ) ) ) return;
539 :
540 0 : if( FD_LIKELY( fd_forest_data_shred_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, shred->idx, shred->fec_set_idx, slot_complete, ref_tick, src, mr, cmr, rx_tick ) ) ) {
541 0 : if( FD_UNLIKELY( src == SHRED_SRC_REPAIR && ( rtt = fd_inflights_shred_match( ctx->inflights, FD_REPAIR_KIND_SHRED, nonce, shred->slot, shred->idx, NULL, &peer, NULL, fd_clock_tile_now( ctx->clock ) ) ) > 0 ) ) {
542 0 : fd_policy_peer_response_update( ctx->policy, &peer, rtt );
543 0 : fd_histf_sample( ctx->metrics->response_latency, (ulong)rtt );
544 0 : blk->response_cnt++;
545 0 : blk->last_repair_resp_ts = rx_tick;
546 0 : }
547 0 : }
548 0 : } else {
549 0 : fd_forest_code_shred_insert( ctx->forest, shred->slot, shred->idx, rx_tick );
550 0 : }
551 0 : }
552 :
553 : /* Kicks off the chained merkle verification starting at a slot with
554 : a confirmed, canonical block_id. Either finishes successfully and
555 : returns early, or detects an incorrect FEC set and clears it. In
556 : this case the verification is paused and state is saved at where
557 : it left off. Verification can be re-triggered in after_fec as well. */
558 : static inline void
559 : check_confirmed( ctx_t * ctx,
560 : fd_forest_blk_t * blk,
561 0 : fd_hash_t const * confirmed_bid ) {
562 :
563 0 : if( FD_LIKELY( !blk->chain_confirmed && blk->complete_idx != UINT_MAX ) ) {
564 : /* The above conditions say that all the shreds of the block have arrived. */
565 0 : fd_forest_blk_t * bad_blk = fd_forest_fec_chain_verify( ctx->forest, blk, confirmed_bid );
566 0 : if( FD_LIKELY( !bad_blk ) ) {
567 : /* chain verified successfully from blk to as far as we have fec data */
568 0 : return;
569 0 : }
570 :
571 0 : uint bad_fec_idx = fd_forest_merkle_last_incorrect_idx( bad_blk );
572 : /* bad_fec_idx UINT_MAX implies this slot is fully validated. Not
573 : possible here because that implies bad_blk is NULL and would
574 : early exit above. */
575 0 : FD_TEST( bad_fec_idx != UINT_MAX );
576 :
577 0 : fd_forest_mr_t const * bad_mroots = fd_forest_blk_mroots( ctx->forest, bad_blk );
578 0 : fd_hash_t const * expected = (bad_fec_idx == bad_blk->complete_idx - (FD_FEC_SHRED_CNT - 1)) ? &bad_blk->confirmed_bid : &bad_mroots[(bad_fec_idx / 32) + 1].cmr;
579 :
580 0 : FD_BASE58_ENCODE_32_BYTES( confirmed_bid->uc, confirmed_bid_b58 );
581 0 : FD_BASE58_ENCODE_32_BYTES( expected->uc, expected_mr );
582 0 : FD_BASE58_ENCODE_32_BYTES( bad_mroots[bad_fec_idx / 32].mr.uc, recorded_mr );
583 :
584 0 : FD_LOG_WARNING(( "[%s] slot %lu block_id %s confirmation detected incorrect FECs. bad FEC is slot %lu fec set %u. expected mr (%s) != recorded mr (%s)",
585 0 : __func__,
586 0 : blk->slot,
587 0 : confirmed_bid_b58,
588 0 : bad_blk->slot,
589 0 : bad_fec_idx,
590 0 : expected_mr,
591 0 : recorded_mr ));
592 :
593 0 : ctx->metrics->failed_chain_verify_cnt++;
594 0 : ctx->metrics->failed_chain_verify_slot = bad_blk->slot;
595 0 : bad_blk->chain_verify_failed = 1;
596 :
597 : /* If we have a bad block, we need to dump and repair backwards from
598 : the point where the merkle root is incorrect.
599 : We start only by dumping the last incorrect FEC. It's possible that
600 : this is the only incorrect one. If it isn't though, when the slot
601 : recompletes, this function will trigger again and we will dump the
602 : second to last incorrect FEC. */
603 :
604 0 : fd_forest_fec_clear( ctx->forest, bad_blk->slot, bad_fec_idx, FD_FEC_SHRED_CNT - 1 );
605 0 : }
606 0 : }
607 :
608 : /* Returns 1 if the fec is guaranteed invalid, 0 otherwise. */
609 : static inline int
610 : after_fec( ctx_t * ctx,
611 : fd_shred_t * shred,
612 : fd_hash_t * mr,
613 : fd_hash_t * cmr,
614 0 : long rx_tick ) {
615 :
616 : /* When this is a FEC completes msg, it is implied that all the
617 : other shreds in the FEC set can also be inserted. Shred inserts
618 : into the forest are idempotent so it is fine to insert the same
619 : shred multiple times. */
620 :
621 0 : int slot_complete = !!( shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE );
622 0 : int ref_tick = shred->data.flags & FD_SHRED_DATA_REF_TICK_MASK;
623 :
624 : /* Similar to after_shred, do not insert a slot that chains to a slot older than root */
625 0 : if( FD_UNLIKELY( shred->slot <= fd_forest_root_slot( ctx->forest ) ||
626 0 : shred->slot - shred->data.parent_off < fd_forest_root_slot( ctx->forest ) ) ) return 0;
627 :
628 0 : ulong evicted = ULONG_MAX;
629 0 : fd_forest_blk_t * ele = fd_forest_blk_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, &evicted );
630 0 : if( FD_UNLIKELY( !blk_insert_check( ctx, ele, shred->slot, evicted ) ) ) return 0;
631 0 : ele = fd_forest_fec_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, shred->idx, shred->fec_set_idx, slot_complete, ref_tick, mr, cmr, rx_tick );
632 0 : if( FD_UNLIKELY( !ele ) ) return 1;
633 :
634 : /* metrics for completed slots */
635 0 : if( FD_UNLIKELY( ele->complete_idx != UINT_MAX && ele->buffered_idx==ele->complete_idx ) ) {
636 0 : long now = fd_tickcount();
637 0 : long start_ts = ele->first_req_ts == 0 || ele->slot >= ctx->turbine_slot0 ? ele->first_shred_ts : ele->first_req_ts;
638 0 : ulong duration_ticks = (ulong)(now - start_ts);
639 0 : fd_histf_sample( ctx->metrics->slot_compl_time, duration_ticks );
640 0 : fd_repair_metrics_add_slot( ctx->slot_metrics, ele->slot, start_ts, now, ele->repair_cnt, ele->turbine_cnt );
641 : /* Note: this log does not imply that the slot is fully executable.
642 : It's possible that we have a slot that doesn't chain verify,
643 : which could be un-executable. */
644 0 : FD_BASE58_ENCODE_32_BYTES( fd_forest_blk_mroots( ctx->forest, ele )[ele->complete_idx / 32].mr.uc, block_id );
645 0 : FD_BASE58_ENCODE_32_BYTES( mr->uc, fec_mr );
646 0 : FD_LOG_INFO(( "[%s] slot is complete %lu. num_data_shreds: %u, num_repaired: %u, num_turbine: %u, num_recovered: %u, duration: %.2f ms. last recvd fec: %u, mr %s. current block_id: %s",
647 0 : __func__,
648 0 : ele->slot,
649 0 : ele->complete_idx + 1,
650 0 : ele->repair_cnt,
651 0 : ele->turbine_cnt,
652 0 : ele->recovered_cnt,
653 0 : (double)fd_metrics_convert_ticks_to_nanoseconds(duration_ticks) / 1e6,
654 0 : shred->fec_set_idx,
655 0 : fec_mr,
656 0 : block_id ));
657 0 : }
658 :
659 : /* re-trigger continuation of chained merkle verification if slot is
660 : complete. TODO MOVE TO AFTER_SHRED? */
661 0 : fd_hash_t empty_mr = (fd_hash_t){ .ul = { 0, 0, 0, 0 } };
662 0 : if( FD_UNLIKELY( ele->buffered_idx == ele->complete_idx && !fd_hash_eq( &ele->confirmed_bid, &empty_mr ) ) ) {
663 0 : check_confirmed( ctx, ele, &ele->confirmed_bid /* if lowest_verified_fec is not UINT_MAX, confirmed_bid must be populated */ );
664 0 : }
665 0 : return 0;
666 0 : }
667 :
668 : static inline void
669 : after_net( ctx_t * ctx,
670 0 : ulong sz ) {
671 0 : fd_eth_hdr_t * eth; fd_ip4_hdr_t * ip4; fd_udp_hdr_t * udp;
672 0 : uchar * data; ulong data_sz;
673 0 : if( FD_UNLIKELY( !fd_ip4_udp_hdr_strip( ctx->net_buf, sz, &data, &data_sz, ð, &ip4, &udp ) ) ) {
674 0 : ctx->metrics->malformed_ping++;
675 0 : return;
676 0 : }
677 0 : fd_ip4_port_t peer_addr = { .addr=ip4->saddr, .port=udp->net_sport };
678 :
679 0 : fd_repair_ping_t ping[1];
680 0 : int err = fd_repair_ping_de( ping, data, data_sz );
681 0 : if( FD_UNLIKELY( err ) ) {
682 0 : ctx->metrics->malformed_ping++;
683 0 : return;
684 0 : }
685 :
686 0 : fd_policy_peer_t * peer = fd_policy_peer_query( ctx->policy, &ping->ping.from );
687 0 : if( FD_UNLIKELY( !peer ) ) {
688 0 : ctx->metrics->unknown_peer_ping++;
689 0 : return;
690 0 : }
691 0 : if( FD_UNLIKELY( peer->ping ) ) return;
692 0 : if( FD_UNLIKELY( fd_signs_queue_full( ctx->pong_queue ) ) ) return;
693 :
694 0 : fd_sha512_t sha[1];
695 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS != fd_ed25519_verify( ping->ping.hash.uc, 32UL, ping->ping.sig, ping->ping.from.uc, sha ) ) ) {
696 0 : ctx->metrics->fail_sigverify_ping++;
697 0 : return;
698 0 : }
699 :
700 : /* Any gossip peer can send a ping, but they are bounded to at most
701 : one ping in the queue so they can't evict others' pings without
702 : multiple gossip identities. */
703 :
704 0 : fd_repair_msg_t * pong = fd_repair_pong( ctx->protocol, &ping->ping.hash );
705 0 : fd_signs_queue_push( ctx->pong_queue, (sign_pending_t){ .msg = *pong, .pong_data = { .peer_addr = peer_addr, .hash = ping->ping.hash, .daddr = ip4->daddr, .key = ping->ping.from } } );
706 0 : peer->ping++;
707 0 : }
708 :
709 : static inline void
710 : after_evict( ctx_t * ctx,
711 0 : fd_fec_evicted_t * evicted ) {
712 0 : fd_forest_fec_clear( ctx->forest, evicted->slot, evicted->fec_set_idx, FD_FEC_SHRED_CNT - 1 );
713 0 : }
714 :
715 : static inline void
716 : after_tower( ctx_t * ctx,
717 : ulong sig,
718 0 : uchar const * chunk ) {
719 :
720 0 : switch( sig ) {
721 0 : case FD_TOWER_SIG_SLOT_DONE: {
722 0 : fd_tower_slot_done_t const * msg = (fd_tower_slot_done_t const *)fd_type_pun_const( chunk );
723 0 : if( FD_LIKELY( msg->root_slot!=ULONG_MAX && msg->root_slot > fd_forest_root_slot( ctx->forest ) ) ) fd_forest_publish( ctx->forest, msg->root_slot );
724 0 : break;
725 0 : }
726 0 : case FD_TOWER_SIG_SLOT_CONFIRMED: {
727 0 : fd_tower_slot_confirmed_t const * msg = (fd_tower_slot_confirmed_t const *)fd_type_pun_const( chunk );
728 0 : if( msg->slot > fd_forest_root_slot( ctx->forest ) && (msg->level >= FD_TOWER_SLOT_CONFIRMED_DUPLICATE ) ) {
729 0 : fd_forest_blk_t * blk = fd_forest_query( ctx->forest, msg->slot );
730 0 : if( FD_UNLIKELY( !blk ) ) {
731 : /* If we receive a confirmation for a slot we don't have,
732 : create a sentinel forest block that we can repair from. */
733 0 : ulong evicted = ULONG_MAX;
734 0 : blk = fd_forest_blk_insert( ctx->forest, msg->slot, ULONG_MAX, &evicted );
735 0 : FD_LOG_INFO(("[%s] creating sentinel for duplicate confirmed block %lu", __func__, msg->slot));
736 0 : if( FD_UNLIKELY( !blk_insert_check( ctx, blk, msg->slot, evicted ) ) ) return;
737 0 : }
738 :
739 : /* Confirm the block */
740 0 : blk->confirmed_bid = msg->block_id;
741 0 : check_confirmed( ctx, blk, &msg->block_id );
742 0 : }
743 0 : break;
744 0 : }
745 0 : default: return;
746 0 : }
747 0 : }
748 :
749 : static void
750 : after_frag( ctx_t * ctx,
751 : ulong in_idx,
752 : ulong seq FD_PARAM_UNUSED,
753 : ulong sig,
754 : ulong sz,
755 : ulong tsorig,
756 : ulong tspub,
757 0 : fd_stem_context_t * stem ) {
758 0 : if( FD_UNLIKELY( ctx->skip_frag ) ) return;
759 :
760 0 : long now_tick = fd_tickcount();
761 0 : long rx_tick = fd_frag_meta_ts_decomp( tsorig, now_tick );
762 0 : if( FD_UNLIKELY( rx_tick>now_tick ) ) rx_tick -= 1L<<32;
763 :
764 0 : ctx->stem = stem;
765 0 : in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
766 0 : uint in_kind = ctx->in_kind[ in_idx ];
767 :
768 0 : switch( in_kind ) {
769 : /* Unreliable frags */
770 0 : case IN_KIND_NET: {
771 0 : after_net( ctx, sz );
772 0 : break;
773 0 : }
774 0 : case IN_KIND_SIGN: {
775 0 : after_sign( ctx, in_idx, sig, stem );
776 0 : break;
777 0 : }
778 : /* Reliable frags read directly from dcache */
779 0 : case IN_KIND_SNAP: {
780 0 : after_snap( ctx, sig, fd_chunk_to_laddr( ctx->in_links[ in_idx ].mem, ctx->snap_out_chunk ) );
781 0 : break;
782 0 : }
783 0 : case IN_KIND_GOSSIP: {
784 0 : fd_gossip_update_message_t const * msg = (fd_gossip_update_message_t const *)fd_type_pun_const( fd_chunk_to_laddr( in_ctx->mem, ctx->chunk ) );
785 0 : after_gossip( ctx, msg, sig );
786 0 : break;
787 0 : }
788 0 : case IN_KIND_REPLAY: {
789 0 : fd_replay_fec_evicted_t const * msg = (fd_replay_fec_evicted_t const *)fd_type_pun_const( fd_chunk_to_laddr( in_ctx->mem, ctx->chunk ) );
790 0 : fd_forest_fec_clear( ctx->forest, msg->slot, msg->fec_set_idx, FD_FEC_SHRED_CNT - 1 );
791 0 : break;
792 0 : }
793 0 : case IN_KIND_TOWER: {
794 0 : after_tower( ctx, sig, fd_chunk_to_laddr( in_ctx->mem, ctx->chunk ) );
795 0 : break;
796 0 : }
797 0 : case IN_KIND_SHRED: {
798 :
799 : /* There are 3 message types from shred:
800 : 1. resolver evict - incomplete FEC set is evicted by resolver
801 : 2. fec complete - FEC set is completed by resolver. Also contains a shred.
802 : 3. shred - new shred
803 :
804 : Msgs 2 and 3 have a shred header in the dcache. Msg 1 is empty. */
805 :
806 0 : uint sig_src = fd_shred_sig_src( sig );
807 0 : int sig_res = fd_shred_sig_res( sig );
808 :
809 0 : if( FD_UNLIKELY( sig_src==SHRED_SIG_FEC_EVICTED ) ) {
810 0 : fd_fec_evicted_t * evicted = (fd_fec_evicted_t *)fd_type_pun( fd_chunk_to_laddr( in_ctx->mem, ctx->chunk ) );
811 0 : after_evict( ctx, evicted );
812 0 : return;
813 0 : }
814 :
815 0 : uchar * src = fd_chunk_to_laddr( in_ctx->mem, ctx->chunk );
816 0 : fd_shred_base_t * shred_msg = (fd_shred_base_t *)fd_type_pun( src );
817 0 : fd_shred_t * shred = &shred_msg->shred; /* completes & shred messages all have a shred header at the same offset (after merkle root) */
818 :
819 0 : if( FD_UNLIKELY( shred->slot > ctx->metrics->current_slot && sig_src == SHRED_SIG_SRC_TURBINE ) ) {
820 0 : FD_LOG_INFO(( "[Turbine] slot: %lu, root: %lu", shred->slot, fd_forest_root_slot( ctx->forest ) ));
821 0 : ctx->metrics->current_slot = shred->slot;
822 0 : }
823 :
824 0 : if( FD_UNLIKELY( ctx->turbine_slot0 == ULONG_MAX && sig_src == SHRED_SIG_SRC_TURBINE ) ) {
825 0 : ctx->turbine_slot0 = shred->slot;
826 0 : fd_repair_metrics_set_turbine_slot0( ctx->slot_metrics, shred->slot );
827 0 : fd_policy_set_turbine_slot0( ctx->policy, shred->slot );
828 :
829 : /* On first turbine shred, seed repair by queuing highest_shred
830 : requests for slots between snapshot and turbine_slot0. This
831 : bypasses forest entirely and dispatches directly via the sign
832 : queue. Cap at half queue capacity to leave room for pongs. */
833 0 : ulong root = fd_forest_root_slot( ctx->forest );
834 0 : if( FD_LIKELY( root != ULONG_MAX && shred->slot > root ) ) {
835 0 : ulong capacity = fd_signs_queue_max( ctx->pong_queue ) - fd_signs_queue_cnt( ctx->pong_queue );
836 0 : ulong seed_cnt = fd_ulong_min( shred->slot-root, capacity/2 );
837 0 : long now_ms = fd_clock_tile_now( ctx->clock )/(long)1e6;
838 0 : for( ulong i=1; i<=seed_cnt; i++ ) {
839 0 : ulong slot = root + i;
840 0 : fd_pubkey_t const * peer = fd_policy_peer_select( ctx->policy );
841 0 : if( FD_UNLIKELY( !peer ) ) break;
842 0 : fd_repair_msg_t * msg = fd_repair_highest_shred( ctx->protocol, peer, (ulong)now_ms, 0, slot, 0 );
843 0 : if( FD_LIKELY( msg ) ) fd_signs_queue_push( ctx->pong_queue, (sign_pending_t){ .msg = *msg } );
844 0 : }
845 0 : }
846 0 : }
847 :
848 0 : if( FD_UNLIKELY( sig==SHRED_SIG_FEC_COMPLETE || sig==SHRED_SIG_FEC_COMPLETE_LEADER ) ) {
849 0 : FD_TEST( sz==sizeof(fd_fec_complete_t) );
850 0 : fd_fec_complete_t * in_msg = (fd_fec_complete_t *)fd_type_pun( src );
851 0 : int invalid = after_fec( ctx, &in_msg->last_shred_hdr, &in_msg->merkle_root, &in_msg->chained_merkle_root, rx_tick );
852 0 : ulong fwd_sig = invalid ? REPAIR_SIG_FEC_INVALID : (sig==SHRED_SIG_FEC_COMPLETE_LEADER ? REPAIR_SIG_FEC_LEADER : REPAIR_SIG_FEC);
853 :
854 0 : fd_repair_fec_complete_t * complete_msg = fd_chunk_to_laddr( ctx->repair_out_ctx->mem, ctx->repair_out_ctx->chunk );
855 0 : complete_msg->fec = *in_msg;
856 :
857 0 : fd_fec_complete_metrics_t * m = &complete_msg->metrics;
858 0 : m->fec_completed_ts_nanos = fd_clock_epoch_y( ctx->clock->epoch, rx_tick );
859 :
860 0 : m->stats_valid = 0U;
861 0 : fd_forest_blk_t const * blk = fd_forest_query( ctx->forest, in_msg->last_shred_hdr.slot );
862 0 : if( FD_LIKELY( blk ) ) {
863 0 : m->stats_valid = 1U;
864 0 : m->blk_turbine_cnt = blk->turbine_cnt;
865 0 : m->blk_repair_cnt = blk->repair_cnt;
866 0 : m->blk_recovered_cnt = blk->recovered_cnt;
867 0 : ulong word_cnt = fd_forest_blk_idxs_word_cnt( ctx->forest->shred_max );
868 0 : m->blk_data_cnt = (uint)fd_forest_blk_idxs_cnt( fd_forest_blk_idxs( ctx->forest, blk ), word_cnt );
869 0 : m->blk_parity_cnt = (uint)fd_forest_blk_idxs_cnt( fd_forest_blk_code( ctx->forest, blk ), word_cnt );
870 0 : m->blk_lowest_verified_fec = blk->lowest_verified_fec;
871 0 : m->blk_chain_confirmed = blk->chain_confirmed;
872 0 : m->blk_slot_complete = (uchar)( blk->complete_idx!=UINT_MAX );
873 0 : m->blk_req_window_cnt = blk->req_window_cnt;
874 0 : m->blk_req_highest_cnt = blk->req_highest_cnt;
875 0 : m->blk_req_orphan_cnt = blk->req_orphan_cnt;
876 0 : m->blk_req_retransmit_cnt = blk->req_retransmit_cnt;
877 0 : m->blk_repair_responses = blk->response_cnt;
878 0 : m->blk_chain_verify_failed = blk->chain_verify_failed;
879 :
880 : /* 0 = never stamped (e.g. no repair request was ever sent). */
881 0 : m->blk_first_shred_ts_nanos = blk->first_shred_ts ? (ulong)fd_clock_epoch_y( ctx->clock->epoch, blk->first_shred_ts ) : 0UL;
882 0 : m->blk_last_shred_ts_nanos = blk->last_shred_ts ? (ulong)fd_clock_epoch_y( ctx->clock->epoch, blk->last_shred_ts ) : 0UL;
883 0 : m->blk_first_req_ts_nanos = blk->first_req_ts ? (ulong)fd_clock_epoch_y( ctx->clock->epoch, blk->first_req_ts ) : 0UL;
884 0 : m->blk_last_repair_resp_ts_nanos = blk->last_repair_resp_ts ? (ulong)fd_clock_epoch_y( ctx->clock->epoch, blk->last_repair_resp_ts ) : 0UL;
885 0 : }
886 :
887 : /* indiscriminately forward FEC complete messages along to replay */
888 0 : fd_stem_publish( ctx->stem, ctx->repair_out_ctx->idx, fwd_sig, ctx->repair_out_ctx->chunk, sizeof(fd_repair_fec_complete_t), 0UL, 0UL, tspub );
889 0 : ctx->repair_out_ctx->chunk = fd_dcache_compact_next( ctx->repair_out_ctx->chunk, sizeof(fd_repair_fec_complete_t), ctx->repair_out_ctx->chunk0, ctx->repair_out_ctx->wmark );
890 0 : } else if( FD_LIKELY( sig_res!=SHRED_SIG_RESULT_EQVOC ) ) {
891 0 : fd_hash_t * cmr = (fd_hash_t *)fd_type_pun(shred_msg->shred_ + fd_shred_chain_off( shred->variant ));
892 0 : after_shred( ctx, sig, shred, shred_msg->rnonce, &shred_msg->merkle_root, cmr, rx_tick );
893 0 : }
894 0 : return;
895 0 : }
896 0 : default: FD_LOG_ERR(( "bad in_kind %u", in_kind )); /* Should never reach here since before_frag should have filtered out any unexpected frags. */
897 0 : }
898 0 : }
899 :
900 : /* Defer a request by adding it to the outstanding inflights table so it
901 : can be re-requested after a timeout window. Nonce is 0 because these
902 : are not real requests made to the network, and cannot be matched
903 : by a shred response. */
904 : static void
905 0 : defer_inflight_request( ctx_t * ctx, ulong slot, ulong shred_idx, long now ) {
906 0 : fd_hash_t hash = { .ul[0] = 0 };
907 0 : fd_inflight_key_t inflight_req[1];
908 0 : fd_inflight_key_init( inflight_req, FD_REPAIR_KIND_SHRED, slot, shred_idx, 0UL, NULL );
909 0 : if( FD_LIKELY( !fd_inflight_map_ele_query( ctx->inflights->map, inflight_req, NULL, ctx->inflights->pool ) ) ) {
910 0 : fd_inflights_shred_insert( ctx->inflights, FD_REPAIR_KIND_SHRED, 0, &hash, slot, shred_idx, NULL, NULL, now );
911 0 : }
912 0 : }
913 :
914 : /* Should be called for any regular FD_REPAIR_KIND_SHRED request made. */
915 : static void
916 0 : record_inflight_request( ctx_t * ctx, ulong nonce, fd_pubkey_t const * peer, ulong slot, ulong shred_idx, long now ) {
917 0 : fd_inflights_shred_insert( ctx->inflights, FD_REPAIR_KIND_SHRED, nonce, peer, slot, shred_idx, NULL, NULL, now );
918 0 : fd_policy_peer_request_update( ctx->policy, peer );
919 0 : }
920 :
921 : /* Repair request invariants:
922 :
923 : Highest window index requests and orphan requests are much less
924 : fragile than regular shred requests. This is because we request
925 : orphans and highest window index requests whenever we "need" them,
926 : i.e. we fire and forget them.
927 :
928 : But regular shred requests are only iterated once per shred per slot.
929 : This is for performance reasons - we don't want to iterate over the
930 : forest more than necessary. If we let knowledge of a regular shred
931 : request get dropped, it's possible the request could get lost
932 : forever. This is why any regular shred request that is conceived
933 : needs to be handled at the repair tile level. All regular shred
934 : requests must be added to the inflights outstanding table so it can
935 : be re-requested after a timeout window.
936 :
937 : An inflight entry should only ever be permanently removed upon
938 : receipt of a shred response.
939 :
940 : There are two methods through which we make regular shred requests:
941 : 1. policy_next
942 : 2. fd_inflights_pop
943 :
944 : In general, policy_next makes the first request for shred X, and
945 : fd_inflights_pop makes all subsequent requests for shred X at
946 : DEDUP_TIMEOUT intervals. This is to give each request a fair chance
947 : to be received, but also to avoid colliding nonces. This is because
948 : we want to be as accurate as possible when tracking per-peer response
949 : latency.
950 :
951 : With eviction, it's possible for policy_next to make the same request
952 : for shred X multiple times, in short timeout intervals. Therefore we
953 : need to have both policy_next and fd_inflights_pop requests
954 : pass through the same dedup cache. At the point of eviction, we
955 : leave requests for that slot in the dedup cache and in the inflights
956 : table. If an old request from before eviction happens to
957 : dedup a request on the new insertion of the slot, these requests must
958 : be queued up in inflights table so they can be re-requested after a
959 : timeout window. */
960 :
961 : static inline void
962 : after_credit( ctx_t * ctx,
963 : fd_stem_context_t * stem FD_PARAM_UNUSED,
964 : int * opt_poll_in FD_PARAM_UNUSED,
965 0 : int * charge_busy ) {
966 0 : long now = fd_clock_tile_now( ctx->clock );
967 :
968 0 : if( FD_UNLIKELY( ctx->halt_signing ) ) {
969 0 : *charge_busy = 1;
970 0 : return;
971 0 : }
972 :
973 : /* Verify that there is at least one sign tile with available credits.
974 : If not, we can't send any requests and leave early. */
975 0 : out_ctx_t * sign_out = sign_avail_credits( ctx );
976 0 : if( FD_UNLIKELY( !sign_out ) ) {
977 0 : ctx->metrics->sign_tile_unavail++;
978 0 : return;
979 0 : }
980 :
981 : /* If inflights is at capacity, then the only thing we can send is:
982 : pongs, initial highest window index requests, or resend things that
983 : are already inflight. Any new requests that would cause an
984 : inflight to be added to the queue must be deferred. */
985 :
986 0 : if( FD_UNLIKELY( !fd_signs_queue_empty( ctx->pong_queue ) ) ) {
987 0 : sign_pending_t signable = fd_signs_queue_pop( ctx->pong_queue );
988 0 : fd_repair_send_sign_request( ctx, sign_out, &signable.msg, signable.msg.kind == FD_REPAIR_KIND_PONG ? &signable.pong_data : NULL );
989 0 : *charge_busy = 1;
990 0 : return;
991 0 : }
992 :
993 0 : if( FD_UNLIKELY( fd_inflights_should_drain( ctx->inflights, now ) ) ) {
994 0 : fd_inflight_t req[1];
995 0 : fd_inflights_pop( ctx->inflights, req );
996 0 : *charge_busy = 1;
997 0 : ulong nonce = req->key.nonce;
998 0 : ulong slot = req->key.slot;
999 0 : ulong shred_idx = req->key.idx;
1000 :
1001 0 : fd_forest_blk_t * blk = fd_forest_query( ctx->forest, slot );
1002 0 : if( FD_UNLIKELY( blk && shred_idx <= blk->complete_idx && !fd_forest_blk_idxs_test( fd_forest_blk_idxs( ctx->forest, blk ), shred_idx ) ) ) {
1003 0 : fd_pubkey_t const * peer = fd_policy_peer_select( ctx->policy );
1004 :
1005 0 : if( FD_UNLIKELY( !peer || fd_reqlim_next( ctx->dedup, fd_reqlim_key( FD_REPAIR_KIND_SHRED, slot, (uint)shred_idx ), now ) ) ) {
1006 : /* No peers available, park the request in inflights. */
1007 0 : defer_inflight_request( ctx, slot, shred_idx, now );
1008 0 : } else {
1009 0 : ctx->metrics->rerequest++;
1010 0 : blk->req_retransmit_cnt++;
1011 0 : nonce = fd_rnonce_ss_compute( ctx->repair_nonce_ss, 1, slot, (uint)shred_idx, now );
1012 0 : fd_repair_msg_t * msg = fd_repair_shred( ctx->protocol, peer, (ulong)now/(ulong)1e6, (uint)nonce, slot, shred_idx );
1013 0 : fd_repair_send_sign_request( ctx, sign_out, msg, NULL );
1014 0 : record_inflight_request( ctx, nonce, peer, slot, shred_idx, now ); /* Request is definitely a regular shred request. */
1015 0 : return;
1016 0 : }
1017 0 : }
1018 0 : }
1019 :
1020 0 : if( FD_UNLIKELY( fd_inflights_outstanding_free( ctx->inflights ) <= fd_signs_map_key_cnt( ctx->signs_map ) ) ) return; /* no new requests allowed */
1021 :
1022 0 : fd_repair_msg_t const * cout = fd_policy_next( ctx->policy, ctx->dedup, ctx->forest, ctx->protocol, now, ctx->metrics->current_slot, charge_busy );
1023 0 : if( FD_UNLIKELY( !cout ) ) return;
1024 :
1025 0 : if( ( cout->kind == FD_REPAIR_KIND_SHRED && fd_reqlim_next( ctx->dedup, fd_reqlim_key( FD_REPAIR_KIND_SHRED, cout->shred.slot, (uint)cout->shred.shred_idx ), now ) ) ) {
1026 : /* Here if policy_next is re-requesting a shred that's already been
1027 : requested. This could be happen during a race - imagine we make a
1028 : request for shred 0 in slot Y. Then eviction causes slot Y to be
1029 : removed and then readded. policy_next will re-request shred 0,
1030 : but if we let it get dropped here, it's possible the request
1031 : could get lost forever. */
1032 0 : defer_inflight_request( ctx, cout->shred.slot, cout->shred.shred_idx, now );
1033 0 : return;
1034 0 : }
1035 :
1036 : /* finally, send the request made by policy */
1037 0 : fd_repair_send_sign_request( ctx, sign_out, cout, NULL );
1038 0 : if( FD_LIKELY( cout->kind == FD_REPAIR_KIND_SHRED ) ) {
1039 0 : record_inflight_request( ctx, cout->shred.nonce, &cout->shred.to, cout->shred.slot, cout->shred.shred_idx, now );
1040 0 : fd_forest_blk_t * blk = fd_forest_query( ctx->forest, cout->shred.slot );
1041 0 : if( FD_LIKELY( blk ) ) {
1042 0 : blk->req_window_cnt++;
1043 0 : if( FD_UNLIKELY( !blk->first_req_ts ) ) blk->first_req_ts = fd_tickcount();
1044 0 : }
1045 0 : }
1046 0 : }
1047 :
1048 : static void
1049 0 : signs_queue_update_identity( ctx_t * ctx ) {
1050 0 : ulong queue_cnt = fd_signs_queue_cnt( ctx->pong_queue );
1051 0 : for( ulong i=0UL; i<queue_cnt; i++ ) {
1052 0 : sign_pending_t signable = fd_signs_queue_pop( ctx->pong_queue );
1053 0 : switch( signable.msg.kind ) {
1054 0 : case FD_REPAIR_KIND_PONG:
1055 0 : memcpy( signable.msg.pong.from.uc, ctx->identity_public_key.uc, sizeof(fd_pubkey_t) );
1056 0 : break;
1057 0 : case FD_REPAIR_KIND_SHRED:
1058 0 : memcpy( signable.msg.shred.from.uc, ctx->identity_public_key.uc, sizeof(fd_pubkey_t) );
1059 0 : break;
1060 0 : case FD_REPAIR_KIND_HIGHEST_SHRED:
1061 0 : memcpy( signable.msg.highest_shred.from.uc, ctx->identity_public_key.uc, sizeof(fd_pubkey_t) );
1062 0 : break;
1063 0 : case FD_REPAIR_KIND_ORPHAN:
1064 0 : memcpy( signable.msg.orphan.from.uc, ctx->identity_public_key.uc, sizeof(fd_pubkey_t) );
1065 0 : break;
1066 0 : default:
1067 0 : FD_LOG_CRIT(( "Unhandled repair kind %u", signable.msg.kind ));
1068 0 : break;
1069 0 : }
1070 0 : fd_signs_queue_push( ctx->pong_queue, signable );
1071 0 : }
1072 0 : }
1073 :
1074 : static inline void
1075 0 : during_housekeeping( ctx_t * ctx ) {
1076 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) fd_clock_tile_recal( ctx->clock );
1077 :
1078 : # if DEBUG_LOGGING
1079 : long now = fd_log_wallclock();
1080 : if( FD_UNLIKELY( now - ctx->tsdebug > (long)10e9 ) ) {
1081 : fd_forest_print( ctx->forest );
1082 : ctx->tsdebug = fd_log_wallclock();
1083 : }
1084 : # endif
1085 :
1086 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_UNHALT_PENDING ) ) {
1087 0 : FD_LOG_DEBUG(( "keyswitch: unhalting" ));
1088 0 : FD_CHECK_CRIT( ctx->halt_signing, "state machine corruption" );
1089 0 : ctx->halt_signing = 0;
1090 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
1091 :
1092 : /* Peers key their ping-pong cache by our pubkey, so every peer is
1093 : cold for the new identity. Re-warm them all. */
1094 0 : fd_policy_peers_t * peers = &ctx->policy->peers;
1095 0 : fd_policy_peer_dlist_t * lists[ 2 ] = { peers->fast, peers->slow };
1096 0 : for( ulong l=0UL; l<2UL; l++ ) {
1097 0 : for( fd_policy_peer_dlist_iter_t iter = fd_policy_peer_dlist_iter_fwd_init( lists[ l ], peers->pool );
1098 0 : !fd_policy_peer_dlist_iter_done( iter, lists[ l ], peers->pool ) && !fd_signs_queue_full( ctx->pong_queue );
1099 0 : iter = fd_policy_peer_dlist_iter_fwd_next( iter, lists[ l ], peers->pool ) ) {
1100 0 : fd_policy_peer_t const * peer = fd_policy_peer_dlist_iter_ele_const( iter, lists[ l ], peers->pool );
1101 0 : fd_repair_msg_t * init = fd_repair_shred( ctx->protocol, &peer->key, (ulong)fd_clock_tile_now( ctx->clock )/1000000UL, 0, 0, 0 );
1102 0 : fd_signs_queue_push( ctx->pong_queue, (sign_pending_t){ .msg = *init } );
1103 0 : }
1104 0 : }
1105 0 : }
1106 :
1107 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
1108 :
1109 0 : if( !ctx->halt_signing ) {
1110 : /* At this point, stop sending new sign requests to the sign tile
1111 : and wait for all outstanding sign requests to be received back
1112 : from the sign tile. We also need to update any pending
1113 : outgoing sign requests with the new identity key. */
1114 0 : FD_LOG_DEBUG(( "keyswitch: halting signing" ));
1115 0 : ctx->halt_signing = 1;
1116 0 : memcpy( ctx->identity_public_key.uc, ctx->keyswitch->bytes, 32UL );
1117 0 : ctx->protocol->identity_key = ctx->identity_public_key;
1118 0 : signs_queue_update_identity( ctx );
1119 0 : }
1120 :
1121 0 : if( fd_signs_map_key_cnt( ctx->signs_map )==0UL ) {
1122 : /* Once there are no more in flight sign requests, we are ready to
1123 : say that the keyswitch is completed. */
1124 0 : FD_LOG_DEBUG(( "keyswitch: completed, no more outstanding stale sign requests" ));
1125 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
1126 0 : }
1127 0 : }
1128 0 : }
1129 :
1130 : static void
1131 : privileged_init( fd_topo_t const * topo,
1132 0 : fd_topo_tile_t const * tile ) {
1133 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1134 :
1135 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1136 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
1137 0 : fd_memset( ctx, 0, sizeof(ctx_t) );
1138 :
1139 0 : uchar const * identity_key = fd_keyload_load( tile->repair.identity_key_path, /* pubkey only: */ 1 );
1140 0 : fd_memcpy( ctx->identity_public_key.uc, identity_key, sizeof(fd_pubkey_t) );
1141 :
1142 0 : FD_TEST( fd_rng_secure( &ctx->repair_seed, sizeof(ulong) ) );
1143 :
1144 0 : ulong rnonce_ss_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "rnonce_ss" );
1145 0 : FD_TEST( rnonce_ss_id!=ULONG_MAX );
1146 0 : memcpy( ctx->repair_nonce_ss, fd_topo_obj_laddr( topo, rnonce_ss_id ), sizeof(fd_rnonce_ss_t) );
1147 0 : }
1148 :
1149 : static void
1150 : unprivileged_init( fd_topo_t const * topo,
1151 0 : fd_topo_tile_t const * tile ) {
1152 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1153 :
1154 0 : ulong total_sign_depth = tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt;
1155 0 : int lg_sign_depth = fd_ulong_find_msb( fd_ulong_pow2_up(total_sign_depth) ) + 1;
1156 :
1157 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1158 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
1159 0 : ctx->protocol = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_align(), fd_repair_footprint() );
1160 0 : ctx->forest = FD_SCRATCH_ALLOC_APPEND( l, fd_forest_align(), fd_forest_footprint( tile->repair.slot_max, tile->repair.max_shreds_per_block ) );
1161 0 : ctx->policy = FD_SCRATCH_ALLOC_APPEND( l, fd_policy_align(), fd_policy_footprint( FD_REPAIR_PEER_MAX ) );
1162 0 : ctx->dedup = FD_SCRATCH_ALLOC_APPEND( l, fd_reqlim_align(), fd_reqlim_footprint( FD_REQLIM_CACHE_MAX ) );
1163 0 : ctx->inflights = FD_SCRATCH_ALLOC_APPEND( l, fd_inflights_align(), fd_inflights_footprint() );
1164 0 : ctx->signs_map = FD_SCRATCH_ALLOC_APPEND( l, fd_signs_map_align(), fd_signs_map_footprint( lg_sign_depth ) );
1165 0 : ctx->pong_queue = FD_SCRATCH_ALLOC_APPEND( l, fd_signs_queue_align(), fd_signs_queue_footprint() );
1166 0 : ctx->slot_metrics = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_metrics_align(), fd_repair_metrics_footprint() );
1167 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
1168 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
1169 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
1170 :
1171 0 : ctx->protocol = fd_repair_join ( fd_repair_new ( ctx->protocol, &ctx->identity_public_key ) );
1172 0 : ctx->forest = fd_forest_join ( fd_forest_new ( ctx->forest, tile->repair.slot_max, tile->repair.max_shreds_per_block, ctx->repair_seed ) );
1173 0 : ctx->policy = fd_policy_join ( fd_policy_new ( ctx->policy, FD_REPAIR_PEER_MAX, ctx->repair_seed, ctx->repair_nonce_ss ) );
1174 0 : ctx->dedup = fd_reqlim_join ( fd_reqlim_new ( ctx->dedup, FD_REQLIM_CACHE_MAX, ctx->repair_seed ) );
1175 0 : ctx->inflights = fd_inflights_join ( fd_inflights_new ( ctx->inflights, ctx->repair_seed+1234UL ) );
1176 0 : ctx->signs_map = fd_signs_map_join ( fd_signs_map_new ( ctx->signs_map, lg_sign_depth, 0UL ) );
1177 0 : ctx->pong_queue = fd_signs_queue_join ( fd_signs_queue_new ( ctx->pong_queue ) );
1178 0 : ctx->slot_metrics = fd_repair_metrics_join( fd_repair_metrics_new( ctx->slot_metrics ) );
1179 :
1180 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
1181 0 : FD_TEST( ctx->keyswitch );
1182 :
1183 0 : ctx->halt_signing = 0;
1184 :
1185 : /* Process in links */
1186 :
1187 0 : if( FD_UNLIKELY( tile->in_cnt > MAX_IN_LINKS ) ) FD_LOG_ERR(( "repair tile has too many input links" ));
1188 :
1189 0 : uint sign_repair_in_idx[ MAX_SIGN_TILE_CNT ] = {0};
1190 0 : uint sign_repair_idx = 0;
1191 0 : ulong sign_link_depth = 0;
1192 :
1193 0 : for( uint in_idx=0U; in_idx<(tile->in_cnt); in_idx++ ) {
1194 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ in_idx ] ];
1195 0 : if( 0==strcmp( link->name, "net_repair" ) ) {
1196 0 : ctx->in_kind[ in_idx ] = IN_KIND_NET;
1197 0 : fd_net_rx_bounds_init( &ctx->in_links[ in_idx ].net_rx, link->dcache );
1198 0 : continue;
1199 0 : } else if( 0==strcmp( link->name, "sign_repair" ) ) {
1200 0 : ctx->in_kind[ in_idx ] = IN_KIND_SIGN;
1201 0 : sign_repair_in_idx[ sign_repair_idx++ ] = in_idx;
1202 0 : sign_link_depth = link->depth;
1203 0 : }
1204 0 : else if( 0==strcmp( link->name, "gossip_ciaddr" ) ) ctx->in_kind[ in_idx ] = IN_KIND_GOSSIP;
1205 0 : else if( 0==strcmp( link->name, "tower_out" ) ) ctx->in_kind[ in_idx ] = IN_KIND_TOWER;
1206 0 : else if( 0==strcmp( link->name, "shred_out" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SHRED;
1207 0 : else if( 0==strcmp( link->name, "snapin_manif" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SNAP;
1208 0 : else if( 0==strcmp( link->name, "replay_slot" ) ) ctx->in_kind[ in_idx ] = IN_KIND_REPLAY;
1209 0 : else FD_LOG_ERR(( "repair tile has unexpected input link %s", link->name ));
1210 :
1211 0 : ctx->in_links[ in_idx ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1212 0 : ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
1213 0 : ctx->in_links[ in_idx ].wmark = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
1214 0 : ctx->in_links[ in_idx ].mtu = link->mtu;
1215 :
1216 0 : FD_TEST( fd_dcache_compact_is_safe( ctx->in_links[in_idx].mem, link->dcache, link->mtu, link->depth ) );
1217 0 : }
1218 :
1219 0 : ctx->net_out_ctx->idx = UINT_MAX;
1220 0 : ctx->repair_out_ctx->idx = UINT_MAX;
1221 0 : ctx->repair_sign_cnt = 0;
1222 0 : ctx->sign_rrobin_idx = 0;
1223 :
1224 0 : for( uint out_idx=0U; out_idx<(tile->out_cnt); out_idx++ ) {
1225 0 : fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ out_idx ] ];
1226 :
1227 0 : if( 0==strcmp( link->name, "repair_net" ) ) {
1228 :
1229 0 : if( ctx->net_out_ctx->idx!=UINT_MAX ) continue; /* only use first net link */
1230 0 : ctx->net_out_ctx->idx = out_idx;
1231 0 : ctx->net_out_ctx->mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1232 0 : ctx->net_out_ctx->chunk0 = fd_dcache_compact_chunk0( ctx->net_out_ctx->mem, link->dcache );
1233 0 : ctx->net_out_ctx->wmark = fd_dcache_compact_wmark( ctx->net_out_ctx->mem, link->dcache, link->mtu );
1234 0 : ctx->net_out_ctx->chunk = ctx->net_out_ctx->chunk0;
1235 :
1236 0 : } else if( 0==strcmp( link->name, "repair_out" ) ) {
1237 :
1238 0 : out_ctx_t * replay_out = ctx->repair_out_ctx;
1239 0 : replay_out->idx = out_idx;
1240 0 : replay_out->mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1241 0 : replay_out->chunk0 = fd_dcache_compact_chunk0( replay_out->mem, link->dcache );
1242 0 : replay_out->wmark = fd_dcache_compact_wmark( replay_out->mem, link->dcache, link->mtu );
1243 0 : replay_out->chunk = replay_out->chunk0;
1244 :
1245 0 : } else if( 0==strcmp( link->name, "repair_sign" ) ) {
1246 :
1247 0 : out_ctx_t * repair_sign_out = &ctx->repair_sign_out_ctx[ ctx->repair_sign_cnt ];
1248 0 : repair_sign_out->idx = out_idx;
1249 0 : repair_sign_out->mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1250 0 : repair_sign_out->chunk0 = fd_dcache_compact_chunk0( repair_sign_out->mem, link->dcache );
1251 0 : repair_sign_out->wmark = fd_dcache_compact_wmark( repair_sign_out->mem, link->dcache, link->mtu );
1252 0 : repair_sign_out->chunk = repair_sign_out->chunk0;
1253 0 : repair_sign_out->in_idx = sign_repair_in_idx[ ctx->repair_sign_cnt++ ]; /* match to the sign_repair input link */
1254 0 : repair_sign_out->max_credits = sign_link_depth;
1255 0 : repair_sign_out->credits = sign_link_depth;
1256 :
1257 0 : } else {
1258 0 : FD_LOG_ERR(( "repair tile has unexpected output link %s", link->name ));
1259 0 : }
1260 0 : }
1261 0 : FD_TEST( ctx->net_out_ctx->idx!=UINT_MAX );
1262 0 : FD_TEST( ctx->repair_out_ctx->idx!=UINT_MAX );
1263 0 : if( FD_UNLIKELY( ctx->repair_sign_cnt!=sign_repair_idx ) ) {
1264 0 : FD_LOG_ERR(( "Mismatch between repair_sign output links (%lu) and sign_repair input links (%u)", ctx->repair_sign_cnt, sign_repair_idx ));
1265 0 : }
1266 0 : if( FD_UNLIKELY( fd_signs_map_key_max( ctx->signs_map ) < tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt ) ) {
1267 0 : FD_LOG_ERR(( "Repair pending signs tracking map is too small: %lu < %lu.", fd_signs_map_key_max( ctx->signs_map ), tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt ));
1268 0 : }
1269 :
1270 0 : ctx->wksp = topo->workspaces[ topo->objs[ tile->tile_obj_id ].wksp_id ].wksp;
1271 0 : ctx->repair_intake_addr.port = fd_ushort_bswap( tile->repair.repair_client_listen_port );
1272 :
1273 : /* TODO clean these up */
1274 0 : ctx->net_id = (ushort)0;
1275 0 : fd_ip4_udp_hdr_init( ctx->intake_hdr, 0, 0, tile->repair.repair_client_listen_port );
1276 :
1277 : /* Repair set up */
1278 :
1279 0 : ctx->turbine_slot0 = ULONG_MAX;
1280 0 : FD_LOG_INFO(( "repair my addr - intake addr: " FD_IP4_ADDR_FMT ":%u",
1281 0 : FD_IP4_ADDR_FMT_ARGS( ctx->repair_intake_addr.addr ), fd_ushort_bswap( ctx->repair_intake_addr.port )
1282 0 : ));
1283 :
1284 0 : memset( ctx->metrics, 0, sizeof(ctx->metrics) );
1285 :
1286 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->slot_compl_time, FD_MHIST_EDGES_NS( REPAIR, SLOT_COMPLETE_DURATION_NANOS ) ) );
1287 0 : fd_histf_join( fd_histf_new( ctx->metrics->response_latency, FD_MHIST_MIN( REPAIR, RESPONSE_LATENCY_NANOS ),
1288 0 : FD_MHIST_MAX( REPAIR, RESPONSE_LATENCY_NANOS ) ) );
1289 :
1290 0 : fd_clock_tile_init( ctx->clock );
1291 :
1292 0 : ctx->tsdebug = fd_log_wallclock();
1293 0 : ctx->pending_key_next = 0;
1294 0 : }
1295 :
1296 : static ulong
1297 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
1298 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1299 : ulong out_cnt,
1300 0 : struct sock_filter * out ) {
1301 0 : populate_sock_filter_policy_fd_repair_tile( out_cnt, out, (uint)fd_log_private_logfile_fd() );
1302 0 : return sock_filter_policy_fd_repair_tile_instr_cnt;
1303 0 : }
1304 :
1305 : static ulong
1306 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
1307 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
1308 : ulong out_fds_cnt,
1309 0 : int * out_fds ) {
1310 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
1311 :
1312 0 : ulong out_cnt = 0UL;
1313 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
1314 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
1315 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
1316 0 : return out_cnt;
1317 0 : }
1318 :
1319 : static inline void
1320 0 : metrics_write( ctx_t * ctx ) {
1321 0 : FD_MGAUGE_SET( REPAIR, HIGHEST_TURBINE_SLOT, ctx->metrics->current_slot );
1322 0 : FD_MGAUGE_SET( REPAIR, HIGHEST_REPAIRED_SLOT, fd_forest_highest_repaired_slot( ctx->forest ) );
1323 0 : FD_MCNT_SET( REPAIR, SHRED_OLD_IGNORED, ctx->metrics->old_shred );
1324 0 : FD_MGAUGE_SET( REPAIR, PEER_COUNT, fd_policy_peer_pool_used( ctx->policy->peers.pool ) );
1325 0 : FD_MCNT_SET( REPAIR, SIGN_BACKPRESSURE, ctx->metrics->sign_tile_unavail );
1326 0 : FD_MCNT_SET( REPAIR, SHRED_REREQUESTED, ctx->metrics->rerequest );
1327 :
1328 0 : FD_MGAUGE_SET( REPAIR, LAST_REQUESTED_SLOT, ctx->metrics->last_requested_slot );
1329 0 : FD_MGAUGE_SET( REPAIR, LAST_REQUESTED_ORPHAN_SLOT, ctx->metrics->last_requested_orphan );
1330 0 : FD_MGAUGE_SET( REPAIR, REQUEST_IN_FLIGHT, fd_inflights_outstanding_cnt( ctx->inflights ) );
1331 :
1332 0 : FD_MCNT_SET ( REPAIR, PKT_TX, ctx->metrics->send_pkt_cnt );
1333 0 : FD_MCNT_ENUM_COPY( REPAIR, REQUEST_TX, ctx->metrics->sent_pkt_types );
1334 :
1335 0 : FD_MHIST_COPY( REPAIR, SLOT_COMPLETE_DURATION_NANOS, ctx->metrics->slot_compl_time );
1336 0 : FD_MHIST_COPY( REPAIR, RESPONSE_LATENCY_NANOS, ctx->metrics->response_latency );
1337 :
1338 0 : FD_MCNT_SET ( REPAIR, BLOCK_EVICTED, ctx->metrics->blk_evicted );
1339 0 : FD_MCNT_SET ( REPAIR, BLOCK_INSERT_FAILED, ctx->metrics->blk_failed_insert );
1340 0 : FD_MGAUGE_SET( REPAIR, LAST_EVICTED_SLOT, ctx->metrics->slot_evicted );
1341 0 : FD_MGAUGE_SET( REPAIR, LAST_EVICTION_CAUSE_SLOT, ctx->metrics->slot_evicted_by );
1342 0 : FD_MGAUGE_SET( REPAIR, LAST_INSERT_FAILED_SLOT, ctx->metrics->slot_failed_insert );
1343 :
1344 0 : FD_MCNT_SET ( REPAIR, CHAIN_VERIFY_FAILED, ctx->metrics->failed_chain_verify_cnt );
1345 0 : FD_MGAUGE_SET( REPAIR, LAST_CHAIN_VERIFY_FAILED_SLOT, ctx->metrics->failed_chain_verify_slot );
1346 :
1347 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_UNKNOWN_PEER, ctx->metrics->unknown_peer_ping );
1348 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_MALFORMED, ctx->metrics->malformed_ping );
1349 0 : FD_MCNT_SET( REPAIR, PING_REJECTED_SIGNATURE_FAILED, ctx->metrics->fail_sigverify_ping );
1350 0 : }
1351 :
1352 : #undef DEBUG_LOGGING
1353 :
1354 : /* At most one sign request is made in after_credit. Then at most one
1355 : message is published in after_frag. */
1356 0 : #define STEM_BURST (2UL)
1357 :
1358 : /* Set LAZY to a reasonable value that keeps housekeeping time low.
1359 : Repair tile's only reliable consumer is replay. */
1360 0 : #define STEM_LAZY (64000)
1361 :
1362 0 : #define STEM_CALLBACK_CONTEXT_TYPE ctx_t
1363 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(ctx_t)
1364 :
1365 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
1366 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
1367 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
1368 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
1369 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
1370 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1371 :
1372 : #include "../../disco/stem/fd_stem.c"
1373 :
1374 : fd_topo_run_tile_t fd_tile_repair = {
1375 : .name = "repair",
1376 : .populate_allowed_seccomp = populate_allowed_seccomp,
1377 : .populate_allowed_fds = populate_allowed_fds,
1378 : .scratch_align = scratch_align,
1379 : .scratch_footprint = scratch_footprint,
1380 : .unprivileged_init = unprivileged_init,
1381 : .privileged_init = privileged_init,
1382 : .run = stem_run,
1383 : };
|
