feat(repair): implement blk repair structure and API

Author: lidatong

.gitignore

@@ -73,6 +73,7 @@ mainnet.sh
 localnet.sh
 sim.sh
 backtest.sh
+local.toml
 
 # Python
 venv

src/ballet/shred/fd_shred.h

@@ -149,17 +149,25 @@ FD_STATIC_ASSERT( sizeof(fd_bmtree_node_t) == FD_SHRED_MERKLE_ROOT_SZ, update FD
 #define FD_SHRED_DATA_FLAG_DATA_COMPLETE ((uchar)0x40)
 
 /* Maximum number of data shreds in a slot, also maximum number of parity shreds in a slot */
-#define FD_SHRED_MAX_PER_SLOT (1 << 15UL) /* 32,768 shreds */
+#define FD_SHRED_BLK_MAX (1 << 15UL) /* 32,768 shreds */
+#define FD_SHRED_IDX_MAX (FD_SHRED_BLK_MAX - 1)
 
 /* Many static bounds are specified around the assumption that this is a
    protocol limit on the max number of shreds in a slot. If this limit
    changes, all the relevant usages in other areas of the Firedancer
    codebase should be updated before modifying this assertion. */
 
-FD_STATIC_ASSERT( FD_SHRED_MAX_PER_SLOT == 32768, check all usages before changing this limit! );
+FD_STATIC_ASSERT( FD_SHRED_BLK_MAX == 32768, check all usages before changing this limit! );
+
+/* Many static bounds are specified around the assumption that this is a
+   protocol limit on the max number of shreds in a slot. If this limit
+   changes, all the relevant usages in other areas of the Firedancer
+   codebase should be updated before modifying this assertion. */
+
+FD_STATIC_ASSERT( FD_SHRED_BLK_MAX == 32768, check all usages before changing this limit! );
 
 /* 36,536,320 bytes per slot */
-#define FD_SHRED_DATA_PAYLOAD_MAX_PER_SLOT (FD_SHRED_DATA_PAYLOAD_MAX * FD_SHRED_MAX_PER_SLOT)
+#define FD_SHRED_DATA_PAYLOAD_MAX_PER_SLOT (FD_SHRED_DATA_PAYLOAD_MAX * FD_SHRED_BLK_MAX)
 
 /* Offset of the shred variant. Used for parsing. */
 #define FD_SHRED_VARIANT_OFF 0x40

src/disco/fd_disco_base.h

@@ -186,8 +186,8 @@ fd_disco_shred_repair_shred_sig( int   completes,
                                  int   is_code,
                                  uint  shred_idx_or_data_cnt ) {
    ulong slot_ul                  = fd_ulong_min( slot, (ulong)UINT_MAX );
-   ulong shred_idx_or_data_cnt_ul = fd_ulong_min( (ulong)shred_idx_or_data_cnt, (ulong)FD_SHRED_MAX_PER_SLOT );
-   ulong fec_set_idx_ul           = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_MAX_PER_SLOT );
+   ulong shred_idx_or_data_cnt_ul = fd_ulong_min( (ulong)shred_idx_or_data_cnt, (ulong)FD_SHRED_BLK_MAX );
+   ulong fec_set_idx_ul           = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_BLK_MAX );
    ulong completes_ul             = !!completes;
    ulong is_code_ul               = !!is_code;
 
