@@ -311,6 +311,13 @@ impl SharedState {
311311 . expect ( "refresh TTFE mutex poisoned" ) = None ;
312312 }
313313
314+ fn finish_refresh ( & self ) {
315+ * self
316+ . refresh_started_at
317+ . lock ( )
318+ . expect ( "refresh start mutex poisoned" ) = None ;
319+ }
320+
314321 fn request_time_to_first_env ( & self ) -> Option < Duration > {
315322 * self
316323 . first_env_since_refresh
@@ -486,12 +493,15 @@ impl PetClient {
486493 /// Refresh environments
487494 pub fn refresh ( & mut self , params : Option < Value > ) -> Result < ( RefreshResult , Duration ) , String > {
488495 // The operation begins immediately before request serialization and pipe I/O.
489- // Clearing prior observations is intentionally outside the measured boundary.
496+ // Inventory/progress clearing is intentionally outside the measured boundary.
490497 self . state . prepare_refresh ( ) ;
491498 let started_at = ( self . clock ) ( ) ;
492499 self . state . begin_refresh_at ( started_at) ;
493- let result = self . send_request ( "refresh" , params. unwrap_or ( json ! ( { } ) ) ) ?;
494- let elapsed = ( self . clock ) ( )
500+ let result = self . send_request ( "refresh" , params. unwrap_or ( json ! ( { } ) ) ) ;
501+ let completed_at = ( self . clock ) ( ) ;
502+ self . state . finish_refresh ( ) ;
503+ let result = result?;
504+ let elapsed = completed_at
495505 . checked_duration_since ( started_at)
496506 . expect ( "refresh completion must not precede its request boundary" ) ;
497507
@@ -902,10 +912,13 @@ fn collect_refresh_diagnostics(
902912struct ChannelWriter {
903913 sender : mpsc:: Sender < Vec < u8 > > ,
904914 buffered : Vec < u8 > ,
915+ clock : Arc < Mutex < Instant > > ,
916+ write_delay : Duration ,
905917}
906918
907919impl Write for ChannelWriter {
908920 fn write ( & mut self , bytes : & [ u8 ] ) -> std:: io:: Result < usize > {
921+ * self . clock . lock ( ) . expect ( "fixture clock mutex poisoned" ) += self . write_delay ;
909922 self . buffered . extend_from_slice ( bytes) ;
910923 Ok ( bytes. len ( ) )
911924 }
@@ -947,17 +960,45 @@ fn frame_jsonrpc(value: &Value) -> Vec<u8> {
947960 format ! ( "Content-Length: {}\r \n \r \n {}" , body. len( ) , body) . into_bytes ( )
948961}
949962
950- #[ test]
951- fn refresh_path_measures_client_round_trip_and_resets_ttfe_per_request ( ) {
952- let base = Instant :: now ( ) ;
963+ fn timed_fixture_client (
964+ base : Instant ,
965+ write_delay : Duration ,
966+ ) -> ( PetClient , mpsc:: Receiver < Vec < u8 > > , mpsc:: Sender < TimedChunk > ) {
953967 let clock_value = Arc :: new ( Mutex :: new ( base) ) ;
954968 let clock = {
955969 let clock_value = clock_value. clone ( ) ;
956970 Arc :: new ( move || * clock_value. lock ( ) . expect ( "fixture clock mutex poisoned" ) )
957- as Arc < dyn Fn ( ) -> Instant + Send + Sync >
958971 } ;
959972 let ( request_sender, request_receiver) = mpsc:: channel ( ) ;
960973 let ( response_sender, response_receiver) = mpsc:: channel ( ) ;
974+ let client = PetClient {
975+ process : None ,
976+ stdin : Box :: new ( ChannelWriter {
977+ sender : request_sender,
978+ buffered : Vec :: new ( ) ,
979+ clock : clock_value. clone ( ) ,
980+ write_delay,
981+ } ) ,
982+ stdout : Box :: new ( BufReader :: new ( TimedChannelReader {
983+ receiver : response_receiver,
984+ current : Cursor :: new ( Vec :: new ( ) ) ,
985+ clock : clock_value,
986+ } ) ) ,
987+ stderr_tail : Arc :: new ( Mutex :: new ( VecDeque :: new ( ) ) ) ,
988+ interpreter_probe_timeouts : Arc :: new ( Mutex :: new ( BTreeMap :: new ( ) ) ) ,
989+ stderr_handle : None ,
990+ state : Arc :: new ( SharedState :: new ( false ) ) ,
991+ start_time : base,
992+ clock,
993+ } ;
994+ ( client, request_receiver, response_sender)
995+ }
996+
997+ #[ test]
998+ fn refresh_path_measures_client_round_trip_and_resets_ttfe_per_request ( ) {
999+ let base = Instant :: now ( ) ;
1000+ let ( mut client, request_receiver, response_sender) =
1001+ timed_fixture_client ( base, Duration :: from_millis ( 5 ) ) ;
9611002 let fixture = thread:: spawn ( move || {
9621003 for ( environment_at_ms, response_at_ms, server_duration_ms) in [ ( 40 , 80 , 2 ) , ( 7 , 50 , 1 ) ] {
9631004 let request = request_receiver
@@ -1002,26 +1043,6 @@ fn refresh_path_measures_client_round_trip_and_resets_ttfe_per_request() {
10021043 }
10031044 } ) ;
10041045
1005- let state = Arc :: new ( SharedState :: new ( false ) ) ;
1006- let mut client = PetClient {
1007- process : None ,
1008- stdin : Box :: new ( ChannelWriter {
1009- sender : request_sender,
1010- buffered : Vec :: new ( ) ,
1011- } ) ,
1012- stdout : Box :: new ( BufReader :: new ( TimedChannelReader {
1013- receiver : response_receiver,
1014- current : Cursor :: new ( Vec :: new ( ) ) ,
1015- clock : clock_value,
1016- } ) ) ,
1017- stderr_tail : Arc :: new ( Mutex :: new ( VecDeque :: new ( ) ) ) ,
1018- interpreter_probe_timeouts : Arc :: new ( Mutex :: new ( BTreeMap :: new ( ) ) ) ,
1019- stderr_handle : None ,
1020- state,
1021- start_time : base,
1022- clock,
1023- } ;
1024-
10251046 let ( first_result, first_round_trip) =
10261047 client. refresh ( None ) . expect ( "first refresh must succeed" ) ;
10271048 assert_eq ! ( first_result. duration, 2 ) ;
@@ -1048,6 +1069,117 @@ fn refresh_path_measures_client_round_trip_and_resets_ttfe_per_request() {
10481069 fixture. join ( ) . expect ( "fixture server must finish" ) ;
10491070}
10501071
1072+ #[ test]
1073+ fn refresh_timing_window_closes_before_later_notifications ( ) {
1074+ for ( mut response, expected_error) in [
1075+ ( json ! ( { "result" : { "duration" : 2 } } ) , None ) ,
1076+ (
1077+ json ! ( { "error" : { "code" : -4 , "message" : "fixture refresh failed" } } ) ,
1078+ Some ( "JSONRPC error" ) ,
1079+ ) ,
1080+ (
1081+ json ! ( { "result" : { "duration" : "invalid" } } ) ,
1082+ Some ( "Failed to parse refresh result" ) ,
1083+ ) ,
1084+ ] {
1085+ let base = Instant :: now ( ) ;
1086+ let ( mut client, request_receiver, response_sender) =
1087+ timed_fixture_client ( base, Duration :: ZERO ) ;
1088+ let fixture = thread:: spawn ( move || {
1089+ for ( method, received_at_ms) in [ ( "refresh" , 10 ) , ( "info" , 30 ) ] {
1090+ let request = request_receiver. recv ( ) . expect ( "client must send request" ) ;
1091+ let request = read_jsonrpc_message ( & mut BufReader :: new ( Cursor :: new ( request) ) )
1092+ . expect ( "fixture must parse request" ) ;
1093+ assert_eq ! ( request. get( "method" ) . and_then( Value :: as_str) , Some ( method) ) ;
1094+ if method == "info" {
1095+ response_sender
1096+ . send ( TimedChunk {
1097+ received_at : base + Duration :: from_millis ( 20 ) ,
1098+ bytes : frame_jsonrpc ( & json ! ( {
1099+ "jsonrpc" : "2.0" ,
1100+ "method" : "environment" ,
1101+ "params" : { "executable" : "python" }
1102+ } ) ) ,
1103+ } )
1104+ . expect ( "client must receive late environment" ) ;
1105+ response = json ! ( { "result" : { } } ) ;
1106+ }
1107+ response[ "jsonrpc" ] = json ! ( "2.0" ) ;
1108+ response[ "id" ] = request. get ( "id" ) . cloned ( ) . expect ( "request must have an id" ) ;
1109+ response_sender
1110+ . send ( TimedChunk {
1111+ received_at : base + Duration :: from_millis ( received_at_ms) ,
1112+ bytes : frame_jsonrpc ( & response) ,
1113+ } )
1114+ . expect ( "client must receive response" ) ;
1115+ }
1116+ } ) ;
1117+ let result = client. refresh ( None ) ;
1118+ if let Some ( expected_error) = expected_error {
1119+ assert ! ( result
1120+ . expect_err( "refresh must fail" )
1121+ . contains( expected_error) ) ;
1122+ } else {
1123+ result. expect ( "refresh must succeed" ) ;
1124+ }
1125+ assert ! ( client. request_time_to_first_env( ) . is_none( ) ) ;
1126+ client
1127+ . send_request ( "info" , json ! ( { } ) )
1128+ . expect ( "info must succeed" ) ;
1129+ assert ! ( client. request_time_to_first_env( ) . is_none( ) ) ;
1130+ assert_eq ! (
1131+ client. startup_time_to_first_env( ) ,
1132+ Some ( Duration :: from_millis( 20 ) )
1133+ ) ;
1134+ assert_eq ! ( client. get_environments( ) . len( ) , 1 ) ;
1135+ fixture. join ( ) . expect ( "fixture server must finish" ) ;
1136+ }
1137+ }
1138+
1139+ #[ test]
1140+ fn refresh_timing_window_closes_after_transport_errors ( ) {
1141+ for write_fails in [ true , false ] {
1142+ let base = Instant :: now ( ) ;
1143+ let ( mut client, request_receiver, response_sender) =
1144+ timed_fixture_client ( base, Duration :: ZERO ) ;
1145+ drop ( response_sender) ;
1146+ if write_fails {
1147+ drop ( request_receiver) ;
1148+ }
1149+ let error = client. refresh ( None ) . expect_err ( "transport must fail" ) ;
1150+ if write_fails {
1151+ assert ! ( error. contains( "Failed to flush stdin" ) ) ;
1152+ } else {
1153+ assert ! ( error. contains( "fixture stopped" ) ) ;
1154+ }
1155+ client
1156+ . state
1157+ . handle_notification_at (
1158+ "environment" ,
1159+ json ! ( { "executable" : "python" } ) ,
1160+ base + Duration :: from_millis ( 20 ) ,
1161+ )
1162+ . unwrap ( ) ;
1163+ assert ! ( client. request_time_to_first_env( ) . is_none( ) ) ;
1164+ }
1165+ }
1166+
1167+ fn required_request_ttfe ( client : & PetClient , workload : & str , iteration : usize ) -> Duration {
1168+ client. request_time_to_first_env ( ) . unwrap_or_else ( || {
1169+ panic ! (
1170+ "{workload} iteration {} produced no environment notification" ,
1171+ iteration + 1
1172+ )
1173+ } )
1174+ }
1175+
1176+ #[ test]
1177+ #[ should_panic( expected = "Full refresh iteration 1 produced no environment notification" ) ]
1178+ fn required_request_ttfe_rejects_missing_samples ( ) {
1179+ let ( client, _requests, _responses) = timed_fixture_client ( Instant :: now ( ) , Duration :: ZERO ) ;
1180+ required_request_ttfe ( & client, "Full refresh" , 0 ) ;
1181+ }
1182+
10511183#[ test]
10521184fn malformed_environment_cannot_become_a_first_environment_sample ( ) {
10531185 let state = SharedState :: new ( false ) ;
@@ -1261,9 +1393,8 @@ fn test_full_refresh_performance() {
12611393 server_duration_stats. add ( result. duration ) ;
12621394 client_duration_stats. add ( client_elapsed. as_millis ( ) ) ;
12631395
1264- if let Some ( time_to_first) = client. request_time_to_first_env ( ) {
1265- request_time_to_first_env_stats. add ( time_to_first. as_millis ( ) ) ;
1266- }
1396+ let time_to_first = required_request_ttfe ( & client, "Full refresh" , i) ;
1397+ request_time_to_first_env_stats. add ( time_to_first. as_millis ( ) ) ;
12671398
12681399 // Track counts from last iteration
12691400 env_count = environments. len ( ) ;
@@ -1290,9 +1421,7 @@ fn test_full_refresh_performance() {
12901421 println ! ( ) ;
12911422 server_duration_stats. print_summary ( "Server duration" ) ;
12921423 client_duration_stats. print_summary ( "Client duration" ) ;
1293- if request_time_to_first_env_stats. count ( ) > 0 {
1294- request_time_to_first_env_stats. print_summary ( "Request-to-first environment" ) ;
1295- }
1424+ request_time_to_first_env_stats. print_summary ( "Request-to-first environment" ) ;
12961425 println ! ( "Environments discovered: {}" , env_count) ;
12971426 println ! ( "Managers discovered: {}" , manager_count) ;
12981427 println ! ( "Environment kinds: {:?}" , kind_counts) ;
@@ -1693,12 +1822,7 @@ fn test_performance_summary() {
16931822 "Cold refresh" ,
16941823 iteration,
16951824 ) ;
1696- let cold_request_ttfe = cold_client. request_time_to_first_env ( ) . unwrap_or_else ( || {
1697- panic ! (
1698- "Cold refresh iteration {} produced no environment notification" ,
1699- iteration + 1
1700- )
1701- } ) ;
1825+ let cold_request_ttfe = required_request_ttfe ( & cold_client, "Cold refresh" , iteration) ;
17021826 let cold_startup_ttfe = cold_client. startup_time_to_first_env ( ) . unwrap_or_else ( || {
17031827 panic ! (
17041828 "Cold refresh iteration {} produced no startup-relative TTFE" ,
@@ -1742,12 +1866,7 @@ fn test_performance_summary() {
17421866 "Warm refresh" ,
17431867 iteration,
17441868 ) ;
1745- let warm_request_ttfe = warm_client. request_time_to_first_env ( ) . unwrap_or_else ( || {
1746- panic ! (
1747- "Warm refresh iteration {} produced no environment notification" ,
1748- iteration + 1
1749- )
1750- } ) ;
1869+ let warm_request_ttfe = required_request_ttfe ( & warm_client, "Warm refresh" , iteration) ;
17511870 let warm_startup_ttfe = warm_client. startup_time_to_first_env ( ) . unwrap_or_else ( || {
17521871 panic ! (
17531872 "Warm refresh iteration {} produced no startup-relative TTFE" ,
0 commit comments