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.h"
9 : #include "fd_blockdb.h"
10 : #include "../rotor/fd_rotor_tile.h"
11 : #include "../../disco/fd_disco_base.h"
12 : #include "../../disco/keyguard/fd_keyguard_client.h"
13 : #include "../../disco/keyguard/fd_keyguard.h"
14 : #include "../../disco/keyguard/fd_keyload.h"
15 : #include "../../disco/keyguard/fd_keyswitch.h"
16 : #include "../../disco/metrics/fd_metrics.h"
17 : #include "../../disco/net/fd_net_tile.h"
18 : #include "../../disco/shred/fd_shred_tile.h"
19 : #include "../../disco/store/fd_store.h"
20 : #include "../../disco/topo/fd_topo.h"
21 : #include "../../util/pod/fd_pod_format.h"
22 : #include "../../flamenco/gossip/fd_gossip_message.h"
23 : #include "../../util/net/fd_net_headers.h"
24 :
25 : #include <linux/futex.h>
26 : #include "generated/fd_rserve_tile_seccomp.h"
27 :
28 0 : #define IN_KIND_NET (0)
29 0 : #define IN_KIND_SIGN (1)
30 0 : #define IN_KIND_SHRED (2)
31 0 : #define IN_KIND_ROTOR (3)
32 :
33 : #define MAX_IN_LINKS FD_TOPO_MAX_TILE_IN_LINKS
34 :
35 0 : #define FD_RSERVE_MAX_PACKET_SIZE 1232
36 :
37 : /* The maximum number of parent slots to look for. */
38 0 : #define FD_RSERVE_MAX_ORPHAN_SLOTS 11
39 :
40 : /* 10 minutes in milliseconds. */
41 0 : #define FD_RSERVE_SIGNED_REPAIR_WINDOW (60L*10L*1000L)
42 :
43 : /* static map from request type to response metric array index */
44 : static uint response_metric_index[AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID + 1] = {
45 : [FD_REPAIR_KIND_PING] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_PING_IDX,
46 : [FD_REPAIR_KIND_SHRED] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_WINDOW_INDEX_IDX,
47 : [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_HIGHEST_WINDOW_INDEX_IDX,
48 : [FD_REPAIR_KIND_ORPHAN] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_ORPHAN_IDX,
49 : [AG_REPAIR_KIND_PARENT_FEC_COUNT] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_PARENT_FEC_SET_COUNT_IDX,
50 : [AG_REPAIR_KIND_FEC_ROOT] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_FEC_SET_ROOT_IDX,
51 : [AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID] = FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_WINDOW_INDEX_FOR_BLOCK_ID_IDX,
52 : };
53 :
54 : /* static map from request type to received request metric array index */
55 : static uint request_metric_index[AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID + 1] = {
56 : [FD_REPAIR_KIND_PONG] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_PONG_IDX,
57 : [FD_REPAIR_KIND_SHRED] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_WINDOW_INDEX_IDX,
58 : [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_HIGHEST_WINDOW_INDEX_IDX,
59 : [FD_REPAIR_KIND_ORPHAN] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_ORPHAN_IDX,
60 : [AG_REPAIR_KIND_PARENT_FEC_COUNT] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_PARENT_FEC_SET_COUNT_IDX,
61 : [AG_REPAIR_KIND_FEC_ROOT] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_FEC_SET_ROOT_IDX,
62 : [AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID] = FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_WINDOW_INDEX_FOR_BLOCK_ID_IDX,
63 : };
64 :
65 :
66 : typedef union {
67 : struct {
68 : fd_wksp_t * mem;
69 : ulong chunk0;
70 : ulong wmark;
71 : ulong mtu;
72 : };
73 : fd_net_rx_bounds_t net_rx;
74 : } in_ctx_t;
75 :
76 : typedef struct ctx {
77 : fd_net_rx_bounds_t net_rx;
78 :
79 : uint in_kind [ MAX_IN_LINKS ];
80 : in_ctx_t in_links[ MAX_IN_LINKS ];
81 :
82 : uint net_out_idx;
83 : fd_wksp_t * net_out_mem;
84 : ulong net_out_chunk0;
85 : ulong net_out_wmark;
86 : ulong net_out_chunk;
87 :
88 : ulong seed;
89 : uchar rserve_secret[ 32 ];
90 : fd_rserve_t * rserve;
91 : fd_blockdb_t * blockdb; /* NULL if disabled */
92 : fd_store_t * store;
93 : int disk_fd;
94 : ulong max_shreds_per_block;
95 :
96 : /* Used for verifying incoming requests, and signing outgoing responses. */
97 : fd_sha512_t sha512[1];
98 : fd_keyguard_client_t keyguard_client[1];
99 : fd_keyswitch_t * keyswitch;
100 : fd_pubkey_t identity_public_key;
101 : int halt_signing;
102 :
103 : fd_ip4_udp_hdrs_t serve_hdr[1];
104 : ushort net_id;
105 :
106 : uchar net_buf[ FD_NET_MTU ];
107 : fd_rotor_block_t rotor_buf[1];
108 :
109 : struct {
110 : ulong received_request_count[FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_CNT];
111 : ulong received_request_bytes;
112 : ulong received_malformed_count[FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_CNT];
113 :
114 : ulong send_pkt_cnt;
115 : ulong sent_pkt_types [FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_CNT];
116 : ulong sent_response_bytes;
117 :
118 : ulong unanswered [FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_CNT];
119 : ulong dropped [FD_METRICS_ENUM_REQUEST_DROP_REASON_CNT];
120 : ulong disk_read[FD_METRICS_ENUM_DISK_READ_RESULT_CNT];
121 : ulong disk_read_bytes;
122 : ulong shreds_current;
123 : ulong disk_current_bytes;
124 : ulong disk_allocated_bytes;
125 : fd_histf_t disk_write_timing[1];
126 : ulong disk_inserted;
127 : ulong disk_write_failed;
128 : ulong disk_write_bytes;
129 : fd_histf_t fec_preevict_timing[1];
130 : ulong fec_preevict_write_cnt;
131 : ulong fec_preevict_write_bytes;
132 :
133 : ulong ping_cache_entries;
134 : ulong ping_cache_capacity;
135 : ulong ping_cache_evictions;
136 : } metrics[ 1 ];
137 : } ctx_t;
138 :
139 : FD_FN_CONST static inline ulong
140 0 : scratch_align( void ) {
141 0 : return 128UL;
142 0 : }
143 :
144 : FD_FN_PURE static inline ulong
145 0 : scratch_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
146 0 : ulong l = FD_LAYOUT_INIT;
147 0 : l = FD_LAYOUT_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
148 0 : l = FD_LAYOUT_APPEND( l, fd_rserve_align(), fd_rserve_footprint( tile->rserve.ping_cache_entries) );
149 0 : if( FD_LIKELY( tile->rserve.blockdb_max ) ) {
150 0 : l = FD_LAYOUT_APPEND( l, fd_blockdb_align(), fd_blockdb_footprint( tile->rserve.blockdb_max ) );
151 0 : }
152 0 : return FD_LAYOUT_FINI( l, scratch_align() );
153 0 : }
154 :
155 : static void
156 : send_packet( ctx_t * ctx,
157 : fd_stem_context_t * stem,
158 : uint dst_ip_addr,
159 : ushort dst_port,
160 : uint src_ip_addr,
161 : uchar const * payload,
162 : ulong payload_sz,
163 0 : ulong tsorig ) {
164 0 : ctx->metrics->send_pkt_cnt++;
165 0 : ctx->metrics->sent_response_bytes += payload_sz;
166 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
167 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
168 0 : *hdr = *ctx->serve_hdr;
169 :
170 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
171 0 : ip4->saddr = src_ip_addr;
172 0 : ip4->daddr = dst_ip_addr;
173 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
174 0 : ip4->check = 0U;
175 0 : ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
176 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
177 :
178 0 : fd_udp_hdr_t * udp = hdr->udp;
179 0 : udp->net_dport = dst_port;
180 0 : udp->net_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
181 0 : fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
182 0 : hdr->udp->check = 0U;
183 :
184 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
185 0 : ulong sig = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
186 0 : ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
187 0 : ulong chunk = ctx->net_out_chunk;
188 0 : fd_stem_publish( stem, ctx->net_out_idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
189 0 : ctx->net_out_chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
190 0 : }
191 :
192 : static inline void
193 : handle_pong( ctx_t * ctx,
194 : uchar const * payload,
195 : ulong payload_sz,
196 : uint saddr,
197 0 : ushort sport ) {
198 0 : if( FD_UNLIKELY( payload_sz!=sizeof(uint)+sizeof(fd_repair_pong_t) ) ) return;
199 0 : fd_repair_msg_t const * msg = (fd_repair_msg_t const *)fd_type_pun_const( payload );
200 0 : fd_repair_pong_t const * request = &msg->pong;
201 :
202 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &request->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( request->hash.uc, 32UL, request->sig, request->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, request->hash.uc, &request->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 = request->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 0 : ulong msg_sz = payload_sz-4UL;
267 :
268 0 : if( FD_UNLIKELY( tag==FD_REPAIR_KIND_PONG ) ) {
269 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
270 0 : ctx->metrics->received_request_bytes += payload_sz;
271 0 : handle_pong( ctx, payload, payload_sz, ip4->saddr, udp->net_sport );
272 0 : return;
273 0 : }
274 : /* Block id requests need the blockdb, which only exists when
275 : alpenglow is enabled. */
276 0 : int is_bid = tag==AG_REPAIR_KIND_PARENT_FEC_COUNT || tag==AG_REPAIR_KIND_FEC_ROOT || tag==AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID;
277 0 : if( FD_UNLIKELY( tag!=FD_REPAIR_KIND_SHRED &&
278 0 : tag!=FD_REPAIR_KIND_HIGHEST_SHRED &&
279 0 : tag!=FD_REPAIR_KIND_ORPHAN &&
280 0 : !( is_bid && ctx->blockdb ) ) ) {
281 0 : if( tag==FD_REPAIR_KIND_PING ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNSUPPORTED_PING_IDX ]++;
282 0 : else if( tag==FD_REPAIR_KIND_ANCESTOR_HASHES ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNSUPPORTED_ANCESTOR_HASHES_IDX ]++;
283 0 : else ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_UNKNOWN_TAG_IDX ]++;
284 0 : return;
285 0 : }
286 :
287 : /* Validate exact message size for each request type. We must check
288 : this before constructing the signable payload to avoid OOB reads. */
289 0 : ulong expected_sz;
290 0 : switch( tag ) {
291 0 : case FD_REPAIR_KIND_ORPHAN: expected_sz = sizeof(fd_repair_orphan_req_t); break;
292 0 : case AG_REPAIR_KIND_PARENT_FEC_COUNT: expected_sz = sizeof(ag_repair_parent_fec_count_req_t); break;
293 0 : case AG_REPAIR_KIND_FEC_ROOT: expected_sz = sizeof(ag_repair_fec_root_req_t); break;
294 0 : case AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID: expected_sz = sizeof(ag_repair_shred_block_id_req_t); break;
295 0 : default: expected_sz = sizeof(fd_repair_shred_req_t); break;
296 0 : }
297 0 : if( FD_UNLIKELY( msg_sz!=expected_sz ) ) {
298 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_REQUEST_MALFORMED_REASON_V_WRONG_SIZE_IDX ]++;
299 0 : return;
300 0 : }
301 :
302 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
303 0 : ctx->metrics->received_request_bytes += payload_sz;
304 :
305 0 : fd_repair_req_header_t header[1];
306 0 : memcpy( header, payload+4UL, sizeof(fd_repair_req_header_t) );
307 :
308 0 : if( FD_UNLIKELY( !fd_pubkey_eq( &ctx->identity_public_key, &header->to ) ) ) {
309 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_WRONG_RECIPIENT_IDX ]++;
310 0 : return;
311 0 : }
312 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &header->from ) ) ) {
313 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_FROM_SELF_IDX ]++;
314 0 : return;
315 0 : }
316 :
317 0 : long current = FD_NANOSEC_TO_MILLI( fd_log_wallclock() );
318 0 : long ts_diff = current - (long)header->ts;
319 0 : if( FD_UNLIKELY( ts_diff < 0L ) ) ts_diff = -ts_diff;
320 0 : if( FD_UNLIKELY( ts_diff > FD_RSERVE_SIGNED_REPAIR_WINDOW ) ) {
321 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_TIMESTAMP_SKEW_IDX ]++;
322 0 : return;
323 0 : }
324 :
325 : /* Most requests are for shreds we never stored, probe before doing
326 : a full ed25519 verify. */
327 0 : if( FD_LIKELY( tag==FD_REPAIR_KIND_SHRED || tag==FD_REPAIR_KIND_HIGHEST_SHRED ) ) {
328 0 : fd_repair_shred_req_t msg[1];
329 0 : memcpy( msg, payload+4UL, sizeof(fd_repair_shred_req_t) );
330 0 : if( FD_UNLIKELY( msg->shred_idx>=ctx->max_shreds_per_block ) ) {
331 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
332 0 : return;
333 0 : }
334 0 : if( FD_LIKELY( !fd_store_disk_probe( ctx->store, msg->slot, tag==FD_REPAIR_KIND_SHRED ? (uint)msg->shred_idx : UINT_MAX ) ) ) {
335 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
336 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
337 0 : return;
338 0 : }
339 0 : }
340 :
341 : /* Verify the signature. */
342 :
343 : /* The signed bytes are the tag followed by everything after the
344 : signature. */
345 0 : uchar signable[ FD_REPAIR_MAX_PREIMAGE_SZ ];
346 0 : ulong signable_sz = payload_sz - sizeof(fd_ed25519_sig_t);
347 0 : FD_TEST( signable_sz<=sizeof(signable) );
348 0 : fd_memcpy( signable, payload, 4 );
349 0 : fd_memcpy( signable+4UL, payload+68UL, signable_sz-4 );
350 :
351 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS!=fd_ed25519_verify( signable, signable_sz, header->sig, header->from.uc, ctx->sha512 ) ) ) {
352 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SIGNATURE_FAILED_IDX ]++;
353 0 : return;
354 0 : }
355 :
356 : /* Check whether we've heard a pong response from this peer at this
357 : exact source address. Keying on (pubkey, address) means a peer that
358 : ponged from one address cannot have repair responses redirected to
359 : a spoofed source address. */
360 0 : ping_cache_key_t key[1];
361 0 : memset( key, 0, sizeof(ping_cache_key_t) );
362 0 : key->pubkey = header->from;
363 0 : key->ip4 = ip4->saddr;
364 0 : key->port = udp->net_sport;
365 0 : ping_cache_entry_t * entry = ping_map_ele_query( ctx->rserve->ping_map, key, NULL, ctx->rserve->ping_pool );
366 0 : if( FD_LIKELY( entry ) ) {
367 0 : switch( tag ) {
368 0 : case FD_REPAIR_KIND_SHRED:
369 0 : case FD_REPAIR_KIND_HIGHEST_SHRED: {
370 0 : fd_repair_shred_req_t msg[1];
371 0 : memcpy( msg, payload+4UL, sizeof(fd_repair_shred_req_t) );
372 :
373 0 : ulong slot = msg->slot;
374 0 : ulong shred_idx = msg->shred_idx;
375 :
376 0 : if( FD_UNLIKELY( shred_idx>=ctx->max_shreds_per_block ) ) {
377 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
378 0 : return;
379 0 : }
380 :
381 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
382 0 : int len;
383 0 : for( ulong retry=0UL;; retry++ ) {
384 0 : if( tag==FD_REPAIR_KIND_SHRED ) {
385 0 : len = fd_store_disk_query( ctx->store, ctx->disk_fd, slot, (uint)shred_idx, payload );
386 0 : } else {
387 0 : len = fd_store_disk_query_highest( ctx->store, ctx->disk_fd, slot, (uint)shred_idx, payload );
388 0 : }
389 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
390 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
391 0 : }
392 0 : if( FD_UNLIKELY( len<0 ) ) {
393 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
394 0 : else if( len==FD_STORE_DISK_QUERY_SCAN_LIMIT) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SCAN_LIMIT_IDX ]++;
395 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
396 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
397 0 : return;
398 0 : }
399 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
400 0 : ctx->metrics->disk_read_bytes += (ulong)len;
401 :
402 0 : fd_memcpy( payload+len, &header->nonce, sizeof(uint) );
403 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
404 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
405 0 : return;
406 0 : }
407 0 : case FD_REPAIR_KIND_ORPHAN: {
408 : /* Orphan repair works by giving us a "root" slot to start at,
409 : and has us walk back through the parent slots sending the
410 : highest shred of each slot.
411 : We may send up to FD_RSERVE_MAX_ORPHAN_SLOTS of these shreds
412 : (including the root one). */
413 0 : fd_repair_orphan_req_t msg[1];
414 0 : memcpy( msg, payload+4UL, sizeof(fd_repair_orphan_req_t) );
415 :
416 0 : ulong current = msg->slot;
417 0 : for( uint i=0; i<FD_RSERVE_MAX_ORPHAN_SLOTS; i++ ) {
418 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
419 0 : int len;
420 0 : for( ulong retry=0UL;; retry++ ) {
421 0 : len = fd_store_disk_query_highest( ctx->store, ctx->disk_fd, current, 0, payload );
422 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
423 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
424 0 : }
425 0 : if( FD_UNLIKELY( len<0 ) ) {
426 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
427 0 : else if( len==FD_STORE_DISK_QUERY_SCAN_LIMIT) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SCAN_LIMIT_IDX ]++;
428 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
429 0 : ctx->metrics->unanswered[ FD_METRICS_ENUM_REPAIR_REQUEST_TYPE_V_ORPHAN_IDX ]++;
430 0 : return;
431 0 : }
432 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
433 0 : ctx->metrics->disk_read_bytes += (ulong)len;
434 0 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( payload );
435 0 : memcpy( payload+len, &header->nonce, sizeof(uint) );
436 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
437 0 : ctx->metrics->sent_pkt_types[ FD_METRICS_ENUM_REPAIR_RESPONSE_TYPE_V_ORPHAN_IDX ]++;
438 0 : ushort parent_off = shred->data.parent_off;
439 : /* fd_shred_parse ensures that parent_off will be 0 if and only if slot is 0 */
440 0 : if( FD_UNLIKELY( parent_off==0 ) ) break;
441 0 : current = current - parent_off;
442 0 : }
443 0 : break;
444 0 : }
445 0 : case AG_REPAIR_KIND_PARENT_FEC_COUNT: {
446 : /* Serves the block's parent, FEC set count, and the proof of its
447 : parent-info leaf, which follows the fec_set_cnt FEC root leaves. */
448 0 : ag_repair_parent_fec_count_req_t msg[1];
449 0 : memcpy( msg, payload+4UL, sizeof(ag_repair_parent_fec_count_req_t) );
450 :
451 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
452 0 : if( FD_UNLIKELY( !blk ) ) {
453 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
454 0 : return;
455 0 : }
456 :
457 0 : uchar proof[ FD_BLOCKDB_PROOF_NODE_MAX*FD_SHRED_MERKLE_NODE_SZ ];
458 0 : int proof_len = fd_blockdb_proof( ctx->blockdb, blk, blk->fec_set_cnt, proof );
459 0 : FD_TEST( proof_len>=0 );
460 :
461 0 : uchar buf[ AG_REPAIR_RESPONSE_MAX_SZ ];
462 0 : ulong sz = ag_repair_parent_fec_count_ser( buf, sizeof(buf), blk->fec_set_cnt, blk->parent_slot, &blk->parent_block_id,
463 0 : proof, (ulong)proof_len, header->nonce );
464 0 : FD_TEST( sz );
465 :
466 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, buf, sz, fd_frag_meta_ts_comp( fd_tickcount() ) );
467 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
468 0 : return;
469 0 : }
470 0 : case AG_REPAIR_KIND_FEC_ROOT: {
471 : /* Serves the merkle root prefix of FEC set fec_set_idx and the
472 : proof of its leaf. */
473 0 : ag_repair_fec_root_req_t msg[1];
474 0 : memcpy( msg, payload+4UL, sizeof(ag_repair_fec_root_req_t) );
475 :
476 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
477 0 : if( FD_UNLIKELY( !blk ) ) {
478 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
479 0 : return;
480 0 : }
481 0 : if( FD_UNLIKELY( msg->fec_set_idx%FD_FEC_SHRED_CNT || msg->fec_set_idx/FD_FEC_SHRED_CNT>=blk->fec_set_cnt ) ) {
482 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_FEC_SET_INDEX_INVALID_IDX ]++;
483 0 : return;
484 0 : }
485 0 : ulong leaf_idx = msg->fec_set_idx/FD_FEC_SHRED_CNT;
486 :
487 0 : uchar proof[ FD_BLOCKDB_PROOF_NODE_MAX*FD_SHRED_MERKLE_NODE_SZ ];
488 0 : int proof_len = fd_blockdb_proof( ctx->blockdb, blk, leaf_idx, proof );
489 0 : FD_TEST( proof_len>=0 );
490 :
491 0 : uchar buf[ AG_REPAIR_RESPONSE_MAX_SZ ];
492 0 : ulong sz = ag_repair_fec_set_root_ser( buf, sizeof(buf), blk->merkle_roots[ leaf_idx ], proof, (ulong)proof_len, header->nonce );
493 0 : FD_TEST( sz );
494 :
495 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, buf, sz, fd_frag_meta_ts_comp( fd_tickcount() ) );
496 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
497 0 : return;
498 0 : }
499 0 : case AG_REPAIR_KIND_SHRED_FOR_BLOCK_ID: {
500 : /* Serves the data shred at shred_idx of block (slot, block_id),
501 : found by its FEC set's merkle root. */
502 0 : ag_repair_shred_block_id_req_t msg[1];
503 0 : memcpy( msg, payload+4UL, sizeof(ag_repair_shred_block_id_req_t) );
504 :
505 0 : fd_blockdb_blk_t const * blk = fd_blockdb_query( ctx->blockdb, msg->slot, &msg->block_id );
506 0 : if( FD_UNLIKELY( !blk ) ) {
507 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
508 0 : return;
509 0 : }
510 0 : if( FD_UNLIKELY( (ulong)msg->shred_idx>=(ulong)blk->fec_set_cnt*FD_FEC_SHRED_CNT ) ) {
511 0 : ctx->metrics->dropped[ FD_METRICS_ENUM_REQUEST_DROP_REASON_V_SHRED_INDEX_INVALID_IDX ]++;
512 0 : return;
513 0 : }
514 :
515 0 : uchar const * root = blk->merkle_roots[ msg->shred_idx/FD_FEC_SHRED_CNT ];
516 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
517 0 : int len;
518 0 : for( ulong retry=0UL;; retry++ ) {
519 0 : len = fd_store_disk_query_root( ctx->store, ctx->disk_fd, root, msg->shred_idx, payload );
520 0 : if( FD_LIKELY( len!=FD_STORE_DISK_QUERY_BUSY || retry==7UL ) ) break;
521 0 : for( ulong pause=0UL; pause<(1UL<<retry); pause++ ) FD_SPIN_PAUSE();
522 0 : }
523 0 : if( FD_UNLIKELY( len<0 ) ) {
524 0 : if( len==FD_STORE_DISK_QUERY_BUSY ) ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_BUSY_IDX ]++;
525 0 : else ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_MISS_IDX ]++;
526 0 : ctx->metrics->unanswered[ request_metric_index[tag] ]++;
527 0 : return;
528 0 : }
529 0 : ctx->metrics->disk_read[ FD_METRICS_ENUM_DISK_READ_RESULT_V_SUCCESS_IDX ]++;
530 0 : ctx->metrics->disk_read_bytes += (ulong)len;
531 :
532 0 : memcpy( payload+len, &header->nonce, sizeof(uint) );
533 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
534 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
535 0 : return;
536 0 : }
537 0 : }
538 0 : } else {
539 : /* Derive a ping token bound to this peer's pubkey and the source
540 : address the request arrived from. The peer can only produce a
541 : valid pong if it actually receives this ping at that address. */
542 0 : uchar token[ 32UL ];
543 0 : fd_rserve_ping_token( ctx->rserve, token, &header->from, ip4->saddr, udp->net_sport );
544 :
545 : /* Sign the token. */
546 0 : uchar signature[ 64UL ];
547 0 : fd_keyguard_client_sign( ctx->keyguard_client, signature, token, 32UL, FD_KEYGUARD_SIGN_TYPE_ED25519 );
548 :
549 : /* Like Agave, block id metadata requests are answered with the
550 : BlockIdRepairResponse ping variant, and ShredForBlockId, like the
551 : legacy requests, with the RepairResponse one. */
552 0 : fd_repair_ping_t msg[ 1 ];
553 0 : msg->kind = (tag==AG_REPAIR_KIND_PARENT_FEC_COUNT || tag==AG_REPAIR_KIND_FEC_ROOT) ? AG_REPAIR_KIND_PING : FD_REPAIR_KIND_PING;
554 0 : msg->ping.from = ctx->identity_public_key;
555 0 : memcpy( msg->ping.sig, signature, 64 );
556 0 : memcpy( msg->ping.hash.uc, token, 32 );
557 :
558 : /* Send the ping packet back to the source. */
559 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, (uchar const *)fd_type_pun_const( msg ), sizeof(fd_repair_ping_t), 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_ROTOR: {
590 0 : if( FD_UNLIKELY( sig!=ROTOR_SIG_BLOCK ) ) break;
591 0 : if( FD_UNLIKELY( sz<FD_ROTOR_BLOCK_SZ( 0 ) || sz>sizeof(fd_rotor_block_t) || chunk<in_ctx->chunk0 || chunk>in_ctx->wmark ) )
592 0 : FD_LOG_ERR(( "rotor_rserve chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
593 0 : fd_memcpy( ctx->rotor_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_ROTOR ) ) {
611 0 : if( FD_UNLIKELY( sig!=ROTOR_SIG_BLOCK ) ) return;
612 0 : fd_rotor_block_t const * msg = ctx->rotor_buf;
613 0 : if( FD_UNLIKELY( sz!=FD_ROTOR_BLOCK_SZ( msg->fec_set_cnt ) ) )
614 0 : FD_LOG_ERR(( "rotor_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_ROTOR: 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, "rotor_rserve" ) ) {
876 0 : if( FD_UNLIKELY( !ctx->blockdb ) ) FD_LOG_ERR(( "rserve has a rotor_rserve link but blockdb_max is 0" ));
877 0 : ctx->in_kind[ in_idx ] = IN_KIND_ROTOR;
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 : };
|
