@@ -3,7 +3,10 @@ use std::{
33 net:: SocketAddr ,
44 path:: { Path , PathBuf } ,
55 str:: FromStr ,
6- sync:: { Arc , Mutex as StdMutex } ,
6+ sync:: {
7+ Arc , Mutex as StdMutex ,
8+ atomic:: { AtomicU64 , Ordering } ,
9+ } ,
710 time:: { Duration , SystemTime , UNIX_EPOCH } ,
811} ;
912
@@ -28,7 +31,7 @@ use tokio::{
2831use tokio_util:: sync:: CancellationToken ;
2932use tower_http:: trace:: TraceLayer ;
3033use tracing:: { Span , debug, error, info} ;
31- use wsrx:: { ProxyStats , TrafficObserver , proxy_observed} ;
34+ use wsrx:: { ProxyStats , TrafficDirection , TrafficObserver , proxy_observed} ;
3235
3336use crate :: cli:: {
3437 capture:: {
@@ -155,6 +158,57 @@ struct ConnectionAudit {
155158 capture_truncated : bool ,
156159}
157160
161+ struct ConnectionObserver {
162+ client_to_target : AtomicU64 ,
163+ target_to_client : AtomicU64 ,
164+ capture : Option < Arc < dyn TrafficObserver > > ,
165+ }
166+
167+ impl ConnectionObserver {
168+ fn new ( capture : Option < Arc < dyn TrafficObserver > > ) -> Self {
169+ Self {
170+ client_to_target : AtomicU64 :: new ( 0 ) ,
171+ target_to_client : AtomicU64 :: new ( 0 ) ,
172+ capture,
173+ }
174+ }
175+
176+ fn stats ( & self ) -> ProxyStats {
177+ ProxyStats {
178+ client_to_target : self . client_to_target . load ( Ordering :: Relaxed ) ,
179+ target_to_client : self . target_to_client . load ( Ordering :: Relaxed ) ,
180+ }
181+ }
182+ }
183+
184+ impl TrafficObserver for ConnectionObserver {
185+ fn observe ( & self , direction : TrafficDirection , data : & [ u8 ] ) {
186+ match direction {
187+ TrafficDirection :: ClientToTarget => {
188+ self . client_to_target
189+ . fetch_add ( data. len ( ) as u64 , Ordering :: Relaxed ) ;
190+ }
191+ TrafficDirection :: TargetToClient => {
192+ self . target_to_client
193+ . fetch_add ( data. len ( ) as u64 , Ordering :: Relaxed ) ;
194+ }
195+ }
196+ if let Some ( capture) = & self . capture {
197+ capture. observe ( direction, data) ;
198+ }
199+ }
200+ }
201+
202+ fn classify_proxy_error ( error : & str , stats : ProxyStats ) -> String {
203+ if ( stats. client_to_target > 0 || stats. target_to_client > 0 )
204+ && error. contains ( "Connection reset without closing handshake" )
205+ {
206+ "completed_ungraceful_close" . into ( )
207+ } else {
208+ format ! ( "proxy_error:{error}" )
209+ }
210+ }
211+
158212#[ derive( Clone , FromRef ) ]
159213pub struct GlobalState {
160214 pub secret : String ,
@@ -393,22 +447,25 @@ async fn process_traffic(
393447 }
394448 }
395449
396- let observer = capture
450+ let capture_observer = capture
397451 . as_ref ( )
398452 . map ( |session| session. observer . clone ( ) as Arc < dyn TrafficObserver > ) ;
453+ let connection_observer = Arc :: new ( ConnectionObserver :: new ( capture_observer) ) ;
454+ let observer = Some ( connection_observer. clone ( ) as Arc < dyn TrafficObserver > ) ;
399455 match timeout (
400456 state. connection_timeout ,
401457 proxy_observed ( socket. into ( ) , tcp, CancellationToken :: new ( ) , observer) ,
402458 )
403459 . await
404460 {
405- Ok ( Ok ( proxy_stats ) ) => stats = proxy_stats ,
461+ Ok ( Ok ( _ ) ) => { }
406462 Ok ( Err ( err) ) => {
407463 debug ! ( target = %record. target, "tunnel closed: {err}" ) ;
408- result = format ! ( "proxy_error:{ err}" ) ;
464+ result = classify_proxy_error ( & err. to_string ( ) , connection_observer . stats ( ) ) ;
409465 }
410466 Err ( _) => result = "connection_timeout" . into ( ) ,
411467 }
468+ stats = connection_observer. stats ( ) ;
412469
413470 let ( captured_bytes, capture_truncated, capture_path) = match & capture {
414471 Some ( session) => {
@@ -702,3 +759,36 @@ fn spawn_expiry_cleanup(state: GlobalState) {
702759 }
703760 } ) ;
704761}
762+
763+ #[ cfg( test) ]
764+ mod tests {
765+ use super :: * ;
766+
767+ #[ test]
768+ fn connection_observer_keeps_counts_for_failed_proxies ( ) {
769+ let observer = ConnectionObserver :: new ( None ) ;
770+ observer. observe ( TrafficDirection :: ClientToTarget , b"request" ) ;
771+ observer. observe ( TrafficDirection :: TargetToClient , b"response" ) ;
772+
773+ assert_eq ! (
774+ observer. stats( ) ,
775+ ProxyStats {
776+ client_to_target: 7 ,
777+ target_to_client: 8 ,
778+ }
779+ ) ;
780+ }
781+
782+ #[ test]
783+ fn reset_after_transfer_is_not_reported_as_failed_traffic ( ) {
784+ let result = classify_proxy_error (
785+ "WebSocket protocol error: Connection reset without closing handshake" ,
786+ ProxyStats {
787+ client_to_target : 7 ,
788+ target_to_client : 8 ,
789+ } ,
790+ ) ;
791+
792+ assert_eq ! ( result, "completed_ungraceful_close" ) ;
793+ }
794+ }
0 commit comments