gui: add support for source ip / tpu

Author: jherrera-jump

book/api/websocket.md

@@ -1276,16 +1276,27 @@ new validator identity.
         "transactions": {
             "start_timestamp_nanos": "1739657041688346791",
             "target_end_timestamp_nanos": "1739657042088346880",
-            "txn_start_timestamps_nanos": ["1739657041706960598"],
-            "txn_stop_timestamps_nanos": ["1739657041707477011"],
-            "txn_compute_units_requested": [0],
-            "txn_compute_units_estimated": [3428],
-            "txn_compute_units_rebated": [0],
-            "txn_micro_lamports_per_cu": ["0"],
+            "txn_arrival_timestamps_nanos": ["1754409729593613895"],
+            "txn_bank_idx": [0],
+            "txn_compute_units_consumed": [3428],
+            "txn_compute_units_requested": [3428],
+            "txn_end_timstamps_nanos": ["1754409729594477657"],
             "txn_error_code": [0],
             "txn_from_bundle": [false],
             "txn_is_simple_vote": [true],
-            "txn_bank_idx": [0]
+            "txn_landed": [true],
+            "txn_load_end_timstamps_nanos": ["1754409729594455631"],
+            "txn_mb_end_timestamps_nanos": ["1754409729594625003"],
+            "txn_mb_start_timestamps_nanos": ["1754409729594431327"],
+            "txn_microblock_id": [0],
+            "txn_preload_end_timstamps_nanos": ["1754409729594432846"],
+            "txn_priority_fee": ["0"],
+            "txn_signature": ["2BfWBnhTP1ZZwFZutwThj5VT1hX71X9otbgFr21W2XJfcppXakbPCvJ2eCh8eBcS74Lfjar5AuowuppAjsEceSuW"],
+            "txn_start_timstamps_nanos": ["1754409729594451074"],
+            "txn_transaction_fee": [0],
+            "txn_tips": ["0"],
+            "txn_source_ipv4": ["123.123.123.123"],
+            "txn_source_tpu": ["gossip"],
         }
     }
 }
@@ -1378,9 +1389,22 @@ new validator identity.
 | txn_error_code                    | `number[]`  | `txn_error_code[i]` is the error code that explains the failure for the `i`-th transaction in the slot. See below for more details |
 | txn_from_bundle                   | `boolean[]` | `txn_from_bundle[i]` is `true` if the `i`-th transaction in the slot came from a bundle and `false` otherwise.  A bundle is a microblock with 1-5 transactions that atomically fail or succeed. It is sent to the validator from a compatible block engine (e.g. jito) that can additionally collect MEV rewards that are distributed to stakers (i.e. tips) |
 | txn_is_simple_vote                | `boolean[]` | `txn_is_simple_vote[i]` is `true` if the `i`-th transaction in the slot is a simple vote and `false` otherwise |
+| txn_landed                        | `boolean[]` | `txn_landed[i]` is `true` if the `i`-th transaction in the slot was included in the produced block. A scheduled transaction may not be included in the block for any number of reasons (e.g. a failed bundle, a duplicate transaction, invalid fee-payer) |
 | txn_bank_idx                      | `number[]`  | `txn_bank_idx[i]` is the index of the bank tile that executed the `i`-th transaction in the slot |
 | txn_microblock_id                 | `string[]`  | `txn_microblock_id[i]` is the index of the microblock for the `i`-th transaction in the slot.  Microblocks are collections of 1+ transactions.  All of the transactions from a bundle share the same microblock. Microblock ids are monotonically increasing in the order they appear in the block and start at 0 for each slot |
 | txn_signature                     | `string[]`  | `txn_signature[i]` is the base58 signature of the `i`-th transaction in the slot |
