Skip to content

snapshots: fix pipeline state machine edge cases #7570

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amass-jump merged 1 commit into from
Jan 6, 2026
Merged

snapshots: fix pipeline state machine edge cases #7570

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/discof/restore/fd_snapct_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
/* FIXME: Handle cases where no explicitly allowed peers advertise RPC */
/* FIXME: Make the code more strict about duplicate IP:port's */
/* FIXME: Handle cases where the slot number we start downloading differs from advertised */
/* FIXME: Ensure local files are not selected again if they fail the first time. */

#define GOSSIP_PEERS_MAX (FD_CONTACT_INFO_TABLE_SIZE)
#define SERVER_PEERS_MAX (FD_TOPO_SNAPSHOTS_SERVERS_MAX_RESOLVED)
Expand Down
25 changes: 12 additions & 13 deletions src/discof/restore/fd_snapdc_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,7 +82,7 @@ metrics_write( fd_snapdc_tile_t * ctx ) {
FD_MGAUGE_SET( SNAPDC, INCREMENTAL_COMPRESSED_BYTES_READ, ctx->metrics.incremental.compressed_bytes_read );
FD_MGAUGE_SET( SNAPDC, INCREMENTAL_DECOMPRESSED_BYTES_WRITTEN, ctx->metrics.incremental.decompressed_bytes_written );

FD_MGAUGE_SET( SNAPDC, STATE, (ulong)(ctx->state) );
FD_MGAUGE_SET( SNAPDC, STATE, (ulong)(ctx->state) );
}

