kbhargava-jump · Oct 26, 2026 · Oct 21, 2026
diff --git a/src/app/firedancer/config/default.toml b/src/app/firedancer/config/default.toml
@@ -2267,6 +2267,15 @@ telemetry = true
         # development purposes.
         report_runtime_diffs = false
 
+        # Maximum upload speed for telemetry data in megabits per second (Mbps).
+        # The steady state of collection sits well under the default here, and
+        # it is set higher to allow draining a backlog of data in the event of
+        # server outage.
+        #
+        # It is not recommended to change this value except in test or
+        # development environments.
+        max_upload_rate_mbps = 32
+
     # The following options relate to the 'firedance-dev pktgen' tool
     # only.  pktgen tests net tile transmit throughput by generating a
     # flood of non-routable Ethernet frames.  Only use 'firedancer-dev

diff --git a/src/app/firedancer/topology.c b/src/app/firedancer/topology.c
@@ -1378,6 +1378,7 @@ fd_topo_configure_tile( fd_topo_tile_t * tile,
 
     fd_cstr_ncpy( tile->event.identity_key_path, config->paths.identity_key, sizeof(tile->event.identity_key_path) );
     fd_cstr_ncpy( tile->event.url, config->tiles.event.url, sizeof(tile->event.url) );
+    tile->event.tx_rate_mbps = config->development.event.max_upload_rate_mbps;
     fd_cstr_ncpy( tile->event.action, config->action, sizeof(tile->event.action) );
     FD_TEST( fd_cstr_printf_check( tile->event.accounts_path,  sizeof(tile->event.accounts_path),  NULL, "%s", config->paths.accounts  ) );
     FD_TEST( fd_cstr_printf_check( tile->event.snapshots_path, sizeof(tile->event.snapshots_path), NULL, "%s", config->paths.snapshots ) );

diff --git a/src/app/shared/fd_config.c b/src/app/shared/fd_config.c
@@ -549,6 +549,8 @@ fd_config_validate( fd_config_t const * config ) {
     FD_LOG_ERR(( "invalid [development.bench.max_cost_per_block]: must be 0 or in [%lu,%u)", FD_PACK_MAX_COST_PER_BLOCK_UPPER_BOUND, UINT_MAX ));
   if( FD_UNLIKELY( bench_shreds && (bench_shreds%FD_SHRED_BLK_MAX || bench_shreds>FD_SHRED_BLK_MAX_RAISED) ) )
     FD_LOG_ERR(( "invalid [development.bench.max_shreds_per_block]: must be 0 or a multiple of %lu up to %lu", (ulong)FD_SHRED_BLK_MAX, FD_SHRED_BLK_MAX_RAISED ));
+  if( FD_UNLIKELY( config->development.event.max_upload_rate_mbps>(ulong)LONG_MAX/125000UL ) )
+    FD_LOG_ERR(( "invalid [development.event.max_upload_rate_mbps]: must be at most %lu", (ulong)LONG_MAX/125000UL ));
 
   if( 0==strcmp( config->net.provider, "xdp" ) ) {
     if( 0!=strcmp( config->net.xdp.xdp_mode, "skb"     ) &&

diff --git a/src/app/shared/fd_config.h b/src/app/shared/fd_config.h
@@ -322,9 +322,10 @@ struct fd_config {
     } bundle;
 
     struct {
-      int report_shreds;
-      int report_transactions;
-      int report_runtime_diffs;
+      int   report_shreds;
+      int   report_transactions;
+      int   report_runtime_diffs;
+      ulong max_upload_rate_mbps;
     } event;
 
     struct {

diff --git a/src/app/shared/fd_config_json.c b/src/app/shared/fd_config_json.c
@@ -12,7 +12,7 @@
    or knowingly skipped) before the constant is bumped.  String keys of
    the user's own file are separately forced through the classification
    lists below. */
-FD_STATIC_ASSERT( sizeof(fd_config_t)==24458368UL, update_fd_config_to_json_for_the_layout_change );
+FD_STATIC_ASSERT( sizeof(fd_config_t)==24458384UL, update_fd_config_to_json_for_the_layout_change );
 
 #define REDACTED "[redacted]"
 
@@ -505,7 +505,8 @@ fd_config_to_json( fd_config_t const * config,
     jw_obj_open( &w, "event" );
       jw_bool( &w, "report_shreds",            config->development.event.report_shreds );
       jw_bool( &w, "report_transactions",      config->development.event.report_transactions );
-      jw_bool( &w, "report_runtime_diffs",      config->development.event.report_runtime_diffs );
+      jw_bool ( &w, "report_runtime_diffs",      config->development.event.report_runtime_diffs );
+      jw_ulong( &w, "max_upload_rate_mbps",      config->development.event.max_upload_rate_mbps );
     jw_obj_close( &w );
     jw_obj_open( &w, "pktgen" );
       jw_str( &w, "affinity",    config->development.pktgen.affinity );

diff --git a/src/app/shared/fd_config_parse.c b/src/app/shared/fd_config_parse.c
@@ -278,6 +278,7 @@ fd_config_extract_pod( uchar *       pod,
   CFG_POP      ( bool,   development.event.report_shreds                  );
   CFG_POP      ( bool,   development.event.report_transactions            );
   CFG_POP      ( bool,   development.event.report_runtime_diffs           );
+  CFG_POP      ( ulong,  development.event.max_upload_rate_mbps           );
 
   CFG_POP      ( cstr,   development.pktgen.affinity                      );
   CFG_POP      ( cstr,   development.pktgen.fake_dst_ip                   );

diff --git a/src/disco/events/fd_event_client.c b/src/disco/events/fd_event_client.c
@@ -41,9 +41,9 @@
 #define FD_EVENT_CLIENT_HEARTBEAT_NANOS (15L*(long)1e9)
 #define FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS (60L*(long)1e9)
 
-#define FD_EVENT_CLIENT_TX_RATE_BPS (4L*1000L*1000L)
 #define FD_EVENT_CLIENT_TX_BURST (256L<<10)
 #define FD_EVENT_CLIENT_CREDIT_STALL_NANOS (20L*(long)1e9)
+#define FD_EVENT_CLIENT_TX_DRAIN_MAX (1024UL)
 
 /* The edge proxy fails a request after 100 MB of body.  Rotate to a new
    StreamEvents request on the same connection before that.  The server
@@ -102,6 +102,7 @@ struct fd_event_client {
 
   long  tx_tokens;
   long  tx_tokens_ns;
+  long  tx_byte_rate;
   long  stall_since;
   ulong stall_rem;
   ulong stall_events_sent;
@@ -240,7 +241,8 @@ fd_event_client_new( void *                     shmem,
                      ulong                      machine_id,
                      ulong                      buf_max,
                      int                        use_tls,
-                     fd_x509_ca_store_t const * ca_store ) {
+                     fd_x509_ca_store_t const * ca_store,
+                     ulong                      tx_rate_mbps ) {
   if( FD_UNLIKELY( !shmem ) ) {
     FD_LOG_WARNING(( "NULL shmem" ));
     return NULL;
@@ -319,6 +321,7 @@ fd_event_client_new( void *                     shmem,
   client->now                  = 0L;
   client->last_stream_send_ns  = 0L;
   client->last_response_ns     = 0L;
+  client->tx_byte_rate         = (long)( ( tx_rate_mbps ? tx_rate_mbps : FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT )*125000UL );
   client->tx_tokens            = FD_EVENT_CLIENT_TX_BURST;
   client->tx_tokens_ns         = 0L;
   client->stall_since          = 0L;
@@ -881,11 +884,13 @@ fd_event_client_grpc_ping_ack( void * app_ctx ) {
 static long
 pace_refill( fd_event_client_t * client,
              long                now ) {
-  long ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS;
-  long dt          = fd_long_max( now-client->tx_tokens_ns, 0L );
-  long grant       = fd_long_min( dt/ns_per_byte, FD_EVENT_CLIENT_TX_BURST-client->tx_tokens );
+  ulong   rate  = (ulong)client->tx_byte_rate;
+  long    dt    = fd_long_max( now-client->tx_tokens_ns, 0L );
+  uint128 avail = (uint128)(ulong)dt*rate/1000000000UL;
+  ulong   room  = (ulong)( FD_EVENT_CLIENT_TX_BURST-client->tx_tokens );
+  long    grant = (long)( avail<room ? (ulong)avail : room );
   client->tx_tokens   += grant;
-  client->tx_tokens_ns = client->tx_tokens==FD_EVENT_CLIENT_TX_BURST ? now : client->tx_tokens_ns+grant*ns_per_byte;
+  client->tx_tokens_ns = client->tx_tokens==FD_EVENT_CLIENT_TX_BURST ? now : client->tx_tokens_ns+(long)( ( (uint128)(ulong)grant*1000000000UL+rate-1UL )/rate );
   return client->tx_tokens;
 }
 
@@ -1076,11 +1081,12 @@ fd_event_client_next_deadline( fd_event_client_t const * client,
   if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING ) ) deadline = fd_long_min( deadline, client->auth_deadline );
   if( FD_UNLIKELY( client->auth_send_pending && !fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) deadline = now;
   if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED && client->event_stream ) ) {
-    deadline = fd_long_min( deadline, fd_long_min( client->last_response_ns   +FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS,
-                                                   client->last_stream_send_ns+FD_EVENT_CLIENT_HEARTBEAT_NANOS ) );
-    if( FD_UNLIKELY( client->tx_tokens<=0L && fd_circq_unsent_cnt( client->circq ) ) ) {
-      deadline = fd_long_min( deadline, client->tx_tokens_ns + (1L-client->tx_tokens)*(long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS );
+    long send_deadline = client->last_stream_send_ns+FD_EVENT_CLIENT_HEARTBEAT_NANOS;
+    if( FD_UNLIKELY( client->tx_tokens<=0L && fd_circq_unsent_cnt( client->circq ) ) ) { /* paced: no heartbeat until tokens return */
+      ulong rate = (ulong)client->tx_byte_rate;
+      send_deadline = client->tx_tokens_ns + (long)( ( (uint128)(ulong)(1L-client->tx_tokens)*1000000000UL+rate-1UL )/rate );
     }
+    deadline = fd_long_min( deadline, fd_long_min( client->last_response_ns+FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS, send_deadline ) );
     if( FD_UNLIKELY( client->stall_since ) ) deadline = fd_long_min( deadline, client->stall_since+FD_EVENT_CLIENT_CREDIT_STALL_NANOS );
     if( FD_UNLIKELY( client->zbatch_cnt ) ) deadline = fd_long_min( deadline, client->zbatch_ns+FD_EVENT_CLIENT_ZSTD_FLUSH_NANOS );
   }
@@ -1164,25 +1170,42 @@ poll1( fd_event_client_t * client,
   }
 }
 
+static int
+tx_flush( fd_event_client_t * client,
+          long                now ) {
+  if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED || !fd_grpc_client_tx_pending( client->grpc_client ) ) ) return 0;
+  int flush_err = client->use_tls ? fd_grpc_client_tls_flush     ( client->grpc_client, client->sockfd )
+                                  : fd_grpc_client_tx_flush_socket( client->grpc_client, client->sockfd );
+  if( FD_UNLIKELY( -1==flush_err ) ) {
+    disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 );
+    return -1;
+  }
+  return 0;
+}
+
 void
 fd_event_client_poll( fd_event_client_t * client,
                       long                now,
                       int *               charge_busy ) {
   client->now = now;
   poll1( client, now, charge_busy );
 
-  /* poll1 flushes before tx() enqueues: flush once more so bytes queued
-     this poll go out now instead of arming EPOLLOUT on a writable
-     socket (a spurious waker roundtrip per message). */
-  if( FD_UNLIKELY( client->state!=FD_EVENT_CLIENT_STATE_DISCONNECTED && fd_grpc_client_tx_pending( client->grpc_client ) ) ) {
-    int flush_err = client->use_tls ? fd_grpc_client_tls_flush     ( client->grpc_client, client->sockfd )
-                                    : fd_grpc_client_tx_flush_socket( client->grpc_client, client->sockfd );
-    if( FD_UNLIKELY( -1==flush_err ) ) {
-      disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 );
-      return;
+  for( ulong i=0UL; i<FD_EVENT_CLIENT_TX_DRAIN_MAX; i++ ) {
+    if( FD_UNLIKELY( client->state!=FD_EVENT_CLIENT_STATE_CONNECTED ) ) break;
+    if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) {
+      if( FD_UNLIKELY( tx_flush( client, now ) ) ) return;
+      if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) break;
     }
+    if( FD_LIKELY( !fd_circq_unsent_cnt( client->circq ) ) ) break;
+    if( FD_UNLIKELY( client->tx_tokens<=0L ) ) break;
+    int progress = 0;
+    tx( client, now, &progress );
+    if( FD_UNLIKELY( !progress ) ) break;
+    *charge_busy = 1;
   }
 
