disco: send transactions with metadata

Author: jherrera-jump

agave

@@ -219,10 +219,7 @@ function tango.dissector (tvb, pinfo, tree)
   if link_name == "net_netmux" or link_name == "quic_netmux" or link_name == "shred_netmux" or link_name == "net_quic" or link_name == "quic_net" or link_name == "shred_net" or link_name == "net_shred" then
     local dissector = Dissector.get("eth_withoutfcs")
     dissector:call(dcache_contents, pinfo, dcache_tree)
-  elseif link_name == "gossip_dedup" then
-    local dissector = Dissector.get("solana.tpu.udp")
-    dissector:call(dcache_contents, pinfo, dcache_tree)
-  elseif link_name == "verify_dedup" or link_name == "dedup_pack" or link_name == "dedup_resolv" or link_name == "resolv_pack" or link_name == "bundle_verif" then
+  elseif link_name == "verify_dedup" or link_name == "dedup_pack" or link_name == "dedup_resolv" or link_name == "resolv_pack" or link_name == "bundle_verif" or "quic_verify" or link_name == "gossip_verif" or link_name == "send_txns" or link_name == "gossip_dedup" then
     local dissector = Dissector.get("fd_txn_m_t")
     dissector:call(dcache_contents, pinfo, dcache_tree)
   elseif link_name == "poh_shred" then
@@ -264,9 +261,6 @@ function tango.dissector (tvb, pinfo, tree)
   elseif link_name == "poh_pack" then
     local dissector = Dissector.get("fd_became_leader_t")
     dissector:call(dcache_contents, pinfo, dcache_tree)
-  elseif link_name == "quic_verify" then
-    local dissector = Dissector.get("solana.tpu.udp")
-    dissector:call(dcache_contents, pinfo, dcache_tree)
   end
 end
 
@@ -544,6 +538,16 @@ local fd_txnm = Proto("fd_txn_m_t", "FD Transaction with Payload and Metadata")
 -- Define fields
 local f_ref_slot = ProtoField.uint64("fd_txn_m_t.reference_slot", "Reference Slot"  )
 local f_txn_t_sz = ProtoField.uint16("fd_txn_m_t.txn_t_sz",       "Size of fd_txn_t")
+local f_source_ipv4 = ProtoField.ipv4("fd_txn_m_t.source_ipv4",       "IP Address")
+
+local source_tpu_enum = {
+    [1] = "QUIC",
+    [2] = "UDP",
+    [4] = "GOSSIP",
+    [8] = "BUNDLE",
+    [16] = "SEND"
+}
+local f_source_tpu = ProtoField.uint8("fd_txn_m_t.source_tpu",       "Source TPU", base.DEC, source_tpu_enum)
 local f_payload_sz = ProtoField.uint16("fd_txn_m_t.payload_sz",       "Size of payload")
 local f_bundle_id = ProtoField.uint64("fd_txn_m_t.bundle_id",       "Bundle ID")
 local f_bundle_txn_cnt = ProtoField.uint64("fd_txn_m_t.bundle_txn_cnt",       "Bundle Transaction Count")
