Line data Source code
1 : /* fd_rserve_tile serves incoming repair requests from other nodes */
2 :
3 : #define _GNU_SOURCE
4 : #include "fd_rserve.h"
5 : #include "fd_repair.h"
6 : #include "../../disco/fd_disco_base.h"
7 : #include "../../disco/keyguard/fd_keyguard_client.h"
8 : #include "../../disco/keyguard/fd_keyguard.h"
9 : #include "../../disco/keyguard/fd_keyload.h"
10 : #include "../../disco/keyguard/fd_keyswitch.h"
11 : #include "../../disco/metrics/fd_metrics.h"
12 : #include "../../disco/net/fd_net_tile.h"
13 : #include "../../disco/shred/fd_shred_tile.h"
14 : #include "../../disco/store/fd_shredb.h"
15 : #include "../../disco/topo/fd_topo.h"
16 : #include "../../flamenco/gossip/fd_gossip_message.h"
17 : #include "../../util/net/fd_net_headers.h"
18 :
19 : #include "generated/fd_rserve_tile_seccomp.h"
20 :
21 0 : #define IN_KIND_NET (0)
22 0 : #define IN_KIND_SHRED (1)
23 0 : #define IN_KIND_SIGN (2)
24 :
25 : #define MAX_IN_LINKS 32
26 :
27 0 : #define FD_RSERVE_MAX_PACKET_SIZE 1232
28 :
29 : /* The maximum number of parent slots to look for. */
30 0 : #define FD_RSERVE_MAX_ORPHAN_SLOTS 11
31 :
32 : /* 10 minutes in milliseconds. */
33 : #define FD_RSERVE_SIGNED_REPAIR_WINDOW (60L*10L*1000L)
34 :
35 : /* static map from request type to response metric array index */
36 : static uint response_metric_index[FD_REPAIR_KIND_ORPHAN + 1] = {
37 : [FD_REPAIR_KIND_PING] = FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_PING_IDX,
38 : [FD_REPAIR_KIND_SHRED] = FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_WINDOW_IDX,
39 : [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_HIGHEST_WINDOW_IDX,
40 : [FD_REPAIR_KIND_ORPHAN] = FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_ORPHAN_IDX,
41 : };
42 :
43 : /* static map from request type to received request metric array index */
44 : static uint request_metric_index[FD_REPAIR_KIND_ORPHAN + 1] = {
45 : [FD_REPAIR_KIND_PONG] = FD_METRICS_ENUM_RSERVE_REQUEST_TYPES_V_PONG_IDX,
46 : [FD_REPAIR_KIND_SHRED] = FD_METRICS_ENUM_RSERVE_REQUEST_TYPES_V_WINDOW_INDEX_IDX,
47 : [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_RSERVE_REQUEST_TYPES_V_HIGHEST_WINDOW_INDEX_IDX,
48 : [FD_REPAIR_KIND_ORPHAN] = FD_METRICS_ENUM_RSERVE_REQUEST_TYPES_V_ORPHAN_IDX,
49 : };
50 :
51 :
52 : typedef union {
53 : struct {
54 : fd_wksp_t * mem;
55 : ulong chunk0;
56 : ulong wmark;
57 : ulong mtu;
58 : };
59 : fd_net_rx_bounds_t net_rx;
60 : } in_ctx_t;
61 :
62 : typedef struct ctx {
63 : fd_net_rx_bounds_t net_rx;
64 :
65 : uint in_kind [ MAX_IN_LINKS ];
66 : in_ctx_t in_links[ MAX_IN_LINKS ];
67 :
68 : uint net_out_idx;
69 : fd_wksp_t * net_out_mem;
70 : ulong net_out_chunk0;
71 : ulong net_out_wmark;
72 : ulong net_out_chunk;
73 :
74 : ulong seed;
75 : fd_rserve_t * rserve;
76 : fd_shredb_t * shredb;
77 :
78 : /* Used for verifying incoming requests, and signing outgoing responses. */
79 : fd_sha512_t sha512[1];
80 : fd_keyguard_client_t keyguard_client[1];
81 : fd_keyswitch_t * keyswitch;
82 : fd_pubkey_t identity_public_key;
83 : int halt_signing;
84 :
85 : fd_ip4_udp_hdrs_t serve_hdr[1];
86 : ushort net_id;
87 :
88 : struct {
89 : ulong received_request_count[FD_METRICS_ENUM_RSERVE_REQUEST_TYPES_CNT];
90 : ulong received_request_bytes;
91 : ulong received_malformed_count[FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_CNT];
92 :
93 : ulong send_pkt_cnt;
94 : ulong sent_pkt_types [FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_CNT];
95 : ulong sent_response_bytes;
96 :
97 : ulong missed_pkt_types[FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_CNT];
98 : ulong fail_sigverify_request;
99 : ulong fail_own_key;
100 : ulong fail_not_for_us;
101 : ulong fail_invalid_token;
102 : ulong fail_outdated;
103 : ulong fail_invalid_shred_idx;
104 : ulong fail_ping_cache_lookup;
105 :
106 : ulong shreds_current;
107 : ulong disk_allocated_bytes;
108 :
109 : ulong ping_cache_entries;
110 : ulong ping_cache_evictions;
111 : } metrics[ 1 ];
112 : } ctx_t;
113 :
114 : FD_FN_CONST static inline ulong
115 0 : scratch_align( void ) {
116 0 : return 128UL;
117 0 : }
118 :
119 : FD_FN_PURE static inline ulong
120 0 : scratch_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
121 0 : ulong l = FD_LAYOUT_INIT;
122 0 : l = FD_LAYOUT_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
123 0 : l = FD_LAYOUT_APPEND( l, fd_shredb_align(), fd_shredb_footprint( tile->rserve.shred_storage_limit_gib ) );
124 0 : l = FD_LAYOUT_APPEND( l, fd_rserve_align(), fd_rserve_footprint( tile->rserve.ping_cache_entries) );
125 0 : return FD_LAYOUT_FINI( l, scratch_align() );
126 0 : }
127 :
128 : static void
129 : send_packet( ctx_t * ctx,
130 : fd_stem_context_t * stem,
131 : uint dst_ip_addr,
132 : ushort dst_port,
133 : uint src_ip_addr,
134 : uchar const * payload,
135 : ulong payload_sz,
136 0 : ulong tsorig ) {
137 0 : ctx->metrics->send_pkt_cnt++;
138 0 : ctx->metrics->sent_response_bytes += payload_sz;
139 0 : uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
140 0 : fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
141 0 : *hdr = *ctx->serve_hdr;
142 :
143 0 : fd_ip4_hdr_t * ip4 = hdr->ip4;
144 0 : ip4->saddr = src_ip_addr;
145 0 : ip4->daddr = dst_ip_addr;
146 0 : ip4->net_id = fd_ushort_bswap( ctx->net_id++ );
147 0 : ip4->check = 0U;
148 0 : ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
149 0 : ip4->check = fd_ip4_hdr_check_fast( ip4 );
150 :
151 0 : fd_udp_hdr_t * udp = hdr->udp;
152 0 : udp->net_dport = dst_port;
153 0 : udp->net_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
154 0 : fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
155 0 : hdr->udp->check = 0U;
156 :
157 0 : ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
158 0 : ulong sig = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
159 0 : ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
160 0 : ulong chunk = ctx->net_out_chunk;
161 0 : fd_stem_publish( stem, ctx->net_out_idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
162 0 : ctx->net_out_chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
163 0 : }
164 :
165 : static inline void
166 : handle_pong( ctx_t * ctx,
167 : uchar const * payload,
168 0 : ulong payload_sz ) {
169 0 : if( FD_UNLIKELY( payload_sz!=sizeof(uint)+sizeof(fd_repair_pong_t) ) ) return;
170 0 : fd_repair_msg_t const * msg = (fd_repair_msg_t const *)fd_type_pun_const( payload );
171 0 : fd_repair_pong_t const * request = &msg->pong;
172 :
173 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &request->from ) ) ) {
174 : /* We've received our own repair request, ignore. */
175 0 : ctx->metrics->fail_own_key++;
176 0 : return;
177 0 : }
178 :
179 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS!=fd_ed25519_verify( request->hash.uc, 32UL, request->sig, request->from.uc, ctx->sha512 ) ) ) {
180 : /* Invalid signature, ignore. */
181 0 : ctx->metrics->fail_sigverify_request++;
182 0 : return;
183 0 : }
184 :
185 : /* Verify that the pong hash corresponds to either our current or
186 : previous rotating token. This prevents replayed or stale pongs. */
187 0 : if( FD_UNLIKELY( !fd_rserve_pong_token_verify( ctx->rserve, request->hash.uc ) ) ) {
188 0 : ctx->metrics->fail_invalid_token++;
189 0 : return;
190 0 : }
191 :
192 0 : fd_rserve_t * rserve = ctx->rserve;
193 0 : ping_cache_entry_t * entry = ping_map_ele_query( rserve->ping_map, &request->from, NULL, rserve->ping_pool );
194 :
195 0 : if( FD_LIKELY( !entry ) ) {
196 : /* New entry, evict LRU if pool is full. */
197 0 : if( FD_UNLIKELY( !ping_pool_free( rserve->ping_pool ) ) ) {
198 0 : ping_cache_entry_t * victim = ping_dlist_ele_pop_head( rserve->ping_dlist, rserve->ping_pool );
199 0 : ping_map_ele_remove_fast( rserve->ping_map, victim, rserve->ping_pool );
200 0 : ping_pool_ele_release( rserve->ping_pool, victim );
201 0 : ctx->metrics->ping_cache_entries--;
202 0 : ctx->metrics->ping_cache_evictions++;
203 0 : }
204 0 : entry = ping_pool_ele_acquire( rserve->ping_pool );
205 0 : entry->addr = request->from;
206 0 : ping_map_ele_insert( rserve->ping_map, entry, rserve->ping_pool );
207 0 : ctx->metrics->ping_cache_entries++;
208 0 : } else {
209 : /* Existing entry, move to tail. */
210 0 : ping_dlist_ele_remove( rserve->ping_dlist, entry, rserve->ping_pool );
211 0 : }
212 :
213 0 : entry->timestamp = (ulong)fd_log_wallclock();
214 0 : ping_dlist_ele_push_tail( rserve->ping_dlist, entry, rserve->ping_pool );
215 0 : return;
216 0 : }
217 :
218 : static inline void
219 : handle_net_request( ctx_t * ctx,
220 : fd_stem_context_t * stem,
221 : uchar const * payload,
222 : ulong payload_sz,
223 : fd_udp_hdr_t * udp,
224 0 : fd_ip4_hdr_t * ip4 ) {
225 0 : if( FD_UNLIKELY( payload_sz<4UL ) ) {
226 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_TOO_SMALL_IDX ]++;
227 0 : return;
228 0 : }
229 0 : uint tag = FD_LOAD( uint, payload );
230 0 : ulong msg_sz = payload_sz-4UL;
231 :
232 0 : if( FD_UNLIKELY( tag==FD_REPAIR_KIND_PONG ) ) {
233 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
234 0 : ctx->metrics->received_request_bytes += payload_sz;
235 0 : handle_pong( ctx, payload, payload_sz );
236 0 : return;
237 0 : }
238 0 : if( FD_UNLIKELY( tag!=FD_REPAIR_KIND_SHRED &&
239 0 : tag!=FD_REPAIR_KIND_HIGHEST_SHRED &&
240 0 : tag!=FD_REPAIR_KIND_ORPHAN ) ) {
241 0 : if( tag==FD_REPAIR_KIND_PING ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_PING_IDX ]++;
242 0 : else if( tag==FD_REPAIR_KIND_ANCESTOR_HASHES ) ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_ANCESTOR_HASHES_IDX ]++;
243 0 : else ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_UNKNOWN_TAG_IDX ]++;
244 0 : return;
245 0 : }
246 :
247 : /* Validate exact message size for each request type. We must check
248 : this before constructing the signable payload to avoid OOB reads. */
249 0 : if( FD_UNLIKELY( tag==FD_REPAIR_KIND_ORPHAN ) ) {
250 0 : if( FD_UNLIKELY( msg_sz!=sizeof(fd_repair_orphan_req_t) ) ) {
251 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_WRONG_SIZE_IDX ]++;
252 0 : return;
253 0 : }
254 0 : } else {
255 0 : if( FD_UNLIKELY( msg_sz!=sizeof(fd_repair_shred_req_t) ) ) {
256 0 : ctx->metrics->received_malformed_count[ FD_METRICS_ENUM_RSERVE_MALFORMED_TYPES_V_WRONG_SIZE_IDX ]++;
257 0 : return;
258 0 : }
259 0 : }
260 :
261 0 : ctx->metrics->received_request_count[ request_metric_index[tag] ]++;
262 0 : ctx->metrics->received_request_bytes += payload_sz;
263 :
264 0 : fd_repair_req_header_t header[1];
265 0 : memcpy( header, payload+4UL, sizeof(fd_repair_req_header_t) );
266 :
267 0 : if( FD_UNLIKELY( !fd_pubkey_eq( &ctx->identity_public_key, &header->to ) ) ) {
268 0 : ctx->metrics->fail_not_for_us++;
269 0 : return;
270 0 : }
271 0 : if( FD_UNLIKELY( fd_pubkey_eq( &ctx->identity_public_key, &header->from ) ) ) {
272 0 : ctx->metrics->fail_own_key++;
273 0 : return;
274 0 : }
275 :
276 0 : long current = FD_NANOSEC_TO_MILLI( fd_log_wallclock() );
277 0 : long ts_diff = current - (long)header->ts;
278 0 : if( FD_UNLIKELY( ts_diff < 0L ) ) ts_diff = -ts_diff;
279 0 : if( FD_UNLIKELY( ts_diff > FD_RSERVE_SIGNED_REPAIR_WINDOW ) ) {
280 0 : ctx->metrics->fail_outdated++;
281 0 : return;
282 0 : }
283 :
284 : /* Verify the signature. */
285 :
286 : /* The largest signable payload size is 96 bytes, that being
287 : 160-64=96, as the signature itself is not included. */
288 0 : uchar signable[ 96 ];
289 0 : uchar signable_sz = tag==FD_REPAIR_KIND_ORPHAN ? 88 : 96;
290 0 : fd_memcpy( signable, payload, 4 );
291 0 : fd_memcpy( signable+4UL, payload+68UL, signable_sz-4 );
292 :
293 0 : if( FD_UNLIKELY( FD_ED25519_SUCCESS!=fd_ed25519_verify( signable, signable_sz, header->sig, header->from.uc, ctx->sha512 ) ) ) {
294 0 : ctx->metrics->fail_sigverify_request++;
295 0 : return;
296 0 : }
297 :
298 : /* Check whether we've heard a pong response from them. */
299 0 : ping_cache_entry_t * entry = ping_map_ele_query( ctx->rserve->ping_map, &header->from, NULL, ctx->rserve->ping_pool );
300 0 : if( FD_LIKELY( entry ) ) {
301 0 : switch( tag ) {
302 0 : case FD_REPAIR_KIND_SHRED:
303 0 : case FD_REPAIR_KIND_HIGHEST_SHRED: {
304 0 : fd_repair_shred_req_t msg[1];
305 0 : memcpy( msg, payload+4UL, sizeof(fd_repair_shred_req_t) );
306 :
307 0 : ulong slot = msg->slot;
308 0 : ulong shred_idx = msg->shred_idx;
309 :
310 0 : if( FD_UNLIKELY( (shred_idx & fd_ulong_mask( 15, 64 ))!=0 ) ) {
311 0 : ctx->metrics->fail_invalid_shred_idx++;
312 0 : return;
313 0 : }
314 :
315 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
316 0 : int len;
317 0 : if( tag==FD_REPAIR_KIND_SHRED ) {
318 0 : len = fd_shredb_query( ctx->shredb, slot, (uint)shred_idx, payload );
319 0 : } else {
320 0 : len = fd_shredb_query_highest( ctx->shredb, slot, (uint)shred_idx, payload );
321 0 : }
322 0 : if( FD_UNLIKELY( len<0 ) ) {
323 0 : ctx->metrics->missed_pkt_types[ response_metric_index[tag] ]++;
324 0 : return;
325 0 : }
326 :
327 0 : fd_memcpy( payload+len, &header->nonce, sizeof(uint) );
328 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
329 0 : ctx->metrics->sent_pkt_types[ response_metric_index[tag] ]++;
330 0 : return;
331 0 : }
332 0 : case FD_REPAIR_KIND_ORPHAN: {
333 : /* Orphan repair works by giving us a "root" slot to start at,
334 : and has us walk back through the parent slots sending the
335 : highest shred of each slot.
336 : We may send up to FD_RSERVE_MAX_ORPHAN_SLOTS of these shreds
337 : (including the root one). */
338 0 : fd_repair_orphan_req_t msg[1];
339 0 : memcpy( msg, payload+4UL, sizeof(fd_repair_orphan_req_t) );
340 :
341 0 : ulong current = msg->slot;
342 0 : for( uint i=0; i<FD_RSERVE_MAX_ORPHAN_SLOTS; i++ ) {
343 0 : uchar payload[ FD_SHRED_MAX_SZ+sizeof(uint) ];
344 0 : int len = fd_shredb_query_highest( ctx->shredb, current, 0, payload );
345 0 : if( FD_UNLIKELY( len<0 ) ) {
346 0 : ctx->metrics->missed_pkt_types[ FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_ORPHAN_IDX ]++;
347 0 : return;
348 0 : }
349 0 : fd_shred_t const * shred = (fd_shred_t const *)fd_type_pun_const( payload );
350 0 : memcpy( payload+len, &header->nonce, sizeof(uint) );
351 0 : send_packet( ctx, stem, ip4->saddr, udp->net_sport, ip4->daddr, payload, (ulong)len+sizeof(uint), fd_frag_meta_ts_comp( fd_tickcount() ) );
352 0 : ctx->metrics->sent_pkt_types[ FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_ORPHAN_IDX ]++;
353 0 : ushort parent_off = shred->data.parent_off;
354 : /* fd_shred_parse ensures that parent_off will be 0 if and only if slot is 0 */
355 0 : if( FD_UNLIKELY( parent_off==0 ) ) break;
356 0 : current = current - parent_off;
357 0 : }
358 0 : break;
359 0 : }
360 0 : }
361 0 : } else {
362 0 : ctx->metrics->fail_ping_cache_lookup++;
363 :
364 : /* Use the current rotating token. */
365 0 : uchar const * token = ctx->rserve->token_cur;
366 :
367 : /* Sign the token. */
368 0 : uchar signature[ 64UL ];
369 0 : fd_keyguard_client_sign( ctx->keyguard_client, signature, token, 32UL, FD_KEYGUARD_SIGN_TYPE_ED25519 );
370 :
371 0 : fd_repair_ping_t msg[ 1 ];
372 0 : msg->kind = FD_REPAIR_KIND_PING;
373 0 : msg->ping.from = ctx->identity_public_key;
374 0 : memcpy( msg->ping.sig, signature, 64 );
375 0 : memcpy( msg->ping.hash.uc, token, 32 );
376 :
377 : /* Send the ping packet back to the source. */
378 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() ) );
379 0 : ctx->metrics->sent_pkt_types[ FD_METRICS_ENUM_RSERVE_SENT_RESPONSE_TYPES_V_PING_IDX ]++;
380 0 : }
381 0 : }
382 :
383 : static inline void
384 : handle_shred( ctx_t * ctx,
385 : uchar const * payload,
386 0 : ulong sig ) {
387 0 : uint src = fd_shred_sig_src( sig );
388 :
389 0 : if( FD_UNLIKELY( src>=SHRED_SIG_FEC_EVICTED ) ) return;
390 :
391 0 : fd_shred_base_t const * msg = (fd_shred_base_t const *)fd_type_pun_const( payload );
392 0 : fd_shred_t const * shred = &msg->shred;
393 :
394 0 : if( FD_UNLIKELY( !(fd_shred_type( shred->variant ) & FD_SHRED_TYPEMASK_DATA) ) ) return;
395 :
396 0 : fd_shredb_insert( ctx->shredb, shred );
397 0 : ctx->metrics->shreds_current = ctx->shredb->cnt;
398 0 : }
399 :
400 : static inline int
401 : returnable_frag( ctx_t * ctx,
402 : ulong in_idx,
403 : ulong seq FD_PARAM_UNUSED,
404 : ulong sig,
405 : ulong chunk,
406 : ulong sz,
407 : ulong ctl,
408 : ulong tsorig FD_PARAM_UNUSED,
409 : ulong tspub FD_PARAM_UNUSED,
410 0 : fd_stem_context_t * stem ) {
411 0 : uint in_kind = ctx->in_kind[ in_idx ];
412 0 : in_ctx_t const * in_ctx = &ctx->in_links[ in_idx ];
413 :
414 0 : switch( in_kind ) {
415 0 : case IN_KIND_NET: {
416 0 : if( FD_UNLIKELY( ctx->halt_signing ) ) return 1;
417 0 : if( fd_disco_netmux_sig_proto( sig )!=DST_PROTO_RSERVE ) return 0;
418 :
419 0 : uchar const * buffer = fd_net_rx_translate_frag( &in_ctx->net_rx, chunk, ctl, sz );
420 0 : uchar * payload; ulong payload_sz;
421 0 : fd_udp_hdr_t * udp;
422 0 : fd_ip4_hdr_t * ip4;
423 0 : if( FD_UNLIKELY( !fd_ip4_udp_hdr_strip( buffer, sz, &payload, &payload_sz, NULL, &ip4, &udp ) ) ) {
424 0 : FD_LOG_WARNING(( "rserve: malformed packet (sz=%lu)", sz ));
425 0 : return 0;
426 0 : }
427 0 : handle_net_request( ctx, stem, payload, payload_sz, udp, ip4 );
428 0 : return 0;
429 0 : }
430 0 : case IN_KIND_SHRED: {
431 0 : if( FD_UNLIKELY( sz==0UL ) ) return 0;
432 0 : if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) )
433 0 : FD_LOG_ERR(( "chunk %lu %lu from in %u corrupt, not in range [%lu,%lu]", chunk, sz, in_kind, in_ctx->chunk0, in_ctx->wmark ));
434 :
435 0 : uchar const * buffer = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
436 0 : handle_shred( ctx, buffer, sig );
437 0 : return 0;
438 0 : }
439 0 : case IN_KIND_SIGN: return 0; /* handled internally by keyguard_client */
440 0 : default: FD_LOG_ERR(( "unexpected input kind (%u)", in_kind ));
441 0 : }
442 0 : }
443 :
444 : static inline void
445 0 : during_housekeeping( ctx_t * ctx ) {
446 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_UNHALT_PENDING ) ) {
447 0 : FD_LOG_DEBUG(( "keyswitch: unhalting" ));
448 0 : FD_DCHECK_CRIT( ctx->halt_signing, "state machine corruption" );
449 0 : fd_memcpy( ctx->identity_public_key.uc, ctx->keyswitch->bytes, sizeof(fd_pubkey_t) );
450 0 : ctx->halt_signing = 0;
451 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
452 0 : }
453 :
454 0 : if( FD_UNLIKELY( fd_keyswitch_state_query( ctx->keyswitch )==FD_KEYSWITCH_STATE_SWITCH_PENDING ) ) {
455 0 : ctx->halt_signing = 1;
456 0 : fd_keyswitch_state( ctx->keyswitch, FD_KEYSWITCH_STATE_COMPLETED );
457 0 : }
458 :
459 0 : fd_rserve_t * rserve = ctx->rserve;
460 0 : ulong now_ns = (ulong)fd_log_wallclock();
461 0 : if( FD_UNLIKELY( now_ns-rserve->last_rotate_ts > FD_RSERVE_TOKEN_ROTATE_NS ) ) {
462 0 : rserve->last_rotate_ts = now_ns;
463 0 : rserve->token_idx++;
464 0 : memcpy( rserve->token_prev, rserve->token_cur, 32UL );
465 0 : fd_rserve_derive_token( rserve->token_cur, rserve->seed, rserve->token_idx );
466 0 : }
467 :
468 : /* Evict expired entries from the head (oldest) of the LRU list. */
469 0 : while( !ping_dlist_is_empty( rserve->ping_dlist, rserve->ping_pool ) ) {
470 0 : ping_cache_entry_t * head = ping_dlist_ele_peek_head( rserve->ping_dlist, rserve->ping_pool );
471 0 : if( FD_LIKELY( now_ns-head->timestamp <= FD_RSERVE_PING_CACHE_TTL_NS ) ) break;
472 0 : ping_dlist_ele_pop_head( rserve->ping_dlist, rserve->ping_pool );
473 0 : ping_map_ele_remove_fast( rserve->ping_map, head, rserve->ping_pool );
474 0 : ping_pool_ele_release( rserve->ping_pool, head );
475 0 : ctx->metrics->ping_cache_entries--;
476 0 : }
477 0 : }
478 :
479 : static inline void
480 0 : metrics_write( ctx_t * ctx ) {
481 0 : FD_MCNT_ENUM_COPY( RSERVE, RECEIVED_REQUEST_COUNT, ctx->metrics->received_request_count );
482 0 : FD_MCNT_SET( RSERVE, RECEIVED_REQUEST_BYTES, ctx->metrics->received_request_bytes );
483 0 : FD_MCNT_ENUM_COPY( RSERVE, RECEIVED_MALFORMED_COUNT, ctx->metrics->received_malformed_count );
484 :
485 0 : FD_MCNT_SET( RSERVE, TOTAL_PKT_COUNT, ctx->metrics->send_pkt_cnt );
486 0 : FD_MCNT_ENUM_COPY( RSERVE, SENT_RESPONSE_TYPES, ctx->metrics->sent_pkt_types );
487 0 : FD_MCNT_SET( RSERVE, SENT_RESPONSE_BYTES, ctx->metrics->sent_response_bytes );
488 :
489 0 : FD_MCNT_ENUM_COPY( RSERVE, MISSED_RESPONSE_TYPES, ctx->metrics->missed_pkt_types );
490 0 : FD_MCNT_SET( RSERVE, FAILED_SIGVERIFY, ctx->metrics->fail_sigverify_request );
491 0 : FD_MCNT_SET( RSERVE, FAILED_OWN_KEY, ctx->metrics->fail_own_key );
492 0 : FD_MCNT_SET( RSERVE, FAILED_INVALID_TOKEN, ctx->metrics->fail_invalid_token );
493 0 : FD_MCNT_SET( RSERVE, FAILED_NOT_FOR_US, ctx->metrics->fail_not_for_us );
494 0 : FD_MCNT_SET( RSERVE, FAILED_OUTDATED, ctx->metrics->fail_outdated );
495 0 : FD_MCNT_SET( RSERVE, FAILED_INVALID_SHRED_INDEX, ctx->metrics->fail_invalid_shred_idx );
496 0 : FD_MCNT_SET( RSERVE, FAILED_PING_CACHE_LOOKUP, ctx->metrics->fail_ping_cache_lookup );
497 :
498 0 : FD_MGAUGE_SET( RSERVE, SHREDS_CURRENT, ctx->metrics->shreds_current );
499 0 : FD_MGAUGE_SET( RSERVE, DISK_CURRENT_BYTES, ctx->metrics->shreds_current*sizeof(fd_shredb_entry_t) );
500 0 : FD_MGAUGE_SET( RSERVE, DISK_ALLOCATED_BYTES, ctx->shredb->file_shreds*sizeof(fd_shredb_entry_t) );
501 :
502 0 : FD_MCNT_SET( RSERVE, PING_CACHE_ENTRIES, ctx->metrics->ping_cache_entries );
503 0 : FD_MCNT_SET( RSERVE, PING_CACHE_EVICTIONS, ctx->metrics->ping_cache_evictions );
504 0 : }
505 :
506 : static void
507 : privileged_init( fd_topo_t const * topo,
508 0 : fd_topo_tile_t const * tile ) {
509 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
510 :
511 0 : ulong size_limit = tile->rserve.shred_storage_limit_gib;
512 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
513 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
514 0 : ctx->shredb = FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_align(), fd_shredb_footprint( size_limit ) );
515 :
516 0 : FD_TEST( fd_rng_secure( &ctx->seed, sizeof(ulong) ) );
517 :
518 0 : FD_LOG_INFO(( "creating shredb (size_limit=%luGiB)", size_limit ));
519 0 : ctx->shredb = fd_shredb_join( fd_shredb_new( ctx->shredb, size_limit, tile->rserve.shredb_path, ctx->seed ) );
520 0 : if( FD_UNLIKELY( !ctx->shredb ) ) FD_LOG_ERR(( "failed to initialize shredb" ));
521 :
522 0 : uchar const * identity_public_key = fd_keyload_load( tile->rserve.identity_key_path, /* pubkey only: */ 1 );
523 0 : fd_memcpy( ctx->identity_public_key.uc, identity_public_key, sizeof(fd_pubkey_t) );
524 0 : }
525 :
526 : static void
527 : unprivileged_init( fd_topo_t const * topo,
528 0 : fd_topo_tile_t const * tile ) {
529 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
530 :
531 0 : ulong ping_cache_entries = tile->rserve.ping_cache_entries;
532 :
533 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
534 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
535 0 : FD_SCRATCH_ALLOC_APPEND( l, fd_shredb_align(), fd_shredb_footprint( tile->rserve.shred_storage_limit_gib ) );
536 0 : ctx->rserve = FD_SCRATCH_ALLOC_APPEND( l, fd_rserve_align(), fd_rserve_footprint( ping_cache_entries ) );
537 0 : FD_TEST( FD_SCRATCH_ALLOC_FINI( l, scratch_align() )==(ulong)scratch + scratch_footprint( tile ) );
538 :
539 0 : (void)ctx->shredb; /* Initialized in privileged_init */
540 0 : ctx->rserve = fd_rserve_join ( fd_rserve_new( ctx->rserve, ping_cache_entries, ctx->seed ) );
541 0 : ctx->keyswitch = fd_keyswitch_join( fd_topo_obj_laddr( topo, tile->id_keyswitch_obj_id ) );
542 0 : FD_TEST( ctx->keyswitch );
543 :
544 0 : fd_memset( ctx->metrics, 0, sizeof(ctx->metrics) );
545 0 : FD_MGAUGE_SET( RSERVE, SHREDS_MAX, ctx->shredb->max_shreds );
546 :
547 0 : ctx->halt_signing = 0;
548 0 : ctx->net_id = (ushort)0;
549 0 : fd_ip4_udp_hdr_init( ctx->serve_hdr, FD_RSERVE_MAX_PACKET_SIZE, 0, tile->rserve.repair_serve_listen_port );
550 0 : fd_sha512_new( ctx->sha512 );
551 :
552 0 : ulong sign_in_idx = fd_topo_find_tile_in_link ( topo, tile, "sign_rserve", tile->kind_id );
553 0 : ulong sign_out_idx = fd_topo_find_tile_out_link( topo, tile, "rserve_sign", tile->kind_id );
554 0 : FD_TEST( sign_in_idx!=ULONG_MAX );
555 0 : fd_topo_link_t const * sign_in = &topo->links[ tile->in_link_id[ sign_in_idx ] ];
556 0 : fd_topo_link_t const * sign_out = &topo->links[ tile->out_link_id[ sign_out_idx ] ];
557 0 : if( FD_UNLIKELY( !fd_keyguard_client_join( fd_keyguard_client_new( ctx->keyguard_client,
558 0 : sign_out->mcache,
559 0 : sign_out->dcache,
560 0 : sign_in->mcache,
561 0 : sign_in->dcache,
562 0 : sign_out->mtu ) ) ) ) {
563 0 : FD_LOG_ERR(( "failed to construct keyguard" ));
564 0 : }
565 :
566 0 : FD_TEST( tile->in_cnt>=1UL );
567 0 : for( ulong in_idx=0UL; in_idx<tile->in_cnt; in_idx++ ) {
568 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ in_idx ] ];
569 0 : if( 0==strcmp( link->name, "net_rserve" ) ) {
570 0 : ctx->in_kind[ in_idx ] = IN_KIND_NET;
571 0 : fd_net_rx_bounds_init( &ctx->in_links[ in_idx ].net_rx, link->dcache );
572 0 : continue;
573 0 : }
574 0 : else if( 0==strcmp( link->name, "shred_out" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SHRED;
575 0 : else if( 0==strcmp( link->name, "sign_rserve" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SIGN;
576 0 : else FD_LOG_ERR(( "rserve tile has unexpected input link: %s", link->name ));
577 :
578 0 : ctx->in_links[ in_idx ].mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
579 0 : ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
580 0 : ctx->in_links[ in_idx ].wmark = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
581 0 : ctx->in_links[ in_idx ].mtu = link->mtu;
582 0 : }
583 :
584 0 : ctx->net_out_idx = UINT_MAX;
585 0 : for( uint out_idx=0U; out_idx<tile->out_cnt; out_idx++ ) {
586 0 : fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ out_idx ] ];
587 0 : if( 0==strcmp( link->name, "rserve_net" ) ) {
588 0 : if( ctx->net_out_idx!=UINT_MAX ) continue; /* only use the first net link */
589 0 : ctx->net_out_idx = out_idx;
590 0 : ctx->net_out_mem = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
591 0 : ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, link->dcache );
592 0 : ctx->net_out_wmark = fd_dcache_compact_wmark( ctx->net_out_mem, link->dcache, link->mtu );
593 0 : ctx->net_out_chunk = ctx->net_out_chunk0;
594 0 : }
595 0 : else if( 0==strcmp( link->name, "rserve_sign" ) ) { /* Handled above for keyguard. */ }
596 0 : else FD_LOG_ERR(( "rserve tile has unexpected output link: %s", link->name ));
597 0 : }
598 0 : if( FD_UNLIKELY( ctx->net_out_idx==UINT_MAX ) ) FD_LOG_ERR(( "Missing rserve_net output link" ));
599 0 : }
600 :
601 : static ulong
602 : populate_allowed_seccomp( fd_topo_t const * topo FD_PARAM_UNUSED,
603 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
604 : ulong out_cnt,
605 0 : struct sock_filter * out ) {
606 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
607 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
608 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
609 0 : populate_sock_filter_policy_fd_rserve_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->shredb->fd );
610 0 : return sock_filter_policy_fd_rserve_tile_instr_cnt;
611 0 : }
612 :
613 : static ulong
614 : populate_allowed_fds( fd_topo_t const * topo FD_PARAM_UNUSED,
615 : fd_topo_tile_t const * tile FD_PARAM_UNUSED,
616 : ulong out_fds_cnt,
617 0 : int * out_fds ) {
618 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
619 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
620 0 : ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
621 :
622 0 : if( FD_UNLIKELY( out_fds_cnt<3UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
623 :
624 0 : ulong out_cnt = 0UL;
625 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
626 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
627 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
628 0 : out_fds[ out_cnt++ ] = ctx->shredb->fd;
629 0 : return out_cnt;
630 0 : }
631 :
632 : /* For orphan responses, we may send up to 11 net packets out.
633 : All other implemented codepaths will send at most 1. */
634 0 : #define STEM_BURST FD_RSERVE_MAX_ORPHAN_SLOTS
635 0 : #define STEM_LAZY (64000UL)
636 :
637 0 : #define STEM_CALLBACK_CONTEXT_TYPE ctx_t
638 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(ctx_t)
639 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
640 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
641 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
642 :
643 : #include "../../disco/stem/fd_stem.c"
644 :
645 : fd_topo_run_tile_t fd_tile_rserve = {
646 : .name = "rserve",
647 : .populate_allowed_seccomp = populate_allowed_seccomp,
648 : .populate_allowed_fds = populate_allowed_fds,
649 : .scratch_align = scratch_align,
650 : .scratch_footprint = scratch_footprint,
651 : .privileged_init = privileged_init,
652 : .unprivileged_init = unprivileged_init,
653 : .run = stem_run,
654 : };
|
