@@ -89,11 +89,23 @@ fn media_preprocessing_permits(reservation_bytes: u64) -> u32 {
8989 . expect ( "media preprocessing reservation exceeds semaphore capacity type" )
9090}
9191
92+ /// The scan roots yu-server last told us about, paired with the `generation`
93+ /// that delivered them. Both live under one lock on purpose: the
94+ /// compare-and-apply in `scan_roots_changed` must not interleave with another
95+ /// request's, and a separate counter would let two calls both pass the
96+ /// staleness check and then land in the wrong order.
97+ pub struct ScanRoots {
98+ pub roots : Vec < PathBuf > ,
99+ /// Highest generation applied so far. yu-server's counter starts at 1, so
100+ /// 0 means nothing but the startup contract has been applied yet.
101+ pub generation : u64 ,
102+ }
103+
92104#[ derive( Clone ) ]
93105pub struct AppState {
94106 pub started_at : std:: time:: Instant ,
95107 pub instance_id : String ,
96- pub scan_roots : Arc < RwLock < Vec < PathBuf > > > ,
108+ pub scan_roots : Arc < RwLock < ScanRoots > > ,
97109 pub auth_token : String ,
98110 pub wd_cache_dir : PathBuf ,
99111 pub wd_infer : Arc < RwLock < HashMap < String , Arc < infer_core:: WdInferEngine > > > > ,
@@ -151,6 +163,16 @@ async fn healthz(State(state): State<AppState>) -> Json<serde_json::Value> {
151163#[ derive( Deserialize ) ]
152164pub struct ScanRootsChangedRequest {
153165 pub scan_roots : Vec < String > ,
166+ /// Monotonic counter yu-server assigns in config-write order. Its own send
167+ /// lock only orders the sends it starts; a send it gave up on (5s timeout)
168+ /// can still reach us after a newer one, so anything not strictly newer
169+ /// than what we already applied is dropped here.
170+ ///
171+ /// Absent means a yu-server predating the field: apply unconditionally,
172+ /// which is exactly what we did before, and leave the stored generation
173+ /// alone so a mixed deployment cannot wedge us.
174+ #[ serde( default ) ]
175+ pub generation : Option < u64 > ,
154176}
155177
156178async fn scan_roots_changed (
@@ -166,13 +188,23 @@ async fn scan_roots_changed(
166188 . into_response ( ) ;
167189 }
168190
169- let roots = body
191+ // Canonicalize before taking the lock -- it hits the filesystem.
192+ let roots: Vec < PathBuf > = body
170193 . scan_roots
171194 . into_iter ( )
172195 . filter_map ( |path| std:: fs:: canonicalize ( path) . ok ( ) )
173196 . collect ( ) ;
174- * state. scan_roots . write ( ) . unwrap ( ) = roots;
175- Json ( json ! ( { "ok" : true } ) ) . into_response ( )
197+
198+ let mut scan_roots = state. scan_roots . write ( ) . unwrap ( ) ;
199+ if let Some ( generation) = body. generation {
200+ if generation <= scan_roots. generation {
201+ return Json ( json ! ( { "ok" : true , "applied" : false , "stale" : true } ) ) . into_response ( ) ;
202+ }
203+ scan_roots. generation = generation;
204+ }
205+ scan_roots. roots = roots;
206+ drop ( scan_roots) ;
207+ Json ( json ! ( { "ok" : true , "applied" : true } ) ) . into_response ( )
176208}
177209
178210#[ derive( Deserialize ) ]
@@ -250,7 +282,11 @@ fn check_auth(state: &AppState, headers: &HeaderMap) -> bool {
250282
251283fn check_scan_roots ( state : & AppState , real_path : & std:: path:: Path ) -> bool {
252284 let scan_roots = state. scan_roots . read ( ) . unwrap ( ) ;
253- !scan_roots. is_empty ( ) && scan_roots. iter ( ) . any ( |root| real_path. starts_with ( root) )
285+ !scan_roots. roots . is_empty ( )
286+ && scan_roots
287+ . roots
288+ . iter ( )
289+ . any ( |root| real_path. starts_with ( root) )
254290}
255291
256292fn check_model_id ( model_id : & str ) -> bool {
@@ -1771,7 +1807,10 @@ mod tests {
17711807 AppState {
17721808 started_at : std:: time:: Instant :: now ( ) ,
17731809 instance_id : "test-instance" . to_string ( ) ,
1774- scan_roots : Arc :: new ( RwLock :: new ( scan_roots) ) ,
1810+ scan_roots : Arc :: new ( RwLock :: new ( ScanRoots {
1811+ roots : scan_roots,
1812+ generation : 0 ,
1813+ } ) ) ,
17751814 auth_token : "test-token" . to_string ( ) ,
17761815 wd_cache_dir : std:: env:: temp_dir ( ) . join ( "yu-infer-test-cache" ) ,
17771816 wd_infer : Arc :: new ( RwLock :: new ( HashMap :: new ( ) ) ) ,
@@ -2083,6 +2122,100 @@ mod tests {
20832122 assert_eq ! ( decoded. height( ) , 64 ) ;
20842123 }
20852124
2125+ async fn scan_roots_changed_request (
2126+ app : axum:: Router ,
2127+ roots : & [ & std:: path:: Path ] ,
2128+ generation : Option < u64 > ,
2129+ ) -> StatusCode {
2130+ let mut payload = json ! ( {
2131+ "scan_roots" : roots. iter( ) . map( |p| p. to_str( ) . unwrap( ) ) . collect:: <Vec <_>>( ) ,
2132+ } ) ;
2133+ if let Some ( generation) = generation {
2134+ payload[ "generation" ] = json ! ( generation) ;
2135+ }
2136+ app. oneshot (
2137+ Request :: builder ( )
2138+ . method ( "POST" )
2139+ . uri ( "/_internal/scan-roots-changed" )
2140+ . header ( "content-type" , "application/json" )
2141+ . header ( "authorization" , "Bearer test-token" )
2142+ . body ( Body :: from ( payload. to_string ( ) ) )
2143+ . unwrap ( ) ,
2144+ )
2145+ . await
2146+ . unwrap ( )
2147+ . status ( )
2148+ }
2149+
2150+ #[ tokio:: test]
2151+ async fn scan_roots_changed_drops_a_generation_it_already_passed ( ) {
2152+ let base = std:: env:: temp_dir ( ) . join ( "yu-infer-scan-roots-generation-test" ) ;
2153+ let newer = base. join ( "newer" ) ;
2154+ let older = base. join ( "older" ) ;
2155+ std:: fs:: create_dir_all ( & newer) . unwrap ( ) ;
2156+ std:: fs:: create_dir_all ( & older) . unwrap ( ) ;
2157+ let canonical_newer = std:: fs:: canonicalize ( & newer) . unwrap ( ) ;
2158+
2159+ let state = test_state ( vec ! [ ] ) ;
2160+
2161+ // Generation 2 lands first; the send that carried generation 1 timed
2162+ // out on yu-server's side and only reaches us afterwards.
2163+ let status =
2164+ scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & newer] , Some ( 2 ) ) . await ;
2165+ assert_eq ! ( status, StatusCode :: OK ) ;
2166+ let status =
2167+ scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & older] , Some ( 1 ) ) . await ;
2168+ assert_eq ! ( status, StatusCode :: OK ) ;
2169+ assert_eq ! (
2170+ state. scan_roots. read( ) . unwrap( ) . roots,
2171+ vec![ canonical_newer. clone( ) ] ,
2172+ "a stale generation must not clobber the newer roots"
2173+ ) ;
2174+
2175+ // An equal generation is a replay, not news, so it is dropped too.
2176+ let status =
2177+ scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & older] , Some ( 2 ) ) . await ;
2178+ assert_eq ! ( status, StatusCode :: OK ) ;
2179+ assert_eq ! (
2180+ state. scan_roots. read( ) . unwrap( ) . roots,
2181+ vec![ canonical_newer]
2182+ ) ;
2183+
2184+ // A strictly newer generation still applies.
2185+ let status =
2186+ scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & older] , Some ( 3 ) ) . await ;
2187+ assert_eq ! ( status, StatusCode :: OK ) ;
2188+ assert_eq ! (
2189+ state. scan_roots. read( ) . unwrap( ) . roots,
2190+ vec![ std:: fs:: canonicalize( & older) . unwrap( ) ]
2191+ ) ;
2192+ assert_eq ! ( state. scan_roots. read( ) . unwrap( ) . generation, 3 ) ;
2193+ }
2194+
2195+ #[ tokio:: test]
2196+ async fn scan_roots_changed_applies_a_request_without_a_generation ( ) {
2197+ // A yu-server predating the field must keep working, and must not
2198+ // move the stored generation.
2199+ let base = std:: env:: temp_dir ( ) . join ( "yu-infer-scan-roots-legacy-test" ) ;
2200+ let first = base. join ( "first" ) ;
2201+ let later = base. join ( "later" ) ;
2202+ std:: fs:: create_dir_all ( & first) . unwrap ( ) ;
2203+ std:: fs:: create_dir_all ( & later) . unwrap ( ) ;
2204+ let state = test_state ( vec ! [ ] ) ;
2205+
2206+ let status =
2207+ scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & first] , Some ( 7 ) ) . await ;
2208+ assert_eq ! ( status, StatusCode :: OK ) ;
2209+ let status = scan_roots_changed_request ( build_router ( state. clone ( ) ) , & [ & later] , None ) . await ;
2210+ assert_eq ! ( status, StatusCode :: OK ) ;
2211+ assert_eq ! (
2212+ state. scan_roots. read( ) . unwrap( ) . roots,
2213+ vec![ std:: fs:: canonicalize( & later) . unwrap( ) ] ,
2214+ "a request without a generation must still apply"
2215+ ) ;
2216+ assert_eq ! ( state. scan_roots. read( ) . unwrap( ) . generation, 7 ) ;
2217+ }
2218+
20862219 #[ tokio:: test]
20872220 async fn infer_wd_rejects_missing_auth_header ( ) {
20882221 let app = build_router ( test_state ( vec ! [ ] ) ) ;
0 commit comments