static inline void
Expand All @@ -96,7 +96,6 @@ handle_control_frag( fd_snapdc_tile_t * ctx,
/* All control messages cause us to want to reset the decompression stream */
ulong error = ZSTD_DCtx_reset( ctx->zstd, ZSTD_reset_session_only );
if( FD_UNLIKELY( ZSTD_isError( error ) ) ) FD_LOG_ERR(( "ZSTD_DCtx_reset failed (%lu-%s)", error, ZSTD_getErrorName( error ) ));
ctx->dirty = 0;

switch( sig ) {
case FD_SNAPSHOT_MSG_CTRL_INIT_FULL: {
Expand All @@ -106,6 +105,7 @@ handle_control_frag( fd_snapdc_tile_t * ctx,
ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
ctx->full = 1;
ctx->is_zstd = !!msg->zstd;
ctx->dirty = 0;
ctx->in.frag_pos = 0UL;
ctx->metrics.full.compressed_bytes_read = 0UL;
ctx->metrics.full.decompressed_bytes_written = 0UL;
Expand All @@ -118,27 +118,26 @@ handle_control_frag( fd_snapdc_tile_t * ctx,
ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
ctx->full = 0;
ctx->is_zstd = !!msg->zstd;
ctx->dirty = 0;
ctx->in.frag_pos = 0UL;
ctx->metrics.incremental.compressed_bytes_read = 0UL;
ctx->metrics.incremental.decompressed_bytes_written = 0UL;
break;
}
case FD_SNAPSHOT_MSG_CTRL_FAIL:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
Comment on lines -127 to -128
Copy link
Contributor Author

@amass-jump amass-jump Dec 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's also possible for tiles to receive the FAIL control message in the IDLE state. This is the cause of several of the FD_TEST assertions we've seen.

This happens when snapct sends out a DONE, and an early tile immediately handles it and goes to IDLE. But a later tile may fail and generate ERROR, which causes snapct to send out a FAIL control message which is looped back through the pipeline

ctx->state = FD_SNAPSHOT_STATE_IDLE;
break;
case FD_SNAPSHOT_MSG_CTRL_NEXT:
case FD_SNAPSHOT_MSG_CTRL_DONE:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
if( FD_UNLIKELY( ctx->is_zstd && ctx->dirty ) ) {
FD_LOG_WARNING(( "encountered end-of-file in the middle of a compressed frame" ));
ctx->state = FD_SNAPSHOT_STATE_ERROR;
fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
return;
}
Comment on lines -135 to -140
Copy link
Contributor Author

@amass-jump amass-jump Dec 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This check could never trigger before (we always reset dirty to 0 above), but also returning here deadlocks the pipeline

ctx->state = FD_SNAPSHOT_STATE_IDLE;
if( FD_LIKELY( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ) ) {
if( FD_UNLIKELY( ctx->is_zstd && ctx->dirty ) ) {
FD_LOG_WARNING(( "encountered end-of-file in the middle of a compressed frame" ));
ctx->state = FD_SNAPSHOT_STATE_ERROR;
fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
} else {
ctx->state = FD_SNAPSHOT_STATE_IDLE;
}
} else FD_TEST( ctx->state==FD_SNAPSHOT_STATE_ERROR );
break;
case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
Expand Down
34 changes: 17 additions & 17 deletions src/discof/restore/fd_snapin_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -87,7 +87,7 @@ metrics_write( fd_snapin_tile_t * ctx ) {
FD_MGAUGE_SET( SNAPIN, FULL_BYTES_READ, ctx->metrics.full_bytes_read );
FD_MGAUGE_SET( SNAPIN, INCREMENTAL_BYTES_READ, ctx->metrics.incremental_bytes_read );
FD_MGAUGE_SET( SNAPIN, ACCOUNTS_INSERTED, ctx->metrics.accounts_inserted );
FD_MGAUGE_SET( SNAPIN, STATE, (ulong)ctx->state );
FD_MGAUGE_SET( SNAPIN, STATE, (ulong)ctx->state );
}

/* verify_slot_deltas_with_slot_history verifies the 'SlotHistory'
Expand Down Expand Up @@ -161,6 +161,7 @@ verify_slot_deltas_with_bank_slot( fd_snapin_tile_t * ctx,
static void
transition_malformed( fd_snapin_tile_t * ctx,
fd_stem_context_t * stem ) {
if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) return;
ctx->state = FD_SNAPSHOT_STATE_ERROR;
fd_stem_publish( stem, ctx->out_ct_idx, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
}
Comment on lines 162 to 167
Copy link
Contributor Author

@amass-jump amass-jump Dec 9, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not a bug, but no need to generate extra ERROR messages when we are already in that state

Expand Down Expand Up @@ -593,22 +594,21 @@ handle_control_frag( fd_snapin_tile_t * ctx,
break;

case FD_SNAPSHOT_MSG_CTRL_FAIL:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
ctx->state = FD_SNAPSHOT_STATE_IDLE;
if( ctx->state!=FD_SNAPSHOT_STATE_IDLE ) {
ctx->state = FD_SNAPSHOT_STATE_IDLE;

if( ctx->use_vinyl ) {
fd_snapin_vinyl_wd_fini( ctx );
if( ctx->vinyl.txn_active ) {
fd_snapin_vinyl_txn_cancel( ctx );
}
} else {
if( ctx->full ) {
fd_accdb_clear( ctx->accdb_admin );
if( ctx->use_vinyl ) {
fd_snapin_vinyl_wd_fini( ctx );
if( ctx->vinyl.txn_active ) {
fd_snapin_vinyl_txn_cancel( ctx );
}
} else {
fd_accdb_cancel( ctx->accdb_admin, ctx->xid );
fd_funk_txn_xid_copy( ctx->xid, fd_funk_last_publish( ctx->accdb_admin->funk ) );
if( ctx->full ) {
fd_accdb_clear( ctx->accdb_admin );
} else {
fd_accdb_cancel( ctx->accdb_admin, ctx->xid );
fd_funk_txn_xid_copy( ctx->xid, fd_funk_last_publish( ctx->accdb_admin->funk ) );
}
}
}
break;
Expand All @@ -619,7 +619,7 @@ handle_control_frag( fd_snapin_tile_t * ctx,
ctx->state==FD_SNAPSHOT_STATE_ERROR );
if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
transition_malformed( ctx, stem );
return;
break;
Copy link
Contributor

@cali-jumptrading cali-jumptrading Dec 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why the break here instead of return? Do we still need to forward the control message if we are generating an error?

Copy link
Contributor Author

@amass-jump amass-jump Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, every control message needs to be forwarded down the pipeline immediately, or deferred for later (such as with snapls). But we can't outright drop a control message or the pipeline will be locked on waiting for that message to be flushed forever (each control message is only generated a single time and we do not generate new control messages until it's been flushed).

}
ctx->state = FD_SNAPSHOT_STATE_IDLE;

Expand All @@ -642,7 +642,7 @@ handle_control_frag( fd_snapin_tile_t * ctx,
ctx->state==FD_SNAPSHOT_STATE_ERROR );
if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
transition_malformed( ctx, stem );
return;
break;
}
ctx->state = FD_SNAPSHOT_STATE_IDLE;

Expand Down
20 changes: 10 additions & 10 deletions src/discof/restore/fd_snapla_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,7 @@ metrics_write( fd_snapla_tile_t * ctx ) {
static void
transition_malformed( fd_snapla_tile_t * ctx,
fd_stem_context_t * stem ) {
if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_ERROR ) ) return;
ctx->state = FD_SNAPSHOT_STATE_ERROR;
fd_stem_publish( stem, FD_SNAPLA_OUT_CTRL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
}
Expand Down Expand Up @@ -248,23 +249,22 @@ handle_control_frag( fd_snapla_tile_t * ctx,
case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
case FD_SNAPSHOT_MSG_CTRL_INIT_INCR:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_IDLE );
ctx->full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
ctx->full = sig==FD_SNAPSHOT_MSG_CTRL_INIT_FULL;
ctx->state = FD_SNAPSHOT_STATE_PROCESSING;
ctx->accounts_seen = 0UL;
ctx->hash_account = 0;
ctx->acc_data_sz = 0UL;
fd_memset( &ctx->account_hdr, 0, sizeof(ctx->account_hdr) );
fd_lthash_zero( &ctx->running_lthash );
fd_ssparse_reset( ctx->ssparse );
fd_ssmanifest_parser_init( ctx->manifest_parser, ctx->manifest );
fd_lthash_adder_new( ctx->adder );
if( ctx->full ) ctx->metrics.full.accounts_hashed = 0UL;
ctx->metrics.incremental.accounts_hashed = 0UL;
break;

case FD_SNAPSHOT_MSG_CTRL_FAIL:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
ctx->state = FD_SNAPSHOT_STATE_IDLE;
fd_lthash_zero( &ctx->running_lthash );
fd_ssparse_reset( ctx->ssparse );
fd_ssmanifest_parser_init( ctx->manifest_parser, ctx->manifest );
fd_lthash_adder_new( ctx->adder );
break;

case FD_SNAPSHOT_MSG_CTRL_NEXT:
Expand All @@ -274,7 +274,7 @@ handle_control_frag( fd_snapla_tile_t * ctx,
ctx->state==FD_SNAPSHOT_STATE_ERROR );
if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
transition_malformed( ctx, stem );
return;
break;
}
fd_lthash_adder_flush( ctx->adder, &ctx->running_lthash );
uchar * lthash_out = fd_chunk_to_laddr( ctx->out.wksp, ctx->out.chunk );
Expand Down
57 changes: 29 additions & 28 deletions src/discof/restore/fd_snapld_tile.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,11 @@ typedef struct fd_snapld_tile {
char path[ PATH_MAX ];
} config;

int state;
int load_full;
int load_file;
int sent_meta;

int awaiting_ack;
ulong awaiting_ack_sig;
int state;
ulong pending_ctrl_sig;
int load_full;
int load_file;
int sent_meta;

int local_full_fd;
int local_incr_fd;
Expand Down Expand Up @@ -171,7 +169,8 @@ unprivileged_init( fd_topo_t * topo,

fd_memcpy( ctx->config.path, tile->snapld.snapshots_path, PATH_MAX );

ctx->state = FD_SNAPSHOT_STATE_IDLE;
ctx->state = FD_SNAPSHOT_STATE_IDLE;
ctx->pending_ctrl_sig = 0UL;

FD_TEST( tile->in_cnt==1UL );
fd_topo_link_t const * in_link = &topo->links[ tile->in_link_id[ 0 ] ];
Expand All @@ -187,10 +186,6 @@ unprivileged_init( fd_topo_t * topo,
ctx->out_dc.chunk = ctx->out_dc.chunk0;
ctx->out_dc.mtu = out_link->mtu;

ctx->awaiting_ack = 0;
ctx->awaiting_ack_sig = 0;
ctx->is_https = 0;

/* We can only close the temporary socket file descriptor after
entering the sandbox because the sandbox checks all file
descriptors are existent. */
Expand All @@ -215,11 +210,16 @@ after_credit( fd_snapld_tile_t * ctx,
fd_stem_context_t * stem,
int * opt_poll_in FD_PARAM_UNUSED,
int * charge_busy ) {
if( FD_UNLIKELY( ctx->awaiting_ack && ctx->state==FD_SNAPSHOT_STATE_FINISHING ) ) {
ctx->awaiting_ack = 0;
ctx->state = FD_SNAPSHOT_STATE_IDLE;
fd_stem_publish( stem, 0UL, ctx->awaiting_ack_sig, 0UL, 0UL, 0UL, 0UL, 0UL );
return;
if( FD_UNLIKELY( ctx->pending_ctrl_sig ) ) {
FD_TEST( !ctx->load_file && ctx->is_https );
FD_TEST( ctx->pending_ctrl_sig==FD_SNAPSHOT_MSG_CTRL_NEXT ||
ctx->pending_ctrl_sig==FD_SNAPSHOT_MSG_CTRL_DONE );
if( ctx->state==FD_SNAPSHOT_STATE_FINISHING || ctx->state==FD_SNAPSHOT_STATE_ERROR ) {
fd_stem_publish( stem, 0UL, ctx->pending_ctrl_sig, 0UL, 0UL, 0UL, 0UL, 0UL );
ctx->pending_ctrl_sig = 0UL;
if( ctx->state!=FD_SNAPSHOT_STATE_ERROR ) ctx->state = FD_SNAPSHOT_STATE_IDLE;
return;
} else FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING );
}

if( ctx->state!=FD_SNAPSHOT_STATE_PROCESSING ) {
Expand Down Expand Up @@ -279,6 +279,7 @@ after_credit( fd_snapld_tile_t * ctx,
case FD_SSHTTP_ADVANCE_ERROR:
ctx->state = FD_SNAPSHOT_STATE_ERROR;
fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_CTRL_ERROR, 0UL, 0UL, 0UL, 0UL, 0UL );
fd_sshttp_cancel( ctx->sshttp );
break;
default: FD_LOG_ERR(( "unexpected fd_sshttp_advance result %d", result ));
}
Expand All @@ -296,6 +297,8 @@ returnable_frag( fd_snapld_tile_t * ctx,
ulong tsorig FD_PARAM_UNUSED,
ulong tspub FD_PARAM_UNUSED,
fd_stem_context_t * stem ) {
FD_TEST( !ctx->pending_ctrl_sig );

switch( sig ) {

case FD_SNAPSHOT_MSG_CTRL_INIT_FULL:
Expand Down Expand Up @@ -323,9 +326,7 @@ returnable_frag( fd_snapld_tile_t * ctx,
}

case FD_SNAPSHOT_MSG_CTRL_FAIL:
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
FD_TEST( ctx->state!=FD_SNAPSHOT_MSG_CTRL_SHUTDOWN );
fd_sshttp_cancel( ctx->sshttp );
ctx->state = FD_SNAPSHOT_STATE_IDLE;
break;
Expand All @@ -335,19 +336,19 @@ returnable_frag( fd_snapld_tile_t * ctx,
FD_TEST( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ||
ctx->state==FD_SNAPSHOT_STATE_FINISHING ||
ctx->state==FD_SNAPSHOT_STATE_ERROR );
if( FD_UNLIKELY( ctx->state!=FD_SNAPSHOT_STATE_FINISHING ) ) {
if( FD_UNLIKELY( ctx->state==FD_SNAPSHOT_STATE_PROCESSING ) ) {
/* snapld should be in the finishing state when reading from a
file or downloading from http. It is only allowed to still
be in progress for shutting down an https connection. */
be in progress for shutting down an https connection. Save
the sig here and send the message when snapld is in the
finishing state. */
FD_TEST( ctx->is_https );
/* snapld might not be done with shutting down an https
connection. Save the sig here and send the message when
snapld is in the finishing state. */
ctx->awaiting_ack = 1;
ctx->awaiting_ack_sig = sig;
ctx->pending_ctrl_sig = sig;
return 0; /* return directly to avoid fowarding the message */
}
ctx->state = FD_SNAPSHOT_STATE_IDLE;
else if( FD_LIKELY( ctx->state==FD_SNAPSHOT_STATE_FINISHING ) ) {
ctx->state = FD_SNAPSHOT_STATE_IDLE;
}
break;

case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN:
Expand Down
Loading
Loading