fix(gateway): keep players connected when a move's session retires before activation

crates/chunk-bench/src/control.rs

@@ -175,6 +175,7 @@ impl Client {
         }
         let activated: ActivateResult = self.call("activate", &operation, &()).await?;
         ensure!(!activated.waiting, "activation waiting for a roster");
+        ensure!(!activated.retired, "synthetic player's session retired before activation");
         let mut claims = self.gateway.claims.clone();
         claims
             .wait_for(|claims| claims.get(&operation) == Some(&ClaimPhase::Arrived))

crates/chunk-control/src/delivery.rs

@@ -13,18 +13,20 @@ use crate::{
 impl Control {
     /// Records admission intent for the claim `identity` names; native Minecraft login attaches the prepared delivery.
     /// # Errors
-    /// Rejects stale identity and retains authority after ambiguous runtime replies. Reports `Busy` until recovery
-    /// has fenced surviving JVMs.
+    /// Rejects stale identity and retains authority after ambiguous runtime replies. Rejects with
+    /// [`crate::DESTINATION_RETIRED`] a claim whose session retired before admitting it, unless control canceled it
+    /// first or a newer claim owns its player. Reports `Busy` until recovery has fenced surviving JVMs.
     pub async fn activate(&self, identity: ClaimIdentity) -> Result<Assignment> {
         self.admit()?;
         let operation = self.operation(&identity.operation_id)?;
         let _guard = operation.lock().await;
         let admitted = self.update(|state| {
+            // Checked before the move's source: once that is fenced and this claim released, nothing owns the player.
+            if lost_to_retirement(state, &identity) {
+                return Err(Error::Invalid(crate::DESTINATION_RETIRED));
+            }
             crate::moves::authorize_destination(state, &identity)?;
             let claim = state.claims.get(&identity.operation_id).ok_or(Error::Invalid("unknown claim"))?;
-            if claim.phase == Phase::Reserved && state.sessions[&claim.session].retired {
-                return Err(Error::Invalid("destination draining"));
-            }
             if claim.identity(&identity.operation_id) != identity
                 || claim.assignment.is_none()
                 || state.players.get(&claim.player).and_then(|p| p.current.as_ref()) != Some(&identity.operation_id)
@@ -84,10 +86,15 @@ impl Control {
         let host = state.sessions.get(&claim.session).ok_or(Error::Invalid("missing session"))?.host.clone();
         // The claim may have been released since it was read, such as with its host's capacity.
         let released = self.update(|s| {
-            let claim = s.claims.get_mut(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
+            let claim = s.claims.get(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
             if claim.phase == Phase::Released {
                 return Ok(true);
             }
+            // A claim its JVM already withdraws, or whose session retired, ends for that reason instead.
+            let ending = claim.phase == Phase::Withdrawing
+                || s.sessions.get(&claim.session).is_some_and(|session| session.retired);
+            let claim = s.claims.get_mut(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
+            claim.canceled |= !ending;
             set_phase(claim, Phase::Withdrawing).map(|()| false)
         })?;
         if released {
@@ -228,6 +235,24 @@ pub(crate) fn apply(state: &mut State, host: &str, status: &JvmDeliveryStatus) -
     }
 }
 
+/// Whether the claim `identity` names is lost to its session's retirement and still its player's newest.
+fn lost_to_retirement(state: &State, identity: &ClaimIdentity) -> bool {
+    let operation = &identity.operation_id;
+    let Some(claim) = state.claims.get(operation) else { return false };
+    let superseded = state.players.get(&claim.player).is_some_and(|owner| {
+        [&owner.current, &owner.pending].into_iter().flatten().any(|other| {
+            other != operation && state.claims.get(other).is_some_and(|other| other.generation > claim.generation)
+        })
+    });
+    claim.identity(operation) == *identity && retired(state, claim) && !superseded
+}
+
+/// Whether `claim` never admitted its player and its session retired, without control canceling it while that session
+/// still ran. Read from the session, so it holds however the claim itself ended; a session no longer kept has finished.
+pub(crate) fn retired(state: &State, claim: &Claim) -> bool {
+    !claim.activated && !claim.canceled && state.sessions.get(&claim.session).is_none_or(|session| session.retired)
+}
+
 pub(crate) fn release(state: &mut State, operation: &str) -> Result<()> {
     let claim = state.claims.get_mut(operation).ok_or(Error::Invalid("unknown claim"))?;
     set_phase(claim, Phase::Released)?;

crates/chunk-control/src/lib.rs

@@ -74,6 +74,10 @@ pub const ALREADY_OWNED: &str = "player already owned";
 /// Control's reply to an activation while other members of its roster have yet to activate. Nothing was admitted.
 pub const ROSTER_WAITING: &str = "roster awaiting members";
 
+/// Control's reply to claiming or activating a claim whose session retired before admitting its player. Nothing was
+/// admitted, and the claim only ends.
+pub const DESTINATION_RETIRED: &str = "destination session retired";
+
 /// Control's reply to an operation ID first used for a different claim request.
 pub const OPERATION_CHANGED: &str = "claim operation changed";
 

crates/chunk-control/src/placement.rs

@@ -17,8 +17,9 @@ pub(crate) use select::{admit_move, select_room};
 impl Control {
     /// Reserves capacity durably, coalesces demand, and prepares a non-active delivery.
     /// # Errors
-    /// Rejects duplicate membership, changed operations, unknown session types and unresolved hosts. Reports `Busy`
-    /// until recovery has fenced surviving JVMs, even for a restored claim: the log may have lost its cancellation.
+    /// Rejects duplicate membership, changed operations, unknown session types and unresolved hosts, and with
+    /// [`crate::DESTINATION_RETIRED`] a reservation whose session retired. Reports `Busy` until recovery has fenced
+    /// surviving JVMs, even for a restored claim: the log may have lost its cancellation.
     pub async fn claim(&self, request: ClaimRequest) -> Result<Assignment> {
         self.claim_approved(request, "").await
     }
@@ -51,6 +52,9 @@ impl Control {
         let state = self.state()?;
         let claim = state.claims.get(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
         claim.matches(&request)?;
+        if crate::delivery::retired(&state, claim) {
+            return Err(Error::Invalid(crate::DESTINATION_RETIRED));
+        }
         if matches!(claim.phase, Phase::Withdrawing | Phase::Released) {
             return Err(Error::Invalid("claim closed"));
         }
@@ -68,9 +72,25 @@ impl Control {
             return Ok(assignment);
         }
         // Control's desired state already asks the JVM for this session, and its topic for this delivery.
-        self.session_ready(&session.host, &runtime, &claim.session).await?;
+        let prepared = async {
+            self.session_ready(&session.host, &runtime, &claim.session).await?;
+            self.prepared(&runtime, &request.operation_id, claim.generation).await
+        };
+        let preparation = match prepared.await {
+            Ok(preparation) => preparation,
+            // The session's retirement is what stopped or closed the delivery. Its report reaches the JVM's link
+            // within its commit, so taking the writer once waits for that commit to publish.
+            Err(error) => {
+                drop(self.authority.writer()?);
+                let state = self.state()?;
+                let retired = state
+                    .claims
+                    .get(&request.operation_id)
+                    .is_some_and(|claim| crate::delivery::retired(&state, claim));
+                return Err(if retired { Error::Invalid(crate::DESTINATION_RETIRED) } else { error });
+            }
+        };
         let mut delivery = delivery(deployment, &runtime, &config, claim, &request);
-        let preparation = self.prepared(&runtime, &request.operation_id, claim.generation).await?;
         delivery.identity = None;
         let assigned_at_ms = crate::now_ms();
         let assignment = Assignment {
@@ -87,6 +107,10 @@ impl Control {
             if state.released(&session.host) {
                 return Err(Error::Stopped);
             }
+            let claim = state.claims.get(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
+            if crate::delivery::retired(state, claim) {
+                return Err(Error::Invalid(crate::DESTINATION_RETIRED));
+            }
             let claim = state.claims.get_mut(&request.operation_id).ok_or(Error::Invalid("unknown claim"))?;
             if claim.phase != Phase::Reserved {
                 return Err(Error::Invalid("claim no longer reserved"));
@@ -306,6 +330,7 @@ pub(crate) fn insert_claim(
             assigned_at_ms: None,
             released_at_ms: None,
             roster,
+            canceled: false,
         },
     );
     Ok(())

crates/chunk-control/src/recovery.rs

@@ -319,5 +319,6 @@ fn tombstone(status: &JvmDeliveryStatus) -> Claim {
         assigned_at_ms: None,
         released_at_ms: Some(now),
         roster: None,
+        canceled: false,
     }
 }

crates/chunk-control/src/state/entities.rs

@@ -350,6 +350,8 @@ pub(crate) struct Claim {
     pub released_at_ms: Option<u64>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub roster: Option<String>,
+    /// Whether control began withdrawing the claim while its session still ran.
+    pub canceled: bool,
 }
 
 impl Claim {

crates/chunk-control/src/state/store.rs

@@ -248,7 +248,7 @@ fn schema() -> DatabaseSchema {
             "generation": generation, "session": string, "phase": {"type": "enum", "values": Phase::NAMES},
             "assignment?": string, "activated": boolean, "created_at_ms": integer, "assigned_at_ms?": integer,
             "released_at_ms?": integer,
-            "roster?": string,
+            "roster?": string, "canceled": boolean,
         },
         MOVES: {
             "request": string, "canceled": boolean, "sequence": integer,

crates/chunk-control/src/tests/gateway.rs

@@ -22,6 +22,7 @@ fn add(control: &Control, gateway: &str, operations: impl IntoIterator<Item = St
                     assigned_at_ms: None,
                     released_at_ms: None,
                     roster: None,
+                    canceled: false,
                 };
                 state.claims.insert(operation, claim);
             }

crates/chunk-control/src/tests/jvm.rs

@@ -128,6 +128,16 @@ impl FakeRuntime {
         }
     }
 
+    /// Ends `session` and closes `operations`' deliveries together, as `scope.finish()` does, so the next report
+    /// carries both.
+    pub(super) fn finish(&self, session: &str, operations: &[&str]) {
+        let mut bindings = self.bindings.lock().unwrap();
+        for operation in operations {
+            bindings.get_mut(*operation).unwrap().phase = JvmDeliveryPhase::Closed;
+        }
+        self.ended_sessions.lock().unwrap().insert(session.into());
+    }
+
     /// Prepares `operation`'s delivery, or closes it once withdrawn.
     fn delivery(&self, operation: &str, wanted: &JvmDelivery) {
         let mut bindings = self.bindings.lock().unwrap();

crates/chunk-control/src/tests/moves.rs

@@ -1,7 +1,8 @@
-//! Moves a JVM asks for, fenced to the claim of the player it holds.
+//! Moves a JVM asks for, fenced to the claim of the player it holds, and moves whose destination retires.
 
 use super::*;
 use chunk_contract::MoveRefusal;
+use chunk_proto::control::v1::ClaimIdentity;
 
 #[tokio::test]
 async fn a_jvm_moves_only_players_it_hosts_under_their_current_generation() {
@@ -35,3 +36,142 @@ async fn a_jvm_moves_only_players_it_hosts_under_their_current_generation() {
     assert_eq!(pending_move(&control, &source).map(|moved| moved.operation_id).as_deref(), Some("move"));
     fixture.close().await;
 }
+
+/// Logs `player` in as `source`, queues their move to an arena as `move` and reserves it.
+async fn moving(fixture: &Fixture, control: &Control, player: &str) -> (ClaimRequest, ClaimRequest, ClaimIdentity) {
+    let source = request("source", player);
+    let assignment = control.claim(source.clone()).await.unwrap();
+    fixture.arrive(control, "source").await;
+    control.activate(assignment.claim.unwrap()).await.unwrap();
+    let demand = SessionDemand { key: "arena".into(), ..source.demand.clone().unwrap() };
+    let command = MoveRequest { operation_id: "move".into(), player_id: player.into(), demand, source: None };
+    let destination = control.move_player(command).unwrap();
+    let identity = control.claim(destination.clone()).await.unwrap().claim.unwrap();
+    (source, destination, identity)
+}
+
+/// Retires the session of `operation`'s claim, leaving the claim in `phase`.
+fn retire(control: &Control, operation: &str, phase: Phase) {
+    control
+        .update(|state| {
+            let claim = state.claims.get_mut(operation).unwrap();
+            claim.phase = phase;
+            state.sessions.get_mut(&claim.session).unwrap().retired = true;
+            Ok(())
+        })
+        .unwrap();
+}
+
+fn retired<T>(result: &Result<T>) -> bool {
+    matches!(result, Err(Error::Invalid(crate::DESTINATION_RETIRED)))
+}
+
+#[tokio::test]
+async fn a_move_into_a_session_that_retires_before_cutover_fails_and_leaves_the_source_active() {
+    let fixture = Fixture::new();
+    let control = fixture.control().await;
+    let uuid = uuid::Uuid::new_v4().to_string();
+    let (source, destination, identity) = moving(&fixture, &control, &uuid).await;
+    retire(&control, "move", Phase::Reserved);
+
+    assert!(retired(&control.claim(destination.clone()).await));
+    assert!(retired(&control.activate(identity.clone()).await));
+    // A claim named at another generation is stale, whatever became of its session.
+    let older = ClaimIdentity { delivery_generation: identity.delivery_generation - 1, ..identity };
+    let outdated = control.activate(older).await;
+    assert!(matches!(outdated, Err(Error::Invalid(_))) && !retired(&outdated), "{outdated:?}");
+
+    control.abandon_move(destination, crate::DESTINATION_RETIRED.into()).await.unwrap();
+    control.reconcile_all().await.unwrap();
+    let state = control.state().unwrap();
+    assert!(state.claims["move"].phase == Phase::Released);
+    assert_eq!(state.players[&uuid].current.as_deref(), Some("source"));
+    assert!(state.players[&uuid].pending.is_none());
+    assert_eq!(control.inspect(&source).unwrap().phase, ClaimPhase::Arrived as i32);
+    assert_eq!(players(&control)[0].1.last_move_failure.as_ref().unwrap().reason, crate::DESTINATION_RETIRED);
+    fixture.close().await;
+}
+
+#[tokio::test]
+async fn a_destination_that_retires_after_its_source_was_fenced_frees_the_player_to_log_in() {
+    let fixture = Fixture::new();
+    let control = fixture.control().await;
+    let uuid = uuid::Uuid::new_v4().to_string();
+    let (source, destination, identity) = moving(&fixture, &control, &uuid).await;
+    control.cancel(source).await.unwrap();
+
+    // A claim canceled while its session still runs is stale, and stays so when the session retires afterwards.
+    control.cancel(destination).await.unwrap();
+    let stale = control.activate(identity.clone()).await;
+    assert!(matches!(stale, Err(Error::Invalid(_))) && !retired(&stale), "{stale:?}");
+    retire(&control, "move", Phase::Released);
+    let canceled = control.activate(identity).await;
+    assert!(matches!(canceled, Err(Error::Invalid(_))) && !retired(&canceled), "{canceled:?}");
+
+    assert!(!control.state().unwrap().players.contains_key(&uuid));
+    control.claim(request("again", &uuid)).await.unwrap();
+    fixture.close().await;
+}
+
+#[tokio::test]
+async fn reconciliation_ending_a_claim_in_a_retired_session_is_reported_until_a_newer_claim_owns_the_player() {
+    let fixture = Fixture::new();
+    let control = fixture.control().await;
+    let uuid = uuid::Uuid::new_v4().to_string();
+    let (source, destination, identity) = moving(&fixture, &control, &uuid).await;
+    control.cancel(source).await.unwrap();
+    retire(&control, "move", Phase::Reserved);
+    control.reconcile_all().await.unwrap();
+    let claim = control.state().unwrap().claims["move"].clone();
+    assert!(!claim.canceled, "{:?}", claim.phase as u8);
+    assert!(retired(&control.activate(identity.clone()).await));
+
+    control.claim(request("again", &uuid)).await.unwrap();
+    let superseded = control.activate(identity).await;
+    assert!(matches!(superseded, Err(Error::Invalid(_))) && !retired(&superseded), "{superseded:?}");
+    drop(destination);
+    fixture.close().await;
+}
+
+#[tokio::test]
+async fn a_report_closing_a_delivery_and_ending_its_session_together_leaves_the_move_retired() {
+    let fixture = Fixture::new();
+    let control = fixture.control().await;
+    let uuid = uuid::Uuid::new_v4().to_string();
+    let (source, _, identity) = moving(&fixture, &control, &uuid).await;
+    control.cancel(source).await.unwrap();
+    let session = control.state().unwrap().claims["move"].session.clone();
+
+    fixture.runtime.finish(&session, &["move"]);
+    eventually(|| {
+        let state = control.state().unwrap();
+        state.claims["move"].phase == Phase::Released && state.sessions[&session].retired
+    })
+    .await;
+    assert!(retired(&control.activate(identity.clone()).await));
+    // The claim outlives its finished session's row.
+    control.reconcile_all().await.unwrap();
+    assert!(!control.state().unwrap().sessions.contains_key(&session));
+    assert!(retired(&control.activate(identity).await));
+    fixture.close().await;
+}
+
+#[tokio::test]
+async fn a_login_whose_delivery_closes_with_its_session_while_control_prepares_it_is_retired() {
+    let fixture = Fixture::new();
+    fixture.runtime.stalled_preparation.store(true, Ordering::Release);
+    let control = fixture.control().await;
+    let login = request("login", &uuid::Uuid::new_v4().to_string());
+    let claiming = tokio::spawn({
+        let (control, login) = (control.clone(), login.clone());
+        async move { control.claim(login).await }
+    });
+    eventually(|| fixture.runtime.bindings.lock().unwrap().contains_key("login")).await;
+    let session = control.state().unwrap().claims["login"].session.clone();
+
+    fixture.runtime.finish(&session, &["login"]);
+    let claimed = claiming.await.unwrap();
+    assert!(retired(&claimed), "{claimed:?}");
+    assert!(retired(&control.claim(login).await));
+    fixture.close().await;
+}

crates/chunk-control/src/tests/recovery.rs

@@ -170,6 +170,7 @@ fn recovery_fences_its_hosts_unassigned_reservations_but_never_another_hosts() {
             assigned_at_ms: None,
             released_at_ms: None,
             roster: None,
+            canceled: false,
         };
         state.claims.insert(host.into(), claim);
     }