LCOV - code coverage report
Current view: top level - discof/repair - fd_repair_tile.c (source / functions) Hit Total Coverage
Test: cov.lcov Lines: 0 614 0.0 %
Date: 2025-12-07 04:58:33 Functions: 0 48 0.0 %

          Line data    Source code
       1             : /* The repair tile is responsible for repairing missing shreds that were
       2             :    not received via Turbine.
       3             : 
       4             :    Generally there are two distinct traffic patterns:
       5             : 
       6             :    1. Firedancer boots up and fires off a large number of repairs to
       7             :       recover all the blocks between the snapshot on which it is booting
       8             :       and the head of the chain.  In this mode, repair tile utilization
       9             :       is very high along with net and sign utilization.
      10             : 
      11             :    2. Firedancer catches up to the head of the chain and enters steady
      12             :       state where most shred traffic is delivered over turbine.  In this
      13             :       state, repairs are only occasionally needed to recover shreds lost
      14             :       due to anomalies like packet loss, transmitter (leader) never sent
      15             :       them or even a malicious leader etc. */
      16             : 
      17             : #define _GNU_SOURCE
      18             : 
      19             : #include "../genesis/fd_genesi_tile.h"
      20             : #include "../../disco/topo/fd_topo.h"
      21             : #include "generated/fd_repair_tile_seccomp.h"
      22             : #include "../../disco/fd_disco.h"
      23             : #include "../../disco/keyguard/fd_keyload.h"
      24             : #include "../../disco/keyguard/fd_keyguard.h"
      25             : #include "../../disco/net/fd_net_tile.h"
      26             : #include "../../flamenco/gossip/fd_gossip_types.h"
      27             : #include "../tower/fd_tower_tile.h"
      28             : #include "../../discof/restore/utils/fd_ssmsg.h"
      29             : #include "../../util/net/fd_net_headers.h"
      30             : #include "../../tango/fd_tango_base.h"
      31             : 
      32             : #include "../forest/fd_forest.h"
      33             : #include "fd_repair_metrics.h"
      34             : #include "fd_inflight.h"
      35             : #include "fd_repair.h"
      36             : #include "fd_policy.h"
      37             : 
      38             : #define LOGGING       1
      39             : #define DEBUG_LOGGING 0
      40             : 
      41             : #define IN_KIND_CONTACT (0)
      42           0 : #define IN_KIND_NET     (1)
      43           0 : #define IN_KIND_TOWER   (2)
      44           0 : #define IN_KIND_SHRED   (3)
      45           0 : #define IN_KIND_SIGN    (4)
      46           0 : #define IN_KIND_SNAP    (5)
      47           0 : #define IN_KIND_STAKE   (6)
      48           0 : #define IN_KIND_GOSSIP  (7)
      49           0 : #define IN_KIND_GENESIS (8)
      50             : 
      51             : #define MAX_IN_LINKS    (32)
      52             : 
      53             : #define MAX_REPAIR_PEERS   40200UL
      54             : #define MAX_BUFFER_SIZE    ( MAX_REPAIR_PEERS * sizeof( fd_shred_dest_wire_t ) )
      55             : #define MAX_SHRED_TILE_CNT ( 16UL )
      56             : #define MAX_SIGN_TILE_CNT  ( 16UL )
      57             : 
      58             : /* Maximum size of a network packet */
      59           0 : #define FD_REPAIR_MAX_PACKET_SIZE 1232
      60             : /* Max number of validators that can be actively queried */
      61           0 : #define FD_ACTIVE_KEY_MAX (FD_CONTACT_INFO_TABLE_SIZE)
      62             : /* Max number of pending shred requests */
      63           0 : #define FD_NEEDED_KEY_MAX (1<<20)
      64             : 
      65             : /* static map from request type to metric array index */
      66             : static uint metric_index[FD_REPAIR_KIND_ORPHAN + 1] = {
      67             :   [FD_REPAIR_KIND_SHRED]         = FD_METRICS_ENUM_REPAIR_SENT_REQUEST_TYPES_V_NEEDED_WINDOW_IDX,
      68             :   [FD_REPAIR_KIND_HIGHEST_SHRED] = FD_METRICS_ENUM_REPAIR_SENT_REQUEST_TYPES_V_NEEDED_HIGHEST_WINDOW_IDX,
      69             :   [FD_REPAIR_KIND_ORPHAN]        = FD_METRICS_ENUM_REPAIR_SENT_REQUEST_TYPES_V_NEEDED_ORPHAN_IDX,
      70             : };
      71             : 
      72             : typedef union {
      73             :   struct {
      74             :     fd_wksp_t * mem;
      75             :     ulong       chunk0;
      76             :     ulong       wmark;
      77             :     ulong       mtu;
      78             :   };
      79             :   fd_net_rx_bounds_t net_rx;
      80             : } in_ctx_t;
      81             : 
      82             : struct out_ctx {
      83             :   ulong         idx;
      84             :   fd_wksp_t *   mem;
      85             :   ulong         chunk0;
      86             :   ulong         wmark;
      87             :   ulong         chunk;
      88             : 
      89             :   /* Repair tile directly tracks credit outside of stem for these
      90             :      asynchronous sign links.  In particular, credits tracks the RETURN
      91             :      sign_repair link.  This is because repair_sign is reliable, and
      92             :      sign_repair is unreliable.  If both links were reliable, and the
      93             :      links filled completely, stem would get into a deadlock. Neither
      94             :      repair or sign would have credits, which would prevent frags from
      95             :      getting polled in repair or sign, which would prevent any credits
      96             :      from getting returned back to the tiles.  credits / max_credits are
      97             :      used by the repair_sign link.  In particular, credits manages the
      98             :      RETURN sign_repair link.
      99             : 
     100             :      Thus the sign_repair link must be unreliable. This is mostly ok,
     101             :      because repair_sign is still reliable, so in theory repair_tile
     102             :      would never publish enough frags such that sign_repair would get
     103             :      overrun.
     104             : 
     105             :      However, there is a fairly common case that breaks this.  Consider
     106             :      the scenario
     107             : 
     108             :              repair_sign (depth 128)        sign_repair (depth 128)
     109             :      repair  ---------------------->  sign ------------------------> repair
     110             :              [rest free, r130, r129]       [r128, r127, ... , r1] (full)
     111             : 
     112             :      This would happen because repair is publishing too many requests
     113             :      too fast(common in catchup), and not polling enough frags from
     114             :      sign. Nothing is stopping repair from publishing more requests,
     115             :      because sign is functioning fast enough to handle the requests.
     116             :      However, nothing is stopping sign from polling the next request and
     117             :      signing it, and PUBLISHING it on the sign_repair link that is
     118             :      already full, because the sign_repair link is unreliable.
     119             : 
     120             :      In fact the only time we could stop repair from publishing more
     121             :      requests is if repair_sign was full, and repair would get
     122             :      backpressured, but sign would still be able to poll requests and
     123             :      overrun the sign_repair link.
     124             : 
     125             :      This is why we need to manually track credits for the sign_repair
     126             :      link. We must ensure that there are never more than 128 items in
     127             :      the ENTIRE repair_sign -> sign tile -> sign_repair work queue, else
     128             :      there is always a possibility of an overrun in the sign_repair
     129             :      link.
     130             : 
     131             :      To lose a frag to overrun isn't necessarily critical, but in
     132             :      general the repair tile relies on the fact that a signing task
     133             :      published to sign tile will always come back.  If we lose a frag to
     134             :      overrun, then there will be an entry in the pending signs structure
     135             :      that is never removed, and theoretically the map could fill up.
     136             :      Conceptually, with a reliable sign->repair->sign structure, there
     137             :      should be no eviction needed in this pending signs structure. */
     138             : 
     139             :   ulong in_idx;      /* index of the incoming link */
     140             :   ulong credits;     /* available credits for link */
     141             :   ulong max_credits; /* maximum credits (depth) */
     142             : };
     143             : typedef struct out_ctx out_ctx_t;
     144             : 
     145             : struct fd_fec_sig {
     146             :   ulong            key; /* map key. 32 msb = slot, 32 lsb = fec_set_idx */
     147             :   fd_ed25519_sig_t sig; /* Ed25519 sig identifier of the FEC. */
     148             : };
     149             : typedef struct fd_fec_sig fd_fec_sig_t;
     150             : 
     151             : #define MAP_NAME    fd_fec_sig
     152           0 : #define MAP_T       fd_fec_sig_t
     153             : #define MAP_MEMOIZE 0
     154             : #include "../../util/tmpl/fd_map_dynamic.c"
     155             : 
     156             : /* Data needed to sign and send a pong that is not contained in the
     157             :    pong msg itself. */
     158             : 
     159             : struct pong_data {
     160             :   fd_ip4_port_t  peer_addr;
     161             :   fd_hash_t      hash;
     162             :   uint           daddr;
     163             : };
     164             : typedef struct pong_data pong_data_t;
     165             : 
     166             : struct sign_req {
     167             :   ulong       key;        /* map key, ctx->pending_key_next */
     168             :   ulong       buflen;
     169             :   union {
     170             :     uchar           buf[sizeof(fd_repair_msg_t)];
     171             :     fd_repair_msg_t msg;
     172             :   };
     173             :   pong_data_t  pong_data; /* populated only for pong msgs */
     174             : };
     175             : typedef struct sign_req sign_req_t;
     176             : 
     177             : #define MAP_NAME         fd_signs_map
     178           0 : #define MAP_KEY          key
     179           0 : #define MAP_KEY_NULL     ULONG_MAX
     180           0 : #define MAP_KEY_INVAL(k) (k==ULONG_MAX)
     181           0 : #define MAP_T            sign_req_t
     182             : #define MAP_MEMOIZE      0
     183             : #include "../../util/tmpl/fd_map_dynamic.c"
     184             : 
     185             : /* Because the sign tiles could be all busy when a contact info arrives,
     186             :    we need to save ping messages to be signed in a queue and dispatched
     187             :    in after_credit when there are sign tiles available.  The size of the
     188             :    queue was determined by the following: we can limit the size of this
     189             :    queue to be the maximum number of active keys - which is equal to the
     190             :    number of warm up requests we might queue.  The queue will also hold
     191             :    pongs, but in order for the ping to arrive the warm up request must
     192             :    have left the queue.  It is possible that we start up and get
     193             :    FD_ACTIVE_KEY_MAX peers gossiped to us, and as we are queueing up
     194             :    their pings they all drop and another FD_ACTIVE_KEY_MAX new peers
     195             :    gossip to us, causing us to fill up the queue.  Idk overall this
     196             :    scenario is highly unlikely and it's not the end of the world if we
     197             :    drop a warmup req or ping  to a peer because the first req to them
     198             :    will retrigger it anyway.
     199             : 
     200             :    Typical flow is that a pong will get added to the sign_queue during
     201             :    an after_frag call.  Then on the following after_credit will get
     202             :    popped from the sign_queue and added to sign_map, and then dispatched
     203             :    to the sign tile. */
     204             : 
     205             : struct sign_pending {
     206             :   fd_repair_msg_t msg;
     207             :   pong_data_t     pong_data; /* populated only for pong msgs */
     208             : };
     209             : typedef struct sign_pending sign_pending_t;
     210             : 
     211             : #define QUEUE_NAME       fd_signs_queue
     212           0 : #define QUEUE_T          sign_pending_t
     213           0 : #define QUEUE_MAX        2*FD_ACTIVE_KEY_MAX
     214             : #include "../../util/tmpl/fd_queue.c"
     215             : 
     216             : struct ctx {
     217             :   long tsdebug; /* timestamp for debug printing */
     218             : 
     219             :   ulong repair_seed;
     220             : 
     221             :   fd_ip4_port_t repair_intake_addr;
     222             :   fd_ip4_port_t repair_serve_addr;
     223             : 
     224             :   fd_forest_t    * forest;
     225             :   fd_fec_sig_t   * fec_sigs;
     226             :   fd_policy_t    * policy;
     227             :   fd_inflights_t * inflights;
     228             :   fd_repair_t    * protocol;
     229             : 
     230             :   fd_pubkey_t identity_public_key;
     231             : 
     232             :   fd_wksp_t * wksp;
     233             : 
     234             :   fd_stem_context_t * stem;
     235             : 
     236             :   uchar    in_kind[ MAX_IN_LINKS ];
     237             :   in_ctx_t in_links[ MAX_IN_LINKS ];
     238             : 
     239             :   int skip_frag;
     240             : 
     241             :   uint        net_out_idx;
     242             :   fd_wksp_t * net_out_mem;
     243             :   ulong       net_out_chunk0;
     244             :   ulong       net_out_wmark;
     245             :   ulong       net_out_chunk;
     246             : 
     247             :   ulong snap_out_chunk;
     248             : 
     249             :   uint      shred_tile_cnt;
     250             :   out_ctx_t shred_out_ctx[ MAX_SHRED_TILE_CNT ];
     251             : 
     252             :   /* repair_sign links (to sign tiles 1+) - for round-robin distribution */
     253             : 
     254             :   ulong     repair_sign_cnt;
     255             :   out_ctx_t repair_sign_out_ctx[ MAX_SIGN_TILE_CNT ];
     256             : 
     257             :   ulong     sign_rrobin_idx;
     258             : 
     259             :   /* Pending sign requests for async operations */
     260             : 
     261             :   uint             pending_key_next;
     262             :   sign_req_t     * signs_map;  /* contains any request currently in the repair->sign or sign->repair dcache */
     263             :   sign_pending_t * sign_queue; /* contains any request waiting to be dispatched to repair->sign */
     264             : 
     265             :   ushort net_id;
     266             :   uchar buffer[ MAX_BUFFER_SIZE ]; /* includes Ethernet, IP, UDP headers */
     267             :   fd_ip4_udp_hdrs_t intake_hdr[1];
     268             :   fd_ip4_udp_hdrs_t serve_hdr [1];
     269             : 
     270             :   ulong manifest_slot;
     271             :   struct {
     272             :     ulong send_pkt_cnt;
     273             :     ulong sent_pkt_types[FD_METRICS_ENUM_REPAIR_SENT_REQUEST_TYPES_CNT];
     274             :     ulong repaired_slots;
     275             :     ulong current_slot;
     276             :     ulong sign_tile_unavail;
     277             :     ulong rerequest;
     278             :     ulong malformed_ping;
     279             :     fd_histf_t slot_compl_time[ 1 ];
     280             :     fd_histf_t response_latency[ 1 ];
     281             :   } metrics[ 1 ];
     282             : 
     283             :   /* Slot-level metrics */
     284             : 
     285             :   fd_repair_metrics_t * slot_metrics;
     286             :   ulong turbine_slot0;  // catchup considered complete after this slot
     287             :   struct {
     288             :     int   enabled;
     289             :     ulong end_slot;
     290             :     int   complete;
     291             :   } profiler;
     292             : };
     293             : typedef struct ctx ctx_t;
     294             : 
     295             : FD_FN_CONST static inline ulong
     296           0 : scratch_align( void ) {
     297           0 :   return 128UL;
     298           0 : }
     299             : 
     300             : FD_FN_PURE static inline ulong
     301           0 : loose_footprint( fd_topo_tile_t const * tile FD_PARAM_UNUSED ) {
     302           0 :   return 1UL * FD_SHMEM_GIGANTIC_PAGE_SZ;
     303           0 : }
     304             : 
     305             : FD_FN_PURE static inline ulong
     306           0 : scratch_footprint( fd_topo_tile_t const * tile ) {
     307           0 :   ulong total_sign_depth = tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt;
     308           0 :   int   lg_sign_depth    = fd_ulong_find_msb( fd_ulong_pow2_up(total_sign_depth) ) + 1;
     309             : 
     310           0 :   ulong l = FD_LAYOUT_INIT;
     311           0 :   l = FD_LAYOUT_APPEND( l, alignof(ctx_t),            sizeof(ctx_t)                                                    );
     312           0 :   l = FD_LAYOUT_APPEND( l, fd_repair_align(),         fd_repair_footprint     ()                                       );
     313           0 :   l = FD_LAYOUT_APPEND( l, fd_forest_align(),         fd_forest_footprint     ( tile->repair.slot_max )                );
     314           0 :   l = FD_LAYOUT_APPEND( l, fd_policy_align(),         fd_policy_footprint     ( FD_NEEDED_KEY_MAX, FD_ACTIVE_KEY_MAX ) );
     315           0 :   l = FD_LAYOUT_APPEND( l, fd_inflights_align(),      fd_inflights_footprint  ()                                       );
     316           0 :   l = FD_LAYOUT_APPEND( l, fd_fec_sig_align(),        fd_fec_sig_footprint    ( 20 )                                   );
     317           0 :   l = FD_LAYOUT_APPEND( l, fd_signs_map_align(),      fd_signs_map_footprint  ( lg_sign_depth )                        );
     318           0 :   l = FD_LAYOUT_APPEND( l, fd_signs_queue_align(),    fd_signs_queue_footprint()                                       );
     319           0 :   l = FD_LAYOUT_APPEND( l, fd_repair_metrics_align(), fd_repair_metrics_footprint()                                    );
     320           0 :   return FD_LAYOUT_FINI( l, scratch_align() );
     321           0 : }
     322             : 
     323             : /* Below functions manage the current pending sign requests. */
     324             : 
     325             : sign_req_t *
     326             : sign_map_insert( ctx_t *                 ctx,
     327             :                  fd_repair_msg_t const * msg,
     328           0 :                  pong_data_t const     * opt_pong_data ) {
     329             : 
     330             :   /* Check if there is any space for a new pending sign request. Should never fail as long as credit management is working. */
     331           0 :   if( FD_UNLIKELY( fd_signs_map_key_cnt( ctx->signs_map )==fd_signs_map_key_max( ctx->signs_map ) ) ) return NULL;
     332             : 
     333           0 :   sign_req_t * pending = fd_signs_map_insert( ctx->signs_map, ctx->pending_key_next++ );
     334           0 :   if( FD_UNLIKELY( !pending ) ) return NULL; // Not possible, unless the same nonce is used twice.
     335           0 :   pending->msg    = *msg;
     336           0 :   pending->buflen = fd_repair_sz( msg );
     337           0 :   if( FD_UNLIKELY( opt_pong_data ) ) pending->pong_data = *opt_pong_data;
     338           0 :   return pending;
     339           0 : }
     340             : 
     341             : int
     342             : sign_map_remove( ctx_t * ctx,
     343           0 :                  ulong   key  ) {
     344           0 :   sign_req_t * pending = fd_signs_map_query( ctx->signs_map, key, NULL );
     345           0 :   if( FD_UNLIKELY( !pending ) ) return -1;
     346           0 :   fd_signs_map_remove( ctx->signs_map, pending );
     347           0 :   return 0;
     348           0 : }
     349             : 
     350             : static void
     351             : send_packet( ctx_t             * ctx,
     352             :              fd_stem_context_t * stem,
     353             :              int                 is_intake,
     354             :              uint                dst_ip_addr,
     355             :              ushort              dst_port,
     356             :              uint                src_ip_addr,
     357             :              uchar const *       payload,
     358             :              ulong               payload_sz,
     359           0 :              ulong               tsorig ) {
     360           0 :   ctx->metrics->send_pkt_cnt++;
     361           0 :   uchar * packet = fd_chunk_to_laddr( ctx->net_out_mem, ctx->net_out_chunk );
     362           0 :   fd_ip4_udp_hdrs_t * hdr = (fd_ip4_udp_hdrs_t *)packet;
     363           0 :   *hdr = *(is_intake ? ctx->intake_hdr : ctx->serve_hdr);
     364             : 
     365           0 :   fd_ip4_hdr_t * ip4 = hdr->ip4;
     366           0 :   ip4->saddr       = src_ip_addr;
     367           0 :   ip4->daddr       = dst_ip_addr;
     368           0 :   ip4->net_id      = fd_ushort_bswap( ctx->net_id++ );
     369           0 :   ip4->check       = 0U;
     370           0 :   ip4->net_tot_len = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_ip4_hdr_t)+sizeof(fd_udp_hdr_t)) );
     371           0 :   ip4->check       = fd_ip4_hdr_check_fast( ip4 );
     372             : 
     373           0 :   fd_udp_hdr_t * udp = hdr->udp;
     374           0 :   udp->net_dport = dst_port;
     375           0 :   udp->net_len   = fd_ushort_bswap( (ushort)(payload_sz + sizeof(fd_udp_hdr_t)) );
     376           0 :   fd_memcpy( packet+sizeof(fd_ip4_udp_hdrs_t), payload, payload_sz );
     377           0 :   hdr->udp->check = 0U;
     378             : 
     379           0 :   ulong tspub     = fd_frag_meta_ts_comp( fd_tickcount() );
     380           0 :   ulong sig       = fd_disco_netmux_sig( dst_ip_addr, dst_port, dst_ip_addr, DST_PROTO_OUTGOING, sizeof(fd_ip4_udp_hdrs_t) );
     381           0 :   ulong packet_sz = payload_sz + sizeof(fd_ip4_udp_hdrs_t);
     382           0 :   ulong chunk     = ctx->net_out_chunk;
     383           0 :   fd_stem_publish( stem, ctx->net_out_idx, sig, chunk, packet_sz, 0UL, tsorig, tspub );
     384           0 :   ctx->net_out_chunk = fd_dcache_compact_next( chunk, packet_sz, ctx->net_out_chunk0, ctx->net_out_wmark );
     385           0 : }
     386             : 
     387             : /* Returns a sign_out context with max available credits.
     388             :    If no sign_out context has available credits, returns NULL. */
     389             : static out_ctx_t *
     390           0 : sign_avail_credits( ctx_t * ctx ) {
     391           0 :   out_ctx_t * sign_out = NULL;
     392           0 :   ulong max_credits = 0;
     393           0 :   for( uint i = 0; i < ctx->repair_sign_cnt; i++ ) {
     394           0 :     if( ctx->repair_sign_out_ctx[i].credits > max_credits ) {
     395           0 :       max_credits =  ctx->repair_sign_out_ctx[i].credits;
     396           0 :       sign_out    = &ctx->repair_sign_out_ctx[i];
     397           0 :     }
     398           0 :   }
     399           0 :   return sign_out;
     400           0 : }
     401             : 
     402             : /* Prepares the signing preimage and publishes a signing request that
     403             :    will be signed asynchronously by the sign tile.  The signed data will
     404             :    be returned via dcache as a frag. */
     405             : static void
     406             : fd_repair_send_sign_request( ctx_t                 * ctx,
     407             :                              out_ctx_t             * sign_out,
     408             :                              fd_repair_msg_t const * msg,
     409           0 :                              pong_data_t     const * opt_pong_data ){
     410             :   /* New sign request */
     411           0 :   sign_req_t * pending = sign_map_insert( ctx, msg, opt_pong_data );
     412           0 :   if( FD_UNLIKELY( !pending ) ) return;
     413             : 
     414           0 :   ulong   sig         = 0;
     415           0 :   ulong   preimage_sz = 0;
     416           0 :   uchar * dst         = fd_chunk_to_laddr( sign_out->mem, sign_out->chunk );
     417             : 
     418           0 :   if( FD_UNLIKELY( msg->kind == FD_REPAIR_KIND_PONG ) ) {
     419           0 :     uchar pre_image[FD_REPAIR_PONG_PREIMAGE_SZ];
     420           0 :     preimage_pong( &opt_pong_data->hash, pre_image, sizeof(pre_image) );
     421           0 :     preimage_sz = FD_REPAIR_PONG_PREIMAGE_SZ;
     422           0 :     fd_memcpy( dst, pre_image, preimage_sz );
     423           0 :     sig = ((ulong)pending->key << 32) | (uint)FD_KEYGUARD_SIGN_TYPE_SHA256_ED25519;
     424           0 :   } else {
     425             :     /* Sign and prepare the message directly into the pending buffer */
     426           0 :     uchar * preimage = preimage_req( &pending->msg, &preimage_sz );
     427           0 :     fd_memcpy( dst, preimage, preimage_sz );
     428           0 :     sig = ((ulong)pending->key << 32) | (uint)FD_KEYGUARD_SIGN_TYPE_ED25519;
     429           0 :   }
     430             : 
     431           0 :   fd_stem_publish( ctx->stem, sign_out->idx, sig, sign_out->chunk, preimage_sz, 0UL, 0UL, 0UL );
     432           0 :   sign_out->chunk = fd_dcache_compact_next( sign_out->chunk, preimage_sz, sign_out->chunk0, sign_out->wmark );
     433             : 
     434           0 :   ctx->metrics->sent_pkt_types[metric_index[msg->kind]]++;
     435           0 :   sign_out->credits--;
     436           0 : }
     437             : 
     438             : static inline int
     439             : before_frag( ctx_t * ctx,
     440             :              ulong   in_idx,
     441             :              ulong   seq FD_PARAM_UNUSED,
     442           0 :              ulong   sig ) {
     443           0 :   uint in_kind = ctx->in_kind[ in_idx ];
     444           0 :   if( FD_LIKELY  ( in_kind==IN_KIND_NET   ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
     445           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) return fd_int_if( fd_forest_root_slot( ctx->forest )==ULONG_MAX, -1, 0 ); /* not ready to read frag */
     446           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_GOSSIP ) ) {
     447           0 :     return sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO &&
     448           0 :            sig!=FD_GOSSIP_UPDATE_TAG_CONTACT_INFO_REMOVE;
     449           0 :   }
     450           0 :   return 0;
     451           0 : }
     452             : 
     453             : static void
     454             : during_frag( ctx_t * ctx,
     455             :              ulong                  in_idx,
     456             :              ulong                  seq FD_PARAM_UNUSED,
     457             :              ulong                  sig FD_PARAM_UNUSED,
     458             :              ulong                  chunk,
     459             :              ulong                  sz,
     460           0 :              ulong                  ctl ) {
     461           0 :   ctx->skip_frag = 0;
     462             : 
     463           0 :   uint             in_kind =  ctx->in_kind[ in_idx ];
     464           0 :   in_ctx_t const * in_ctx  = &ctx->in_links[ in_idx ];
     465             : 
     466           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_TOWER ) ) {
     467           0 :     if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) ) {
     468           0 :       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
     469           0 :     }
     470           0 :     uchar const * dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
     471           0 :     fd_memcpy( ctx->buffer, dcache_entry, sz );
     472           0 :     return;
     473           0 :   }
     474             : 
     475           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_GENESIS ) ) {
     476           0 :     return;
     477           0 :   }
     478           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_NET ) ) {
     479           0 :     uchar const * dcache_entry = fd_net_rx_translate_frag( &in_ctx->net_rx, chunk, ctl, sz );
     480           0 :     fd_memcpy( ctx->buffer, dcache_entry, sz );
     481           0 :     return;
     482           0 :   }
     483             : 
     484           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_GOSSIP ) ) {
     485           0 :     if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) ) {
     486           0 :       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
     487           0 :     }
     488           0 :     uchar const * dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
     489           0 :     fd_memcpy( ctx->buffer, dcache_entry, sz );
     490           0 :     return;
     491           0 :   }
     492             : 
     493           0 :   if( FD_LIKELY  ( in_kind==IN_KIND_SHRED  ) ) {
     494           0 :     if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) ) {
     495           0 :       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
     496           0 :     }
     497           0 :     uchar const * dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
     498           0 :     if( FD_LIKELY( sz > 0 ) ) fd_memcpy( ctx->buffer, dcache_entry, sz );
     499           0 :     return;
     500           0 :   }
     501             : 
     502           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_STAKE ) ) {
     503           0 :     return;
     504           0 :   }
     505             : 
     506           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SNAP ) ) {
     507           0 :     if( FD_UNLIKELY( fd_ssmsg_sig_message( sig )!=FD_SSMSG_DONE ) ) ctx->snap_out_chunk = chunk;
     508           0 :     return;
     509           0 :   }
     510             : 
     511           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SIGN ) ) {
     512           0 :     if( FD_UNLIKELY( chunk<in_ctx->chunk0 || chunk>in_ctx->wmark || sz>in_ctx->mtu ) ) {
     513           0 :       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, in_ctx->chunk0, in_ctx->wmark ));
     514           0 :     }
     515           0 :     uchar const * dcache_entry = fd_chunk_to_laddr_const( in_ctx->mem, chunk );
     516           0 :     fd_memcpy( ctx->buffer, dcache_entry, sz );
     517           0 :     return;
     518           0 :   }
     519             : 
     520           0 :   FD_LOG_ERR(( "Frag from unknown link (kind=%u in_idx=%lu)", in_kind, in_idx ));
     521           0 : }
     522             : 
     523             : static inline void
     524             : after_snap( ctx_t * ctx,
     525             :                  ulong                  sig,
     526           0 :                  uchar const          * chunk ) {
     527           0 :   if( FD_UNLIKELY( fd_ssmsg_sig_message( sig )!=FD_SSMSG_DONE ) ) return;
     528           0 :   fd_snapshot_manifest_t * manifest = (fd_snapshot_manifest_t *)chunk;
     529             : 
     530           0 :   fd_forest_init( ctx->forest, manifest->slot );
     531           0 :   FD_TEST( fd_forest_root_slot( ctx->forest )!=ULONG_MAX );
     532             : 
     533           0 : }
     534             : 
     535             : static inline void
     536           0 : after_contact( ctx_t * ctx, fd_gossip_update_message_t const * msg ) {
     537           0 :   fd_contact_info_t const * contact_info = msg->contact_info.contact_info;
     538           0 :   fd_ip4_port_t repair_peer = contact_info->sockets[ FD_CONTACT_INFO_SOCKET_SERVE_REPAIR ];
     539           0 :   if( FD_UNLIKELY( !repair_peer.addr || !repair_peer.port ) ) return;
     540           0 :   fd_policy_peer_t const * peer = fd_policy_peer_insert( ctx->policy, &contact_info->pubkey, &repair_peer );
     541           0 :   if( peer ) {
     542             :     /* The repair process uses a Ping-Pong protocol that incurs one
     543             :        round-trip time (RTT) for the initial repair request.  To
     544             :        optimize this, we proactively send a placeholder repair request
     545             :        as soon as we receive a peer's contact information for the first
     546             :        time, effectively prepaying the RTT cost. */
     547           0 :     fd_repair_msg_t * init = fd_repair_shred( ctx->protocol, &contact_info->pubkey, (ulong)fd_log_wallclock()/1000000L, 0, 0, 0 );
     548           0 :     fd_signs_queue_push( ctx->sign_queue, (sign_pending_t){ .msg = *init } );
     549           0 :   }
     550           0 : }
     551             : 
     552             : static inline void
     553             : after_sign( ctx_t             * ctx,
     554             :             ulong               in_idx,
     555             :             ulong               sig,
     556           0 :             fd_stem_context_t * stem ) {
     557           0 :   ulong pending_key = sig >> 32;
     558             :   /* Look up the pending request. Since the repair_sign links are
     559             :      reliable, the incoming sign_repair fragments represent a complete
     560             :      set of the previously sent outgoing messages. However, with
     561             :      multiple sign tiles, the responses may arrive interleaved. */
     562             : 
     563             :   /* Find which sign tile sent this response and increment its credits */
     564           0 :   for( uint i = 0; i < ctx->repair_sign_cnt; i++ ) {
     565           0 :     if( ctx->repair_sign_out_ctx[i].in_idx == in_idx ) {
     566           0 :       if( ctx->repair_sign_out_ctx[i].credits < ctx->repair_sign_out_ctx[i].max_credits ) {
     567           0 :         ctx->repair_sign_out_ctx[i].credits++;
     568           0 :       }
     569           0 :       break;
     570           0 :     }
     571           0 :   }
     572             : 
     573           0 :   sign_req_t * pending_ = fd_signs_map_query( ctx->signs_map, pending_key, NULL );
     574           0 :   if( FD_UNLIKELY( !pending_ ) ) FD_LOG_CRIT(( "No pending request found for key %lu", pending_key ));
     575             : 
     576           0 :   sign_req_t   pending[1] = { *pending_ }; /* Make a copy of the pending request so we can sign_map_remove immediately. */
     577           0 :   sign_map_remove( ctx, pending_key );
     578             : 
     579             :   /* Thhis is a pong message */
     580           0 :   if( FD_UNLIKELY( pending->msg.kind == FD_REPAIR_KIND_PONG ) ) {
     581           0 :     fd_memcpy( pending->msg.pong.sig, ctx->buffer, 64UL );
     582           0 :     send_packet( ctx, stem, 1, pending->pong_data.peer_addr.addr, pending->pong_data.peer_addr.port, pending->pong_data.daddr, pending->buf, fd_repair_sz( &pending->msg ), fd_frag_meta_ts_comp( fd_tickcount() ) );
     583           0 :     return;
     584           0 :   }
     585             : 
     586             :   /* Inject the signature into the pending request */
     587           0 :   fd_memcpy( pending->buf + 4, ctx->buffer, 64UL );
     588           0 :   uint  src_ip4 = 0U;
     589             : 
     590             :   /* This is a warmup message */
     591           0 :   if( FD_UNLIKELY( pending->msg.kind == FD_REPAIR_KIND_SHRED && pending->msg.shred.slot == 0 ) ) {
     592           0 :     fd_policy_peer_t * active = fd_policy_peer_query( ctx->policy, &pending->msg.shred.to );
     593           0 :     if( FD_UNLIKELY( active ) ) send_packet( ctx, stem, 1, active->ip4, active->port, src_ip4, pending->buf, pending->buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
     594           0 :     else { /* This is a warmup request for a peer that is no longer active.  There's no reason to pick another peer for a warmup rq, so just drop it. */ }
     595           0 :     return;
     596           0 :   }
     597             : 
     598             :   /* This is a regular repair shred request
     599             : 
     600             :      TODO: anyways to make this less complicated? Essentially we need to
     601             :      ensure we always send out any shred requests we have, because policy_next
     602             :      has no way to revisit a shred.  But the fact that peers can drop out
     603             :      of the active peer list makes this complicated.
     604             : 
     605             :      1. If the peer is still there (common), it's fine.
     606             :      2. If the peer is not there, we can select another peer and send the request.
     607             :      3. If the peer is not there, and we have no other peers, we can add
     608             :         this request to the inflights table, pretend we've sent it and
     609             :         let the inflight timeout request it down the line.
     610             :   */
     611           0 :   fd_policy_peer_t * active         = fd_policy_peer_query( ctx->policy, &pending->msg.shred.to );
     612           0 :   int                is_regular_req = pending->msg.kind == FD_REPAIR_KIND_SHRED && pending->msg.shred.nonce > 0; // not a highest/orphan request
     613             : 
     614           0 :   if( FD_UNLIKELY( !active ) ) {
     615           0 :     fd_pubkey_t const * new_peer = fd_policy_peer_select( ctx->policy );
     616           0 :     if( FD_LIKELY( new_peer ) ) {
     617             :       /* We have a new peer, so we can send the request */
     618           0 :       pending->msg.shred.to = *new_peer;
     619           0 :       fd_signs_queue_push( ctx->sign_queue, (sign_pending_t){ .msg = pending->msg } );
     620           0 :     }
     621             : 
     622           0 :     if( FD_UNLIKELY( !new_peer && is_regular_req ) ) {
     623             :       /* This is real devastation - we clearly had a peer at the time of
     624             :          making this request, but for some reason we now have ZERO
     625             :          peers. The only thing we can do is to add this artificially to
     626             :          the inflights table, pretend we've sent it and let the inflight
     627             :          timeout request it down the line. */
     628           0 :       fd_inflights_request_insert( ctx->inflights, pending->msg.shred.nonce, &pending->msg.shred.to, pending->msg.shred.slot, pending->msg.shred.shred_idx );
     629           0 :     }
     630           0 :     return;
     631           0 :   }
     632             :   /* Happy path - all is well, our peer didn't drop out from beneath us. */
     633           0 :   if( FD_LIKELY( is_regular_req ) ) {
     634           0 :     fd_inflights_request_insert( ctx->inflights, pending->msg.shred.nonce, &pending->msg.shred.to, pending->msg.shred.slot, pending->msg.shred.shred_idx );
     635           0 :     fd_policy_peer_request_update( ctx->policy, &pending->msg.shred.to );
     636           0 :   }
     637           0 :   send_packet( ctx, stem, 1, active->ip4, active->port, src_ip4, pending->buf, pending->buflen, fd_frag_meta_ts_comp( fd_tickcount() ) );
     638           0 : }
     639             : 
     640             : static inline void
     641             : after_shred( ctx_t      * ctx,
     642             :              ulong        sig,
     643             :              fd_shred_t * shred,
     644           0 :              ulong        nonce ) {
     645             :   /* Insert the shred sig (shared by all shred members in the FEC set)
     646             :       into the map. */
     647           0 :   int is_code = fd_shred_is_code( fd_shred_type( shred->variant ) );
     648           0 :   int src = fd_disco_shred_out_shred_sig_is_turbine( sig ) ? SHRED_SRC_TURBINE : SHRED_SRC_REPAIR;
     649           0 :   if( FD_LIKELY( !is_code ) ) {
     650           0 :     long rtt = 0;
     651           0 :     fd_pubkey_t peer;
     652           0 :     if( FD_UNLIKELY( src == SHRED_SRC_REPAIR && ( rtt = fd_inflights_request_remove( ctx->inflights, nonce, &peer ) ) > 0 ) ) {
     653           0 :       fd_policy_peer_response_update( ctx->policy, &peer, rtt );
     654           0 :       fd_histf_sample( ctx->metrics->response_latency, (ulong)rtt );
     655           0 :     }
     656             : 
     657           0 :     int slot_complete = !!(shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE);
     658           0 :     int ref_tick      = shred->data.flags & FD_SHRED_DATA_REF_TICK_MASK;
     659           0 :     fd_forest_blk_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off );
     660           0 :     if( FD_UNLIKELY( ctx->profiler.enabled && shred->slot == ctx->profiler.end_slot ) ) fd_forest_blk_parent_update( ctx->forest, shred->slot, shred->slot - shred->data.parent_off );
     661           0 :     fd_forest_data_shred_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, shred->idx, shred->fec_set_idx, slot_complete, ref_tick, src );
     662           0 :   } else {
     663           0 :     fd_forest_code_shred_insert( ctx->forest, shred->slot, shred->idx );
     664           0 :   }
     665           0 : }
     666             : 
     667             : static inline void
     668             : after_fec( ctx_t      * ctx,
     669           0 :            fd_shred_t * shred ) {
     670             : 
     671             :   /* When this is a FEC completes msg, it is implied that all the
     672             :      other shreds in the FEC set can also be inserted.  Shred inserts
     673             :      into the forest are idempotent so it is fine to insert the same
     674             :      shred multiple times. */
     675             : 
     676           0 :   int slot_complete = !!( shred->data.flags & FD_SHRED_DATA_FLAG_SLOT_COMPLETE );
     677           0 :   int ref_tick      = shred->data.flags & FD_SHRED_DATA_REF_TICK_MASK;
     678             : 
     679           0 :   fd_forest_blk_t * ele = fd_forest_blk_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off );
     680           0 :   fd_forest_fec_insert( ctx->forest, shred->slot, shred->slot - shred->data.parent_off, shred->idx, shred->fec_set_idx, slot_complete, ref_tick );
     681           0 :   fd_fec_sig_t * fec_sig = fd_fec_sig_query( ctx->fec_sigs, (shred->slot << 32) | shred->fec_set_idx, NULL );
     682           0 :   if( FD_LIKELY( fec_sig ) ) fd_fec_sig_remove( ctx->fec_sigs, fec_sig );
     683           0 :   FD_TEST( ele ); /* must be non-empty */
     684             : 
     685             :   /* metrics for completed slots */
     686           0 :   if( FD_UNLIKELY( ele->complete_idx != UINT_MAX && ele->buffered_idx==ele->complete_idx &&
     687           0 :                    0==memcmp( ele->cmpl, ele->fecs, sizeof(fd_forest_blk_idxs_t) * fd_forest_blk_idxs_word_cnt ) ) ) {
     688           0 :     long now = fd_tickcount();
     689           0 :     long start_ts = ele->first_req_ts == 0 || ele->slot >= ctx->turbine_slot0 ? ele->first_shred_ts : ele->first_req_ts;
     690           0 :     ulong duration_ticks = (ulong)(now - start_ts);
     691           0 :     fd_histf_sample( ctx->metrics->slot_compl_time, duration_ticks );
     692           0 :     fd_repair_metrics_add_slot( ctx->slot_metrics, ele->slot, start_ts, now, ele->repair_cnt, ele->turbine_cnt );
     693           0 :     FD_LOG_INFO(( "slot is complete %lu. num_data_shreds: %u, num_repaired: %u, num_turbine: %u, num_recovered: %u, duration: %.2f ms", ele->slot, ele->complete_idx + 1, ele->repair_cnt, ele->turbine_cnt, ele->recovered_cnt, (double)fd_metrics_convert_ticks_to_nanoseconds(duration_ticks) / 1e6 ));
     694           0 :   }
     695             : 
     696           0 :   if( FD_UNLIKELY( ctx->profiler.enabled ) ) {
     697             :     // If turbine slot 0 is in the consumed frontier, and it satisfies the
     698             :     // above conditions for completions, then catchup is complete
     699           0 :     fd_forest_blk_t * turbine0     = fd_forest_query( ctx->forest, ctx->turbine_slot0 );
     700           0 :     ulong             turbine0_idx = fd_forest_pool_idx( fd_forest_pool( ctx->forest ), turbine0 );
     701           0 :     fd_forest_ref_t * consumed     = fd_forest_consumed_ele_query( fd_forest_consumed( ctx->forest ), &turbine0_idx, NULL, fd_forest_conspool( ctx->forest ) );
     702           0 :     if( FD_UNLIKELY( consumed && turbine0->complete_idx != UINT_MAX && turbine0->complete_idx == turbine0->buffered_idx &&
     703           0 :                      0==memcmp( turbine0->cmpl, turbine0->fecs, sizeof(fd_forest_blk_idxs_t) * fd_forest_blk_idxs_word_cnt ) ) ) {
     704           0 :       FD_COMPILER_MFENCE();
     705           0 :       FD_VOLATILE( ctx->profiler.complete ) = 1;
     706           0 :     }
     707           0 :   }
     708           0 : }
     709             : 
     710             : static inline void
     711             : after_net( ctx_t * ctx,
     712           0 :            ulong   sz  ) {
     713           0 :   fd_eth_hdr_t * eth; fd_ip4_hdr_t * ip4; fd_udp_hdr_t * udp;
     714           0 :   uchar * data; ulong data_sz;
     715           0 :   FD_TEST( fd_ip4_udp_hdr_strip( ctx->buffer, sz, &data, &data_sz, &eth, &ip4, &udp ) );
     716           0 :   fd_ip4_port_t peer_addr = { .addr=ip4->saddr, .port=udp->net_sport };
     717           0 :   if( FD_UNLIKELY( data_sz != sizeof(fd_repair_ping_t) ) ) {
     718           0 :     ctx->metrics->malformed_ping++;
     719           0 :     return;
     720           0 :   }
     721           0 :   fd_repair_ping_t * res = (fd_repair_ping_t *)fd_type_pun( data );
     722           0 :   if( FD_UNLIKELY( res->kind != FD_REPAIR_KIND_PING ) ) {
     723           0 :     ctx->metrics->malformed_ping++;
     724           0 :     return;
     725           0 :   }
     726           0 :   fd_repair_msg_t * pong = fd_repair_pong( ctx->protocol, &res->ping.hash );
     727           0 :   fd_signs_queue_push( ctx->sign_queue, (sign_pending_t){ .msg = *pong, .pong_data = { .peer_addr = peer_addr, .hash = res->ping.hash, .daddr = ip4->daddr } } );
     728           0 : }
     729             : 
     730             : static inline void
     731             : after_evict( ctx_t * ctx,
     732           0 :              ulong   sig ) {
     733           0 :   ulong spilled_slot        = fd_disco_shred_out_shred_sig_slot       ( sig );
     734           0 :   uint  spilled_fec_set_idx = fd_disco_shred_out_shred_sig_fec_set_idx( sig );
     735           0 :   uint  spilled_max_idx     = fd_disco_shred_out_shred_sig_data_cnt   ( sig );
     736             : 
     737           0 :   fd_forest_fec_clear( ctx->forest, spilled_slot, spilled_fec_set_idx, spilled_max_idx );
     738           0 : }
     739             : 
     740             : static void
     741             : after_frag( ctx_t * ctx,
     742             :             ulong                  in_idx,
     743             :             ulong                  seq    FD_PARAM_UNUSED,
     744             :             ulong                  sig,
     745             :             ulong                  sz,
     746             :             ulong                  tsorig FD_PARAM_UNUSED,
     747             :             ulong                  tspub  FD_PARAM_UNUSED,
     748           0 :             fd_stem_context_t *    stem ) {
     749           0 :   if( FD_UNLIKELY( ctx->skip_frag ) ) return;
     750             : 
     751           0 :   ctx->stem = stem;
     752             : 
     753           0 :   uint in_kind = ctx->in_kind[ in_idx ];
     754           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_GENESIS && sig==GENESI_SIG_BOOTSTRAP_COMPLETED ) ) {
     755           0 :     fd_forest_init( ctx->forest, 0 );
     756           0 :     return;
     757           0 :   }
     758             : 
     759           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_GOSSIP ) ) {
     760           0 :     fd_gossip_update_message_t const * msg = (fd_gossip_update_message_t const *)fd_type_pun_const( ctx->buffer );
     761           0 :     if( FD_LIKELY( sig==FD_GOSSIP_UPDATE_TAG_CONTACT_INFO ) ){
     762           0 :       after_contact( ctx, msg );
     763           0 :     } else {
     764           0 :       fd_policy_peer_remove( ctx->policy, (fd_pubkey_t const *)fd_type_pun_const( msg->origin_pubkey ) );
     765           0 :     }
     766           0 :     return;
     767           0 :   }
     768             : 
     769           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_TOWER ) ) {
     770           0 :     if( FD_LIKELY( sig==FD_TOWER_SIG_SLOT_DONE ) ) {
     771           0 :       fd_tower_slot_done_t const * msg = (fd_tower_slot_done_t const *)fd_type_pun_const( ctx->buffer );
     772           0 :       if( FD_LIKELY( msg->root_slot!=ULONG_MAX ) ) fd_forest_publish( ctx->forest, msg->root_slot );
     773           0 :     }
     774           0 :     return;
     775           0 :   }
     776             : 
     777           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SIGN ) ) {
     778           0 :     after_sign( ctx, in_idx, sig, stem );
     779           0 :     return;
     780           0 :   }
     781             : 
     782           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) {
     783             : 
     784             :     /* There are 3 message types from shred:
     785             :         1. resolver evict - incomplete FEC set is evicted by resolver
     786             :         2. fec complete   - FEC set is completed by resolver. Also contains a shred.
     787             :         3. shred          - new shred
     788             : 
     789             :         Msgs 2 and 3 have a shred header in ctx->buffer */
     790             : 
     791           0 :     int resolver_evicted = sz == 0;
     792           0 :     int fec_completes    = sz == FD_SHRED_DATA_HEADER_SZ + sizeof(fd_hash_t) + sizeof(fd_hash_t) + sizeof(int);
     793           0 :     if( FD_UNLIKELY( resolver_evicted ) ) {
     794           0 :       after_evict( ctx, sig );
     795           0 :       return;
     796           0 :     }
     797             : 
     798           0 :     fd_shred_t * shred = (fd_shred_t *)fd_type_pun( ctx->buffer );
     799           0 :     uint         nonce = FD_LOAD(uint, ctx->buffer + fd_shred_header_sz( shred->variant ) );
     800           0 :     if( FD_UNLIKELY( shred->slot <= fd_forest_root_slot( ctx->forest ) ) ) {
     801           0 :       FD_LOG_INFO(( "shred %lu %u %u too old, ignoring", shred->slot, shred->idx, shred->fec_set_idx ));
     802           0 :       return;
     803           0 :     };
     804             : 
     805           0 :     if( FD_UNLIKELY( ctx->profiler.enabled && ctx->turbine_slot0 != ULONG_MAX && shred->slot > ctx->turbine_slot0 ) ) return;
     806           0 : #   if LOGGING
     807           0 :     if( FD_UNLIKELY( shred->slot > ctx->metrics->current_slot ) ) {
     808           0 :       FD_LOG_INFO(( "\n\n[Turbine]\n"
     809           0 :                     "slot:             %lu\n"
     810           0 :                     "root:             %lu\n",
     811           0 :                     shred->slot,
     812           0 :                     fd_forest_root_slot( ctx->forest ) ));
     813           0 :     }
     814           0 : #   endif
     815           0 :     ctx->metrics->current_slot  = fd_ulong_max( shred->slot, ctx->metrics->current_slot );
     816           0 :     if( FD_UNLIKELY( ctx->turbine_slot0 == ULONG_MAX ) ) {
     817             : 
     818           0 :       if( FD_UNLIKELY( ctx->profiler.enabled ) ) {
     819             :         /* we wait until the first turbine shred arrives to kick off
     820             :            the profiler.  This is to let gossip peers accumulate similar
     821             :            to a regular Firedancer run. */
     822           0 :         fd_forest_blk_insert( ctx->forest, ctx->profiler.end_slot, ctx->profiler.end_slot - 1 );
     823           0 :         fd_forest_code_shred_insert( ctx->forest, ctx->profiler.end_slot, 0 );
     824             : 
     825           0 :         ctx->turbine_slot0 = ctx->profiler.end_slot;
     826           0 :         fd_repair_metrics_set_turbine_slot0( ctx->slot_metrics, ctx->profiler.end_slot );
     827           0 :         fd_policy_set_turbine_slot0( ctx->policy, ctx->profiler.end_slot );
     828           0 :         return;
     829           0 :       }
     830             : 
     831           0 :       ctx->turbine_slot0 = shred->slot;
     832           0 :       fd_repair_metrics_set_turbine_slot0( ctx->slot_metrics, shred->slot );
     833           0 :       fd_policy_set_turbine_slot0( ctx->policy, shred->slot );
     834           0 :     }
     835             : 
     836           0 :     if( FD_UNLIKELY( fec_completes ) ) {
     837           0 :       after_fec( ctx, shred );
     838           0 :     } else {
     839             :       /* Don't want to reinsert the shred sig for an already complete FEC set */
     840           0 :       fd_fec_sig_t * fec_sig = fd_fec_sig_query( ctx->fec_sigs, (shred->slot << 32) | shred->fec_set_idx, NULL );
     841           0 :       if( FD_UNLIKELY( !fec_sig ) ) {
     842           0 :         fec_sig = fd_fec_sig_insert( ctx->fec_sigs, (shred->slot << 32) | shred->fec_set_idx );
     843           0 :         memcpy( fec_sig->sig, shred->signature, sizeof(fd_ed25519_sig_t) );
     844           0 :       }
     845           0 :       after_shred( ctx, sig, shred, nonce );
     846           0 :     }
     847             : 
     848             :     /* Check if there are FECs to force complete. Algorithm: window
     849             :        through the idxs in interval [i, j). If j = next fec_set_idx
     850             :        then we know we can force complete the FEC set interval [i, j)
     851             :        (assuming it wasn't already completed based on `cmpl`). */
     852             : 
     853           0 :     fd_forest_blk_t * blk = fd_forest_query( ctx->forest, shred->slot );
     854           0 :     if( blk ) {
     855           0 :       uint i = blk->consumed_idx + 1;
     856           0 :       for( uint j = i; j < blk->buffered_idx + 1; j++ ) {
     857           0 :         if( FD_UNLIKELY( fd_forest_blk_idxs_test( blk->fecs, j ) ) ) {
     858           0 :           if( FD_UNLIKELY( fd_forest_blk_idxs_test( blk->cmpl, j ) ) ) {
     859             :             /* already been completed without force complete */
     860           0 :           } else {
     861             :             /* force completeable */
     862           0 :             fd_fec_sig_t * fec_sig  = fd_fec_sig_query( ctx->fec_sigs, (shred->slot << 32) | i, NULL );
     863           0 :             if( FD_LIKELY( fec_sig ) ) {
     864           0 :               ulong          sig      = fd_ulong_load_8( fec_sig->sig );
     865           0 :               ulong          tile_idx = sig % ctx->shred_tile_cnt;
     866           0 :               uint           last_idx = j - i;
     867             : 
     868           0 :               uchar * chunk = fd_chunk_to_laddr( ctx->shred_out_ctx[tile_idx].mem, ctx->shred_out_ctx[tile_idx].chunk );
     869           0 :               memcpy( chunk, fec_sig->sig, sizeof(fd_ed25519_sig_t) );
     870           0 :               fd_fec_sig_remove( ctx->fec_sigs, fec_sig );
     871           0 :               fd_stem_publish( stem, ctx->shred_out_ctx[tile_idx].idx, last_idx, ctx->shred_out_ctx[tile_idx].chunk, sizeof(fd_ed25519_sig_t), 0UL, 0UL, 0UL );
     872           0 :               ctx->shred_out_ctx[tile_idx].chunk = fd_dcache_compact_next( ctx->shred_out_ctx[tile_idx].chunk, sizeof(fd_ed25519_sig_t), ctx->shred_out_ctx[tile_idx].chunk0, ctx->shred_out_ctx[tile_idx].wmark );
     873           0 :             }
     874           0 :           }
     875             :           /* advance consumed */
     876           0 :           blk->consumed_idx = j;
     877           0 :           i = j + 1;
     878           0 :         }
     879           0 :       }
     880           0 :     }
     881             :     /* update metrics */
     882           0 :     ctx->metrics->repaired_slots = fd_forest_highest_repaired_slot( ctx->forest );
     883           0 :     return;
     884           0 :   }
     885             : 
     886           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_STAKE ) ) {
     887           0 :     return;
     888           0 :   }
     889             : 
     890           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_SNAP ) ) {
     891           0 :     after_snap( ctx, sig, fd_chunk_to_laddr( ctx->in_links[ in_idx ].mem, ctx->snap_out_chunk ) );
     892           0 :     return;
     893           0 :   }
     894             : 
     895           0 :   if( FD_UNLIKELY( in_kind==IN_KIND_NET ) ) {
     896           0 :     after_net( ctx, sz );
     897           0 :     return;
     898           0 :   }
     899           0 : }
     900             : 
     901             : static inline void
     902             : after_credit( ctx_t *             ctx,
     903             :               fd_stem_context_t * stem FD_PARAM_UNUSED,
     904             :               int *               opt_poll_in FD_PARAM_UNUSED,
     905           0 :               int *               charge_busy ) {
     906           0 :   long now = fd_log_wallclock();
     907             : 
     908             :   /* Verify that there is at least one sign tile with available credits.
     909             :      If not, we can't send any requests and leave early. */
     910           0 :   out_ctx_t * sign_out = sign_avail_credits( ctx );
     911           0 :   if( FD_UNLIKELY( !sign_out ) ) {
     912           0 :     ctx->metrics->sign_tile_unavail++;
     913           0 :     return;
     914           0 :   }
     915           0 :   if( FD_UNLIKELY( !fd_signs_queue_empty( ctx->sign_queue ) ) ) {
     916           0 :     sign_pending_t signable = fd_signs_queue_pop( ctx->sign_queue );
     917           0 :     fd_repair_send_sign_request( ctx, sign_out, &signable.msg, signable.msg.kind == FD_REPAIR_KIND_PONG ? &signable.pong_data : NULL );
     918           0 :     *charge_busy = 1;
     919           0 :     return;
     920           0 :   }
     921             : 
     922           0 :   if( FD_UNLIKELY( fd_inflights_should_drain( ctx->inflights, now ) ) ) {
     923           0 :     ulong nonce; ulong slot; ulong shred_idx;
     924           0 :     *charge_busy = 1;
     925           0 :     fd_inflights_request_pop( ctx->inflights, &nonce, &slot, &shred_idx );
     926           0 :     fd_forest_blk_t * blk = fd_forest_query( ctx->forest, slot );
     927           0 :     if( FD_UNLIKELY( blk && !fd_forest_blk_idxs_test( blk->idxs, shred_idx ) ) ) {
     928           0 :       fd_pubkey_t const * peer = fd_policy_peer_select( ctx->policy );
     929           0 :       ctx->metrics->rerequest++;
     930           0 :       if( FD_UNLIKELY( !peer ) ) {
     931             :         /* No peers. But we CANNOT lose this request. */
     932             :         /* Add this request to the inflights table, pretend we've sent it and let the inflight timeout request it down the line. */
     933           0 :         fd_hash_t hash = { .ul[0] = 0 };
     934           0 :         fd_inflights_request_insert( ctx->inflights, ctx->policy->nonce++, &hash, slot, shred_idx );
     935           0 :       } else {
     936           0 :         fd_repair_msg_t * msg = fd_repair_shred( ctx->protocol, peer, (ulong)((ulong)now / 1e6L), ctx->policy->nonce++, slot, shred_idx );
     937           0 :         fd_repair_send_sign_request( ctx, sign_out, msg, NULL );
     938           0 :         return;
     939           0 :       }
     940           0 :     }
     941           0 :   }
     942             : 
     943           0 :   fd_repair_msg_t const * cout = fd_policy_next( ctx->policy, ctx->forest, ctx->protocol, now, ctx->metrics->current_slot, charge_busy );
     944           0 :   if( FD_UNLIKELY( !cout ) ) return;
     945             : 
     946           0 :   fd_repair_send_sign_request( ctx, sign_out, cout, NULL );
     947           0 : }
     948             : 
     949             : static inline void
     950           0 : during_housekeeping( ctx_t * ctx ) {
     951           0 :   (void)ctx;
     952             : # if DEBUG_LOGGING
     953             :   long now = fd_log_wallclock();
     954             :   if( FD_UNLIKELY( now - ctx->tsdebug > (long)10e9 ) ) {
     955             :     fd_forest_print( ctx->forest );
     956             :     ctx->tsdebug = fd_log_wallclock();
     957             :   }
     958             : # endif
     959           0 : }
     960             : 
     961             : static void
     962             : privileged_init( fd_topo_t *      topo,
     963           0 :                  fd_topo_tile_t * tile ) {
     964           0 :   void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
     965             : 
     966           0 :   FD_SCRATCH_ALLOC_INIT( l, scratch );
     967           0 :   ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t), sizeof(ctx_t) );
     968           0 :   fd_memset( ctx, 0, sizeof(ctx_t) );
     969             : 
     970           0 :   uchar const * identity_key = fd_keyload_load( tile->repair.identity_key_path, /* pubkey only: */ 0 );
     971           0 :   fd_memcpy( ctx->identity_public_key.uc, identity_key + 32UL, sizeof(fd_pubkey_t) );
     972             : 
     973           0 :   FD_TEST( fd_rng_secure( &ctx->repair_seed, sizeof(ulong) ) );
     974           0 : }
     975             : 
     976             : static void
     977             : unprivileged_init( fd_topo_t *      topo,
     978           0 :                    fd_topo_tile_t * tile ) {
     979           0 :   void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
     980             : 
     981           0 :   ulong total_sign_depth = tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt;
     982           0 :   int   lg_sign_depth    = fd_ulong_find_msb( fd_ulong_pow2_up(total_sign_depth) ) + 1;
     983             : 
     984           0 :   FD_SCRATCH_ALLOC_INIT( l, scratch );
     985           0 :   ctx_t * ctx       = FD_SCRATCH_ALLOC_APPEND( l, alignof(ctx_t),            sizeof(ctx_t)                                                    );
     986           0 :   ctx->protocol     = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_align(),         fd_repair_footprint     ()                                       );
     987           0 :   ctx->forest       = FD_SCRATCH_ALLOC_APPEND( l, fd_forest_align(),         fd_forest_footprint     ( tile->repair.slot_max )                );
     988           0 :   ctx->policy       = FD_SCRATCH_ALLOC_APPEND( l, fd_policy_align(),         fd_policy_footprint     ( FD_NEEDED_KEY_MAX, FD_ACTIVE_KEY_MAX ) );
     989           0 :   ctx->inflights    = FD_SCRATCH_ALLOC_APPEND( l, fd_inflights_align(),      fd_inflights_footprint  ()                                       );
     990           0 :   ctx->fec_sigs     = FD_SCRATCH_ALLOC_APPEND( l, fd_fec_sig_align(),        fd_fec_sig_footprint    ( 20 )                                   );
     991           0 :   ctx->signs_map    = FD_SCRATCH_ALLOC_APPEND( l, fd_signs_map_align(),      fd_signs_map_footprint  ( lg_sign_depth )                        );
     992           0 :   ctx->sign_queue   = FD_SCRATCH_ALLOC_APPEND( l, fd_signs_queue_align(),    fd_signs_queue_footprint()                                       );
     993           0 :   ctx->slot_metrics = FD_SCRATCH_ALLOC_APPEND( l, fd_repair_metrics_align(), fd_repair_metrics_footprint()                                    );
     994           0 :   FD_TEST( FD_SCRATCH_ALLOC_FINI( l, scratch_align() ) == (ulong)scratch + scratch_footprint( tile ) );
     995             : 
     996           0 :   ctx->protocol     = fd_repair_join        ( fd_repair_new        ( ctx->protocol, &ctx->identity_public_key                              ) );
     997           0 :   ctx->forest       = fd_forest_join        ( fd_forest_new        ( ctx->forest,   tile->repair.slot_max, ctx->repair_seed                ) );
     998           0 :   ctx->policy       = fd_policy_join        ( fd_policy_new        ( ctx->policy,   FD_NEEDED_KEY_MAX, FD_ACTIVE_KEY_MAX, ctx->repair_seed ) );
     999           0 :   ctx->inflights    = fd_inflights_join     ( fd_inflights_new     ( ctx->inflights                                                        ) );
    1000           0 :   ctx->fec_sigs     = fd_fec_sig_join       ( fd_fec_sig_new       ( ctx->fec_sigs, 20, 0UL                                                ) );
    1001           0 :   ctx->signs_map    = fd_signs_map_join     ( fd_signs_map_new     ( ctx->signs_map, lg_sign_depth, 0UL                                    ) );
    1002           0 :   ctx->sign_queue   = fd_signs_queue_join   ( fd_signs_queue_new   ( ctx->sign_queue                                                       ) );
    1003           0 :   ctx->slot_metrics = fd_repair_metrics_join( fd_repair_metrics_new( ctx->slot_metrics                                                     ) );
    1004             : 
    1005             :   /* Process in links */
    1006             : 
    1007           0 :   if( FD_UNLIKELY( tile->in_cnt > MAX_IN_LINKS ) ) FD_LOG_ERR(( "repair tile has too many input links" ));
    1008             : 
    1009           0 :   uint  sign_repair_in_idx[ MAX_SIGN_TILE_CNT ] = {0};
    1010           0 :   uint  sign_repair_idx  = 0;
    1011           0 :   ulong sign_link_depth  = 0;
    1012             : 
    1013           0 :   for( uint in_idx=0U; in_idx<(tile->in_cnt); in_idx++ ) {
    1014           0 :     fd_topo_link_t * link = &topo->links[ tile->in_link_id[ in_idx ] ];
    1015           0 :     if( 0==strcmp( link->name, "net_repair" ) ) {
    1016           0 :       ctx->in_kind[ in_idx ] = IN_KIND_NET;
    1017           0 :       fd_net_rx_bounds_init( &ctx->in_links[ in_idx ].net_rx, link->dcache );
    1018           0 :       continue;
    1019           0 :     } else if( 0==strcmp( link->name, "sign_repair" ) ) {
    1020           0 :       ctx->in_kind[ in_idx ]                  = IN_KIND_SIGN;
    1021           0 :       sign_repair_in_idx[ sign_repair_idx++ ] = in_idx;
    1022           0 :       sign_link_depth                         = link->depth;
    1023           0 :     }
    1024           0 :     else if( 0==strcmp( link->name, "gossip_out"   ) ) ctx->in_kind[ in_idx ] = IN_KIND_GOSSIP;
    1025           0 :     else if( 0==strcmp( link->name, "tower_out"    ) ) ctx->in_kind[ in_idx ] = IN_KIND_TOWER;
    1026           0 :     else if( 0==strcmp( link->name, "shred_out"    ) ) ctx->in_kind[ in_idx ] = IN_KIND_SHRED;
    1027           0 :     else if( 0==strcmp( link->name, "snapin_manif" ) ) ctx->in_kind[ in_idx ] = IN_KIND_SNAP;
    1028           0 :     else if( 0==strcmp( link->name, "replay_stake" ) ) ctx->in_kind[ in_idx ] = IN_KIND_STAKE;
    1029           0 :     else if( 0==strcmp( link->name, "genesi_out"   ) ) ctx->in_kind[ in_idx ] = IN_KIND_GENESIS;
    1030           0 :     else FD_LOG_ERR(( "repair tile has unexpected input link %s", link->name ));
    1031             : 
    1032           0 :     ctx->in_links[ in_idx ].mem    = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
    1033           0 :     ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
    1034           0 :     ctx->in_links[ in_idx ].wmark  = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
    1035           0 :     ctx->in_links[ in_idx ].mtu    = link->mtu;
    1036             : 
    1037           0 :     FD_TEST( fd_dcache_compact_is_safe( ctx->in_links[in_idx].mem, link->dcache, link->mtu, link->depth ) );
    1038           0 :   }
    1039             : 
    1040           0 :   ctx->net_out_idx       = UINT_MAX;
    1041           0 :   ctx->shred_tile_cnt    = 0;
    1042           0 :   ctx->repair_sign_cnt   = 0;
    1043           0 :   ctx->sign_rrobin_idx   = 0;
    1044             : 
    1045           0 :   for( uint out_idx=0U; out_idx<(tile->out_cnt); out_idx++ ) {
    1046           0 :     fd_topo_link_t * link = &topo->links[ tile->out_link_id[ out_idx ] ];
    1047             : 
    1048           0 :     if( 0==strcmp( link->name, "repair_net" ) ) {
    1049             : 
    1050           0 :       if( ctx->net_out_idx!=UINT_MAX ) continue; /* only use first net link */
    1051           0 :       ctx->net_out_idx    = out_idx;
    1052           0 :       ctx->net_out_mem    = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
    1053           0 :       ctx->net_out_chunk0 = fd_dcache_compact_chunk0( ctx->net_out_mem, link->dcache );
    1054           0 :       ctx->net_out_wmark  = fd_dcache_compact_wmark( ctx->net_out_mem, link->dcache, link->mtu );
    1055           0 :       ctx->net_out_chunk  = ctx->net_out_chunk0;
    1056             : 
    1057           0 :     } else if( 0==strcmp( link->name, "repair_shred" ) ) {
    1058             : 
    1059           0 :       out_ctx_t * shred_out = &ctx->shred_out_ctx[ ctx->shred_tile_cnt++ ];
    1060           0 :       shred_out->idx        = out_idx;
    1061           0 :       shred_out->mem        = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
    1062           0 :       shred_out->chunk0     = fd_dcache_compact_chunk0( shred_out->mem, link->dcache );
    1063           0 :       shred_out->wmark      = fd_dcache_compact_wmark( shred_out->mem, link->dcache, link->mtu );
    1064           0 :       shred_out->chunk      = shred_out->chunk0;
    1065             : 
    1066           0 :     } else if( 0==strcmp( link->name, "repair_sign" ) ) {
    1067             : 
    1068           0 :       out_ctx_t * repair_sign_out  = &ctx->repair_sign_out_ctx[ ctx->repair_sign_cnt ];
    1069           0 :       repair_sign_out->idx         = out_idx;
    1070           0 :       repair_sign_out->mem         = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;
    1071           0 :       repair_sign_out->chunk0      = fd_dcache_compact_chunk0( repair_sign_out->mem, link->dcache );
    1072           0 :       repair_sign_out->wmark       = fd_dcache_compact_wmark( repair_sign_out->mem, link->dcache, link->mtu );
    1073           0 :       repair_sign_out->chunk       = repair_sign_out->chunk0;
    1074           0 :       repair_sign_out->in_idx      = sign_repair_in_idx[ ctx->repair_sign_cnt++ ]; /* match to the sign_repair input link */
    1075           0 :       repair_sign_out->max_credits = sign_link_depth;
    1076           0 :       repair_sign_out->credits     = sign_link_depth;
    1077             : 
    1078           0 :     } else {
    1079           0 :       FD_LOG_ERR(( "repair tile has unexpected output link %s", link->name ));
    1080           0 :     }
    1081           0 :   }
    1082           0 :   if( FD_UNLIKELY( ctx->net_out_idx==UINT_MAX       ) ) FD_LOG_ERR(( "Missing repair_net link" ));
    1083           0 :   if( FD_UNLIKELY( ctx->repair_sign_cnt!=sign_repair_idx ) ) {
    1084           0 :     FD_LOG_ERR(( "Mismatch between repair_sign output links (%lu) and sign_repair input links (%u)", ctx->repair_sign_cnt, sign_repair_idx ));
    1085           0 :   }
    1086             : 
    1087           0 :   FD_TEST( ctx->shred_tile_cnt == fd_topo_tile_name_cnt( topo, "shred" ) );
    1088             : 
    1089             : # if DEBUG_LOGGING
    1090             :   if( fd_signs_map_key_max( ctx->signs_map ) < tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt ) {
    1091             :     FD_LOG_ERR(( "repair pending signs tracking map is too small: %lu < %lu.  Increase the key_max", fd_signs_map_key_max( ctx->signs_map ), tile->repair.repair_sign_depth * tile->repair.repair_sign_cnt ));
    1092             :   }
    1093             : # endif
    1094             : 
    1095           0 :   ctx->wksp = topo->workspaces[ topo->objs[ tile->tile_obj_id ].wksp_id ].wksp;
    1096           0 :   ctx->repair_intake_addr.port = fd_ushort_bswap( tile->repair.repair_intake_listen_port );
    1097           0 :   ctx->repair_serve_addr.port  = fd_ushort_bswap( tile->repair.repair_serve_listen_port  );
    1098             : 
    1099           0 :   ctx->net_id = (ushort)0;
    1100           0 :   fd_ip4_udp_hdr_init( ctx->intake_hdr, FD_REPAIR_MAX_PACKET_SIZE, 0, tile->repair.repair_intake_listen_port );
    1101           0 :   fd_ip4_udp_hdr_init( ctx->serve_hdr,  FD_REPAIR_MAX_PACKET_SIZE, 0, tile->repair.repair_serve_listen_port  );
    1102             : 
    1103             :   /* Repair set up */
    1104             : 
    1105           0 :   ctx->turbine_slot0 = ULONG_MAX;
    1106           0 :   FD_LOG_INFO(( "repair my addr - intake addr: " FD_IP4_ADDR_FMT ":%u, serve_addr: " FD_IP4_ADDR_FMT ":%u",
    1107           0 :     FD_IP4_ADDR_FMT_ARGS( ctx->repair_intake_addr.addr ), fd_ushort_bswap( ctx->repair_intake_addr.port ),
    1108           0 :     FD_IP4_ADDR_FMT_ARGS( ctx->repair_serve_addr.addr ), fd_ushort_bswap( ctx->repair_serve_addr.port ) ));
    1109             : 
    1110           0 :   memset( ctx->metrics, 0, sizeof(ctx->metrics) );
    1111             : 
    1112           0 :   fd_histf_join( fd_histf_new( ctx->metrics->slot_compl_time, FD_MHIST_SECONDS_MIN( REPAIR, SLOT_COMPLETE_TIME ),
    1113           0 :                                                               FD_MHIST_SECONDS_MAX( REPAIR, SLOT_COMPLETE_TIME ) ) );
    1114           0 :   fd_histf_join( fd_histf_new( ctx->metrics->response_latency, FD_MHIST_MIN( REPAIR, RESPONSE_LATENCY ),
    1115           0 :                                                                FD_MHIST_MAX( REPAIR, RESPONSE_LATENCY ) ) );
    1116             : 
    1117           0 :   ctx->tsdebug = fd_log_wallclock();
    1118           0 :   ctx->pending_key_next = 0;
    1119           0 :   ctx->profiler.enabled  = tile->repair.end_slot != 0UL;
    1120           0 :   ctx->profiler.end_slot = tile->repair.end_slot;
    1121           0 :   if( ctx->profiler.enabled ) {
    1122           0 :     ctx->metrics->current_slot = tile->repair.end_slot + 1; /* +1 to allow the turbine slot 0 to be completed */
    1123           0 :     ctx->profiler.complete     = 0;
    1124           0 :   }
    1125           0 : }
    1126             : 
    1127             : static ulong
    1128             : populate_allowed_seccomp( fd_topo_t const *      topo FD_PARAM_UNUSED,
    1129             :                           fd_topo_tile_t const * tile FD_PARAM_UNUSED,
    1130             :                           ulong                  out_cnt,
    1131           0 :                           struct sock_filter *   out ) {
    1132           0 :   populate_sock_filter_policy_fd_repair_tile(
    1133           0 :     out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)-1 );
    1134           0 :   return sock_filter_policy_fd_repair_tile_instr_cnt;
    1135           0 : }
    1136             : 
    1137             : static ulong
    1138             : populate_allowed_fds( fd_topo_t const *      topo FD_PARAM_UNUSED,
    1139             :                       fd_topo_tile_t const * tile FD_PARAM_UNUSED,
    1140             :                       ulong                  out_fds_cnt,
    1141           0 :                       int *                  out_fds ) {
    1142           0 :   if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
    1143             : 
    1144           0 :   ulong out_cnt = 0UL;
    1145           0 :   out_fds[ out_cnt++ ] = 2; /* stderr */
    1146           0 :   if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
    1147           0 :     out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
    1148           0 :   return out_cnt;
    1149           0 : }
    1150             : 
    1151             : static inline void
    1152           0 : metrics_write( ctx_t * ctx ) {
    1153           0 :   FD_MCNT_SET( REPAIR, CURRENT_SLOT,      ctx->metrics->current_slot );
    1154           0 :   FD_MCNT_SET( REPAIR, REPAIRED_SLOTS,    ctx->metrics->repaired_slots );
    1155           0 :   FD_MCNT_SET( REPAIR, REQUEST_PEERS,     fd_peer_pool_used( ctx->policy->peers.pool ) );
    1156           0 :   FD_MCNT_SET( REPAIR, SIGN_TILE_UNAVAIL, ctx->metrics->sign_tile_unavail );
    1157           0 :   FD_MCNT_SET( REPAIR, REREQUEST_QUEUE,   ctx->metrics->rerequest );
    1158             : 
    1159           0 :   FD_MCNT_SET      ( REPAIR, TOTAL_PKT_COUNT, ctx->metrics->send_pkt_cnt   );
    1160           0 :   FD_MCNT_ENUM_COPY( REPAIR, SENT_PKT_TYPES,  ctx->metrics->sent_pkt_types );
    1161             : 
    1162           0 :   FD_MHIST_COPY( REPAIR, SLOT_COMPLETE_TIME, ctx->metrics->slot_compl_time );
    1163           0 :   FD_MHIST_COPY( REPAIR, RESPONSE_LATENCY,   ctx->metrics->response_latency );
    1164           0 : }
    1165             : 
    1166             : #undef DEBUG_LOGGING
    1167             : 
    1168             : /* TODO: This is not correct, but is temporary and will be fixed
    1169             :    when fixed FEC 32 goes in, and we can finally get rid of force
    1170             :    completes BS. */
    1171           0 : #define STEM_BURST (64UL)
    1172             : 
    1173           0 : #define STEM_CALLBACK_CONTEXT_TYPE  ctx_t
    1174           0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(ctx_t)
    1175             : 
    1176           0 : #define STEM_CALLBACK_AFTER_CREDIT        after_credit
    1177           0 : #define STEM_CALLBACK_BEFORE_FRAG         before_frag
    1178           0 : #define STEM_CALLBACK_DURING_FRAG         during_frag
    1179           0 : #define STEM_CALLBACK_AFTER_FRAG          after_frag
    1180           0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
    1181           0 : #define STEM_CALLBACK_METRICS_WRITE       metrics_write
    1182             : 
    1183             : #include "../../disco/stem/fd_stem.c"
    1184             : 
    1185             : fd_topo_run_tile_t fd_tile_repair = {
    1186             :   .name                     = "repair",
    1187             :   .loose_footprint          = loose_footprint,
    1188             :   .populate_allowed_seccomp = populate_allowed_seccomp,
    1189             :   .populate_allowed_fds     = populate_allowed_fds,
    1190             :   .scratch_align            = scratch_align,
    1191             :   .scratch_footprint        = scratch_footprint,
    1192             :   .unprivileged_init        = unprivileged_init,
    1193             :   .privileged_init          = privileged_init,
    1194             :   .run                      = stem_run,
    1195             : };

Generated by: LCOV version 1.14