snapshots: highest manifest slot

cali-jumptrading · cali-jumptrading · commit 2673ae16842e · 2025-08-08T20:16:30.000Z
diff --git a/src/discof/replay/fd_replay_tile.c b/src/discof/replay/fd_replay_tile.c
@@ -792,7 +792,9 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
                      fd_stem_context_t *    stem,
                      ulong                  in_idx,
                      ulong                  chunk,
-                     ulong                  sig ) {
+                     ulong                  sig,
+                     ulong                  tsorig FD_PARAM_UNUSED,
+                     ulong                  tspub  FD_PARAM_UNUSED) {
   ulong msg = fd_ssmsg_sig_message( sig );
   if( FD_LIKELY( msg==FD_SSMSG_DONE ) ) {
     /* An end of message notification indicates the snapshot is loaded.
@@ -801,19 +803,6 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
        state machine and set the state here accordingly. */
     FD_LOG_INFO(("Snapshot loaded, replay can start executing"));
     ctx->snapshot_init_done = 1;
-    /* Kickoff repair orphans after the snapshots are done loading. If
-       we kickoff repair after we receive a full manifest, we might try
-       to repair a slot that is potentially huge amount of slots behind
-       turbine causing our repair buffers to fill up. Instead, we should
-       wait until we are done receiving all the snapshots.
-
-       TODO: Eventually, this logic should be cased out more:
-       1. If we just have a full snapshot, load in the slot_ctx for the
-          slot ctx and kickoff repair as soon as the manifest is
-          received.
-       2. If we are loading a full and incremental snapshot, we should
-          only load in the slot_ctx and kickoff repair for the
-          incremental snapshot. */
     kickoff_repair_orphans( ctx, stem );
     init_from_snapshot( ctx, stem );
     ulong curr_slot = fd_bank_slot_get( ctx->slot_ctx->bank );
@@ -824,6 +813,16 @@ on_snapshot_message( fd_replay_tile_ctx_t * ctx,
   }
 
   switch( msg ) {
+    case FD_SSMSG_HIGHEST_MANIFEST_SLOT: {
+      /* snapin sends the highest manifest slot so far to notify any
+         listeners of the highest rooted slot that will be loaded
+         by the snapshot tiles.  The highest manifest slot may change
+         if the snapshot tiles retry snapshot loading, but it is
+         guaranteed to be monotonically increasing. */
+      /* TODO: currently a no-op.  Repair cannot handle an receiving an
+         incrementally changing rooted slot and stake weights yet. */
+      break;
+    }
     case FD_SSMSG_MANIFEST_FULL:
     case FD_SSMSG_MANIFEST_INCREMENTAL: {
       /* We may either receive a full snapshot manifest or an
@@ -944,8 +943,8 @@ after_frag( fd_replay_tile_ctx_t *   ctx,
             ulong                    seq FD_PARAM_UNUSED,
             ulong                    sig,
             ulong                    sz FD_PARAM_UNUSED,
-            ulong                    tsorig FD_PARAM_UNUSED,
-            ulong                    tspub FD_PARAM_UNUSED,
+            ulong                    tsorig,
+            ulong                    tspub,
             fd_stem_context_t *      stem FD_PARAM_UNUSED ) {
   if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_ROOT ) ) {
     ulong root = sig;
@@ -974,7 +973,7 @@ after_frag( fd_replay_tile_ctx_t *   ctx,
 
     fd_fseq_update( ctx->published_wmark, root );
   } else if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_SNAP ) ) {
-    on_snapshot_message( ctx, stem, in_idx, ctx->_snap_out_chunk, sig );
+    on_snapshot_message( ctx, stem, in_idx, ctx->_snap_out_chunk, sig, tsorig, tspub );
   } else if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_REPAIR ) ) {
 
     /* Forks form a partial ordering over FEC sets. The Repair tile
diff --git a/src/discof/restore/fd_snapdc_tile.c b/src/discof/restore/fd_snapdc_tile.c
@@ -101,9 +101,11 @@ transition_malformed( fd_snapdc_tile_t *  ctx,
 static inline void
 handle_control_frag( fd_snapdc_tile_t *  ctx,
                      fd_stem_context_t * stem,
-                     ulong               sig ) {
+                     ulong               sig,
+                     ulong               tsorig,
+                     ulong               tspub ) {
   /* 1. Pass the control message downstream to the next consumer. */
-  fd_stem_publish( stem, 0UL, sig, ctx->out.chunk, 0UL, 0UL, 0UL, 0UL );
+  fd_stem_publish( stem, 0UL, sig, ctx->out.chunk, 0UL, 0UL, tsorig, tspub );
   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 ) ));
 
@@ -148,6 +150,9 @@ handle_control_frag( fd_snapdc_tile_t *  ctx,
       FD_TEST( ctx->state==FD_SNAPDC_STATE_DONE );
       ctx->state = FD_SNAPDC_STATE_SHUTDOWN;
       break;
+    case FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT:
+      /* Informational control message forwarded to snapin */
+      break;
     default:
       FD_LOG_ERR(( "unexpected control sig %lu", sig ));
       return;
@@ -253,7 +258,7 @@ returnable_frag( fd_snapdc_tile_t *  ctx,
   FD_TEST( ctx->state!=FD_SNAPDC_STATE_SHUTDOWN );
 
   if( FD_LIKELY( sig==FD_SNAPSHOT_MSG_DATA ) ) return handle_data_frag( ctx, stem, chunk, sz );
-  else                                                handle_control_frag( ctx,stem, sig );
+  else                                                handle_control_frag( ctx,stem, sig, tsorig, tspub );
 
   return 0;
 }
diff --git a/src/discof/restore/fd_snapin_tile.c b/src/discof/restore/fd_snapin_tile.c
@@ -214,7 +214,9 @@ handle_data_frag( fd_snapin_tile_t *  ctx,
 static void
 handle_control_frag( fd_snapin_tile_t *  ctx,
                      fd_stem_context_t * stem,
-                     ulong               sig ) {
+                     ulong               sig,
+                     ulong               tsorig,
+                     ulong               tspub ) {
   switch( sig ) {
     case FD_SNAPSHOT_MSG_CTRL_RESET_FULL:
       ctx->full = 1;
@@ -257,6 +259,11 @@ handle_control_frag( fd_snapin_tile_t *  ctx,
     case FD_SNAPSHOT_MSG_CTRL_SHUTDOWN:
       ctx->state = FD_SNAPIN_STATE_SHUTDOWN;
       break;
+    case FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT:
+      /* Informational control message forwarded to snap_out */
+      fd_stem_publish( stem, 0UL, FD_SSMSG_HIGHEST_MANIFEST_SLOT, 0UL, 0UL, 0UL, tsorig, tspub );
+      /* No need to send an ack for an informational control message */
+      return;
     default:
       FD_LOG_ERR(( "unexpected control sig %lu", sig ));
       return;
@@ -288,7 +295,7 @@ returnable_frag( fd_snapin_tile_t *  ctx,
   FD_TEST( ctx->state!=FD_SNAPIN_STATE_SHUTDOWN );
 
   if( FD_UNLIKELY( sig==FD_SNAPSHOT_MSG_DATA ) ) handle_data_frag( ctx, chunk, sz, stem );
-  else                                           handle_control_frag( ctx, stem, sig  );
+  else                                           handle_control_frag( ctx, stem, sig, tsorig, tspub  );
 
   return 0;
 }
diff --git a/src/discof/restore/fd_snaprd_tile.c b/src/discof/restore/fd_snaprd_tile.c
@@ -2,6 +2,7 @@
 #include "utils/fd_sshttp.h"
 #include "utils/fd_ssctrl.h"
 #include "utils/fd_ssarchive.h"
+#include "utils/fd_ssmsg.h"
 
 #include "../../disco/topo/fd_topo.h"
 #include "../../disco/metrics/fd_metrics.h"
@@ -74,10 +75,17 @@ struct fd_snaprd_tile {
   } local_in;
 
   struct {
-    char full_path[ PATH_MAX ];
-    ulong full_len;
-    char incremental_path[ PATH_MAX ];
-    ulong incremental_len;
+
+    struct {
+      char  path[ PATH_MAX ];
+      ulong len;
+    } full;
+
+    struct {
+      char  path[ PATH_MAX ];
+      ulong len;
+    } incremental;
+
   } http;
 
   struct {
@@ -328,11 +336,11 @@ rename_snapshots( fd_snaprd_tile_t * ctx ) {
   if( FD_UNLIKELY( -1==ctx->local_out.dir_fd ) ) return;
 
   if( FD_LIKELY( -1!=ctx->local_out.full_snapshot_fd ) ) {
-    if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.full_path ) ) )
+    if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.full.path ) ) )
       FD_LOG_ERR(( "renameat() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
   }
   if( FD_LIKELY( -1!=ctx->local_out.incremental_snapshot_fd ) ) {
-    if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "incremental-snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.incremental_path ) ) )
+    if( FD_UNLIKELY( -1==renameat( ctx->local_out.dir_fd, "incremental-snapshot.tar.bz2-partial", ctx->local_out.dir_fd, ctx->http.incremental.path ) ) )
       FD_LOG_ERR(( "renameat() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
   }
 }
@@ -361,7 +369,7 @@ after_credit( fd_snaprd_tile_t *  ctx,
 
   switch ( ctx->state ) {
     case FD_SNAPRD_STATE_WAITING_FOR_PEERS: {
-      fd_sspeer_t best = fd_ssping_best( ctx->ssping );
+      fd_sspeer_t best = fd_ssping_best( ctx->ssping, 0UL );
       if( FD_LIKELY( best.addr.l ) ) {
         ctx->state = FD_SNAPRD_STATE_COLLECTING_PEERS;
         ctx->deadline_nanos = now + 500L*1000L*1000L;
@@ -371,7 +379,14 @@ after_credit( fd_snaprd_tile_t *  ctx,
     case FD_SNAPRD_STATE_COLLECTING_PEERS: {
       if( FD_UNLIKELY( now<ctx->deadline_nanos ) ) break;
 
-      fd_sspeer_t best = fd_ssping_best( ctx->ssping );
+      ulong highest_slot = 0UL;
+      if( FD_LIKELY( ctx->peer.addr.l ) ) {
+        highest_slot = ctx->peer.snapshot_info->incremental.slot!=ULONG_MAX ?
+                       ctx->peer.snapshot_info->incremental.slot :
+                       ctx->peer.snapshot_info->full.slot;
+      }
+
+      fd_sspeer_t best = fd_ssping_best( ctx->ssping, highest_slot );
       if( FD_UNLIKELY( !best.addr.l ) ) {
         ctx->state = FD_SNAPRD_STATE_WAITING_FOR_PEERS;
         break;
@@ -384,17 +399,29 @@ after_credit( fd_snaprd_tile_t *  ctx,
       } else {
         char path[ PATH_MAX ];
         char encoded_full_hash[ FD_BASE58_ENCODED_32_SZ ];
-        char encoded_incremental_hash[ FD_BASE58_ENCODED_32_SZ ];
+
+        /* Generate the http paths */
         fd_base58_encode_32( best.snapshot_info->full.hash, NULL, encoded_full_hash );
-        fd_base58_encode_32( best.snapshot_info->incremental.hash, NULL, encoded_incremental_hash );
-        FD_TEST( fd_cstr_printf_check( ctx->http.full_path, PATH_MAX, &ctx->http.full_len, "snapshot-%lu-%s.tar.zst", best.snapshot_info->full.slot, encoded_full_hash ) );
-        FD_TEST( fd_cstr_printf_check( ctx->http.incremental_path, PATH_MAX, &ctx->http.incremental_len, "incremental-snapshot-%lu-%lu-%s.tar.zst", best.snapshot_info->incremental.base_slot, best.snapshot_info->incremental.slot, encoded_incremental_hash ) );
-        FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.full_path ) );
+        FD_TEST( fd_cstr_printf_check( ctx->http.full.path, PATH_MAX, &ctx->http.full.len, "snapshot-%lu-%s.tar.zst", best.snapshot_info->full.slot, encoded_full_hash ) );
+        FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.full.path ) );
 
-        FD_LOG_NOTICE(( "downloading full snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( best.addr.addr ), best.addr.port, ctx->http.full_path ));
+        if( ctx->config.incremental_snapshot_fetch ) {
+          char encoded_incremental_hash[ FD_BASE58_ENCODED_32_SZ ];
+          fd_base58_encode_32( best.snapshot_info->incremental.hash, NULL, encoded_incremental_hash );
+          FD_TEST( fd_cstr_printf_check( ctx->http.incremental.path, PATH_MAX, &ctx->http.incremental.len, "incremental-snapshot-%lu-%lu-%s.tar.zst", best.snapshot_info->incremental.base_slot, best.snapshot_info->incremental.slot, encoded_incremental_hash ) );
+        }
+
+        uint low;
+        uint high;
+        /* send the highest manifest slot */
+        ulong highest_manifest_slot = ctx->config.incremental_snapshot_fetch ? best.snapshot_info->incremental.slot : best.snapshot_info->full.slot;
+        fd_ssmsg_slot_to_frag( highest_manifest_slot, &low, &high );
+        fd_stem_publish( stem, 0UL, FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT, 0UL, 0UL, 0UL, low, high );
+
+        FD_LOG_NOTICE(( "downloading full snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( best.addr.addr ), best.addr.port, ctx->http.full.path ));
         ctx->peer  = best;
         ctx->state = FD_SNAPRD_STATE_READING_FULL_HTTP;
-        fd_sshttp_init( ctx->sshttp, best.addr, path, ctx->http.full_len + 1UL, now );
+        fd_sshttp_init( ctx->sshttp, best.addr, path, ctx->http.full.len + 1UL, now );
       }
       break;
     }
@@ -459,9 +486,9 @@ after_credit( fd_snaprd_tile_t *  ctx,
       }
 
       char path[ PATH_MAX ];
-      FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.incremental_path ) );
+      FD_TEST( fd_cstr_printf_check( path, PATH_MAX, NULL, "/%s", ctx->http.incremental.path ) );
       FD_LOG_NOTICE(( "downloading incremental snapshot from http://" FD_IP4_ADDR_FMT ":%hu/%s", FD_IP4_ADDR_FMT_ARGS( ctx->peer.addr.addr ), ctx->peer.addr.port, path));
-      fd_sshttp_init( ctx->sshttp, ctx->peer.addr, path, ctx->http.incremental_len + 1UL, fd_log_wallclock() );
+      fd_sshttp_init( ctx->sshttp, ctx->peer.addr, path, ctx->http.incremental.len + 1UL, fd_log_wallclock() );
       ctx->state = FD_SNAPRD_STATE_READING_INCREMENTAL_HTTP;
       break;
     case FD_SNAPRD_STATE_FLUSHING_FULL_HTTP_RESET:
@@ -691,6 +718,8 @@ unprivileged_init( fd_topo_t *      topo,
   ctx->out.wmark  = fd_dcache_compact_wmark ( ctx->out.wksp, topo->links[ tile->out_link_id[ 0 ] ].dcache, topo->links[ tile->out_link_id[ 0 ] ].mtu );
   ctx->out.chunk  = ctx->out.chunk0;
   ctx->out.mtu    = topo->links[ tile->out_link_id[ 0 ] ].mtu;
+
+  fd_memset( &ctx->peer, 0, sizeof(ctx->peer) );
 }
 
 #define STEM_BURST 2UL /* One control message, and one data message */
diff --git a/src/discof/restore/utils/fd_ssctrl.h b/src/discof/restore/utils/fd_ssctrl.h
@@ -40,13 +40,23 @@
 
 #define FD_SNAPSHOT_MSG_DATA                   (0UL) /* Fragment represents some snapshot data */
 
-#define FD_SNAPSHOT_MSG_CTRL_RESET_FULL        (1UL) /* Reset to start loading a fresh full snapshot */
-#define FD_SNAPSHOT_MSG_CTRL_EOF_FULL          (2UL) /* Full snapshot data is done, incremental data starting now */
-#define FD_SNAPSHOT_MSG_CTRL_RESET_INCREMENTAL (3UL) /* Incremental data being retried, start incremental over */
-#define FD_SNAPSHOT_MSG_CTRL_DONE              (4UL) /* Snapshot load is over, data is finished for this tile */
-#define FD_SNAPSHOT_MSG_CTRL_SHUTDOWN          (5UL) /* All tiles have acknowledged snapshot load is done, can now shutdown */
-
-#define FD_SNAPSHOT_MSG_CTRL_ACK               (6UL) /* Sent from tiles back to snaprd, meaning they ACK whatever control message was pending */
-#define FD_SNAPSHOT_MSG_CTRL_MALFORMED         (7UL) /* Sent from tiles back to snaprd, meaning they consider the current snapshot malformed */
+/* The HIGHEST_MANIFEST_SLOT message is an informational control message
+   that is forwarded from the snaprd tile to the snapin tile, which
+   forwards the message to the snap_out link.  The HIGHEST_MANIFEST_SLOT
+   message contains the highest manifest slot so far.  It is typically
+   the incremental snapshot slot but can be the full snapshot slot if
+   incremental snapshots are disabled.  It is guaranteed to be
+   monotonically increasing and is forwarded regardless of any snapshot
+   loading error / retry. */
+#define FD_SNAPSHOT_MSG_HIGHEST_MANIFEST_SLOT  (1UL) /* Fragment contains the highest manifest slot so far, guaranteed to be monotonically increasing */
+
+#define FD_SNAPSHOT_MSG_CTRL_RESET_FULL        (2UL) /* Reset to start loading a fresh full snapshot */
+#define FD_SNAPSHOT_MSG_CTRL_EOF_FULL          (3UL) /* Full snapshot data is done, incremental data starting now */
+#define FD_SNAPSHOT_MSG_CTRL_RESET_INCREMENTAL (4UL) /* Incremental data being retried, start incremental over */
+#define FD_SNAPSHOT_MSG_CTRL_DONE              (5UL) /* Snapshot load is over, data is finished for this tile */
+#define FD_SNAPSHOT_MSG_CTRL_SHUTDOWN          (6UL) /* All tiles have acknowledged snapshot load is done, can now shutdown */
+
+#define FD_SNAPSHOT_MSG_CTRL_ACK               (7UL) /* Sent from tiles back to snaprd, meaning they ACK whatever control message was pending */
+#define FD_SNAPSHOT_MSG_CTRL_MALFORMED         (8UL) /* Sent from tiles back to snaprd, meaning they consider the current snapshot malformed */
 
 #endif /* HEADER_fd_src_discof_restore_utils_fd_ssctrl_h */
diff --git a/src/discof/restore/utils/fd_sshttp.c b/src/discof/restore/utils/fd_sshttp.c
@@ -36,9 +36,6 @@ struct fd_sshttp_private {
   ulong response_len;
   char  response[ USHORT_MAX ];
 
-  char full_snapshot_name[ PATH_MAX ];
-  char incremental_snapshot_name[ PATH_MAX ];
-
   ulong content_len;
 
   ulong magic;
@@ -292,14 +289,6 @@ read_body( fd_sshttp_t * http,
   return FD_SSHTTP_ADVANCE_DATA;
 }
 
-void
-fd_sshttp_snapshot_names( fd_sshttp_t * http,
-                         char const **  full_snapshot_name,
-                         char const **  incremental_snapshot_name ) {
-  *full_snapshot_name        = http->full_snapshot_name;
-  *incremental_snapshot_name = http->incremental_snapshot_name;
-}
-
 int
 fd_sshttp_advance( fd_sshttp_t * http,
                    ulong *       data_len,
diff --git a/src/discof/restore/utils/fd_sshttp.h b/src/discof/restore/utils/fd_sshttp.h
@@ -24,12 +24,6 @@ fd_sshttp_new( void * shmem );
 fd_sshttp_t *
 fd_sshttp_join( void * sshttp );
 
-/* Sets points to snapshot names */
-void
-fd_sshttp_snapshot_names( fd_sshttp_t * http,
-                         char const **  full_snapshot_name,
-                         char const **  incremental_snapshot_name );
-
 void
 fd_sshttp_init( fd_sshttp_t * http,
                 fd_ip4_port_t addr,
diff --git a/src/discof/restore/utils/fd_ssmsg.h b/src/discof/restore/utils/fd_ssmsg.h
@@ -3,9 +3,10 @@
 
 #include "../../../flamenco/types/fd_types.h"
 
-#define FD_SSMSG_MANIFEST_FULL        (0) /* A snapshot manifest message from the full snapshot */
-#define FD_SSMSG_MANIFEST_INCREMENTAL (1) /* A snapshot manifest message from the incremental snapshot */
-#define FD_SSMSG_DONE                 (2) /* Indicates the snapshot is fully loaded and tiles are shutting down */
+#define FD_SSMSG_MANIFEST_FULL         (0) /* A snapshot manifest message from the full snapshot */
+#define FD_SSMSG_MANIFEST_INCREMENTAL  (1) /* A snapshot manifest message from the incremental snapshot */
+#define FD_SSMSG_HIGHEST_MANIFEST_SLOT (2) /* The highest manifest slot so far, guaranteed to be monotonically increasing */
+#define FD_SSMSG_DONE                  (3) /* Indicates the snapshot is fully loaded and tiles are shutting down */
 
 /* This number is exhagerately high, but is consisent with Frankendancer.
    We need this to be about the number of validators, so roughly
@@ -24,6 +25,24 @@ fd_ssmsg_sig( ulong message,
 FD_FN_CONST static inline ulong fd_ssmsg_sig_manifest_size( ulong sig ) { return (sig >> 2); }
 FD_FN_CONST static inline ulong fd_ssmsg_sig_message( ulong sig ) { return (sig & 0x3UL); }
 
+/* The FD_SSMSG_HIGHEST_MANIFEST_SLOT uses the tsorig and tspub fields
+   of the fd_frag_meta_t struct to store the low and high 32 bits of
+   the slot number. */
+static inline void
+fd_ssmsg_slot_to_frag( ulong slot,
+                       uint* low,
+                       uint* high ) {
+   *low = (uint)(slot & 0xFFFFFFFFUL);
+   *high = (uint)((slot >> 32UL) & 0xFFFFFFFFUL);
+}
+
+static inline void
+fd_ssmsg_frag_to_slot( ulong low,
+                       ulong high,
+                       ulong* slot ) {
+   *slot = (high << 32UL) | low;
+}
+
 struct fd_snapshot_manifest_vote_account {
   /* The pubkey of the vote account */
   uchar vote_account_pubkey[ 32UL ];
diff --git a/src/discof/restore/utils/fd_ssping.c b/src/discof/restore/utils/fd_ssping.c
diff --git a/src/discof/restore/utils/fd_ssping.h b/src/discof/restore/utils/fd_ssping.h
diff --git a/src/discof/restore/utils/fd_ssresolve.c b/src/discof/restore/utils/fd_ssresolve.c
diff --git a/src/discof/restore/utils/fd_ssresolve.h b/src/discof/restore/utils/fd_ssresolve.h