+| txn_source_ipv4                   | `number[]`  | `txn_signature[i]` is the source ipv4 address for the `i`-th transaction in the slot |
+| txn_source_tpu                    | `string[]`  | `txn_signature[i]` is the transaction processing unit (TPU) which handled the `i`-th transaction in the slot |
+
+The source tpu for a transaction can be one of the following
+
+| TPU    | Description |
+|--------|-------------|
+| quic   | the primary ingress tpu for user transactions.  Utilizes the quic protocol to recieve packets |
+| udp    | ingress transactions recieved as simple UDP packets |
+| gossip | vote transactions recieved from the gossip network |
+| bundle | bundle transacionts recieved by the bundle tile from a block builder.  Utilizes a grpc connection to recieve packets |
+| send   | vote transactions procuded by this validator recieved from the send tile. These transactions are meant for the active cluster leader |
 
 These are the possible error codes that might be included in `txn_error_code` and their meanings.
 

src/disco/bundle/fd_bundle_client.c

@@ -565,7 +565,7 @@ fd_bundle_client_visit_pb_bundle_txn(
     return true;
   }
 
-  uint _ip4; uint ip4 = fd_uint_if( packet.has_meta, fd_cstr_to_ip4_addr( packet.meta.addr, &_ip4 ) ? _ip4 : 0U, 0U );
+  uint _ip4; uint ip4 = fd_uint_if( packet.has_meta, fd_cstr_to_ip4_addr( packet.meta.addr, &_ip4 ) ? _ip4 : ctx->server_ip4_addr, ctx->server_ip4_addr );
   fd_bundle_tile_publish_bundle_txn(
       ctx,
       packet.data.bytes, packet.data.size,

src/disco/gui/fd_gui.c

@@ -1805,6 +1805,8 @@ fd_gui_microblock_execution_begin( fd_gui_t *   gui,
     txn_entry->priority_fee                = priority_rewards;
     txn_entry->transaction_fee             = sig_rewards;
     txn_entry->timestamp_delta_start_nanos = (int)((double)(tickcount - slot->txs.reference_ticks) / fd_tempo_tick_per_ns( NULL ));
+    txn_entry->source_ipv4                 = txn_payload->source_ipv4;
+    txn_entry->source_tpu                  = txn_payload->source_tpu;
     txn_entry->microblock_idx              = microblock_idx;
     txn_entry->flags                      |= (uchar)FD_GUI_TXN_FLAGS_STARTED;
     txn_entry->flags                      &= (uchar)(~(uchar)(FD_GUI_TXN_FLAGS_IS_SIMPLE_VOTE | FD_GUI_TXN_FLAGS_FROM_BUNDLE));

src/disco/gui/fd_gui.h

@@ -266,7 +266,9 @@ struct __attribute__((packed)) fd_gui_txn {
   uchar txn_end_pct;
   uchar txn_preload_end_pct;
   uchar flags; /* assigned with the FD_GUI_TXN_FLAGS_* macros */
-  uint microblock_idx;
+  uchar source_tpu; /* FD_TXN_M_TPU_SOURCE_* */
+  uint  source_ipv4;
+  uint  microblock_idx;
 };
 
 

src/disco/gui/fd_gui_printf.c

@@ -5,6 +5,7 @@
 
 #include "../../waltz/http/fd_http_server_private.h"
 #include "../../ballet/utf8/fd_utf8.h"
+#include "../../disco/fd_txn_m_t.h"
 
 #ifdef __has_include
 #if __has_include("../../app/fdctl/version.h")
@@ -1324,6 +1325,40 @@ fd_gui_printf_slot_transactions_request( fd_gui_t * gui,
           jsonp_open_array( gui, "txn_tips" );
             for( ulong i=0UL; i<txn_cnt; i++) jsonp_ulong_as_str( gui, NULL, gui->txs[ (slot->txs.start_offset + i)%FD_GUI_TXN_HISTORY_SZ ]->tips );
           jsonp_close_array( gui );
+          jsonp_open_array( gui, "txn_source_ipv4" );
+            for( ulong i=0UL; i<txn_cnt; i++) {
+              char addr[ 64 ];
+              fd_cstr_printf_check( addr, sizeof(addr), NULL, FD_IP4_ADDR_FMT, FD_IP4_ADDR_FMT_ARGS( gui->txs[ (slot->txs.start_offset + i)%FD_GUI_TXN_HISTORY_SZ ]->source_ipv4 ) );
+              jsonp_string( gui, NULL, addr );
+            }
+          jsonp_close_array( gui );
+          jsonp_open_array( gui, "txn_source_tpu" );
+            for( ulong i=0UL; i<txn_cnt; i++) {
+              switch ( gui->txs[ (slot->txs.start_offset + i)%FD_GUI_TXN_HISTORY_SZ ]->source_tpu ) {
+                case FD_TXN_M_TPU_SOURCE_QUIC: {
+                  jsonp_string( gui, NULL, "quic");
+                  break;
+                }
+                case FD_TXN_M_TPU_SOURCE_UDP   : {
+                  jsonp_string( gui, NULL, "udp");
+                  break;
+                }
+                case FD_TXN_M_TPU_SOURCE_GOSSIP: {
+                  jsonp_string( gui, NULL, "gossip");
+                  break;
+                }
+                case FD_TXN_M_TPU_SOURCE_BUNDLE: {
+                  jsonp_string( gui, NULL, "bundle");
+                  break;
+                }
+                case FD_TXN_M_TPU_SOURCE_SEND  : {
+                  jsonp_string( gui, NULL, "send");
+                  break;
+                }
+                default: FD_LOG_ERR(("unknown tpu"));
+              }
+            }
+          jsonp_close_array( gui );
           jsonp_open_array( gui, "txn_microblock_id" );
             for( ulong i=0UL; i<txn_cnt; i++) jsonp_ulong( gui, NULL, gui->txs[ (slot->txs.start_offset + i)%FD_GUI_TXN_HISTORY_SZ ]->microblock_idx );
           jsonp_close_array( gui );

src/disco/pack/fd_microblock.h

@@ -71,6 +71,11 @@ struct __attribute__((aligned(64))) fd_txn_p {
   };
   /* The time that the transaction arrived to the pack tile in ticks. Set by pack and intended to be read from a transaction on a pack->bank link. */
   long scheduler_arrival_time_nanos;
+
+  /* Source ipv4 address and tpu pipeline for this transaction. TPU is one of FD_TXN_M_TPU_SOURCE_* */
+  uchar source_tpu;
+  uint  source_ipv4;
+
   /* Populated by pack, bank.  A combination of the bitfields
      FD_TXN_P_FLAGS_* defined above.  The bank sets the high byte with
      the transaction result code. */

src/disco/pack/fd_pack.c

@@ -1982,8 +1982,10 @@ fd_pack_schedule_impl( fd_pack_t          * pack,
          copy anyway. */
       FD_STATIC_ASSERT( offsetof(fd_txn_p_t, payload_sz     )+sizeof(((fd_txn_p_t*)NULL)->payload_sz    )<=1280UL, nt_memcpy );
       FD_STATIC_ASSERT( offsetof(fd_txn_p_t, blockhash_slot )+sizeof(((fd_txn_p_t*)NULL)->blockhash_slot)<=1280UL, nt_memcpy );
-      FD_STATIC_ASSERT( offsetof(fd_txn_p_t, flags          )+sizeof(((fd_txn_p_t*)NULL)->flags         )<=1280UL, nt_memcpy );
       FD_STATIC_ASSERT( offsetof(fd_txn_p_t, scheduler_arrival_time_nanos )+sizeof(((fd_txn_p_t*)NULL)->scheduler_arrival_time_nanos )<=1280UL, nt_memcpy );
+      FD_STATIC_ASSERT( offsetof(fd_txn_p_t, source_tpu     )+sizeof(((fd_txn_p_t*)NULL)->source_tpu    )<=1280UL, nt_memcpy );
+      FD_STATIC_ASSERT( offsetof(fd_txn_p_t, source_ipv4    )+sizeof(((fd_txn_p_t*)NULL)->source_ipv4   )<=1280UL, nt_memcpy );
+      FD_STATIC_ASSERT( offsetof(fd_txn_p_t, flags          )+sizeof(((fd_txn_p_t*)NULL)->flags         )<=1280UL, nt_memcpy );
       FD_STATIC_ASSERT( offsetof(fd_txn_p_t, _              )                                            <=1280UL, nt_memcpy );
       const ulong offset_into_txn = 1280UL - offsetof(fd_txn_p_t, _ );
       fd_memcpy( offset_into_txn+(uchar *)TXN(out), offset_into_txn+(uchar const *)txn,
@@ -1995,8 +1997,10 @@ fd_pack_schedule_impl( fd_pack_t          * pack,
       out->payload_sz                      = cur->txn->payload_sz;
       out->pack_cu.requested_exec_plus_acct_data_cus = cur->txn->pack_cu.requested_exec_plus_acct_data_cus;
       out->pack_cu.non_execution_cus       = cur->txn->pack_cu.non_execution_cus;
-      out->flags                           = cur->txn->flags;
       out->scheduler_arrival_time_nanos    = cur->txn->scheduler_arrival_time_nanos;
+      out->source_tpu                      = cur->txn->source_tpu;
+      out->source_ipv4                     = cur->txn->source_ipv4;
+      out->flags                           = cur->txn->flags;
     }
     out++;
 
@@ -2408,8 +2412,10 @@ fd_pack_try_schedule_bundle( fd_pack_t  * pack,
     out->payload_sz                      = cur->txn->payload_sz;
     out->pack_cu.requested_exec_plus_acct_data_cus = cur->txn->pack_cu.requested_exec_plus_acct_data_cus;
     out->pack_cu.non_execution_cus       = cur->txn->pack_cu.non_execution_cus;
-    out->flags                           = cur->txn->flags;
     out->scheduler_arrival_time_nanos    = cur->txn->scheduler_arrival_time_nanos;
+    out->source_tpu                      = cur->txn->source_tpu;
+    out->source_ipv4                     = cur->txn->source_ipv4;
+    out->flags                           = cur->txn->flags;
     out++;
 
     pack->cumulative_block_cost += cur->compute_est;

src/disco/pack/fd_pack_tile.c

@@ -818,8 +818,10 @@ during_frag( fd_pack_ctx_t * ctx,
       FD_LOG_ERR(( "chunk %lu %lu corrupt, not in range [%lu,%lu]", chunk, sz, ctx->in[ in_idx ].chunk0, ctx->in[ in_idx ].wmark ));
 
     fd_txn_m_t * txnm = (fd_txn_m_t *)dcache_entry;
-    ulong payload_sz = txnm->payload_sz;
-    ulong txn_t_sz   = txnm->txn_t_sz;
+    ulong payload_sz  = txnm->payload_sz;
+    ulong txn_t_sz    = txnm->txn_t_sz;
+    uint  source_ipv4 = txnm->source_ipv4;
+    uchar source_tpu  = txnm->source_tpu;
     FD_TEST( payload_sz<=FD_TPU_MTU    );
     FD_TEST( txn_t_sz  <=FD_TXN_MAX_SZ );
     fd_txn_t * txn  = fd_txn_m_txn_t( txnm );
@@ -898,8 +900,10 @@ during_frag( fd_pack_ctx_t * ctx,
     fd_memcpy( ctx->cur_spot->txnp->payload, fd_txn_m_payload( txnm ), payload_sz    );
     fd_memcpy( TXN(ctx->cur_spot->txnp),     txn,                      txn_t_sz      );
     fd_memcpy( ctx->cur_spot->alt_accts,     fd_txn_m_alut( txnm ),    addr_table_sz );
-    ctx->cur_spot->txnp->payload_sz = payload_sz;
     ctx->cur_spot->txnp->scheduler_arrival_time_nanos = ctx->approx_wallclock_ns + (long)((double)(fd_tickcount() - ctx->approx_tickcount) / ctx->ticks_per_ns);
+    ctx->cur_spot->txnp->payload_sz  = payload_sz;
+    ctx->cur_spot->txnp->source_ipv4 = source_ipv4;
+    ctx->cur_spot->txnp->source_tpu  = source_tpu;
 
     break;
   }