+  if( FD_UNLIKELY( tx_flush( client, now ) ) ) return;
+
   /* Arm EPOLLOUT only on write demand: unsent HTTP/2 bytes, or a TLS
      handshake blocked on write (which is also what an in-progress
      TCP connect looks like: the handshake write EAGAINed, and connect

diff --git a/src/disco/events/fd_event_client.h b/src/disco/events/fd_event_client.h
@@ -71,7 +71,10 @@ fd_event_client_new( void *                     shmem,
                      ulong                      machine_id,
                      ulong                      buf_max,
                      int                        use_tls,
-                     fd_x509_ca_store_t const * ca_store );
+                     fd_x509_ca_store_t const * ca_store,
+                     ulong                      tx_rate_mbps );
+
+#define FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT (32UL)
 
 fd_event_client_t *
 fd_event_client_join( void * shec );

diff --git a/src/disco/events/fd_event_tile.c b/src/disco/events/fd_event_tile.c
@@ -648,7 +648,8 @@ unprivileged_init( fd_topo_t const *      topo,
                                                            ctx->machine_id,
                                                            GRPC_BUF_MAX,
                                                            ctx->use_tls,
-                                                           ctx->ca_store ) );
+                                                           ctx->ca_store,
+                                                           tile->event.tx_rate_mbps ) );
   FD_TEST( ctx->client );
   ctx->next_poll_deadline = 0L;
 

diff --git a/src/disco/events/test_event_client.c b/src/disco/events/test_event_client.c
@@ -100,7 +100,7 @@ FD_UNIT_TEST( conn_tls_lifecycle ) {
   static fd_x509_ca_store_t ca_store[1]; /* empty: the pinned mock cert bypasses chain verification */
   fd_event_client_t * client = fd_event_client_join( fd_event_client_new(
       client_mem, NULL, rng, circq, g_epoll_fd, 1<<20, "https://localhost:1", identity_pubkey, "0.0.0",
-      "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, 4096UL, 1, ca_store ) );
+      "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, 4096UL, 1, ca_store, 0UL ) );
   FD_TEST( client );
 
   static fd_tlsrec_conn_t server_conn[1];
@@ -207,7 +207,8 @@ FD_UNIT_TEST( stream_heartbeat ) {
       3UL,
       4096UL,
       0,
-      NULL ) );
+      NULL,
+      0UL ) );
   FD_TEST( client );
 
   fd_grpc_client_t * grpc = client->grpc_client;
