Line data Source code
1 : #include "fd_gossip.h"
2 : #include "fd_bloom.h"
3 : #include "fd_gossip_message.h"
4 : #include "fd_gossip_txbuild.h"
5 : #include "fd_active_set.h"
6 : #include "fd_ping_tracker.h"
7 : #include "fd_prune_finder.h"
8 : #include "fd_gossip_wsample.h"
9 : #include "../../disco/keyguard/fd_keyguard.h"
10 : #include "../../ballet/sha256/fd_sha256.h"
11 : #include "../leaders/fd_leaders_base.h"
12 :
13 : FD_STATIC_ASSERT( FD_METRICS_ENUM_GOSSIP_MESSAGE_CNT==FD_GOSSIP_MESSAGE_CNT,
14 : "FD_METRICS_ENUM_GOSSIP_MESSAGE_CNT must match FD_GOSSIP_MESSAGE_CNT" );
15 :
16 : FD_STATIC_ASSERT( FD_METRICS_ENUM_CRDS_VALUE_CNT==FD_GOSSIP_VALUE_CNT,
17 : "FD_METRICS_ENUM_CRDS_VALUE_CNT must match FD_GOSSIP_VALUE_CNT" );
18 :
19 0 : #define BLOOM_FALSE_POSITIVE_RATE (0.1)
20 0 : #define BLOOM_NUM_KEYS (8.0)
21 :
22 : struct stake {
23 : fd_pubkey_t pubkey;
24 : ulong stake;
25 :
26 : struct {
27 : ulong prev;
28 : ulong next;
29 : } map;
30 :
31 : struct {
32 : ulong next;
33 : } pool;
34 : };
35 :
36 : typedef struct stake stake_t;
37 :
38 : /* NOTE: Since the staked count is known at the time we populate
39 : the map, we can treat the pool as an array instead. This means we
40 : can bypass the acquire/release model and quickly iterate through the
41 : pool when we repopulate the map on every fd_gossip_stakes_update
42 : iteration. */
43 : #define POOL_NAME stake_pool
44 0 : #define POOL_T stake_t
45 : #define POOL_IDX_T ulong
46 0 : #define POOL_NEXT pool.next
47 : #include "../../util/tmpl/fd_pool.c"
48 :
49 : #define MAP_NAME stake_map
50 0 : #define MAP_KEY pubkey
51 : #define MAP_ELE_T stake_t
52 : #define MAP_KEY_T fd_pubkey_t
53 0 : #define MAP_PREV map.prev
54 0 : #define MAP_NEXT map.next
55 0 : #define MAP_KEY_EQ(k0,k1) fd_pubkey_eq( k0, k1 )
56 0 : #define MAP_KEY_HASH(key,seed) (seed^fd_ulong_load_8( (key)->uc ))
57 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
58 : #include "../../util/tmpl/fd_map_chain.c"
59 :
60 : struct fd_gossip_private {
61 : uchar identity_pubkey[ 32UL ];
62 : ulong identity_stake;
63 :
64 : fd_gossip_metrics_t metrics[1];
65 :
66 : fd_gossip_wsample_t * wsample;
67 : fd_crds_t * crds;
68 : fd_gossip_purged_t * purged;
69 : fd_active_set_t * active_set;
70 : fd_ping_tracker_t * ping_tracker;
71 : fd_prune_finder_t * prune_finder;
72 :
73 : fd_sha256_t sha256[1];
74 : fd_sha512_t sha512[1];
75 :
76 : ulong entrypoints_cnt;
77 : fd_ip4_port_t entrypoints[ 16UL ];
78 :
79 : fd_rng_t * rng;
80 :
81 : struct {
82 : ulong count;
83 : stake_t * pool;
84 : stake_map_t * map;
85 : } stake;
86 :
87 : struct {
88 : long next_pull_request;
89 : long next_active_set_refresh;
90 : long next_contact_info_refresh;
91 : long next_flush_push_state;
92 : } timers;
93 :
94 : /* Token-bucket rate limiter for outbound pull response data.
95 : Matches Agave's DataBudget: replenished every 100ms with
96 : num_staked*1024 bytes, capped at 5x that amount. Only
97 : pull responses are rate-limited; push messages are not. */
98 : struct {
99 : ulong remaining; /* bytes remaining in budget (signed) */
100 : long last_replenish_nanos; /* last replenish timestamp in nanos */
101 : } outbound_budget;
102 :
103 : /* Per-request iteration budget for the CRDS treap scan in
104 : rx_pull_request. Reset at the start of each request. */
105 : struct {
106 : ulong remaining;
107 : } scan_budget;
108 :
109 : /* Callbacks */
110 : fd_gossip_sign_fn sign_fn;
111 : void * sign_ctx;
112 :
113 : fd_gossip_send_fn send_fn;
114 : void * send_ctx;
115 :
116 : fd_ping_tracker_change_fn ping_tracker_change_fn;
117 : void * ping_tracker_change_fn_ctx;
118 :
119 : struct {
120 : uchar crds_val[ FD_GOSSIP_VALUE_MAX_SZ ];
121 : ulong crds_val_sz;
122 : fd_gossip_value_t ci[1];
123 : } my_contact_info;
124 :
125 : fd_gossip_out_ctx_t * gossip_net_out;
126 : };
127 :
128 : FD_FN_CONST ulong
129 0 : fd_gossip_align( void ) {
130 0 : return 128uL;
131 0 : }
132 :
133 : FD_FN_CONST ulong
134 : fd_gossip_footprint( ulong max_values,
135 0 : ulong entrypoints_len ) {
136 0 : ulong l;
137 0 : l = FD_LAYOUT_INIT;
138 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_gossip_t), sizeof(fd_gossip_t) );
139 0 : l = FD_LAYOUT_APPEND( l, fd_gossip_purged_align(), fd_gossip_purged_footprint( max_values ) );
140 0 : l = FD_LAYOUT_APPEND( l, fd_gossip_wsample_align(),fd_gossip_wsample_footprint( FD_CONTACT_INFO_TABLE_SIZE ) );
141 0 : l = FD_LAYOUT_APPEND( l, fd_crds_align(), fd_crds_footprint( max_values ) );
142 0 : l = FD_LAYOUT_APPEND( l, fd_active_set_align(), fd_active_set_footprint() );
143 0 : l = FD_LAYOUT_APPEND( l, fd_ping_tracker_align(), fd_ping_tracker_footprint( entrypoints_len ) );
144 0 : l = FD_LAYOUT_APPEND( l, fd_prune_finder_align(), fd_prune_finder_footprint() );
145 0 : l = FD_LAYOUT_APPEND( l, stake_pool_align(), stake_pool_footprint( MAX_SHRED_DESTS ) );
146 0 : l = FD_LAYOUT_APPEND( l, stake_map_align(), stake_map_footprint( stake_map_chain_cnt_est( MAX_SHRED_DESTS ) ) );
147 0 : l = FD_LAYOUT_FINI( l, fd_gossip_align() );
148 0 : return l;
149 0 : }
150 :
151 : static void
152 : ping_tracker_change( void * _ctx,
153 : uchar const * peer_pubkey,
154 : fd_ip4_port_t peer_address,
155 : long now,
156 0 : int change_type ) {
157 0 : fd_gossip_t * ctx = (fd_gossip_t *)_ctx;
158 :
159 0 : if( FD_UNLIKELY( !memcmp( peer_pubkey, ctx->identity_pubkey, 32UL ) ) ) return;
160 :
161 0 : if( FD_LIKELY( change_type==FD_PING_TRACKER_CHANGE_TYPE_ACTIVE ) ) {
162 0 : fd_gossip_purged_drain_no_contact_info( ctx->purged, peer_pubkey );
163 0 : }
164 :
165 0 : ulong ci_idx = fd_crds_ci_idx( ctx->crds, peer_pubkey );
166 0 : if( FD_UNLIKELY( ci_idx!=ULONG_MAX ) ) {
167 0 : switch( change_type ) {
168 0 : case FD_PING_TRACKER_CHANGE_TYPE_ACTIVE:
169 0 : fd_gossip_wsample_ping_tracked( ctx->wsample, ci_idx, 1 );
170 0 : break;
171 0 : case FD_PING_TRACKER_CHANGE_TYPE_INACTIVE:
172 0 : case FD_PING_TRACKER_CHANGE_TYPE_INACTIVE_STAKED:
173 0 : fd_gossip_wsample_ping_tracked( ctx->wsample, ci_idx, 0 );
174 0 : fd_active_set_remove_peer( ctx->active_set, ci_idx );
175 0 : break;
176 0 : default: FD_LOG_ERR(( "Unknown change type %d", change_type )); return;
177 0 : }
178 0 : }
179 :
180 0 : ctx->ping_tracker_change_fn( ctx->ping_tracker_change_fn_ctx, peer_pubkey, peer_address, now, change_type );
181 0 : }
182 :
183 : static inline void
184 : refresh_contact_info( fd_gossip_t * gossip,
185 0 : long now ) {
186 0 : fd_memcpy( gossip->my_contact_info.ci->origin, gossip->identity_pubkey, 32UL );
187 0 : gossip->my_contact_info.ci->wallclock = (ulong)FD_NANOSEC_TO_MILLI( now );
188 0 : long sz = fd_gossip_value_serialize( gossip->my_contact_info.ci, gossip->my_contact_info.crds_val, FD_GOSSIP_VALUE_MAX_SZ );
189 0 : FD_TEST( sz!=-1L );
190 0 : gossip->my_contact_info.crds_val_sz = (ulong)sz;
191 :
192 0 : gossip->sign_fn( gossip->sign_ctx,
193 0 : gossip->my_contact_info.crds_val+64UL,
194 0 : gossip->my_contact_info.crds_val_sz-64UL,
195 0 : FD_KEYGUARD_SIGN_TYPE_ED25519,
196 0 : gossip->my_contact_info.crds_val );
197 :
198 : /* We don't have stem_ctx here so we pre-empt in next
199 : fd_gossip_advance iteration instead. */
200 0 : gossip->timers.next_contact_info_refresh = now;
201 0 : }
202 :
203 : void *
204 : fd_gossip_new( void * shmem,
205 : fd_rng_t * rng,
206 : ulong max_values,
207 : ulong entrypoints_len,
208 : fd_ip4_port_t const * entrypoints,
209 : uchar const * identity_pubkey,
210 : fd_gossip_contact_info_t const * my_contact_info,
211 : long now,
212 : fd_gossip_send_fn send_fn,
213 : void * send_ctx,
214 : fd_gossip_sign_fn sign_fn,
215 : void * sign_ctx,
216 : fd_ping_tracker_change_fn ping_tracker_change_fn,
217 : void * ping_tracker_change_fn_ctx,
218 : fd_gossip_activity_update_fn activity_update_fn,
219 : void * activity_update_fn_ctx,
220 : fd_gossip_out_ctx_t * gossip_update_out,
221 0 : fd_gossip_out_ctx_t * gossip_net_out ) {
222 0 : if( FD_UNLIKELY( !shmem ) ) {
223 0 : FD_LOG_WARNING(( "NULL shmem" ));
224 0 : return NULL;
225 0 : }
226 :
227 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, fd_gossip_align() ) ) ) {
228 0 : FD_LOG_WARNING(( "misaligned shmem" ));
229 0 : return NULL;
230 0 : }
231 :
232 0 : if( FD_UNLIKELY( entrypoints_len>16UL ) ) {
233 0 : FD_LOG_WARNING(( "entrypoints_cnt must be in [0, 16]" ));
234 0 : return NULL;
235 0 : }
236 :
237 0 : if( FD_UNLIKELY( !fd_ulong_is_pow2( max_values ) ) ) {
238 0 : FD_LOG_WARNING(( "max_values must be a power of 2" ));
239 0 : return NULL;
240 0 : }
241 :
242 0 : FD_SCRATCH_ALLOC_INIT( l, shmem );
243 0 : fd_gossip_t * gossip = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_gossip_t), sizeof(fd_gossip_t) );
244 0 : void * purged = FD_SCRATCH_ALLOC_APPEND( l, fd_gossip_purged_align(), fd_gossip_purged_footprint( max_values ) );
245 0 : void * wsample = FD_SCRATCH_ALLOC_APPEND( l, fd_gossip_wsample_align(), fd_gossip_wsample_footprint( FD_CONTACT_INFO_TABLE_SIZE ) );
246 0 : void * crds = FD_SCRATCH_ALLOC_APPEND( l, fd_crds_align(), fd_crds_footprint( max_values ) );
247 0 : void * active_set = FD_SCRATCH_ALLOC_APPEND( l, fd_active_set_align(), fd_active_set_footprint() );
248 0 : void * ping_tracker = FD_SCRATCH_ALLOC_APPEND( l, fd_ping_tracker_align(), fd_ping_tracker_footprint( entrypoints_len ) );
249 0 : void * prune_finder = FD_SCRATCH_ALLOC_APPEND( l, fd_prune_finder_align(), fd_prune_finder_footprint() );
250 0 : void * stake_pool = FD_SCRATCH_ALLOC_APPEND( l, stake_pool_align(), stake_pool_footprint( MAX_SHRED_DESTS ) );
251 0 : void * stake_weights = FD_SCRATCH_ALLOC_APPEND( l, stake_map_align(), stake_map_footprint( stake_map_chain_cnt_est( MAX_SHRED_DESTS ) ) );
252 :
253 0 : gossip->gossip_net_out = gossip_net_out;
254 :
255 0 : gossip->entrypoints_cnt = entrypoints_len;
256 0 : fd_memcpy( gossip->entrypoints, entrypoints, entrypoints_len*sizeof(fd_ip4_port_t) );
257 :
258 0 : gossip->purged = fd_gossip_purged_join( fd_gossip_purged_new( purged, rng, max_values ) );
259 0 : FD_TEST( gossip->purged );
260 :
261 0 : gossip->wsample = fd_gossip_wsample_join( fd_gossip_wsample_new( wsample, rng, FD_CONTACT_INFO_TABLE_SIZE ) );
262 0 : FD_TEST( gossip->wsample );
263 :
264 0 : gossip->crds = fd_crds_join( fd_crds_new( crds, entrypoints, entrypoints_len, gossip->wsample, active_set, rng, max_values, gossip->purged, activity_update_fn, activity_update_fn_ctx, gossip_update_out ) );
265 0 : FD_TEST( gossip->crds );
266 :
267 0 : gossip->active_set = fd_active_set_join( fd_active_set_new( active_set, gossip->wsample, gossip->crds, rng, identity_pubkey, 0UL, send_fn, send_ctx ) );
268 0 : FD_TEST( gossip->active_set );
269 :
270 0 : gossip->ping_tracker = fd_ping_tracker_join( fd_ping_tracker_new( ping_tracker, rng, gossip->entrypoints_cnt, gossip->entrypoints, ping_tracker_change, gossip ) );
271 0 : FD_TEST( gossip->ping_tracker );
272 :
273 0 : gossip->prune_finder = fd_prune_finder_join( fd_prune_finder_new( prune_finder, fd_rng_ulong( rng ) ) );
274 0 : FD_TEST( gossip->prune_finder );
275 :
276 0 : gossip->stake.count = 0UL;
277 0 : gossip->stake.pool = stake_pool_join( stake_pool_new( stake_pool, MAX_SHRED_DESTS ) );
278 0 : FD_TEST( gossip->stake.pool );
279 :
280 0 : gossip->stake.map = stake_map_join( stake_map_new( stake_weights, stake_map_chain_cnt_est( MAX_SHRED_DESTS ), fd_rng_ulong( rng ) ) );
281 0 : FD_TEST( gossip->stake.map );
282 :
283 0 : FD_TEST( fd_sha256_join( fd_sha256_new( gossip->sha256 ) ) );
284 0 : FD_TEST( fd_sha512_join( fd_sha512_new( gossip->sha512 ) ) );
285 :
286 0 : gossip->rng = rng;
287 :
288 0 : gossip->timers.next_pull_request = 0L;
289 0 : gossip->timers.next_active_set_refresh = 0L;
290 0 : gossip->timers.next_contact_info_refresh = 0L;
291 0 : gossip->timers.next_flush_push_state = 0L;
292 :
293 0 : gossip->outbound_budget.remaining = 0UL;
294 0 : gossip->outbound_budget.last_replenish_nanos = now;
295 :
296 0 : gossip->send_fn = send_fn;
297 0 : gossip->send_ctx = send_ctx;
298 0 : gossip->sign_fn = sign_fn;
299 0 : gossip->sign_ctx = sign_ctx;
300 0 : gossip->ping_tracker_change_fn = ping_tracker_change_fn;
301 0 : gossip->ping_tracker_change_fn_ctx = ping_tracker_change_fn_ctx;
302 :
303 0 : gossip->my_contact_info.ci->tag = FD_GOSSIP_VALUE_CONTACT_INFO;
304 0 : *gossip->my_contact_info.ci->contact_info = *my_contact_info;
305 0 : fd_memcpy( gossip->identity_pubkey, identity_pubkey, 32UL );
306 0 : gossip->identity_stake = 0UL;
307 0 : refresh_contact_info( gossip, now );
308 :
309 0 : fd_memset( gossip->metrics, 0, sizeof(fd_gossip_metrics_t) );
310 :
311 0 : return gossip;
312 0 : }
313 :
314 : fd_gossip_t *
315 0 : fd_gossip_join( void * shgossip ) {
316 0 : if( FD_UNLIKELY( !shgossip ) ) {
317 0 : FD_LOG_WARNING(( "NULL shgossip" ));
318 0 : return NULL;
319 0 : }
320 :
321 0 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shgossip, fd_gossip_align() ) ) ) {
322 0 : FD_LOG_WARNING(( "misaligned shgossip" ));
323 0 : return NULL;
324 0 : }
325 :
326 0 : return (fd_gossip_t *)shgossip;
327 0 : }
328 :
329 : fd_gossip_metrics_t const *
330 0 : fd_gossip_metrics( fd_gossip_t const * gossip ) {
331 0 : return gossip->metrics;
332 0 : }
333 :
334 : fd_crds_metrics_t const *
335 0 : fd_gossip_crds_metrics( fd_gossip_t const * gossip ) {
336 0 : return fd_crds_metrics( gossip->crds );
337 0 : }
338 :
339 : fd_ping_tracker_metrics_t const *
340 0 : fd_gossip_ping_tracker_metrics( fd_gossip_t const * gossip ) {
341 0 : return fd_ping_tracker_metrics( gossip->ping_tracker );
342 0 : }
343 :
344 : fd_gossip_purged_metrics_t const *
345 0 : fd_gossip_purged_metrics2( fd_gossip_t const * gossip ) {
346 0 : return fd_gossip_purged_metrics( gossip->purged );
347 0 : }
348 :
349 : fd_active_set_metrics_t const *
350 0 : fd_gossip_active_set_metrics2( fd_gossip_t const * gossip ) {
351 0 : return fd_active_set_metrics( gossip->active_set );
352 0 : }
353 :
354 : static fd_ip4_port_t
355 0 : random_entrypoint( fd_gossip_t const * gossip ) {
356 0 : ulong idx = fd_rng_ulong_roll( gossip->rng, gossip->entrypoints_cnt );
357 0 : return gossip->entrypoints[ idx ];
358 0 : }
359 :
360 : static ulong
361 : get_stake( fd_gossip_t const * gossip,
362 0 : uchar const * pubkey ) {
363 0 : stake_t const * entry = stake_map_ele_query_const( gossip->stake.map, (fd_pubkey_t const *)pubkey, NULL, gossip->stake.pool );
364 0 : if( FD_UNLIKELY( !entry ) ) return 0UL;
365 0 : return entry->stake;
366 0 : }
367 :
368 : void
369 : fd_gossip_set_identity( fd_gossip_t * gossip,
370 : uchar const * identity_pubkey,
371 : long now,
372 0 : ulong identity_outset ) {
373 0 : int identity_changed = memcmp( gossip->identity_pubkey, identity_pubkey, 32UL );
374 0 : if( FD_UNLIKELY( !identity_changed ) ) return;
375 :
376 0 : ulong new_ci_idx = fd_crds_ci_idx( gossip->crds, identity_pubkey );
377 :
378 : /* The new identity may already exist in CRDS as a normal peer (active
379 : in the wsample and potentially present in the active set). We
380 : must deactivate it before updating identity_pubkey to maintain the
381 : invariant that our own identity is never sampleable. */
382 0 : if( FD_UNLIKELY( new_ci_idx!=ULONG_MAX ) ) fd_active_set_remove_peer( gossip->active_set, new_ci_idx );
383 :
384 : /* Also remove the new identity from the ping tracker (we don't
385 : track ourselves). */
386 0 : fd_ping_tracker_remove( gossip->ping_tracker, identity_pubkey, now );
387 :
388 0 : fd_memcpy( gossip->identity_pubkey, identity_pubkey, 32UL );
389 0 : gossip->identity_stake = get_stake( gossip, identity_pubkey );
390 0 : fd_gossip_wsample_set_identity( gossip->wsample, new_ci_idx );
391 0 : fd_gossip_wsample_self_stake( gossip->wsample, gossip->identity_stake );
392 0 : fd_active_set_set_identity( gossip->active_set, gossip->identity_pubkey, gossip->identity_stake );
393 0 : fd_prune_finder_set_identity( gossip->prune_finder, gossip->identity_pubkey, gossip->identity_stake );
394 :
395 : /* For identity swaps, refresh the contact info outset so this
396 : instance can override older contact info for the same identity. */
397 0 : gossip->my_contact_info.ci->contact_info->outset = identity_outset;
398 0 : refresh_contact_info( gossip, now );
399 0 : }
400 :
401 : void
402 : fd_gossip_set_shred_version( fd_gossip_t * gossip,
403 : ushort shred_version,
404 0 : long now ) {
405 0 : gossip->my_contact_info.ci->contact_info->shred_version = shred_version;
406 0 : refresh_contact_info( gossip, now );
407 0 : }
408 :
409 : void
410 : fd_gossip_stakes_update( fd_gossip_t * gossip,
411 : fd_stake_weight_t const * stake_weights,
412 0 : ulong stake_weights_cnt ) {
413 0 : for( ulong i=0UL; i<stake_weights_cnt; i++ ) {
414 0 : if( FD_LIKELY( stake_weights[ i ].stake<FD_GOSSIP_STAKED_THRESHOLD ) ) continue;
415 0 : if( FD_UNLIKELY( get_stake( gossip, stake_weights[ i ].key.uc )<FD_GOSSIP_STAKED_THRESHOLD ) )
416 0 : fd_gossip_purged_drain_no_contact_info( gossip->purged, stake_weights[ i ].key.uc );
417 0 : }
418 :
419 0 : stake_map_reset( gossip->stake.map );
420 0 : stake_pool_reset( gossip->stake.pool );
421 :
422 0 : for( ulong i=0UL; i<stake_weights_cnt; i++ ) {
423 0 : stake_t * entry = stake_pool_ele_acquire( gossip->stake.pool );
424 0 : entry->pubkey = stake_weights[i].key;
425 0 : entry->stake = stake_weights[i].stake;
426 0 : stake_map_ele_insert( gossip->stake.map, entry, gossip->stake.pool );
427 0 : }
428 :
429 0 : gossip->identity_stake = get_stake( gossip, gossip->identity_pubkey );
430 0 : fd_gossip_wsample_self_stake( gossip->wsample, gossip->identity_stake );
431 0 : fd_active_set_set_identity( gossip->active_set, gossip->identity_pubkey, gossip->identity_stake );
432 0 : fd_prune_finder_set_identity( gossip->prune_finder, gossip->identity_pubkey, gossip->identity_stake );
433 0 : gossip->stake.count = stake_pool_used( gossip->stake.pool );
434 0 : }
435 :
436 : /* Outbound data budget constants (matching Agave's DataBudget for gossip).
437 : Budget is replenished every BUDGET_REPLENISH_INTERVAL_NS with
438 : num_staked * BUDGET_BYTES_PER_INTERVAL bytes, capped at
439 : BUDGET_MAX_MULTIPLE * num_staked * BUDGET_BYTES_PER_INTERVAL. */
440 :
441 : #define BUDGET_REPLENISH_INTERVAL_NS (100L*1000L*1000L) /* 100 ms */
442 0 : #define BUDGET_BYTES_PER_INTERVAL (1024UL) /* per staked validator */
443 0 : #define BUDGET_MAX_MULTIPLE (5UL) /* max accumulation */
444 0 : #define BUDGET_MIN_STAKED (2UL) /* floor for num_staked */
445 :
446 : /* Lazily replenish the outbound pull-response budget if at least
447 : BUDGET_REPLENISH_INTERVAL_NS have elapsed since last replenish.
448 : Returns current remaining budget in bytes. */
449 :
450 : static inline ulong
451 : outbound_budget_replenish( fd_gossip_t * gossip,
452 0 : long now ) {
453 0 : long elapsed = now-gossip->outbound_budget.last_replenish_nanos;
454 :
455 0 : if( FD_LIKELY( elapsed>=BUDGET_REPLENISH_INTERVAL_NS ) ) {
456 0 : ulong num_staked = fd_ulong_max( gossip->stake.count, BUDGET_MIN_STAKED );
457 0 : ulong increment = num_staked * BUDGET_BYTES_PER_INTERVAL;
458 0 : ulong cap = BUDGET_MAX_MULTIPLE * increment;
459 0 : ulong remaining = gossip->outbound_budget.remaining + increment;
460 0 : gossip->outbound_budget.remaining = fd_ulong_min( remaining, cap );
461 0 : gossip->outbound_budget.last_replenish_nanos = now;
462 0 : }
463 0 : return gossip->outbound_budget.remaining;
464 0 : }
465 :
466 : static inline void
467 : txbuild_flush( fd_gossip_t * gossip,
468 : fd_gossip_txbuild_t * txbuild,
469 : fd_stem_context_t * stem,
470 : fd_ip4_port_t dest_addr,
471 0 : long now ) {
472 0 : if( FD_UNLIKELY( !txbuild->crds_len ) ) return;
473 :
474 : /* Debit the outbound data budget (gossip payload bytes only, not
475 : including IP/UDP headers — matching Agave's DataBudget which
476 : operates on serialized gossip-layer packet sizes). */
477 0 : gossip->outbound_budget.remaining -= fd_ulong_min( txbuild->bytes_len, gossip->outbound_budget.remaining );
478 :
479 0 : gossip->send_fn( gossip->send_ctx, stem, txbuild->bytes, txbuild->bytes_len, &dest_addr, (ulong)now );
480 :
481 0 : gossip->metrics->message_tx[ txbuild->tag ]++;
482 0 : gossip->metrics->message_tx_bytes[ txbuild->tag ] += txbuild->bytes_len+42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
483 0 : for( ulong i=0UL; i<txbuild->crds_len; i++ ) {
484 0 : gossip->metrics->crds_tx_pull_response[ txbuild->crds[ i ].tag ]++;
485 0 : gossip->metrics->crds_tx_pull_response_bytes[ txbuild->crds[ i ].tag ] += txbuild->crds[ i ].sz;
486 0 : }
487 :
488 0 : fd_gossip_txbuild_init( txbuild, gossip->identity_pubkey, txbuild->tag );
489 0 : }
490 :
491 : /* pull_scan_range iterates CRDS entries in [start_hash, end_hash] and
492 : appends values the caller is missing to pull_resp. Returns 1 if a
493 : budget was exhausted (iteration or outbound data), 0 if the range was
494 : fully scanned. */
495 :
496 : static int
497 : pull_scan_range( fd_gossip_t * gossip,
498 : ulong start_hash,
499 : ulong end_hash,
500 : fd_bloom_t * filter,
501 : ulong adjusted_wallclock_ms,
502 : fd_gossip_txbuild_t * pull_resp,
503 : fd_stem_context_t * stem,
504 : fd_ip4_port_t peer_addr,
505 0 : long now ) {
506 0 : uchar iter_mem[ 16UL ];
507 :
508 0 : for( fd_crds_mask_iter_t * it = fd_crds_mask_iter_init_range( gossip->crds, start_hash, end_hash, iter_mem );
509 0 : !fd_crds_mask_iter_done( it, gossip->crds );
510 0 : it=fd_crds_mask_iter_next( it, gossip->crds ) ) {
511 0 : if( FD_UNLIKELY( !gossip->scan_budget.remaining ) ) return 1;
512 0 : gossip->scan_budget.remaining--;
513 :
514 0 : fd_crds_entry_t const * candidate = fd_crds_mask_iter_entry( it, gossip->crds );
515 :
516 0 : if( FD_UNLIKELY( fd_crds_entry_wallclock( candidate )>adjusted_wallclock_ms ) ) continue;
517 :
518 0 : if( FD_UNLIKELY( fd_bloom_contains( filter, fd_crds_entry_hash( candidate ), 32UL ) ) ) continue;
519 :
520 0 : uchar const * crds_val;
521 0 : ulong crds_size;
522 0 : fd_crds_entry_value( candidate, &crds_val, &crds_size );
523 0 : if( FD_UNLIKELY( !fd_gossip_txbuild_can_fit( pull_resp, crds_size ) ) ) txbuild_flush( gossip, pull_resp, stem, peer_addr, now );
524 0 : fd_gossip_txbuild_append( pull_resp, crds_size, crds_val );
525 :
526 0 : if( FD_UNLIKELY( !gossip->outbound_budget.remaining ) ) return 1;
527 0 : }
528 0 : return 0;
529 0 : }
530 :
531 : static void
532 : rx_pull_request( fd_gossip_t * gossip,
533 : fd_gossip_pull_request_t const * pr_view,
534 : fd_ip4_port_t peer_addr,
535 : fd_stem_context_t * stem,
536 0 : long now ) {
537 : /* Replenish and check outbound data budget. If the budget is
538 : exhausted, skip generating pull responses entirely. */
539 0 : if( FD_UNLIKELY( !outbound_budget_replenish( gossip, now ) ) ) return;
540 :
541 : /* When responding to a pull request, we skip CRDS entries whose
542 : wallclock is newer than the caller's wallclock + a random jitter.
543 : The jitter is drawn uniformly from [0, TIMEOUT/4) ms, matching
544 : Agave's behavior (CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS = 15000ms).
545 : This prevents all responders from consistently excluding the same
546 : set of very-recent CRDS values. */
547 0 : #define FD_GOSSIP_PULL_JITTER_BOUND_MS (15000UL/4UL)
548 :
549 : /* Generate a random jitter in [0, 3750) ms, added to the caller's
550 : wallclock. CRDS entries newer than this adjusted threshold are
551 : excluded from the response. The jitter prevents all responders
552 : from consistently excluding the same near-boundary entries,
553 : improving cluster-wide convergence of recent values. */
554 0 : ulong caller_wallclock_ms = pr_view->contact_info->wallclock;
555 0 : ulong jitter_ms = fd_rng_ulong_roll( gossip->rng, FD_GOSSIP_PULL_JITTER_BOUND_MS );
556 0 : ulong adjusted_wallclock_ms = caller_wallclock_ms + jitter_ms;
557 :
558 0 : ulong keys[ sizeof(pr_view->crds_filter->filter->keys)/sizeof(ulong) ];
559 0 : ulong bits[ sizeof(pr_view->crds_filter->filter->bits)/sizeof(ulong) ];
560 0 : fd_memcpy( keys, pr_view->crds_filter->filter->keys, sizeof(pr_view->crds_filter->filter->keys) );
561 0 : fd_memcpy( bits, pr_view->crds_filter->filter->bits, sizeof(pr_view->crds_filter->filter->bits) );
562 :
563 0 : fd_bloom_t filter[1];
564 0 : filter->keys_len = pr_view->crds_filter->filter->keys_len;
565 0 : filter->keys = keys;
566 :
567 0 : filter->bits_len = pr_view->crds_filter->filter->bits_len;
568 0 : filter->bits = bits;
569 :
570 0 : fd_gossip_txbuild_t pull_resp[1];
571 0 : fd_gossip_txbuild_init( pull_resp, gossip->identity_pubkey, FD_GOSSIP_MESSAGE_PULL_RESPONSE );
572 :
573 : /* CPU budget for the CRDS treap scan. An honest sender picks
574 :
575 : mask_bits = ceil(log2(num_items / max_items))
576 : num_items >= MIN_NUM_BLOOM_ITEMS (65536)
577 :
578 : max_items is computed by fd_bloom_max_items as:
579 :
580 : max_items = ceil( max_bits / D(K) )
581 : D(K) = -K / ln(1 - exp(ln(p)/K)), p=0.1
582 :
583 : where K is the number of bloom hash functions. D(K) is minimized
584 : at K=3 where D(3)=4.808. max_bits is bounded by the bloom struct's
585 : bits[] array which holds at most 151 u64 words = 9664 bits (see
586 : fd_gossip_message.h). Therefore,
587 :
588 : max_items <= ceil(9664/4.808) = 2010 < 2048.
589 :
590 : The sender's mask_bits partitions the hash space into 2^mask_bits
591 : partitions, each covering ~max_items of the sender's entries.
592 : What's the worst case iteration count for the receiver given an
593 : honest sender?
594 :
595 : 2^mask_bits >= MIN_NUM_BLOOM_ITEMS / max_items
596 : iterations <= crds_len / 2^mask_bits
597 : iterations <= crds_len * max_items / MIN_NUM_BLOOM_ITEMS
598 : <= crds_len * 2048 / 65536
599 : <= crds_len / 32
600 :
601 : A floor of 1024 handles small tables where the budget
602 : would otherwise round to near-zero. */
603 :
604 0 : gossip->scan_budget.remaining = fd_ulong_max( fd_crds_len( gossip->crds ) / 32UL, 1024UL );
605 :
606 : /* Compute the hash range [start_hash, end_hash] for this mask
607 : partition, then pick a random starting point within it.
608 : We iterate [mid_hash, end_hash] then [start_hash, mid_hash)
609 : so that when the iteration budget truncates the scan, different
610 : entries are skipped each time rather than always penalizing the
611 : tail of the partition. */
612 :
613 0 : ulong start_hash, end_hash;
614 0 : fd_gossip_purged_generate_masks( pr_view->crds_filter->mask, pr_view->crds_filter->mask_bits, &start_hash, &end_hash );
615 0 : ulong range = end_hash - start_hash;
616 0 : ulong mid_hash = start_hash + ( range==ULONG_MAX ? fd_rng_ulong( gossip->rng ) : fd_rng_ulong( gossip->rng ) % (range + 1UL) );
617 :
618 0 : int exhausted = pull_scan_range( gossip, mid_hash, end_hash, filter, adjusted_wallclock_ms, pull_resp, stem, peer_addr, now );
619 0 : if( FD_LIKELY( !exhausted && mid_hash>start_hash ) ) pull_scan_range( gossip, start_hash, mid_hash-1UL, filter, adjusted_wallclock_ms, pull_resp, stem, peer_addr, now );
620 :
621 0 : txbuild_flush( gossip, pull_resp, stem, peer_addr, now );
622 0 : }
623 :
624 : static void
625 : rx_values( fd_gossip_t * gossip,
626 : ulong values_len,
627 : fd_gossip_value_t const * values,
628 : uchar const * payload,
629 : uchar const * failed,
630 : fd_stem_context_t * stem,
631 : long now,
632 0 : long results[ static 17UL ] ) {
633 0 : for( ulong i=0UL; i<values_len; i++ ) {
634 0 : fd_gossip_value_t const * value = &values[ i ];
635 :
636 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) {
637 0 : uchar candidate_hash[ 32UL ];
638 0 : fd_sha256_hash( payload+value->offset, value->length, candidate_hash );
639 0 : if( FD_LIKELY( failed[ i ]==FD_GOSSIP_FAILED_NO_CONTACT_INFO ) ) fd_gossip_purged_insert_no_contact_info( gossip->purged, value->origin, candidate_hash, now );
640 0 : else fd_gossip_purged_insert_failed_insert( gossip->purged, candidate_hash, now );
641 0 : continue;
642 0 : }
643 :
644 0 : ulong origin_stake = get_stake( gossip, value->origin );
645 0 : int is_me = !memcmp( value->origin, gossip->identity_pubkey, 32UL );
646 :
647 0 : int origin_ping_tracker_active = 0;
648 0 : fd_ip4_port_t origin_addr = { 0 };
649 0 : if( FD_UNLIKELY( value->tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
650 0 : origin_addr = (fd_ip4_port_t){
651 0 : .addr = value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4,
652 0 : .port = value->contact_info->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port
653 0 : };
654 0 : origin_ping_tracker_active = fd_ping_tracker_active( gossip->ping_tracker, value->origin, origin_addr );
655 0 : }
656 :
657 0 : results[ i ] = fd_crds_insert( gossip->crds, value, payload+value->offset, value->length, origin_stake, origin_ping_tracker_active, is_me, now, stem );
658 0 : if( FD_UNLIKELY( results[ i ] ) ) continue;
659 :
660 0 : if( FD_UNLIKELY( value->tag==FD_GOSSIP_VALUE_CONTACT_INFO ) ) {
661 0 : if( FD_LIKELY( !is_me ) ) fd_ping_tracker_track( gossip->ping_tracker, value->origin, origin_stake, origin_addr, now );
662 :
663 : /* We just learned this peer's contact info. Drain any
664 : no_contact_info hashes associated with this origin from the
665 : purged set so peers re-send those CRDS values. */
666 0 : if( FD_LIKELY( fd_ping_tracker_active( gossip->ping_tracker, value->origin, origin_addr ) ) ) fd_gossip_purged_drain_no_contact_info( gossip->purged, value->origin );
667 0 : }
668 :
669 0 : fd_active_set_push( gossip->active_set, payload+value->offset, value->length, value->origin, origin_stake, stem, now, 0 );
670 0 : }
671 0 : }
672 :
673 : static void
674 : rx_pull_response( fd_gossip_t * gossip,
675 : fd_gossip_pull_response_t const * pull_response,
676 : uchar const * payload,
677 : uchar const * failed,
678 : fd_stem_context_t * stem,
679 0 : long now ) {
680 0 : long results[ 17UL ];
681 0 : rx_values( gossip, pull_response->values_len, pull_response->values, payload, failed, stem, now, results );
682 0 : for( ulong i=0UL; i<pull_response->values_len; i++ ) {
683 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) continue;
684 0 : if( FD_LIKELY( !results[ i ] ) ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_UPSERTED_PULL_RESPONSE_IDX ]++;
685 0 : else if( results[ i ]<0L ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_DROPPED_PULL_RESPONSE_STALE_IDX ]++;
686 0 : else gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_DROPPED_PULL_RESPONSE_DUPLICATE_IDX ]++;
687 0 : }
688 0 : }
689 :
690 : /* tx_prune constructs, signs, and sends a prune message telling
691 : `relayer` to stop pushing CRDS values originating from `origin`.
692 :
693 : On-wire layout (bincode):
694 : Protocol tag 4 (FD_GOSSIP_MESSAGE_PRUNE = 3)
695 : sender pubkey 32 (= identity_pubkey, outer PruneMessage field)
696 : PruneData.pubkey 32 (= identity_pubkey)
697 : prunes_len 8
698 : prunes[1] 32
699 : signature 64
700 : destination 32
701 : wallclock 8
702 :
703 : The signable data (input to Ed25519 sign) is the PruneData fields
704 : excluding signature:
705 : prefix[26] + pubkey[32] + prunes_len[8] + prunes[32] + destination[32] + wallclock[8]
706 : This must match fd_keyguard_payload_matches_prune_data (106 + 32 bytes). */
707 :
708 : static void
709 : tx_prune( fd_gossip_t * gossip,
710 : uchar const * relayer,
711 : uchar const * origin,
712 : fd_stem_context_t * stem,
713 0 : long now ) {
714 0 : ulong ci_idx = fd_crds_ci_idx( gossip->crds, relayer );
715 0 : if( FD_UNLIKELY( ci_idx==ULONG_MAX ) ) return;
716 :
717 0 : fd_gossip_contact_info_t const * ci = fd_crds_ci( gossip->crds, ci_idx );
718 0 : fd_ip4_port_t dest_addr = {
719 0 : .addr = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0U : ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4,
720 0 : .port = ci->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port
721 0 : };
722 0 : if( FD_UNLIKELY( !dest_addr.addr || !dest_addr.port ) ) return;
723 :
724 0 : ulong wallclock = (ulong)FD_NANOSEC_TO_MILLI( now );
725 :
726 : /* Build the signable payload:
727 : prefix[26] + pubkey[32] + prunes_len[8] + prunes[32] + destination[32] + wallclock[8] */
728 0 : uchar signable[ 26UL + 32UL + 8UL + 32UL + 32UL + 8UL ];
729 0 : uchar * p = signable;
730 0 : FD_STORE( ulong, p, 18UL ); p += 8UL;
731 0 : fd_memcpy( p, "\xffSOLANA_PRUNE_DATA", 18UL ); p += 18UL;
732 0 : fd_memcpy( p, gossip->identity_pubkey, 32UL ); p += 32UL;
733 0 : FD_STORE( ulong, p, 1UL ); p += 8UL;
734 0 : fd_memcpy( p, origin, 32UL ); p += 32UL;
735 0 : fd_memcpy( p, relayer, 32UL ); p += 32UL;
736 0 : FD_STORE( ulong, p, wallclock ); p += 8UL;
737 :
738 0 : uchar signature[ 64UL ];
739 0 : gossip->sign_fn( gossip->sign_ctx, signable, sizeof(signable), FD_KEYGUARD_SIGN_TYPE_ED25519, signature );
740 :
741 : /* Build the on-wire packet:
742 : tag(4) + sender(32) + pubkey(32) + prunes_len(8) + prunes[32]
743 : + signature(64) + destination(32) + wallclock(8) */
744 0 : uchar pkt[ 4UL + 32UL + 32UL + 8UL + 32UL + 64UL + 32UL + 8UL ];
745 0 : uchar * q = pkt;
746 0 : FD_STORE( uint, q, FD_GOSSIP_MESSAGE_PRUNE ); q += 4UL;
747 0 : fd_memcpy( q, gossip->identity_pubkey, 32UL ); q += 32UL; /* sender */
748 0 : fd_memcpy( q, gossip->identity_pubkey, 32UL ); q += 32UL; /* PruneData.pubkey */
749 0 : FD_STORE( ulong, q, 1UL ); q += 8UL;
750 0 : fd_memcpy( q, origin, 32UL ); q += 32UL;
751 0 : fd_memcpy( q, signature, 64UL ); q += 64UL;
752 0 : fd_memcpy( q, relayer, 32UL ); q += 32UL;
753 0 : FD_STORE( ulong, q, wallclock ); q += 8UL;
754 :
755 0 : gossip->send_fn( gossip->send_ctx, stem, pkt, sizeof(pkt), &dest_addr, (ulong)now );
756 :
757 0 : gossip->metrics->message_tx[ FD_GOSSIP_MESSAGE_PRUNE ]++;
758 0 : gossip->metrics->message_tx_bytes[ FD_GOSSIP_MESSAGE_PRUNE ] += sizeof(pkt) + 42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
759 0 : }
760 :
761 : static void
762 : tx_prunes( fd_gossip_t * gossip,
763 : fd_stem_context_t * stem,
764 0 : long now ) {
765 0 : uchar const * relayer;
766 0 : uchar const * origin;
767 0 : while( fd_prune_finder_pop_prune( gossip->prune_finder, &relayer, &origin ) ) {
768 0 : tx_prune( gossip, relayer, origin, stem, now );
769 0 : }
770 0 : }
771 :
772 : static void
773 : rx_push( fd_gossip_t * gossip,
774 : fd_gossip_push_t const * push,
775 : uchar const * payload,
776 : uchar const * failed,
777 : long now,
778 0 : fd_stem_context_t * stem ) {
779 0 : long results[ 17UL ];
780 0 : rx_values( gossip, push->values_len, push->values, payload, failed, stem, now, results );
781 :
782 0 : for( ulong i=0UL; i<push->values_len; i++ ) {
783 0 : if( FD_UNLIKELY( failed[ i ] && failed[ i ]!=FD_GOSSIP_FAILED_DUPLICATE ) ) continue;
784 0 : if( FD_LIKELY( !results[ i ] ) ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_UPSERTED_PUSH_IDX ]++;
785 0 : else if( results[ i ]<0L ) gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_DROPPED_PUSH_STALE_IDX ]++;
786 0 : else gossip->metrics->crds_rx_count[ FD_METRICS_ENUM_GOSSIP_CRDS_OUTCOME_V_DROPPED_PUSH_DUPLICATE_IDX ]++;
787 :
788 0 : ulong num_dups;
789 0 : if( FD_LIKELY( !results[ i ] ) ) num_dups = 0UL;
790 0 : else if( FD_UNLIKELY( results[ i ]<0L ) ) num_dups = ULONG_MAX; /* stale => never timely */
791 0 : else num_dups = (ulong)results[ i ];
792 :
793 0 : ulong origin_stake = get_stake( gossip, push->values[ i ].origin );
794 0 : fd_prune_finder_record( gossip->prune_finder, push->values[ i ].origin, origin_stake, push->from, get_stake( gossip, push->from ), num_dups );
795 0 : }
796 :
797 0 : tx_prunes( gossip, stem, now );
798 0 : }
799 :
800 : static void
801 : rx_prune( fd_gossip_t * gossip,
802 0 : fd_gossip_prune_t const * prune ) {
803 0 : for( ulong i=0UL; i<prune->prunes_len; i++ ) {
804 0 : fd_active_set_prune( gossip->active_set,
805 0 : prune->pubkey,
806 0 : prune->prunes[ i ],
807 0 : get_stake( gossip, prune->prunes[ i ] ) );
808 0 : }
809 0 : }
810 :
811 :
812 : static void
813 : rx_ping( fd_gossip_t * gossip,
814 : fd_gossip_ping_t const * ping,
815 : fd_ip4_port_t peer_address,
816 : fd_stem_context_t * stem,
817 0 : long now ) {
818 0 : uchar out_payload[ sizeof(fd_gossip_pong_t)+4UL];
819 0 : FD_STORE( uint, out_payload, FD_GOSSIP_MESSAGE_PONG );
820 :
821 0 : fd_gossip_pong_t * out_pong = (fd_gossip_pong_t *)(out_payload + 4UL);
822 0 : fd_memcpy( out_pong->from, gossip->identity_pubkey, 32UL );
823 :
824 : /* fd_keyguard checks payloads for certain patterns before performing the
825 : sign. Pattern-matching can't be done on hashed data, so we need
826 : to supply the pre-hashed image to the sign fn (fd_keyguard will hash when
827 : supplied with FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519) while also hashing
828 : the image ourselves onto pong->ping_hash */
829 :
830 0 : uchar pre_image[ 48UL ];
831 0 : fd_memcpy( pre_image, "SOLANA_PING_PONG", 16UL );
832 0 : fd_memcpy( pre_image+16UL, ping->token, 32UL );
833 :
834 0 : fd_sha256_hash( pre_image, 48UL, out_pong->hash );
835 :
836 0 : gossip->sign_fn( gossip->sign_ctx, pre_image, 48UL, FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519, out_pong->signature );
837 0 : gossip->send_fn( gossip->send_ctx, stem, out_payload, sizeof(out_payload), &peer_address, (ulong)now );
838 :
839 0 : gossip->metrics->message_tx[ FD_GOSSIP_MESSAGE_PONG ]++;
840 0 : gossip->metrics->message_tx_bytes[ FD_GOSSIP_MESSAGE_PONG ] += sizeof(out_payload)+42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
841 0 : }
842 :
843 : static void
844 : rx_pong( fd_gossip_t * gossip,
845 : fd_gossip_pong_t const * pong,
846 : fd_ip4_port_t peer_address,
847 0 : long now ) {
848 0 : ulong stake = get_stake( gossip, pong->from );
849 0 : fd_ping_tracker_register( gossip->ping_tracker, pong->from, stake, peer_address, pong->hash, now );
850 0 : }
851 :
852 : void
853 : fd_gossip_rx( fd_gossip_t * gossip,
854 : fd_ip4_port_t peer,
855 : uchar const * data,
856 : ulong data_sz,
857 : long now,
858 0 : fd_stem_context_t * stem ) {
859 : /* TODO: Implement traffic shaper / bandwidth limiter */
860 0 : FD_TEST( data_sz>=sizeof(fd_gossip_message_t)+FD_GOSSIP_MESSAGE_MAX_CRDS );
861 0 : fd_gossip_message_t const * message = (fd_gossip_message_t const *)data;
862 0 : uchar const * failed = data+sizeof(fd_gossip_message_t);
863 0 : uchar const * payload = data+sizeof(fd_gossip_message_t)+FD_GOSSIP_MESSAGE_MAX_CRDS;
864 :
865 0 : switch( message->tag ) {
866 0 : case FD_GOSSIP_MESSAGE_PULL_REQUEST: rx_pull_request( gossip, message->pull_request, peer, stem, now ); break;
867 0 : case FD_GOSSIP_MESSAGE_PULL_RESPONSE: rx_pull_response( gossip, message->pull_response, payload, failed, stem, now ); break;
868 0 : case FD_GOSSIP_MESSAGE_PUSH: rx_push( gossip, message->push, payload, failed, now, stem ); break;
869 0 : case FD_GOSSIP_MESSAGE_PRUNE: rx_prune( gossip, message->prune ); break;
870 0 : case FD_GOSSIP_MESSAGE_PING: rx_ping( gossip, message->ping, peer, stem, now ); break;
871 0 : case FD_GOSSIP_MESSAGE_PONG: rx_pong( gossip, message->pong, peer, now ); break;
872 0 : default:
873 0 : FD_LOG_CRIT(( "Unknown gossip message type %u", message->tag ));
874 0 : break;
875 0 : }
876 0 : }
877 :
878 : static int
879 : fd_gossip_push( fd_gossip_t * gossip,
880 : fd_gossip_value_t const * value,
881 : fd_stem_context_t * stem,
882 0 : long now ) {
883 0 : uchar serialized[ FD_GOSSIP_VALUE_MAX_SZ ];
884 0 : long serialized_sz = fd_gossip_value_serialize( value, serialized, sizeof(serialized) );
885 0 : FD_TEST( serialized_sz!=-1L );
886 0 : gossip->sign_fn( gossip->sign_ctx, serialized+64UL, (ulong)serialized_sz-64UL, FD_KEYGUARD_SIGN_TYPE_ED25519, serialized );
887 :
888 0 : int origin_active = 0; /* Value doesn't matter, since is_me=1 it's never used. */
889 0 : if( FD_UNLIKELY( fd_crds_insert( gossip->crds, value, serialized, (ulong)serialized_sz, gossip->identity_stake, origin_active, 1, now, stem ) ) ) return -1;
890 :
891 0 : fd_active_set_push( gossip->active_set, serialized, (ulong)serialized_sz, gossip->identity_pubkey, gossip->identity_stake, stem, now, 1 );
892 0 : return 0;
893 0 : }
894 :
895 : int
896 : fd_gossip_push_vote( fd_gossip_t * gossip,
897 : uchar const * txn,
898 : ulong txn_sz,
899 : fd_stem_context_t * stem,
900 0 : long now ) {
901 0 : fd_gossip_value_t value = {
902 0 : .tag = FD_GOSSIP_VALUE_VOTE,
903 0 : .wallclock = (ulong)FD_NANOSEC_TO_MILLI( now ),
904 0 : .vote = {{
905 0 : .index = 0UL, /* TODO */
906 0 : .transaction_len = txn_sz,
907 0 : }}
908 0 : };
909 0 : fd_memcpy( value.origin, gossip->identity_pubkey, 32UL );
910 0 : FD_TEST( txn_sz<=sizeof(value.vote->transaction) );
911 0 : fd_memcpy( value.vote->transaction, txn, txn_sz );
912 :
913 0 : return fd_gossip_push( gossip, &value, stem, now );
914 0 : }
915 :
916 : int
917 : fd_gossip_push_duplicate_shred( fd_gossip_t * gossip,
918 : fd_gossip_duplicate_shred_t const * duplicate_shred,
919 : fd_stem_context_t * stem,
920 0 : long now ) {
921 0 : fd_gossip_value_t value = {
922 0 : .tag = FD_GOSSIP_VALUE_DUPLICATE_SHRED,
923 0 : .wallclock = (ulong)FD_NANOSEC_TO_MILLI( now ),
924 0 : };
925 0 : fd_memcpy( value.origin, gossip->identity_pubkey, 32UL );
926 0 : *value.duplicate_shred = *duplicate_shred;
927 :
928 0 : return fd_gossip_push( gossip, &value, stem, now );
929 0 : }
930 :
931 : static void
932 : tx_ping( fd_gossip_t * gossip,
933 : fd_stem_context_t * stem,
934 : long now,
935 0 : int * charge_busy ) {
936 0 : uchar out_payload[ sizeof(fd_gossip_ping_t) + 4UL ];
937 0 : FD_STORE( uint, out_payload, FD_GOSSIP_MESSAGE_PING );
938 :
939 0 : fd_gossip_ping_t * out_ping = (fd_gossip_ping_t *)( out_payload+4UL );
940 0 : fd_memcpy( out_ping->from, gossip->identity_pubkey, 32UL );
941 :
942 0 : uchar const * peer_pubkey;
943 0 : uchar const * ping_token;
944 0 : fd_ip4_port_t const * peer_address;
945 0 : while( fd_ping_tracker_pop_request( gossip->ping_tracker,
946 0 : now,
947 0 : &peer_pubkey,
948 0 : &peer_address,
949 0 : &ping_token ) ) {
950 0 : fd_memcpy( out_ping->token, ping_token, 32UL );
951 :
952 0 : gossip->sign_fn( gossip->sign_ctx, out_ping->token, 32UL, FD_KEYGUARD_SIGN_TYPE_ED25519, out_ping->signature );
953 0 : gossip->send_fn( gossip->send_ctx, stem, out_payload, sizeof(out_payload), peer_address, (ulong)now );
954 :
955 0 : gossip->metrics->message_tx[ FD_GOSSIP_MESSAGE_PING ]++;
956 0 : gossip->metrics->message_tx_bytes[ FD_GOSSIP_MESSAGE_PING ] += sizeof(out_payload) + 42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
957 0 : if( charge_busy ) *charge_busy = 1;
958 0 : }
959 0 : }
960 :
961 : /* Construct and send a pull request to a random peer. The pull
962 : request contains a bloom filter over our known CRDS hashes so that
963 : the peer can respond with values we are missing.
964 :
965 : NOTE: Divergence from Agave:
966 : - Agave builds up to 2^mask_bits filters per pull period
967 : (sampling up to 1024), each covering a distinct partition of
968 : the hash space. We build and send exactly one filter per
969 : pull period, covering 1/2^mask_bits of the space.
970 :
971 : Maximum bloom filter bits in a PullRequest packet:
972 :
973 : PACKET_DATA_SIZE = 1232 (= 1280 - 40 - 8)
974 :
975 : Bytes consumed by non-bloom fields:
976 : discriminant(4) + keys_len(8) + keys(8*num_keys) +
977 : has_bits(1) + bloom_vec_len(8) + bloom_bits_count(8) +
978 : bloom_num_bits_set(8) + mask(8) + mask_bits(4)
979 : + contact_info_crds_val(crds_val_sz)
980 : = 49 + 8*num_keys + crds_val_sz
981 :
982 : The bitvec is serialized as u64 words, so the bitvec storage is
983 : ceil(num_bits/64)*8 bytes. The remaining packet bytes must
984 : accommodate this.
985 :
986 : Agave determines the max_bytes parameter (input to Bloom::random)
987 : via an empirical cache (get_max_bloom_filter_bytes). max_bytes*8
988 : is passed as the max_bits cap to Bloom::random, but actual
989 : num_bits is only ~83% of max_bits (the E/D ratio for p=0.1).
990 : We replicate this with a closed-form inversion: the largest
991 : max_bytes where ceil(num_bits/64)*8 fits in remaining space is
992 : max_bytes = floor(D * floor(64*W/E) / 8), where W is the max
993 : number of u64 words, E and D are the bloom filter constants.
994 :
995 : num_keys depends on the bloom sizing, which depends on the
996 : overhead, which depends on num_keys. However there is a closed
997 : form: compute num_keys from the pessimistic KEYS=8 overhead, then
998 : recompute the tight overhead with the true num_keys. This always
999 : converges in one step because the optimal key count is
1000 : D*ln(2) ≈ 3.32 (where D = ln(p)/ln(1/2^ln2)), far from any
1001 : rounding boundary. For p=0.1 and KEYS=8, num_keys is always 3.
1002 :
1003 : NB: The has_bits(1) + bloom_vec_len(8) are only written when
1004 : num_bits>=1. fd_bloom_num_bits clamps to [1, max_bits], so
1005 : num_bits>=1 always holds and this layout is correct. */
1006 :
1007 : static void
1008 : tx_pull_request( fd_gossip_t * gossip,
1009 : fd_stem_context_t * stem,
1010 0 : long now ) {
1011 0 : ulong total_crds_vals = fd_crds_len( gossip->crds ) + fd_gossip_purged_len( gossip->purged );
1012 0 : ulong num_items = fd_ulong_max( 65536UL, total_crds_vals );
1013 0 : ulong crds_val_sz = gossip->my_contact_info.crds_val_sz;
1014 :
1015 : /* Step 1: Compute num_keys from the pessimistic KEYS=8 overhead
1016 : (same initial estimate Agave uses in CrdsFilterSet::new). */
1017 0 : ulong pessimistic_overhead = 49UL + 8UL*(ulong)BLOOM_NUM_KEYS + crds_val_sz;
1018 0 : FD_TEST( pessimistic_overhead<FD_GOSSIP_MTU );
1019 0 : double pessimistic_max_bits = (double)( 8UL*( FD_GOSSIP_MTU - pessimistic_overhead ) );
1020 0 : double pessimistic_items = fd_bloom_max_items( pessimistic_max_bits, BLOOM_NUM_KEYS, BLOOM_FALSE_POSITIVE_RATE );
1021 0 : FD_TEST( pessimistic_items>0.0 );
1022 0 : ulong pessimistic_num_bits = fd_bloom_num_bits( pessimistic_items, BLOOM_FALSE_POSITIVE_RATE, pessimistic_max_bits );
1023 0 : ulong num_keys = fd_bloom_num_keys( (double)pessimistic_num_bits, pessimistic_items );
1024 :
1025 : /* Step 2: Recompute with the tight overhead using the true num_keys.
1026 : Find the largest max_bytes parameter (matching Agave's
1027 : get_max_bloom_filter_bytes cache) such that the resulting bitvec
1028 : fits in the remaining packet space.
1029 :
1030 : Given:
1031 : max_items = ceil(max_bits / D) where D = -K / ln(1-exp(ln(p)/K))
1032 : num_bits = ceil(max_items * E) where E = ln(p) / ln(1/2^ln2)
1033 :
1034 : We need ceil(num_bits/64)*8 <= remaining, i.e. num_bits <= 64*W
1035 : where W = floor(remaining/8). Working backwards:
1036 : max_items <= I where I = floor(64*W / E)
1037 : max_bytes <= D*I / 8
1038 :
1039 : So max_bytes = floor(D * floor(64*W/E) / 8). */
1040 0 : ulong overhead = 49UL + 8UL*num_keys + crds_val_sz;
1041 0 : FD_TEST( overhead<FD_GOSSIP_MTU );
1042 0 : ulong remaining = FD_GOSSIP_MTU - overhead;
1043 0 : ulong max_words = remaining / 8UL; /* max u64 words for bitvec */
1044 :
1045 0 : double E = log( BLOOM_FALSE_POSITIVE_RATE ) / log( 1.0 / pow( 2.0, log( 2.0 ) ) );
1046 0 : double D = -BLOOM_NUM_KEYS / log( 1.0 - exp( log( BLOOM_FALSE_POSITIVE_RATE ) / BLOOM_NUM_KEYS ) );
1047 0 : ulong I = (ulong)floor( 64.0 * (double)max_words / E );
1048 0 : ulong max_bytes = (ulong)floor( D * (double)I / 8.0 );
1049 :
1050 0 : double max_bits = (double)( max_bytes * 8UL );
1051 0 : double max_items = fd_bloom_max_items( max_bits, BLOOM_NUM_KEYS, BLOOM_FALSE_POSITIVE_RATE );
1052 0 : FD_TEST( max_items>0.0 );
1053 0 : ulong num_bits = fd_bloom_num_bits( max_items, BLOOM_FALSE_POSITIVE_RATE, max_bits );
1054 0 : FD_TEST( num_bits>=1UL );
1055 0 : FD_TEST( (num_bits+63UL)/64UL<=max_words ); /* verify bitvec fits */
1056 0 : FD_TEST( fd_bloom_num_keys( (double)num_bits, max_items )==num_keys ); /* verify convergence */
1057 0 : FD_TEST( num_keys<=(ulong)BLOOM_NUM_KEYS );
1058 :
1059 0 : double _mask_bits = ceil( log2( (double)num_items / max_items ) );
1060 0 : uint mask_bits = _mask_bits >= 0.0 ? fd_uint_min( (uint)_mask_bits, 63U ) : 0U;
1061 0 : ulong mask = fd_rng_ulong( gossip->rng ) | (~0UL>>(mask_bits));
1062 :
1063 0 : uchar payload[ FD_GOSSIP_MTU ] = {0};
1064 :
1065 0 : uchar * keys_ptr, * bits_ptr, * bits_set;
1066 0 : long payload_sz = fd_gossip_pull_request_init( payload,
1067 0 : FD_GOSSIP_MTU,
1068 0 : num_keys,
1069 0 : num_bits,
1070 0 : mask,
1071 0 : mask_bits,
1072 0 : gossip->my_contact_info.crds_val,
1073 0 : gossip->my_contact_info.crds_val_sz,
1074 0 : &keys_ptr,
1075 0 : &bits_ptr,
1076 0 : &bits_set );
1077 0 : FD_TEST( -1L!=payload_sz );
1078 :
1079 : /* Bloom filter fields are not naturally aligned in the serialized packet.
1080 : Build the filter in aligned storage and copy it into the packet. */
1081 0 : ulong bloom_keys[ (ulong)BLOOM_NUM_KEYS ];
1082 0 : ulong bloom_bits[ (FD_GOSSIP_MTU+sizeof(ulong)-1UL)/sizeof(ulong) ] = {0};
1083 0 : ulong bloom_word_cnt = (num_bits+63UL)/64UL;
1084 0 : fd_bloom_t filter[1];
1085 0 : fd_bloom_init_inplace( bloom_keys, bloom_bits, num_keys, num_bits, 0, gossip->rng, BLOOM_FALSE_POSITIVE_RATE, filter );
1086 :
1087 0 : uchar iter_mem[ 16UL ];
1088 0 : for( fd_crds_mask_iter_t * it = fd_crds_mask_iter_init( gossip->crds, mask, mask_bits, iter_mem );
1089 0 : !fd_crds_mask_iter_done( it, gossip->crds );
1090 0 : it = fd_crds_mask_iter_next( it, gossip->crds ) ) {
1091 0 : fd_bloom_insert( filter, fd_crds_entry_hash( fd_crds_mask_iter_entry( it, gossip->crds ) ), 32UL );
1092 0 : }
1093 :
1094 0 : for( fd_gossip_purged_mask_iter_t * it = fd_gossip_purged_mask_iter_init( gossip->purged, mask, mask_bits, iter_mem );
1095 0 : !fd_gossip_purged_mask_iter_done( it, gossip->purged );
1096 0 : it = fd_gossip_purged_mask_iter_next( it, gossip->purged ) ){
1097 0 : fd_bloom_insert( filter, fd_gossip_purged_mask_iter_hash( it, gossip->purged ), 32UL );
1098 0 : }
1099 :
1100 0 : int num_bits_set = 0;
1101 0 : for( ulong i=0UL; i<bloom_word_cnt; i++ ) num_bits_set += fd_ulong_popcnt( bloom_bits[ i ] );
1102 0 : fd_memcpy( keys_ptr, bloom_keys, num_keys*sizeof(ulong) );
1103 0 : fd_memcpy( bits_ptr, bloom_bits, bloom_word_cnt*sizeof(ulong) );
1104 0 : FD_STORE( ulong, bits_set, (ulong)num_bits_set );
1105 :
1106 0 : ulong idx = fd_gossip_wsample_sample_pull_request( gossip->wsample );
1107 0 : fd_ip4_port_t peer_addr;
1108 0 : if( FD_UNLIKELY( idx==ULONG_MAX ) ) {
1109 0 : if( FD_UNLIKELY( !gossip->entrypoints_cnt ) ) {
1110 : /* We are the bootstrapping node, and nobody else is present in
1111 : the cluster. Nowhere to send the pull request. */
1112 0 : return;
1113 0 : }
1114 0 : peer_addr = random_entrypoint( gossip );
1115 0 : } else {
1116 0 : fd_gossip_contact_info_t const * peer = fd_crds_ci( gossip->crds, idx );
1117 0 : peer_addr.addr = peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].is_ipv6 ? 0 : peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].ip4;
1118 0 : peer_addr.port = peer->sockets[ FD_GOSSIP_CONTACT_INFO_SOCKET_GOSSIP ].port;
1119 0 : }
1120 0 : gossip->send_fn( gossip->send_ctx, stem, payload, (ulong)payload_sz, &peer_addr, (ulong)now );
1121 :
1122 0 : gossip->metrics->message_tx[ FD_GOSSIP_MESSAGE_PULL_REQUEST ]++;
1123 0 : gossip->metrics->message_tx_bytes[ FD_GOSSIP_MESSAGE_PULL_REQUEST ] += (ulong)payload_sz + 42UL; /* 42 = sizeof(fd_ip4_udp_hdrs_t) */
1124 0 : }
1125 :
1126 : void
1127 : fd_gossip_advance( fd_gossip_t * gossip,
1128 : long now,
1129 : fd_stem_context_t * stem,
1130 0 : int * charge_busy ) {
1131 0 : outbound_budget_replenish( gossip, now );
1132 :
1133 0 : fd_gossip_purged_expire( gossip->purged, now );
1134 0 : fd_active_set_advance( gossip->active_set, stem, now, charge_busy );
1135 0 : fd_crds_advance( gossip->crds, now, stem, charge_busy );
1136 :
1137 0 : tx_ping( gossip, stem, now, charge_busy );
1138 0 : if( FD_UNLIKELY( now>=gossip->timers.next_pull_request ) ) {
1139 0 : tx_pull_request( gossip, stem, now );
1140 0 : if( charge_busy ) *charge_busy = 1;
1141 : /* 1.6ms (625/s). Agave sends min(1024, ceil(2^mask_bits/8))
1142 : filters every 500ms. For a typical mainnet table (~65k items,
1143 : mask_bits≈7) that is ~16 filters/500ms = one every 31ms. We
1144 : send a single filter per round, so we fire ~20× more often to
1145 : compensate for sending one filter instead of many per period.
1146 :
1147 : We considered dynamically matching Agave's exact rate by
1148 : computing 500ms/filters_per_round from mask_bits each round,
1149 : but this caused slow table fill on startup (mask_bits starts
1150 : low -> long intervals -> few pulls -> slow CRDS population).
1151 : Adaptive boosting (counter-based, timestamp-based, and
1152 : threshold-based) all added complexity without clear benefit:
1153 : counter decay lost state between send and response arrival,
1154 : timestamp checks never disarmed because trickle inserts kept
1155 : refreshing the window, and threshold heuristics required
1156 : tuning constants that varied by cluster size.
1157 :
1158 : A fixed 1.6ms is simpler and robust: the cost of a redundant
1159 : pull request is negligible (a single 1232-byte packet whose
1160 : reply will be empty if we're already caught up), and it
1161 : guarantees fast table fill on startup without any adaptive
1162 : machinery. */
1163 0 : gossip->timers.next_pull_request = now+1600L*1000L;
1164 0 : }
1165 0 : if( FD_UNLIKELY( now>=gossip->timers.next_contact_info_refresh ) ) {
1166 : /* TODO: Frequency of this? More often if observing? */
1167 0 : refresh_contact_info( gossip, now );
1168 0 : int origin_active = 0; /* Value doesn't matter, since is_me=1 it's never used. */
1169 0 : fd_crds_insert( gossip->crds, gossip->my_contact_info.ci, gossip->my_contact_info.crds_val, gossip->my_contact_info.crds_val_sz, gossip->identity_stake, origin_active, 1, now, stem );
1170 0 : fd_active_set_push( gossip->active_set, gossip->my_contact_info.crds_val, gossip->my_contact_info.crds_val_sz, gossip->identity_pubkey, gossip->identity_stake, stem, now, 1 );
1171 0 : gossip->timers.next_contact_info_refresh = now+15L*500L*1000L*1000L; /* TODO: Jitter */
1172 0 : if( charge_busy ) *charge_busy = 1;
1173 0 : }
1174 0 : }
1175 :
1176 : void
1177 : fd_gossip_ping_tracker_track( fd_gossip_t * gossip,
1178 : uchar const * peer_pubkey,
1179 : fd_ip4_port_t peer_address,
1180 0 : long now ) {
1181 : /* Don't track ourselves. */
1182 0 : if( FD_UNLIKELY( !memcmp( peer_pubkey, gossip->identity_pubkey, 32UL ) ) ) return;
1183 :
1184 0 : ulong origin_stake = get_stake( gossip, peer_pubkey );
1185 0 : fd_ping_tracker_track( gossip->ping_tracker, peer_pubkey, origin_stake, peer_address, now );
1186 0 : }
|
