quic: fix scheduling misses, svc_schedule validates conn

      reset next_timeout in svc_schedule

      fix retx timeout ticks/ns issue

      prep schedule in gen_frames

      CRIT if svc_timer_cnt != active_conn_cnt

      schedule idle_timeout from last_activity, not now

      test_quic_conformance: update for service heap
      fix test_quic_keep_alive

Author: akhinvasara-jumptrading

src/waltz/quic/fd_quic.c

@@ -690,6 +690,16 @@ fd_quic_svc_schedule1( fd_quic_conn_t * conn ) {
   fd_quic_svc_schedule( fd_quic_get_state(conn->quic)->svc_timers, conn );
 }
 
+/* Validation Helper */
+static inline void
+svc_cnt_eq_active_conn( fd_quic_svc_timers_t * timers, fd_quic_t * quic ) {
+  ulong const event_cnt = fd_quic_svc_cnt_events( timers );
+  ulong const conn_cnt  = quic->metrics.conn_active_cnt;
+  if( FD_UNLIKELY( event_cnt != conn_cnt ) ) {
+    FD_LOG_CRIT(( "only %lu out of %lu connections are in timer", event_cnt, conn_cnt ));
+  }
+}
+
 /* validates the free conn list doesn't cycle, point nowhere, leak, or point to live conn */
 static void
 fd_quic_conn_free_validate( fd_quic_t * quic ) {
@@ -2252,6 +2262,8 @@ fd_quic_process_quic_packet_v1( fd_quic_t *     quic,
         break;
     }
 
+    fd_quic_svc_schedule( state->svc_timers, conn );
+
     if( FD_UNLIKELY( rc == FD_QUIC_PARSE_FAIL ) ) {
       FD_DEBUG( FD_LOG_DEBUG(( "Rejected packet (type=%d)", long_packet_type )); )
       return FD_QUIC_PARSE_FAIL;
@@ -2268,6 +2280,9 @@ fd_quic_process_quic_packet_v1( fd_quic_t *     quic,
     ulong dst_conn_id = fd_ulong_load_8( cur_ptr+1 );
     conn = fd_quic_conn_query( state->conn_map, dst_conn_id );
     rc = fd_quic_handle_v1_one_rtt( quic, conn, pkt, cur_ptr, cur_sz );
+
+    fd_quic_svc_schedule( state->svc_timers, conn );
+
     if( FD_UNLIKELY( rc == FD_QUIC_PARSE_FAIL ) ) {
       return FD_QUIC_PARSE_FAIL;
     }
@@ -2283,6 +2298,7 @@ fd_quic_process_quic_packet_v1( fd_quic_t *     quic,
   int ack_type = fd_quic_lazy_ack_pkt( quic, conn, pkt );
   quic->metrics.ack_tx[ ack_type ]++;
 
+  /* fd_quic_lazy_ack_pkt may have prepped schedule */
   fd_quic_svc_schedule( state->svc_timers, conn );
 
   if( pkt->rtt_ack_time ) {
@@ -2462,6 +2478,7 @@ fd_quic_process_packet_impl( fd_quic_t * quic,
       }
 
       rc = fd_quic_process_quic_packet_v1( quic, &pkt, cur_ptr, cur_sz );
+      svc_cnt_eq_active_conn( state->svc_timers, quic );
 
       /* 0UL means no progress, so fail */
       if( FD_UNLIKELY( ( rc == FD_QUIC_PARSE_FAIL ) |
@@ -2491,6 +2508,7 @@ fd_quic_process_packet_impl( fd_quic_t * quic,
   /* short header packet
      only one_rtt packets currently have short headers */
   fd_quic_process_quic_packet_v1( quic, &pkt, cur_ptr, cur_sz );
+  svc_cnt_eq_active_conn( state->svc_timers, quic );
 }
 
 void
@@ -2875,7 +2893,7 @@ fd_quic_svc_poll( fd_quic_t *      quic,
             max_idle_timeout value advertised by both endpoints." */
         FD_DEBUG( FD_LOG_WARNING(("%s  conn %p  conn_idx: %u  closing due to idle timeout (%g ms)",
             conn->server?"SERVER":"CLIENT",
-            (void *)conn, conn->conn_idx, (double)fd_quic_ticks_to_us(conn->idle_timeout_ticks) / 1e3 )); )
+            (void *)conn, conn->conn_idx, (double)fd_quic_ticks_to_us( quic, conn->idle_timeout_ticks ) / 1e3 )); )
 
         fd_quic_set_conn_state( conn, FD_QUIC_CONN_STATE_DEAD );
         quic->metrics.conn_timeout_cnt++;
@@ -2909,7 +2927,7 @@ fd_quic_svc_poll( fd_quic_t *      quic,
     break;
   default:
     /* prep idle timeout or keep alive at idle timeout/2 */
-    fd_quic_svc_prep_schedule( conn, state->now + (conn->idle_timeout_ticks>>(quic->config.keep_alive)) );
+    fd_quic_svc_prep_schedule( conn, conn->last_activity + (conn->idle_timeout_ticks>>(quic->config.keep_alive)) );
     fd_quic_svc_schedule( state->svc_timers, conn );
     break;
   }
@@ -2927,6 +2945,7 @@ fd_quic_service( fd_quic_t * quic ) {
   long now_ticks = fd_tickcount();
 
   fd_quic_svc_timers_t * timers = state->svc_timers;
+  svc_cnt_eq_active_conn( timers, quic );
   fd_quic_svc_event_t    next   = fd_quic_svc_timers_next( timers, now, 1 /* pop */);
   if( FD_UNLIKELY( next.conn == NULL ) ) {
     return 0;
@@ -3513,6 +3532,8 @@ fd_quic_gen_frames( fd_quic_conn_t           * conn,
     }
   }
 
+  fd_quic_svc_prep_schedule( conn, pkt_meta_tmpl->expiry );
+
   return payload_ptr;
 }
 
@@ -3576,7 +3597,8 @@ fd_quic_conn_tx( fd_quic_t      * quic,
   ulong now = fd_quic_get_state( quic )->now;
 
   /* initialize expiry and tx_time */
-  fd_quic_pkt_meta_t pkt_meta_tmpl[1] = {{.expiry = now+500000000UL, .tx_time = now}};
+  ulong const retx_timeout = fd_quic_us_to_ticks( quic, 500000UL ); /* 500 ms */
+  fd_quic_pkt_meta_t pkt_meta_tmpl[1] = {{.expiry = now+retx_timeout, .tx_time = now}};
   // pkt_meta_tmpl->expiry = fd_quic_calc_expiry( conn, now );
   //ulong margin = (ulong)(conn->rtt->smoothed_rtt) + (ulong)(3 * conn->rtt->var_rtt);
   //if( margin < pkt_meta->expiry ) {
@@ -3855,7 +3877,6 @@ fd_quic_conn_tx( fd_quic_t      * quic,
 void
 fd_quic_conn_service( fd_quic_t * quic, fd_quic_conn_t * conn, ulong now ) {
   (void)now;
-  conn->svc_meta.next_timeout = ULONG_MAX;
 
   /* Send new rtt measurement probe? */
   if( FD_UNLIKELY(now > conn->last_ack + (ulong)conn->rtt_period_ticks) ) {
@@ -3947,6 +3968,7 @@ fd_quic_conn_service( fd_quic_t * quic, fd_quic_conn_t * conn, ulong now ) {
     case FD_QUIC_CONN_STATE_INVALID:
       /* fall thru */
     default:
+      FD_LOG_CRIT(( "invalid conn state %u", conn->state ));
       return;
   }
 
@@ -3972,12 +3994,6 @@ fd_quic_conn_free( fd_quic_t *      quic,
 
   fd_quic_state_t * state = fd_quic_get_state( quic );
 
-  /* no need to remove this connection from the events queue
-     free is called from two places:
-       fini    - service will never be called again. All events are destroyed
-       service - removes event before calling free. Event only allowed to be
-       enqueued once */
-
   /* remove all stream ids from map, and free stream */
 
   /* remove used streams */
@@ -4034,12 +4050,12 @@ fd_quic_conn_free( fd_quic_t *      quic,
   }
   conn->tls_hs = NULL;
 
+  /* remove from service queue */
   fd_quic_svc_cancel( state->svc_timers, conn );
 
   /* put connection back in free list */
   conn->free_conn_next  = state->free_conn_list;
   state->free_conn_list = conn->conn_idx;
-  fd_quic_set_conn_state( conn, FD_QUIC_CONN_STATE_INVALID );
 
   quic->metrics.conn_active_cnt--;
 
@@ -4066,7 +4082,6 @@ fd_quic_connect( fd_quic_t *  quic,
     }
   }
 
-
   fd_rng_t * rng = state->_rng;
 
   /* create conn ids for us and them

src/waltz/quic/fd_quic_svc_q.c

@@ -19,10 +19,7 @@ typedef fd_quic_svc_event_t fd_quic_svc_queue_prq_t;
 
 ulong
 fd_quic_svc_timers_footprint( ulong max_conn ) {
-  ulong offset = 0UL;
-  offset       = fd_ulong_align_up( offset, fd_quic_svc_queue_prq_align() );
-  offset      += fd_quic_svc_queue_prq_footprint( max_conn );
-  return offset;
+  return fd_quic_svc_queue_prq_footprint( max_conn );
 }
 
 ulong
@@ -78,9 +75,19 @@ fd_quic_svc_cancel( fd_quic_svc_timers_t * timers,
 void
 fd_quic_svc_schedule( fd_quic_svc_timers_t * timers,
                       fd_quic_conn_t       * conn ) {
+
+  /* if conn null or invalid, do not schedule */
+  if( FD_UNLIKELY( !conn || conn->state == FD_QUIC_CONN_STATE_INVALID ) ) {
+    /* cleaner/safer to check in here for now. If function call overhead
+       becomes a constraint, move check to caller */
+    return;
+  }
+
   ulong idx    = conn->svc_meta.idx;
   ulong expiry = conn->svc_meta.next_timeout;
 
+  conn->svc_meta.next_timeout = ULONG_MAX; /* reset next_timeout */
+
   if( FD_UNLIKELY( idx != FD_QUIC_SVC_IDX_INVAL ) ) {
     /* find current expiry */
     fd_quic_svc_event_t * event      = timers + idx;
@@ -95,6 +102,7 @@ fd_quic_svc_schedule( fd_quic_svc_timers_t * timers,
       conn->svc_meta.idx = FD_QUIC_SVC_IDX_INVAL;
     }
   }
+  /* TODO: potential perf improvement combining remove and insert(?) */
 
   /* insert new element */
   fd_quic_svc_event_t e = {
@@ -161,3 +169,8 @@ fd_quic_svc_get_event( fd_quic_svc_timers_t * timers,
   if( idx == FD_QUIC_SVC_IDX_INVAL ) return NULL;
   return timers + idx;
 }
+
+ulong
+fd_quic_svc_cnt_events( fd_quic_svc_timers_t * timers ) {
+  return fd_quic_svc_queue_prq_cnt( timers );
+}

src/waltz/quic/fd_quic_svc_q.h

@@ -56,7 +56,8 @@ fd_quic_svc_timers_init_conn( fd_quic_conn_t * conn );
 
 /* fd_quic_svc_schedule schedules a connection timer.
    Uses conn->svc_meta.next_timeout as the expiry time.
-   If already scheduled, keeps the earlier time. */
+   If already scheduled, keeps the earlier time. Resets
+   next_timeout to ULONG_MAX. */
 void
 fd_quic_svc_schedule( fd_quic_svc_timers_t * timers,
                       fd_quic_conn_t       * conn );
@@ -90,4 +91,10 @@ fd_quic_svc_event_t*
 fd_quic_svc_get_event( fd_quic_svc_timers_t * timers,
                        fd_quic_conn_t       * conn );
 
+
+/* fd_quic_svc_cnt_events returns the number of conns with active timers
+   Primarily used for testing/validation. */
+ulong
+fd_quic_svc_cnt_events( fd_quic_svc_timers_t * timers );
+
 #endif /* HEADER_fd_src_waltz_quic_fd_quic_svc_q_h */

src/waltz/quic/tests/fd_quic_sandbox.c

@@ -315,6 +315,9 @@ fd_quic_sandbox_new_conn_established( fd_quic_sandbox_t * sandbox,
   conn->srx->rx_max_data_ackd = 0UL;
   conn->tx_initial_max_stream_data_uni = 0UL;
 
+  fd_quic_state_t * state = fd_quic_get_state( quic );
+  fd_quic_svc_schedule( state->svc_timers, conn );
+
   /* TODO set a realistic packet number */
 
   return conn;

src/waltz/quic/tests/test_quic_conformance.c

@@ -120,7 +120,7 @@ test_quic_ping_frame( fd_quic_sandbox_t * sandbox,
   fd_quic_sandbox_init( sandbox, FD_QUIC_ROLE_SERVER );
   fd_quic_conn_t * conn = fd_quic_sandbox_new_conn_established( sandbox, rng );
   conn->ack_gen->is_elicited = 0;
-  FD_TEST( conn->svc_meta.idx == FD_QUIC_SVC_IDX_INVAL );
+  FD_TEST( conn->svc_meta.idx != FD_QUIC_SVC_IDX_INVAL );
 
   uchar buf[1] = {0x01};
   fd_quic_sandbox_send_lone_frame( sandbox, conn, buf, sizeof(buf) );

src/waltz/quic/tests/test_quic_keep_alive.c

@@ -129,6 +129,9 @@ main( int argc, char ** argv ) {
   server_quic->config.idle_timeout = 1e7;
   client_quic->config.idle_timeout = 1e9;
 
+  server_quic->config.ack_delay = 1e6;
+  client_quic->config.ack_delay = 1e6;
+
   fd_quic_virtual_pair_t vp;
   fd_quic_virtual_pair_init( &vp, server_quic, client_quic );
 

src/waltz/quic/tests/test_quic_svc_q.c

@@ -105,6 +105,7 @@ test_multiple_connections( fd_quic_svc_timers_t * timers,
   fd_quic_conn_t * conns[conn_cnt]; /* array of conn ptrs */
   for( uint i=0; i<conn_cnt; i++ ) {
     conns[i] = (fd_quic_conn_t *)(conn_base + i * conn_sz);
+    conns[i]->state = FD_QUIC_CONN_STATE_ACTIVE;
   }
 
   ulong now = 1000UL;
@@ -192,6 +193,7 @@ main( int argc, char ** argv ) {
 
   {
     fd_quic_conn_t * conn = (fd_quic_conn_t *)create_mock_conns( &limits , 1);
+    conn->state = FD_QUIC_CONN_STATE_ACTIVE;
     test_svc_schedule( timers, conn );
     test_svc_cancel( timers, conn );
     free( conn );