@@ -256,15 +257,16 @@ FD_UNIT_TEST( stream_heartbeat ) {
 }
 
 static fd_event_client_t *
-test_connected_client( fd_circq_t * circq,
-                       fd_rng_t *   rng,
-                       ulong        buf_max ) {
+test_connected_client_rate( fd_circq_t * circq,
+                            fd_rng_t *   rng,
+                            ulong        buf_max,
+                            ulong        tx_rate_mbps ) {
   void * client_mem = aligned_alloc( fd_event_client_align(), fd_event_client_footprint( buf_max ) );
   FD_TEST( client_mem );
   uchar identity_pubkey[32] = {0};
   fd_event_client_t * client = fd_event_client_join( fd_event_client_new(
       client_mem, NULL, rng, circq, g_epoll_fd, 1<<20, "http://localhost:1", identity_pubkey, "0.0.0",
-      "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, buf_max, 0, NULL ) );
+      "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, buf_max, 0, NULL, tx_rate_mbps ) );
   FD_TEST( client );
   fd_grpc_client_t * grpc = client->grpc_client;
   client->state       = FD_EVENT_CLIENT_STATE_CONNECTED;
@@ -282,6 +284,13 @@ test_connected_client( fd_circq_t * circq,
   return client;
 }
 
+static fd_event_client_t *
+test_connected_client( fd_circq_t * circq,
+                       fd_rng_t *   rng,
+                       ulong        buf_max ) {
+  return test_connected_client_rate( circq, rng, buf_max, 0UL );
+}
+
 /* Queue an event of sz bytes, incompressible unless zero. */
 static void
 test_push( fd_circq_t * circq,
@@ -360,7 +369,7 @@ FD_UNIT_TEST( tx_pacing ) {
   FD_TEST( dl>now && dl<=now+(long)1e9 );
 
   /* Polls shorter than one token's worth of time keep accruing. */
-  long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS;
+  long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL );
   long tokens0 = client->tx_tokens, tokens_ns0 = client->tx_tokens_ns;
   for( long i=1L; i<ns_per_byte; i++ ) {
     test_poll_tx( client, now+i );
@@ -411,7 +420,7 @@ FD_UNIT_TEST( tx_pacing_large_msg ) {
   test_poll_tx( client, now );
   FD_TEST( client->metrics.events_sent==1UL && !test_drain( grpc ) );
   client->last_response_ns = now;
-  long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS;
+  long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL );
   long dl = fd_event_client_next_deadline( client, now );
   FD_TEST( dl==now+(1L-client->tx_tokens)*ns_per_byte );
 
@@ -429,6 +438,61 @@ FD_UNIT_TEST( tx_pacing_large_msg ) {
   fd_rng_delete( fd_rng_leave( rng ) );
 }
 
+/* An explicit rate is honored (0 keeps the default), and a paced event
+   goes out at its deadline even when a byte is not a whole number of
+   nanoseconds. */
+FD_UNIT_TEST( tx_pacing_rates ) {
+  static uchar circq_mem[ 4096UL+(1UL<<20) ] __attribute__((aligned(FD_CIRCQ_ALIGN)));
+  ulong const mbps[ 3 ] = { 0UL, 3UL, 64UL };
+  for( ulong i=0UL; i<3UL; i++ ) {
+    fd_circq_t * circq = fd_circq_join( fd_circq_new( circq_mem, 1UL<<20 ) );
+    FD_TEST( circq );
+    fd_rng_t rng_mem[1];
+    fd_rng_t * rng = fd_rng_join( fd_rng_new( rng_mem, 0U, 1UL ) );
+    fd_event_client_t * client = test_connected_client_rate( circq, rng, 1UL<<20, mbps[ i ] );
+    long const rate = (long)( ( mbps[ i ] ? mbps[ i ] : FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT )*125000UL );
+    FD_TEST( client->tx_byte_rate==rate );
+
+    test_push( circq, rng, 16UL, 0 );
+    long now = fd_log_wallclock();
+    client->tx_tokens = -1000L; client->tx_tokens_ns = now;
+    client->last_response_ns = now; client->last_stream_send_ns = now;
+    long dl = fd_event_client_next_deadline( client, now );
+    FD_TEST( dl==now+( 1001L*(long)1e9+rate-1L )/rate );
+    FD_TEST( (ulong)( (uint128)(ulong)( dl-1L-now )*(ulong)rate/1000000000UL )<=1000UL );
+    test_poll_tx( client, dl );
+    FD_TEST( client->zbatch_cnt==1UL );
+
+    free( client );
+    fd_rng_delete( fd_rng_leave( rng ) );
+  }
+}
+
+/* At a low rate a large batch leaves more than a heartbeat interval of
+   debt.  No heartbeat can go out while events are queued, so the
+   deadline is the refill, not an already expired heartbeat. */
+FD_UNIT_TEST( tx_pacing_debt_past_heartbeat ) {
+  static uchar circq_mem[ 4096UL+(1UL<<20) ] __attribute__((aligned(FD_CIRCQ_ALIGN)));
+  fd_circq_t * circq = fd_circq_join( fd_circq_new( circq_mem, 1UL<<20 ) );
+  FD_TEST( circq );
+  fd_rng_t rng_mem[1];
+  fd_rng_t * rng = fd_rng_join( fd_rng_new( rng_mem, 0U, 1UL ) );
+  fd_event_client_t * client = test_connected_client_rate( circq, rng, 1UL<<20, 1UL );
+  long const rate = 125000L;
+
+  test_push( circq, rng, 16UL, 0 );
+  long now = fd_log_wallclock();
+  client->tx_tokens = -3000000L; client->tx_tokens_ns = now; /* 24 s of debt */
+  client->last_response_ns    = now;
+  client->last_stream_send_ns = now-FD_EVENT_CLIENT_HEARTBEAT_NANOS-1L;
+  long dl = fd_event_client_next_deadline( client, now );
+  FD_TEST( dl==now+( 3000001L*(long)1e9+rate-1L )/rate );
+  FD_TEST( dl>now+FD_EVENT_CLIENT_HEARTBEAT_NANOS );
+
+  free( client );
+  fd_rng_delete( fd_rng_leave( rng ) );
+}
+
 /* Zero peer credit with no progress for the stall window reconnects;
    genuine progress resets the timer. */
 FD_UNIT_TEST( credit_stall ) {
@@ -466,7 +530,7 @@ FD_UNIT_TEST( credit_stall ) {
 
   /* The bucket keeps refilling while parked, so an expired pacing
      deadline does not pin the tile awake. */
-  long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS;
+  long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL );
   test_push( circq, rng, 16UL, 1 );
   client->tx_tokens = -1000L; client->tx_tokens_ns = now;
   long t0 = now+2000L*ns_per_byte;

diff --git a/src/disco/topo/fd_topo.h b/src/disco/topo/fd_topo.h
@@ -347,6 +347,7 @@ struct fd_topo_tile {
       long   boot_timestamp_nanos;
       ulong  launch_cpus[ 1024UL/64UL ];
       char   launch_cgroup[ PATH_MAX ];
+      ulong  tx_rate_mbps;
     } event;
 
     struct {