@@ -212,8 +212,8 @@ FD_FN_CONST static inline uint  fd_disco_shred_repair_shred_sig_data_cnt   ( ulo
 FD_FN_CONST static inline ulong
 fd_disco_shred_repair_fec_sig( ulong slot, uint fec_set_idx, uint data_cnt, int is_slot_complete, int is_batch_complete ) {
   ulong slot_ul          = fd_ulong_min( slot, (ulong)UINT_MAX );
-  ulong fec_set_idx_ul   = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_MAX_PER_SLOT );
-  ulong data_cnt_ul      = fd_ulong_min( (ulong)data_cnt, (ulong)FD_SHRED_MAX_PER_SLOT );
+  ulong fec_set_idx_ul   = fd_ulong_min( (ulong)fec_set_idx, (ulong)FD_SHRED_BLK_MAX );
+  ulong data_cnt_ul      = fd_ulong_min( (ulong)data_cnt, (ulong)FD_SHRED_BLK_MAX );
   ulong is_slot_complete_ul = !!is_slot_complete;
   ulong is_batch_complete_ul = !!is_batch_complete;
   return slot_ul << 32 | fec_set_idx_ul << 17 | data_cnt_ul << 2 | is_slot_complete_ul << 1 | is_batch_complete_ul;
@@ -242,7 +242,7 @@ fd_disco_repair_replay_sig( ulong slot, uint data_cnt, ushort parent_off, int sl
    | [32, 63]  | [17, 31]      | [1, 16]         | [0]
   */
   ulong slot_ul          = fd_ulong_min( slot, (ulong)UINT_MAX );
-  ulong data_cnt_ul      = fd_ulong_min( (ulong)data_cnt, (ulong)FD_SHRED_MAX_PER_SLOT );
+  ulong data_cnt_ul      = fd_ulong_min( (ulong)data_cnt, (ulong)FD_SHRED_BLK_MAX );
   ulong parent_off_ul    = (ulong)parent_off;
   ulong slot_complete_ul = !!slot_complete;
   return slot_ul << 32 | data_cnt_ul << 17 | parent_off_ul << 1 | slot_complete_ul;

src/disco/net/sock/fd_sock_tile.c

@@ -188,18 +188,18 @@ privileged_init( fd_topo_t *      topo,
   static char const * udp_port_links[] = {
     "net_quic",   /* legacy_transaction_listen_port */
     "net_quic",   /* quic_transaction_listen_port */
-    "net_shred",  /* shred_listen_port */
+    "net_shred",  /* shred_listen_port (turbine) */
     "net_gossip", /* gossip_listen_port */
-    "net_repair", /* repair_intake_listen_port */
-    "net_repair"  /* repair_intake_listen_port */
+    "net_shred",  /* shred_listen_port (repair) */
+    "net_repair"  /* repair_serve_listen_port */
   };
   static uchar const udp_port_protos[] = {
     DST_PROTO_TPU_UDP,  /* legacy_transaction_listen_port */
     DST_PROTO_TPU_QUIC, /* quic_transaction_listen_port */
-    DST_PROTO_SHRED,    /* shred_listen_port */
+    DST_PROTO_SHRED,    /* shred_listen_port (turbine) */
     DST_PROTO_GOSSIP,   /* gossip_listen_port */
-    DST_PROTO_REPAIR,   /* repair_intake_listen_port */
-    DST_PROTO_REPAIR    /* repair_intake_listen_port */
+    DST_PROTO_SHRED,    /* shred_listen_port (repair) */
+    DST_PROTO_REPAIR    /* repair_serve_listen_port */
   };
   for( uint candidate_idx=0U; candidate_idx<6; candidate_idx++ ) {
     if( !udp_port_candidates[ candidate_idx ] ) continue;

src/disco/net/xdp/fd_xdp_tile.c

@@ -39,6 +39,14 @@
 
 #define FD_XDP_STATS_INTERVAL_NS (11e6) /* 11ms */
 
+/* REPAIR_PING_SZ is the sz of a ping packet for the repair protocol.
+   Because pings are routed to the same port as shreds without any
+   discriminant encoding, we have to use the packet sz to interpret the
+   payload.  Note that any valid shred must be either FD_SHRED_MAX_SZ
+   or FD_SHRED_MIN_SZ ie. will never be FD_REPAIR_PING_SZ.*/
+
+#define REPAIR_PING_SZ (174UL)
+
 /* fd_net_in_ctx_t contains consumer information for an incoming tango
    link.  It is used as part of the TX path. */
 

src/disco/shred/fd_shred_tile.c

@@ -309,11 +309,13 @@ before_frag( fd_shred_ctx_t * ctx,
              ulong            in_idx,
              ulong            seq,
              ulong            sig ) {
-  if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) ctx->poh_in_expect_seq = seq+1UL;
-
-  if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) )     return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_SHRED;
-  else if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) return  (fd_disco_poh_sig_pkt_type( sig )!=POH_PKT_TYPE_MICROBLOCK) &
-                                                                      (fd_disco_poh_sig_pkt_type( sig )!=POH_PKT_TYPE_FEAT_ACT_SLOT);
+  if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_POH ) ) {
+    ctx->poh_in_expect_seq = seq+1UL;
+    return (fd_disco_poh_sig_pkt_type( sig )!=POH_PKT_TYPE_MICROBLOCK) & (fd_disco_poh_sig_pkt_type( sig )!=POH_PKT_TYPE_FEAT_ACT_SLOT);
+  }
+  if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
+    return (fd_disco_netmux_sig_proto( sig )!=DST_PROTO_SHRED) && (fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR);
+  }
   return 0;
 }
 
