snapshots: highest manifest slot

Author: cali-jumptrading

src/discof/replay/fd_replay_tile.c

@@ -921,7 +921,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.
@@ -930,25 +932,22 @@ 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 );
     return;
   }
 
   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
@@ -1043,8 +1042,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_TOWER ) ) {
     ulong root = sig;
@@ -1074,7 +1073,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

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;
@@ -254,7 +259,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;
 }

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;
@@ -289,7 +296,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;
 }

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 ) ));
   }
 }
@@ -362,7 +370,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;
@@ -372,7 +380,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;
@@ -385,17 +400,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 ) );
+
+        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 ));
+        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;
     }
@@ -460,9 +487,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:
@@ -692,6 +719,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 */

src/discof/restore/utils/fd_ssctrl.h

@@ -40,13 +40,20 @@
 
 #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 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 */

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 ];

src/discof/restore/utils/fd_ssping.c

@@ -242,14 +242,14 @@ fd_ssping_add( fd_ssping_t * ssping,
     if( FD_UNLIKELY( !peer_pool_free( ssping->pool ) ) ) return;
     peer = peer_pool_ele_acquire( ssping->pool );
     FD_TEST( peer );
-    peer->refcnt = 0UL;
-    peer->state  = PEER_STATE_UNPINGED;
-    peer->addr   = addr;
+    peer->refcnt                              = 0UL;
+    peer->state                               = PEER_STATE_UNPINGED;
+    peer->addr                                = addr;
     peer->snapshot_info.full.slot             = ULONG_MAX;
     peer->snapshot_info.incremental.base_slot = ULONG_MAX;
     peer->snapshot_info.incremental.slot      = ULONG_MAX;
-    peer->full_latency_nanos        = 0UL;
-    peer->incremental_latency_nanos = 0UL;
+    peer->full_latency_nanos                  = 0UL;
+    peer->incremental_latency_nanos           = 0UL;
     peer_map_ele_insert( ssping->map, peer, ssping->pool );
     deadline_list_ele_push_tail( ssping->unpinged, peer, ssping->pool );
   }
@@ -580,21 +580,24 @@ fd_ssping_advance( fd_ssping_t * ssping,
 }
 
 fd_sspeer_t
-fd_ssping_best( fd_ssping_t const * ssping ) {
-  score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( ssping->score_treap, ssping->pool );
-  if( FD_UNLIKELY( score_treap_fwd_iter_done( iter ) ) ) {
-    return (fd_sspeer_t){ 
-      .addr = {
-        .l = 0UL
-      },
-      .snapshot_info = NULL,
-    };
+fd_ssping_best( fd_ssping_t const * ssping,
+                ulong               highest_slot ) {
+  for( score_treap_fwd_iter_t iter = score_treap_fwd_iter_init( ssping->score_treap, ssping->pool );
+       !score_treap_fwd_iter_done( iter );
+       iter = score_treap_fwd_iter_next( iter, ssping->pool ) ) {
+    fd_ssping_peer_t const * best = score_treap_fwd_iter_ele_const( iter, ssping->pool );
+    if( FD_LIKELY( best->snapshot_info.incremental.slot>=highest_slot ) ) {
+      return (fd_sspeer_t){
+        .addr = best->addr,
+        .snapshot_info = &best->snapshot_info,
+      };
+    }
   }
 
-  fd_ssping_peer_t const * best = score_treap_fwd_iter_ele_const( iter, ssping->pool );
-
   return (fd_sspeer_t){
-    .addr = best->addr,
-    .snapshot_info = &best->snapshot_info,
+    .addr = {
+      .l = 0UL
+    },
+    .snapshot_info = NULL,
   };
 }