···

       
83
       
83
       
 
       
        };

     

       
84
       
84
       
 
       
        loop {

     

       
85
       
85
       
 
       
            if let Some((t, item)) = get().await {

     

       
86
       
86
       
-
       
                let expected_release = t + self.delay;

     

       
87
       
86
       
 
       
                let now = Instant::now();

     

       
88
       
88
       
-
       
                if expected_release > now {

     

       
87
       
87
       
+
       
                let expected_release = t + self.delay;

     

       
88
       
88
       
+
       
                if expected_release.saturating_duration_since(now) > Duration::from_millis(1) {

     

       
89
       
89
       
 
       
                    tokio::time::sleep_until(expected_release.into()).await;

     

       
90
       
90
       
 
       
                    metrics::counter!("delay_queue_emit_total", "early" => "yes").increment(1);

     

       
91
       
91
       
+
       
                    metrics::histogram!("delay_queue_emit_overshoot").record(0);

     

       
91
       
92
       
 
       
                } else {

     

       
93
       
93
       
+
       
                    let overshoot = now.saturating_duration_since(expected_release);

     

       
92
       
94
       
 
       
                    metrics::counter!("delay_queue_emit_total", "early" => "no").increment(1);

     

       
93
       
93
       
-
       
                    let overshoot = now - expected_release;

     

       
94
       
95
       
 
       
                    metrics::histogram!("delay_queue_emit_overshoot").record(overshoot.as_secs_f64());

     

       
95
       
96
       
 
       
                }

     

       
96
       
97
       
 
       
                return Some(item)