22// SPDX-License-Identifier: Apache-2.0
33
44use tokio:: sync:: { mpsc, oneshot} ;
5- use tracing:: error;
5+ use tokio:: time:: Duration ;
6+ use tracing:: warn;
67
78use crate :: traces:: span_dedup:: { DedupKey , Deduper } ;
89
@@ -12,6 +13,8 @@ pub enum DedupError {
1213 SendError ( mpsc:: error:: SendError < DedupCommand > ) ,
1314 #[ error( "Failed to receive response from deduper: {0}" ) ]
1415 RecvError ( oneshot:: error:: RecvError ) ,
16+ #[ error( "Timeout waiting for response from deduper" ) ]
17+ Timeout ,
1518}
1619
1720pub enum DedupCommand {
@@ -34,14 +37,29 @@ impl DedupHandle {
3437 ///
3538 /// # Errors
3639 ///
37- /// Returns an error if the command cannot be sent to the deduper service
38- /// or if the response cannot be received.
39- pub async fn check_and_add ( & self , key : DedupKey ) -> Result < bool , DedupError > {
40+ /// Returns an error if the command cannot be sent to the deduper service,
41+ /// if the response cannot be received, or if the operation times out after 5 seconds.
42+ pub async fn check_and_add (
43+ & self ,
44+ key : DedupKey ,
45+ timeout : Option < Duration > ,
46+ ) -> Result < bool , DedupError > {
4047 let ( response_tx, response_rx) = oneshot:: channel ( ) ;
4148 self . tx
4249 . send ( DedupCommand :: CheckAndAdd ( key, response_tx) )
4350 . map_err ( DedupError :: SendError ) ?;
44- response_rx. await . map_err ( DedupError :: RecvError )
51+
52+ // Sometimes the dedup service fails to send a response for unknown reasons, so we
53+ // add a timeout to avoid blocking the caller forever. We may remove the
54+ // timeout if we can figure out and fix the root cause.
55+ if let Some ( timeout) = timeout {
56+ tokio:: time:: timeout ( timeout, response_rx)
57+ . await
58+ . map_err ( |_| DedupError :: Timeout ) ?
59+ . map_err ( DedupError :: RecvError )
60+ } else {
61+ response_rx. await . map_err ( DedupError :: RecvError )
62+ }
4563 }
4664}
4765
@@ -77,7 +95,7 @@ impl DedupService {
7795 DedupCommand :: CheckAndAdd ( key, response_tx) => {
7896 let was_added = self . deduper . check_and_add ( key) ;
7997 if let Err ( e) = response_tx. send ( was_added) {
80- error ! ( "Failed to send check_and_add response: {e:?}" ) ;
98+ warn ! ( "Failed to send check_and_add response: {e:?}" ) ;
8199 }
82100 }
83101 }
@@ -112,17 +130,17 @@ mod tests {
112130 let key2 = DedupKey :: new ( 100 , 456 ) ;
113131
114132 // First call should return true (key was added)
115- assert ! ( handle. check_and_add( key1) . await . unwrap( ) ) ;
133+ assert ! ( handle. check_and_add( key1, None ) . await . unwrap( ) ) ;
116134
117135 // Second call should return false (key already exists)
118- assert ! ( !handle. check_and_add( key1) . await . unwrap( ) ) ;
136+ assert ! ( !handle. check_and_add( key1, None ) . await . unwrap( ) ) ;
119137
120138 // Different key should return true again
121- assert ! ( handle. check_and_add( key2) . await . unwrap( ) ) ;
139+ assert ! ( handle. check_and_add( key2, None ) . await . unwrap( ) ) ;
122140
123141 // Calling again on already-added keys should return false
124- assert ! ( !handle. check_and_add( key1) . await . unwrap( ) ) ;
125- assert ! ( !handle. check_and_add( key2) . await . unwrap( ) ) ;
142+ assert ! ( !handle. check_and_add( key1, None ) . await . unwrap( ) ) ;
143+ assert ! ( !handle. check_and_add( key2, None ) . await . unwrap( ) ) ;
126144 }
127145
128146 #[ tokio:: test]
@@ -139,17 +157,17 @@ mod tests {
139157 let key4 = DedupKey :: new ( 4 , 40 ) ;
140158
141159 // Add 3 keys
142- assert ! ( handle. check_and_add( key1) . await . unwrap( ) ) ;
143- assert ! ( handle. check_and_add( key2) . await . unwrap( ) ) ;
144- assert ! ( handle. check_and_add( key3) . await . unwrap( ) ) ;
160+ assert ! ( handle. check_and_add( key1, None ) . await . unwrap( ) ) ;
161+ assert ! ( handle. check_and_add( key2, None ) . await . unwrap( ) ) ;
162+ assert ! ( handle. check_and_add( key3, None ) . await . unwrap( ) ) ;
145163
146164 // Add a 4th key, should evict the oldest (key1)
147- assert ! ( handle. check_and_add( key4) . await . unwrap( ) ) ;
165+ assert ! ( handle. check_and_add( key4, None ) . await . unwrap( ) ) ;
148166
149167 // Now key1 should be addable again (was evicted)
150- assert ! ( handle. check_and_add( key1) . await . unwrap( ) ) ;
168+ assert ! ( handle. check_and_add( key1, None ) . await . unwrap( ) ) ;
151169
152170 // But key2 should now be evicted
153- assert ! ( handle. check_and_add( key2) . await . unwrap( ) ) ;
171+ assert ! ( handle. check_and_add( key2, None ) . await . unwrap( ) ) ;
154172 }
155173}
0 commit comments