Repair & Sign: async repair requests

Author: npatel-jump

src/app/firedancer/config/default.toml

@@ -689,10 +689,10 @@ user = ""
     shred_tile_count = 1
 
     # How many sign tiles to run. Should be set to 2. This is
-    # configurable and designed to scale for the repair tile.  
+    # configurable and designed to scale for the repair tile.
     # The 0th tile gets used by other tiles for signing messages.  While
     # the remaining tiles distribute the workload of signing repair
-    # requests. 
+    # requests.
     sign_tile_count = 2
 
 # All memory that will be used in Firedancer is pre-allocated in two

src/app/firedancer/topology.c

@@ -348,15 +348,15 @@ fd_topo_initialize( config_t * config ) {
   /**/                 fd_topob_link( topo, "send_net",     "net_send",     config->net.ingress_buffer_size,          FD_NET_MTU,                    2UL );
 
   /**/                 fd_topob_link( topo, "repair_net",   "net_repair",   config->net.ingress_buffer_size,          FD_NET_MTU,                    1UL );
-  /**/                 fd_topob_link( topo, "ping_sign",  "repair_sign",  128UL,                                    2048UL,                        1UL );
+  /**/                 fd_topob_link( topo, "ping_sign",  "repair_sign",  1024UL,                                    2048UL,                        1UL );
   FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_repair", "shred_repair", pending_fec_shreds_depth,                 FD_SHRED_REPAIR_MTU,           2UL /* at most 2 msgs per after_frag */ );
 
 
   FOR(shred_tile_cnt)  fd_topob_link( topo, "repair_shred", "shred_repair", pending_fec_shreds_depth,                 sizeof(fd_ed25519_sig_t),      1UL );
 
-  /**/                 fd_topob_link( topo, "sign_ping",  "repair_sign",  128UL,                                    64UL,                          1UL );
-  FOR(sign_tile_cnt-1) fd_topob_link( topo, "repair_sign",  "repair_sign",  128UL,                                    2048UL,                        1UL );
-  FOR(sign_tile_cnt-1) fd_topob_link( topo, "sign_repair",  "repair_sign",  128UL,                                    64UL,                          1UL );
+  /**/                 fd_topob_link( topo, "sign_ping",  "repair_sign",  1024UL,                                    64UL,                          1UL );
+  FOR(sign_tile_cnt-1) fd_topob_link( topo, "repair_sign",  "repair_sign",  1024UL,                                    2048UL,                        1UL );
+  FOR(sign_tile_cnt-1) fd_topob_link( topo, "sign_repair",  "repair_sign",  1024UL,                                    64UL,                          1UL );
 
   /**/                 fd_topob_link( topo, "repair_repla", "repair_repla", 65536UL,                                  sizeof(fd_fec_out_t),                                        1UL );
   FOR(bank_tile_cnt)   fd_topob_link( topo, "replay_poh",   "replay_poh",   128UL,                                    (4096UL*sizeof(fd_txn_p_t))+sizeof(fd_microblock_trailer_t), 1UL  );

src/disco/shred/fd_shred_tile.c

@@ -857,7 +857,6 @@ after_frag( fd_shred_ctx_t *    ctx,
         ulong sz = fd_shred_header_sz( shred->variant );
         fd_memcpy( fd_chunk_to_laddr( ctx->repair_out_mem, ctx->repair_out_chunk ), shred, sz );
         ulong tspub = fd_frag_meta_ts_comp( fd_tickcount() );
-        // FD_LOG_INFO(("Published shred %lu %u %u, time: %ld", shred->slot, shred->idx, shred->fec_set_idx, fd_log_wallclock()));
         fd_stem_publish( stem, ctx->repair_out_idx, sig, ctx->repair_out_chunk, sz, 0UL, ctx->tsorig, tspub );
         ctx->repair_out_chunk = fd_dcache_compact_next( ctx->repair_out_chunk, sz, ctx->repair_out_chunk0, ctx->repair_out_wmark );
       }

src/disco/sign/fd_sign_tile.c

@@ -121,9 +121,10 @@ during_frag_sensitive( void * _ctx,
 
   int   role = ctx->in[ in_idx ].role;
   ulong mtu  = ctx->in[ in_idx ].mtu;
-  
+
   if( chunk<ctx->in[ in_idx ].chunk0 || chunk>ctx->in[ in_idx ].wmark || sz>mtu ) {
-    FD_LOG_EMERG(( "oversz or out of bounds signing request (role=%d chunk=%lu sz=%lu mtu=%lu, chunk0=%lu, wmark=%lu)", role, chunk, sz, mtu, ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
+    FD_LOG_EMERG(( "oversz or out of bounds signing request (role=%d chunk=%lu sz=%lu mtu=%lu, chunk0=%lu, wmark=%lu)",
+      role, chunk, sz, mtu, ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
   }
 
   void * src = fd_chunk_to_laddr( ctx->in[ in_idx ].mem, chunk );

src/discof/repair/fd_repair_tile.c

@@ -130,6 +130,8 @@ struct fd_repair_tile_ctx {
 
   fd_wksp_t * wksp;
 
+  fd_stem_context_t * stem;
+
   uchar              in_kind[ MAX_IN_LINKS ];
   fd_repair_in_ctx_t in_links[ MAX_IN_LINKS ];
 
@@ -235,19 +237,19 @@ repair_signer_async( void *        signer_ctx,
                      ulong         len,
                      int           sign_type ) {
   fd_repair_tile_ctx_t * ctx = (fd_repair_tile_ctx_t *) signer_ctx;
-  
+
   if( FD_UNLIKELY( ctx->repair_sign_cnt == 0 ) ) {
     FD_LOG_ERR(( "No repair_sign links configured for async signing" ));
   }
-  
+
   uint round_robin_idx = (uint)(nonce % ctx->repair_sign_cnt);
   fd_repair_out_ctx_t * sign_out = &ctx->repair_sign_out_ctx[ round_robin_idx ];
-  
+
   uchar * dst = fd_chunk_to_laddr( sign_out->mem, sign_out->chunk );
   fd_memcpy( dst, buffer, len );
-  
+
   ulong sig = ((ulong)nonce << 32) | (ulong)(uint)sign_type;
-  fd_stem_publish( ctx->stem, sign_out->idx, sig, sign_out->chunk, len, 0UL, 0UL, 0UL );  
+  fd_stem_publish( ctx->stem, sign_out->idx, sig, sign_out->chunk, len, 0UL, 0UL, 0UL );
   sign_out->chunk = fd_dcache_compact_next( sign_out->chunk, len, sign_out->chunk0, sign_out->wmark );
 
   ctx->request_seq = fd_seq_inc( ctx->request_seq, 1UL );
@@ -424,7 +426,7 @@ fd_repair_sign_and_send( fd_repair_tile_ctx_t *  repair_tile_ctx,
   }
 }
 
-/*  REPAIR TILE REQUEST HANDLING ARCHITECTURE
+/*  REQUEST HANDLING ARCHITECTURE
     =========================================
 
     The repair tile implements two distinct request handling patterns
@@ -440,11 +442,7 @@ fd_repair_sign_and_send( fd_repair_tile_ctx_t *  repair_tile_ctx,
     - PINGs & PONGs: Handles peer connectivity and liveness with simple
       round-trip messages.
 
-    - PEER WARM UPs: On receiving peer information in
-      handle_new_cluster_contact_info, we prepay the RTT cost by sending
-      a placeholder Repair request immediately.
-
-    2. ASYNCHRONOUS REQUEST HANDLING  
+    2. ASYNCHRONOUS REQUEST HANDLING
     --------------------------------
     Used strictly for repair requests. These requests are sent to the
     sign tile, and the repair tile continues handling other operations
@@ -456,15 +454,19 @@ fd_repair_sign_and_send( fd_repair_tile_ctx_t *  repair_tile_ctx,
     - WINDOW_INDEX (exact shred): Requests for a specific shred at a
       known slot and index. Used when the repair tile knows exactly
       which shred is missing from a FEC set.
-  
+
     - HIGHEST_WINDOW_INDEX: Requests for the highest shred in a slot.
       Used to determine the end boundary of a slot when the exact count
       is unknown.
-  
+
     - ORPHAN: Requests for the highest shred in the parent slot of an
       orphaned slot. Used to establish the chain of slot ancestry when a
       slot's parent is missing.
 
+    - PEER WARM UPs: On receiving peer information in
+      handle_new_cluster_contact_info, we prepay the RTT cost by sending
+      a placeholder Repair request immediately.
+
     Async requests can be distributed across multiple sign tiles using
     round-robin based on the request nonce. This provides load balancing
     and prevents any single sign tile from becoming a bottleneck. */
@@ -519,32 +521,29 @@ fd_repair_send_request_async( fd_repair_tile_ctx_t *   ctx,
                               long                     now ){
     fd_active_elem_t * peer = fd_active_table_query(glob->actives, recipient, NULL);
     if (!peer) return;
-  
+
     uint nonce = (uint)glob->next_nonce;
 
     fd_repair_protocol_t protocol;
     fd_repair_construct_request_protocol( glob, &protocol, type, slot, shred_index, recipient, glob->next_nonce, now );
     glob->next_nonce++;
     glob->metrics.send_pkt_cnt++;
-  
+
     if( FD_UNLIKELY( fd_repair_pending_sign_req_deque_full( ctx->pending_sign_req_deque ) ) ) {
         return;
     }
-    
+
     fd_repair_pending_sign_req_t * pending = fd_repair_pending_sign_req_deque_push_tail_nocopy( ctx->pending_sign_req_deque );
-    
-    /* Use the unified sign_and_send function in async mode */
+
     fd_repair_sign_and_send( ctx, &protocol, &peer->addr, pending->buf, sizeof(pending->buf), 1, nonce );
-    
-    /* Since async requests don't complete the buffer with signature,
-       we need to track the encoded length */
+
     fd_bincode_encode_ctx_t encode_ctx = { .data = pending->buf, .dataend = pending->buf + sizeof(pending->buf) };
     if( FD_UNLIKELY( fd_repair_protocol_encode( &protocol, &encode_ctx ) != FD_BINCODE_SUCCESS ) ) {
         FD_LOG_CRIT(( "Failed to encode repair message (type %#x)", protocol.discriminant ));
     }
-    
+
     ulong buflen = (ulong)encode_ctx.data - (ulong)pending->buf;
-    
+
     pending->buflen      = buflen;
     pending->sig_offset  = 4;
     pending->dst_ip_addr = peer->addr.addr;
@@ -593,12 +592,12 @@ handle_new_cluster_contact_info( fd_repair_tile_ctx_t * ctx,
     };
     int dup = fd_repair_add_active_peer( ctx->repair, &repair_peer, in_dests[i].pubkey );
     if( !dup ) {
-/* The repair process uses a Ping-Pong protocol that incurs one
-   round-trip time (RTT) for the initial repair request. To optimize
-   this, we proactively send a placeholder Repair request as soon as we
-   receive a peer's contact information for the first time, effectively
-   prepaying the RTT cost. */
-      fd_repair_send_request(ctx, ctx->stem, ctx->repair, 0, 0, 0, in_dests[i].pubkey, fd_log_wallclock());
+    /* The repair process uses a Ping-Pong protocol that incurs one
+      round-trip time (RTT) for the initial repair request. To optimize
+      this, we proactively send a placeholder Repair request as soon as we
+      receive a peer's contact information for the first time, effectively
+      prepaying the RTT cost. */
+      fd_repair_send_request_async(ctx, ctx->stem, ctx->repair, 0, 0, 0, in_dests[i].pubkey, fd_log_wallclock());
       ulong hash_src = 0xfffffUL & fd_ulong_hash( (ulong)in_dests[i].ip4_addr | ((ulong)repair_peer.port<<32) );
       FD_LOG_INFO(( "Added repair peer: pubkey %s hash_src %lu", FD_BASE58_ENC_32_ALLOCA(in_dests[i].pubkey), hash_src ));
     }
@@ -611,7 +610,6 @@ before_frag( fd_repair_tile_ctx_t * ctx,
              ulong                  in_idx,
              ulong                  seq FD_PARAM_UNUSED,
              ulong                  sig ) {
-  // FD_LOG_NOTICE(( "repair: before_frag %lu", in_idx ));
   uint in_kind = ctx->in_kind[ in_idx ];
   if( FD_LIKELY( in_kind==IN_KIND_NET ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
   return 0;
@@ -670,7 +668,7 @@ during_frag( fd_repair_tile_ctx_t * ctx,
   fd_memcpy( ctx->buffer, dcache_entry, dcache_entry_sz );
 }
 
-static ulong
+static ulong FD_FN_UNUSED
 fd_repair_send_ping( fd_repair_tile_ctx_t        * repair_tile_ctx,
                      fd_repair_t                 * glob,
                      fd_pinged_elem_t            * val,
@@ -697,7 +695,7 @@ fd_repair_send_ping( fd_repair_tile_ctx_t        * repair_tile_ctx,
   return (ulong)((uchar*)ctx.data - buf);
 }
 
-static void
+static void FD_FN_UNUSED
 fd_repair_recv_pong(fd_repair_t * glob, fd_gossip_ping_t const * pong, fd_gossip_peer_addr_t const * from) {
   fd_pinged_elem_t * val = fd_pinged_table_query(glob->pinged, from, NULL);
   if( val == NULL || !fd_pubkey_eq( &val->id, &pong->from ) )
@@ -731,10 +729,6 @@ fd_repair_recv_pong(fd_repair_t * glob, fd_gossip_ping_t const * pong, fd_gossip
   val->good = 1;
 }
 
-/* Pass a raw service request packet into the protocol.
-   src_addr is the address of the sender
-   dst_ip4_addr is the dst IPv4 address of the incoming packet (i.e. our IP) */
-
 static void
 after_frag( fd_repair_tile_ctx_t * ctx,
             ulong                  in_idx,
@@ -747,8 +741,9 @@ after_frag( fd_repair_tile_ctx_t * ctx,
 
   if( FD_UNLIKELY( ctx->skip_frag ) ) return;
 
+  ctx->stem = stem;
+
   uint in_kind = ctx->in_kind[ in_idx ];
-  // FD_LOG_INFO(( "in_idx: %lu, in_kind: %u", in_idx, in_kind ));
   if( FD_UNLIKELY( in_kind==IN_KIND_CONTACT ) ) {
     handle_new_cluster_contact_info( ctx, ctx->buffer, sz );
     return;
@@ -763,37 +758,37 @@ after_frag( fd_repair_tile_ctx_t * ctx,
     /* Nonce was packed into sig, so we need to unpack it */
     ulong response_nonce = sig >> 32;
     fd_repair_pending_sign_req_t pending;
-    
+
     /* Iterate over all pending requests, as every request sent to the
        sign tile will be returned. Since the repair_sign links are
        reliable, the incoming sign_repair fragments represent a complete
        set of the previously sent outgoing messages. However, with
-       multiple sign tiles, the responses may not arrive in order. But, 
+       multiple sign tiles, the responses may not arrive in order. But,
        we can safely process them sequentially as we encounter them in
        the deque. */
     while( !fd_repair_pending_sign_req_deque_empty( ctx->pending_sign_req_deque ) ) {
       fd_repair_pending_sign_req_t * head_req = fd_repair_pending_sign_req_deque_peek_head( ctx->pending_sign_req_deque );
-      
+
       if( head_req->nonce > response_nonce ) {
         break;
       }
-      
+
       pending = fd_repair_pending_sign_req_deque_pop_head( ctx->pending_sign_req_deque );
-      
+
       if( pending.nonce == response_nonce ) {
         fd_memcpy( pending.buf + pending.sig_offset, ctx->buffer, 64UL );
         ulong tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
-    
+
         uint src_ip4_addr = 0U;
         send_packet( ctx, stem, 1, pending.dst_ip_addr, pending.dst_port, src_ip4_addr, pending.buf, pending.buflen, tsorig );
         return;
       }
     }
-    
+
     FD_LOG_WARNING(( "No matching request found for nonce %lu", response_nonce ));
     return;
   }
-  
+
   if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) {
 
     /* Initialize the forest, which requires the root to be ready.  This
@@ -941,10 +936,6 @@ after_credit( fd_repair_tile_ctx_t * ctx,
               int *                  opt_poll_in FD_PARAM_UNUSED,
               int *                  charge_busy ) {
 
-  /* TODO: Don't charge the tile as busy if after_credit isn't actually
-     doing any work. */
-  *charge_busy = 1;
-
   if( FD_LIKELY( !fd_fec_out_empty( ctx->fec_chainer->out ) && ctx->store ) ) {
 
     fd_fec_out_t out = fd_fec_out_pop_head( ctx->fec_chainer->out );
@@ -1036,7 +1027,6 @@ after_credit( fd_repair_tile_ctx_t * ctx,
       fd_stem_publish( ctx->stem, REPLAY_OUT_IDX, sig, 0, 0, 0, 0, tspub );
       if( FD_UNLIKELY( out.slot_complete ) ) {
         fd_reasm_remove( ctx->reasm, reasm );
-        // FD_LOG_INFO(( "SLOT COMPLETE: %lu, time: %ld", out.slot, fd_log_wallclock() ));
       }
     }
     *opt_poll_in = 1;
@@ -1192,17 +1182,7 @@ unprivileged_init( fd_topo_t *      topo,
     ctx->in_links[ in_idx ].chunk0 = fd_dcache_compact_chunk0( ctx->in_links[ in_idx ].mem, link->dcache );
     ctx->in_links[ in_idx ].wmark  = fd_dcache_compact_wmark ( ctx->in_links[ in_idx ].mem, link->dcache, link->mtu );
     ctx->in_links[ in_idx ].mtu    = link->mtu;
-    
-    if( ctx->in_kind[ in_idx ] == IN_KIND_SIGN ) {
-      // fd_wksp_t * wksp = fd_wksp_containing( link->dcache );
-      // FD_LOG_NOTICE(( "repair tile: link %s[%lu] mem=%p, dcache=%p, wksp=%p, wksp_name=%s, in_idx=%u", 
-                      // link->name, link->kind_id,
-                      // (void*)ctx->in_links[ in_idx ].mem,
-                      // link->dcache,
-                      // (void*)wksp,
-                      // wksp ? fd_wksp_name( wksp ) : "NULL",
-                      // in_idx ));
-    }
+
     FD_TEST( fd_dcache_compact_is_safe( ctx->in_links[in_idx].mem, link->dcache, link->mtu, link->depth ) );
   }
 
@@ -1247,7 +1227,7 @@ unprivileged_init( fd_topo_t *      topo,
       ctx->shredcap_out_chunk0 = fd_dcache_compact_chunk0( ctx->shredcap_out_mem, link->dcache );
       ctx->shredcap_out_wmark  = fd_dcache_compact_wmark( ctx->shredcap_out_mem, link->dcache, link->mtu );
       ctx->shredcap_out_chunk  = ctx->shredcap_out_chunk0;
-    
+
     } else if( 0==strcmp( link->name, "ping_sign" ) ) {
       ctx->ping_sign_out_idx = out_idx;
       ctx->ping_sign_out_mem    = topo->workspaces[ topo->objs[ link->dcache_obj_id ].wksp_id ].wksp;

src/discof/send/fd_send_tile.c

@@ -2,6 +2,7 @@
 #include "../../disco/topo/fd_topo.h"
 #include "../../disco/keyguard/fd_keyload.h"
 #include "../../disco/keyguard/fd_keyguard.h"
+#include "../../disco/fd_txn_m_t.h"
 #include "generated/fd_send_tile_seccomp.h"
 
 

src/util/hist/fd_histf.h

@@ -7,7 +7,6 @@
 
 #include <math.h> /* FIXME: HMMM */
 #include "../log/fd_log.h"
-
 #if FD_HAS_AVX
 #include "../simd/fd_avx.h"
 #endif