@@ -724,22 +726,24 @@ after_frag( fd_shred_ctx_t *    ctx,
     }
 
     if( (rv==FD_FEC_RESOLVER_SHRED_OKAY) | (rv==FD_FEC_RESOLVER_SHRED_COMPLETES) ) {
-      /* Relay this shred */
-      ulong max_dest_cnt[1];
-      do {
-        /* If we've validated the shred and it COMPLETES but we can't
-           compute the destination for whatever reason, don't forward
-           the shred, but still send it to the blockstore. */
-        fd_shred_dest_t * sdest = fd_stake_ci_get_sdest_for_slot( ctx->stake_ci, shred->slot );
-        if( FD_UNLIKELY( !sdest ) ) break;
-        fd_shred_dest_idx_t * dests = fd_shred_dest_compute_children( sdest, &shred, 1UL, ctx->scratchpad_dests, 1UL, fanout, fanout, max_dest_cnt );
-        if( FD_UNLIKELY( !dests ) ) break;
-
-        send_shred( ctx, *out_shred, ctx->adtl_dest, ctx->tsorig );
-        for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, *out_shred, fd_shred_dest_idx_to_dest( sdest, dests[ j ]), ctx->tsorig );
-      } while( 0 );
-
-      if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) { /* Optimize for full firedancer. The branch predictor will learn quickly if this is frankendancer instead */
+      if( FD_LIKELY( fd_disco_netmux_sig_proto( sig ) != DST_PROTO_REPAIR ) ) {
+        /* Relay this shred */
+        ulong max_dest_cnt[1];
+        do {
+          /* If we've validated the shred and it COMPLETES but we can't
+            compute the destination for whatever reason, don't forward
+            the shred, but still send it to the blockstore. */
+          fd_shred_dest_t * sdest = fd_stake_ci_get_sdest_for_slot( ctx->stake_ci, shred->slot );
+          if( FD_UNLIKELY( !sdest ) ) break;
+          fd_shred_dest_idx_t * dests = fd_shred_dest_compute_children( sdest, &shred, 1UL, ctx->scratchpad_dests, 1UL, fanout, fanout, max_dest_cnt );
+          if( FD_UNLIKELY( !dests ) ) break;
+
+          send_shred( ctx, *out_shred, ctx->adtl_dest, ctx->tsorig );
+          for( ulong j=0UL; j<*max_dest_cnt; j++ ) send_shred( ctx, *out_shred, fd_shred_dest_idx_to_dest( sdest, dests[ j ]), ctx->tsorig );
+        } while( 0 );
+      }
+
+      if( FD_LIKELY( ctx->repair_out_idx!=ULONG_MAX ) ) { /* Only send to repair in full Firedancer */
 
         /* Construct the sig from the shred. */
 
@@ -796,7 +800,7 @@ after_frag( fd_shred_ctx_t *    ctx,
   ulong sz2 = sizeof(fd_shred34_t) - (34UL - s34[ 2 ].shred_cnt)*FD_SHRED_MAX_SZ;
   ulong sz3 = sizeof(fd_shred34_t) - (34UL - s34[ 3 ].shred_cnt)*FD_SHRED_MAX_SZ;
 
-  if( FD_LIKELY( ctx->blockstore ) ) { /* firedancer topo compiler hint */
+  if( FD_LIKELY( ctx->blockstore ) ) {
 
     /* Insert shreds into the blockstore. Note we do this regardless of
        whether the shreds are for one of our leader slots or not. Even

src/discof/forest/Local.mk

@@ -0,0 +1,4 @@
+ifdef FD_HAS_INT128
+$(call add-objs,fd_forest,fd_discof)
+$(call make-unit-test,test_forest,test_forest,fd_discof fd_disco fd_flamenco fd_tango fd_ballet fd_util)
+endif