@@ -91,11 +91,11 @@ pub struct SupervisorCredential {
9191pub enum ObservedInstance {
9292 /// No host session is connected.
9393 NoHost ,
94- /// A host is connected but has not reconciled to the snapshot that
95- /// assigned this route .
94+ /// A host is connected, but this gateway has not recovered the route's
95+ /// assignment yet or the host has not reconciled the snapshot assigning it .
9696 Pending ,
97- /// The host reconciled to that snapshot (or the route is not assigned)
98- /// and does not host the route .
97+ /// The host reconciled to the snapshot assigning this route and does
98+ /// not host it .
9999 Absent ,
100100 /// The host's latest report of the instance.
101101 Reported ( SandboxInstanceStatus ) ,
@@ -200,7 +200,7 @@ struct Inner {
200200 desired : Mutex < Desired > ,
201201 /// Latest supervisor token the gateway minted; `None` until the first.
202202 credential : watch:: Sender < Option < SupervisorCredential > > ,
203- /// Current desired revision; 0 while uninitialized .
203+ /// Monotonic local revision, also bumped at the start of a new term .
204204 revision : watch:: Sender < u64 > ,
205205 observed : Mutex < Observed > ,
206206}
@@ -371,8 +371,8 @@ impl SupervisorHostController {
371371 lock ( & self . inner . desired ) . initialized
372372 }
373373
374- /// Current desired revision: 0 while uninitialized. Read it before
375- /// deriving a full list for [`Self::replace_assignments`]; the
374+ /// Current desired revision, including the fence for this leadership
375+ /// term. Read it before deriving a full list for [`Self::replace_assignments`]; the
376376 /// coordinator does so through
377377 /// [`SharedSupervision::begin_recovery`](crate::SharedSupervision::begin_recovery).
378378 #[ must_use]
@@ -393,8 +393,8 @@ impl SupervisorHostController {
393393 /// durable records while the revision was `baseline`.
394394 ///
395395 /// Initializes the controller; edits made before initialization are
396- /// applied on top of the list. After initialization the replacement is
397- /// compare-and-replace: when the revision moved past `baseline` while
396+ /// applied on top of the list. The replacement is compare-and-replace:
397+ /// when the revision moved past `baseline` while
398398 /// the list was being derived, nothing changes and the conflict is
399399 /// returned so the caller derives the list again.
400400 pub fn replace_assignments (
@@ -531,6 +531,16 @@ impl SupervisorHostController {
531531 if observed. session . is_none ( ) {
532532 return ObservedInstance :: NoHost ;
533533 }
534+ let Some ( assigned) = desired
535+ . assignments
536+ . get ( route)
537+ . filter ( |_| desired. initialized )
538+ else {
539+ // A create can have reached another replica before this leader's
540+ // recovery discovers its durable assignment. Never suspend that
541+ // workload based on a snapshot that did not assign it.
542+ return ObservedInstance :: Pending ;
543+ } ;
534544 let Some ( status) = & observed. status else {
535545 return ObservedInstance :: Pending ;
536546 } ;
@@ -541,11 +551,10 @@ impl SupervisorHostController {
541551 {
542552 return ObservedInstance :: Reported ( instance. clone ( ) ) ;
543553 }
544- match desired. assignments . get ( route) {
545- Some ( assigned) if assigned. revision > status. applied_revision => {
546- ObservedInstance :: Pending
547- }
548- _ => ObservedInstance :: Absent ,
554+ if assigned. revision > status. applied_revision {
555+ ObservedInstance :: Pending
556+ } else {
557+ ObservedInstance :: Absent
549558 }
550559 }
551560
@@ -1000,10 +1009,12 @@ mod tests {
10001009 assert ! ( controller. replace_assignments( old_baseline, [ ] ) . is_err( ) ) ;
10011010 // No authoritative empty snapshot before the new term's full recovery.
10021011 assert ! ( controller. snapshot( ) . is_none( ) ) ;
1012+ let ( session, revoked) = controller. begin_session ( identity. clone ( ) , & hello) . unwrap ( ) ;
1013+ controller. record_status ( session, status ( 0 , & [ ] ) ) . unwrap ( ) ;
1014+ assert_eq ! ( controller. observed( "rt-new" ) , ObservedInstance :: Pending ) ;
10031015 controller
10041016 . replace_assignments ( controller. revision ( ) , [ assignment ( "rt-new" , b"new" ) ] )
10051017 . unwrap ( ) ;
1006- let ( session, revoked) = controller. begin_session ( identity. clone ( ) , & hello) . unwrap ( ) ;
10071018 controller. record_sent ( session, controller. revision ( ) ) ;
10081019 controller
10091020 . record_status ( session, status ( controller. revision ( ) , & [ "rt-new" ] ) )
@@ -1108,7 +1119,7 @@ mod tests {
11081119 // A newer assignment is pending until the host reconciles past it.
11091120 controller. assign ( assignment ( "rt-b" , b"b" ) ) . unwrap ( ) ;
11101121 assert_eq ! ( controller. observed( "rt-b" ) , ObservedInstance :: Pending ) ;
1111- assert_eq ! ( controller. observed( "rt-unknown" ) , ObservedInstance :: Absent ) ;
1122+ assert_eq ! ( controller. observed( "rt-unknown" ) , ObservedInstance :: Pending ) ;
11121123 controller. record_sent ( current, 2 ) ;
11131124 controller. record_status ( current, status ( 2 , & [ ] ) ) . unwrap ( ) ;
11141125 assert_eq ! ( controller. observed( "rt-b" ) , ObservedInstance :: Absent ) ;
0 commit comments