fix: harden OpenSim session readiness

crates/metacrate-grid-agent/src/session.rs

@@ -20,6 +20,7 @@ const MAX_OFFLINE_WORK: usize = 1_024;
 const MAX_SEEN_WORK: usize = 4_096;
 const MAX_BACKOFF: Duration = Duration::from_hours(1);
 const MAX_STABLE_RESET: Duration = Duration::from_hours(24);
+const MAX_READINESS_TIMEOUT: Duration = Duration::from_mins(10);
 const MAX_SHUTDOWN: Duration = Duration::from_mins(1);
 
 pub type SessionFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
@@ -201,6 +202,7 @@ pub trait GridSessionBackend: Send + Sync + 'static {
 pub struct ReconnectPolicy {
     pub initial_delay: Duration,
     pub maximum_delay: Duration,
+    pub readiness_timeout: Duration,
     pub stable_reset_after: Duration,
     pub shutdown_deadline: Duration,
     pub jitter_basis_points: u16,
@@ -213,6 +215,7 @@ impl Default for ReconnectPolicy {
         Self {
             initial_delay: Duration::from_secs(1),
             maximum_delay: Duration::from_mins(1),
+            readiness_timeout: Duration::from_secs(30),
             stable_reset_after: Duration::from_mins(2),
             shutdown_deadline: Duration::from_secs(10),
             jitter_basis_points: 2_000,
@@ -237,6 +240,8 @@ impl ReconnectPolicy {
         if self.initial_delay.is_zero()
             || self.initial_delay > self.maximum_delay
             || self.maximum_delay > MAX_BACKOFF
+            || self.readiness_timeout.is_zero()
+            || self.readiness_timeout > MAX_READINESS_TIMEOUT
             || self.stable_reset_after.is_zero()
             || self.stable_reset_after > MAX_STABLE_RESET
             || self.shutdown_deadline.is_zero()
@@ -608,6 +613,7 @@ enum LoginEvent {
 enum ActiveEvent {
     Cancelled,
     Command(Option<Command>),
+    ReadinessTimedOut,
     Signal(SessionSignal),
 }
 
@@ -635,6 +641,7 @@ impl Runtime {
         let mut desired = DesiredState::Running;
         let mut session: Option<Box<dyn GridSession>> = None;
         let mut ready = false;
+        let mut readiness_deadline = None;
         let mut stable_since = None;
 
         loop {
@@ -711,6 +718,8 @@ impl Runtime {
                         Ok(connected) if connected.generation() == generation => {
                             session = Some(connected);
                             ready = false;
+                            readiness_deadline =
+                                Some(Instant::now() + self.policy.readiness_timeout);
                             stable_since = None;
                             self.transition(
                                 SessionState::Degraded,
@@ -756,9 +765,18 @@ impl Runtime {
                         };
                         let signal = active.next_signal(token);
                         tokio::pin!(signal);
+                        let readiness = async {
+                            if let Some(deadline) = readiness_deadline {
+                                tokio::time::sleep_until(deadline).await;
+                            } else {
+                                std::future::pending::<()>().await;
+                            }
+                        };
+                        tokio::pin!(readiness);
                         tokio::select! {
                             () = self.cancellation.token().cancelled() => ActiveEvent::Cancelled,
                             command = self.commands.recv() => ActiveEvent::Command(command),
+                            () = &mut readiness => ActiveEvent::ReadinessTimedOut,
                             next = &mut signal => ActiveEvent::Signal(next),
                         }
                     };
@@ -766,6 +784,21 @@ impl Runtime {
                         ActiveEvent::Cancelled => {
                             desired = DesiredState::Shutdown;
                         }
+                        ActiveEvent::ReadinessTimedOut => {
+                            ready = false;
+                            readiness_deadline = None;
+                            stable_since = None;
+                            self.fence.invalidate();
+                            if let Some(unready) = session.take() {
+                                self.close_session(unready).await;
+                            }
+                            self.failures = self.failures.saturating_add(1);
+                            desired = self
+                                .backoff(SessionFailure::new(
+                                    SessionFailureKind::TransientTransport,
+                                ))
+                                .await;
+                        }
                         ActiveEvent::Command(command) => match command {
                             Some(Command::Submit(work, response)) => {
                                 let disposition = self.handle_work(work.clone(), ready, generation);
@@ -796,6 +829,7 @@ impl Runtime {
                         ActiveEvent::Signal(next) => match next {
                             SessionSignal::Ready => {
                                 ready = true;
+                                readiness_deadline = None;
                                 stable_since.get_or_insert_with(Instant::now);
                                 self.transition(
                                     SessionState::Online,
@@ -807,6 +841,8 @@ impl Runtime {
                             }
                             SessionSignal::Degraded => {
                                 ready = false;
+                                readiness_deadline =
+                                    Some(Instant::now() + self.policy.readiness_timeout);
                                 stable_since = None;
                                 self.transition(
                                     SessionState::Degraded,
@@ -846,6 +882,7 @@ impl Runtime {
                             self.close_session(active).await;
                         }
                         ready = false;
+                        readiness_deadline = None;
                         stable_since = None;
                     }
                 }