Line data Source code
1 : /* Repair tile runs the repair protocol for a Firedancer node. */
2 : #include "fd_fec_chainer.h"
3 : #define _GNU_SOURCE
4 :
5 : #include "../../disco/topo/fd_topo.h"
6 : #include "generated/fd_repair_tile_seccomp.h"
7 :
8 : #include "../../flamenco/repair/fd_repair.h"
9 : #include "../../flamenco/runtime/fd_blockstore.h"
10 : #include "../../flamenco/leaders/fd_leaders_base.h"
11 : #include "../../disco/fd_disco.h"
12 : #include "../../disco/keyguard/fd_keyload.h"
13 : #include "../../disco/keyguard/fd_keyguard_client.h"
14 : #include "../../disco/keyguard/fd_keyguard.h"
15 : #include "../../disco/net/fd_net_tile.h"
16 : #include "../../discof/replay/fd_exec.h"
17 : #include "../../util/pod/fd_pod_format.h"
18 : #include "../../util/net/fd_net_headers.h"
19 :
20 : #include "../forest/fd_forest.h"
21 : #include "fd_fec_repair.h"
22 : #include "fd_fec_chainer.h"
23 :
24 : #include <errno.h>
25 :
26 0 : #define IN_KIND_NET (0)
27 0 : #define IN_KIND_CONTACT (1)
28 0 : #define IN_KIND_STAKE (2)
29 0 : #define IN_KIND_SHRED (3)
30 0 : #define IN_KIND_SIGN (4)
31 : #define MAX_IN_LINKS (16)
32 :
33 : #define NET_OUT_IDX (0)
34 : #define SIGN_OUT_IDX (1)
35 0 : #define REPLAY_OUT_IDX (2)
36 : #define ARCHIVE_OUT_IDX (3)
37 :
38 : #define MAX_REPAIR_PEERS 40200UL
39 : #define MAX_BUFFER_SIZE ( MAX_REPAIR_PEERS * sizeof(fd_shred_dest_wire_t))
40 : #define MAX_SHRED_TILE_CNT (16UL)
41 :
42 : typedef union {
43 : struct {
44 : fd_wksp_t * mem;
45 : ulong chunk0;
46 : ulong wmark;
47 : ulong mtu;
48 : };
49 : fd_net_rx_bounds_t net_rx;
50 : } fd_repair_in_ctx_t;
51 :
52 : struct fd_repair_out_ctx {
53 : ulong idx;
54 : fd_wksp_t * mem;
55 : ulong chunk0;
56 : ulong wmark;
57 : ulong chunk;
58 : };
59 : typedef struct fd_repair_out_ctx fd_repair_out_ctx_t;
60 :
61 : struct fd_fec_sig {
62 : ulong key; /* map key. 32 msb = slot, 32 lsb = fec_set_idx */
63 : fd_ed25519_sig_t sig; /* Ed25519 sig identifier of the FEC. */
64 : };
65 : typedef struct fd_fec_sig fd_fec_sig_t;
66 :
67 : #define MAP_NAME fd_fec_sig
68 0 : #define MAP_T fd_fec_sig_t
69 : #define MAP_MEMOIZE 0
70 : #include "../../util/tmpl/fd_map_dynamic.c"
71 :
72 : struct fd_reasm {
73 : ulong slot;
74 : uint cnt;
75 : };
76 : typedef struct fd_reasm fd_reasm_t;
77 :
78 : #define MAP_NAME fd_reasm
79 0 : #define MAP_T fd_reasm_t
80 0 : #define MAP_KEY slot
81 : #define MAP_MEMOIZE 0
82 : #include "../../util/tmpl/fd_map_dynamic.c"
83 :
84 : struct fd_repair_tile_ctx {
85 : long tsprint; /* timestamp for printing */
86 : long tsrepair; /* timestamp for repair */
87 : long tsreset; /* timestamp for resetting iterator */
88 : ulong * wmark;
89 : ulong prev_wmark;
90 :
91 : fd_repair_t * repair;
92 : fd_repair_config_t repair_config;
93 :
94 : ulong repair_seed;
95 :
96 : fd_repair_peer_addr_t repair_intake_addr;
97 : fd_repair_peer_addr_t repair_serve_addr;
98 :
99 : ushort repair_intake_listen_port;
100 : ushort repair_serve_listen_port;
101 :
102 : fd_forest_t * forest;
103 : fd_fec_sig_t * fec_sigs;
104 : fd_reasm_t * reasm;
105 : fd_fec_chainer_t * fec_chainer;
106 : fd_forest_iter_t repair_iter;
107 :
108 : ulong * turbine_slot0;
109 : ulong * turbine_slot;
110 :
111 : uchar identity_private_key[ 32 ];
112 : fd_pubkey_t identity_public_key;
113 :
114 : fd_wksp_t * wksp;
115 :
116 : uchar in_kind[ MAX_IN_LINKS ];
117 : fd_repair_in_ctx_t in_links[ MAX_IN_LINKS ];
118 :
119 : int skip_frag;
120 :
121 : uint net_out_idx;
122 : fd_wksp_t * net_out_mem;
123 : ulong net_out_chunk0;
124 : ulong net_out_wmark;
125 : ulong net_out_chunk;
126 :
127 : fd_wksp_t * replay_out_mem;
128 : ulong replay_out_chunk0;
129 : ulong replay_out_wmark;
130 : ulong replay_out_chunk;
131 :
132 : uint shred_tile_cnt;
133 : fd_repair_out_ctx_t shred_out_ctx[ MAX_SHRED_TILE_CNT ];
134 :
135 : ushort net_id;
136 : /* Includes Ethernet, IP, UDP headers */
137 : uchar buffer[ MAX_BUFFER_SIZE ];
138 : fd_ip4_udp_hdrs_t intake_hdr[1];
139 : fd_ip4_udp_hdrs_t serve_hdr [1];
140 :
141 : fd_stem_context_t * stem;
142 :
143 : fd_wksp_t * blockstore_wksp;
144 : fd_blockstore_t blockstore_ljoin;
145 : fd_blockstore_t * blockstore;
146 :
147 : fd_keyguard_client_t keyguard_client[1];
148 : };
149 : typedef struct fd_repair_tile_ctx fd_repair_tile_ctx_t;
150 :
151 : FD_FN_CONST static inline ulong
152 0 : scratch_align( void ) {
153 0 : return 128UL;
154 0 : }
155 :
156 : FD_FN_PURE static inline ulong
157 0 : loose_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
158 0 : return 1UL * FD_SHMEM_GIGANTIC_PAGE_SZ;
159 0 : }
160 :
161 : FD_FN_PURE static inline ulong
162 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
163 :
164 0 : ulong l = FD_LAYOUT_INIT;
165 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_repair_tile_ctx_t), sizeof(fd_repair_tile_ctx_t) );
166 0 : l = FD_LAYOUT_APPEND( l, fd_repair_align(), fd_repair_footprint() );
167 0 : l = FD_LAYOUT_APPEND( l, fd_forest_align(), fd_forest_footprint( tile->repair.slot_max ) );
168 0 : l = FD_LAYOUT_APPEND( l, fd_fec_sig_align(), fd_fec_sig_footprint( 20 ) );
169 0 : l = FD_LAYOUT_APPEND( l, fd_reasm_align(), fd_reasm_footprint( 20 ) );
170 : // l = FD_LAYOUT_APPEND( l, fd_fec_repair_align(), fd_fec_repair_footprint( ( 1<<20 ), tile->repair.shred_tile_cnt ) );
171 0 : l = FD_LAYOUT_APPEND( l, fd_fec_chainer_align(), fd_fec_chainer_footprint( 1 << 20 ) ); // TODO: fix this
172 0 : l = FD_LAYOUT_APPEND( l, fd_scratch_smem_align(), fd_scratch_smem_footprint( FD_REPAIR_SCRATCH_MAX ) );
173 0 : l = FD_LAYOUT_APPEND( l, fd_scratch_fmem_align(), fd_scratch_fmem_footprint( FD_REPAIR_SCRATCH_DEPTH ) );
174 0 : return FD_LAYOUT_FINI( l, scratch_align() );
175 0 : }
176 :
177 : static void
178 : repair_signer( void * signer_ctx,
179 : uchar signature[ static 64 ],
180 : uchar const * buffer,
181 : ulong len,
182 0 : int sign_type ) {
183 0 : fd_repair_tile_ctx_t * ctx = (fd_repair_tile_ctx_t *) signer_ctx;
184 0 : fd_keyguard_client_sign( ctx->keyguard_client, signature, buffer, len, sign_type );
185 0 : }
186 :
187 : static void
188 : send_packet( fd_repair_tile_ctx_t * ctx,
189 : fd_stem_context_t * stem,
190 : int is_intake,
191 : uint dst_ip_addr,
192 : ushort dst_port,
193 : uint src_ip_addr,
194 : uchar const * payload,
195 : ulong payload_sz,
196 0 : ulong tsorig ) {
197 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
198 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
199 0 : *hdr = *(is_intake ? ctx->intake_hdr : ctx->serve_hdr);
200 :
201 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
202 0 : ip4->saddr = src_ip_addr;
203 0 : ip4->daddr = dst_ip_addr;
204 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
205 0 : ip4->check = 0U;
206 0 : ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
207 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
208 :
209 0 : fd_udp_hdr_t * udp = hdr->udp;
210 0 : udp->net_dport = dst_port;
211 0 : udp->net_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
212 0 : fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
213 0 : hdr->udp->check = 0U;
214 :
215 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
216 0 : ulong sig = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
217 0 : ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
218 0 : ulong chunk = ctx->net_out_chunk;
219 0 : fd_stem_publish( stem, ctx->net_out_idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
220 0 : ctx->net_out_chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
221 0 : }
222 :
223 : static inline void
224 : handle_new_cluster_contact_info( fd_repair_tile_ctx_t * ctx,
225 : uchar const * buf,
226 0 : ulong buf_sz ) {
227 0 : fd_shred_dest_wire_t const * in_dests = (fd_shred_dest_wire_t const *)fd_type_pun_const( buf );
228 :
229 0 : ulong dest_cnt = buf_sz;
230 0 : if( FD_UNLIKELY( dest_cnt >= MAX_REPAIR_PEERS ) ) {
231 0 : FD_LOG_WARNING(( "Cluster nodes had %lu destinations, which was more than the max of %lu", dest_cnt, MAX_REPAIR_PEERS ));
232 0 : return;
233 0 : }
234 :
235 : /* Stop adding peers after we reach the peer max, but we may want to
236 : consider an eviction policy. */
237 0 : for( ulong i=0UL; i<dest_cnt; i++ ) {
238 0 : if( FD_UNLIKELY( ctx->repair->peer_cnt >= FD_ACTIVE_KEY_MAX ) ) break;// FIXME: aiming to move all peer tracking out of lib into tile, leaving like this for now
239 0 : fd_repair_peer_addr_t repair_peer = {
240 0 : .addr = in_dests[i].ip4_addr,
241 0 : .port = fd_ushort_bswap( in_dests[i].udp_port ),
242 0 : };
243 0 : int dup = fd_repair_add_active_peer( ctx->repair, &repair_peer, in_dests[i].pubkey );
244 0 : if( !dup ) {
245 0 : ulong hash_src = 0xfffffUL & fd_ulong_hash( (ulong)in_dests[i].ip4_addr | ((ulong)repair_peer.port<<32) );
246 0 : FD_LOG_INFO(( "Added repair peer: pubkey %s hash_src %lu", FD_BASE58_ENC_32_ALLOCA(in_dests[i].pubkey), hash_src ));
247 0 : }
248 0 : }
249 0 : }
250 :
251 : ulong
252 : fd_repair_handle_ping( fd_repair_tile_ctx_t * repair_tile_ctx,
253 : fd_repair_t * glob,
254 : fd_gossip_ping_t const * ping,
255 : fd_gossip_peer_addr_t const * peer_addr FD_PARAM_UNUSED,
256 : uint self_ip4_addr FD_PARAM_UNUSED,
257 : uchar * msg_buf,
258 0 : ulong msg_buf_sz ) {
259 0 : fd_repair_protocol_t protocol;
260 0 : fd_repair_protocol_new_disc(&protocol, fd_repair_protocol_enum_pong);
261 0 : fd_gossip_ping_t * pong = &protocol.inner.pong;
262 :
263 0 : pong->from = *glob->public_key;
264 :
265 : /* Generate response hash token */
266 0 : uchar pre_image[FD_PING_PRE_IMAGE_SZ];
267 0 : memcpy( pre_image, "SOLANA_PING_PONG", 16UL );
268 0 : memcpy( pre_image+16UL, ping->token.uc, 32UL);
269 :
270 : /* Generate response hash token */
271 0 : fd_sha256_hash( pre_image, FD_PING_PRE_IMAGE_SZ, &pong->token );
272 :
273 : /* Sign it */
274 0 : repair_signer( repair_tile_ctx, pong->signature.uc, pre_image, FD_PING_PRE_IMAGE_SZ, FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519 );
275 :
276 0 : fd_bincode_encode_ctx_t ctx;
277 0 : ctx.data = msg_buf;
278 0 : ctx.dataend = msg_buf + msg_buf_sz;
279 0 : FD_TEST(0 == fd_repair_protocol_encode(&protocol, &ctx));
280 0 : ulong buflen = (ulong)((uchar*)ctx.data - msg_buf);
281 0 : return buflen;
282 0 : }
283 :
284 : /* Pass a raw client response packet into the protocol. addr is the address of the sender */
285 : static int
286 : fd_repair_recv_clnt_packet( fd_repair_tile_ctx_t * repair_tile_ctx,
287 : fd_stem_context_t * stem,
288 : fd_repair_t * glob,
289 : uchar const * msg,
290 : ulong msglen,
291 : fd_repair_peer_addr_t const * src_addr,
292 0 : uint dst_ip4_addr ) {
293 0 : glob->metrics.recv_clnt_pkt++;
294 :
295 0 : FD_SCRATCH_SCOPE_BEGIN {
296 0 : while( 1 ) {
297 0 : ulong decoded_sz;
298 0 : fd_repair_response_t * gmsg = fd_bincode_decode1_scratch(
299 0 : repair_response, msg, msglen, NULL, &decoded_sz );
300 0 : if( FD_UNLIKELY( !gmsg ) ) {
301 : /* Solana falls back to assuming we got a shred in this case
302 : https://github.com/solana-labs/solana/blob/master/core/src/repair/serve_repair.rs#L1198 */
303 0 : break;
304 0 : }
305 0 : if( FD_UNLIKELY( decoded_sz != msglen ) ) {
306 0 : break;
307 0 : }
308 :
309 0 : switch( gmsg->discriminant ) {
310 0 : case fd_repair_response_enum_ping:
311 0 : {
312 0 : uchar buf[1024];
313 0 : ulong buflen = fd_repair_handle_ping( repair_tile_ctx, glob, &gmsg->inner.ping, src_addr, dst_ip4_addr, buf, sizeof(buf) );
314 0 : ulong tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
315 0 : send_packet( repair_tile_ctx, stem, 1, src_addr->addr, src_addr->port, dst_ip4_addr, buf, buflen, tsorig );
316 0 : break;
317 0 : }
318 0 : }
319 :
320 0 : return 0;
321 0 : }
322 0 : } FD_SCRATCH_SCOPE_END;
323 0 : return 0;
324 0 : }
325 :
326 : static ulong
327 : fd_repair_sign_and_send( fd_repair_tile_ctx_t * repair_tile_ctx,
328 : fd_repair_protocol_t * protocol,
329 : fd_gossip_peer_addr_t * addr FD_PARAM_UNUSED,
330 : uchar * buf,
331 0 : ulong buflen ) {
332 :
333 0 : FD_TEST( buflen >= 1024UL );
334 0 : fd_bincode_encode_ctx_t ctx = { .data = buf, .dataend = buf + buflen };
335 0 : if( FD_UNLIKELY( fd_repair_protocol_encode( protocol, &ctx ) != FD_BINCODE_SUCCESS ) ) {
336 0 : FD_LOG_CRIT(( "Failed to encode repair message (type %#x)", protocol->discriminant ));
337 0 : }
338 :
339 0 : buflen = (ulong)ctx.data - (ulong)buf;
340 0 : if( FD_UNLIKELY( buflen<68 ) ) {
341 0 : FD_LOG_CRIT(( "Attempted to sign unsigned repair message type (type %#x)", protocol->discriminant ));
342 0 : }
343 :
344 : /* At this point buffer contains
345 :
346 : [ discriminant ] [ signature ] [ payload ]
347 : ^ ^ ^
348 : 0 4 68 */
349 :
350 : /* https://github.com/solana-labs/solana/blob/master/core/src/repair/serve_repair.rs#L1258 */
351 :
352 0 : fd_memcpy( buf+64, buf, 4 );
353 0 : buf += 64UL;
354 0 : buflen -= 64UL;
355 :
356 : /* Now it contains
357 :
358 : [ discriminant ] [ payload ]
359 : ^ ^
360 : buf buf+4 */
361 :
362 0 : fd_signature_t sig;
363 0 : repair_signer( repair_tile_ctx, sig.uc, buf, buflen, FD_KEYGUARD_SIGN_TYPE_ED25519 );
364 :
365 : /* Reintroduce the signature */
366 :
367 0 : buf -= 64UL;
368 0 : buflen += 64UL;
369 0 : fd_memcpy( buf + 4U, &sig, 64U );
370 :
371 0 : return buflen;
372 0 : }
373 :
374 :
375 : static void
376 : fd_repair_send_request( fd_repair_tile_ctx_t * repair_tile_ctx,
377 : fd_stem_context_t * stem,
378 : fd_repair_t * glob,
379 : enum fd_needed_elem_type type,
380 : ulong slot,
381 : uint shred_index,
382 : fd_pubkey_t const * recipient,
383 0 : long now ) {
384 :
385 : /* Send requests starting where we left off last time. i.e. if n < current_nonce, seek forward */
386 : /* Track statistics */
387 0 : fd_repair_protocol_t protocol;
388 0 : fd_repair_construct_request_protocol( glob, &protocol, type, slot, shred_index, recipient, glob->next_nonce, now );
389 0 : glob->next_nonce++;
390 0 : fd_active_elem_t * active = fd_active_table_query( glob->actives, recipient, NULL );
391 :
392 0 : active->avg_reqs++;
393 0 : glob->metrics.send_pkt_cnt++;
394 :
395 0 : uchar buf[1024];
396 0 : ulong buflen = fd_repair_sign_and_send( repair_tile_ctx, &protocol, &active->addr, buf, sizeof(buf) );
397 0 : ulong tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
398 0 : uint src_ip4_addr = 0U; /* unknown */
399 0 : send_packet( repair_tile_ctx, stem, 1, active->addr.addr, active->addr.port, src_ip4_addr, buf, buflen, tsorig );
400 0 : }
401 :
402 : static void
403 : fd_repair_send_requests( fd_repair_tile_ctx_t * ctx,
404 : fd_stem_context_t * stem,
405 : enum fd_needed_elem_type type,
406 : ulong slot,
407 : uint shred_index,
408 0 : long now ){
409 0 : fd_repair_t * glob = ctx->repair;
410 :
411 0 : for( uint i=0; i<FD_REPAIR_NUM_NEEDED_PEERS; i++ ) {
412 0 : fd_pubkey_t const * id = &glob->peers[ glob->peer_idx++ ].key;
413 0 : fd_repair_send_request( ctx, stem, glob, type, slot, shred_index, id, now );
414 0 : if( FD_UNLIKELY( glob->peer_idx >= glob->peer_cnt ) ) glob->peer_idx = 0; /* wrap around */
415 0 : }
416 0 : }
417 :
418 :
419 : static inline int
420 : before_frag( fd_repair_tile_ctx_t * ctx,
421 : ulong in_idx,
422 : ulong seq FD_PARAM_UNUSED,
423 0 : ulong sig ) {
424 0 : uint in_kind = ctx->in_kind[ in_idx ];
425 0 : if( FD_LIKELY( in_kind==IN_KIND_NET ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
426 0 : return 0;
427 0 : }
428 :
429 : static int
430 0 : is_fec_completes_msg( ulong sz ) {
431 0 : return sz == FD_SHRED_DATA_HEADER_SZ + FD_SHRED_MERKLE_ROOT_SZ;
432 0 : }
433 :
434 : static void
435 : during_frag( fd_repair_tile_ctx_t * ctx,
436 : ulong in_idx,
437 : ulong seq FD_PARAM_UNUSED,
438 : ulong sig FD_PARAM_UNUSED,
439 : ulong chunk,
440 : ulong sz,
441 0 : ulong ctl ) {
442 0 : ctx->skip_frag = 0;
443 :
444 0 : uchar const * dcache_entry;
445 0 : ulong dcache_entry_sz;
446 :
447 : // TODO: check for sz>MTU for failure once MTUs are decided
448 0 : uint in_kind = ctx->in_kind[ in_idx ];
449 0 : fd_repair_in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
450 0 : if( FD_LIKELY( in_kind==IN_KIND_NET ) ) {
451 0 : dcache_entry = fd_net_rx_translate_frag( &in_ctx->net_rx, chunk, ctl, sz );
452 0 : dcache_entry_sz = sz;
453 :
454 0 : } else if( FD_UNLIKELY( in_kind==IN_KIND_CONTACT ) ) {
455 0 : if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) ) {
456 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
457 0 : }
458 0 : dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
459 0 : dcache_entry_sz = sz * sizeof(fd_shred_dest_wire_t);
460 :
461 0 : } else if( FD_UNLIKELY( in_kind==IN_KIND_STAKE ) ) {
462 0 : if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) ) {
463 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
464 0 : }
465 0 : dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
466 0 : fd_stake_weight_msg_t const * msg = fd_type_pun_const( dcache_entry );
467 0 : fd_repair_set_stake_weights_init( ctx->repair, msg->weights, msg->staked_cnt );
468 0 : return;
469 :
470 0 : } else if( FD_LIKELY( in_kind==IN_KIND_SHRED ) ) {
471 0 : if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) ) {
472 0 : FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
473 0 : }
474 0 : dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
475 0 : dcache_entry_sz = sz;
476 :
477 0 : } else {
478 0 : FD_LOG_ERR(( "Frag from unknown link (kind=%u in_idx=%lu)", in_kind, in_idx ));
479 0 : }
480 :
481 0 : fd_memcpy( ctx->buffer, dcache_entry, dcache_entry_sz );
482 0 : }
483 :
484 : static ulong
485 : fd_repair_send_ping( fd_repair_tile_ctx_t * repair_tile_ctx,
486 : fd_repair_t * glob,
487 : fd_pinged_elem_t * val,
488 : uchar * buf,
489 0 : ulong buflen ) {
490 0 : fd_repair_response_t gmsg;
491 0 : fd_repair_response_new_disc( &gmsg, fd_repair_response_enum_ping );
492 0 : fd_gossip_ping_t * ping = &gmsg.inner.ping;
493 0 : ping->from = *glob->public_key;
494 :
495 0 : uchar pre_image[FD_PING_PRE_IMAGE_SZ];
496 0 : memcpy( pre_image, "SOLANA_PING_PONG", 16UL );
497 0 : memcpy( pre_image+16UL, val->token.uc, 32UL );
498 :
499 0 : fd_sha256_hash( pre_image, FD_PING_PRE_IMAGE_SZ, &ping->token );
500 :
501 0 : repair_signer( repair_tile_ctx, ping->signature.uc, pre_image, FD_PING_PRE_IMAGE_SZ, FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519 );
502 :
503 0 : fd_bincode_encode_ctx_t ctx;
504 0 : FD_TEST( buflen >= 1024UL );
505 0 : ctx.data = buf;
506 0 : ctx.dataend = buf + buflen;
507 0 : FD_TEST(0 == fd_repair_response_encode(&gmsg, &ctx));
508 0 : return (ulong)((uchar*)ctx.data - buf);
509 0 : }
510 :
511 : static void
512 0 : fd_repair_recv_pong(fd_repair_t * glob, fd_gossip_ping_t const * pong, fd_gossip_peer_addr_t const * from) {
513 0 : fd_pinged_elem_t * val = fd_pinged_table_query(glob->pinged, from, NULL);
514 0 : if( val == NULL || !fd_pubkey_eq( &val->id, &pong->from ) )
515 0 : return;
516 :
517 : /* Verify response hash token */
518 0 : uchar pre_image[FD_PING_PRE_IMAGE_SZ];
519 0 : memcpy( pre_image, "SOLANA_PING_PONG", 16UL );
520 0 : memcpy( pre_image+16UL, val->token.uc, 32UL );
521 :
522 0 : fd_hash_t pre_image_hash;
523 0 : fd_sha256_hash( pre_image, FD_PING_PRE_IMAGE_SZ, pre_image_hash.uc );
524 :
525 0 : fd_sha256_t sha[1];
526 0 : fd_sha256_init( sha );
527 0 : fd_sha256_append( sha, "SOLANA_PING_PONG", 16UL );
528 0 : fd_sha256_append( sha, pre_image_hash.uc, 32UL );
529 0 : fd_hash_t golden;
530 0 : fd_sha256_fini( sha, golden.uc );
531 :
532 0 : fd_sha512_t sha2[1];
533 0 : if( fd_ed25519_verify( /* msg */ golden.uc,
534 0 : /* sz */ 32U,
535 0 : /* sig */ pong->signature.uc,
536 0 : /* public_key */ pong->from.uc,
537 0 : sha2 )) {
538 0 : FD_LOG_WARNING(("Failed sig verify for pong"));
539 0 : return;
540 0 : }
541 :
542 0 : val->good = 1;
543 0 : }
544 :
545 :
546 : static long
547 : repair_get_shred( ulong slot,
548 : uint shred_idx,
549 : void * buf,
550 : ulong buf_max,
551 0 : void * arg ) {
552 0 : fd_repair_tile_ctx_t * ctx = (fd_repair_tile_ctx_t *)arg;
553 0 : fd_blockstore_t * blockstore = ctx->blockstore;
554 0 : if( FD_UNLIKELY( blockstore == NULL ) ) {
555 0 : return -1;
556 0 : }
557 :
558 0 : if( shred_idx == UINT_MAX ) {
559 0 : int err = FD_MAP_ERR_AGAIN;
560 0 : while( err == FD_MAP_ERR_AGAIN ) {
561 0 : fd_block_map_query_t query[1] = { 0 };
562 0 : err = fd_block_map_query_try( blockstore->block_map, &slot, NULL, query, 0 );
563 0 : fd_block_info_t * meta = fd_block_map_query_ele( query );
564 0 : if( FD_UNLIKELY( err == FD_MAP_ERR_KEY ) ) return -1L;
565 0 : if( FD_UNLIKELY( err == FD_MAP_ERR_AGAIN ) ) continue;
566 0 : shred_idx = (uint)meta->slot_complete_idx;
567 0 : err = fd_block_map_query_test( query );
568 0 : }
569 0 : }
570 0 : long sz = fd_buf_shred_query_copy_data( blockstore, slot, shred_idx, buf, buf_max );
571 0 : return sz;
572 0 : }
573 :
574 : static ulong
575 : repair_get_parent( ulong slot,
576 0 : void * arg ) {
577 0 : fd_repair_tile_ctx_t * ctx = (fd_repair_tile_ctx_t *)arg;
578 0 : fd_blockstore_t * blockstore = ctx->blockstore;
579 0 : if( FD_UNLIKELY( blockstore == NULL ) ) {
580 0 : return FD_SLOT_NULL;
581 0 : }
582 0 : return fd_blockstore_parent_slot_query( blockstore, slot );
583 0 : }
584 :
585 :
586 : /* Pass a raw service request packet into the protocol.
587 : src_addr is the address of the sender
588 : dst_ip4_addr is the dst IPv4 address of the incoming packet (i.e. our IP) */
589 :
590 : static int
591 : fd_repair_recv_serv_packet( fd_repair_tile_ctx_t * repair_tile_ctx,
592 : fd_stem_context_t * stem,
593 : fd_repair_t * glob,
594 : uchar * msg,
595 : ulong msglen,
596 : fd_repair_peer_addr_t const * peer_addr,
597 0 : uint self_ip4_addr ) {
598 : //ulong recv_serv_packet;
599 : //ulong recv_serv_pkt_types[FD_METRICS_ENUM_SENT_REQUEST_TYPES_CNT];
600 :
601 0 : FD_SCRATCH_SCOPE_BEGIN {
602 0 : ulong decoded_sz;
603 0 : fd_repair_protocol_t * protocol = fd_bincode_decode1_scratch(
604 0 : repair_protocol, msg, msglen, NULL, &decoded_sz );
605 0 : if( FD_UNLIKELY( !protocol ) ) {
606 0 : glob->metrics.recv_serv_corrupt_pkt++;
607 0 : FD_LOG_WARNING(( "Failed to decode repair request packet" ));
608 0 : return 0;
609 0 : }
610 :
611 0 : glob->metrics.recv_serv_pkt++;
612 :
613 0 : if( FD_UNLIKELY( decoded_sz != msglen ) ) {
614 0 : FD_LOG_WARNING(( "failed to decode repair request packet" ));
615 0 : return 0;
616 0 : }
617 :
618 0 : fd_repair_request_header_t * header;
619 0 : switch( protocol->discriminant ) {
620 0 : case fd_repair_protocol_enum_pong:
621 0 : glob->metrics.recv_serv_pkt_types[FD_METRICS_ENUM_REPAIR_SERV_PKT_TYPES_V_PONG_IDX]++;
622 0 : fd_repair_recv_pong( glob, &protocol->inner.pong, peer_addr );
623 :
624 0 : return 0;
625 0 : case fd_repair_protocol_enum_window_index: {
626 0 : glob->metrics.recv_serv_pkt_types[FD_METRICS_ENUM_REPAIR_SERV_PKT_TYPES_V_WINDOW_IDX]++;
627 0 : fd_repair_window_index_t * wi = &protocol->inner.window_index;
628 0 : header = &wi->header;
629 0 : break;
630 0 : }
631 0 : case fd_repair_protocol_enum_highest_window_index: {
632 0 : glob->metrics.recv_serv_pkt_types[FD_METRICS_ENUM_REPAIR_SERV_PKT_TYPES_V_HIGHEST_WINDOW_IDX]++;
633 0 : fd_repair_highest_window_index_t * wi = &protocol->inner.highest_window_index;
634 0 : header = &wi->header;
635 0 : break;
636 0 : }
637 0 : case fd_repair_protocol_enum_orphan: {
638 0 : glob->metrics.recv_serv_pkt_types[FD_METRICS_ENUM_REPAIR_SERV_PKT_TYPES_V_ORPHAN_IDX]++;
639 0 : fd_repair_orphan_t * wi = &protocol->inner.orphan;
640 0 : header = &wi->header;
641 0 : break;
642 0 : }
643 0 : default: {
644 0 : glob->metrics.recv_serv_pkt_types[FD_METRICS_ENUM_REPAIR_SERV_PKT_TYPES_V_UNKNOWN_IDX]++;
645 0 : FD_LOG_WARNING(( "received repair request of unknown type: %d", (int)protocol->discriminant ));
646 0 : return 0;
647 0 : }
648 0 : }
649 :
650 0 : if( FD_UNLIKELY( !fd_pubkey_eq( &header->recipient, glob->public_key ) ) ) {
651 0 : FD_LOG_WARNING(( "received repair request with wrong recipient, %s instead of %s", FD_BASE58_ENC_32_ALLOCA( header->recipient.uc ), FD_BASE58_ENC_32_ALLOCA( glob->public_key ) ));
652 0 : return 0;
653 0 : }
654 :
655 : /* Verify the signature */
656 0 : fd_sha512_t sha2[1];
657 0 : fd_signature_t sig;
658 0 : fd_memcpy( &sig, header->signature.uc, sizeof(sig) );
659 0 : fd_memcpy( (uchar *)msg + 64U, msg, 4U );
660 0 : if( fd_ed25519_verify( /* msg */ msg + 64U,
661 0 : /* sz */ msglen - 64U,
662 0 : /* sig */ sig.uc,
663 0 : /* public_key */ header->sender.uc,
664 0 : sha2 )) {
665 0 : glob->metrics.recv_serv_invalid_signature++;
666 0 : FD_LOG_WARNING(( "received repair request with with invalid signature" ));
667 0 : return 0;
668 0 : }
669 :
670 0 : fd_pinged_elem_t * val = fd_pinged_table_query( glob->pinged, peer_addr, NULL) ;
671 0 : if( val == NULL || !val->good || !fd_pubkey_eq( &val->id, &header->sender ) ) {
672 : /* Need to ping this client */
673 0 : if( val == NULL ) {
674 0 : if( fd_pinged_table_is_full( glob->pinged ) ) {
675 0 : FD_LOG_WARNING(( "pinged table is full" ));
676 :
677 0 : glob->metrics.recv_serv_full_ping_table++;
678 0 : return 0;
679 0 : }
680 0 : val = fd_pinged_table_insert( glob->pinged, peer_addr );
681 0 : for ( ulong i = 0; i < FD_HASH_FOOTPRINT / sizeof(ulong); i++ )
682 0 : val->token.ul[i] = fd_rng_ulong(glob->rng);
683 0 : }
684 0 : val->id = header->sender;
685 0 : val->good = 0;
686 0 : uchar buf[1024];
687 0 : ulong buflen = fd_repair_send_ping( repair_tile_ctx, glob, val, buf, sizeof(buf) );
688 0 : send_packet( repair_tile_ctx, stem, 0, peer_addr->addr, peer_addr->port, self_ip4_addr, buf, buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
689 0 : } else {
690 0 : uchar buf[FD_SHRED_MAX_SZ + sizeof(uint)];
691 0 : switch( protocol->discriminant ) {
692 0 : case fd_repair_protocol_enum_window_index: {
693 0 : fd_repair_window_index_t const * wi = &protocol->inner.window_index;
694 0 : long sz = repair_get_shred( wi->slot, (uint)wi->shred_index, buf, FD_SHRED_MAX_SZ, repair_tile_ctx );
695 0 : if( sz < 0 ) break;
696 0 : *(uint *)(buf + sz) = wi->header.nonce;
697 0 : send_packet( repair_tile_ctx, stem, 0, peer_addr->addr, peer_addr->port, self_ip4_addr, buf, (ulong)sz + sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
698 0 : break;
699 0 : }
700 :
701 0 : case fd_repair_protocol_enum_highest_window_index: {
702 0 : fd_repair_highest_window_index_t const * wi = &protocol->inner.highest_window_index;
703 0 : long sz = repair_get_shred( wi->slot, UINT_MAX, buf, FD_SHRED_MAX_SZ, repair_tile_ctx );
704 0 : if( sz < 0 ) break;
705 0 : *(uint *)(buf + sz) = wi->header.nonce;
706 0 : send_packet( repair_tile_ctx, stem, 0, peer_addr->addr, peer_addr->port, self_ip4_addr, buf, (ulong)sz + sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
707 0 : break;
708 0 : }
709 :
710 0 : case fd_repair_protocol_enum_orphan: {
711 0 : fd_repair_orphan_t const * wi = &protocol->inner.orphan;
712 0 : ulong slot = wi->slot;
713 0 : for(unsigned i = 0; i < 10; ++i) {
714 0 : slot = repair_get_parent( slot, repair_tile_ctx );
715 : /* We cannot serve slots <= 1 since they are empy and created at genesis. */
716 0 : if( slot == FD_SLOT_NULL || slot <= 1UL ) break;
717 0 : long sz = repair_get_shred( slot, UINT_MAX, buf, FD_SHRED_MAX_SZ, repair_tile_ctx );
718 0 : if( sz < 0 ) continue;
719 0 : *(uint *)(buf + sz) = wi->header.nonce;
720 0 : send_packet( repair_tile_ctx, stem, 0, peer_addr->addr, peer_addr->port, self_ip4_addr, buf, (ulong)sz + sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
721 0 : }
722 0 : break;
723 0 : }
724 :
725 0 : default:
726 0 : break;
727 0 : }
728 0 : }
729 :
730 :
731 0 : } FD_SCRATCH_SCOPE_END;
732 0 : return 0;
733 0 : }
734 :
735 : static void
736 : after_frag( fd_repair_tile_ctx_t * ctx,
737 : ulong in_idx,
738 : ulong seq FD_PARAM_UNUSED,
739 : ulong sig FD_PARAM_UNUSED,
740 : ulong sz,
741 : ulong tsorig,
742 : ulong tspub FD_PARAM_UNUSED,
743 0 : fd_stem_context_t * stem ) {
744 :
745 0 : if( FD_UNLIKELY( ctx->skip_frag ) ) return;
746 :
747 0 : ctx->stem = stem;
748 :
749 0 : uint in_kind = ctx->in_kind[ in_idx ];
750 0 : if( FD_UNLIKELY( in_kind==IN_KIND_CONTACT ) ) {
751 0 : handle_new_cluster_contact_info( ctx, ctx->buffer, sz );
752 0 : return;
753 0 : }
754 :
755 0 : if( FD_UNLIKELY( in_kind==IN_KIND_STAKE ) ) {
756 0 : fd_repair_set_stake_weights_fini( ctx->repair );
757 0 : return;
758 0 : }
759 :
760 0 : if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) {
761 :
762 : /* Initialize the forest, which requires the root to be ready. This
763 : must be the case if we have received a frag from shred, because
764 : shred requires stake weights, which implies a genesis or snapshot
765 : slot has been loaded. */
766 :
767 0 : ulong wmark = fd_fseq_query( ctx->wmark );
768 0 : if( FD_UNLIKELY( fd_forest_root_slot( ctx->forest ) == ULONG_MAX ) ) {
769 0 : fd_forest_init( ctx->forest, wmark );
770 0 : uchar mr[ FD_SHRED_MERKLE_ROOT_SZ ] = { 0 }; /* FIXME */
771 0 : fd_fec_chainer_init( ctx->fec_chainer, wmark, mr );
772 0 : FD_TEST( fd_forest_root_slot( ctx->forest ) != ULONG_MAX );
773 0 : FD_LOG_NOTICE(( "Forest initialized with root %lu", wmark ));
774 0 : ctx->prev_wmark = wmark;
775 0 : }
776 0 : if( FD_UNLIKELY( ctx->prev_wmark < wmark ) ) {
777 0 : fd_forest_publish( ctx->forest, wmark );
778 0 : ctx->prev_wmark = wmark;
779 : // invalidate our repair iterator
780 0 : ctx->repair_iter = fd_forest_iter_init( ctx->forest );
781 0 : }
782 :
783 0 : fd_shred_t * shred = (fd_shred_t *)fd_type_pun( ctx->buffer );
784 0 : if( FD_UNLIKELY( shred->slot <= fd_forest_root_slot( ctx->forest ) ) ) {
785 0 : FD_LOG_WARNING(( "shred %lu %u %u too old, ignoring", shred->slot, shred->idx, shred->fec_set_idx ));
786 0 : return;
787 0 : };
788 :
789 : /* Update turbine_slot0 and turbine_slot. */
790 :
791 0 : if( FD_UNLIKELY( fd_fseq_query( ctx->turbine_slot0 )==ULONG_MAX ) ) {
792 0 : fd_fseq_update( ctx->turbine_slot0, shred->slot );
793 0 : FD_LOG_NOTICE(("First turbine slot %lu", shred->slot));
794 0 : }
795 0 : fd_fseq_update( ctx->turbine_slot, fd_ulong_max( shred->slot, fd_fseq_query( ctx->turbine_slot ) ) );
796 0 : if( FD_UNLIKELY( shred->slot <= fd_forest_root_slot( ctx->forest ) ) ) return; /* shred too old */
797 :
798 :
799 : /* Insert the shred sig (shared by all shred members in the FEC set)
800 : into the map. */
801 :
802 0 : fd_fec_sig_t * fec_sig = fd_fec_sig_query( ctx->fec_sigs, (shred->slot << 32) | shred->fec_set_idx, NULL );
803 0 : if( FD_UNLIKELY( !fec_sig ) ) {
804 0 : fec_sig = fd_fec_sig_insert( ctx->fec_sigs, (shred->slot << 32) | shred->fec_set_idx );
805 0 : memcpy( fec_sig->sig, shred->signature, sizeof(fd_ed25519_sig_t) );
806 0 : }
807 :
808 : /* When this is a FEC completes msg, it is implied that all the
809 : other shreds in the FEC set can also be inserted. Shred inserts
810 : into the forest are idempotent so it is fine to insert the same
811 : shred multiple times. */
812 :
813 0 : if( FD_UNLIKELY( is_fec_completes_msg( sz ) ) ) {
814 0 : fd_forest_ele_t * ele = NULL;
815 0 : for( uint idx = shred->fec_set_idx; idx <= shred->idx; idx++ ) {
816 0 : ele = fd_forest_data_shred_insert( ctx->forest, shred->slot, shred->data.parent_off, idx, shred->fec_set_idx, 0, 0 );
817 0 : }
818 0 : FD_TEST( ele ); /* must be non-empty */
819 0 : fd_forest_ele_idxs_insert( ele->cmpl, shred->fec_set_idx );
820 :
821 0 : uchar * merkle = ctx->buffer + FD_SHRED_DATA_HEADER_SZ;
822 0 : int data_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_DATA_COMPLETE);
823 0 : int slot_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE);
824 :
825 0 : FD_TEST( fd_fec_pool_free( ctx->fec_chainer->pool ) );
826 0 : FD_TEST( !fd_fec_chainer_query( ctx->fec_chainer, shred->slot, shred->fec_set_idx ) );
827 0 : FD_TEST( fd_fec_chainer_insert( ctx->fec_chainer, shred->slot, shred->fec_set_idx, (ushort)(shred->idx - shred->fec_set_idx + 1), data_complete, slot_complete, shred->data.parent_off, merkle, merkle /* FIXME */ ) );
828 :
829 0 : while( FD_LIKELY( !fd_fec_out_empty( ctx->fec_chainer->out ) ) ) {
830 0 : fd_fec_out_t out = fd_fec_out_pop_head( ctx->fec_chainer->out );
831 0 : if( FD_UNLIKELY( out.err != FD_FEC_CHAINER_SUCCESS ) ) FD_LOG_ERR(( "fec chainer err %d", out.err ));
832 0 : fd_reasm_t * reasm = fd_reasm_query( ctx->reasm, out.slot, NULL );
833 0 : if( FD_UNLIKELY( !reasm ) ) {
834 0 : reasm = fd_reasm_insert( ctx->reasm, out.slot );
835 0 : reasm->cnt = 0;
836 0 : }
837 0 : if( FD_UNLIKELY( out.data_complete ) ) {
838 0 : uint cnt = out.fec_set_idx + out.data_cnt - reasm->cnt;
839 0 : ulong sig = fd_disco_repair_replay_sig( out.slot, out.parent_off, cnt, out.slot_complete );
840 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
841 0 : reasm->cnt = out.fec_set_idx + out.data_cnt;
842 0 : fd_stem_publish( ctx->stem, REPLAY_OUT_IDX, sig, 0, 0, 0, tsorig, tspub );
843 0 : if( FD_UNLIKELY( out.slot_complete ) ) {
844 0 : fd_reasm_remove( ctx->reasm, reasm );
845 0 : }
846 0 : }
847 0 : }
848 0 : }
849 :
850 : /* Insert the shred into the map. */
851 :
852 0 : int is_code = fd_shred_is_code( fd_shred_type( shred->variant ) );
853 : // FD_LOG_NOTICE(( "shred %lu %u %u %d", shred->slot, shred->idx, shred->fec_set_idx, is_code ));
854 0 : if( FD_LIKELY( !is_code ) ) {
855 0 : fd_repair_inflight_remove( ctx->repair, shred->slot, shred->idx );
856 :
857 0 : int data_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_DATA_COMPLETE);
858 0 : int slot_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE);
859 0 : fd_forest_ele_t * ele = fd_forest_data_shred_insert( ctx->forest, shred->slot, shred->data.parent_off, shred->idx, shred->fec_set_idx, data_complete, slot_complete );
860 :
861 : /* Check if there are FECs to force complete. Algorithm: window
862 : through the idxs in interval [i, j). If j = next fec_set_idx
863 : then we know we can force complete the FEC set interval [i, j)
864 : (assuming it wasn't already completed based on `cmpl`). */
865 :
866 0 : uint i = 0;
867 0 : for( uint j = 1; j < ele->buffered_idx + 1; j++ ) { /* TODO iterate by word */
868 0 : if( FD_UNLIKELY( fd_forest_ele_idxs_test( ele->cmpl, i ) && fd_forest_ele_idxs_test( ele->fecs, j ) ) ) {
869 0 : i = j;
870 0 : } else if( FD_UNLIKELY( fd_forest_ele_idxs_test( ele->fecs, j ) || j == ele->complete_idx ) ) {
871 0 : if ( j == ele->complete_idx ) j++;
872 0 : fd_forest_ele_idxs_insert( ele->cmpl, i );
873 :
874 : /* Find the shred tile owning this FEC set. */
875 :
876 0 : fd_fec_sig_t * fec_sig = fd_fec_sig_query( ctx->fec_sigs, (shred->slot << 32) | i, NULL );
877 :
878 0 : ulong sig = fd_ulong_load_8( fec_sig->sig );
879 0 : ulong tile_idx = sig % ctx->shred_tile_cnt;
880 0 : uint last_idx = j - i - 1;
881 :
882 0 : uchar * chunk = fd_chunk_to_laddr( ctx->shred_out_ctx[tile_idx].mem, ctx->shred_out_ctx[tile_idx].chunk );
883 0 : memcpy( chunk, fec_sig->sig, sizeof(fd_ed25519_sig_t) );
884 0 : fd_stem_publish( stem, ctx->shred_out_ctx[tile_idx].idx, last_idx, ctx->shred_out_ctx[tile_idx].chunk, sizeof(fd_ed25519_sig_t), 0UL, 0UL, 0UL );
885 0 : ctx->shred_out_ctx[tile_idx].chunk = fd_dcache_compact_next( ctx->shred_out_ctx[tile_idx].chunk, sizeof(fd_ed25519_sig_t), ctx->shred_out_ctx[tile_idx].chunk0, ctx->shred_out_ctx[tile_idx].wmark );
886 0 : i = j;
887 0 : } else {
888 : // FD_LOG_NOTICE(( "not a fec boundary %lu %u", ele->slot, j ));
889 0 : }
890 0 : }
891 0 : }
892 0 : return;
893 0 : }
894 :
895 0 : fd_eth_hdr_t const * eth = (fd_eth_hdr_t const *)ctx->buffer;
896 0 : fd_ip4_hdr_t const * ip4 = (fd_ip4_hdr_t const *)( (ulong)eth + sizeof(fd_eth_hdr_t) );
897 0 : fd_udp_hdr_t const * udp = (fd_udp_hdr_t const *)( (ulong)ip4 + FD_IP4_GET_LEN( *ip4 ) );
898 0 : uchar * data = (uchar *)( (ulong)udp + sizeof(fd_udp_hdr_t) );
899 0 : if( FD_UNLIKELY( (ulong)udp+sizeof(fd_udp_hdr_t) > (ulong)eth+sz ) ) return;
900 0 : ulong udp_sz = fd_ushort_bswap( udp->net_len );
901 0 : if( FD_UNLIKELY( udp_sz<sizeof(fd_udp_hdr_t) ) ) return;
902 0 : ulong data_sz = udp_sz-sizeof(fd_udp_hdr_t);
903 0 : if( FD_UNLIKELY( (ulong)data+data_sz > (ulong)eth+sz ) ) return;
904 :
905 0 : fd_gossip_peer_addr_t peer_addr = { .addr=ip4->saddr, .port=udp->net_sport };
906 0 : ushort dport = udp->net_dport;
907 0 : if( ctx->repair_intake_addr.port == dport ) {
908 0 : fd_repair_recv_clnt_packet( ctx, stem, ctx->repair, data, data_sz, &peer_addr, ip4->daddr );
909 0 : } else if( ctx->repair_serve_addr.port == dport ) {
910 0 : fd_repair_recv_serv_packet( ctx, stem, ctx->repair, data, data_sz, &peer_addr, ip4->daddr );
911 0 : } else {
912 0 : FD_LOG_WARNING(( "Unexpectedly received packet for port %u", (uint)fd_ushort_bswap( dport ) ));
913 0 : }
914 0 : }
915 :
916 0 : #define MAX_REQ_PER_CREDIT 1
917 :
918 : static inline void
919 : after_credit( fd_repair_tile_ctx_t * ctx,
920 : fd_stem_context_t * stem FD_PARAM_UNUSED,
921 : int * opt_poll_in FD_PARAM_UNUSED,
922 0 : int * charge_busy ) {
923 : /* TODO: Don't charge the tile as busy if after_credit isn't actually
924 : doing any work. */
925 0 : *charge_busy = 1;
926 :
927 0 : if( FD_UNLIKELY( ctx->forest->root == ULONG_MAX ) ) return;
928 0 : if( FD_UNLIKELY( ctx->repair->peer_cnt == 0 ) ) return; /* no peers to send requests to */
929 :
930 0 : long now = fd_log_wallclock();
931 :
932 : #if MAX_REQ_PER_CREDIT > FD_REPAIR_NUM_NEEDED_PEERS
933 : /* If the requests are > 1 per credit then we need to starve
934 : after_credit for after_frag to get the chance to be called. We could
935 : get rid of this all together considering max requests per credit is
936 : 1 currently, but it could be useful for benchmarking purposes in the
937 : future. */
938 : if( FD_UNLIKELY( now - ctx->tsrepair < (long)20e6 ) ) {
939 : return;
940 : }
941 : ctx->tsrepair = now;
942 : #endif
943 :
944 0 : fd_forest_t * forest = ctx->forest;
945 0 : fd_forest_ele_t * pool = fd_forest_pool( forest );
946 0 : fd_forest_orphaned_t * orphaned = fd_forest_orphaned( forest );
947 :
948 : // Always request orphans
949 :
950 0 : int total_req = 0;
951 0 : for( fd_forest_orphaned_iter_t iter = fd_forest_orphaned_iter_init( orphaned, pool );
952 0 : !fd_forest_orphaned_iter_done( iter, orphaned, pool );
953 0 : iter = fd_forest_orphaned_iter_next( iter, orphaned, pool ) ) {
954 0 : fd_forest_ele_t * orphan = fd_forest_orphaned_iter_ele( iter, orphaned, pool );
955 0 : if( fd_repair_need_orphan( ctx->repair, orphan->slot ) ) {
956 0 : fd_repair_send_requests( ctx, stem, fd_needed_orphan, orphan->slot, UINT_MAX, now );
957 0 : total_req += FD_REPAIR_NUM_NEEDED_PEERS;
958 0 : }
959 0 : }
960 :
961 0 : if( FD_UNLIKELY( total_req >= MAX_REQ_PER_CREDIT ) ) {
962 0 : fd_repair_continue( ctx->repair );
963 0 : return; /* we have already sent enough requests */
964 0 : }
965 :
966 : // Travel down frontier
967 :
968 : /* Every so often we'll need to reset the frontier iterator to the
969 : head of frontier, because we could end up traversing down a very
970 : long tree if we are far behind. */
971 :
972 0 : if( FD_UNLIKELY( now - ctx->tsreset > (long)40e6 ) ) {
973 : // reset iterator to the beginning of the forest frontier
974 0 : ctx->repair_iter = fd_forest_iter_init( ctx->forest );
975 0 : ctx->tsreset = now;
976 0 : }
977 :
978 : /* We are at the head of the turbine, so we should give turbine the
979 : chance to complete the shreds. !ele handles an edgecase where all
980 : frontier are fully complete and the iter is done */
981 :
982 0 : fd_forest_ele_t const * ele = fd_forest_pool_ele_const( pool, ctx->repair_iter.ele_idx );
983 0 : if( FD_LIKELY( !ele || ( ele->slot == fd_fseq_query( ctx->turbine_slot ) && ( now - ctx->tsreset ) < (long)30e6 ) ) ){
984 0 : return;
985 0 : }
986 :
987 0 : while( total_req < MAX_REQ_PER_CREDIT ){
988 0 : ele = fd_forest_pool_ele_const( pool, ctx->repair_iter.ele_idx );
989 : // Request first, advance iterator second.
990 0 : if( ctx->repair_iter.shred_idx == UINT_MAX && fd_repair_need_highest_window_index( ctx->repair, ele->slot, 0 ) ){
991 0 : fd_repair_send_requests( ctx, stem, fd_needed_highest_window_index, ele->slot, 0, now );
992 0 : total_req += FD_REPAIR_NUM_NEEDED_PEERS;
993 0 : } else if( fd_repair_need_window_index( ctx->repair, ele->slot, ctx->repair_iter.shred_idx ) ) {
994 0 : fd_repair_send_requests( ctx, stem, fd_needed_window_index, ele->slot, ctx->repair_iter.shred_idx, now );
995 0 : total_req += FD_REPAIR_NUM_NEEDED_PEERS;
996 0 : }
997 :
998 0 : ctx->repair_iter = fd_forest_iter_next( ctx->repair_iter, forest );
999 :
1000 0 : if( FD_UNLIKELY( fd_forest_iter_done( ctx->repair_iter, forest ) ) ) {
1001 : /* No more elements in the forest frontier, or the iterator got
1002 : invalidated, so we can start from top again. */
1003 0 : ctx->repair_iter = fd_forest_iter_init( forest );
1004 0 : break;
1005 0 : }
1006 0 : }
1007 :
1008 0 : fd_repair_continue( ctx->repair );
1009 0 : }
1010 :
1011 : static inline void
1012 0 : during_housekeeping( fd_repair_tile_ctx_t * ctx ) {
1013 0 : fd_repair_settime( ctx->repair, fd_log_wallclock() );
1014 :
1015 0 : long now = fd_log_wallclock();
1016 0 : if( FD_UNLIKELY( now - ctx->tsprint > (long)1e9 ) ) {
1017 0 : fd_forest_print( ctx->forest );
1018 0 : ctx->tsprint = fd_log_wallclock();
1019 0 : }
1020 :
1021 0 : if( FD_UNLIKELY( !ctx->stem ) ) {
1022 0 : return;
1023 0 : }
1024 0 : }
1025 : static void
1026 : privileged_init( fd_topo_t * topo,
1027 0 : fd_topo_tile_t * tile ) {
1028 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1029 :
1030 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1031 0 : fd_repair_tile_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_tile_ctx_t), sizeof(fd_repair_tile_ctx_t) );
1032 0 : fd_memset( ctx, 0, sizeof(fd_repair_tile_ctx_t) );
1033 :
1034 0 : uchar const * identity_key = fd_keyload_load( tile->repair.identity_key_path, /* pubkey only: */ 0 );
1035 0 : fd_memcpy( ctx->identity_private_key, identity_key, sizeof(fd_pubkey_t) );
1036 0 : fd_memcpy( ctx->identity_public_key.uc, identity_key + 32UL, sizeof(fd_pubkey_t) );
1037 :
1038 0 : ctx->repair_config.private_key = ctx->identity_private_key;
1039 0 : ctx->repair_config.public_key = &ctx->identity_public_key;
1040 :
1041 0 : tile->repair.good_peer_cache_file_fd = open( tile->repair.good_peer_cache_file, O_RDWR | O_CREAT, 0644 );
1042 0 : if( FD_UNLIKELY( tile->repair.good_peer_cache_file_fd==-1 ) ) {
1043 0 : FD_LOG_WARNING(( "Failed to open the good peer cache file (%s) (%i-%s)", tile->repair.good_peer_cache_file, errno, fd_io_strerror( errno ) ));
1044 0 : }
1045 0 : ctx->repair_config.good_peer_cache_file_fd = tile->repair.good_peer_cache_file_fd;
1046 :
1047 0 : FD_TEST( fd_rng_secure( &ctx->repair_seed, sizeof(ulong) ) );
1048 0 : }
1049 :
1050 : static void
1051 : unprivileged_init( fd_topo_t * topo,
1052 0 : fd_topo_tile_t * tile ) {
1053 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
1054 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
1055 0 : fd_repair_tile_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_repair_tile_ctx_t), sizeof(fd_repair_tile_ctx_t) );
1056 0 : ctx->tsprint = fd_log_wallclock();
1057 0 : ctx->tsrepair = fd_log_wallclock();
1058 0 : ctx->tsreset = fd_log_wallclock();
1059 :
1060 0 : if( FD_UNLIKELY( tile->in_cnt > MAX_IN_LINKS ) ) FD_LOG_ERR(( "repair tile has too many input links" ));
1061 :
1062 0 : uint sign_link_in_idx = UINT_MAX;
1063 0 : for( uint in_idx=0U; in_idx<(tile->in_cnt); in_idx++ ) {
1064 0 : fd_topo_link_t * link = &topo->links[ tile->in_link_id[ in_idx ] ];
1065 0 : if( 0==strcmp( link->name, "net_repair" ) ) {
1066 0 : ctx->in_kind[ in_idx ] = IN_KIND_NET;
1067 0 : fd_net_rx_bounds_init( &ctx->in_links[ in_idx ].net_rx, link->dcache );
1068 0 : continue;
1069 0 : } else if( 0==strcmp( link->name, "gossip_repai" ) ) {
1070 0 : ctx->in_kind[ in_idx ] = IN_KIND_CONTACT;
1071 0 : } else if( 0==strcmp( link->name, "stake_out" ) ) {
1072 0 : ctx->in_kind[ in_idx ] = IN_KIND_STAKE;
1073 0 : } else if( 0==strcmp( link->name, "shred_repair" ) ) {
1074 0 : ctx->in_kind[ in_idx ] = IN_KIND_SHRED;
1075 0 : } else if( 0==strcmp( link->name, "sign_repair" ) ) {
1076 0 : ctx->in_kind[ in_idx ] = IN_KIND_SIGN;
1077 0 : sign_link_in_idx = in_idx;
1078 0 : } else {
1079 0 : FD_LOG_ERR(( "repair tile has unexpected input link %s", link->name ));
1080 0 : }
1081 :
1082 0 : ctx->in_links[ in_idx ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1083 0 : ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
1084 0 : ctx->in_links[ in_idx ].wmark = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
1085 0 : ctx->in_links[ in_idx ].mtu = link->mtu;
1086 0 : FD_TEST( fd_dcache_compact_is_safe( ctx->in_links[in_idx].mem, link->dcache, link->mtu, link->depth ) );
1087 0 : }
1088 0 : if( FD_UNLIKELY( sign_link_in_idx==UINT_MAX ) ) FD_LOG_ERR(( "Missing sign_repair link" ));
1089 :
1090 0 : uint net_link_out_idx = UINT_MAX;
1091 0 : uint sign_link_out_idx = UINT_MAX;
1092 0 : uint shred_tile_idx = 0;
1093 0 : for( uint out_idx=0U; out_idx<(tile->out_cnt); out_idx++ ) {
1094 0 : fd_topo_link_t * link = &topo->links[ tile->out_link_id[ out_idx ] ];
1095 :
1096 0 : if( 0==strcmp( link->name, "repair_net" ) ) {
1097 :
1098 0 : if( net_link_out_idx!=UINT_MAX ) continue; /* only use first net link */
1099 0 : net_link_out_idx = out_idx;
1100 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1101 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, link->dcache );
1102 0 : ctx->net_out_wmark = fd_dcache_compact_wmark( ctx->net_out_mem, link->dcache, link->mtu );
1103 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
1104 :
1105 0 : } else if( 0==strcmp( link->name, "repair_sign" ) ) {
1106 :
1107 0 : sign_link_out_idx = out_idx;
1108 :
1109 0 : } else if( 0==strcmp( link->name, "repair_repla" ) ) {
1110 :
1111 0 : ctx->replay_out_mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1112 0 : ctx->replay_out_chunk0 = fd_dcache_compact_chunk0( ctx->replay_out_mem, link->dcache );
1113 0 : ctx->replay_out_wmark = fd_dcache_compact_wmark( ctx->replay_out_mem, link->dcache, link->mtu );
1114 0 : ctx->replay_out_chunk = ctx->replay_out_chunk0;
1115 :
1116 0 : } else if ( 0==strcmp( link->name, "repair_shred" ) ) {
1117 :
1118 0 : fd_repair_out_ctx_t * shred_out = &ctx->shred_out_ctx[ shred_tile_idx++ ];
1119 0 : shred_out->idx = out_idx;
1120 0 : shred_out->mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
1121 0 : shred_out->chunk0 = fd_dcache_compact_chunk0( shred_out->mem, link->dcache );
1122 0 : shred_out->wmark = fd_dcache_compact_wmark( shred_out->mem, link->dcache, link->mtu );
1123 0 : shred_out->chunk = shred_out->chunk0;
1124 :
1125 0 : } else {
1126 0 : FD_LOG_ERR(( "repair tile has unexpected output link %s", link->name ));
1127 0 : }
1128 :
1129 0 : }
1130 0 : if( FD_UNLIKELY( sign_link_out_idx==UINT_MAX ) ) FD_LOG_ERR(( "Missing repair_sign link" ));
1131 0 : if( FD_UNLIKELY( net_link_out_idx ==UINT_MAX ) ) FD_LOG_ERR(( "Missing repair_net link" ));
1132 0 : ctx->shred_tile_cnt = shred_tile_idx;
1133 0 : FD_TEST( ctx->shred_tile_cnt == fd_topo_tile_name_cnt( topo, "shred" ) );
1134 :
1135 : /* Scratch mem setup */
1136 :
1137 0 : ctx->blockstore = &ctx->blockstore_ljoin;
1138 0 : ctx->repair = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_align(), fd_repair_footprint() );
1139 0 : ctx->forest = FD_SCRATCH_ALLOC_APPEND( l, fd_forest_align(), fd_forest_footprint( tile->repair.slot_max ) );
1140 0 : ctx->fec_sigs = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_sig_align(), fd_fec_sig_footprint( 20 ) );
1141 0 : ctx->reasm = FD_SCRATCH_ALLOC_APPEND( l, fd_reasm_align(), fd_reasm_footprint( 20 ) );
1142 : // ctx->fec_repair = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_repair_align(), fd_fec_repair_footprint( ( 1<<20 ), tile->repair.shred_tile_cnt ) );
1143 : /* Look at fec_repair.h for an explanation of this fec_max. */
1144 :
1145 0 : ctx->fec_chainer = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_chainer_align(), fd_fec_chainer_footprint( 1 << 20 ) );
1146 :
1147 0 : void * smem = FD_SCRATCH_ALLOC_APPEND( l, fd_scratch_smem_align(), fd_scratch_smem_footprint( FD_REPAIR_SCRATCH_MAX ) );
1148 0 : void * fmem = FD_SCRATCH_ALLOC_APPEND( l, fd_scratch_fmem_align(), fd_scratch_fmem_footprint( FD_REPAIR_SCRATCH_DEPTH ) );
1149 :
1150 0 : FD_TEST( ( !!smem ) & ( !!fmem ) );
1151 0 : fd_scratch_attach( smem, fmem, FD_REPAIR_SCRATCH_MAX, FD_REPAIR_SCRATCH_DEPTH );
1152 :
1153 0 : ctx->wksp = topo->workspaces[ topo->objs[ tile->tile_obj_id ].wksp_id ].wksp;
1154 :
1155 0 : ctx->repair_intake_addr.port = fd_ushort_bswap( tile->repair.repair_intake_listen_port );
1156 0 : ctx->repair_serve_addr.port = fd_ushort_bswap( tile->repair.repair_serve_listen_port );
1157 :
1158 0 : ctx->repair_intake_listen_port = tile->repair.repair_intake_listen_port;
1159 0 : ctx->repair_serve_listen_port = tile->repair.repair_serve_listen_port;
1160 :
1161 0 : ctx->net_id = (ushort)0;
1162 :
1163 0 : fd_ip4_udp_hdr_init( ctx->intake_hdr, FD_REPAIR_MAX_PACKET_SIZE, 0, ctx->repair_intake_listen_port );
1164 0 : fd_ip4_udp_hdr_init( ctx->serve_hdr, FD_REPAIR_MAX_PACKET_SIZE, 0, ctx->repair_serve_listen_port );
1165 :
1166 : /* Keyguard setup */
1167 0 : fd_topo_link_t * sign_in = &topo->links[ tile->in_link_id[ sign_link_in_idx ] ];
1168 0 : fd_topo_link_t * sign_out = &topo->links[ tile->out_link_id[ sign_link_out_idx ] ];
1169 0 : if( fd_keyguard_client_join( fd_keyguard_client_new( ctx->keyguard_client,
1170 0 : sign_out->mcache,
1171 0 : sign_out->dcache,
1172 0 : sign_in->mcache,
1173 0 : sign_in->dcache ) ) == NULL ) {
1174 0 : FD_LOG_ERR(( "Keyguard join failed" ));
1175 0 : }
1176 :
1177 : /* Blockstore setup */
1178 0 : ulong blockstore_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "blockstore" );
1179 0 : FD_TEST( blockstore_obj_id!=ULONG_MAX );
1180 0 : ctx->blockstore_wksp = topo->workspaces[ topo->objs[ blockstore_obj_id ].wksp_id ].wksp;
1181 :
1182 0 : if( ctx->blockstore_wksp==NULL ) {
1183 0 : FD_LOG_ERR(( "no blocktore workspace" ));
1184 0 : }
1185 :
1186 0 : ctx->blockstore = fd_blockstore_join( &ctx->blockstore_ljoin, fd_topo_obj_laddr( topo, blockstore_obj_id ) );
1187 0 : FD_TEST( ctx->blockstore!=NULL );
1188 :
1189 0 : FD_LOG_NOTICE(( "repair starting" ));
1190 :
1191 : /* Repair set up */
1192 :
1193 0 : ctx->repair = fd_repair_join( fd_repair_new( ctx->repair, ctx->repair_seed ) );
1194 0 : ctx->forest = fd_forest_join( fd_forest_new( ctx->forest, tile->repair.slot_max, ctx->repair_seed ) );
1195 : // ctx->fec_repair = fd_fec_repair_join( fd_fec_repair_new( ctx->fec_repair, ( tile->repair.max_pending_shred_sets + 2 ), tile->repair.shred_tile_cnt, 0 ) );
1196 0 : ctx->fec_sigs = fd_fec_sig_join( fd_fec_sig_new( ctx->fec_sigs, 20 ) );
1197 0 : ctx->reasm = fd_reasm_join( fd_reasm_new( ctx->reasm, 20 ) );
1198 0 : ctx->fec_chainer = fd_fec_chainer_join( fd_fec_chainer_new( ctx->fec_chainer, 1 << 20, 0 ) );
1199 0 : ctx->repair_iter = fd_forest_iter_init( ctx->forest );
1200 0 : FD_TEST( fd_forest_iter_done( ctx->repair_iter, ctx->forest ) );
1201 :
1202 : /**********************************************************************/
1203 : /* turbine_slot fseq */
1204 : /**********************************************************************/
1205 :
1206 0 : ulong turbine_slot0_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "turbine_slot0" );
1207 0 : FD_TEST( turbine_slot0_obj_id!=ULONG_MAX );
1208 0 : ctx->turbine_slot0 = fd_fseq_join( fd_topo_obj_laddr( topo, turbine_slot0_obj_id ) );
1209 0 : FD_TEST( ctx->turbine_slot0 );
1210 0 : FD_TEST( fd_fseq_query( ctx->turbine_slot0 )==ULONG_MAX );
1211 :
1212 0 : ulong turbine_slot_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "turbine_slot" );
1213 0 : FD_TEST( turbine_slot_obj_id!=ULONG_MAX );
1214 0 : ctx->turbine_slot = fd_fseq_join( fd_topo_obj_laddr( topo, turbine_slot_obj_id ) );
1215 0 : FD_TEST( ctx->turbine_slot );
1216 0 : fd_fseq_update( ctx->turbine_slot, 0UL );
1217 :
1218 0 : FD_LOG_NOTICE(( "repair my addr - intake addr: " FD_IP4_ADDR_FMT ":%u, serve_addr: " FD_IP4_ADDR_FMT ":%u",
1219 0 : FD_IP4_ADDR_FMT_ARGS( ctx->repair_intake_addr.addr ), fd_ushort_bswap( ctx->repair_intake_addr.port ),
1220 0 : FD_IP4_ADDR_FMT_ARGS( ctx->repair_serve_addr.addr ), fd_ushort_bswap( ctx->repair_serve_addr.port ) ));
1221 :
1222 0 : ulong root_slot_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "root_slot" );
1223 0 : FD_TEST( root_slot_obj_id!=ULONG_MAX );
1224 0 : ctx->wmark = fd_fseq_join( fd_topo_obj_laddr( topo, root_slot_obj_id ) );
1225 0 : if( FD_UNLIKELY( !ctx->wmark ) ) FD_LOG_ERR(( "replay tile has no root_slot fseq" ));
1226 0 : ctx->prev_wmark = fd_fseq_query( ctx->wmark );
1227 :
1228 0 : if( fd_repair_set_config( ctx->repair, &ctx->repair_config ) ) {
1229 0 : FD_LOG_ERR( ( "error setting repair config" ) );
1230 0 : }
1231 :
1232 0 : fd_repair_update_addr( ctx->repair, &ctx->repair_intake_addr, &ctx->repair_serve_addr );
1233 :
1234 0 : fd_repair_settime( ctx->repair, fd_log_wallclock() );
1235 0 : fd_repair_start( ctx->repair );
1236 :
1237 0 : ulong scratch_top = FD_SCRATCH_ALLOC_FINI( l, 1UL );
1238 0 : if( FD_UNLIKELY( scratch_top > (ulong)scratch + scratch_footprint( tile ) ) )
1239 0 : FD_LOG_ERR(( "scratch overflow %lu %lu %lu", scratch_top - (ulong)scratch - scratch_footprint( tile ), scratch_top, (ulong)scratch + scratch_footprint( tile ) ));
1240 0 : }
1241 :
1242 : static ulong
1243 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
1244 : fd_topo_tile_t const * tile,
1245 : ulong out_cnt,
1246 0 : struct sock_filter * out ) {
1247 0 : populate_sock_filter_policy_fd_repair_tile(
1248 0 : out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)tile->repair.good_peer_cache_file_fd );
1249 0 : return sock_filter_policy_fd_repair_tile_instr_cnt;
1250 0 : }
1251 :
1252 : static ulong
1253 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
1254 : fd_topo_tile_t const * tile,
1255 : ulong out_fds_cnt,
1256 0 : int * out_fds ) {
1257 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
1258 :
1259 0 : ulong out_cnt = 0UL;
1260 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
1261 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
1262 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
1263 0 : if( FD_LIKELY( -1!=tile->repair.good_peer_cache_file_fd ) )
1264 0 : out_fds[ out_cnt++ ] = tile->repair.good_peer_cache_file_fd; /* good peer cache file */
1265 0 : return out_cnt;
1266 0 : }
1267 :
1268 : static inline void
1269 0 : fd_repair_update_repair_metrics( fd_repair_metrics_t * metrics ) {
1270 0 : FD_MCNT_SET( REPAIR, RECV_CLNT_PKT, metrics->recv_clnt_pkt );
1271 0 : FD_MCNT_SET( REPAIR, RECV_SERV_PKT, metrics->recv_serv_pkt );
1272 0 : FD_MCNT_SET( REPAIR, RECV_SERV_CORRUPT_PKT, metrics->recv_serv_corrupt_pkt );
1273 0 : FD_MCNT_SET( REPAIR, RECV_SERV_INVALID_SIGNATURE, metrics->recv_serv_invalid_signature );
1274 0 : FD_MCNT_SET( REPAIR, RECV_SERV_FULL_PING_TABLE, metrics->recv_serv_full_ping_table );
1275 0 : FD_MCNT_ENUM_COPY( REPAIR, RECV_SERV_PKT_TYPES, metrics->recv_serv_pkt_types );
1276 0 : FD_MCNT_SET( REPAIR, RECV_PKT_CORRUPTED_MSG, metrics->recv_pkt_corrupted_msg );
1277 0 : FD_MCNT_SET( REPAIR, SEND_PKT_CNT, metrics->send_pkt_cnt );
1278 0 : FD_MCNT_ENUM_COPY( REPAIR, SENT_PKT_TYPES, metrics->sent_pkt_types );
1279 0 : }
1280 :
1281 : static inline void
1282 0 : metrics_write( fd_repair_tile_ctx_t * ctx ) {
1283 : /* Repair-protocol-specific metrics */
1284 0 : fd_repair_update_repair_metrics( fd_repair_get_metrics( ctx->repair ) );
1285 0 : }
1286 :
1287 : /* TODO: This is probably not correct. */
1288 0 : #define STEM_BURST (2UL)
1289 :
1290 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_repair_tile_ctx_t
1291 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_repair_tile_ctx_t)
1292 :
1293 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
1294 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
1295 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
1296 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
1297 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
1298 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1299 :
1300 : #include "../../disco/stem/fd_stem.c"
1301 :
1302 : fd_topo_run_tile_t fd_tile_repair = {
1303 : .name = "repair",
1304 : .loose_footprint = loose_footprint,
1305 : .populate_allowed_seccomp = populate_allowed_seccomp,
1306 : .populate_allowed_fds = populate_allowed_fds,
1307 : .scratch_align = scratch_align,
1308 : .scratch_footprint = scratch_footprint,
1309 : .unprivileged_init = unprivileged_init,
1310 : .privileged_init = privileged_init,
1311 : .run = stem_run,
1312 : };
|
