Line data Source code
1 : #define _GNU_SOURCE
2 : #include "utils/fd_ssctrl.h"
3 : #include "utils/fd_ssload.h"
4 : #include "utils/fd_ssmsg.h"
5 : #include "utils/fd_ssparse.h"
6 : #include "utils/fd_ssmanifest_parser.h"
7 : #include "utils/fd_wfs.h"
8 : #include "utils/fd_slot_delta_parser.h"
9 : #include "../../util/fd_hash32.h"
10 : #include "../../util/bits/fd_float.h"
11 : #include "../../util/bits/fd_uwide.h"
12 :
13 : #include "../../disco/topo/fd_topo.h"
14 : #include "../../disco/metrics/fd_metrics.h"
15 : #include "../../disco/gui/fd_gui_config_parse.h"
16 : #include "../../flamenco/runtime/fd_txncache.h"
17 : #include "../../flamenco/runtime/fd_system_ids.h"
18 : #include "../../flamenco/runtime/fd_hashes.h"
19 : #include "../../flamenco/runtime/sysvar/fd_sysvar_cache.h"
20 : #include "../../flamenco/runtime/sysvar/fd_sysvar_epoch_schedule.h"
21 : #include "../../flamenco/runtime/sysvar/fd_sysvar_slot_history.h"
22 : #include "../../flamenco/rewards/fd_rewards_base.h"
23 :
24 : #include "../../flamenco/runtime/fd_bank.h"
25 : #include "../../flamenco/features/fd_features.h"
26 : #include "../../flamenco/stakes/fd_stake_types.h"
27 : #include "../../disco/stem/fd_stem.h"
28 : #include "../../flamenco/accdb/fd_accdb.h"
29 : #include "../../disco/events/generated/fd_event_gen.h"
30 :
31 : #include <linux/futex.h>
32 : #include "generated/fd_snapin_tile_seccomp.h"
33 :
34 : #include <errno.h>
35 : #include <fcntl.h>
36 : #include <unistd.h>
37 :
38 : #define NAME "snapin"
39 :
40 0 : #define FD_SNAPIN_WRITE_BUF_SZ (16UL<<20)
41 : #define FD_SNAPIN_WRITE_ACCOUNT_MAX (FD_SNAPIN_WRITE_BUF_SZ/sizeof(fd_accdb_disk_meta_t))
42 : /* Accounts are written with O_DIRECT so that snapin tiles do not
43 : serialize on the inode lock taken by buffered writes. Each flush is
44 : padded to FD_SNAPIN_DIRECT_ALIGN with a dead record header. The
45 : buffer fills only to FD_SNAPIN_WRITE_BUF_MAX so the header always
46 : fits. */
47 0 : #define FD_SNAPIN_DIRECT_ALIGN (4096UL)
48 0 : #define FD_SNAPIN_WRITE_BUF_MAX (FD_SNAPIN_WRITE_BUF_SZ-sizeof(fd_accdb_disk_meta_t))
49 :
50 : FD_STATIC_ASSERT( FD_SNAPIN_WRITE_BUF_SZ%FD_SNAPIN_DIRECT_ALIGN==0UL, write_buf_align );
51 :
52 : /* Raw state values exported by the metrics (states= enums) */
53 : FD_STATIC_ASSERT( FD_SNAPSHOT_STATE_IDLE==FD_METRICS_ENUM_SNAPSHOT_STREAM_STATE_V_IDLE_IDX, metrics_state );
54 : FD_STATIC_ASSERT( FD_SNAPSHOT_STATE_PROCESSING==FD_METRICS_ENUM_SNAPSHOT_STREAM_STATE_V_PROCESSING_IDX, metrics_state );
55 : FD_STATIC_ASSERT( FD_SNAPSHOT_STATE_FINISHING==FD_METRICS_ENUM_SNAPSHOT_STREAM_STATE_V_FINISHING_IDX, metrics_state );
56 : FD_STATIC_ASSERT( FD_SNAPSHOT_STATE_ERROR==FD_METRICS_ENUM_SNAPSHOT_STREAM_STATE_V_ERROR_IDX, metrics_state );
57 : FD_STATIC_ASSERT( FD_SNAPSHOT_STATE_SHUTDOWN==FD_METRICS_ENUM_SNAPSHOT_STREAM_STATE_V_SHUTDOWN_IDX, metrics_state );
58 : FD_STATIC_ASSERT( FD_SNAPSHOT_DATA_MTU<FD_SNAPIN_WRITE_BUF_MAX, write_buf );
59 : FD_STATIC_ASSERT( sizeof(fd_accdb_disk_meta_t)+FD_RUNTIME_ACC_SZ_MAX<=FD_SNAPIN_WRITE_BUF_MAX, max_account );
60 :
61 : /* The snapin tiles are state machines that parse and load a full and
62 : optionally an incremental snapshot. They are responsible for loading
63 : accounts into the accounts database and writing their records to
64 : disk. */
65 :
66 : struct fd_blockhash_entry {
67 : fd_hash_t blockhash;
68 :
69 : struct {
70 : ulong prev;
71 : ulong next;
72 : } map;
73 : };
74 :
75 : typedef struct fd_blockhash_entry fd_blockhash_entry_t;
76 :
77 : #define MAP_NAME blockhash_map
78 0 : #define MAP_KEY blockhash
79 : #define MAP_KEY_T fd_hash_t
80 : #define MAP_ELE_T fd_blockhash_entry_t
81 0 : #define MAP_KEY_EQ(k0,k1) (!memcmp((k0),(k1), sizeof(fd_hash_t)))
82 0 : #define MAP_KEY_HASH(key,seed) (fd_hash32( (key)->uc, (seed) ))
83 0 : #define MAP_PREV map.prev
84 0 : #define MAP_NEXT map.next
85 : #define MAP_OPTIMIZE_RANDOM_ACCESS_REMOVAL 1
86 : #include "../../util/tmpl/fd_map_chain.c"
87 :
88 : /* One link of the txncache chain rebuilt by populate_txncache, from one
89 : entry of the snapshot's blockhash queue. */
90 :
91 : struct fd_snapin_bank {
92 : ulong hash_index;
93 : uchar blockhash[ 32UL ];
94 : fd_txncache_fork_id_t fork_id;
95 : ulong txnhash_offset;
96 : };
97 :
98 : typedef struct fd_snapin_bank fd_snapin_bank_t;
99 :
100 : #define SORT_NAME fd_snapin_bank_sort_newest_first
101 0 : #define SORT_KEY_T fd_snapin_bank_t
102 0 : #define SORT_BEFORE(a,b) ((a).hash_index>(b).hash_index)
103 : #include "../../util/tmpl/fd_sort.c"
104 :
105 : /* For a transaction to be valid to be inserted into the txncache, it
106 : must reference a blockhash that is in the set of recent blockhashes.
107 : This means that only transactions executed in the latest 151 rooted
108 : blocks can be in the txncache. The remaining entries can be ignored.
109 :
110 : A slot delta holds the transactions of one block, each twice ,once as
111 : signature and once as message hash. When slot deltas are ingested,
112 : their entries are staged in a grid of FD_TXNCACHE_MAX_SLOT_DELTAS
113 : rows of txncache_max_entries_per_slot entries per row, one row per
114 : retained slot delta. */
115 :
116 : FD_STATIC_ASSERT( FD_TXNCACHE_MAX_SLOT_DELTAS<=FD_SLOT_DELTA_MAX_ENTRIES, txncache_staging_slot_cnt );
117 :
118 : /* A blockhash group is one (slot, blockhash) pair in the status cache,
119 : holding the txnhash offset for that blockhash and the entries of the
120 : transactions executed in that slot which referenced the blockhash.
121 : Agave inserts every processed transaction into the status cache twice
122 : (once with message hash and once with signature), both times under
123 : the same blockhash and slot. Groups can reference any blockhash,
124 : including durable nonces that are not in the recent blockhash queue.
125 : So an honest slot has at most txncache_max_groups_per_slot groups
126 : (config->limits.max_txn_per_slot). Snapshots exceeding this are
127 : rejected as malformed. */
128 :
129 : struct blockhash_group {
130 : uchar blockhash[ 32UL ];
131 : uint txnhash_offset;
132 : uint txncache_entry_cnt;
133 : };
134 :
135 : typedef struct blockhash_group blockhash_group_t;
136 :
137 : FD_STATIC_ASSERT( sizeof(blockhash_group_t)==40UL, blockhash_group );
138 :
139 : /* Each slot delta's groups are staged in the txncache at a fixed
140 : stride of txncache_staging_row_sz bytes. */
141 :
142 : FD_FN_CONST static inline ulong
143 0 : txncache_staging_row_sz( ulong max_groups_per_slot ) {
144 0 : return fd_ulong_align_up( max_groups_per_slot*sizeof(blockhash_group_t), FD_TXNCACHE_SNAPIN_STAGE_ALIGN );
145 0 : }
146 :
147 : /* After filtering with the recent blockhash queue, there is at most one
148 : group per retained (slot, recent blockhash) pair. Filtering can only
149 : happen after we receive the manifest, hence we still need
150 : FD_TXNCACHE_MAX_SLOT_DELTAS*txncache_max_groups_per_slot worst case
151 : slot deltas buffered. */
152 0 : #define FD_SNAPIN_MAX_RECENT_GROUPS (FD_TXNCACHE_MAX_SLOT_DELTAS*FD_TXNCACHE_MAX_SLOT_DELTAS)
153 :
154 : struct recent_blockhash_group {
155 : ulong blockhash_bank_i;
156 : ulong execution_slot;
157 : ulong execution_bank_i;
158 : ulong txnhash_offset;
159 : ulong txncache_entry_idx;
160 : ulong txncache_entry_cnt;
161 : };
162 :
163 : typedef struct recent_blockhash_group recent_blockhash_group_t;
164 :
165 : struct txncache_staging_slot {
166 : ulong slot;
167 : ulong entry_cnt;
168 : ulong group_cnt;
169 : };
170 :
171 : typedef struct txncache_staging_slot txncache_staging_slot_t;
172 :
173 : struct fd_snapin_out_link {
174 : ulong idx;
175 : fd_wksp_t * mem;
176 : ulong chunk0;
177 : ulong wmark;
178 : ulong chunk;
179 : ulong mtu;
180 : };
181 : typedef struct fd_snapin_out_link fd_snapin_out_link_t;
182 :
183 : struct fd_snapin_account_batch {
184 : ulong cnt;
185 : ulong lamports [ FD_SNAPIN_WRITE_ACCOUNT_MAX ];
186 : uint slots [ FD_SNAPIN_WRITE_ACCOUNT_MAX ];
187 : uint data_lens [ FD_SNAPIN_WRITE_ACCOUNT_MAX ];
188 : int executables [ FD_SNAPIN_WRITE_ACCOUNT_MAX ];
189 : };
190 :
191 : typedef struct fd_snapin_account_batch fd_snapin_account_batch_t;
192 :
193 : /* Only tile 0 uses this state. */
194 : struct fd_snapin_lead {
195 : uint init_completed : 1; /* did INIT complete for this attempt? */
196 :
197 : ulong seed;
198 : long boot_timestamp;
199 :
200 : fd_txncache_t * txncache;
201 : fd_bank_t * bank;
202 :
203 : fd_ssmanifest_parser_t * manifest_parser;
204 : fd_slot_delta_parser_t * slot_delta_parser;
205 :
206 : struct {
207 : int manifest_done;
208 : int status_cache_done;
209 : int manifest_processed;
210 : } flags;
211 :
212 : ulong advertised_slot;
213 : ulong bank_slot;
214 : ulong epoch;
215 :
216 : fd_epoch_schedule_t epoch_schedule;
217 :
218 : ulong full_genesis_creation_time_seconds;
219 : uchar advertised_hash[ FD_HASH_FOOTPRINT ];
220 :
221 : ulong manifest_capitalization; /* capitalization according to the current snapshot manifest */
222 :
223 : struct {
224 : ulong loaded;
225 : ulong duplicates;
226 : } account_counts;
227 :
228 : struct {
229 : ulong capitalization;
230 : fd_accdb_snapshot_recovery_t accdb_metadata;
231 : } recovery; /* stores state from the last full snapshot for incremental revert */
232 :
233 : fd_sysvar_cache_t sysvar_cache[1]; /* verify_sysvars scratch */
234 :
235 : blockhash_group_t * blockhash_groups; /* groups of one staged slot delta */
236 : ulong blockhash_groups_cnt; /* every group parsed, including those from dropped slots */
237 : recent_blockhash_group_t * recent_groups;
238 : ulong recent_groups_len;
239 :
240 : fd_sstxncache_hash_t * txncache_entries;
241 : txncache_staging_slot_t txncache_slots[ FD_TXNCACHE_MAX_SLOT_DELTAS ];
242 : ulong txncache_slots_len;
243 : ulong txncache_current_slot_idx;
244 : ulong txncache_current_slot_group_cnt;
245 : ulong txncache_current_slot_entry_cnt; /* entries of the slot delta being parsed, retained or not */
246 : ulong txncache_max_groups_per_slot; /* config->limits.max_txn_per_slot */
247 : ulong txncache_max_entries_per_slot; /* 2x, signature and message hash entries */
248 :
249 : fd_accdb_fork_id_t accdb_root_fork_id;
250 : fd_accdb_fork_id_t accdb_incr_fork_id; /* child fork for incremental writes (purge on failure) */
251 : fd_txncache_fork_id_t txncache_root_fork_id;
252 :
253 : fd_snapin_out_link_t manifest_out;
254 :
255 : /* Roll back a failure at the next INIT. */
256 : struct {
257 : int pending;
258 : int full; /* failed attempt type */
259 : fd_accdb_fork_id_t fork; /* failed incremental fork */
260 : } rollback;
261 : };
262 :
263 : typedef struct fd_snapin_lead fd_snapin_lead_t;
264 :
265 : /* Shared state for parallel snapin tiles. */
266 : struct fd_snapin_shmem {
267 : /* Tile 0 publishes the accdb fork before acknowledging INIT. */
268 : ulong fork_id;
269 :
270 : /* Stake delegations fork for incremental writes, USHORT_MAX for full. */
271 : ushort stake_fork;
272 :
273 : /* Per-tile attempt values. Lamport totals are 128-bit <*_hi,*>:
274 : they count every account version, which can pass 2^64. */
275 : struct __attribute__((aligned(128))) {
276 : ulong loaded;
277 : ulong duplicates;
278 : ulong input_lamports;
279 : ulong duplicate_lamports;
280 : ulong input_lamports_hi;
281 : ulong duplicate_lamports_hi;
282 : } values[ FD_TOPO_MAX_TILE_IN_LINKS ];
283 :
284 : /* Atomic index of the next unclaimed appendvec. */
285 : ulong next_appendvec_ticket __attribute__((aligned(128)));
286 : };
287 :
288 : typedef struct fd_snapin_shmem fd_snapin_shmem_t;
289 :
290 : struct fd_snapin_tile {
291 : int state;
292 : uint full : 1; /* loading a full snapshot? */
293 :
294 : fd_snapin_lead_t lead;
295 :
296 : ulong tile_idx; /* tile kind ID */
297 : ulong lane_cnt;
298 : ulong expected_frame;
299 : ulong pending_control; /* control message expected from snapdc tiles */
300 : uchar control_seen[ FD_TOPO_MAX_TILE_IN_LINKS ];
301 :
302 : fd_accdb_t * accdb;
303 :
304 : /* Shared stake data updated by every tile. */
305 : fd_stake_delegations_t * stake_delegations;
306 :
307 : fd_ssparse_t ssparse[1];
308 :
309 : struct {
310 : ulong full_bytes_read;
311 : ulong incremental_bytes_read;
312 :
313 : /* Account counters (full + incremental) */
314 : ulong accounts_loaded;
315 : ulong accounts_replaced;
316 : ulong accounts_ignored;
317 :
318 : /* Account counters (snapshot taken for full snapshot only) */
319 : ulong full_accounts_loaded;
320 : ulong full_accounts_replaced;
321 : ulong full_accounts_ignored;
322 :
323 : /* Persistent counters */
324 : ulong total_accounts_processed;
325 : ulong total_account_batches_processed;
326 : ulong disk_bytes_written;
327 : } metrics;
328 :
329 : struct {
330 : fd_wksp_t * wksp;
331 : ulong chunk0;
332 : ulong wmark;
333 : ulong mtu;
334 : ulong pos;
335 : } in[ FD_TOPO_MAX_TILE_IN_LINKS ];
336 :
337 : fd_snapin_out_link_t ct_out;
338 : fd_snapin_out_link_t gui_out;
339 :
340 : /* Shared snapshot state. */
341 : fd_snapin_shmem_t * shmem;
342 :
343 : /* Parse state for one attempt. */
344 : ulong appendvec_seq; /* next appendvec number */
345 : ulong claimed_appendvec; /* current claim */
346 : ulong incr_fork; /* insert fork; USHORT_MAX for full */
347 :
348 : struct {
349 : uchar buf[ FD_SNAPIN_WRITE_BUF_SZ ] __attribute__((aligned(FD_SNAPIN_DIRECT_ALIGN)));
350 : ulong buf_used;
351 : int accdb_direct_fd; /* O_DIRECT fd to accounts.db */
352 : fd_snapin_account_batch_t batch;
353 : } writer;
354 :
355 : /* Buffer streamed account data before inserting. */
356 : struct {
357 : int executable;
358 : ulong slot;
359 : ulong lamports;
360 : ulong data_len;
361 : ulong bytes_received;
362 : uchar pubkey[ 32UL ];
363 : uchar owner [ 32UL ];
364 : uchar data[ FD_RUNTIME_ACC_SZ_MAX ] __attribute__((aligned(64)));
365 : } staged;
366 :
367 : ulong wfs_slot;
368 : fd_hash_t wfs_bank_hash;
369 : ushort wfs_shred_version;
370 : int wfs_hash_is_zero;
371 : };
372 :
373 : typedef struct fd_snapin_tile fd_snapin_tile_t;
374 :
375 : static inline int
376 0 : is_lead( fd_snapin_tile_t const * ctx ) {
377 0 : return ctx->tile_idx==0UL;
378 0 : }
379 :
380 : static void
381 0 : format_count( char * out, ulong out_sz, ulong n ) {
382 0 : if( n>=1000000UL ) FD_TEST( fd_cstr_printf_check( out, out_sz, NULL, "%.1fM", (double)n/1e6 ) );
383 0 : else if( n>=1000UL ) FD_TEST( fd_cstr_printf_check( out, out_sz, NULL, "%.1fK", (double)n/1e3 ) );
384 0 : else FD_TEST( fd_cstr_printf_check( out, out_sz, NULL, "%lu", n ) );
385 0 : }
386 :
387 : static inline int
388 0 : should_shutdown( fd_snapin_tile_t * ctx ) {
389 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN && is_lead( ctx ) ) ) {
390 0 : long elapsed_ns = fd_log_wallclock() - ctx->lead.boot_timestamp;
391 0 : char loaded_buf[ 32 ];
392 0 : char dup_buf [ 32 ];
393 0 : format_count( loaded_buf, sizeof(loaded_buf), ctx->lead.account_counts.loaded );
394 0 : format_count( dup_buf, sizeof(dup_buf), ctx->lead.account_counts.duplicates );
395 0 : FD_LOG_NOTICE(( "loaded %s accounts %s(%s dups)%s from snapshot in %.3f seconds",
396 0 : loaded_buf, fd_log_style_dim(), dup_buf, fd_log_style_normal(), (double)elapsed_ns/1e9 ));
397 0 : }
398 0 : return ctx->state==FD_SNAPSHOT_STATE_SHUTDOWN;
399 0 : }
400 :
401 : static ulong
402 0 : scratch_align( void ) {
403 0 : return FD_SNAPIN_DIRECT_ALIGN;
404 0 : }
405 :
406 : static ulong
407 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
408 0 : ulong l = FD_LAYOUT_INIT;
409 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapin_tile_t), sizeof(fd_snapin_tile_t) );
410 0 : l = FD_LAYOUT_APPEND( l, fd_accdb_align(), fd_accdb_footprint( tile->snapin.max_live_slots, 0 ) );
411 :
412 : /* Only tile 0 publishes the manifest */
413 0 : if( FD_LIKELY( !tile->kind_id ) ) {
414 0 : l = FD_LAYOUT_APPEND( l, fd_txncache_align(), fd_txncache_footprint( tile->snapin.max_live_slots ) );
415 0 : l = FD_LAYOUT_APPEND( l, fd_ssmanifest_parser_align(), fd_ssmanifest_parser_footprint() );
416 0 : l = FD_LAYOUT_APPEND( l, fd_slot_delta_parser_align(), fd_slot_delta_parser_footprint() );
417 0 : l = FD_LAYOUT_APPEND( l, alignof(recent_blockhash_group_t), sizeof(recent_blockhash_group_t)*FD_SNAPIN_MAX_RECENT_GROUPS );
418 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_sstxncache_hash_t), sizeof(fd_sstxncache_hash_t)*FD_TXNCACHE_MAX_SLOT_DELTAS*2UL*tile->snapin.max_txn_per_slot );
419 0 : l = FD_LAYOUT_APPEND( l, FD_TXNCACHE_SNAPIN_STAGE_ALIGN, txncache_staging_row_sz( tile->snapin.max_txn_per_slot ) );
420 0 : }
421 0 : return FD_LAYOUT_FINI( l, scratch_align() );
422 0 : }
423 :
424 : static void
425 0 : metrics_write( fd_snapin_tile_t * ctx ) {
426 0 : fd_accdb_flush_metrics( ctx->accdb );
427 :
428 0 : FD_MGAUGE_SET( SNAPIN, STREAM_STATE, (ulong)ctx->state );
429 0 : FD_MGAUGE_SET( SNAPIN, PARSE_PROGRESS_BYTES_FULL, ctx->metrics.full_bytes_read );
430 0 : FD_MGAUGE_SET( SNAPIN, PARSE_PROGRESS_BYTES_INCREMENTAL, ctx->metrics.incremental_bytes_read );
431 0 : FD_MCNT_SET ( SNAPIN, DISK_WRITTEN_BYTES, ctx->metrics.disk_bytes_written );
432 0 : FD_MGAUGE_SET( SNAPIN, ACCOUNT_PROGRESS_NEW, ctx->metrics.accounts_loaded );
433 0 : FD_MGAUGE_SET( SNAPIN, ACCOUNT_PROGRESS_REPLACED, ctx->metrics.accounts_replaced );
434 0 : FD_MGAUGE_SET( SNAPIN, ACCOUNT_PROGRESS_IGNORED, ctx->metrics.accounts_ignored );
435 0 : FD_MCNT_SET ( SNAPIN, ACCOUNT_PROCESSED, ctx->metrics.total_accounts_processed );
436 0 : FD_MCNT_SET ( SNAPIN, ACCOUNT_BATCH_PROCESSED, ctx->metrics.total_account_batches_processed );
437 0 : }
438 :
439 : /* verify_slot_deltas_with_slot_history verifies the 'SlotHistory'
440 : sysvar account (data, data_len) after loading a snapshot. Returns 0
441 : if verification passed, -1 if not. */
442 :
443 : static int
444 : verify_slot_deltas_with_slot_history( fd_snapin_tile_t * ctx,
445 : uchar const * data,
446 0 : ulong data_len ) {
447 0 : if( FD_UNLIKELY( data_len!=FD_SYSVAR_SLOT_HISTORY_BINCODE_SZ ) ) {
448 0 : FD_LOG_WARNING(( "SlotHistory sysvar account data size is %lu, expected %lu", data_len, FD_SYSVAR_SLOT_HISTORY_BINCODE_SZ ));
449 0 : return -1;
450 0 : }
451 :
452 0 : fd_slot_history_view_t view[1];
453 0 : if( FD_UNLIKELY( !fd_sysvar_slot_history_view( view, data, data_len ) ) ) {
454 0 : FD_LOG_WARNING(( "SlotHistory sysvar account data is corrupt" ));
455 0 : return -1;
456 0 : }
457 :
458 : /* Sanity checks for slot history:
459 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L586 */
460 :
461 0 : ulong newest_slot = view->next_slot - 1UL;
462 0 : if( FD_UNLIKELY( newest_slot!=ctx->lead.bank_slot ) ) {
463 : /* VerifySlotHistoryError::InvalidNewestSlot
464 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L621 */
465 0 : FD_LOG_WARNING(( "SlotHistory sysvar has an invalid newest slot: %lu != bank slot: %lu", newest_slot, ctx->lead.bank_slot ));
466 0 : return -1;
467 0 : }
468 :
469 0 : if( FD_UNLIKELY( view->bits_len!=FD_SLOT_HISTORY_MAX_ENTRIES ) ) {
470 : /* VerifySlotHistoryError::InvalidNumEntries
471 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L625 */
472 0 : FD_LOG_WARNING(( "SlotHistory sysvar has invalid number of entries: %lu != expected: %lu", view->bits_len, FD_SLOT_HISTORY_MAX_ENTRIES ));
473 0 : return -1;
474 0 : }
475 :
476 : /* Stricter than Agave, which only checks bits_len (the bv crate
477 : merely requires bits_len<=blocks_len*64). An exact block count is
478 : what makes fd_sysvar_slot_history_find_slot's (slot/64)%blocks_len
479 : indexing agree with Agave's bits.get(slot%MAX_ENTRIES); the runtime
480 : always writes exactly this many blocks. */
481 0 : if( FD_UNLIKELY( view->blocks_len!=FD_SLOT_HISTORY_MAX_ENTRIES/64UL ) ) {
482 0 : FD_LOG_WARNING(( "SlotHistory sysvar has invalid bitvec block count: %lu != expected: %lu", view->blocks_len, FD_SLOT_HISTORY_MAX_ENTRIES/64UL ));
483 0 : return -1;
484 0 : }
485 :
486 : /* All slots in slot deltas should be present in the slot history */
487 0 : fd_slot_delta_slot_set_t slot_set = fd_slot_delta_parser_slot_set( ctx->lead.slot_delta_parser );
488 0 : for( ulong i=0UL; i<slot_set.ele_cnt; i++ ) {
489 0 : ulong slot = slot_set.pool[ i ].slot;
490 0 : if( FD_UNLIKELY( fd_sysvar_slot_history_find_slot( view, slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) ) {
491 : /* VerifySlotDeltasError::SlotNotFoundInHistory
492 : https://github.com/anza-xyz/agave/blob/v3.1.8/snapshots/src/error.rs#L144
493 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L593 */
494 0 : FD_LOG_WARNING(( "slot %lu missing from SlotHistory sysvar account", slot ));
495 0 : return -1;
496 0 : }
497 0 : }
498 :
499 0 : ulong scan_cnt = fd_ulong_min( view->next_slot, FD_SLOT_HISTORY_MAX_ENTRIES );
500 0 : ulong found_cnt = 0UL;
501 0 : for( ulong i=0UL; i<scan_cnt && found_cnt<FD_SLOT_DELTA_MAX_ENTRIES; i++ ) {
502 0 : ulong slot = newest_slot - i;
503 0 : if( FD_UNLIKELY( fd_sysvar_slot_history_find_slot( view, slot )!=FD_SLOT_HISTORY_SLOT_FOUND ) ) continue;
504 0 : found_cnt++;
505 0 : if( FD_UNLIKELY( slot_set_ele_query( slot_set.map, &slot, NULL, slot_set.pool )==NULL ) ) {
506 : /* VerifySlotDeltasError::SlotNotFoundInDeltas
507 : https://github.com/anza-xyz/agave/blob/v3.1.8/snapshots/src/error.rs#L147
508 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L609 */
509 0 : FD_LOG_WARNING(( "slot %lu missing from slot deltas but present in SlotHistory", slot ));
510 0 : return -1;
511 0 : }
512 0 : }
513 :
514 0 : return 0;
515 0 : }
516 :
517 : /* Sysvar accounts checked by verify_sysvars. Agave requires Rent at
518 : restore; Firedancer also needs Clock and SlotHistory. */
519 :
520 : struct snapin_sysvar {
521 : fd_pubkey_t const * id;
522 : char const * name;
523 : int required;
524 : };
525 : typedef struct snapin_sysvar snapin_sysvar_t;
526 :
527 : static snapin_sysvar_t const snapin_sysvar_tbl[ FD_SYSVAR_CACHE_ENTRY_CNT ] = {
528 : [ FD_SYSVAR_clock_IDX ] = { &fd_sysvar_clock_id, "Clock", 1 },
529 : [ FD_SYSVAR_epoch_rewards_IDX ] = { &fd_sysvar_epoch_rewards_id, "EpochRewards", 0 },
530 : [ FD_SYSVAR_epoch_schedule_IDX ] = { &fd_sysvar_epoch_schedule_id, "EpochSchedule", 0 },
531 : [ FD_SYSVAR_last_restart_slot_IDX ] = { &fd_sysvar_last_restart_slot_id, "LastRestartSlot", 0 },
532 : [ FD_SYSVAR_recent_hashes_IDX ] = { &fd_sysvar_recent_block_hashes_id, "RecentBlockhashes", 0 },
533 : [ FD_SYSVAR_rent_IDX ] = { &fd_sysvar_rent_id, "Rent", 1 },
534 : [ FD_SYSVAR_slot_hashes_IDX ] = { &fd_sysvar_slot_hashes_id, "SlotHashes", 0 },
535 : [ FD_SYSVAR_stake_history_IDX ] = { &fd_sysvar_stake_history_id, "StakeHistory", 0 },
536 : };
537 :
538 : /* Rent::try_minimum_balance bounds
539 : https://github.com/anza-xyz/solana-sdk/blob/rent%40v4.4.0/rent/src/lib.rs#L187-L204 */
540 :
541 : #define SNAPIN_RENT_MAX_LAMPORTS_PER_BYTE_THRESHOLD_1 (1759197129867UL)
542 : #define SNAPIN_RENT_MAX_LAMPORTS_PER_BYTE_THRESHOLD_2 ( 879598564933UL)
543 :
544 : static int
545 0 : verify_rent( fd_sysvar_cache_t const * cache ) {
546 0 : fd_rent_t rent[1];
547 0 : fd_sysvar_cache_rent_read( cache, rent ); /* required */
548 0 : ulong threshold_bits = fd_dblbits( rent->exemption_threshold );
549 0 : if( FD_UNLIKELY( ( threshold_bits==fd_dblbits( 1.0 ) && rent->lamports_per_uint8_year>SNAPIN_RENT_MAX_LAMPORTS_PER_BYTE_THRESHOLD_1 ) ||
550 0 : ( threshold_bits==fd_dblbits( 2.0 ) && rent->lamports_per_uint8_year>SNAPIN_RENT_MAX_LAMPORTS_PER_BYTE_THRESHOLD_2 ) ) ) {
551 0 : FD_LOG_WARNING(( "Rent sysvar lamports_per_byte %lu overflows minimum_balance for exemption_threshold %g",
552 0 : rent->lamports_per_uint8_year, rent->exemption_threshold ));
553 0 : return -1;
554 0 : }
555 0 : return 0;
556 0 : }
557 :
558 : /* fd_sysvar_slot_hashes_update needs the full-size account */
559 :
560 : static int
561 : verify_slot_hashes( fd_sysvar_cache_t const * cache,
562 0 : ulong data_len ) {
563 0 : if( FD_UNLIKELY( fd_sysvar_cache_slot_hashes_is_valid( cache ) && data_len<FD_SYSVAR_SLOT_HASHES_BINCODE_SZ ) ) {
564 0 : FD_LOG_WARNING(( "SlotHashes sysvar account data size is %lu, expected at least %lu",
565 0 : data_len, FD_SYSVAR_SLOT_HASHES_BINCODE_SZ ));
566 0 : return -1;
567 0 : }
568 0 : return 0;
569 0 : }
570 :
571 : /* An active EpochRewards sysvar drives fd_rewards_recalculate_partitioned_rewards
572 : at boot; check what that path asserts. */
573 :
574 : static int
575 : verify_epoch_rewards( fd_snapin_tile_t const * ctx,
576 0 : fd_sysvar_cache_t const * cache ) {
577 0 : fd_sysvar_epoch_rewards_t rewards[1];
578 0 : if( !fd_sysvar_cache_epoch_rewards_read( cache, rewards ) || !rewards->active ) return 0;
579 :
580 : /* https://github.com/anza-xyz/agave/blob/v4.4.0-alpha.4/runtime/src/bank/partitioned_epoch_rewards/sysvar.rs#L117 */
581 0 : if( FD_UNLIKELY( rewards->distributed_rewards>rewards->total_rewards ) ) {
582 0 : FD_LOG_WARNING(( "EpochRewards sysvar distributed rewards %lu exceed total rewards %lu",
583 0 : rewards->distributed_rewards, rewards->total_rewards ));
584 0 : return -1;
585 0 : }
586 : /* Bounds asserted by fd_stake_rewards_init and
587 : fd_distribute_partitioned_epoch_rewards */
588 0 : ulong epoch_slot_cnt = fd_epoch_slot_cnt( &ctx->lead.epoch_schedule, ctx->lead.epoch );
589 0 : if( FD_UNLIKELY( !rewards->num_partitions ||
590 0 : rewards->num_partitions>MAX_PARTITIONS_PER_EPOCH ||
591 0 : rewards->num_partitions>=epoch_slot_cnt ) ) {
592 0 : FD_LOG_WARNING(( "EpochRewards sysvar has invalid partition count %lu (epoch has %lu slots)",
593 0 : rewards->num_partitions, epoch_slot_cnt ));
594 0 : return -1;
595 0 : }
596 0 : if( FD_UNLIKELY( rewards->distribution_starting_block_height>ULONG_MAX-rewards->num_partitions ) ) {
597 0 : FD_LOG_WARNING(( "EpochRewards sysvar distribution starting block height %lu overflows with %lu partitions",
598 0 : rewards->distribution_starting_block_height, rewards->num_partitions ));
599 0 : return -1;
600 0 : }
601 : /* read_stake_history in fd_rewards.c aborts without it */
602 0 : if( FD_UNLIKELY( !fd_sysvar_cache_stake_history_is_valid( cache ) ) ) {
603 0 : FD_LOG_WARNING(( "EpochRewards sysvar is active but the StakeHistory sysvar account is not present" ));
604 0 : return -1;
605 0 : }
606 0 : return 0;
607 0 : }
608 :
609 : /* read_sysvar reads a sysvar account into ctx->staged.data. Returns 1
610 : if present, 0 if absent and optional, -1 if absent and required or
611 : not owned by the sysvar program. */
612 :
613 : static int
614 : read_sysvar( fd_snapin_tile_t * ctx,
615 : fd_accdb_fork_id_t fork_id,
616 : snapin_sysvar_t const * sysvar,
617 : ulong * lamports,
618 0 : ulong * data_len ) {
619 0 : int executable;
620 0 : uchar owner[ 32UL ];
621 0 : int source = fd_accdb_read_one_nocache( ctx->accdb, fork_id, sysvar->id->uc,
622 0 : lamports, &executable, owner,
623 0 : ctx->staged.data, data_len );
624 0 : if( source==FD_ACCDB_READ_ONE_NOCACHE_MISS ) {
625 0 : if( FD_UNLIKELY( sysvar->required ) ) {
626 0 : FD_LOG_WARNING(( "%s sysvar account was not present in the accounts database", sysvar->name ));
627 0 : return -1;
628 0 : }
629 0 : return 0;
630 0 : }
631 :
632 0 : if( FD_UNLIKELY( !fd_memeq( owner, fd_sysvar_owner_id.uc, sizeof(fd_pubkey_t) ) ) ) {
633 0 : FD_BASE58_ENCODE_32_BYTES( owner, owner_b58 );
634 0 : FD_LOG_WARNING(( "%s sysvar owner is invalid: %s != sysvar_owner_id", sysvar->name, owner_b58 ));
635 0 : return -1;
636 0 : }
637 0 : return 1;
638 0 : }
639 :
640 : /* verify_sysvars reads the sysvar accounts back from the accounts
641 : database and checks they decode and satisfy the invariants replay
642 : asserts at boot. Call after all FINI acks. Returns 0 on success,
643 : -1 on failure. */
644 :
645 : static int
646 0 : verify_sysvars( fd_snapin_tile_t * ctx ) {
647 0 : fd_accdb_fork_id_t fork_id = ctx->full ? ctx->lead.accdb_root_fork_id
648 0 : : ctx->lead.accdb_incr_fork_id;
649 0 : fd_sysvar_cache_t * cache = fd_sysvar_cache_join( fd_sysvar_cache_new( ctx->lead.sysvar_cache ) );
650 :
651 0 : ulong data_lens[ FD_SYSVAR_CACHE_ENTRY_CNT ] = {0}; /* untruncated */
652 :
653 0 : for( ulong i=0UL; i<FD_SYSVAR_CACHE_ENTRY_CNT; i++ ) {
654 0 : snapin_sysvar_t const * sysvar = &snapin_sysvar_tbl[ i ];
655 :
656 0 : ulong lamports;
657 0 : ulong data_len;
658 0 : int found = read_sysvar( ctx, fork_id, sysvar, &lamports, &data_len );
659 0 : if( FD_UNLIKELY( found<0 ) ) return -1;
660 0 : if( !found ) continue;
661 :
662 : /* Same decode as fd_sysvar_cache_restore at boot */
663 0 : fd_sysvar_cache_restore_one( cache, sysvar->id, lamports, ctx->staged.data, data_len );
664 0 : ulong cached_sz;
665 0 : if( FD_UNLIKELY( !fd_sysvar_cache_data_query( cache, sysvar->id->uc, &cached_sz ) ) ) {
666 0 : FD_LOG_WARNING(( "%s sysvar account data is corrupt (data_len=%lu)", sysvar->name, data_len ));
667 0 : return -1;
668 0 : }
669 :
670 0 : data_lens[ i ] = data_len;
671 0 : }
672 :
673 0 : if( FD_UNLIKELY( verify_rent ( cache ) ) ) return -1;
674 0 : if( FD_UNLIKELY( verify_slot_hashes ( cache, data_lens[ FD_SYSVAR_slot_hashes_IDX ] ) ) ) return -1;
675 0 : if( FD_UNLIKELY( verify_epoch_rewards( ctx, cache ) ) ) return -1;
676 :
677 0 : snapin_sysvar_t const slot_history = { &fd_sysvar_slot_history_id, "SlotHistory", 1 };
678 0 : ulong lamports;
679 0 : ulong data_len;
680 0 : if( FD_UNLIKELY( read_sysvar( ctx, fork_id, &slot_history, &lamports, &data_len )<0 ) ) return -1;
681 0 : return verify_slot_deltas_with_slot_history( ctx, ctx->staged.data, data_len );
682 0 : }
683 :
684 : /* verification of epoch stakes from manifest
685 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L632 */
686 : static int
687 0 : verify_epoch_stakes( fd_snapshot_manifest_t const * manifest ) {
688 0 : fd_epoch_schedule_t epoch_schedule = (fd_epoch_schedule_t){
689 0 : .slots_per_epoch = manifest->epoch_schedule_params.slots_per_epoch,
690 0 : .leader_schedule_slot_offset = manifest->epoch_schedule_params.leader_schedule_slot_offset,
691 0 : .warmup = manifest->epoch_schedule_params.warmup,
692 0 : .first_normal_epoch = manifest->epoch_schedule_params.first_normal_epoch,
693 0 : .first_normal_slot = manifest->epoch_schedule_params.first_normal_slot,
694 0 : };
695 :
696 0 : ulong min_required_epoch = fd_slot_to_epoch( &epoch_schedule, manifest->slot, NULL );
697 0 : ulong max_required_epoch = fd_slot_to_leader_schedule_epoch( &epoch_schedule, manifest->slot );
698 :
699 : /* ensure all required epochs are present in epoch stakes */
700 0 : for( ulong i=min_required_epoch; i<=max_required_epoch; i++ ) {
701 0 : int found = 0;
702 0 : for( ulong j=0UL; j<FD_RUNTIME_MANIFEST_EPOCH_STAKES_LEN; j++ ) {
703 0 : if( manifest->epoch_stakes[j].epoch==i ) {
704 0 : found = 1;
705 0 : break;
706 0 : }
707 0 : }
708 :
709 0 : if( FD_UNLIKELY( !found ) ) {
710 : /* VerifyEpochStakesError::StakesNotFound
711 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L667 */
712 0 : FD_LOG_WARNING(( "stakes not found for epoch %lu in manifest", i ));
713 0 : return -1;
714 0 : }
715 0 : }
716 :
717 0 : return 0;
718 0 : }
719 :
720 : static int
721 : verify_slot_deltas_with_bank_slot( fd_snapin_tile_t * ctx,
722 0 : ulong bank_slot ) {
723 0 : fd_slot_delta_slot_set_t slot_set = fd_slot_delta_parser_slot_set( ctx->lead.slot_delta_parser );
724 0 : for( ulong i=0UL; i<slot_set.ele_cnt; i++ ) {
725 0 : ulong slot = slot_set.pool[ i ].slot;
726 : /* VerifySlotDeltasError::SlotGreaterThanMaxRoot
727 : https://github.com/anza-xyz/agave/blob/v3.1.8/snapshots/src/error.rs#L138
728 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L550 */
729 0 : if( FD_UNLIKELY( slot>bank_slot ) ) {
730 0 : FD_LOG_WARNING(( "entry slot %lu is greater than bank slot %lu", slot, bank_slot ));
731 0 : return -1;
732 0 : }
733 0 : }
734 0 : return 0;
735 0 : }
736 :
737 : static int
738 : verify_bank_hash( fd_snapin_tile_t const * ctx,
739 0 : fd_snapshot_manifest_t const * manifest ) {
740 0 : if( FD_UNLIKELY( manifest->blockhashes_len==0UL ) ) {
741 0 : FD_LOG_WARNING(( "%s manifest for epoch %lu and slot %lu has no blockhashes",
742 0 : ctx->full?"full":"incr", ctx->lead.epoch, manifest->slot ));
743 0 : return -1;
744 0 : }
745 :
746 0 : if( FD_UNLIKELY( !manifest->has_accounts_lthash ) ) {
747 0 : FD_LOG_WARNING(( "%s manifest for epoch %lu and slot %lu is missing accounts lthash",
748 0 : ctx->full?"full":"incr", ctx->lead.epoch, manifest->slot ));
749 0 : return -1;
750 0 : }
751 :
752 : /* find the last blockhash */
753 0 : ulong max_hash_idx = 0UL;
754 0 : ulong last_bh_idx = 0UL;
755 0 : for( ulong i=0UL; i<manifest->blockhashes_len; i++ ) {
756 0 : if( FD_LIKELY( manifest->blockhashes[ i ].hash_index > max_hash_idx ) ) {
757 0 : max_hash_idx = manifest->blockhashes[ i ].hash_index;
758 0 : last_bh_idx = i;
759 0 : }
760 0 : }
761 :
762 : /* fd_lthash_value_t is aligned to 64B but the accounts_lthash in the
763 : manifest may not be because its simply a uchar array. Copy is
764 : needed to avoid undefined behavior. */
765 0 : fd_lthash_value_t accounts_lthash[ 1UL ];
766 0 : fd_memcpy( accounts_lthash, manifest->accounts_lthash, sizeof(fd_lthash_value_t) );
767 :
768 0 : fd_hash_t const * parent_bank_hash = (fd_hash_t const *)fd_type_pun_const( manifest->parent_bank_hash );
769 0 : fd_hash_t const * last_blockhash = (fd_hash_t const *)fd_type_pun_const( manifest->blockhashes[ last_bh_idx ].hash );
770 0 : fd_hash_t computed_bank_hash[ 1UL ];
771 0 : fd_hashes_hash_bank( accounts_lthash, parent_bank_hash, last_blockhash, manifest->signature_count, computed_bank_hash );
772 0 : fd_hashes_apply_hard_forks(
773 0 : computed_bank_hash,
774 0 : manifest->slot,
775 0 : manifest->parent_slot,
776 0 : manifest->hard_forks,
777 0 : manifest->hard_fork_cnt );
778 :
779 0 : if( FD_UNLIKELY( memcmp( computed_bank_hash, manifest->bank_hash, FD_HASH_FOOTPRINT ) ) ) {
780 0 : FD_BASE58_ENCODE_32_BYTES( computed_bank_hash->hash, computed_bank_hash_enc );
781 0 : FD_BASE58_ENCODE_32_BYTES( manifest->bank_hash, manifest_bank_hash_enc );
782 0 : FD_LOG_WARNING(( "%s manifest for epoch %lu and slot %lu bank hash verification failed: computed %s does not match manifest %s",
783 0 : ctx->full?"full":"incr", ctx->lead.epoch, manifest->slot,
784 0 : computed_bank_hash_enc, manifest_bank_hash_enc ));
785 0 : return -1;
786 0 : }
787 :
788 0 : return 0;
789 0 : }
790 :
791 : static inline void
792 0 : clear_control_barrier( fd_snapin_tile_t * ctx ) {
793 0 : ctx->pending_control = ULONG_MAX;
794 0 : fd_memset( ctx->control_seen, 0, sizeof(ctx->control_seen) );
795 0 : }
796 :
797 : static void
798 : transition_malformed( fd_snapin_tile_t * ctx,
799 0 : fd_stem_context_t * stem ) {
800 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) return;
801 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
802 0 : fd_stem_publish( stem, ctx->ct_out.idx, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
803 0 : }
804 :
805 : /* Writes (write!=0) or reads the staged groups of slot delta slot_idx
806 : between the in-memory buffer and the txncache. No-op if slot_idx is
807 : ULONG_MAX. */
808 :
809 : static void
810 : txncache_staging_row_io( fd_snapin_tile_t * ctx,
811 : ulong slot_idx,
812 0 : int write ) {
813 0 : if( FD_UNLIKELY( slot_idx==ULONG_MAX ) ) return;
814 0 : ulong sz = fd_ulong_align_up( ctx->lead.txncache_slots[ slot_idx ].group_cnt*sizeof(blockhash_group_t), FD_TXNCACHE_SNAPIN_STAGE_ALIGN );
815 0 : if( FD_LIKELY( sz ) ) fd_txncache_snapin_stage( ctx->lead.txncache, slot_idx*txncache_staging_row_sz( ctx->lead.txncache_max_groups_per_slot ), ctx->lead.blockhash_groups, sz, write );
816 0 : }
817 :
818 : static void
819 0 : txncache_staging_reset( fd_snapin_tile_t * ctx ) {
820 0 : ctx->lead.blockhash_groups_cnt = 0UL;
821 0 : ctx->lead.recent_groups_len = 0UL;
822 0 : ctx->lead.txncache_slots_len = 0UL;
823 0 : ctx->lead.txncache_current_slot_entry_cnt = 0UL;
824 0 : ctx->lead.txncache_current_slot_idx = ULONG_MAX;
825 0 : ctx->lead.txncache_current_slot_group_cnt = 0UL;
826 0 : }
827 :
828 : static ulong
829 : txncache_staging_slot_begin( fd_snapin_tile_t * ctx,
830 0 : ulong slot ) {
831 0 : txncache_staging_row_io( ctx, ctx->lead.txncache_current_slot_idx, 1 );
832 :
833 0 : ulong candidate_idx;
834 0 : if( FD_LIKELY( ctx->lead.txncache_slots_len<FD_TXNCACHE_MAX_SLOT_DELTAS ) ) {
835 0 : candidate_idx = ctx->lead.txncache_slots_len++;
836 0 : } else {
837 0 : candidate_idx = 0UL;
838 0 : for( ulong i=1UL; i<FD_TXNCACHE_MAX_SLOT_DELTAS; i++ ) {
839 0 : if( ctx->lead.txncache_slots[ i ].slot<ctx->lead.txncache_slots[ candidate_idx ].slot ) candidate_idx = i;
840 0 : }
841 0 : if( FD_UNLIKELY( slot<ctx->lead.txncache_slots[ candidate_idx ].slot ) ) candidate_idx = ULONG_MAX;
842 0 : }
843 :
844 0 : if( FD_LIKELY( candidate_idx!=ULONG_MAX ) ) {
845 0 : ctx->lead.txncache_slots[ candidate_idx ].slot = slot;
846 0 : ctx->lead.txncache_slots[ candidate_idx ].entry_cnt = 0UL;
847 0 : ctx->lead.txncache_slots[ candidate_idx ].group_cnt = 0UL;
848 0 : }
849 0 : ctx->lead.txncache_current_slot_idx = candidate_idx;
850 0 : ctx->lead.txncache_current_slot_entry_cnt = 0UL;
851 0 : ctx->lead.txncache_current_slot_group_cnt = 0UL;
852 0 : return candidate_idx;
853 0 : }
854 :
855 : static int
856 : txncache_staging_group_begin( fd_snapin_tile_t * ctx,
857 : uchar const * blockhash,
858 0 : ulong txnhash_offset ) {
859 0 : if( FD_UNLIKELY( ctx->lead.txncache_current_slot_group_cnt>=ctx->lead.txncache_max_groups_per_slot ) ) return -1;
860 0 : ctx->lead.txncache_current_slot_group_cnt++;
861 0 : ctx->lead.blockhash_groups_cnt++;
862 :
863 0 : ulong slot_idx = ctx->lead.txncache_current_slot_idx;
864 0 : if( FD_UNLIKELY( slot_idx==ULONG_MAX ) ) return 0;
865 :
866 0 : FD_TEST( slot_idx<ctx->lead.txncache_slots_len );
867 0 : txncache_staging_slot_t * staging_slot = &ctx->lead.txncache_slots[ slot_idx ];
868 0 : FD_TEST( staging_slot->group_cnt<ctx->lead.txncache_max_groups_per_slot );
869 0 : blockhash_group_t * group = &ctx->lead.blockhash_groups[ staging_slot->group_cnt ];
870 0 : memcpy( group->blockhash, blockhash, 32UL );
871 0 : group->txnhash_offset = (uint)txnhash_offset;
872 0 : group->txncache_entry_cnt = 0U;
873 0 : staging_slot->group_cnt++;
874 0 : return 0;
875 0 : }
876 :
877 : static int
878 : txncache_staging_entry_add( fd_snapin_tile_t * ctx,
879 : ulong slot,
880 0 : uchar const * txnhash ) {
881 0 : if( FD_UNLIKELY( ctx->lead.txncache_current_slot_entry_cnt>=ctx->lead.txncache_max_entries_per_slot ) ) return -1;
882 0 : ctx->lead.txncache_current_slot_entry_cnt++;
883 :
884 0 : ulong slot_idx = ctx->lead.txncache_current_slot_idx;
885 0 : if( FD_UNLIKELY( slot_idx==ULONG_MAX ) ) return 0; /* discarded older delta */
886 :
887 0 : FD_TEST( slot_idx<ctx->lead.txncache_slots_len );
888 0 : txncache_staging_slot_t * staging_slot = &ctx->lead.txncache_slots[ slot_idx ];
889 0 : FD_TEST( staging_slot->slot==slot );
890 0 : FD_TEST( staging_slot->group_cnt );
891 0 : FD_TEST( staging_slot->entry_cnt<ctx->lead.txncache_max_entries_per_slot );
892 0 : blockhash_group_t * group = &ctx->lead.blockhash_groups[ staging_slot->group_cnt-1UL ];
893 0 : memcpy( ctx->lead.txncache_entries[ slot_idx*ctx->lead.txncache_max_entries_per_slot+staging_slot->entry_cnt ].txnhash, txnhash, sizeof(fd_sstxncache_hash_t) );
894 0 : staging_slot->entry_cnt++;
895 0 : group->txncache_entry_cnt++;
896 0 : return 0;
897 0 : }
898 :
899 : static ulong
900 : txncache_staging_rank_slots( fd_snapin_tile_t const * ctx,
901 0 : ulong execution_bank_i_by_staging_idx[ static FD_TXNCACHE_MAX_SLOT_DELTAS ] ) {
902 0 : ulong staging_slot_cnt = ctx->lead.txncache_slots_len;
903 0 : ulong staging_idx_by_bank_i[ FD_TXNCACHE_MAX_SLOT_DELTAS ];
904 0 : for( ulong staging_idx=0UL; staging_idx<staging_slot_cnt; staging_idx++ ) {
905 0 : ulong bank_i = staging_idx;
906 0 : while( bank_i>0UL && ctx->lead.txncache_slots[ staging_idx_by_bank_i[ bank_i-1UL ] ].slot<ctx->lead.txncache_slots[ staging_idx ].slot ) {
907 0 : staging_idx_by_bank_i[ bank_i ] = staging_idx_by_bank_i[ bank_i-1UL ];
908 0 : bank_i--;
909 0 : }
910 0 : staging_idx_by_bank_i[ bank_i ] = staging_idx;
911 0 : }
912 0 : for( ulong bank_i=0UL; bank_i<staging_slot_cnt; bank_i++ ) {
913 0 : execution_bank_i_by_staging_idx[ staging_idx_by_bank_i[ bank_i ] ] = bank_i;
914 0 : }
915 0 : return staging_slot_cnt ? ctx->lead.txncache_slots[ staging_idx_by_bank_i[ 0UL ] ].slot : ULONG_MAX;
916 0 : }
917 :
918 : static int
919 : txncache_staging_filter_groups( fd_snapin_tile_t * ctx,
920 : blockhash_map_t const * blockhash_map,
921 : fd_blockhash_entry_t const * blockhash_pool,
922 0 : ulong snapshot_slot ) {
923 0 : ulong execution_bank_i_by_staging_idx[ FD_TXNCACHE_MAX_SLOT_DELTAS ];
924 0 : ulong newest_staged_slot = txncache_staging_rank_slots( ctx, execution_bank_i_by_staging_idx );
925 0 : if( FD_UNLIKELY( newest_staged_slot!=ULONG_MAX && newest_staged_slot!=snapshot_slot ) ) {
926 0 : FD_LOG_WARNING(( "corrupt snapshot: newest slot delta is for slot %lu, not the snapshot slot %lu", newest_staged_slot, snapshot_slot ));
927 0 : return 1;
928 0 : }
929 :
930 0 : txncache_staging_row_io( ctx, ctx->lead.txncache_current_slot_idx, 1 );
931 :
932 0 : ctx->lead.recent_groups_len = 0UL;
933 0 : for( ulong staging_idx=0UL; staging_idx<ctx->lead.txncache_slots_len; staging_idx++ ) {
934 0 : txncache_staging_slot_t const * staging_slot = &ctx->lead.txncache_slots[ staging_idx ];
935 0 : txncache_staging_row_io( ctx, staging_idx, 0 );
936 0 : blockhash_group_t const * groups = ctx->lead.blockhash_groups;
937 0 : ulong entry_idx = staging_idx*ctx->lead.txncache_max_entries_per_slot;
938 :
939 0 : for( ulong i=0UL; i<staging_slot->group_cnt; i++ ) {
940 0 : blockhash_group_t const * group = &groups[ i ];
941 0 : fd_hash_t key;
942 0 : fd_memcpy( key.uc, group->blockhash, 32UL );
943 0 : fd_blockhash_entry_t const * entry = blockhash_map_ele_query_const( blockhash_map, &key, NULL, blockhash_pool );
944 0 : if( FD_LIKELY( entry ) ) {
945 0 : if( FD_UNLIKELY( ctx->lead.recent_groups_len>=FD_SNAPIN_MAX_RECENT_GROUPS ) ) {
946 0 : FD_LOG_WARNING(( "corrupt snapshot: more than %lu blockhash groups reference recent blockhashes", FD_SNAPIN_MAX_RECENT_GROUPS ));
947 0 : return -1;
948 0 : }
949 0 : recent_blockhash_group_t * recent = &ctx->lead.recent_groups[ ctx->lead.recent_groups_len++ ];
950 0 : recent->blockhash_bank_i = (ulong)(entry-blockhash_pool);
951 0 : recent->execution_slot = staging_slot->slot;
952 0 : recent->execution_bank_i = execution_bank_i_by_staging_idx[ staging_idx ];
953 0 : recent->txnhash_offset = group->txnhash_offset;
954 0 : recent->txncache_entry_idx = entry_idx;
955 0 : recent->txncache_entry_cnt = group->txncache_entry_cnt;
956 0 : }
957 0 : entry_idx += group->txncache_entry_cnt;
958 0 : }
959 0 : FD_TEST( entry_idx==staging_idx*ctx->lead.txncache_max_entries_per_slot+staging_slot->entry_cnt );
960 0 : }
961 0 : return 0;
962 0 : }
963 :
964 : static int
965 : populate_txncache( fd_snapin_tile_t * ctx,
966 : fd_snapshot_manifest_blockhash_t const blockhashes[ static FD_BLOCKHASHES_MAX ],
967 : ulong blockhashes_len,
968 0 : ulong snapshot_slot ) {
969 : /* Our txncache internally contains the fork structure for the chain,
970 : which we need to recreate here. Because snapshots are only served
971 : for rooted slots, there is actually no forking, and the bank forks
972 : are just a single bank, the root, like
973 :
974 : _root
975 :
976 : But the txncache also must contain the 150 more recent banks prior
977 : to the root (151 rooted banks total), looking like,
978 :
979 :
980 : _root_150 -> _root_149 -> ... -> _root_2 -> _root_1 -> _root
981 :
982 : Our txncache is "slot agnostic" meaning there is no concept of a
983 : slot number in it. It just has a fork tree structure. So long as
984 : the fork tree is isomorphic to the actual bank forks, and each bank
985 : has the correct blockhash, it works.
986 :
987 : So the challenge is simply to create this chain of 151 forks in the
988 : txncache, with correct blockhashes, and then insert all the
989 : transactions into it.
990 :
991 : Constructing the chain of blockhashes is easy. It is just the
992 : BLOCKHASH_QUEUE array in the manifest. This array is unfortunately
993 : not sorted and appears in random order, but it has a hash_index
994 : field, starting at some arbitrary offset, so we can back out the
995 : blockhashes we need from this, by first finding the max hash_index
996 : as _max and then collecting hash entries via,
997 :
998 : _root_150 -> _root_149 -> ... -> _root_2 -> _root_1 -> _root
999 : _max-150 -> _max-149 -> ... -> _max-2 -> _max-1 -> _max
1000 :
1001 : Usually the index is gapless and this is exactly 151 links. An
1002 : index can be missing, though: a hole (see fd_blockhashes.h), left
1003 : when the same blockhash was registered again and Agave's queue kept
1004 : only the newer index. A hole has no blockhash of its own, so it is
1005 : not a link. The chain is the live entries in descending hash_index
1006 : order, and the recent window is still bounded by index distance, so
1007 : it is the live entries within _max-150 ... _max, which may be fewer
1008 : than 151 links.
1009 :
1010 : Now the remaining problem is inserting transactions into this
1011 : chain. Remember each transaction needs to be inserted with:
1012 :
1013 : (a) The fork ID (position of the bank in the chain) it was executed in.
1014 : (b) The blockhash of the bank it referenced.
1015 :
1016 : (b) is trivial to retrieve, as it's in the actual slot_deltas entry
1017 : in the manifest served by Agave. But (a) is mildly annoying. Agave
1018 : serves slot_deltas based on slot, so we need an additional mapping
1019 : from slot to position in our banks chain. It turns out we have to
1020 : go to yet another structure in the manifest to retrieve this, the
1021 : ancestors array. This is just an array of slot values, so we need
1022 : to sort it, and line it up against our banks chain like so (shown
1023 : for the usual gapless case),
1024 :
1025 : _root_150 -> _root_149 -> ... -> _root_2 -> _root_1 -> _root
1026 : _max-150 -> _max-149 -> ... -> _max-2 -> _max-1 -> _max
1027 : _slots_150 -> _slots_149 -> ... -> _slots_2 -> _slots_1 -> _slots
1028 :
1029 : That is what populate does, with one shortcut. Every rooted slot
1030 : with a block has a slot delta, and every rooted block registered
1031 : exactly one blockhash, consuming one hash_index. Both sequences end
1032 : at the snapshot slot, so the slot delta of rank r (newest first)
1033 : executed in the block that registered hash_index _max-r, without
1034 : consulting the ancestors array. That index is turned into a chain
1035 : position through the queue: a live index is its own link, and a
1036 : hole (see fd_blockhashes.h) maps to the nearest newer live link,
1037 : since the block that registered it shares that link's blockhash.
1038 :
1039 : From there we are done.
1040 :
1041 : The Agave status cache stores 20 bytes of each transaction hash.
1042 : We retrieve the offset from the slot_deltas groups and key it into
1043 : the bank of the referenced blockhash in our chain, checking that
1044 : every group for a blockhash agrees on it. */
1045 :
1046 0 : if( FD_UNLIKELY( blockhashes_len>FD_BLOCKHASHES_MAX ) ) {
1047 0 : FD_LOG_WARNING(( "corrupt snapshot: blockhash queue length %lu exceeds maximum %lu", blockhashes_len, FD_BLOCKHASHES_MAX ));
1048 0 : return 1;
1049 0 : }
1050 0 : if( FD_UNLIKELY( !blockhashes_len ) ) {
1051 0 : FD_LOG_WARNING(( "corrupt snapshot: blockhash queue is empty" ));
1052 0 : return 1;
1053 0 : }
1054 :
1055 : /* hash_index values may skip (see fd_blockhashes.h). A skipped index
1056 : has no blockhash of its own, so it is not a link: the chain is the
1057 : entries in descending hash_index order, so copy them out and sort.
1058 : The manifest buffer is shared with other tiles, so it is not
1059 : reordered in place. The newest index is seq_max, and the recent
1060 : window is still bounded by index distance, so it is the entries
1061 : within seq_max-150 ... seq_max. */
1062 :
1063 0 : fd_snapin_bank_t banks[ FD_BLOCKHASHES_MAX ];
1064 0 : for( ulong i=0UL; i<blockhashes_len; i++ ) {
1065 0 : banks[ i ].hash_index = blockhashes[ i ].hash_index;
1066 0 : banks[ i ].fork_id.val = USHORT_MAX;
1067 0 : banks[ i ].txnhash_offset = ULONG_MAX;
1068 0 : memcpy( banks[ i ].blockhash, blockhashes[ i ].hash, 32UL );
1069 0 : }
1070 0 : fd_snapin_bank_sort_newest_first_inplace( banks, blockhashes_len );
1071 :
1072 0 : ulong chain_cnt = blockhashes_len;
1073 0 : ulong seq_max = banks[ 0UL ].hash_index;
1074 0 : ulong recent_cnt = 0UL;
1075 0 : for( ulong i=0UL; i<chain_cnt; i++ ) {
1076 0 : if( FD_UNLIKELY( i && banks[ i ].hash_index==banks[ i-1UL ].hash_index ) ) {
1077 0 : FD_LOG_WARNING(( "corrupt snapshot: duplicate blockhash queue hash_index %lu", banks[ i ].hash_index ));
1078 0 : return 1;
1079 0 : }
1080 0 : if( FD_LIKELY( seq_max-banks[ i ].hash_index<151UL ) ) recent_cnt++;
1081 0 : }
1082 0 : FD_TEST( chain_cnt==blockhashes_len );
1083 0 : FD_TEST( recent_cnt<=151UL );
1084 :
1085 0 : ulong chain_len = recent_cnt;
1086 :
1087 : /* Chain position of the block that registered each hash_index, by
1088 : slot delta rank r (hash_index seq_max-r): the newest link whose
1089 : hash_index is at most seq_max-r. For a present index that is its
1090 : own link; for a skipped one it is the nearest newer link, since the
1091 : block that registered it shares that link's blockhash. Ranks past
1092 : the oldest entry cannot reference a recent blockhash; they map past
1093 : the chain so that the age check below rejects them if they try. */
1094 :
1095 0 : ulong seq_span = seq_max - banks[ chain_cnt-1UL ].hash_index; /* oldest entry is the last link */
1096 0 : ulong exec_pos_by_rank[ FD_TXNCACHE_MAX_SLOT_DELTAS ];
1097 0 : ulong link = 0UL;
1098 0 : for( ulong r=0UL; r<FD_TXNCACHE_MAX_SLOT_DELTAS; r++ ) {
1099 0 : if( FD_UNLIKELY( r>seq_span ) ) { exec_pos_by_rank[ r ] = chain_cnt; continue; }
1100 0 : ulong target = seq_max-r; /* in [oldest,seq_max], so the walk stops before chain_cnt */
1101 0 : while( banks[ link ].hash_index>target ) link++;
1102 0 : exec_pos_by_rank[ r ] = fd_ulong_if( banks[ link ].hash_index==target, link, link-1UL ); /* link>0 when skipped: seq_max is link 0 */
1103 0 : }
1104 :
1105 : /* Now we need a hashset of just the 151 most recent blockhashes,
1106 : anything else is a nonce transaction which we do not insert, or an
1107 : already expired transaction which can also be discarded. */
1108 :
1109 0 : uchar __attribute__((aligned(alignof(blockhash_map_t)))) _map[ blockhash_map_footprint( 1024UL ) ];
1110 0 : blockhash_map_t * blockhash_map = blockhash_map_join( blockhash_map_new( _map, 1024UL, ctx->lead.seed ) );
1111 0 : if( FD_UNLIKELY( !blockhash_map ) ) FD_LOG_ERR(( "failed to create blockhash map" ));
1112 :
1113 0 : fd_blockhash_entry_t blockhash_pool[ 151UL ];
1114 0 : for( ulong i=0UL; i<chain_len; i++ ) {
1115 0 : fd_memcpy( blockhash_pool[ i ].blockhash.uc, banks[ i ].blockhash, 32UL );
1116 :
1117 0 : if( FD_UNLIKELY( blockhash_map_ele_query_const( blockhash_map, &blockhash_pool[ i ].blockhash, NULL, blockhash_pool ) ) ) {
1118 0 : FD_BASE58_ENCODE_32_BYTES( banks[ i ].blockhash, blockhash_b58 );
1119 0 : FD_LOG_WARNING(( "corrupt snapshot: duplicate blockhash %s in 151 most recent blockhashes", blockhash_b58 ));
1120 0 : return 1;
1121 0 : }
1122 :
1123 0 : blockhash_map_ele_insert( blockhash_map, &blockhash_pool[ i ], blockhash_pool );
1124 0 : }
1125 :
1126 : /* The staged groups live in the txncache, so filtering must finish
1127 : before the first txncache insert. */
1128 0 : if( FD_UNLIKELY( txncache_staging_filter_groups( ctx, blockhash_map, blockhash_pool, snapshot_slot ) ) ) return 1;
1129 0 : for( ulong i=0UL; i<ctx->lead.recent_groups_len; i++ ) {
1130 0 : recent_blockhash_group_t * group = &ctx->lead.recent_groups[ i ];
1131 0 : FD_TEST( group->execution_bank_i<FD_TXNCACHE_MAX_SLOT_DELTAS );
1132 0 : group->execution_bank_i = exec_pos_by_rank[ group->execution_bank_i ];
1133 0 : }
1134 :
1135 : /* Now load the blockhash offsets for these blockhashes ... */
1136 0 : if( FD_UNLIKELY( !ctx->lead.blockhash_groups_cnt ) ) {
1137 0 : fd_slot_delta_slot_set_t ss = fd_slot_delta_parser_slot_set( ctx->lead.slot_delta_parser );
1138 : /* Rooted slots with no groups represent an empty status cache.
1139 : No restored entry needs a hash offset, so finalization uses zero. */
1140 0 : if( FD_UNLIKELY( !ss.ele_cnt ) ) {
1141 0 : FD_LOG_WARNING(( "corrupt snapshot: no blockhash offsets found (rooted_slots=%lu)", ss.ele_cnt ));
1142 0 : return 1;
1143 0 : }
1144 0 : FD_LOG_WARNING(( "status cache has no blockhash groups (rooted_slots=%lu); defaulting transaction hash offsets to zero", ss.ele_cnt ));
1145 0 : }
1146 0 : for( ulong i=0UL; i<ctx->lead.recent_groups_len; i++ ) {
1147 0 : recent_blockhash_group_t const * group = &ctx->lead.recent_groups[ i ];
1148 0 : ulong blockhash_bank_i = group->blockhash_bank_i;
1149 0 : FD_TEST( blockhash_bank_i<chain_len );
1150 :
1151 0 : if( FD_UNLIKELY( banks[ blockhash_bank_i ].txnhash_offset!=ULONG_MAX && banks[ blockhash_bank_i ].txnhash_offset!=group->txnhash_offset ) ) {
1152 0 : FD_BASE58_ENCODE_32_BYTES( banks[ blockhash_bank_i ].blockhash, blockhash_b58 );
1153 0 : FD_LOG_WARNING(( "corrupt snapshot: conflicting txnhash offsets for blockhash %s", blockhash_b58 ));
1154 0 : return 1;
1155 0 : }
1156 :
1157 0 : if( FD_UNLIKELY( group->txncache_entry_cnt && blockhash_bank_i<=group->execution_bank_i ) ) {
1158 0 : FD_BASE58_ENCODE_32_BYTES( banks[ blockhash_bank_i ].blockhash, blockhash_b58 );
1159 0 : FD_LOG_WARNING(( "corrupt snapshot: %lu status cache entries of slot %lu reference blockhash %s, which is not older than that slot", group->txncache_entry_cnt, group->execution_slot, blockhash_b58 ));
1160 0 : return 1;
1161 0 : }
1162 :
1163 0 : banks[ blockhash_bank_i ].txnhash_offset = group->txnhash_offset;
1164 0 : }
1165 :
1166 : /* Construct the linear fork chain in the txncache. */
1167 :
1168 0 : fd_txncache_fork_id_t parent = { .val = USHORT_MAX };
1169 0 : for( ulong i=0UL; i<chain_len; i++ ) banks[ chain_len-1UL-i ].fork_id = parent = fd_txncache_attach_child( ctx->lead.txncache, parent );
1170 0 : for( ulong i=0UL; i<chain_len; i++ ) fd_txncache_attach_blockhash( ctx->lead.txncache, banks[ i ].fork_id, banks[ i ].blockhash );
1171 :
1172 0 : for( ulong i=0UL; i<ctx->lead.recent_groups_len; i++ ) {
1173 0 : recent_blockhash_group_t const * group = &ctx->lead.recent_groups[ i ];
1174 0 : fd_sstxncache_hash_t const * entries = &ctx->lead.txncache_entries[ group->txncache_entry_idx ];
1175 0 : for( ulong j=0UL; j<group->txncache_entry_cnt; j++ ) {
1176 0 : fd_txncache_insert( ctx->lead.txncache, banks[ group->execution_bank_i ].fork_id, banks[ group->blockhash_bank_i ].blockhash, entries[ j ].txnhash );
1177 0 : }
1178 0 : }
1179 :
1180 : /* Then finalize all the banks (freezing them) and setting the txnhash
1181 : offset so future queries use the correct offset. If the offset is
1182 : ULONG_MAX this is valid, it means the blockhash had no transactions
1183 : in it, so there's nothing in the status cache under that blockhash.
1184 :
1185 : Just set the offset to 0 in this case. */
1186 0 : for( ulong i=0UL; i<chain_len; i++ ) {
1187 0 : ulong txnhash_offset = banks[ chain_len-1UL-i ].txnhash_offset==ULONG_MAX ? 0UL : banks[ chain_len-1UL-i ].txnhash_offset;
1188 0 : fd_txncache_finalize_fork( ctx->lead.txncache, banks[ chain_len-1UL-i ].fork_id, txnhash_offset, banks[ chain_len-1UL-i ].blockhash );
1189 0 : }
1190 :
1191 0 : for( ulong i=1UL; i<chain_len; i++ ) fd_txncache_advance_root( ctx->lead.txncache, banks[ chain_len-1UL-i ].fork_id );
1192 :
1193 0 : ctx->lead.txncache_root_fork_id = parent;
1194 :
1195 0 : return 0;
1196 0 : }
1197 :
1198 : static void
1199 : process_manifest( fd_snapin_tile_t * ctx,
1200 0 : fd_stem_context_t * stem ) {
1201 0 : fd_snapshot_manifest_t * manifest = fd_chunk_to_laddr( ctx->lead.manifest_out.mem, ctx->lead.manifest_out.chunk );
1202 :
1203 0 : if( FD_UNLIKELY( ctx->lead.advertised_slot!=manifest->slot ) ) {
1204 : /* SnapshotError::MismatchedSlot
1205 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L472 */
1206 0 : FD_LOG_WARNING(( "snapshot manifest bank slot %lu does not match advertised slot %lu from snapshot peer",
1207 0 : manifest->slot, ctx->lead.advertised_slot ));
1208 0 : transition_malformed( ctx, stem );
1209 0 : return;
1210 0 : }
1211 :
1212 0 : if( FD_UNLIKELY( !manifest->has_accounts_lthash ) ) {
1213 : /* The manifest must contain accounts lthash, irrespective of
1214 : whether lthash verification is disabled or not.
1215 : https://github.com/anza-xyz/agave/blob/v3.1.9/runtime/src/serde_snapshot.rs#L482 */
1216 0 : FD_LOG_WARNING(( "snapshot manifest missing accounts lthash" ));
1217 0 : transition_malformed( ctx, stem );
1218 0 : return;
1219 0 : }
1220 :
1221 0 : if( FD_UNLIKELY( !manifest->has_block_id ) ) {
1222 0 : FD_LOG_WARNING(( "snapshot manifest missing block id" ));
1223 0 : transition_malformed( ctx, stem );
1224 0 : return;
1225 0 : }
1226 :
1227 0 : uchar const * sum = manifest->accounts_lthash;
1228 0 : uchar hash32[32]; fd_blake3_hash( sum, FD_LTHASH_LEN_BYTES, hash32 );
1229 0 : FD_BASE58_ENCODE_32_BYTES( sum, sum_enc );
1230 0 : FD_BASE58_ENCODE_32_BYTES( hash32, hash32_enc );
1231 0 : FD_LOG_INFO(( "snapshot manifest slot=%lu indicates lthash[..32]=%s blake3(lthash)=%s",
1232 0 : manifest->slot, sum_enc, hash32_enc ));
1233 :
1234 0 : if( FD_UNLIKELY( memcmp( ctx->lead.advertised_hash, hash32, FD_HASH_FOOTPRINT ) ) ) {
1235 : /* SnapshotError::MismatchedHash
1236 : https://github.com/anza-xyz/agave/blob/v3.1.8/runtime/src/snapshot_bank_utils.rs#L479 */
1237 0 : FD_BASE58_ENCODE_32_BYTES( ctx->lead.advertised_hash, advertised_hash_enc );
1238 0 : FD_LOG_WARNING(( "snapshot manifest accounts lthash %s does not match advertised hash from snapshot peer %s",
1239 0 : hash32_enc, advertised_hash_enc ));
1240 0 : transition_malformed( ctx, stem );
1241 0 : return;
1242 0 : }
1243 :
1244 0 : ctx->lead.bank_slot = manifest->slot;
1245 0 : ctx->lead.manifest_capitalization = manifest->capitalization;
1246 0 : if( FD_UNLIKELY( ctx->lead.manifest_capitalization>LONG_MAX ) ) {
1247 : /* Calculations downstream require capitalization to be treated
1248 : as long (to handle addition and subtraction). */
1249 0 : FD_LOG_WARNING(( "snapshot manifest capitalization %lu exceeds LONG_MAX", ctx->lead.manifest_capitalization ));
1250 0 : transition_malformed( ctx, stem );
1251 0 : return;
1252 0 : }
1253 :
1254 0 : if( FD_UNLIKELY( fd_ssload_manifest_validate( manifest, FD_RUNTIME_MAX_VAT_VOTE_ACCOUNTS, FD_RUNTIME_MAX_STAKE_ACCOUNTS ) ) ) {
1255 0 : FD_LOG_WARNING(( "snapshot manifest validation failed" ));
1256 0 : transition_malformed( ctx, stem );
1257 0 : return;
1258 0 : }
1259 :
1260 0 : fd_epoch_schedule_t epoch_schedule = (fd_epoch_schedule_t){
1261 0 : .slots_per_epoch = manifest->epoch_schedule_params.slots_per_epoch,
1262 0 : .leader_schedule_slot_offset = manifest->epoch_schedule_params.leader_schedule_slot_offset,
1263 0 : .warmup = manifest->epoch_schedule_params.warmup,
1264 0 : .first_normal_epoch = manifest->epoch_schedule_params.first_normal_epoch,
1265 0 : .first_normal_slot = manifest->epoch_schedule_params.first_normal_slot,
1266 0 : };
1267 0 : ctx->lead.epoch = fd_slot_to_epoch( &epoch_schedule, manifest->slot, NULL );
1268 0 : ctx->lead.epoch_schedule = epoch_schedule;
1269 :
1270 0 : if( FD_UNLIKELY( verify_bank_hash( ctx, manifest ) ) ) {
1271 : /* https://github.com/anza-xyz/agave/blob/v3.1.9/runtime/src/bank.rs#L4682 */
1272 0 : transition_malformed( ctx, stem );
1273 0 : return;
1274 0 : }
1275 :
1276 : /* Classify this manifest against WFS (see fd_wfs.h). A verdict on a
1277 : full is provisional (an incremental may supersede it); only on an
1278 : incremental is it final, so ERROR rejects only an incremental. */
1279 0 : int wfs_mode = fd_wfs_mode( ctx->wfs_slot, ctx->wfs_hash_is_zero, (ulong)ctx->wfs_shred_version, manifest->slot );
1280 0 : switch( wfs_mode ) {
1281 0 : case FD_WFS_MODE_ERROR: {
1282 0 : if( FD_UNLIKELY( !ctx->full ) ) {
1283 0 : FD_LOG_WARNING(( "incremental snapshot manifest slot %lu is behind "
1284 0 : "[consensus.wait_for_supermajority_at_slot] %lu; rejecting this snapshot",
1285 0 : manifest->slot, ctx->wfs_slot ));
1286 0 : transition_malformed( ctx, stem );
1287 0 : return;
1288 0 : }
1289 0 : FD_LOG_NOTICE(( "full snapshot manifest slot %lu is behind "
1290 0 : "[consensus.wait_for_supermajority_at_slot] %lu; an incremental snapshot must "
1291 0 : "bridge the gap or the boot will be rejected",
1292 0 : manifest->slot, ctx->wfs_slot ));
1293 0 : break;
1294 0 : }
1295 0 : case FD_WFS_MODE_NOOP: {
1296 0 : FD_BASE58_ENCODE_32_BYTES( ctx->wfs_bank_hash.uc, expected_hash_enc );
1297 0 : FD_LOG_WARNING(( "%s snapshot manifest slot %lu is ahead of [consensus.wait_for_supermajority_at_slot] %lu "
1298 0 : "(wait_for_supermajority_with_bank_hash=%s, expected_shred_version=%lu); configured slot and bank hash not verified",
1299 0 : ctx->full ? "full" : "incremental",
1300 0 : manifest->slot, ctx->wfs_slot, expected_hash_enc, (ulong)ctx->wfs_shred_version ));
1301 0 : break;
1302 0 : }
1303 0 : case FD_WFS_MODE_MATCH:
1304 : /* Replay verifies the bank hash of the slot we boot at. */
1305 0 : case FD_WFS_MODE_DISABLED:
1306 0 : case FD_WFS_MODE_UNRESOLVED:
1307 0 : break;
1308 0 : default:
1309 0 : FD_LOG_CRIT(( "unexpected WFS mode %d (%s) at manifest slot %lu",
1310 0 : wfs_mode, fd_wfs_mode_str( wfs_mode ), manifest->slot ));
1311 0 : }
1312 :
1313 0 : if( FD_UNLIKELY( verify_slot_deltas_with_bank_slot( ctx, manifest->slot ) ) ) {
1314 0 : FD_LOG_WARNING(( "slot deltas verification failed" ));
1315 0 : transition_malformed( ctx, stem );
1316 0 : return;
1317 0 : }
1318 :
1319 0 : if( FD_UNLIKELY( verify_epoch_stakes( manifest ) ) ) {
1320 0 : FD_LOG_WARNING(( "epoch stakes verification failed" ));
1321 0 : transition_malformed( ctx, stem );
1322 0 : return;
1323 0 : }
1324 :
1325 0 : if( FD_UNLIKELY( populate_txncache( ctx, manifest->blockhashes, manifest->blockhashes_len, manifest->slot ) ) ) {
1326 0 : FD_LOG_WARNING(( "populating txncache failed" ));
1327 0 : transition_malformed( ctx, stem );
1328 0 : return;
1329 0 : }
1330 :
1331 0 : if( ctx->full ) {
1332 0 : ctx->lead.full_genesis_creation_time_seconds = manifest->creation_time_seconds;
1333 0 : } else {
1334 0 : if( FD_UNLIKELY( manifest->creation_time_seconds!=ctx->lead.full_genesis_creation_time_seconds ) ) {
1335 0 : FD_LOG_WARNING(( "snapshot manifest genesis creation time seconds %lu does not match full snapshot genesis creation time seconds %lu",
1336 0 : manifest->creation_time_seconds, ctx->lead.full_genesis_creation_time_seconds ));
1337 0 : transition_malformed( ctx, stem );
1338 0 : return;
1339 0 : }
1340 0 : }
1341 :
1342 0 : manifest->accdb_fork_id = fd_ushort_if( ctx->full, ctx->lead.accdb_root_fork_id.val, ctx->lead.accdb_incr_fork_id.val );
1343 0 : manifest->txncache_fork_id = ctx->lead.txncache_root_fork_id.val;
1344 :
1345 0 : ulong sig = ctx->full ? fd_ssmsg_sig( FD_SSMSG_MANIFEST_FULL ) :
1346 0 : fd_ssmsg_sig( FD_SSMSG_MANIFEST_INCREMENTAL );
1347 0 : fd_stem_publish( stem, ctx->lead.manifest_out.idx, sig, ctx->lead.manifest_out.chunk, sizeof(fd_snapshot_manifest_t), 0UL, 0UL, 0UL );
1348 0 : ctx->lead.manifest_out.chunk = fd_dcache_compact_next( ctx->lead.manifest_out.chunk, sizeof(fd_snapshot_manifest_t), ctx->lead.manifest_out.chunk0, ctx->lead.manifest_out.wmark );
1349 0 : }
1350 :
1351 : static void
1352 : snoop_stake_delegation( fd_snapin_tile_t * ctx,
1353 : ushort stake_fork,
1354 : ulong slot,
1355 : fd_pubkey_t const * stake_account,
1356 : ulong lamports,
1357 : fd_stake_state_t const * stake_state,
1358 0 : ulong data_sz ) {
1359 0 : fd_delegation_t const * delegation = &stake_state->stake.stake.delegation;
1360 0 : FD_CHECK_ERR( delegation->activation_epoch ==ULONG_MAX || delegation->activation_epoch <(ulong)USHORT_MAX, "activation_epoch overflow" );
1361 0 : FD_CHECK_ERR( delegation->deactivation_epoch==ULONG_MAX || delegation->deactivation_epoch<(ulong)USHORT_MAX, "deactivation_epoch overflow" );
1362 :
1363 0 : fd_stake_delegations_fork_update(
1364 0 : ctx->stake_delegations,
1365 0 : stake_fork,
1366 0 : slot,
1367 0 : stake_account,
1368 0 : &delegation->voter_pubkey,
1369 0 : delegation->stake,
1370 0 : delegation->activation_epoch,
1371 0 : delegation->deactivation_epoch,
1372 0 : stake_state->stake.stake.credits_observed,
1373 0 : lamports,
1374 0 : (uint)data_sz );
1375 0 : }
1376 :
1377 : /* Write engine */
1378 :
1379 : static void
1380 : writer_pwrite( fd_snapin_tile_t * ctx,
1381 : uchar const * buf,
1382 : ulong sz,
1383 0 : ulong off ) {
1384 0 : ulong done = 0UL;
1385 0 : while( done<sz ) {
1386 0 : long res = pwrite( ctx->writer.accdb_direct_fd, buf+done, sz-done, (long)(off+done) );
1387 0 : if( FD_UNLIKELY( res<=0L ) ) {
1388 0 : int err = res<0L ? errno : EIO;
1389 0 : if( res<0L && err==EINTR ) continue;
1390 0 : FD_LOG_ERR(( "snapshot write failed at offset %lu (%d-%s)", off+done, err, fd_io_strerror( err ) ));
1391 0 : }
1392 0 : done += (ulong)res;
1393 0 : ctx->metrics.disk_bytes_written += (ulong)res;
1394 0 : }
1395 0 : }
1396 :
1397 : static int
1398 0 : writer_flush( fd_snapin_tile_t * ctx ) {
1399 0 : if( FD_UNLIKELY( !ctx->writer.buf_used ) ) return 0;
1400 :
1401 : /* Pad the range to FD_SNAPIN_DIRECT_ALIGN for O_DIRECT. The padding
1402 : is a dead record so compaction reclaims it after snapshot loading
1403 : finishes. */
1404 0 : ulong used = ctx->writer.buf_used;
1405 0 : ulong padded = fd_ulong_align_up( used+sizeof(fd_accdb_disk_meta_t), FD_SNAPIN_DIRECT_ALIGN );
1406 0 : FD_TEST( padded<=FD_SNAPIN_WRITE_BUF_SZ );
1407 0 : fd_memset( ctx->writer.buf+used, 0, padded-used );
1408 0 : fd_accdb_disk_meta_t * dummy_record = (fd_accdb_disk_meta_t *)( ctx->writer.buf+used );
1409 0 : dummy_record->size = (uint)( padded-used-sizeof(fd_accdb_disk_meta_t) );
1410 :
1411 : /* The offset is aligned because partition sizes are multiples of
1412 : FD_SNAPIN_DIRECT_ALIGN (whole GiB), the snapin tiles are the only
1413 : layer-0 writers during the load, and every reservation they make
1414 : is a multiple of FD_SNAPIN_DIRECT_ALIGN. */
1415 0 : ulong base_off = fd_accdb_snapshot_reserve_write( ctx->accdb, padded, padded-used );
1416 0 : FD_TEST( fd_ulong_is_aligned( base_off, FD_SNAPIN_DIRECT_ALIGN ) );
1417 0 : writer_pwrite( ctx, ctx->writer.buf, padded, base_off );
1418 :
1419 0 : fd_accdb_fork_id_t fork_id = { .val = ctx->full ? USHORT_MAX : (ushort)ctx->incr_fork };
1420 0 : ushort stake_fork = FD_VOLATILE_CONST( ctx->shmem->stake_fork );
1421 0 : fd_snapin_account_batch_t * batch = &ctx->writer.batch;
1422 0 : ulong tile_idx = ctx->tile_idx;
1423 :
1424 0 : uchar const * pubkeys[ FD_SSPARSE_ACC_BATCH_MAX ];
1425 0 : ulong slots [ FD_SSPARSE_ACC_BATCH_MAX ];
1426 0 : ulong data_lens [ FD_SSPARSE_ACC_BATCH_MAX ];
1427 0 : ulong file_offsets [ FD_SSPARSE_ACC_BATCH_MAX ];
1428 0 : uchar results [ FD_SSPARSE_ACC_BATCH_MAX ];
1429 0 : ulong buf_off = 0UL;
1430 :
1431 : /* Flush accounts in batches of 8 */
1432 0 : for( ulong batch_off=0UL; batch_off<batch->cnt; batch_off+=FD_SSPARSE_ACC_BATCH_MAX ) {
1433 0 : ulong cnt = fd_ulong_min( FD_SSPARSE_ACC_BATCH_MAX, batch->cnt-batch_off );
1434 0 : ulong input_lamports_hi = 0UL;
1435 0 : ulong input_lamports = 0UL;
1436 :
1437 0 : for( ulong i=0UL; i<cnt; i++ ) {
1438 0 : ulong idx = batch_off+i;
1439 :
1440 0 : pubkeys [ i ] = ctx->writer.buf+buf_off;
1441 0 : slots [ i ] = (ulong)batch->slots [ idx ];
1442 0 : data_lens [ i ] = (ulong)batch->data_lens[ idx ];
1443 0 : file_offsets[ i ] = base_off+buf_off;
1444 :
1445 0 : buf_off += sizeof(fd_accdb_disk_meta_t)+data_lens[ i ];
1446 0 : fd_uwide_inc( &input_lamports_hi, &input_lamports, input_lamports_hi, input_lamports, batch->lamports[ idx ] );
1447 0 : }
1448 :
1449 0 : ulong accounts_ignored, accounts_replaced, accounts_loaded, replaced_lamports, ignored_lamports;
1450 0 : if( FD_UNLIKELY( fd_accdb_snapshot_write_batch( ctx->accdb, fork_id, cnt, pubkeys,
1451 0 : slots, batch->lamports+batch_off,
1452 0 : data_lens, batch->executables+batch_off,
1453 0 : file_offsets, &accounts_ignored, &accounts_replaced,
1454 0 : &accounts_loaded, &replaced_lamports, &ignored_lamports, results ) ) ) {
1455 0 : return 1;
1456 0 : }
1457 :
1458 : /* Update the snooped stake delegations. Anything that is not a
1459 : delegation and replaced a funded version tombstones it, so a tile
1460 : still holding the older version cannot leave it behind. */
1461 0 : for( ulong i=0UL; i<cnt; i++ ) {
1462 0 : if( FD_UNLIKELY( results[ i ]==FD_ACCDB_SNAPSHOT_WRITE_IGNORED ) ) continue;
1463 :
1464 0 : ulong lamports = batch->lamports[ batch_off+i ];
1465 0 : fd_pubkey_t const * pubkey = (fd_pubkey_t const *)pubkeys[ i ];
1466 0 : uchar const * owner = pubkeys[ i ]+offsetof(fd_accdb_disk_meta_t, owner);
1467 0 : uchar const * data = pubkeys[ i ]+sizeof(fd_accdb_disk_meta_t);
1468 :
1469 0 : if( lamports && !memcmp( owner, fd_solana_stake_program_id.uc, 32UL ) ) {
1470 0 : fd_stake_state_t const * stake_state = fd_stake_state_view( data, data_lens[ i ] );
1471 0 : if( stake_state && stake_state->stake_type==FD_STAKE_STATE_STAKE ) {
1472 0 : snoop_stake_delegation( ctx, stake_fork, slots[ i ], pubkey, lamports, stake_state, data_lens[ i ] );
1473 0 : continue;
1474 0 : }
1475 0 : }
1476 :
1477 0 : int replacing_full_entry = results[ i ]==FD_ACCDB_SNAPSHOT_WRITE_REPLACED_CROSS;
1478 0 : if( FD_UNLIKELY( replacing_full_entry || results[ i ]==FD_ACCDB_SNAPSHOT_WRITE_REPLACED ) ) {
1479 0 : fd_stake_delegations_fork_remove( ctx->stake_delegations, stake_fork, slots[ i ], pubkey, replacing_full_entry );
1480 0 : }
1481 0 : }
1482 :
1483 0 : ctx->metrics.accounts_ignored += accounts_ignored;
1484 0 : ctx->metrics.accounts_replaced += accounts_replaced;
1485 0 : ctx->metrics.accounts_loaded += accounts_loaded;
1486 0 : ctx->shmem->values[ tile_idx ].loaded += accounts_loaded;
1487 0 : ctx->shmem->values[ tile_idx ].duplicates += accounts_ignored + accounts_replaced;
1488 0 : ulong duplicate_lamports_hi, duplicate_lamports;
1489 0 : fd_uwide_inc( &duplicate_lamports_hi, &duplicate_lamports, 0UL, replaced_lamports, ignored_lamports );
1490 0 : ulong * in_hi = &ctx->shmem->values[ tile_idx ].input_lamports_hi; ulong * in_lo = &ctx->shmem->values[ tile_idx ].input_lamports;
1491 0 : ulong * dup_hi = &ctx->shmem->values[ tile_idx ].duplicate_lamports_hi; ulong * dup_lo = &ctx->shmem->values[ tile_idx ].duplicate_lamports;
1492 0 : fd_uwide_add( in_hi, in_lo, *in_hi, *in_lo, input_lamports_hi, input_lamports, 0UL );
1493 0 : fd_uwide_add( dup_hi, dup_lo, *dup_hi, *dup_lo, duplicate_lamports_hi, duplicate_lamports, 0UL );
1494 0 : }
1495 :
1496 0 : FD_TEST( buf_off==ctx->writer.buf_used );
1497 0 : ctx->writer.buf_used = 0UL;
1498 0 : ctx->writer.batch.cnt = 0UL;
1499 0 : return 0;
1500 0 : }
1501 :
1502 : static int
1503 : writer_append_account( fd_snapin_tile_t * ctx,
1504 : uchar const * pubkey,
1505 : uchar const * owner,
1506 : uchar const * data,
1507 : ulong slot,
1508 : ulong lamports,
1509 : ulong data_len,
1510 0 : int executable ) {
1511 0 : FD_TEST( slot<=UINT_MAX );
1512 0 : ulong account_sz = sizeof(fd_accdb_disk_meta_t)+data_len;
1513 0 : FD_TEST( account_sz<=FD_SNAPIN_WRITE_BUF_MAX );
1514 :
1515 0 : if( FD_UNLIKELY( account_sz>FD_SNAPIN_WRITE_BUF_MAX-ctx->writer.buf_used && writer_flush( ctx ) ) ) {
1516 0 : return 1;
1517 0 : }
1518 :
1519 0 : FD_TEST( ctx->writer.batch.cnt<FD_SNAPIN_WRITE_ACCOUNT_MAX );
1520 :
1521 0 : ulong idx = ctx->writer.batch.cnt++;
1522 0 : ulong buf_off = ctx->writer.buf_used;
1523 :
1524 : /* Serialize the account metadata into the buffer */
1525 0 : fd_accdb_disk_meta_t meta = {
1526 0 : .size = (uint)data_len,
1527 0 : .generation = 0U,
1528 0 : };
1529 0 : uchar * meta_ptr = ctx->writer.buf+buf_off;
1530 0 : uchar * data_ptr = meta_ptr+sizeof(fd_accdb_disk_meta_t);
1531 0 : fd_memcpy( meta.pubkey, pubkey, 32UL );
1532 0 : fd_memcpy( meta.owner, owner, 32UL );
1533 0 : fd_memcpy( meta_ptr, meta.b, sizeof(fd_accdb_disk_meta_t) );
1534 0 : fd_memcpy( data_ptr, data, data_len );
1535 0 : ctx->writer.buf_used += account_sz;
1536 :
1537 : /* Stage fields needed for index publication after pwrite. */
1538 0 : ctx->writer.batch.lamports [ idx ] = lamports;
1539 0 : ctx->writer.batch.slots [ idx ] = (uint)slot;
1540 0 : ctx->writer.batch.data_lens [ idx ] = (uint)data_len;
1541 0 : ctx->writer.batch.executables [ idx ] = executable;
1542 0 : ctx->metrics.total_accounts_processed++;
1543 0 : return 0;
1544 0 : }
1545 :
1546 : static int
1547 0 : writer_append_staged_account( fd_snapin_tile_t * ctx ) {
1548 0 : if( FD_UNLIKELY( writer_append_account( ctx, ctx->staged.pubkey, ctx->staged.owner, ctx->staged.data,
1549 0 : ctx->staged.slot, ctx->staged.lamports, ctx->staged.data_len,
1550 0 : ctx->staged.executable ) ) ) {
1551 0 : return 1;
1552 0 : }
1553 0 : ctx->metrics.total_account_batches_processed++;
1554 0 : return 0;
1555 0 : }
1556 :
1557 : static int
1558 : process_account_batch( fd_snapin_tile_t * ctx,
1559 0 : fd_ssparse_advance_result_t * result ) {
1560 0 : uchar const * const * entries = result->account_batch.batch;
1561 0 : ulong cnt = result->account_batch.batch_cnt;
1562 0 : ulong batch_slot = result->account_batch.slot;
1563 :
1564 0 : for( ulong i=0UL; i<cnt; i++ ) {
1565 0 : uchar const * e = entries[ i ];
1566 0 : if( FD_UNLIKELY( writer_append_account( ctx, e+16UL, e+64UL, e+136UL, batch_slot,
1567 0 : fd_ulong_load_8_fast( e+48UL ), fd_ulong_load_8_fast( e+8UL ),
1568 0 : (int)e[ 96UL ] ) ) ) {
1569 0 : return 1;
1570 0 : }
1571 0 : }
1572 0 : ctx->metrics.total_account_batches_processed++;
1573 0 : return 0;
1574 0 : }
1575 :
1576 : static int
1577 : process_account_header( fd_snapin_tile_t * ctx,
1578 0 : fd_ssparse_advance_result_t * result ) {
1579 0 : FD_TEST( ctx->staged.bytes_received==ctx->staged.data_len );
1580 0 : FD_TEST( result->account_header.data_len<=FD_RUNTIME_ACC_SZ_MAX );
1581 :
1582 0 : ctx->staged.executable = result->account_header.executable;
1583 0 : ctx->staged.slot = result->account_header.slot;
1584 0 : ctx->staged.lamports = result->account_header.lamports;
1585 0 : ctx->staged.data_len = result->account_header.data_len;
1586 0 : ctx->staged.bytes_received = 0UL;
1587 0 : fd_memcpy( ctx->staged.pubkey, result->account_header.pubkey, 32UL );
1588 0 : fd_memcpy( ctx->staged.owner, result->account_header.owner, 32UL );
1589 :
1590 0 : if( FD_LIKELY( ctx->staged.data_len ) ) {
1591 0 : return 0;
1592 0 : }
1593 :
1594 : /* No account data to receive, so we can just publish the account now. */
1595 0 : return writer_append_staged_account( ctx );
1596 0 : }
1597 :
1598 : static int
1599 : process_account_data( fd_snapin_tile_t * ctx,
1600 0 : fd_ssparse_advance_result_t * result ) {
1601 0 : FD_TEST( ctx->staged.bytes_received<ctx->staged.data_len );
1602 0 : FD_TEST( result->account_data.data_sz<=ctx->staged.data_len-ctx->staged.bytes_received );
1603 :
1604 0 : fd_memcpy( ctx->staged.data+ctx->staged.bytes_received, result->account_data.data, result->account_data.data_sz );
1605 0 : ctx->staged.bytes_received += result->account_data.data_sz;
1606 :
1607 : /* More bytes remain in account data, don't publish yet */
1608 0 : if( FD_LIKELY( ctx->staged.bytes_received<ctx->staged.data_len ) ) {
1609 0 : return 0;
1610 0 : }
1611 :
1612 0 : return writer_append_staged_account( ctx );
1613 0 : }
1614 :
1615 : static int
1616 : handle_data_frag( fd_snapin_tile_t * ctx,
1617 : ulong in_idx,
1618 : ulong chunk,
1619 : ulong sz,
1620 0 : fd_stem_context_t * stem ) {
1621 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_FINISHING ) ) {
1622 0 : FD_LOG_WARNING(( "received unexpected data frag while in state %s (%lu)",
1623 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
1624 0 : transition_malformed( ctx, stem );
1625 0 : return 0;
1626 0 : }
1627 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) {
1628 : /* Ignore all data frags after observing an error in the stream until
1629 : we receive fail & init control messages to restart processing. */
1630 0 : return 0;
1631 0 : }
1632 0 : if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING ) ) {
1633 0 : FD_LOG_ERR(( "received data frag during invalid state %s (%lu)",
1634 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
1635 0 : }
1636 :
1637 0 : if( FD_UNLIKELY( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>ctx->in[ in_idx ].mtu ) ) {
1638 0 : FD_LOG_ERR(( "invalid data frag bounds (chunk=%lu chunk0=%lu wmark=%lu sz=%lu mtu=%lu)", chunk, ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark, sz, ctx->in[ in_idx ].mtu ));
1639 0 : }
1640 :
1641 0 : for(;;) {
1642 0 : if( FD_UNLIKELY( sz-ctx->in[ in_idx ].pos==0UL ) ) break;
1643 :
1644 0 : uchar const * data = (uchar const *)fd_chunk_to_laddr_const( ctx->in[ in_idx ].wksp, chunk ) + ctx->in[ in_idx ].pos;
1645 :
1646 0 : int early_exit = 0;
1647 0 : fd_ssparse_advance_result_t result[1];
1648 0 : int res = fd_ssparse_advance( ctx->ssparse, data, sz-ctx->in[ in_idx ].pos, result );
1649 0 : switch( res ) {
1650 0 : case FD_SSPARSE_ADVANCE_ERROR:
1651 0 : FD_LOG_WARNING(( "error while parsing snapshot stream" ));
1652 0 : transition_malformed( ctx, stem );
1653 0 : return 0;
1654 0 : case FD_SSPARSE_ADVANCE_AGAIN:
1655 0 : break;
1656 0 : case FD_SSPARSE_ADVANCE_APPENDVEC: {
1657 : /* Parse only this tile's claimed appendvecs. */
1658 0 : ulong appendvec_idx = ctx->appendvec_seq++;
1659 0 : if( FD_UNLIKELY( appendvec_idx==ctx->claimed_appendvec ) ) {
1660 0 : ctx->claimed_appendvec = FD_ATOMIC_FETCH_AND_ADD( &ctx->shmem->next_appendvec_ticket, 1UL );
1661 0 : fd_ssparse_appendvec_parse( ctx->ssparse );
1662 0 : }
1663 0 : break;
1664 0 : }
1665 0 : case FD_SSPARSE_ADVANCE_MANIFEST:
1666 0 : case FD_SSPARSE_ADVANCE_MANIFEST_DONE: {
1667 : /* Tile 0 only. */
1668 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
1669 :
1670 0 : if( FD_UNLIKELY( ctx->lead.flags.manifest_done ) ) {
1671 0 : FD_LOG_WARNING(( "excess data after manifest" ));
1672 0 : transition_malformed( ctx, stem );
1673 0 : return 0;
1674 0 : }
1675 0 : int parser_res = fd_ssmanifest_parser_consume( ctx->lead.manifest_parser,
1676 0 : result->manifest.data,
1677 0 : result->manifest.data_sz );
1678 0 : if( FD_UNLIKELY( parser_res==FD_SSMANIFEST_PARSER_ADVANCE_ERROR ) ) {
1679 0 : FD_LOG_WARNING(( "error while parsing snapshot manifest" ));
1680 0 : transition_malformed( ctx, stem );
1681 0 : return 0;
1682 0 : }
1683 0 : if( res==FD_SSPARSE_ADVANCE_MANIFEST_DONE ) {
1684 0 : if( FD_UNLIKELY( fd_ssmanifest_parser_fini( ctx->lead.manifest_parser )!=FD_SSMANIFEST_PARSER_ADVANCE_DONE ) ) {
1685 0 : FD_LOG_WARNING(( "manifest stream ended before parser was done" ));
1686 0 : transition_malformed( ctx, stem );
1687 0 : return 0;
1688 0 : }
1689 0 : ctx->lead.flags.manifest_done = 1;
1690 0 : }
1691 0 : break;
1692 0 : }
1693 0 : case FD_SSPARSE_ADVANCE_STATUS_CACHE: {
1694 : /* Tile 0 only. */
1695 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
1696 :
1697 0 : fd_slot_delta_parser_advance_result_t sd_result[1];
1698 0 : ulong bytes_remaining = result->status_cache.data_sz;
1699 :
1700 0 : while( bytes_remaining ) {
1701 0 : int res = fd_slot_delta_parser_consume( ctx->lead.slot_delta_parser,
1702 0 : result->status_cache.data,
1703 0 : bytes_remaining,
1704 0 : sd_result );
1705 0 : if( FD_UNLIKELY( res<0 ) ) {
1706 0 : FD_LOG_WARNING(( "error while parsing slot deltas in status cache" ));
1707 0 : transition_malformed( ctx, stem );
1708 0 : return 0;
1709 0 : } else if( FD_LIKELY( res==FD_SLOT_DELTA_PARSER_ADVANCE_SLOT ) ) {
1710 0 : txncache_staging_slot_begin( ctx, sd_result->slot );
1711 0 : } else if( FD_LIKELY( res==FD_SLOT_DELTA_PARSER_ADVANCE_GROUP ) ) {
1712 0 : if( FD_UNLIKELY( txncache_staging_group_begin( ctx, sd_result->group.blockhash, sd_result->group.txnhash_offset ) ) ) {
1713 0 : FD_LOG_WARNING(( "blockhash groups overflow for slot %lu, max is %lu", sd_result->group.slot, ctx->lead.txncache_max_groups_per_slot ));
1714 0 : transition_malformed( ctx, stem );
1715 0 : return 0;
1716 0 : }
1717 0 : } else if( FD_LIKELY( res==FD_SLOT_DELTA_PARSER_ADVANCE_ENTRY ) ) {
1718 0 : if( FD_UNLIKELY( txncache_staging_entry_add( ctx, sd_result->entry->slot, sd_result->entry->txnhash ) ) ) {
1719 0 : FD_LOG_WARNING(( "status cache slot %lu has more than %lu entries", sd_result->entry->slot, ctx->lead.txncache_max_entries_per_slot ));
1720 0 : transition_malformed( ctx, stem );
1721 0 : return 0;
1722 0 : }
1723 0 : }
1724 :
1725 0 : bytes_remaining -= sd_result->bytes_consumed;
1726 0 : result->status_cache.data += sd_result->bytes_consumed;
1727 0 : }
1728 :
1729 0 : if( FD_UNLIKELY( result->status_cache.done ) ) {
1730 0 : int fini_res = fd_slot_delta_parser_consume( ctx->lead.slot_delta_parser, result->status_cache.data, 0UL, sd_result );
1731 0 : if( FD_UNLIKELY( fini_res<0 ) ) {
1732 0 : FD_LOG_WARNING(( "error while finalizing slot deltas in status cache" ));
1733 0 : transition_malformed( ctx, stem );
1734 0 : return 0;
1735 0 : }
1736 0 : ctx->lead.flags.status_cache_done = fini_res==FD_SLOT_DELTA_PARSER_ADVANCE_DONE;
1737 0 : }
1738 0 : break;
1739 0 : }
1740 0 : case FD_SSPARSE_ADVANCE_ACCOUNT_HEADER:
1741 0 : early_exit = process_account_header( ctx, result );
1742 0 : if( FD_UNLIKELY( early_exit ) ) {
1743 0 : transition_malformed( ctx, stem );
1744 0 : return 0;
1745 0 : }
1746 0 : break;
1747 0 : case FD_SSPARSE_ADVANCE_ACCOUNT_DATA:
1748 0 : if( FD_UNLIKELY( process_account_data( ctx, result ) ) ) {
1749 0 : transition_malformed( ctx, stem );
1750 0 : return 0;
1751 0 : }
1752 :
1753 : /* Account data may span multiple input chunks. Once complete,
1754 : copy the staged account into gui_out and publish it once.
1755 :
1756 : We expect ConfigKeys Vec to be length 2 (checked via the
1757 : first byte of the accumulated data). We expect the size of
1758 : ConfigProgram-owned accounts to be at most
1759 : FD_GUI_CONFIG_PARSE_MAX_VALID_ACCT_SZ, since this is the
1760 : size that the Solana CLI allocates for them. Although the
1761 : ConfigProgram itself does not enforce these invariants, the
1762 : vast majority of accounts (with a tiny number of exceptions
1763 : on devnet) are maintained with the Solana CLI. */
1764 0 : if( FD_UNLIKELY( ctx->staged.bytes_received==ctx->staged.data_len
1765 0 : && ctx->gui_out.idx!=ULONG_MAX
1766 0 : && !memcmp( ctx->staged.owner, fd_solana_config_program_id.key, sizeof(fd_hash_t) )
1767 0 : && ctx->staged.data_len
1768 0 : && ctx->staged.data_len<=FD_GUI_CONFIG_PARSE_MAX_VALID_ACCT_SZ
1769 0 : && ctx->staged.data[ 0 ]==2UL ) ) {
1770 0 : uchar * acct = fd_chunk_to_laddr( ctx->gui_out.mem, ctx->gui_out.chunk );
1771 0 : fd_memcpy( acct, ctx->staged.data, ctx->staged.data_len );
1772 0 : fd_stem_publish( stem, ctx->gui_out.idx, ctx->staged.slot, ctx->gui_out.chunk, ctx->staged.data_len, 0UL, 0UL, 0UL ); /* sig is the account slot */
1773 0 : ctx->gui_out.chunk = fd_dcache_compact_next( ctx->gui_out.chunk, ctx->staged.data_len, ctx->gui_out.chunk0, ctx->gui_out.wmark );
1774 0 : early_exit = 1;
1775 0 : }
1776 0 : break;
1777 0 : case FD_SSPARSE_ADVANCE_ACCOUNT_BATCH:
1778 0 : if( FD_UNLIKELY( process_account_batch( ctx, result ) ) ) {
1779 0 : transition_malformed( ctx, stem );
1780 0 : return 0;
1781 0 : }
1782 0 : break;
1783 0 : case FD_SSPARSE_ADVANCE_DONE:
1784 0 : ctx->state = FD_SNAPSHOT_STATE_FINISHING;
1785 0 : break;
1786 0 : default:
1787 0 : FD_LOG_ERR(( "unexpected fd_ssparse_advance result %d", res ));
1788 0 : break;
1789 0 : }
1790 :
1791 0 : if( FD_UNLIKELY( is_lead( ctx ) && !ctx->lead.flags.manifest_processed && ctx->lead.flags.manifest_done && ctx->lead.flags.status_cache_done ) ) {
1792 0 : process_manifest( ctx, stem );
1793 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) break;
1794 0 : ctx->lead.flags.manifest_processed = 1;
1795 0 : }
1796 :
1797 0 : ctx->in[ in_idx ].pos += result->bytes_consumed;
1798 0 : if( FD_LIKELY( ctx->full ) ) ctx->metrics.full_bytes_read += result->bytes_consumed;
1799 0 : else ctx->metrics.incremental_bytes_read += result->bytes_consumed;
1800 :
1801 0 : if( FD_UNLIKELY( early_exit ) ) break;
1802 0 : }
1803 :
1804 0 : int reprocess_frag = ctx->in[ in_idx ].pos<sz;
1805 0 : if( FD_LIKELY( !reprocess_frag ) ) ctx->in[ in_idx ].pos = 0UL;
1806 0 : return reprocess_frag;
1807 0 : }
1808 :
1809 : /* Resets local worker state */
1810 :
1811 : static void
1812 0 : reset_attempt_state( fd_snapin_tile_t * ctx ) {
1813 0 : for( ulong lane=0UL; lane<ctx->lane_cnt; lane++ ) {
1814 0 : ctx->in[ lane ].pos = 0UL;
1815 0 : }
1816 0 : ctx->expected_frame = 0UL;
1817 0 : ctx->appendvec_seq = 0UL;
1818 0 : ctx->incr_fork = ULONG_MAX;
1819 0 : ctx->writer.buf_used = 0UL;
1820 0 : ctx->writer.batch.cnt = 0UL;
1821 0 : ctx->staged.data_len = 0UL;
1822 0 : ctx->staged.bytes_received = 0UL;
1823 :
1824 0 : fd_ssparse_init( ctx->ssparse );
1825 0 : fd_ssparse_batch_enable( ctx->ssparse, 1 );
1826 0 : }
1827 :
1828 : static void
1829 0 : start_processing_attempt( fd_snapin_tile_t * ctx ) {
1830 0 : FD_COMPILER_MFENCE();
1831 0 : ctx->incr_fork = FD_VOLATILE_CONST( ctx->shmem->fork_id );
1832 0 : FD_TEST( ctx->full ? ctx->incr_fork==(ulong)USHORT_MAX : ctx->incr_fork<(ulong)USHORT_MAX );
1833 :
1834 : /* Claim before the first data fragment. */
1835 0 : ctx->claimed_appendvec = FD_ATOMIC_FETCH_AND_ADD( &ctx->shmem->next_appendvec_ticket, 1UL );
1836 0 : }
1837 :
1838 : static int
1839 0 : validate_capitalization( fd_snapin_tile_t * ctx ) {
1840 : /* Sum exactly: saturating at 2^64 computed 0 for an honest snapshot
1841 : and could accept a wrong one. */
1842 0 : ulong input_hi = 0UL, input_lo = 0UL;
1843 0 : ulong dup_hi = 0UL, dup_lo = 0UL;
1844 0 : FD_COMPILER_MFENCE();
1845 0 : for( ulong i=0UL; i<FD_TOPO_MAX_TILE_IN_LINKS; i++ ) {
1846 0 : fd_uwide_add( &input_hi, &input_lo, input_hi, input_lo, ctx->shmem->values[ i ].input_lamports_hi, ctx->shmem->values[ i ].input_lamports, 0UL );
1847 0 : fd_uwide_add( &dup_hi, &dup_lo, dup_hi, dup_lo, ctx->shmem->values[ i ].duplicate_lamports_hi, ctx->shmem->values[ i ].duplicate_lamports, 0UL );
1848 0 : }
1849 :
1850 0 : ulong cap_hi = 0UL;
1851 0 : ulong cap_lo = fd_ulong_if( ctx->full, 0UL, ctx->lead.recovery.capitalization );
1852 0 : fd_uwide_add( &cap_hi, &cap_lo, cap_hi, cap_lo, input_hi, input_lo, 0UL );
1853 0 : ulong borrow = fd_uwide_sub( &cap_hi, &cap_lo, cap_hi, cap_lo, dup_hi, dup_lo, 0UL );
1854 0 : if( FD_UNLIKELY( borrow || cap_hi || cap_lo!=ctx->lead.manifest_capitalization ) ) {
1855 : /* SnapshotError::MismatchedCapitalization
1856 : https://github.com/anza-xyz/agave/blob/v4.0.0-beta.2/runtime/src/snapshot_bank_utils.rs#L217 */
1857 0 : if( FD_LIKELY( !borrow && !cap_hi ) ) {
1858 0 : FD_LOG_WARNING(( "%s snapshot manifest capitalization %lu does not match computed capitalization %lu",
1859 0 : ctx->full?"full":"incr", ctx->lead.manifest_capitalization, cap_lo ));
1860 0 : } else {
1861 0 : FD_LOG_WARNING(( "%s snapshot manifest capitalization %lu does not match computed capitalization, which is out of range "
1862 0 : "(input lamports %lu*2^64+%lu, duplicate lamports %lu*2^64+%lu)",
1863 0 : ctx->full?"full":"incr", ctx->lead.manifest_capitalization, input_hi, input_lo, dup_hi, dup_lo ));
1864 0 : }
1865 0 : return -1;
1866 0 : }
1867 0 : return 0;
1868 0 : }
1869 :
1870 : static void
1871 0 : fold_account_counts( fd_snapin_tile_t * ctx ) {
1872 0 : for( ulong i=0UL; i<FD_TOPO_MAX_TILE_IN_LINKS; i++ ) {
1873 0 : ctx->lead.account_counts.loaded += ctx->shmem->values[ i ].loaded;
1874 0 : ctx->lead.account_counts.duplicates += ctx->shmem->values[ i ].duplicates;
1875 0 : }
1876 0 : }
1877 :
1878 : static void
1879 : handle_control_frag( fd_snapin_tile_t * ctx,
1880 : fd_stem_context_t * stem,
1881 : ulong in_idx,
1882 : ulong sig,
1883 : ulong chunk,
1884 0 : ulong sz ) {
1885 0 : if( ctx->state==FD_SNAPSHOT_STATE_ERROR && sig!=FD_SNAPSHOT_MSG_CTRL_FAIL ) {
1886 : /* Control messages move along the snapshot load pipeline. Since
1887 : error conditions can be triggered by any tile in the pipeline,
1888 : it is possible to be in error state and still receive otherwise
1889 : valid messages. Only a fail message can revert this. */
1890 0 : return;
1891 0 : };
1892 :
1893 0 : int forward_msg = 1;
1894 :
1895 0 : switch( sig ) {
1896 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
1897 0 : case FD_SNAPSHOT_MSG_CTRL_INIT_INCR: {
1898 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
1899 0 : ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
1900 0 : ctx->full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
1901 :
1902 0 : reset_attempt_state( ctx );
1903 :
1904 : /* Rewind metric counters (no-op unless recovering from a fail) */
1905 0 : if( sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL ) {
1906 0 : ctx->metrics.accounts_loaded = ctx->metrics.full_accounts_loaded = 0;
1907 0 : ctx->metrics.accounts_replaced = ctx->metrics.full_accounts_replaced = 0;
1908 0 : ctx->metrics.accounts_ignored = ctx->metrics.full_accounts_ignored = 0;
1909 0 : ctx->metrics.full_bytes_read = 0UL;
1910 0 : ctx->metrics.incremental_bytes_read = 0UL;
1911 0 : } else {
1912 0 : ctx->metrics.accounts_loaded = ctx->metrics.full_accounts_loaded;
1913 0 : ctx->metrics.accounts_replaced = ctx->metrics.full_accounts_replaced;
1914 0 : ctx->metrics.accounts_ignored = ctx->metrics.full_accounts_ignored;
1915 0 : ctx->metrics.incremental_bytes_read = 0UL;
1916 0 : }
1917 :
1918 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
1919 :
1920 : /* Roll back before publishing this attempt. */
1921 0 : if( FD_UNLIKELY( ctx->lead.rollback.pending ) ) {
1922 0 : ctx->lead.rollback.pending = 0;
1923 : /* Purge failed incremental state unless a full reset follows. */
1924 0 : if( !ctx->lead.rollback.full && FD_LIKELY( !ctx->full ) ) {
1925 0 : fd_accdb_purge( ctx->accdb, ctx->lead.rollback.fork );
1926 0 : fd_accdb_snapshot_revert_whead( ctx->accdb, &ctx->lead.recovery.accdb_metadata );
1927 0 : fd_stake_delegations_evict_fork( ctx->stake_delegations, FD_VOLATILE_CONST( ctx->shmem->stake_fork ) );
1928 0 : }
1929 0 : }
1930 :
1931 0 : ctx->lead.manifest_capitalization = 0UL;
1932 :
1933 0 : fd_txncache_reset( ctx->lead.txncache );
1934 0 : txncache_staging_reset( ctx );
1935 0 : fd_ssmanifest_parser_init( ctx->lead.manifest_parser, fd_chunk_to_laddr( ctx->lead.manifest_out.mem, ctx->lead.manifest_out.chunk ) );
1936 0 : fd_slot_delta_parser_init( ctx->lead.slot_delta_parser );
1937 0 : fd_memset( &ctx->lead.flags, 0, sizeof(ctx->lead.flags) );
1938 :
1939 0 : ushort stake_fork = USHORT_MAX;
1940 0 : if( sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL ) {
1941 0 : ctx->lead.full_genesis_creation_time_seconds = 0UL;
1942 0 : ctx->lead.recovery.capitalization = 0UL;
1943 0 : fd_memset( &ctx->lead.account_counts, 0, sizeof(ctx->lead.account_counts) );
1944 :
1945 0 : fd_stake_delegations_reset( ctx->stake_delegations );
1946 0 : fd_accdb_reset( ctx->accdb );
1947 0 : fd_accdb_fork_id_t null_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
1948 0 : ctx->lead.accdb_root_fork_id = fd_accdb_attach_child( ctx->accdb, null_fork_id );
1949 :
1950 0 : fd_accdb_snapshot_load_begin( ctx->accdb );
1951 0 : } else {
1952 : /* Create a child fork for incremental writes. On failure,
1953 : fd_accdb_purge(child) reverts just the incremental changes.
1954 : On success, fd_accdb_advance_root(child) promotes them. The
1955 : stake delegations get a fork likewise. */
1956 0 : ctx->lead.accdb_incr_fork_id = fd_accdb_attach_child( ctx->accdb, ctx->lead.accdb_root_fork_id );
1957 0 : stake_fork = fd_stake_delegations_new_fork( ctx->stake_delegations, USHORT_MAX );
1958 0 : }
1959 :
1960 : /* Save the slot advertised by the snapshot peer and verify it
1961 : against the slot in the snapshot manifest. For redirect-based
1962 : HTTP downloads, these are initial estimates from gossip and
1963 : will be updated by the META message below once the redirect
1964 : resolves to a concrete snapshot filename. */
1965 0 : fd_ssctrl_init_t const * msg = fd_chunk_to_laddr_const( ctx->in[ in_idx ].wksp, chunk );
1966 0 : ctx->lead.advertised_slot = msg->slot;
1967 0 : fd_memcpy( ctx->lead.advertised_hash, msg->snapshot_hash, FD_HASH_FOOTPRINT );
1968 0 : ctx->lead.init_completed = 1;
1969 :
1970 : /* Reset shared state before publishing the fork. */
1971 0 : fd_memset( &ctx->shmem->values, 0, sizeof(ctx->shmem->values) );
1972 0 : FD_VOLATILE( ctx->shmem->next_appendvec_ticket ) = 0UL;
1973 0 : FD_COMPILER_MFENCE();
1974 :
1975 : /* Publish before acknowledging INIT. */
1976 0 : FD_VOLATILE( ctx->shmem->fork_id ) = ctx->full ? (ulong)USHORT_MAX : (ulong)ctx->lead.accdb_incr_fork_id.val;
1977 0 : FD_VOLATILE( ctx->shmem->stake_fork ) = stake_fork;
1978 0 : FD_COMPILER_MFENCE();
1979 0 : break;
1980 0 : }
1981 :
1982 0 : case FD_SNAPSHOT_MSG_META: {
1983 0 : forward_msg = 0; /* snapct already receives META directly from snapld */
1984 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
1985 :
1986 : /* For redirect-based HTTP downloads, the META message carries
1987 : the resolved slot and hash from the actual snapshot filename
1988 : the server redirected to. Update the advertised values so
1989 : that process_manifest can verify the manifest against them. */
1990 0 : FD_TEST( sz==sizeof(fd_ssctrl_meta_t) );
1991 0 : fd_ssctrl_meta_t const * meta = fd_chunk_to_laddr_const( ctx->in[ in_idx ].wksp, chunk );
1992 0 : if( meta->resolved_slot!=ULONG_MAX ) {
1993 0 : ctx->lead.advertised_slot = meta->resolved_slot;
1994 0 : fd_memcpy( ctx->lead.advertised_hash, meta->resolved_hash, FD_HASH_FOOTPRINT );
1995 0 : }
1996 0 : break;
1997 0 : }
1998 :
1999 0 : case FD_SNAPSHOT_MSG_CTRL_FINI: {
2000 : /* This is a special case: handle_data_frag must have already
2001 : processed FD_SSPARSE_ADVANCE_DONE and moved the state into
2002 : FD_SNAPSHOT_STATE_FINISHING. Otherwise, treat this as a
2003 : malformed snapshot so that the pipeline can retry. */
2004 0 : if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
2005 0 : FD_LOG_WARNING(( "received FINI while in state %s (%lu), expected FINISHING (possibly truncated tar stream)",
2006 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
2007 0 : transition_malformed( ctx, stem );
2008 0 : forward_msg = 0;
2009 0 : break;
2010 0 : }
2011 :
2012 : /* Flush records before the FINI ack. */
2013 0 : if( FD_UNLIKELY( writer_flush( ctx ) ) ) {
2014 0 : transition_malformed( ctx, stem );
2015 0 : forward_msg = 0;
2016 0 : break;
2017 0 : }
2018 :
2019 0 : fd_accdb_flush_metrics( ctx->accdb );
2020 :
2021 : /* Publish prior total updates before the FINI ack. */
2022 0 : FD_COMPILER_MFENCE(); /* publish before ack */
2023 :
2024 : /* Keep per-tile gauges. Dashboards sum them. */
2025 0 : break;
2026 0 : }
2027 :
2028 0 : case FD_SNAPSHOT_MSG_CTRL_NEXT: {
2029 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
2030 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
2031 :
2032 : /* Backup metric counters */
2033 0 : ctx->metrics.full_accounts_loaded = ctx->metrics.accounts_loaded;
2034 0 : ctx->metrics.full_accounts_replaced = ctx->metrics.accounts_replaced;
2035 0 : ctx->metrics.full_accounts_ignored = ctx->metrics.accounts_ignored;
2036 :
2037 : /* Tile 0 only. */
2038 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
2039 :
2040 : /* FINI acks make shared data stable. */
2041 0 : if( FD_UNLIKELY( verify_sysvars( ctx ) ) ) {
2042 0 : FD_LOG_WARNING(( "sysvar verification failed for full snapshot" ));
2043 0 : transition_malformed( ctx, stem );
2044 0 : forward_msg = 0;
2045 0 : break;
2046 0 : }
2047 :
2048 0 : if( FD_UNLIKELY( validate_capitalization( ctx ) ) ) {
2049 0 : transition_malformed( ctx, stem );
2050 0 : forward_msg = 0;
2051 0 : break;
2052 0 : }
2053 :
2054 0 : fold_account_counts( ctx );
2055 0 : ctx->lead.recovery.capitalization = ctx->lead.manifest_capitalization;
2056 0 : fd_accdb_snapshot_save_whead( ctx->accdb, &ctx->lead.recovery.accdb_metadata );
2057 0 : ctx->lead.init_completed = 0;
2058 0 : break;
2059 0 : }
2060 :
2061 0 : case FD_SNAPSHOT_MSG_CTRL_DONE: {
2062 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_FINISHING );
2063 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
2064 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) break;
2065 :
2066 0 : if( FD_UNLIKELY( verify_sysvars( ctx ) ) ) {
2067 0 : if( ctx->full ) FD_LOG_WARNING(( "sysvar verification failed for full snapshot" ));
2068 0 : else FD_LOG_WARNING(( "sysvar verification failed for incremental snapshot" ));
2069 0 : transition_malformed( ctx, stem );
2070 0 : forward_msg = 0;
2071 0 : break;
2072 0 : }
2073 :
2074 0 : if( FD_UNLIKELY( validate_capitalization( ctx ) ) ) {
2075 0 : transition_malformed( ctx, stem );
2076 0 : forward_msg = 0;
2077 0 : break;
2078 0 : }
2079 :
2080 0 : fold_account_counts( ctx );
2081 0 : if( !ctx->full ) {
2082 0 : fd_accdb_snapshot_recover_delta( ctx->accdb, ctx->lead.accdb_incr_fork_id );
2083 : /* ensure that snapin tile sees all delta changes before rooting */
2084 0 : __atomic_thread_fence( __ATOMIC_SEQ_CST );
2085 0 : fd_accdb_advance_root( ctx->accdb, ctx->lead.accdb_incr_fork_id );
2086 0 : ctx->lead.accdb_root_fork_id = ctx->lead.accdb_incr_fork_id;
2087 0 : ctx->lead.accdb_incr_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
2088 0 : ushort stake_fork = FD_VOLATILE_CONST( ctx->shmem->stake_fork );
2089 0 : fd_stake_delegations_advance_root( 0UL, NULL, NULL, 1, ctx->stake_delegations, stake_fork, NULL );
2090 0 : fd_stake_delegations_evict_fork( ctx->stake_delegations, stake_fork );
2091 0 : }
2092 :
2093 0 : fd_accdb_snapshot_load_end( ctx->accdb );
2094 :
2095 : /* TODO: Pass in tile_idx and tile_cnt when parallelizing snapin */
2096 0 : fd_features_restore_chunk( &ctx->lead.bank->f.features, ctx->accdb, ctx->lead.accdb_root_fork_id, ctx->lead.bank_slot, &ctx->lead.epoch_schedule, 0UL, 1UL );
2097 :
2098 : /* Notify replay when snapshot is fully loaded and verified. */
2099 0 : fd_stem_publish( stem, ctx->lead.manifest_out.idx, fd_ssmsg_sig( FD_SSMSG_DONE ), 0UL, 0UL, 0UL, 0UL, 0UL );
2100 0 : ctx->lead.init_completed = 0;
2101 0 : break;
2102 0 : }
2103 :
2104 0 : case FD_SNAPSHOT_MSG_CTRL_ERROR: {
2105 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
2106 0 : ctx->state = FD_SNAPSHOT_STATE_ERROR;
2107 0 : break;
2108 0 : }
2109 :
2110 0 : case FD_SNAPSHOT_MSG_CTRL_FAIL: {
2111 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
2112 0 : fd_accdb_flush_metrics( ctx->accdb );
2113 :
2114 0 : FD_COMPILER_MFENCE();
2115 :
2116 : /* Reset the worker state for this full/incr loading attempt */
2117 0 : reset_attempt_state( ctx );
2118 :
2119 : /* Defer rollback until the next INIT, which is triggered after
2120 : all workers have sent their FAIL acks. */
2121 0 : if( FD_UNLIKELY( is_lead( ctx ) ) ) {
2122 0 : FD_VOLATILE( ctx->shmem->fork_id ) = ULONG_MAX;
2123 0 : FD_COMPILER_MFENCE();
2124 :
2125 : /* Only a completed INIT has valid state to roll back. */
2126 0 : if( FD_LIKELY( ctx->lead.init_completed ) ) {
2127 0 : ctx->lead.rollback.pending = 1;
2128 0 : ctx->lead.rollback.full = ctx->full;
2129 0 : ctx->lead.rollback.fork = ctx->lead.accdb_incr_fork_id;
2130 0 : if( ctx->full ) ctx->lead.accdb_root_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
2131 0 : ctx->lead.accdb_incr_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
2132 0 : }
2133 0 : ctx->lead.init_completed = 0;
2134 0 : }
2135 :
2136 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
2137 0 : break;
2138 0 : }
2139 :
2140 0 : case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN: {
2141 0 : FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
2142 0 : ctx->state = FD_SNAPSHOT_STATE_SHUTDOWN;
2143 0 : break;
2144 0 : }
2145 :
2146 0 : default: {
2147 0 : FD_LOG_ERR(( "unexpected control frag %s (%lu) in state %s (%lu)",
2148 0 : fd_ssctrl_msg_ctrl_str( sig ), sig,
2149 0 : fd_ssctrl_state_str( (ulong)ctx->state ), (ulong)ctx->state ));
2150 0 : break;
2151 0 : }
2152 0 : }
2153 :
2154 : /* Forward the control message down the pipeline */
2155 0 : if( FD_LIKELY( forward_msg ) ) {
2156 0 : fd_stem_publish( stem, ctx->ct_out.idx, sig, 0UL, 0UL, 0UL, 0UL, 0UL );
2157 0 : }
2158 0 : }
2159 :
2160 : static inline int
2161 0 : all_controls_seen( fd_snapin_tile_t const * ctx ) {
2162 0 : int all_seen = 1;
2163 0 : for( ulong i=0UL; i<ctx->lane_cnt; i++ ) {
2164 0 : all_seen &= !!ctx->control_seen[ i ];
2165 0 : }
2166 0 : return all_seen;
2167 0 : }
2168 :
2169 : static inline int
2170 : before_frag( fd_snapin_tile_t * ctx,
2171 : ulong in_idx,
2172 : ulong seq FD_PARAM_UNUSED,
2173 0 : ulong sig ) {
2174 : /* If we're currently in ERROR state we should only process FAIL
2175 : control frags */
2176 0 : if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) {
2177 0 : return sig!=FD_SNAPSHOT_MSG_CTRL_FAIL;
2178 0 : }
2179 :
2180 0 : if( FD_UNLIKELY( sig==FD_SNAPSHOT_MSG_CTRL_ERROR ) ) {
2181 0 : return 0;
2182 0 : }
2183 :
2184 : /* Once this lane sends the pending control, hold its later frags
2185 : until all snapdc lanes send the same control. */
2186 0 : if( FD_UNLIKELY( ctx->pending_control!=ULONG_MAX && ctx->control_seen[ in_idx ] ) ) {
2187 0 : FD_TEST( sig!=ctx->pending_control );
2188 0 : return -1;
2189 0 : }
2190 :
2191 0 : if( FD_LIKELY( fd_snapdc_sig_type( sig )==FD_SNAPSHOT_MSG_DATA ) ) {
2192 : /* Only accept DATA frags for the expected frame */
2193 0 : ulong frame = fd_snapdc_sig_frame( sig );
2194 0 : FD_CHECK_ERR( frame>=ctx->expected_frame, "snapdc lane repeated a frame" );
2195 0 : if( FD_UNLIKELY( frame!=ctx->expected_frame ) ) return -1;
2196 0 : }
2197 :
2198 0 : return 0;
2199 0 : }
2200 :
2201 : static inline int
2202 : handle_lane_data_frag( fd_snapin_tile_t * ctx,
2203 : fd_stem_context_t * stem,
2204 : ulong in_idx,
2205 : ulong chunk,
2206 : ulong sz,
2207 0 : ulong ctl ) {
2208 0 : if( FD_UNLIKELY( ctx->incr_fork==ULONG_MAX ) ) {
2209 0 : start_processing_attempt( ctx );
2210 0 : }
2211 :
2212 : /* EOM marks the end of a frame */
2213 0 : int eom = !!fd_frag_meta_ctl_eom( ctl );
2214 :
2215 : /* The tar parser can reach EOF before snapdc reports the end of the
2216 : zstd frame. Only the empty EOM is valid (any payload after EOF is
2217 : malformed). */
2218 0 : int trailing_eom = ctx->state==FD_SNAPSHOT_STATE_FINISHING && eom && !sz;
2219 0 : if( FD_UNLIKELY( !trailing_eom && handle_data_frag( ctx, in_idx, chunk, sz, stem ) ) ) {
2220 0 : return 1;
2221 0 : }
2222 :
2223 0 : if( FD_UNLIKELY( eom ) ) {
2224 0 : ctx->expected_frame++;
2225 0 : }
2226 :
2227 0 : return 0;
2228 0 : }
2229 :
2230 : static inline void
2231 : handle_control_barrier( fd_snapin_tile_t * ctx,
2232 : fd_stem_context_t * stem,
2233 : ulong in_idx,
2234 : ulong sig,
2235 : ulong chunk,
2236 0 : ulong sz ) {
2237 : /* Error control frags must be immediately handled. */
2238 0 : if( FD_UNLIKELY( sig==FD_SNAPSHOT_MSG_CTRL_ERROR ) ) {
2239 0 : handle_control_frag( ctx, stem, in_idx, sig, chunk, sz );
2240 0 : return;
2241 0 : }
2242 :
2243 0 : if( FD_UNLIKELY( sig!=ctx->pending_control ) ) {
2244 0 : FD_TEST( ctx->pending_control==ULONG_MAX || sig==FD_SNAPSHOT_MSG_CTRL_FAIL );
2245 0 : clear_control_barrier( ctx );
2246 0 : ctx->pending_control = sig;
2247 0 : }
2248 :
2249 : /* Only process the control frag when all upstream tiles have sent
2250 : the same control message. */
2251 0 : FD_TEST( !ctx->control_seen[ in_idx ] );
2252 0 : ctx->control_seen[ in_idx ] = 1U;
2253 0 : if( FD_LIKELY( !all_controls_seen( ctx ) ) ) {
2254 0 : return;
2255 0 : }
2256 :
2257 : /* All controls received, process the control frag. */
2258 0 : clear_control_barrier( ctx );
2259 0 : handle_control_frag( ctx, stem, in_idx, sig, chunk, sz );
2260 0 : }
2261 :
2262 : static inline int
2263 : returnable_frag( fd_snapin_tile_t * ctx,
2264 : ulong in_idx,
2265 : ulong seq FD_PARAM_UNUSED,
2266 : ulong sig,
2267 : ulong chunk,
2268 : ulong sz,
2269 : ulong ctl,
2270 : ulong tsorig FD_PARAM_UNUSED,
2271 : ulong tspub FD_PARAM_UNUSED,
2272 0 : fd_stem_context_t * stem ) {
2273 0 : FD_TEST( ctx->state!=FD_SNAPSHOT_STATE_SHUTDOWN );
2274 :
2275 0 : if( FD_LIKELY( fd_snapdc_sig_type( sig )==FD_SNAPSHOT_MSG_DATA ) ) return handle_lane_data_frag( ctx, stem, in_idx, chunk, sz, ctl );
2276 0 : else handle_control_barrier( ctx, stem, in_idx, sig, chunk, sz );
2277 :
2278 0 : return 0;
2279 0 : }
2280 :
2281 : static ulong
2282 : populate_allowed_fds( fd_topo_t const * topo,
2283 : fd_topo_tile_t const * tile,
2284 : ulong out_fds_cnt,
2285 0 : int * out_fds ) {
2286 0 : if( FD_UNLIKELY( out_fds_cnt<6UL ) ) FD_LOG_ERR(( "invalid out_fds_cnt %lu", out_fds_cnt ));
2287 0 : fd_snapin_tile_t const * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
2288 :
2289 0 : ulong out_cnt = 0;
2290 0 : out_fds[ out_cnt++ ] = 2UL; /* stderr */
2291 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) ) {
2292 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
2293 0 : }
2294 0 : out_fds[ out_cnt++ ] = FD_ACCDB_FD_RW; /* accounts db */
2295 0 : out_fds[ out_cnt++ ] = FD_STAKE_DELEGATIONS_FD; /* stake delegation disk spill */
2296 0 : out_fds[ out_cnt++ ] = ctx->writer.accdb_direct_fd; /* accounts db, O_DIRECT */
2297 0 : if( FD_LIKELY( !tile->kind_id ) ) out_fds[ out_cnt++ ] = FD_TXNCACHE_FD;
2298 :
2299 0 : return out_cnt;
2300 0 : }
2301 :
2302 : static ulong
2303 : populate_allowed_seccomp( fd_topo_t const * topo,
2304 : fd_topo_tile_t const * tile,
2305 : ulong out_cnt,
2306 0 : struct sock_filter * out ) {
2307 0 : fd_snapin_tile_t const * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
2308 0 : populate_sock_filter_policy_fd_snapin_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), FD_ACCDB_FD_RW, (uint)ctx->writer.accdb_direct_fd, FD_STAKE_DELEGATIONS_FD, (uint)FD_TXNCACHE_FD );
2309 0 : return sock_filter_policy_fd_snapin_tile_instr_cnt;
2310 0 : }
2311 :
2312 : static void
2313 : privileged_init( fd_topo_t const * topo,
2314 0 : fd_topo_tile_t const * tile ) {
2315 0 : fd_snapin_tile_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
2316 0 : memset( ctx, 0, sizeof(fd_snapin_tile_t) );
2317 0 : FD_TEST( fd_rng_secure( &ctx->lead.seed, 8UL ) );
2318 :
2319 0 : char path[ 64 ];
2320 0 : FD_TEST( fd_cstr_printf_check( path, sizeof(path), NULL, "/proc/self/fd/%d", FD_ACCDB_FD_RW ) );
2321 0 : ctx->writer.accdb_direct_fd = open( path, O_WRONLY|O_DIRECT|O_CLOEXEC );
2322 0 : FD_CHECK_ERR( ctx->writer.accdb_direct_fd>=0, "open(accounts.db, O_DIRECT) failed; the filesystem holding [paths.accounts] must support direct IO" );
2323 0 : }
2324 :
2325 : static inline fd_snapin_out_link_t
2326 : out1( fd_topo_t const * topo,
2327 : fd_topo_tile_t const * tile,
2328 : char const * name,
2329 0 : ulong kind_id ) {
2330 0 : ulong idx = fd_topo_find_tile_out_link( topo, tile, name, kind_id );
2331 :
2332 0 : if( FD_UNLIKELY( idx==ULONG_MAX ) ) return (fd_snapin_out_link_t){ .idx = ULONG_MAX, .mem = NULL, .chunk0 = 0, .wmark = 0, .chunk = 0, .mtu = 0 };
2333 :
2334 0 : ulong mtu = topo->links[ tile->out_link_id[ idx ] ].mtu;
2335 0 : if( FD_UNLIKELY( mtu==0UL ) ) return (fd_snapin_out_link_t){ .idx = idx, .mem = NULL, .chunk0 = ULONG_MAX, .wmark = ULONG_MAX, .chunk = ULONG_MAX, .mtu = mtu };
2336 :
2337 0 : void * mem = topo->workspaces[ topo->objs[ topo->links[ tile->out_link_id[ idx ] ].dcache_obj_id ].wksp_id ].wksp;
2338 0 : ulong chunk0 = fd_dcache_compact_chunk0( mem, topo->links[ tile->out_link_id[ idx ] ].dcache );
2339 0 : ulong wmark = fd_dcache_compact_wmark ( mem, topo->links[ tile->out_link_id[ idx ] ].dcache, mtu );
2340 0 : return (fd_snapin_out_link_t){ .idx = idx, .mem = mem, .chunk0 = chunk0, .wmark = wmark, .chunk = chunk0, .mtu = mtu };
2341 0 : }
2342 :
2343 : static void
2344 : unprivileged_init( fd_topo_t const * topo,
2345 0 : fd_topo_tile_t const * tile ) {
2346 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
2347 :
2348 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
2349 0 : fd_snapin_tile_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapin_tile_t), sizeof(fd_snapin_tile_t) );
2350 0 : void * _accdb = FD_SCRATCH_ALLOC_APPEND( l, fd_accdb_align(), fd_accdb_footprint( tile->snapin.max_live_slots, 0 ) );
2351 :
2352 0 : ctx->tile_idx = tile->kind_id;
2353 0 : if( FD_UNLIKELY( ctx->tile_idx>=FD_TOPO_MAX_TILE_IN_LINKS ) ) FD_LOG_ERR(( "tile `" NAME "` has unsupported kind id %lu", tile->kind_id ));
2354 :
2355 0 : ctx->full = 1;
2356 0 : ctx->state = FD_SNAPSHOT_STATE_IDLE;
2357 0 : ctx->lane_cnt = tile->in_cnt;
2358 0 : clear_control_barrier( ctx );
2359 0 : fd_memset( &ctx->metrics, 0, sizeof(ctx->metrics) );
2360 :
2361 0 : void * _accdb_shmem = fd_topo_obj_laddr( topo, tile->snapin.accdb_obj_id );
2362 0 : fd_accdb_shmem_t * accdb_shmem = fd_accdb_shmem_join( _accdb_shmem );
2363 0 : FD_TEST( accdb_shmem );
2364 :
2365 0 : fd_sleep_t * accdb_sleep = topo->sleep_obj_id!=ULONG_MAX ? fd_sleep_join( fd_topo_obj_laddr( topo, topo->sleep_obj_id ) ) : NULL;
2366 0 : ctx->accdb = fd_accdb_join( fd_accdb_new( _accdb, accdb_shmem, FD_ACCDB_FD_RW, 0UL, NULL, accdb_sleep, fd_topo_find_tile( topo, "accdb", 0UL ), 0 ) );
2367 0 : FD_TEST( ctx->accdb );
2368 :
2369 0 : ctx->shmem = fd_topo_obj_laddr( topo, tile->snapin.shmem_obj_id );
2370 :
2371 : /* Every tile updates stakes. Only tile 0 owns the bank. */
2372 0 : fd_banks_t * banks = fd_banks_join( fd_topo_obj_laddr( topo, tile->snapin.banks_obj_id ) );
2373 0 : FD_TEST( banks );
2374 0 : ctx->stake_delegations = fd_banks_stake_delegations_root_query( banks );
2375 0 : FD_TEST( ctx->stake_delegations );
2376 :
2377 0 : ctx->ct_out = out1( topo, tile, "snapin_ct", ctx->tile_idx );
2378 0 : if( FD_UNLIKELY( ctx->ct_out.idx==ULONG_MAX ) ) FD_LOG_ERR(( "tile `" NAME ":%lu` missing required out link `snapin_ct`", ctx->tile_idx ));
2379 :
2380 0 : ctx->gui_out = out1( topo, tile, "snapin_gui", ctx->tile_idx );
2381 :
2382 0 : for( ulong i=0UL; i<ctx->lane_cnt; i++ ) {
2383 0 : fd_topo_link_t const * in_link = &topo->links[ tile->in_link_id[ i ] ];
2384 0 : FD_TEST( 0==strcmp( in_link->name, "snapdc_in" ) );
2385 0 : FD_TEST( in_link->kind_id==i );
2386 0 : fd_topo_wksp_t const * in_wksp = &topo->workspaces[ topo->objs[ in_link->dcache_obj_id ].wksp_id ];
2387 0 : ctx->in[ i ].wksp = in_wksp->wksp;
2388 0 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].wksp, in_link->dcache );
2389 0 : ctx->in[ i ].wmark = fd_dcache_compact_wmark( ctx->in[ i ].wksp, in_link->dcache, in_link->mtu );
2390 0 : ctx->in[ i ].mtu = in_link->mtu;
2391 0 : ctx->in[ i ].pos = 0UL;
2392 0 : }
2393 :
2394 0 : reset_attempt_state( ctx );
2395 0 : if( FD_LIKELY( !is_lead( ctx ) ) ) return;
2396 :
2397 : /* Tile 0 state. */
2398 0 : ctx->lead.init_completed = 0;
2399 0 : ctx->lead.txncache_max_groups_per_slot = tile->snapin.max_txn_per_slot;
2400 0 : ctx->lead.txncache_max_entries_per_slot = 2UL*tile->snapin.max_txn_per_slot;
2401 :
2402 0 : void * _txncache = FD_SCRATCH_ALLOC_APPEND( l, fd_txncache_align(), fd_txncache_footprint( tile->snapin.max_live_slots ) );
2403 0 : void * _manifest_parser = FD_SCRATCH_ALLOC_APPEND( l, fd_ssmanifest_parser_align(), fd_ssmanifest_parser_footprint() );
2404 0 : void * _sd_parser = FD_SCRATCH_ALLOC_APPEND( l, fd_slot_delta_parser_align(), fd_slot_delta_parser_footprint() );
2405 0 : ctx->lead.recent_groups = FD_SCRATCH_ALLOC_APPEND( l, alignof(recent_blockhash_group_t), sizeof(recent_blockhash_group_t)*FD_SNAPIN_MAX_RECENT_GROUPS );
2406 0 : ctx->lead.txncache_entries = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_sstxncache_hash_t), sizeof(fd_sstxncache_hash_t)*FD_TXNCACHE_MAX_SLOT_DELTAS*2UL*tile->snapin.max_txn_per_slot );
2407 0 : ctx->lead.blockhash_groups = FD_SCRATCH_ALLOC_APPEND( l, FD_TXNCACHE_SNAPIN_STAGE_ALIGN, txncache_staging_row_sz( tile->snapin.max_txn_per_slot ) );
2408 :
2409 0 : void * _txncache_shmem = fd_topo_obj_laddr( topo, tile->snapin.txncache_obj_id );
2410 0 : fd_txncache_shmem_t * txncache_shmem = fd_txncache_shmem_join( _txncache_shmem );
2411 0 : FD_TEST( txncache_shmem );
2412 0 : ctx->lead.txncache = fd_txncache_join( fd_txncache_new( _txncache, txncache_shmem, FD_TXNCACHE_FD ) );
2413 0 : FD_TEST( ctx->lead.txncache );
2414 :
2415 0 : txncache_staging_reset( ctx );
2416 :
2417 0 : ctx->lead.bank = fd_banks_init_bank( banks );
2418 0 : FD_TEST( ctx->lead.bank );
2419 0 : FD_TEST( ctx->lead.bank->idx==0UL );
2420 :
2421 0 : ctx->lead.manifest_parser = fd_ssmanifest_parser_join( fd_ssmanifest_parser_new( _manifest_parser ) );
2422 0 : FD_TEST( ctx->lead.manifest_parser );
2423 :
2424 0 : ctx->lead.slot_delta_parser = fd_slot_delta_parser_join( fd_slot_delta_parser_new( _sd_parser ) );
2425 0 : FD_TEST( ctx->lead.slot_delta_parser );
2426 :
2427 0 : ctx->lead.manifest_out = out1( topo, tile, "snapin_manif", 0UL );
2428 0 : if( FD_UNLIKELY( ctx->lead.manifest_out.idx==ULONG_MAX ) ) FD_LOG_ERR(( "tile `" NAME "` missing required out link `snapin_manif`" ));
2429 :
2430 0 : fd_ssmanifest_parser_init( ctx->lead.manifest_parser, fd_chunk_to_laddr( ctx->lead.manifest_out.mem, ctx->lead.manifest_out.chunk ) );
2431 0 : fd_slot_delta_parser_init( ctx->lead.slot_delta_parser );
2432 :
2433 0 : ctx->lead.advertised_slot = 0UL;
2434 0 : ctx->lead.bank_slot = 0UL;
2435 0 : ctx->lead.epoch = 0UL;
2436 0 : ctx->lead.full_genesis_creation_time_seconds = 0UL;
2437 0 : ctx->lead.manifest_capitalization = 0UL;
2438 0 : ctx->lead.recovery.capitalization = 0UL;
2439 0 : fd_memset( &ctx->lead.account_counts, 0, sizeof(ctx->lead.account_counts) );
2440 0 : ctx->lead.accdb_root_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
2441 0 : ctx->lead.accdb_incr_fork_id = (fd_accdb_fork_id_t){ .val = USHORT_MAX };
2442 0 : fd_memset( &ctx->lead.recovery.accdb_metadata, 0, sizeof(ctx->lead.recovery.accdb_metadata) );
2443 0 : fd_memset( &ctx->lead.flags, 0, sizeof(ctx->lead.flags) );
2444 0 : ctx->lead.boot_timestamp = fd_log_wallclock();
2445 :
2446 0 : ctx->wfs_slot = tile->snapin.wait_for_supermajority_at_slot;
2447 0 : ctx->wfs_bank_hash = tile->snapin.wait_for_supermajority_with_bank_hash;
2448 0 : ctx->wfs_shred_version = tile->snapin.expected_shred_version;
2449 0 : ctx->wfs_hash_is_zero = !memcmp( ctx->wfs_bank_hash.uc, ((fd_hash_t){0}).uc, sizeof(fd_hash_t) );
2450 0 : }
2451 :
2452 : /* There are 3 output links that affect the calculation of STEM_BURST:
2453 : 1. snapin_ct - worst case: 1 message (ack or unsolicited ERROR)
2454 : 2. snapin_manif - worst case: 1 message (tile 0 only)
2455 : 3. snapin_gui - worst case: 1 message (config program account)
2456 : The STEM_BURST is the max value across these 3 links (not the sum). */
2457 0 : #define STEM_BURST 1UL
2458 :
2459 0 : #define STEM_LAZY (128L*3000L)
2460 :
2461 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapin_tile_t
2462 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapin_tile_t)
2463 :
2464 0 : #define STEM_CALLBACK_SHOULD_SHUTDOWN should_shutdown
2465 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
2466 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
2467 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
2468 :
2469 : #include "../../disco/stem/fd_stem.c"
2470 :
2471 : static ulong
2472 0 : max_event_sz( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
2473 0 : return sizeof(fd_event_accdb_partition_added_t);
2474 0 : }
2475 :
2476 : static ulong
2477 : snapin_shmem_footprint( fd_topo_t const * topo FD_PARAM_UNUSED,
2478 0 : fd_topo_obj_t const * obj FD_PARAM_UNUSED ) {
2479 0 : return sizeof(fd_snapin_shmem_t);
2480 0 : }
2481 :
2482 : static ulong
2483 : snapin_shmem_align( fd_topo_t const * topo FD_PARAM_UNUSED,
2484 0 : fd_topo_obj_t const * obj FD_PARAM_UNUSED ) {
2485 0 : return alignof(fd_snapin_shmem_t);
2486 0 : }
2487 :
2488 : static void
2489 : snapin_shmem_new( fd_topo_t const * topo,
2490 0 : fd_topo_obj_t const * obj ) {
2491 0 : fd_snapin_shmem_t * shmem = fd_topo_obj_laddr( topo, obj->id );
2492 0 : fd_memset( shmem, 0, sizeof(fd_snapin_shmem_t) );
2493 0 : shmem->fork_id = ULONG_MAX;
2494 0 : }
2495 :
2496 : fd_topo_obj_callbacks_t fd_obj_cb_snapin_shmem = {
2497 : .name = "snapin_shmem",
2498 : .footprint = snapin_shmem_footprint,
2499 : .align = snapin_shmem_align,
2500 : .new = snapin_shmem_new,
2501 : };
2502 :
2503 : fd_topo_run_tile_t fd_tile_snapin = {
2504 : .name = NAME,
2505 : .populate_allowed_fds = populate_allowed_fds,
2506 : .populate_allowed_seccomp = populate_allowed_seccomp,
2507 : .scratch_align = scratch_align,
2508 : .scratch_footprint = scratch_footprint,
2509 : .privileged_init = privileged_init,
2510 : .unprivileged_init = unprivileged_init,
2511 : .max_event_sz = max_event_sz,
2512 : .run = stem_run,
2513 : };
2514 :
2515 : #undef NAME
|
