Line data Source code
1 : /* fd_rserve_tile serves incoming repair requests from other nodes */
2 :
3 : #define _GNU_SOURCE
4 : #include <errno.h>
5 : #include <fcntl.h>
6 :
7 : #include "fd_rserve.h"
8 : #include "fd_repair_tile.h"
9 : #include "fd_repair_serde.h"
10 : #include "../../disco/fd_disco_base.h"
11 : #include "../../disco/keyguard/fd_keyguard_client.h"
12 : #include "../../disco/keyguard/fd_keyguard.h"
13 : #include "../../disco/keyguard/fd_keyload.h"
14 : #include "../../disco/keyguard/fd_keyswitch.h"
15 : #include "../../disco/metrics/fd_metrics.h"
16 : #include "../../disco/net/fd_net_tile.h"
17 : #include "../../disco/shred/fd_shred_tile.h"
18 : #include "../../disco/store/fd_store.h"
19 : #include "../../disco/topo/fd_topo.h"
20 : #include "../../util/pod/fd_pod_format.h"
21 : #include "../../flamenco/gossip/fd_gossip_message.h"
22 : #include "../../util/net/fd_net_headers.h"
23 :
24 : #include <linux/futex.h>
25 : #include "generated/fd_rserve_tile_seccomp.h"
26 :
27 0 : #define IN_KIND_NET (0)
28 0 : #define IN_KIND_SIGN (1)
29 0 : #define IN_KIND_SHRED (2)
30 0 : #define IN_KIND_REPAIR (3)
31 :
32 : #define MAX_IN_LINKS FD_TOPO_MAX_TILE_IN_LINKS
33 :
34 0 : #define FD_RSERVE_MAX_PACKET_SIZE 1232
35 :
36 : /* The maximum number of parent slots to look for. */
37 0 : #define FD_RSERVE_MAX_ORPHAN_SLOTS 11
38 :
39 : /* 10 minutes in milliseconds. */
40 : #define FD_RSERVE_SIGNED_REPAIR_WINDOW (60L*10L*1000L)
41 :
42 : /* static map from request type to response metric array index */
43 : static uint response_metric_index[FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID + 1] = {
44 : [FD_REPAIR_SERDE_TAG_PING] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_PING_IDX,
45 : [FD_REPAIR_SERDE_TAG_WINDOW_INDEX] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_WINDOW_INDEX_IDX,
46 : [FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_HIGHEST_WINDOW_INDEX_IDX,
47 : [FD_REPAIR_SERDE_TAG_ORPHAN] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_ORPHAN_IDX,
48 : [FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_PARENT_FEC_SET_COUNT_IDX,
49 : [FD_REPAIR_SERDE_TAG_FEC_SET_ROOT] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_FEC_SET_ROOT_IDX,
50 : [FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_WINDOW_INDEX_FOR_BLOCK_ID_IDX,
51 : };
52 :
53 : /* static map from request type to received request metric array index */
54 : static uint request_metric_index[FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID + 1] = {
55 : [FD_REPAIR_SERDE_TAG_PONG] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_PONG_IDX,
56 : [FD_REPAIR_SERDE_TAG_WINDOW_INDEX] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_WINDOW_INDEX_IDX,
57 : [FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_HIGHEST_WINDOW_INDEX_IDX,
58 : [FD_REPAIR_SERDE_TAG_ORPHAN] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_ORPHAN_IDX,
59 : [FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_PARENT_FEC_SET_COUNT_IDX,
60 : [FD_REPAIR_SERDE_TAG_FEC_SET_ROOT] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_FEC_SET_ROOT_IDX,
61 : [FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_WINDOW_INDEX_FOR_BLOCK_ID_IDX,
62 : };
63 :
64 :
65 : typedef union {
66 : struct {
67 : fd_wksp_t * mem;
68 : ulong chunk0;
69 : ulong wmark;
70 : ulong mtu;
71 : };
72 : fd_net_rx_bounds_t net_rx;
73 : } in_ctx_t;
74 :
75 : typedef struct ctx {
76 : fd_net_rx_bounds_t net_rx;
77 :
78 : uint in_kind [ MAX_IN_LINKS ];
79 : in_ctx_t in_links[ MAX_IN_LINKS ];
80 :
81 : uint net_out_idx;
82 : fd_wksp_t * net_out_mem;
83 : ulong net_out_chunk0;
84 : ulong net_out_wmark;
85 : ulong net_out_chunk;
86 :
87 : ulong seed;
88 : uchar rserve_secret[ 32 ];
89 : fd_rserve_t * rserve;
90 : fd_blockdb_t * blockdb; /* NULL if disabled */
91 : fd_store_t * store;
92 : int disk_fd;
93 : ulong max_shreds_per_block;
94 :
95 : /* Used for verifying incoming requests, and signing outgoing responses. */
96 : fd_sha512_t sha512[1];
97 : fd_keyguard_client_t keyguard_client[1];
98 : fd_keyswitch_t * keyswitch;
99 : fd_pubkey_t identity_public_key;
100 : int halt_signing;
101 :
102 : fd_ip4_udp_hdrs_t serve_hdr[1];
103 : ushort net_id;
104 :
105 : uchar net_buf[ FD_NET_MTU ];
106 : fd_repair_out_blk_t repair_buf[1];
107 :
108 : struct {
109 : ulong received_request_count[FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_CNT];
110 : ulong received_request_bytes;
111 : ulong received_malformed_count[FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_CNT];
112 :
113 : ulong send_pkt_cnt;
114 : ulong sent_pkt_types [FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_CNT];
115 : ulong sent_response_bytes;
116 :
117 : ulong unanswered [FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_CNT];
118 : ulong dropped [FD_METRICS_ENUM_REQUEST_DROP_REASON_CNT];
119 : ulong disk_read[FD_METRICS_ENUM_DISK_READ_RESULT_CNT];
120 : ulong disk_read_bytes;
121 : ulong shreds_current;
122 : ulong disk_current_bytes;
123 : ulong disk_allocated_bytes;
124 : fd_histf_t disk_write_timing[1];
125 : ulong disk_inserted;
126 : ulong disk_write_failed;
127 : ulong disk_write_bytes;
128 : fd_histf_t fec_preevict_timing[1];
129 : ulong fec_preevict_write_cnt;
130 : ulong fec_preevict_write_bytes;
131 :
132 : ulong ping_cache_entries;
133 : ulong ping_cache_capacity;
134 : ulong ping_cache_evictions;
135 : } metrics[ 1 ];
136 : } ctx_t;
137 :
138 : FD_FN_CONST static inline ulong
139 0 : scratch_align( void ) {
140 0 : return 128UL;
141 0 : }
142 :
143 : FD_FN_PURE static inline ulong
144 0 : scratch_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
145 0 : ulong l = FD_LAYOUT_INIT;
146 0 : l = FD_LAYOUT_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
147 0 : l = FD_LAYOUT_APPEND( l, fd_rserve_align(), fd_rserve_footprint( tile->rserve.ping_cache_entries) );
148 0 : if( FD_LIKELY( tile->rserve.blockdb_max ) ) {
149 0 : l = FD_LAYOUT_APPEND( l, fd_blockdb_align(), fd_blockdb_footprint( tile->rserve.blockdb_max ) );
150 0 : }
151 0 : return FD_LAYOUT_FINI( l, scratch_align() );
152 0 : }
153 :
154 : static void
155 : send_packet( ctx_t * ctx,
156 : fd_stem_context_t * stem,
157 : uint dst_ip_addr,
158 : ushort dst_port,
159 : uint src_ip_addr,
160 : uchar const * payload,
161 : ulong payload_sz,
162 0 : ulong tsorig ) {
163 0 : ctx->metrics->send_pkt_cnt++;
164 0 : ctx->metrics->sent_response_bytes += payload_sz;
165 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
166 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
167 0 : *hdr = *ctx->serve_hdr;
168 :
169 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
170 0 : ip4->saddr = src_ip_addr;
171 0 : ip4->daddr = dst_ip_addr;
172 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
173 0 : ip4->check = 0U;
174 0 : ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
175 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
176 :
177 0 : fd_udp_hdr_t * udp = hdr->udp;
178 0 : udp->net_dport = dst_port;
179 0 : udp->net_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
180 0 : fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
181 0 : hdr->udp->check = 0U;
182 :
183 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
184 0 : ulong sig = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
185 0 : ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
186 0 : ulong chunk = ctx->net_out_chunk;
187 0 : fd_stem_publish( stem, ctx->net_out_idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
188 0 : ctx->net_out_chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
189 0 : }
190 :
191 : static inline void
192 : handle_pong( ctx_t * ctx,
193 : uchar const * payload,
194 : ulong payload_sz,
195 : uint saddr,
196 0 : ushort sport ) {
197 0 : fd_repair_pong_t pong;
198 0 : fd_ed25519_sig_t signature;
199 0 : fd_pubkey_t from;
200 0 : if( FD_UNLIKELY( fd_repair_pong_de( payload, payload_sz, &pong, signature, &from )!=payload_sz ) ) return;
201 :
202 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &from ) ) ) {
203 : /* We've received our own repair request, ignore. */
204 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_FROM_SELF_IDX ]++;
205 0 : return;
206 0 : }
207 :
208 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS!=fd_ed25519_verify( pong.hash.uc, 32UL, signature, from.uc, ctx->sha512 ) ) ) {
209 : /* Invalid signature, ignore. */
210 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SIGNATURE_FAILED_IDX ]++;
211 0 : return;
212 0 : }
213 :
214 : /* Verify that the pong hash corresponds to the token we would have issued
215 : to this (pubkey, address) under either the current or previous rotating
216 : secret. */
217 0 : if( FD_UNLIKELY( !fd_rserve_pong_token_verify( ctx->rserve, pong.hash.uc, &from, saddr, sport ) ) ) {
218 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_PONG_TOKEN_INVALID_IDX ]++;
219 0 : return;
220 0 : }
221 :
222 0 : fd_rserve_t * rserve = ctx->rserve;
223 0 : ping_cache_key_t key[1];
224 0 : memset( key, 0, sizeof(ping_cache_key_t) );
225 0 : key->pubkey = from;
226 0 : key->ip4 = saddr;
227 0 : key->port = sport;
228 :
229 0 : ping_cache_entry_t * entry = ping_map_ele_query( rserve->ping_map, key, NULL, rserve->ping_pool );
230 :
231 0 : if( FD_LIKELY( !entry ) ) {
232 : /* New entry, evict LRU if pool is full. */
233 0 : if( FD_UNLIKELY( !ping_pool_free( rserve->ping_pool ) ) ) {
234 0 : ping_cache_entry_t * victim = ping_dlist_ele_pop_head( rserve->ping_dlist, rserve->ping_pool );
235 0 : ping_map_ele_remove_fast( rserve->ping_map, victim, rserve->ping_pool );
236 0 : ping_pool_ele_release( rserve->ping_pool, victim );
237 0 : ctx->metrics->ping_cache_entries--;
238 0 : ctx->metrics->ping_cache_evictions++;
239 0 : }
240 0 : entry = ping_pool_ele_acquire( rserve->ping_pool );
241 0 : entry->key = *key;
242 0 : ping_map_ele_insert( rserve->ping_map, entry, rserve->ping_pool );
243 0 : ctx->metrics->ping_cache_entries++;
244 0 : } else {
245 : /* Existing entry, move to tail. */
246 0 : ping_dlist_ele_remove( rserve->ping_dlist, entry, rserve->ping_pool );
247 0 : }
248 :
249 0 : entry->timestamp = (ulong)fd_log_wallclock();
250 0 : ping_dlist_ele_push_tail( rserve->ping_dlist, entry, rserve->ping_pool );
251 0 : return;
252 0 : }
253 :
254 : static inline void
255 : handle_net_request( ctx_t * ctx,
256 : fd_stem_context_t * stem,
257 : uchar const * payload,
258 : ulong payload_sz,
259 : fd_udp_hdr_t * udp,
260 0 : fd_ip4_hdr_t * ip4 ) {
261 0 : if( FD_UNLIKELY( payload_sz<4UL ) ) {
262 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_TOO_SMALL_IDX ]++;
263 0 : return;
264 0 : }
265 0 : uint tag = FD_LOAD( uint, payload );
266 :
267 0 : if( FD_UNLIKELY( tag==FD_REPAIR_SERDE_TAG_PONG ) ) {
268 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
269 0 : ctx->metrics->received_request_bytes += payload_sz;
270 0 : handle_pong( ctx, payload, payload_sz, ip4->saddr, udp->net_sport );
271 0 : return;
272 0 : }
273 : /* Block id requests need the blockdb, which only exists when
274 : alpenglow is enabled. */
275 0 : int is_bid = tag==FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT || tag==FD_REPAIR_SERDE_TAG_FEC_SET_ROOT || tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID;
276 0 : if( FD_UNLIKELY( tag!=FD_REPAIR_SERDE_TAG_WINDOW_INDEX &&
277 0 : tag!=FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX &&
278 0 : tag!=FD_REPAIR_SERDE_TAG_ORPHAN &&
279 0 : !( is_bid && ctx->blockdb ) ) ) {
280 0 : if( tag==FD_REPAIR_SERDE_TAG_PING ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNSUPPORTED_PING_IDX ]++;
281 0 : else if( tag==FD_REPAIR_SERDE_TAG_ANCESTOR_HASHES ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNSUPPORTED_ANCESTOR_HASHES_IDX ]++;
282 0 : else ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNKNOWN_TAG_IDX ]++;
283 0 : return;
284 0 : }
285 :
286 : /* The request must be exactly its kind's size, checked before
287 : constructing the signable payload to avoid OOB reads. */
288 0 : fd_ed25519_sig_t signature;
289 0 : fd_pubkey_t sender;
290 0 : fd_pubkey_t recipient;
291 0 : ulong timestamp;
292 0 : uint nonce;
293 0 : union {
294 0 : fd_repair_shred_t shred;
295 0 : fd_repair_highest_shred_t highest_shred;
296 0 : fd_repair_orphan_t orphan;
297 0 : fd_repair_parent_fec_set_count_t parent_fec_set_count;
298 0 : fd_repair_fec_set_root_t fec_set_root;
299 0 : fd_repair_shred_for_block_id_t shred_for_block_id;
300 0 : } req;
301 0 : ulong req_sz = 0UL;
302 0 : switch( tag ) {
303 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX: req_sz = fd_repair_req_window_index_de ( payload, payload_sz, &req.shred, signature, &sender, &recipient, ×tamp, &nonce ); break;
304 0 : case FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX: req_sz = fd_repair_req_highest_window_index_de ( payload, payload_sz, &req.highest_shred, signature, &sender, &recipient, ×tamp, &nonce ); break;
305 0 : case FD_REPAIR_SERDE_TAG_ORPHAN: req_sz = fd_repair_req_orphan_de ( payload, payload_sz, &req.orphan, signature, &sender, &recipient, ×tamp, &nonce ); break;
306 0 : case FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: req_sz = fd_repair_req_parent_and_fec_set_count_de ( payload, payload_sz, &req.parent_fec_set_count, signature, &sender, &recipient, ×tamp, &nonce ); break;
307 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT: req_sz = fd_repair_req_fec_set_root_de ( payload, payload_sz, &req.fec_set_root, signature, &sender, &recipient, ×tamp, &nonce ); break;
308 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: req_sz = fd_repair_req_window_index_for_block_id_de( payload, payload_sz, &req.shred_for_block_id, signature, &sender, &recipient, ×tamp, &nonce ); break;
309 0 : }
310 0 : if( FD_UNLIKELY( req_sz!=payload_sz ) ) {
311 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_WRONG_SIZE_IDX ]++;
312 0 : return;
313 0 : }
314 :
315 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
316 0 : ctx->metrics->received_request_bytes += payload_sz;
317 :
318 0 : if( FD_UNLIKELY( !fd_pubkey_eq( &ctx->identity_public_key, &recipient ) ) ) {
319 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_WRONG_RECIPIENT_IDX ]++;
320 0 : return;
321 0 : }
322 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &sender ) ) ) {
323 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_FROM_SELF_IDX ]++;
324 0 : return;
325 0 : }
326 :
327 0 : long current = FD_NANOSEC_TO_MILLI( fd_log_wallclock() );
328 0 : long ts_diff = current - (long)timestamp;
329 0 : if( FD_UNLIKELY( ts_diff < 0L ) ) ts_diff = -ts_diff;
330 0 : if( FD_UNLIKELY( ts_diff > FD_RSERVE_SIGNED_REPAIR_WINDOW ) ) {
331 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_TIMESTAMP_SKEW_IDX ]++;
332 0 : return;
333 0 : }
334 :
335 : /* Most requests are for shreds we never stored, probe before doing
336 : a full ed25519 verify. */
337 0 : if( FD_LIKELY( tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX || tag==FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX ) ) {
338 0 : fd_repair_shred_t const * msg = &req.shred; /* highest_shred shares its layout */
339 0 : if( FD_UNLIKELY( msg->shred_idx>=ctx->max_shreds_per_block ) ) {
340 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
341 0 : return;
342 0 : }
343 0 : if( FD_LIKELY( !fd_store_disk_probe( ctx->store, msg->slot, tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX ? (uint)msg->shred_idx : UINT_MAX ) ) ) {
344 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
345 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
346 0 : return;
347 0 : }
348 0 : }
349 :
350 : /* Verify the signature. */
351 :
352 : /* The signed bytes are the tag followed by everything after the
353 : signature. */
354 0 : uchar signable[ FD_REPAIR_SIG_SER_MAX ];
355 0 : ulong signable_sz = fd_repair_req_sig_ser( payload, payload_sz, signable );
356 :
357 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS!=fd_ed25519_verify( signable, signable_sz, signature, sender.uc, ctx->sha512 ) ) ) {
358 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SIGNATURE_FAILED_IDX ]++;
359 0 : return;
360 0 : }
361 :
362 : /* Check whether we've heard a pong response from this peer at this
363 : exact source address. Keying on (pubkey, address) means a peer that
364 : ponged from one address cannot have repair responses redirected to
365 : a spoofed source address. */
366 0 : ping_cache_key_t key[1];
367 0 : memset( key, 0, sizeof(ping_cache_key_t) );
368 0 : key->pubkey = sender;
369 0 : key->ip4 = ip4->saddr;
370 0 : key->port = udp->net_sport;
371 0 : ping_cache_entry_t * entry = ping_map_ele_query( ctx->rserve->ping_map, key, NULL, ctx->rserve->ping_pool );
372 0 : if( FD_LIKELY( entry ) ) {
373 0 : switch( tag ) {
374 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX:
375 0 : case FD_REPAIR_SERDE_TAG_HIGHEST_WINDOW_INDEX: {
376 0 : fd_repair_shred_t const * msg = &req.shred; /* highest_shred shares its layout */
377 :
378 0 : ulong slot = msg->slot;
379 0 : ulong shred_idx = msg->shred_idx;
380 :
381 0 : if( FD_UNLIKELY( shred_idx>=ctx->max_shreds_per_block ) ) {
382 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
383 0 : return;
384 0 : }
385 :
386 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
387 0 : int len;
388 0 : for( ulong retry=0UL;; retry++ ) {
389 0 : if( tag==FD_REPAIR_SERDE_TAG_WINDOW_INDEX ) {
390 0 : len = fd_store_disk_query( ctx->store, ctx->disk_fd, slot, (uint)shred_idx, payload );
391 0 : } else {
392 0 : len = fd_store_disk_query_highest( ctx->store, ctx->disk_fd, slot, (uint)shred_idx, payload );
393 0 : }
394 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
395 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
396 0 : }
397 0 : if( FD_UNLIKELY( len<0 ) ) {
398 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
399 0 : else if( len==FD_STORE_DISK_QUERY_SCAN_LIMIT) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SCAN_LIMIT_IDX ]++;
400 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
401 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
402 0 : return;
403 0 : }
404 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
405 0 : ctx->metrics->disk_read_bytes += (ulong)len;
406 :
407 0 : fd_memcpy( payload+len, &nonce, sizeof(uint) );
408 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
409 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
410 0 : return;
411 0 : }
412 0 : case FD_REPAIR_SERDE_TAG_ORPHAN: {
413 : /* Orphan repair works by giving us a "root" slot to start at,
414 : and has us walk back through the parent slots sending the
415 : highest shred of each slot.
416 : We may send up to FD_RSERVE_MAX_ORPHAN_SLOTS of these shreds
417 : (including the root one). */
418 0 : fd_repair_orphan_t const * msg = &req.orphan;
419 :
420 0 : ulong current = msg->slot;
421 0 : for( uint i=0; i<FD_RSERVE_MAX_ORPHAN_SLOTS; i++ ) {
422 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
423 0 : int len;
424 0 : for( ulong retry=0UL;; retry++ ) {
425 0 : len = fd_store_disk_query_highest( ctx->store, ctx->disk_fd, current, 0, payload );
426 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
427 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
428 0 : }
429 0 : if( FD_UNLIKELY( len<0 ) ) {
430 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
431 0 : else if( len==FD_STORE_DISK_QUERY_SCAN_LIMIT) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SCAN_LIMIT_IDX ]++;
432 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
433 0 : ctx->metrics->unanswered[ FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_ORPHAN_IDX ]++;
434 0 : return;
435 0 : }
436 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
437 0 : ctx->metrics->disk_read_bytes += (ulong)len;
438 0 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( payload );
439 0 : memcpy( payload+len, &nonce, sizeof(uint) );
440 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
441 0 : ctx->metrics->sent_pkt_types[ FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_ORPHAN_IDX ]++;
442 0 : ushort parent_off = shred->data.parent_off;
443 : /* fd_shred_parse ensures that parent_off will be 0 if and only if slot is 0 */
444 0 : if( FD_UNLIKELY( parent_off==0 ) ) break;
445 0 : current = current - parent_off;
446 0 : }
447 0 : break;
448 0 : }
449 0 : case FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT: {
450 : /* Serves the block's parent, FEC set count, and the proof of its
451 : parent-info leaf, which follows the fec_set_cnt FEC root leaves. */
452 0 : fd_repair_parent_fec_set_count_t const * msg = &req.parent_fec_set_count;
453 :
454 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
455 0 : if( FD_UNLIKELY( !blk ) ) {
456 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
457 0 : return;
458 0 : }
459 :
460 0 : uchar proof[ FD_BLOCKDB_PROOF_NODE_MAX*FD_SHRED_MERKLE_NODE_SZ ];
461 0 : int proof_len = fd_blockdb_proof( ctx->blockdb, blk, blk->fec_set_cnt, proof );
462 0 : FD_TEST( proof_len>=0 );
463 :
464 0 : uchar buf[ FD_REPAIR_RES_DE_MAX ];
465 0 : ulong sz = fd_repair_res_parent_fec_set_count_ser( blk->fec_set_cnt, blk->parent_slot, &blk->parent_block_id,
466 0 : proof, (ulong)proof_len*FD_SHRED_MERKLE_NODE_SZ, nonce, buf, sizeof(buf) );
467 0 : FD_TEST( sz );
468 :
469 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, buf, sz, fd_frag_meta_ts_comp( fd_tickcount() ) );
470 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
471 0 : return;
472 0 : }
473 0 : case FD_REPAIR_SERDE_TAG_FEC_SET_ROOT: {
474 : /* Serves the merkle root prefix of FEC set fec_set_idx and the
475 : proof of its leaf. */
476 0 : fd_repair_fec_set_root_t const * msg = &req.fec_set_root;
477 :
478 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
479 0 : if( FD_UNLIKELY( !blk ) ) {
480 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
481 0 : return;
482 0 : }
483 0 : if( FD_UNLIKELY( msg->fec_set_idx%FD_FEC_SHRED_CNT || msg->fec_set_idx/FD_FEC_SHRED_CNT>=blk->fec_set_cnt ) ) {
484 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_FEC_SET_INDEX_INVALID_IDX ]++;
485 0 : return;
486 0 : }
487 0 : ulong leaf_idx = msg->fec_set_idx/FD_FEC_SHRED_CNT;
488 :
489 0 : uchar proof[ FD_BLOCKDB_PROOF_NODE_MAX*FD_SHRED_MERKLE_NODE_SZ ];
490 0 : int proof_len = fd_blockdb_proof( ctx->blockdb, blk, leaf_idx, proof );
491 0 : FD_TEST( proof_len>=0 );
492 :
493 0 : uchar buf[ FD_REPAIR_RES_DE_MAX ];
494 0 : ulong sz = fd_repair_res_fec_set_root_ser( blk->merkle_roots[ leaf_idx ], proof, (ulong)proof_len*FD_SHRED_MERKLE_NODE_SZ, nonce, buf, sizeof(buf) );
495 0 : FD_TEST( sz );
496 :
497 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, buf, sz, fd_frag_meta_ts_comp( fd_tickcount() ) );
498 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
499 0 : return;
500 0 : }
501 0 : case FD_REPAIR_SERDE_TAG_WINDOW_INDEX_FOR_BLOCK_ID: {
502 : /* Serves the data shred at shred_idx of block (slot, block_id),
503 : found by its FEC set's merkle root. */
504 0 : fd_repair_shred_for_block_id_t const * msg = &req.shred_for_block_id;
505 :
506 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
507 0 : if( FD_UNLIKELY( !blk ) ) {
508 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
509 0 : return;
510 0 : }
511 0 : if( FD_UNLIKELY( (ulong)msg->shred_idx>=(ulong)blk->fec_set_cnt*FD_FEC_SHRED_CNT ) ) {
512 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
513 0 : return;
514 0 : }
515 :
516 0 : uchar const * root = blk->merkle_roots[ msg->shred_idx/FD_FEC_SHRED_CNT ];
517 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
518 0 : int len;
519 0 : for( ulong retry=0UL;; retry++ ) {
520 0 : len = fd_store_disk_query_root( ctx->store, ctx->disk_fd, root, msg->shred_idx, payload );
521 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
522 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
523 0 : }
524 0 : if( FD_UNLIKELY( len<0 ) ) {
525 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
526 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
527 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
528 0 : return;
529 0 : }
530 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
531 0 : ctx->metrics->disk_read_bytes += (ulong)len;
532 :
533 0 : memcpy( payload+len, &nonce, sizeof(uint) );
534 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
535 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
536 0 : return;
537 0 : }
538 0 : }
539 0 : } else {
540 : /* Derive a ping token bound to this peer's pubkey and the source
541 : address the request arrived from. The peer can only produce a
542 : valid pong if it actually receives this ping at that address. */
543 0 : uchar token[ 32UL ];
544 0 : fd_rserve_ping_token( ctx->rserve, token, &sender, ip4->saddr, udp->net_sport );
545 :
546 : /* Sign the token. */
547 0 : fd_ed25519_sig_t ping_sig;
548 0 : fd_keyguard_client_sign( ctx->keyguard_client, ping_sig, token, 32UL, FD_KEYGUARD_SIGN_TYPE_ED25519 );
549 :
550 : /* Like Agave, block id metadata requests are answered with the
551 : BlockIdRepairResponse ping variant, and ShredForBlockId, like the
552 : legacy requests, with the RepairResponse one. */
553 0 : uchar buf[ FD_REPAIR_PING_DE_SZ ];
554 0 : ulong sz;
555 0 : if( tag==FD_REPAIR_SERDE_TAG_PARENT_AND_FEC_SET_COUNT || tag==FD_REPAIR_SERDE_TAG_FEC_SET_ROOT ) sz = fd_repair_block_id_ping_ser( &ctx->identity_public_key, (fd_hash_t const *)token, ping_sig, buf );
556 0 : else sz = fd_repair_ping_ser ( &ctx->identity_public_key, (fd_hash_t const *)token, ping_sig, buf );
557 :
558 : /* Send the ping packet back to the source. */
559 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, buf, sz, fd_frag_meta_ts_comp( fd_tickcount() ) );
560 0 : ctx->metrics->sent_pkt_types[ FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_PING_IDX ]++;
561 0 : }
562 0 : }
563 :
564 :
565 : static inline int
566 : before_frag( ctx_t * ctx,
567 : ulong in_idx,
568 : ulong seq FD_PARAM_UNUSED,
569 0 : ulong sig ) {
570 0 : if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_RSERVE;
571 0 : return 0;
572 0 : }
573 :
574 : static inline void
575 : during_frag( ctx_t * ctx,
576 : ulong in_idx,
577 : ulong seq FD_PARAM_UNUSED,
578 : ulong sig,
579 : ulong chunk,
580 : ulong sz,
581 0 : ulong ctl ) {
582 0 : in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
583 0 : switch( ctx->in_kind[ in_idx ] ) {
584 0 : case IN_KIND_NET: {
585 0 : uchar const * buffer = fd_net_rx_translate_frag( &in_ctx->net_rx, chunk, ctl, sz );
586 0 : fd_memcpy( ctx->net_buf, buffer, sz );
587 0 : break;
588 0 : }
589 0 : case IN_KIND_REPAIR: {
590 0 : if( FD_UNLIKELY( sig!=REPAIR_SIG_BLK ) ) break;
591 0 : if( FD_UNLIKELY( sz<offsetof(fd_repair_out_blk_t, merkle_roots) || sz>sizeof(fd_repair_out_blk_t) || chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) )
592 0 : FD_LOG_ERR(( "repair_rserve chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
593 0 : fd_memcpy( ctx->repair_buf, fd_chunk_to_laddr_const( in_ctx->mem, chunk ), sz );
594 0 : break;
595 0 : }
596 0 : default: break;
597 0 : }
598 0 : }
599 :
600 : static inline void
601 : after_frag( ctx_t * ctx,
602 : ulong in_idx,
603 : ulong seq FD_PARAM_UNUSED,
604 : ulong sig,
605 : ulong sz,
606 : ulong tsorig FD_PARAM_UNUSED,
607 : ulong tspub FD_PARAM_UNUSED,
608 0 : fd_stem_context_t * stem ) {
609 0 : uint in_kind = ctx->in_kind[ in_idx ];
610 0 : if( FD_UNLIKELY( in_kind==IN_KIND_REPAIR ) ) {
611 0 : if( FD_UNLIKELY( sig!=REPAIR_SIG_BLK ) ) return;
612 0 : fd_repair_out_blk_t const * msg = ctx->repair_buf;
613 0 : if( FD_UNLIKELY( sz!=offsetof(fd_repair_out_blk_t, merkle_roots) + (ulong)msg->fec_set_cnt*FD_SHRED_MERKLE_NODE_SZ ) )
614 0 : FD_LOG_ERR(( "repair_rserve frag sz %lu does not match fec_set_cnt %u", sz, msg->fec_set_cnt ));
615 0 : FD_TEST( fd_blockdb_insert( ctx->blockdb, msg->slot, &msg->block_id, msg->parent_slot, &msg->parent_block_id,
616 0 : msg->fec_set_cnt, (uchar const *)msg->merkle_roots ) );
617 0 : return;
618 0 : }
619 0 : if( FD_LIKELY( in_kind!=IN_KIND_NET ) ) return; /* returnable_frag */
620 :
621 0 : uchar * payload; ulong payload_sz;
622 0 : fd_udp_hdr_t * udp;
623 0 : fd_ip4_hdr_t * ip4;
624 0 : if( FD_UNLIKELY( !fd_ip4_udp_hdr_strip( ctx->net_buf, sz, &payload, &payload_sz, NULL, &ip4, &udp ) ) ) {
625 0 : FD_LOG_WARNING(( "rserve: malformed packet (sz=%lu)", sz ));
626 0 : return;
627 0 : }
628 0 : handle_net_request( ctx, stem, payload, payload_sz, udp, ip4 );
629 0 : }
630 :
631 : static inline int
632 : returnable_frag( ctx_t * ctx,
633 : ulong in_idx,
634 : ulong seq FD_PARAM_UNUSED,
635 : ulong sig,
636 : ulong chunk,
637 : ulong sz,
638 : ulong ctl FD_PARAM_UNUSED,
639 : ulong tsorig FD_PARAM_UNUSED,
640 : ulong tspub FD_PARAM_UNUSED,
641 0 : fd_stem_context_t * stem FD_PARAM_UNUSED ) {
642 0 : uint in_kind = ctx->in_kind[ in_idx ];
643 0 : in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
644 :
645 0 : switch( in_kind ) {
646 0 : case IN_KIND_NET:
647 0 : return ctx->halt_signing; /* after_frag */
648 0 : case IN_KIND_SIGN: return 0; /* handled internally by keyguard_client */
649 0 : case IN_KIND_SHRED: {
650 0 : int shred_result = fd_shred_sig_res( sig );
651 0 : if( FD_UNLIKELY( fd_shred_sig_src( sig )>SHRED_SIG_SRC_BAD_REPAIR ||
652 0 : (shred_result!=SHRED_SIG_RESULT_OKAY && shred_result!=SHRED_SIG_RESULT_COMPLETES) ) ) return 0;
653 0 : if( FD_UNLIKELY( sz!=sizeof(fd_shred_base_t) || chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) )
654 0 : FD_LOG_ERR(( "shred_out chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
655 :
656 0 : fd_shred_base_t const * msg = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
657 0 : fd_shred_t const * shred = &msg->shred;
658 0 : if( FD_UNLIKELY( !fd_shred_is_data( fd_shred_type( shred->variant ) ) ) ) return 0;
659 :
660 0 : long dt = -fd_tickcount();
661 0 : int result = fd_store_disk_insert( ctx->store, ctx->disk_fd, shred, &msg->merkle_root );
662 0 : dt += fd_tickcount();
663 0 : fd_histf_sample( ctx->metrics->disk_write_timing, (ulong)dt );
664 0 : if( FD_LIKELY( result==FD_STORE_DISK_INSERT_SUCCESS ) ) {
665 0 : ctx->metrics->disk_inserted++;
666 0 : ctx->metrics->disk_write_bytes += sizeof(fd_shredb_entry_t);
667 0 : } else if( FD_UNLIKELY( result==FD_STORE_DISK_INSERT_ERR ) ) ctx->metrics->disk_write_failed++;
668 0 : return 0;
669 0 : }
670 0 : case IN_KIND_REPAIR: return 0; /* after_frag */
671 0 : default: FD_LOG_ERR(( "unexpected input kind (%u)", in_kind ));
672 0 : }
673 0 : }
674 :
675 : static inline void
676 : before_credit( ctx_t * ctx,
677 : fd_stem_context_t * stem FD_PARAM_UNUSED,
678 0 : int * charge_busy ) {
679 0 : fd_store_fec_spill_stats_t spill[1];
680 0 : if( FD_UNLIKELY( fd_store_fec_data_preevict( ctx->store, ctx->disk_fd, spill ) ) ) {
681 0 : fd_histf_sample( ctx->metrics->fec_preevict_timing, spill->write_ticks );
682 0 : ctx->metrics->fec_preevict_write_cnt += spill->write_cnt;
683 0 : ctx->metrics->fec_preevict_write_bytes += spill->write_bytes;
684 0 : *charge_busy = 1;
685 0 : }
686 0 : }
687 :
688 : static inline void
689 0 : during_housekeeping( ctx_t * ctx ) {
690 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_UNHALT_PENDING ) ) {
691 0 : FD_LOG_DEBUG(( "keyswitch: unhalting" ));
692 0 : FD_DCHECK_CRIT( ctx->halt_signing, "state machine corruption" );
693 0 : fd_memcpy( ctx->identity_public_key.uc, ctx->keyswitch->bytes, sizeof(fd_pubkey_t) );
694 0 : ctx->halt_signing = 0;
695 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
696 0 : }
697 :
698 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
699 0 : ctx->halt_signing = 1;
700 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
701 0 : }
702 :
703 0 : fd_rserve_t * rserve = ctx->rserve;
704 0 : ulong now_ns = (ulong)fd_log_wallclock();
705 0 : fd_rserve_maybe_rotate( rserve, now_ns );
706 :
707 : /* Evict expired entries from the head (oldest) of the LRU list. */
708 0 : while( !ping_dlist_is_empty( rserve->ping_dlist, rserve->ping_pool ) ) {
709 0 : ping_cache_entry_t * head = ping_dlist_ele_peek_head( rserve->ping_dlist, rserve->ping_pool );
710 0 : if( FD_LIKELY( now_ns-head->timestamp <= FD_RSERVE_PING_CACHE_TTL_NS ) ) break;
711 0 : ping_dlist_ele_pop_head( rserve->ping_dlist, rserve->ping_pool );
712 0 : ping_map_ele_remove_fast( rserve->ping_map, head, rserve->ping_pool );
713 0 : ping_pool_ele_release( rserve->ping_pool, head );
714 0 : ctx->metrics->ping_cache_entries--;
715 0 : }
716 0 : }
717 :
718 : static inline void
719 0 : metrics_write( ctx_t * ctx ) {
720 0 : fd_store_fec_cache_stats_t cache_stats[1] = {{0}};
721 0 : if( FD_LIKELY( ctx->store ) ) {
722 0 : fd_store_disk_stats_t stats[1];
723 0 : if( FD_LIKELY( !fd_store_disk_stats_query( ctx->store, stats ) ) ) {
724 0 : ctx->metrics->shreds_current = stats->shred_cnt;
725 0 : ctx->metrics->disk_current_bytes = stats->current_bytes;
726 0 : ctx->metrics->disk_allocated_bytes = stats->allocated_bytes;
727 0 : }
728 0 : fd_store_fec_cache_stats_query( ctx->store, cache_stats );
729 0 : }
730 :
731 0 : FD_MCNT_ENUM_COPY( RSERVE, REQUEST_RX, ctx->metrics->received_request_count );
732 0 : FD_MCNT_SET ( RSERVE, REQUEST_RX_BYTES, ctx->metrics->received_request_bytes );
733 0 : FD_MCNT_ENUM_COPY( RSERVE, REQUEST_MALFORMED, ctx->metrics->received_malformed_count );
734 0 : FD_MCNT_ENUM_COPY( RSERVE, REQUEST_DROPPED, ctx->metrics->dropped );
735 0 : FD_MCNT_ENUM_COPY( RSERVE, REQUEST_UNANSWERED, ctx->metrics->unanswered );
736 :
737 0 : FD_MCNT_SET ( RSERVE, PKT_TX, ctx->metrics->send_pkt_cnt );
738 0 : FD_MCNT_ENUM_COPY( RSERVE, RESPONSE_TX, ctx->metrics->sent_pkt_types );
739 0 : FD_MCNT_SET ( RSERVE, RESPONSE_TX_BYTES, ctx->metrics->sent_response_bytes );
740 :
741 0 : FD_MCNT_ENUM_COPY( RSERVE, DISK_READ_ATTEMPTED, ctx->metrics->disk_read );
742 0 : FD_MCNT_SET ( RSERVE, DISK_READ_BYTES, ctx->metrics->disk_read_bytes );
743 0 : FD_MGAUGE_SET ( RSERVE, SHRED_USED, ctx->metrics->shreds_current );
744 0 : FD_MGAUGE_SET ( RSERVE, DISK_USED_BYTES, ctx->metrics->disk_current_bytes );
745 0 : FD_MGAUGE_SET ( RSERVE, DISK_PEAK_BYTES, ctx->metrics->disk_allocated_bytes );
746 0 : FD_MHIST_COPY ( RSERVE, DISK_WRITE_DURATION_NANOS, ctx->metrics->disk_write_timing );
747 0 : FD_MCNT_SET ( RSERVE, DISK_SHRED_WRITTEN, ctx->metrics->disk_inserted );
748 0 : FD_MCNT_SET ( RSERVE, DISK_SHRED_WRITE_FAILED, ctx->metrics->disk_write_failed );
749 0 : FD_MCNT_SET ( RSERVE, DISK_SHRED_WRITTEN_BYTES, ctx->metrics->disk_write_bytes );
750 0 : FD_MHIST_COPY ( RSERVE, STORE_FEC_PREEVICT_DURATION_NANOS, ctx->metrics->fec_preevict_timing );
751 0 : FD_MCNT_SET ( RSERVE, STORE_FEC_PREEVICTED, ctx->metrics->fec_preevict_write_cnt );
752 0 : FD_MCNT_SET ( RSERVE, STORE_FEC_PREEVICTED_BYTES, ctx->metrics->fec_preevict_write_bytes );
753 0 : FD_MGAUGE_SET ( RSERVE, STORE_FEC_CACHE_FREE, cache_stats->free_cnt );
754 0 : FD_MGAUGE_SET ( RSERVE, STORE_FEC_CACHE_CAPACITY, cache_stats->max );
755 0 : FD_MGAUGE_SET ( RSERVE, STORE_FEC_CACHE_FREE_TARGET, cache_stats->target );
756 0 : FD_MGAUGE_SET ( RSERVE, STORE_FEC_CACHE_FREE_LOW_WATER, cache_stats->low_water );
757 :
758 0 : FD_MGAUGE_SET ( RSERVE, PING_CACHE_USED, ctx->metrics->ping_cache_entries );
759 0 : FD_MGAUGE_SET ( RSERVE, PING_CACHE_CAPACITY, ctx->metrics->ping_cache_capacity );
760 0 : FD_MCNT_SET ( RSERVE, PING_CACHE_EVICTED, ctx->metrics->ping_cache_evictions );
761 0 : }
762 :
763 : static void
764 : privileged_init( fd_topo_t const * topo,
765 0 : fd_topo_tile_t const * tile ) {
766 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
767 :
768 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
769 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
770 :
771 0 : FD_TEST( fd_rng_secure( &ctx->seed, sizeof(ulong) ) );
772 0 : FD_TEST( fd_rng_secure( ctx->rserve_secret, sizeof(ctx->rserve_secret) ) );
773 :
774 0 : ctx->disk_fd = -1;
775 0 : ulong store_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "store" );
776 0 : if( FD_LIKELY( store_obj_id!=ULONG_MAX ) ) {
777 0 : fd_store_t * store = fd_store_join( fd_topo_obj_laddr( topo, store_obj_id ) );
778 0 : FD_TEST( store );
779 0 : ctx->disk_fd = FD_STORE_FD_RW;
780 0 : if( FD_UNLIKELY( fcntl( ctx->disk_fd, F_GETFD )<0 ) )
781 0 : FD_LOG_ERR(( "store file descriptor was not inherited (%i-%s)", errno, fd_io_strerror( errno ) ));
782 0 : }
783 :
784 0 : uchar const * identity_public_key = fd_keyload_load( tile->rserve.identity_key_path, /* pubkey only: */ 1 );
785 0 : fd_memcpy( ctx->identity_public_key.uc, identity_public_key, sizeof(fd_pubkey_t) );
786 0 : }
787 :
788 : static void
789 : unprivileged_init( fd_topo_t const * topo,
790 0 : fd_topo_tile_t const * tile ) {
791 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
792 :
793 0 : ulong ping_cache_entries = tile->rserve.ping_cache_entries;
794 :
795 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
796 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
797 0 : ctx->rserve = FD_SCRATCH_ALLOC_APPEND( l, fd_rserve_align(), fd_rserve_footprint( ping_cache_entries ) );
798 0 : void * blockdb_mem = NULL;
799 0 : if( FD_LIKELY( tile->rserve.blockdb_max ) ) {
800 0 : blockdb_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_blockdb_align(), fd_blockdb_footprint( tile->rserve.blockdb_max ) );
801 0 : }
802 0 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, scratch_align() )==(ulong)scratch + scratch_footprint( tile ) );
803 :
804 0 : ctx->blockdb = NULL;
805 0 : if( FD_LIKELY( blockdb_mem ) ) {
806 0 : ctx->blockdb = fd_blockdb_join( fd_blockdb_new( blockdb_mem, tile->rserve.blockdb_max, ctx->seed ) );
807 0 : FD_TEST( ctx->blockdb );
808 0 : }
809 :
810 0 : ctx->store = NULL;
811 0 : ulong store_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "store" );
812 0 : if( FD_LIKELY( store_obj_id!=ULONG_MAX ) ) {
813 0 : ctx->store = fd_store_join( fd_topo_obj_laddr( topo, store_obj_id ) );
814 0 : FD_TEST( ctx->store );
815 0 : }
816 0 : if( FD_UNLIKELY( !ctx->store || !fd_store_has_disk( ctx->store ) ) ) {
817 0 : FD_LOG_ERR(( "rserve requires an enabled store disk layer" ));
818 0 : }
819 0 : if( FD_UNLIKELY( ctx->disk_fd<0 ) ) {
820 0 : FD_LOG_ERR(( "rserve could not open the store disk file" ));
821 0 : }
822 :
823 0 : ctx->max_shreds_per_block = tile->rserve.max_shreds_per_block;
824 0 : ctx->rserve = fd_rserve_join ( fd_rserve_new( ctx->rserve, ping_cache_entries, ctx->seed, ctx->rserve_secret ) );
825 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
826 0 : FD_TEST( ctx->keyswitch );
827 :
828 0 : fd_memset( ctx->metrics, 0, sizeof(ctx->metrics) );
829 0 : ctx->metrics->ping_cache_capacity = fd_ulong_pow2_up( ping_cache_entries );
830 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->disk_write_timing, FD_MHIST_EDGES_NS( RSERVE, DISK_WRITE_DURATION_NANOS ) ) );
831 0 : fd_histf_join( fd_metrics_histf_new_ns( ctx->metrics->fec_preevict_timing, FD_MHIST_EDGES_NS( RSERVE, STORE_FEC_PREEVICT_DURATION_NANOS ) ) );
832 0 : FD_MGAUGE_SET( RSERVE, SHRED_CAPACITY, ctx->store ? ctx->store->disk_max_shreds : 0UL );
833 :
834 0 : ctx->halt_signing = 0;
835 0 : ctx->net_id = (ushort)0;
836 0 : fd_ip4_udp_hdr_init( ctx->serve_hdr, FD_RSERVE_MAX_PACKET_SIZE, 0, tile->rserve.repair_serve_listen_port );
837 0 : fd_sha512_new( ctx->sha512 );
838 :
839 0 : ulong sign_in_idx = fd_topo_find_tile_in_link ( topo, tile, "sign_rserve", tile->kind_id );
840 0 : ulong sign_out_idx = fd_topo_find_tile_out_link( topo, tile, "rserve_sign", tile->kind_id );
841 0 : FD_TEST( sign_in_idx!=ULONG_MAX );
842 0 : fd_topo_link_t const * sign_in = &topo->links[ tile->in_link_id[ sign_in_idx ] ];
843 0 : fd_topo_link_t const * sign_out = &topo->links[ tile->out_link_id[ sign_out_idx ] ];
844 :
845 0 : fd_sleep_t * sleep = NULL;
846 0 : if( FD_UNLIKELY( topo->sleep_obj_id!=ULONG_MAX ) ) {
847 0 : sleep = fd_sleep_join( fd_topo_obj_laddr( topo, topo->sleep_obj_id ) );
848 0 : FD_TEST( sleep );
849 0 : }
850 :
851 0 : if( FD_UNLIKELY( !fd_keyguard_client_join( fd_keyguard_client_new( ctx->keyguard_client,
852 0 : sign_out->mcache,
853 0 : sign_out->dcache,
854 0 : sign_in->mcache,
855 0 : sign_in->dcache,
856 0 : sign_out->mtu,
857 0 : sign_in->mtu,
858 0 : sleep,
859 0 : sign_out->id,
860 0 : fd_topo_find_link_consumer( topo, sign_out ) ) ) ) ) {
861 0 : FD_LOG_ERR(( "failed to construct keyguard" ));
862 0 : }
863 :
864 0 : FD_TEST( tile->in_cnt>=1UL );
865 0 : FD_CHECK_ERR( tile->in_cnt<=MAX_IN_LINKS, "too many input links" );
866 0 : for( ulong in_idx=0UL; in_idx<tile->in_cnt; in_idx++ ) {
867 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ in_idx ] ];
868 0 : if( 0==strcmp( link->name, "net_rserve" ) ) {
869 0 : ctx->in_kind[ in_idx ] = IN_KIND_NET;
870 0 : fd_net_rx_bounds_init( &ctx->in_links[ in_idx ].net_rx, link->dcache );
871 0 : continue;
872 0 : }
873 0 : else if( 0==strcmp( link->name, "sign_rserve" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SIGN;
874 0 : else if( 0==strcmp( link->name, "shred_out" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SHRED;
875 0 : else if( 0==strcmp( link->name, "repair_rserve" ) ) {
876 0 : if( FD_UNLIKELY( !ctx->blockdb ) ) FD_LOG_ERR(( "rserve has a repair_rserve link but blockdb_max is 0" ));
877 0 : ctx->in_kind[ in_idx ] = IN_KIND_REPAIR;
878 0 : }
879 0 : else FD_LOG_ERR(( "rserve tile has unexpected input link: %s", link->name ));
880 :
881 0 : ctx->in_links[ in_idx ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
882 0 : ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
883 0 : ctx->in_links[ in_idx ].wmark = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
884 0 : ctx->in_links[ in_idx ].mtu = link->mtu;
885 0 : }
886 :
887 0 : ctx->net_out_idx = UINT_MAX;
888 0 : for( uint out_idx=0U; out_idx<tile->out_cnt; out_idx++ ) {
889 0 : fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ out_idx ] ];
890 0 : if( 0==strcmp( link->name, "rserve_net" ) ) {
891 0 : if( ctx->net_out_idx!=UINT_MAX ) continue; /* only use the first net link */
892 0 : ctx->net_out_idx = out_idx;
893 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
894 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, link->dcache );
895 0 : ctx->net_out_wmark = fd_dcache_compact_wmark( ctx->net_out_mem, link->dcache, link->mtu );
896 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
897 0 : }
898 0 : else if( 0==strcmp( link->name, "rserve_sign" ) ) { /* Handled above for keyguard. */ }
899 0 : else FD_LOG_ERR(( "rserve tile has unexpected output link: %s", link->name ));
900 0 : }
901 0 : if( FD_UNLIKELY( ctx->net_out_idx==UINT_MAX ) ) FD_LOG_ERR(( "Missing rserve_net output link" ));
902 0 : }
903 :
904 : static ulong
905 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
906 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
907 : ulong out_cnt,
908 0 : struct sock_filter * out ) {
909 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
910 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
911 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
912 0 : populate_sock_filter_policy_fd_rserve_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->disk_fd );
913 0 : return sock_filter_policy_fd_rserve_tile_instr_cnt;
914 0 : }
915 :
916 : static ulong
917 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
918 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
919 : ulong out_fds_cnt,
920 0 : int * out_fds ) {
921 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
922 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
923 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
924 :
925 0 : if( FD_UNLIKELY( out_fds_cnt<3UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
926 :
927 0 : ulong out_cnt = 0UL;
928 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
929 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
930 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
931 0 : if( FD_LIKELY( ctx->disk_fd>=0 ) )
932 0 : out_fds[ out_cnt++ ] = ctx->disk_fd;
933 0 : return out_cnt;
934 0 : }
935 :
936 : /* For orphan responses, we may send up to 11 net packets out.
937 : All other implemented codepaths will send at most 1. */
938 0 : #define STEM_BURST FD_RSERVE_MAX_ORPHAN_SLOTS
939 0 : #define STEM_LAZY (64000UL)
940 :
941 0 : #define STEM_CALLBACK_CONTEXT_TYPE ctx_t
942 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(ctx_t)
943 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
944 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
945 0 : #define STEM_CALLBACK_BEFORE_CREDIT before_credit
946 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
947 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
948 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
949 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
950 :
951 : #include "../../disco/stem/fd_stem.c"
952 :
953 : fd_topo_run_tile_t fd_tile_rserve = {
954 : .name = "rserve",
955 : .populate_allowed_seccomp = populate_allowed_seccomp,
956 : .populate_allowed_fds = populate_allowed_fds,
957 : .scratch_align = scratch_align,
958 : .scratch_footprint = scratch_footprint,
959 : .privileged_init = privileged_init,
960 : .unprivileged_init = unprivileged_init,
961 : .run = stem_run,
962 : };
|
