quic: fix scheduling misses, svc_schedule validates conn

      reset next_timeout in svc_schedule

      increase retx timeout to 1250M ticks

      rm unnecessary unacked_sz reset
      WARN if svc_timer_cnt != active_conn_cnt

      test_quic_conformance: update for service heap

Author: akhinvasara-jumptrading

src/waltz/quic/fd_quic.c

@@ -2020,7 +2020,6 @@ fd_quic_lazy_ack_pkt( fd_quic_t *           quic,
     ( ( pkt->enc_level == fd_quic_enc_level_initial_id   ) |
       ( pkt->enc_level == fd_quic_enc_level_handshake_id ) );
   if( ack_sz_threshold_hit | force_instant_ack ) {
-    conn->unacked_sz = 0UL;
     fd_quic_svc_prep_schedule( conn, state->now );
   } else {
     fd_quic_svc_prep_schedule( conn, state->now + quic->config.ack_delay );
@@ -2252,6 +2251,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 +2269,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 +2287,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 +2467,11 @@ fd_quic_process_packet( fd_quic_t * quic,
       }
 
       rc = fd_quic_process_quic_packet_v1( quic, &pkt, cur_ptr, cur_sz );
+      if( FD_UNLIKELY( fd_quic_svc_cnt_events( state->svc_timers ) != quic->metrics.conn_active_cnt ) ) {
+        FD_LOG_WARNING(( "only %lu out of %lu connections are in timer",
+                          fd_quic_svc_cnt_events( state->svc_timers ),
+                          quic->metrics.conn_active_cnt ));
+      }
 
       /* 0UL means no progress, so fail */
       if( FD_UNLIKELY( ( rc == FD_QUIC_PARSE_FAIL ) |
@@ -2491,6 +2501,11 @@ fd_quic_process_packet( 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 );
+  if( FD_UNLIKELY( fd_quic_svc_cnt_events( state->svc_timers ) != quic->metrics.conn_active_cnt ) ) {
+    FD_LOG_WARNING(( "only %lu out of %lu connections are in timer",
+                      fd_quic_svc_cnt_events( state->svc_timers ),
+                      quic->metrics.conn_active_cnt ));
+  }
 }
 
 /* main receive-side entry point */
@@ -2924,6 +2939,11 @@ fd_quic_service( fd_quic_t * quic ) {
   long now_ticks = fd_tickcount();
 
   fd_quic_svc_timers_t * timers = state->svc_timers;
+  if( FD_UNLIKELY( fd_quic_svc_cnt_events( timers ) != quic->metrics.conn_active_cnt ) ) {
+    FD_LOG_WARNING(( "only %lu out of %lu connections are in timer",
+                      fd_quic_svc_cnt_events( timers ),
+                      quic->metrics.conn_active_cnt ));
+  }
   fd_quic_svc_event_t    next   = fd_quic_svc_timers_next( timers, now, 1 /* pop */);
   if( FD_UNLIKELY( next.conn == NULL ) ) {
     return 0;
@@ -3573,7 +3593,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 = 1250UL*1000000UL; /* 1250M ticks ~ 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 ) {
@@ -3852,7 +3873,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) ) {
@@ -3951,6 +3971,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_ERR(( "invalid conn state %u", conn->state ));
       return;
   }
 
@@ -3976,12 +3997,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 */
@@ -4038,12 +4053,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--;
 
@@ -4070,7 +4085,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

@@ -78,9 +78,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 +105,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 +172,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

@@ -637,6 +637,7 @@ test_quic_pktmeta_pktnum_skip( fd_quic_sandbox_t * sandbox,
 
   fd_quic_sandbox_init( sandbox, FD_QUIC_ROLE_SERVER );
   fd_quic_t *                  quic    = sandbox->quic;
+  fd_quic_state_t *            state   = fd_quic_get_state( quic );
   fd_quic_conn_t *             conn    = fd_quic_sandbox_new_conn_established( sandbox, rng );
   fd_quic_metrics_t *          metrics = &conn->quic->metrics;
   fd_quic_pkt_meta_tracker_t * tracker = &conn->pkt_meta_tracker;
@@ -649,7 +650,8 @@ test_quic_pktmeta_pktnum_skip( fd_quic_sandbox_t * sandbox,
     conn->flags          = ( conn->flags & ~FD_QUIC_CONN_FLAGS_PING_SENT ) | FD_QUIC_CONN_FLAGS_PING;
     conn->upd_pkt_number = FD_QUIC_PKT_NUM_PENDING;
     sandbox->wallclock += (ulong)10e6;
-    fd_quic_svc_schedule1( conn, FD_QUIC_SVC_INSTANT );
+    conn->svc_meta.next_timeout = sandbox->wallclock;
+    fd_quic_svc_schedule( state->svc_timers, conn );
     fd_quic_service( quic );
     FD_TEST( conn->pkt_number[2] == next_pkt_number + 1UL );
     next_pkt_number++;
@@ -679,7 +681,8 @@ test_quic_pktmeta_pktnum_skip( fd_quic_sandbox_t * sandbox,
     conn->flags          = ( conn->flags & ~FD_QUIC_CONN_FLAGS_PING_SENT ) | FD_QUIC_CONN_FLAGS_PING;
     conn->upd_pkt_number = FD_QUIC_PKT_NUM_PENDING;
     sandbox->wallclock += (ulong)10e6;
-    fd_quic_svc_schedule1( conn, FD_QUIC_SVC_INSTANT );
+    conn->svc_meta.next_timeout = sandbox->wallclock;
+    fd_quic_svc_schedule( state->svc_timers, conn );
     fd_quic_service( quic );
     FD_TEST( conn->pkt_number[2] == next_pkt_number );
     FD_TEST( *metrics_alloc_fail_cnt > alloc_fail_cnt );
@@ -718,7 +721,8 @@ test_quic_pktmeta_pktnum_skip( fd_quic_sandbox_t * sandbox,
     conn->flags          = ( conn->flags & ~FD_QUIC_CONN_FLAGS_PING_SENT ) | FD_QUIC_CONN_FLAGS_PING;
     conn->upd_pkt_number = FD_QUIC_PKT_NUM_PENDING;
     sandbox->wallclock += (ulong)10e6;
-    fd_quic_svc_schedule1( conn, FD_QUIC_SVC_INSTANT );
+    conn->svc_meta.next_timeout = sandbox->wallclock;
+    fd_quic_svc_schedule( state->svc_timers, conn );
     fd_quic_service( quic );
     FD_TEST( conn->pkt_number[2] == next_pkt_number + 1UL );
     next_pkt_number++;

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 );