@@ -552,7 +556,7 @@ local f_bundle_pubkey = ProtoField.bytes("fd_txn_m_t.bundle_commission_pubkey",
 local f_alt_entry = ProtoField.bytes("fd_txn_m_t.alt_entry",       "Address Lookup Table Account Address")
 
 -- Add the fields to the protocol
-fd_txnm.fields = { f_ref_slot, f_txn_t_sz, f_payload_sz, f_bundle_id, f_bundle_txn_cnt, f_bundle_commission, f_bundle_pubkey, f_alt_entry }
+fd_txnm.fields = { f_ref_slot, f_txn_t_sz, f_source_ipv4, f_source_tpu, f_payload_sz, f_bundle_id, f_bundle_txn_cnt, f_bundle_commission, f_bundle_pubkey, f_alt_entry }
 
 function fd_txnm.dissector(buffer, pinfo, tree)
   local subtree = tree:add(fd_txnm, buffer(), "fd_txn_m_t")
@@ -563,21 +567,26 @@ function fd_txnm.dissector(buffer, pinfo, tree)
   subtree:add_le(f_ref_slot, buffer(0, 8))
   subtree:add_le(f_payload_sz, buffer(8, 2))
   subtree:add_le(f_txn_t_sz, buffer(10, 2))
+  subtree:add(f_source_ipv4, buffer(12, 4))
+  subtree:add_le(f_source_tpu, buffer(16, 1))
+
+  subtree:add_le(f_bundle_id, buffer(24, 8))
+  subtree:add_le(f_bundle_txn_cnt, buffer(32, 8))
+  subtree:add_le(f_bundle_commission, buffer(40, 1))
+  subtree:add(f_bundle_pubkey, buffer(41, 32))
+  local payload_start = 80
 
-  local payload_start = 16
-  if true then
-    subtree:add_le(f_bundle_id, buffer(16, 8))
-    subtree:add_le(f_bundle_txn_cnt, buffer(24, 8))
-    subtree:add_le(f_bundle_commission, buffer(32, 1))
-    subtree:add(f_bundle_pubkey, buffer(33, 32))
-    payload_start = 72
-  end
-  
   local payload_tree = tree:add(buffer(payload_start,payload_sz), "Solana Transaction")
   local udp_dissector = Dissector.get("solana.tpu.udp")
   udp_dissector:call(buffer(payload_start,payload_sz):tvb(), pinfo, payload_tree)
 
   local offset = payload_start + payload_sz
+
+  -- pre-dedup frags don't have fields after the payload
+  if offset == buffer:len() then
+    return
+  end
+
   -- Align to 2
   if offset % 2 == 1 then
     offset = offset+1
@@ -602,6 +611,5 @@ function fd_txnm.dissector(buffer, pinfo, tree)
 end
 
 
-
 local udp_port = DissectorTable.get("udp.port")
 udp_port:add(9001, Dissector.get("solana.tpu.udp"))

contrib/tango.lua

@@ -85,7 +85,7 @@ fd_topo_initialize( config_t * config ) {
   FOR(shred_tile_cnt)  fd_topob_link( topo, "shred_net",    "net_shred",    32768UL,                                  FD_NET_MTU,             1UL );
   FOR(quic_tile_cnt)   fd_topob_link( topo, "quic_verify",  "quic_verify",  config->tiles.verify.receive_buffer_size, FD_TPU_REASM_MTU,       config->tiles.quic.txn_reassembly_count );
   FOR(verify_tile_cnt) fd_topob_link( topo, "verify_dedup", "verify_dedup", config->tiles.verify.receive_buffer_size, FD_TPU_PARSED_MTU,      1UL );
-  /**/                 fd_topob_link( topo, "gossip_dedup", "gossip_dedup", 2048UL,                                   FD_TPU_MTU,             1UL );
+  /**/                 fd_topob_link( topo, "gossip_dedup", "gossip_dedup", 2048UL,                                   FD_TPU_RAW_MTU,             1UL );
   /* dedup_pack is large currently because pack can encounter stalls when running at very high throughput rates that would
      otherwise cause drops. */
   /**/                 fd_topob_link( topo, "dedup_resolv", "dedup_resolv", 65536UL,                                  FD_TPU_PARSED_MTU,      1UL );

src/app/fdctl/topology.c

@@ -335,7 +335,7 @@ fd_topo_initialize( config_t * config ) {
 
   FOR(exec_tile_cnt)   fd_topob_link( topo, "exec_writer",  "exec_writer",  128UL,                                    FD_EXEC_WRITER_MTU,            1UL );
 
-  /**/                 fd_topob_link( topo, "gossip_verif", "gossip_verif", config->tiles.verify.receive_buffer_size, FD_TPU_MTU,                    1UL );
+  /**/                 fd_topob_link( topo, "gossip_verif", "gossip_verif", config->tiles.verify.receive_buffer_size, FD_TPU_RAW_MTU,                1UL );
   /**/                 fd_topob_link( topo, "gossip_tower", "gossip_tower", 128UL,                                    FD_TPU_MTU,                    1UL );
   /**/                 fd_topob_link( topo, "replay_tower", "replay_tower", 128UL,                                    65536UL,                       1UL );
   /**/                 fd_topob_link( topo, "tower_replay", "replay_tower", 128UL,                                    0,                             1UL );
@@ -359,7 +359,7 @@ fd_topo_initialize( config_t * config ) {
   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 );
 
   /**/                 fd_topob_link( topo, "tower_send",   "tower_send",   65536UL,                                  sizeof(fd_txn_p_t),            1UL   );
-  /**/                 fd_topob_link( topo, "send_txns",    "send_txns",    128UL,                                    FD_TXN_MTU,                    1UL   );
+  /**/                 fd_topob_link( topo, "send_txns",    "send_txns",    128UL,                                    FD_TPU_RAW_MTU,                1UL   );
   /**/                 fd_topob_link( topo, "send_sign",    "send_sign",    128UL,                                    FD_TXN_MTU,                    1UL   );
   /**/                 fd_topob_link( topo, "sign_send",    "sign_send",    128UL,                                    64UL,                          1UL   );
 

src/app/firedancer/topology.c

@@ -446,7 +446,8 @@ fd_bundle_tile_publish_bundle_txn(
     fd_bundle_tile_t * ctx,
     void const *       txn,
     ulong              txn_sz,  /* <=FD_TXN_MTU */
-    ulong              bundle_txn_cnt
+    ulong              bundle_txn_cnt,
+    uint               source_ipv4
 ) {
   if( FD_UNLIKELY( !ctx->builder_info_avail ) ) {
     ctx->metrics.missing_builder_info_fail_cnt++; /* unreachable */
@@ -457,7 +458,9 @@ fd_bundle_tile_publish_bundle_txn(
   *txnm = (fd_txn_m_t) {
     .reference_slot = 0UL,
     .payload_sz     = (ushort)txn_sz,
-    .txn_t_sz       = 0,
+    .txn_t_sz       = 0U,
+    .source_ipv4      = source_ipv4,
+    .source_tpu       = FD_TXN_M_TPU_SOURCE_BUNDLE,
     .block_engine   = {
       .bundle_id      = ctx->bundle_seq,
       .bundle_txn_cnt = bundle_txn_cnt,
@@ -486,18 +489,21 @@ static void
 fd_bundle_tile_publish_txn(
     fd_bundle_tile_t * ctx,
     void const *       txn,
-    ulong              txn_sz  /* <=FD_TXN_MTU */
+    ulong              txn_sz,  /* <=FD_TXN_MTU */
+    uint               source_ipv4
 ) {
   fd_txn_m_t * txnm = fd_chunk_to_laddr( ctx->verify_out.mem, ctx->verify_out.chunk );
   *txnm = (fd_txn_m_t) {
     .reference_slot = 0UL,
     .payload_sz     = (ushort)txn_sz,
-    .txn_t_sz       = 0,
+    .txn_t_sz       = 0U,
+    .source_ipv4    = source_ipv4,
+    .source_tpu     = FD_TXN_M_TPU_SOURCE_BUNDLE,
     .block_engine   = {
       .bundle_id         = 0UL,
       .bundle_txn_cnt    = 1UL,
-      .commission        = 0,
-      .commission_pubkey = {0}
+      .commission        = 0U,
+      .commission_pubkey = {0U}
     },
   };
   fd_memcpy( fd_txn_m_payload( txnm ), txn, txn_sz );
@@ -549,15 +555,22 @@ fd_bundle_client_visit_pb_bundle_txn(
     return false;
   }
 
+  if( FD_UNLIKELY( packet.data.size == 0 ) ) {
+    FD_LOG_WARNING(( "Bundle server delivered an empty packet, ignoring" ));
+    return true;
+  }
+
   if( FD_UNLIKELY( packet.data.size > FD_TXN_MTU ) ) {
     FD_LOG_WARNING(( "Bundle server delivered an oversize transaction, ignoring" ));
     return true;
   }
 
+  uint _ip4; uint ip4 = fd_uint_if( packet.has_meta, fd_cstr_to_ip4_addr( packet.meta.addr, &_ip4 ) ? _ip4 : 0U, 0U );
   fd_bundle_tile_publish_bundle_txn(
       ctx,
       packet.data.bytes, packet.data.size,
-      ctx->bundle_txn_cnt
+      ctx->bundle_txn_cnt,
+      ip4
   );
 
   return true;
@@ -672,12 +685,19 @@ fd_bundle_client_visit_pb_packet(
     return false;
   }
 
+  if( FD_UNLIKELY( packet.data.size == 0 ) ) {
+    FD_LOG_WARNING(( "Bundle server delivered an empty packet, ignoring" ));
+    return true;
+  }
+
   if( FD_UNLIKELY( packet.data.size > FD_TXN_MTU ) ) {
     FD_LOG_WARNING(( "Bundle server delivered an oversize transaction, ignoring" ));
     return true;
   }
 
-  fd_bundle_tile_publish_txn( ctx, packet.data.bytes, packet.data.size );
+
+  uint _ip4; uint ip4 = fd_uint_if( packet.has_meta, fd_cstr_to_ip4_addr( packet.meta.addr, &_ip4 ) ? _ip4 : 0U, 0U );
+  fd_bundle_tile_publish_txn( ctx, packet.data.bytes, packet.data.size, ip4 );
   ctx->metrics.packet_received_cnt++;
 
   return true;

src/disco/bundle/fd_bundle_client.c

@@ -22,6 +22,15 @@ test_bundle_rx( fd_wksp_t * wksp ) {
   test_bundle_env_t env[1]; test_bundle_env_create( env, wksp );
   fd_bundle_tile_t * state = env->state;
 
+  /* A SubscribePacketsResponse message with 2 packets included. The
+     first packet is 1 byte { 0x48 }, the second packet is 2 bytes
+     {0x48, 0x48}.
+
+     message SubscribePacketsResponse {
+      shared.Header header = 1;
+      packet.PacketBatch batch = 2;
+    }
+  */
   static uchar subscribe_packets_msg[] = {
     0x12, 0x13, 0x0a, 0x07, 0x0a, 0x01, 0x48, 0x12,
     0x02, 0x08, 0x01, 0x0a, 0x08, 0x0a, 0x02, 0x48,
@@ -39,9 +48,11 @@ test_bundle_rx( fd_wksp_t * wksp ) {
     env->out_mcache[ i ].tspub  = 0U;
   }
 
+  const ulong packet1_sz = 1UL;
+  const ulong packet2_sz = 2UL;
   fd_frag_meta_t expected[2] = {
-    { .seq=0UL, .sig=0UL, .chunk=0, .sz=sizeof(fd_txn_m_t)+8, .ctl=0 },
-    { .seq=1UL, .sig=0UL, .chunk=2, .sz=sizeof(fd_txn_m_t)+8, .ctl=0 }
+    { .seq=0UL, .sig=0UL, .chunk=0, .sz=sizeof(fd_txn_m_t)+packet1_sz, .ctl=0 },
+    { .seq=1UL, .sig=0UL, .chunk=2, .sz=sizeof(fd_txn_m_t)+packet2_sz, .ctl=0 }
   };
   FD_TEST( fd_memeq( env->out_mcache, expected, 2*sizeof(fd_frag_meta_t) ) );
 

src/disco/bundle/test_bundle_client.c

@@ -111,16 +111,18 @@ during_frag( fd_dedup_ctx_t * ctx,
   uchar * dst = (uchar *)fd_chunk_to_laddr( ctx->out_mem, ctx->out_chunk );
 
   if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_GOSSIP ) ) {
-    if( FD_UNLIKELY( sz>FD_TPU_MTU ) ) FD_LOG_ERR(( "received a gossip transaction that was too large" ));
+    if( FD_UNLIKELY( sz>FD_TPU_RAW_MTU ) ) FD_LOG_ERR(( "received a gossip transaction that was too large" ));
+    fd_memcpy( dst, src, sz );
 
-    fd_txn_m_t * txnm = (fd_txn_m_t *)dst;
-    txnm->payload_sz = (ushort)sz;
-    fd_memcpy( fd_txn_m_payload( txnm ), src, sz );
-    txnm->block_engine.bundle_id = 0UL;
+    fd_txn_m_t const * txnm = (fd_txn_m_t const *)dst;
+    if( FD_UNLIKELY( txnm->payload_sz>FD_TPU_MTU ) ) {
+      FD_LOG_ERR(( "vote txn payload size %hu exceeds max %lu", txnm->payload_sz, FD_TPU_MTU ));
+    }
   } else if( FD_UNLIKELY( ctx->in_kind[ in_idx ]==IN_KIND_EXECUTED_TXN ) ) {
+    if( FD_UNLIKELY( sz!=FD_TXN_SIGNATURE_SZ ) ) FD_LOG_ERR(( "received an executed transaction signature message with the wrong size %lu", sz ));
     /* Executed txns just have their signature inserted into the tcache
        so we can dedup them easily. */
-    ulong ha_dedup_tag = fd_hash( ctx->hashmap_seed, src+64UL, 64UL );
+    ulong ha_dedup_tag = fd_hash( ctx->hashmap_seed, src+FD_TXN_SIGNATURE_SZ, FD_TXN_SIGNATURE_SZ );
     int _is_dup;
     FD_TCACHE_INSERT( _is_dup, *ctx->tcache_sync, ctx->tcache_ring, ctx->tcache_depth, ctx->tcache_map, ctx->tcache_map_cnt, ha_dedup_tag );
     (void)_is_dup;

src/disco/dedup/fd_dedup_tile.c

@@ -75,7 +75,7 @@ FD_PROTOTYPES_BEGIN
 FD_FN_CONST static inline ulong
 fd_disco_netmux_sig( uint   hash_ip_addr,
                      ushort hash_port,
-                     uint   dst_ip_addr,
+                     uint   ip_addr,
                      ulong  proto,
                      ulong  hdr_sz ) {
   /* The size of an Ethernet header is 14+4k bytes, where 0<=k<=3 (?) is
@@ -86,12 +86,12 @@ fd_disco_netmux_sig( uint   hash_ip_addr,
      size by just storing i. */
   ulong hdr_sz_i = ((hdr_sz - 42UL)>>2)&0xFUL;
   ulong hash     = 0xfffffUL & fd_ulong_hash( (ulong)hash_ip_addr | ((ulong)hash_port<<32) );
-  return (hash<<44) | ((hdr_sz_i&0xFUL)<<40UL) | ((proto&0xFFUL)<<32UL) | ((ulong)dst_ip_addr);
+  return (hash<<44) | ((hdr_sz_i&0xFUL)<<40UL) | ((proto&0xFFUL)<<32UL) | ((ulong)ip_addr);
 }
 
-FD_FN_CONST static inline ulong fd_disco_netmux_sig_hash  ( ulong sig ) { return (sig>>44UL); }
-FD_FN_CONST static inline ulong fd_disco_netmux_sig_proto ( ulong sig ) { return (sig>>32UL) & 0xFFUL; }
-FD_FN_CONST static inline uint  fd_disco_netmux_sig_dst_ip( ulong sig ) { return (uint)(sig & 0xFFFFFFFFUL); }
+FD_FN_CONST static inline ulong fd_disco_netmux_sig_hash ( ulong sig ) { return (sig>>44UL); }
+FD_FN_CONST static inline ulong fd_disco_netmux_sig_proto( ulong sig ) { return (sig>>32UL) & 0xFFUL; }
+FD_FN_CONST static inline uint  fd_disco_netmux_sig_ip   ( ulong sig ) { return (uint)(sig & 0xFFFFFFFFUL); }
 
 /* fd_disco_netmux_sig_hdr_sz extracts the total size of the Ethernet,
    IP, and UDP headers from the netmux signature field.  The UDP payload

src/disco/fd_disco_base.h

@@ -6,6 +6,12 @@
 
 #include "../ballet/txn/fd_txn.h"
 
+#define FD_TXN_M_TPU_SOURCE_QUIC   (1UL)
+#define FD_TXN_M_TPU_SOURCE_UDP    (2UL)
+#define FD_TXN_M_TPU_SOURCE_GOSSIP (3UL)
+#define FD_TXN_M_TPU_SOURCE_BUNDLE (4UL)
+#define FD_TXN_M_TPU_SOURCE_SEND   (5UL)
+
 struct fd_txn_m {
   /* The computed slot that this transaction is referencing, aka. the
      slot number of the reference_blockhash.  If it could not be
@@ -18,7 +24,12 @@ struct fd_txn_m {
      so we just store this redundantly. */
   ushort    txn_t_sz;
 
-  /* 4 bytes of padding here */
+  /* Source tpu and IP address for this transaction.  Note that
+     source_ipv4 is in big endian. */
+  uint     source_ipv4;
+  uchar    source_tpu;
+
+  /* 7 bytes of padding here */
 
   struct {
     /* If the transaction is part of a bundle, the bundle_id will be
@@ -44,12 +55,14 @@ struct fd_txn_m {
     ulong bundle_txn_cnt;
     uchar commission;
     uchar commission_pubkey[ 32 ];
-  } block_engine;
 
-  /* alignof is 8, so 7 bytes of padding here */
+    /* alignof is 8, so 7 bytes of padding here */
+
+  } block_engine;
 
   /* There are three additional fields at the end here, which are
-     variable length and not included in the size of this struct.
+     variable length and not included in the size of this struct. txn_t
+     and alut are only found in frags after the verify step.
   uchar          payload[ ]
   fd_txn_t       txn_t[ ]
   fd_acct_addr_t alut[ ] */
@@ -100,15 +113,21 @@ fd_txn_m_realized_footprint( fd_txn_m_t const * txnm,
                              int                include_txn_t,
                              int                include_alut ) {
   if( FD_LIKELY( include_txn_t ) ) {
-    return fd_txn_m_footprint( txnm->payload_sz,
-                               fd_txn_m_txn_t_const( txnm )->instr_cnt,
-                               fd_txn_m_txn_t_const( txnm )->addr_table_lookup_cnt,
-                               include_alut ? fd_txn_m_txn_t_const( txnm )->addr_table_adtl_cnt : 0UL );
+    ulong l = FD_LAYOUT_INIT;
+    l = FD_LAYOUT_APPEND( l, alignof(fd_txn_m_t),     sizeof(fd_txn_m_t) );
+    l = FD_LAYOUT_APPEND( l, 1UL,                     txnm->payload_sz );
+    l = FD_LAYOUT_APPEND( l, fd_txn_align(),          fd_txn_footprint( fd_txn_m_txn_t_const( txnm )->instr_cnt, fd_txn_m_txn_t_const( txnm )->addr_table_lookup_cnt ) );
+    l = FD_LAYOUT_APPEND( l, alignof(fd_acct_addr_t), fd_uchar_if(include_alut, fd_txn_m_txn_t_const( txnm )->addr_table_adtl_cnt, 0U)*sizeof(fd_acct_addr_t) );
+
+    /* FD_LAYOUT_FINI is not included since the _realized_ footprint
+       should not include the extra padding typically added after the
+       last struct used to align the entire footprint. */
+    return l;
   } else {
     ulong l = FD_LAYOUT_INIT;
     l = FD_LAYOUT_APPEND( l, alignof(fd_txn_m_t), sizeof(fd_txn_m_t) );
-    l = FD_LAYOUT_APPEND( l, 1UL,                 txnm->payload_sz   );
-    return FD_LAYOUT_FINI( l, fd_txn_m_align() );
+    l = FD_LAYOUT_APPEND( l, 1UL, txnm->payload_sz );
+    return l;
   }
 }
 

src/disco/fd_txn_m_t.h

@@ -408,7 +408,7 @@ poll_rx_socket( fd_sock_tile_t *    ctx,
 
     ctx->metrics.rx_pkt_cnt++;
     ulong chunk = fd_laddr_to_chunk( base, eth_hdr );
-    ulong sig   = fd_disco_netmux_sig( sa->sin_addr.s_addr, fd_ushort_bswap( sa->sin_port ), 0U, proto, hdr_sz );
+    ulong sig   = fd_disco_netmux_sig( sa->sin_addr.s_addr, fd_ushort_bswap( sa->sin_port ), sa->sin_addr.s_addr, proto, hdr_sz );
     ulong tspub = fd_frag_meta_ts_comp( ts );
 
     /* default for repair intake is to send to [shreds] to shred tile.