From 55d8f9578ec28c89ec10164957672704a81c7dc4 Mon Sep 17 00:00:00 2001 From: oglassdev <23040613+oglassdev@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:40:41 -0400 Subject: [PATCH 1/5] fix(gateway): keep players connected when a move's session retires before activation Control refuses to claim a reservation whose session retired, so a move still preparing fails as a move and its player stays active on the source. Activation of a claim whose session retired first is reported as a distinct outcome instead of a stale activation; the gateway then withdraws the claim, records the move as failed and routes the player as a login on the same connection, since the move already fenced its source. A stale generation or a claim withdrawn while its session runs is still rejected. Fixes #292 --- crates/chunk-control/src/delivery.rs | 17 ++-- crates/chunk-control/src/lib.rs | 4 + crates/chunk-control/src/placement.rs | 11 ++- crates/chunk-control/src/tests/moves.rs | 85 ++++++++++++++++++- .../src/core/sync/platform/claims.rs | 9 +- crates/chunk-proxy/src/server/managed.rs | 70 ++++++++++++--- .../src/server/managed/commands/fixture.rs | 3 + .../managed/commands/fixture/core_service.rs | 3 +- .../src/server/managed/commands/tests.rs | 2 +- .../managed/commands/tests/placement.rs | 33 ++++++- .../chunk-proxy/src/server/managed/testing.rs | 4 +- proto/chunk/sync/v1/gateway.proto | 3 + 12 files changed, 216 insertions(+), 28 deletions(-) diff --git a/crates/chunk-control/src/delivery.rs b/crates/chunk-control/src/delivery.rs index 434f34c5..d9fa879c 100644 --- a/crates/chunk-control/src/delivery.rs +++ b/crates/chunk-control/src/delivery.rs @@ -13,18 +13,25 @@ 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`] the current claim whose session retired before admitting it, whether or not that + /// already withdrew the claim. Reports `Busy` until recovery has fenced surviving JVMs. pub async fn activate(&self, identity: ClaimIdentity) -> Result { self.admit()?; let operation = self.operation(&identity.operation_id)?; let _guard = operation.lock().await; let admitted = self.update(|state| { - 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")); + // Checked before the move's source: once that is fenced and this claim released, nothing owns the player. + if claim.identity(&identity.operation_id) == identity + && !claim.activated + && matches!(claim.phase, Phase::Reserved | Phase::Withdrawing | Phase::Released) + && state.sessions.get(&claim.session).is_some_and(|session| session.retired) + { + 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.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) diff --git a/crates/chunk-control/src/lib.rs b/crates/chunk-control/src/lib.rs index 2bc77158..f887b12f 100644 --- a/crates/chunk-control/src/lib.rs +++ b/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"; diff --git a/crates/chunk-control/src/placement.rs b/crates/chunk-control/src/placement.rs index a8873ae2..29ad2f53 100644 --- a/crates/chunk-control/src/placement.rs +++ b/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 { self.claim_approved(request, "").await } @@ -55,6 +56,9 @@ impl Control { return Err(Error::Invalid("claim closed")); } let session = state.sessions.get(&claim.session).ok_or(Error::Invalid("missing session"))?; + if claim.phase == Phase::Reserved && session.retired { + return Err(Error::Invalid(crate::DESTINATION_RETIRED)); + } let deployment = &state.host_release(&session.host)?.deployment; let runtime = self.runtime(&session.host).await?; let config = self.jvm_configuration(deployment, &runtime)?; @@ -91,6 +95,9 @@ impl Control { if claim.phase != Phase::Reserved { return Err(Error::Invalid("claim no longer reserved")); } + if state.sessions.get(&claim.session).is_some_and(|session| session.retired) { + return Err(Error::Invalid(crate::DESTINATION_RETIRED)); + } let mut persisted = assignment.clone(); persisted.configuration = None; persisted.destination = None; diff --git a/crates/chunk-control/src/tests/moves.rs b/crates/chunk-control/src/tests/moves.rs index 68c40c66..9492e823 100644 --- a/crates/chunk-control/src/tests/moves.rs +++ b/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,85 @@ 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(result: &Result) -> 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 withdrawn while its session still runs is stale. + control + .update(|state| { + state.claims.get_mut("move").unwrap().phase = Phase::Withdrawing; + Ok(()) + }) + .unwrap(); + let stale = control.activate(identity.clone()).await; + assert!(matches!(stale, Err(Error::Invalid("stale activation"))), "{stale:?}"); + // A JVM that finishes a session withdraws its deliveries. + retire(&control, "move", Phase::Withdrawing); + assert!(retired(&control.activate(identity).await)); + + control.cancel(destination).await.unwrap(); + assert!(!control.state().unwrap().players.contains_key(&uuid)); + control.claim(request("again", &uuid)).await.unwrap(); + fixture.close().await; +} diff --git a/crates/chunk-environment/src/core/sync/platform/claims.rs b/crates/chunk-environment/src/core/sync/platform/claims.rs index f0702cee..609b4ccf 100644 --- a/crates/chunk-environment/src/core/sync/platform/claims.rs +++ b/crates/chunk-environment/src/core/sync/platform/claims.rs @@ -6,7 +6,9 @@ use super::{ super::{SyncService, app}, decode, errors, }; -use chunk_control::{ALREADY_OWNED, Error as Failure, Generation, ROSTER_WAITING, ROUTE_AGAIN, StoredClaim}; +use chunk_control::{ + ALREADY_OWNED, DESTINATION_RETIRED, Error as Failure, Generation, ROSTER_WAITING, ROUTE_AGAIN, StoredClaim, +}; use chunk_proto::{ control::v1 as control, sync::v1::{ @@ -108,8 +110,9 @@ async fn activate(service: &SyncService, gateway: &str, operation: &str) -> Resu let identity = identity.ok_or_else(|| errors::invalid("no claim under this operation ID"))?; let control = service.control.clone(); match service.operations.admit(async move { control.activate(identity).await }).await { - Ok(_) => Ok(ActivateResult { waiting: false }), - Err(Failure::Unresolved(ROSTER_WAITING)) => Ok(ActivateResult { waiting: true }), + Ok(_) => Ok(ActivateResult::default()), + Err(Failure::Unresolved(ROSTER_WAITING)) => Ok(ActivateResult { waiting: true, retired: false }), + Err(Failure::Invalid(DESTINATION_RETIRED)) => Ok(ActivateResult { waiting: false, retired: true }), Err(failure) => Err(errors::operation(&failure)), } } diff --git a/crates/chunk-proxy/src/server/managed.rs b/crates/chunk-proxy/src/server/managed.rs index 97586a3e..aae17ff8 100644 --- a/crates/chunk-proxy/src/server/managed.rs +++ b/crates/chunk-proxy/src/server/managed.rs @@ -174,7 +174,13 @@ pub(super) async fn serve( let mut lifecycle = Lifecycle::new(platform.clone()); let mut commands = commands::Commands::new(&platform).await?; let mut packs = packs::Packs::default(); + let mut rerouted = false; loop { + if guard.platform.target.deployment != platform.target.deployment { + platform = guard.platform.clone(); + lifecycle.rebind(platform.clone()); + commands = commands::Commands::new(&platform).await?; + } commands.bind(&guard.claim, &assignment.identity)?; // Packs load within what the claim's lifetime leaves, keeping time to connect before it expires. let remaining = assignment.expires.saturating_duration_since(Instant::now()).saturating_sub(CONNECT_RESERVE); @@ -186,9 +192,29 @@ pub(super) async fn serve( deadline.min(PACK_TIMEOUT).min(remaining), ) .await?; - let mut internal = timeout(deadline.min(WAIT_TIMEOUT), open(&assignment, &guard, &authenticated, &settings)) + let opened = timeout(deadline.min(WAIT_TIMEOUT), open(&assignment, &guard, &authenticated, &settings)) .await .map_err(io::Error::other)??; + let Some(mut internal) = opened else { + // The client is configuring and its player holds no other claim, so they are routed as a login, once. + let routed = if rerouted { + Err(io::Error::other("destination session retired after routing again")) + } else { + timeout(deadline.min(WAIT_TIMEOUT), reroute(guard, current)).await.map_err(io::Error::other).flatten() + }; + match routed { + Ok(next) => (guard, assignment) = next, + Err(error) => { + let _ = + configuration::disconnect(&mut authenticated.transport, 0x02, "Server temporarily unavailable") + .await; + return Err(error); + } + } + lifecycle.cutover(&guard.claim); + rerouted = true; + continue; + }; timeout( deadline.min(WAIT_TIMEOUT), Box::pin(configuration::relay(&mut authenticated.transport, &mut internal, &mut settings)), @@ -212,6 +238,7 @@ pub(super) async fn serve( } lifecycle.arrived(&guard.claim)?; commands.arrived(); + rerouted = false; tracing::info!(operation = %guard.claim.operation_id, player = %guard.claim.player.uuid, "player arrived in managed session"); let next = relay::until( &mut authenticated.transport, @@ -239,12 +266,27 @@ pub(super) async fn serve( guard.armed = false; drop(internal); (guard, assignment) = next; - if guard.platform.target.deployment != platform.target.deployment { - platform = guard.platform.clone(); - lifecycle.rebind(platform.clone()); - commands = commands::Commands::new(&platform).await?; + } +} + +/// Routes the player of `retired`, a claim whose session retired before admitting them, as a login again. Core admits +/// that login once `retired` is withdrawn, as the player then holds no claim: a move fenced its source before +/// activating. A move is first recorded as failed. +async fn reroute(mut retired: ClaimGuard, current: &Retarget) -> io::Result<(ClaimGuard, Assignment)> { + let operation = retired.claim.operation_id.clone(); + tracing::warn!(%operation, player = %retired.claim.player.uuid, "session retired before activation; routing its player again"); + if retired.claim.source.is_some() { + retired.failure = Some("destination session retired".into()); + if let Err(error) = retired.cancel().await { + tracing::debug!(%error, %operation, "retired move's failure unrecorded"); } + retired.failure = None; } + withdraw(&retired).await?; + retired.armed = false; + let Claim { connection_id, player, .. } = retired.claim.clone(); + let login = Claim { operation_id: uuid::Uuid::new_v4().to_string(), connection_id, player, ..Claim::default() }; + claim_destination(&login, current).await } /// Routes and claims `login` through the current release. A login whose player may return to the session they left on @@ -327,26 +369,30 @@ fn login_claim(profile: &chunk_protocol::versions::v26_2::LoginSuccess, connecti } } +/// Activates `guard`'s claim and connects to its JVM, or returns `None` when the claim's session retired first. async fn open( assignment: &Assignment, guard: &ClaimGuard, authenticated: &Authenticated, settings: &chunk_protocol::versions::v26_2::ConfigurationClientInformation, -) -> io::Result> { +) -> io::Result>> { if assignment.protocol != authenticated.protocol_version { return Err(invalid_data("destination protocol differs from client")); } - activate(guard).await?; - gameplay::login(authenticated, settings, &assignment.endpoint, assignment.setup.clone()).await + if !activate(guard).await? { + return Ok(None); + } + gameplay::login(authenticated, settings, &assignment.endpoint, assignment.setup.clone()).await.map(Some) } -/// Records admission intent. A roster member waits for the rest of its group; the caller bounds the wait. -async fn activate(guard: &ClaimGuard) -> io::Result<()> { +/// Records admission intent, returning whether core admitted the claim: `false` means its session retired first, which +/// ends the claim. A roster member waits for the rest of its group; the caller bounds the wait. +async fn activate(guard: &ClaimGuard) -> io::Result { loop { let operation = &guard.claim.operation_id; let (result, _): (ActivateResult, _) = guard.platform.call("activate", operation, &(), RPC_TIMEOUT).await?; - if !result.waiting { - return Ok(()); + if result.retired || !result.waiting { + return Ok(!result.retired); } sleep(Duration::from_millis(250)).await; } diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture.rs b/crates/chunk-proxy/src/server/managed/commands/fixture.rs index a55328f8..a09ec93a 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture.rs @@ -44,6 +44,8 @@ pub(super) struct Service { /// Activations answered as waiting for the rest of a roster before one succeeds. pub roster_waits: Arc, pub activations: Arc, + /// The next activation finds its claim's session retired. + pub retiring: Arc, /// Counts claim-state changes; open streams send a new snapshot for each. pub published: Arc>, /// Ends open streams at their next publication and refuses new ones. @@ -150,6 +152,7 @@ impl Fixture { reopen: Arc::default(), roster_waits: Arc::default(), activations: Arc::default(), + retiring: Arc::default(), published: Arc::new(watch::Sender::new(1)), watch_down: Arc::default(), refused_watches: Arc::default(), diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs index f126216a..12db5331 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs @@ -83,7 +83,8 @@ impl Service { self.activations.fetch_add(1, Ordering::SeqCst); let waits = self.roster_waits.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |waits| waits.checked_sub(1)); - ActivateResult { waiting: waits.is_ok() }.encode_to_vec() + let retired = self.retiring.swap(false, Ordering::SeqCst); + ActivateResult { waiting: waits.is_ok(), retired }.encode_to_vec() } "chunk:withdraw" => { self.logins.lock().unwrap().cancels.push(operation); diff --git a/crates/chunk-proxy/src/server/managed/commands/tests.rs b/crates/chunk-proxy/src/server/managed/commands/tests.rs index fd4ab39d..9ff03a7c 100644 --- a/crates/chunk-proxy/src/server/managed/commands/tests.rs +++ b/crates/chunk-proxy/src/server/managed/commands/tests.rs @@ -435,7 +435,7 @@ async fn roster_activation_waits_for_the_group_and_other_failures_still_end_the_ failure: None, }; fixture.service.roster_waits.store(3, Ordering::SeqCst); - crate::server::managed::activate(&guard).await.unwrap(); + assert!(crate::server::managed::activate(&guard).await.unwrap()); assert_eq!(fixture.service.activations.load(Ordering::SeqCst), 4); let mut target = fixture.commands.tasks.platform.target.clone(); target.gateway.credential = "wrong".into(); diff --git a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs index e3231a77..d77da7e6 100644 --- a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs +++ b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs @@ -6,12 +6,15 @@ use crate::{ PlatformTarget, server::{ Retarget, - managed::{Assignment, ClaimGuard, claim_destination, moves::next_move}, + managed::{Assignment, ClaimGuard, activate, claim_destination, moves::next_move, reroute}, }, }; use chunk_proto::sync::v1::{GatewayMove, SessionDemand}; use serde_json::{Value, json}; -use std::{io, sync::RwLock}; +use std::{ + io, + sync::{RwLock, atomic::Ordering}, +}; fn manifest(app: &str) -> Value { let hook = |event: &str, export: &str| json!({"domain": "", "event": event, "export": export}); @@ -211,3 +214,29 @@ async fn a_reserved_move_completes_in_its_deployment_though_the_current_one_drop assert!(hooks(&fixture).iter().all(|(deployment, _)| deployment == "b")); fixture.close().await; } + +#[tokio::test] +async fn a_move_whose_session_retired_before_activation_is_recorded_failed_and_its_player_routed_as_a_login() { + let fixture = Fixture::new().await; + deployments(&fixture, "b"); + let current = retarget(&fixture, "b"); + let (_source, running) = moving(&fixture, ¤t); + let (destination, _) = tokio::time::timeout(Duration::from_secs(5), running).await.unwrap().unwrap().unwrap(); + fixture.service.retiring.store(true, Ordering::SeqCst); + assert!(!activate(&destination).await.unwrap()); + + let (mut login, assignment) = reroute(destination, ¤t).await.unwrap(); + login.armed = false; + assert_eq!(assignment.deployment, "b"); + assert!(login.claim.source.is_none()); + assert_eq!(login.claim.connection_id, fixture.claim.connection_id); + let failure = fixture.service.movement.lock().unwrap().failure.clone(); + assert_eq!(failure, Some(("move".into(), "destination session retired".into()))); + // The retired claim was withdrawn before its player's login was claimed. + { + let logins = fixture.service.logins.lock().unwrap(); + assert_eq!(logins.cancels, ["move"]); + assert_eq!(logins.claims.last().map(|(operation, _)| operation), Some(&login.claim.operation_id)); + } + fixture.close().await; +} diff --git a/crates/chunk-proxy/src/server/managed/testing.rs b/crates/chunk-proxy/src/server/managed/testing.rs index 1cb58136..a30f6f90 100644 --- a/crates/chunk-proxy/src/server/managed/testing.rs +++ b/crates/chunk-proxy/src/server/managed/testing.rs @@ -45,7 +45,9 @@ async fn arrived( }; let guard = ClaimGuard { platform: platform.clone(), claim: login, armed: true, failure: None }; let assignment = claim(&guard).await?.ok_or_else(|| io::Error::other("core asked to route the login again"))?; - activate(&guard).await?; + if !activate(&guard).await? { + return Err(io::Error::other("the claim's session retired before activation")); + } arrive(&guard, assignment.identity.clone()).await?; Ok((guard, assignment)) } diff --git a/proto/chunk/sync/v1/gateway.proto b/proto/chunk/sync/v1/gateway.proto index d3dce642..8568872f 100644 --- a/proto/chunk/sync/v1/gateway.proto +++ b/proto/chunk/sync/v1/gateway.proto @@ -219,6 +219,9 @@ enum ClaimRefusal { message ActivateResult { // Other members of the player's roster have yet to activate, so none was admitted. Call again. bool waiting = 1; + // The claim's session retired before admitting the player, so the claim only ends. Withdraw it; its player, who + // holds no other claim once a move fenced its source, can then log in again. + bool retired = 2; } // Arguments of `chunk:abandon_move`. From 9b33749a57988200437d8fe3859d158cb88fdebd Mon Sep 17 00:00:00 2001 From: oglassdev <23040613+oglassdev@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:53:51 -0400 Subject: [PATCH 2/5] fix(control,proxy): recover from destinations that retire during claim or configuration Record on a claim that its session retired, so activation reports retirement only for claims it ended and not for cancelled, abandoned or superseded ones. A login whose session retires while claiming is refused with a new claim refusal and claimed again under a new operation ID. A move whose claim ends before configuration is abandoned, and one whose destination retires during configuration is routed again as a login. --- crates/chunk-bench/src/control.rs | 1 + crates/chunk-control/src/delivery.rs | 42 +++++-- crates/chunk-control/src/placement.rs | 1 + crates/chunk-control/src/reconcile.rs | 10 ++ crates/chunk-control/src/recovery.rs | 1 + crates/chunk-control/src/state/entities.rs | 2 + crates/chunk-control/src/state/store.rs | 2 +- crates/chunk-control/src/tests/gateway.rs | 1 + crates/chunk-control/src/tests/moves.rs | 25 +++- crates/chunk-control/src/tests/recovery.rs | 1 + .../src/core/sync/platform/claims.rs | 13 ++- crates/chunk-proxy/src/server/managed.rs | 108 ++++++++++++++---- .../src/server/managed/commands/fixture.rs | 2 +- .../managed/commands/fixture/core_service.rs | 2 + .../src/server/managed/commands/tests.rs | 1 + .../managed/commands/tests/placement.rs | 16 +++ .../chunk-proxy/src/server/managed/moves.rs | 7 +- .../chunk-proxy/src/server/managed/testing.rs | 8 +- proto/chunk/sync/v1/gateway.proto | 3 + 19 files changed, 205 insertions(+), 41 deletions(-) diff --git a/crates/chunk-bench/src/control.rs b/crates/chunk-bench/src/control.rs index f8caa5d0..1b8c0237 100644 --- a/crates/chunk-bench/src/control.rs +++ b/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)) diff --git a/crates/chunk-control/src/delivery.rs b/crates/chunk-control/src/delivery.rs index d9fa879c..e8955867 100644 --- a/crates/chunk-control/src/delivery.rs +++ b/crates/chunk-control/src/delivery.rs @@ -14,20 +14,15 @@ 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. Rejects with - /// [`crate::DESTINATION_RETIRED`] the current claim whose session retired before admitting it, whether or not that - /// already withdrew the claim. Reports `Busy` until recovery has fenced surviving JVMs. + /// [`crate::DESTINATION_RETIRED`] a claim that ended unactivated because its session retired, unless its move was + /// abandoned or a newer claim owns its player. Reports `Busy` until recovery has fenced surviving JVMs. pub async fn activate(&self, identity: ClaimIdentity) -> Result { self.admit()?; let operation = self.operation(&identity.operation_id)?; let _guard = operation.lock().await; let admitted = self.update(|state| { - let claim = state.claims.get(&identity.operation_id).ok_or(Error::Invalid("unknown claim"))?; // Checked before the move's source: once that is fenced and this claim released, nothing owns the player. - if claim.identity(&identity.operation_id) == identity - && !claim.activated - && matches!(claim.phase, Phase::Reserved | Phase::Withdrawing | Phase::Released) - && state.sessions.get(&claim.session).is_some_and(|session| session.retired) - { + if lost_to_retirement(state, &identity) { return Err(Error::Invalid(crate::DESTINATION_RETIRED)); } crate::moves::authorize_destination(state, &identity)?; @@ -228,6 +223,9 @@ pub(crate) fn apply(state: &mut State, host: &str, status: &JvmDeliveryStatus) - if phase <= claim.phase { return Ok(()); } + if matches!(phase, Phase::Withdrawing | Phase::Released) { + note_retirement(state, operation); + } if phase == Phase::Released { release(state, operation) } else { @@ -235,6 +233,34 @@ pub(crate) fn apply(state: &mut State, host: &str, status: &JvmDeliveryStatus) - } } +/// Whether the claim `identity` names ended, or is ending, unactivated because its session retired: not because it was +/// cancelled, its move abandoned or a newer claim took its player. +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 ended = claim.retired + || (claim.phase == Phase::Reserved && state.sessions.get(&claim.session).is_some_and(|s| s.retired)); + let abandoned = state.moves.get(operation).is_some_and(|intent| intent.canceled || intent.failure.is_some()); + 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 && !claim.activated && ended && !abandoned && !superseded +} + +/// Records that `operation` is ending because its session retired, if it is still reserved there. +pub(crate) fn note_retirement(state: &mut State, operation: &str) { + let retired = state.claims.get(operation).is_some_and(|claim| { + claim.phase == Phase::Reserved + && !claim.activated + && state.sessions.get(&claim.session).is_some_and(|s| s.retired) + }); + if retired && let Some(claim) = state.claims.get_mut(operation) { + claim.retired = true; + } +} + 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)?; diff --git a/crates/chunk-control/src/placement.rs b/crates/chunk-control/src/placement.rs index 29ad2f53..0a56ae89 100644 --- a/crates/chunk-control/src/placement.rs +++ b/crates/chunk-control/src/placement.rs @@ -313,6 +313,7 @@ pub(crate) fn insert_claim( assigned_at_ms: None, released_at_ms: None, roster, + retired: false, }, ); Ok(()) diff --git a/crates/chunk-control/src/reconcile.rs b/crates/chunk-control/src/reconcile.rs index bea0c1e3..516c50ee 100644 --- a/crates/chunk-control/src/reconcile.rs +++ b/crates/chunk-control/src/reconcile.rs @@ -30,12 +30,22 @@ impl Control { if !cancel { continue; } + let retired = claim.phase == Phase::Reserved && state.sessions[&claim.session].retired; let control = self.clone(); let permits = permits.clone(); tasks.spawn(async move { let Ok(_permit) = permits.acquire_owned().await else { return; }; + if retired + && let Err(error) = control.update(|state| { + crate::delivery::note_retirement(state, &request.operation_id); + Ok(()) + }) + { + tracing::debug!(%error, "control reconciliation could not record a retirement"); + return; + } if let Err(error) = control.cancel(request).await { tracing::debug!(%error, "control reconciliation retains unresolved claim"); } diff --git a/crates/chunk-control/src/recovery.rs b/crates/chunk-control/src/recovery.rs index ee5fdf20..265c29d3 100644 --- a/crates/chunk-control/src/recovery.rs +++ b/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, + retired: false, } } diff --git a/crates/chunk-control/src/state/entities.rs b/crates/chunk-control/src/state/entities.rs index be49ac20..0c91883f 100644 --- a/crates/chunk-control/src/state/entities.rs +++ b/crates/chunk-control/src/state/entities.rs @@ -350,6 +350,8 @@ pub(crate) struct Claim { pub released_at_ms: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub roster: Option, + /// Whether the claim ended unactivated because its session retired, rather than by cancellation. + pub retired: bool, } impl Claim { diff --git a/crates/chunk-control/src/state/store.rs b/crates/chunk-control/src/state/store.rs index fbb30b97..b6d42973 100644 --- a/crates/chunk-control/src/state/store.rs +++ b/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, "retired": boolean, }, MOVES: { "request": string, "canceled": boolean, "sequence": integer, diff --git a/crates/chunk-control/src/tests/gateway.rs b/crates/chunk-control/src/tests/gateway.rs index fcb4791c..27077b0c 100644 --- a/crates/chunk-control/src/tests/gateway.rs +++ b/crates/chunk-control/src/tests/gateway.rs @@ -22,6 +22,7 @@ fn add(control: &Control, gateway: &str, operations: impl IntoIterator return Err(errors::error(Code::OperationMismatch, "the operation ID names a login")), (None, None) => return Err(errors::invalid("no move is queued under this operation ID")), }; - let control = service.control.clone(); + let login = request.source.is_none(); + let (control, ended) = (service.control.clone(), request.clone()); let outcome = match service.operations.admit(async move { control.claim_approved(request, &approved).await }).await { Ok(assignment) => Outcome::Assignment(assigned(service, assignment)?), Err(Failure::Unresolved(ROUTE_AGAIN)) => Outcome::Refusal(ClaimRefusal::RouteAgain.into()), + // A login's claim ends here so the player can claim again; a move is abandoned by its gateway. + Err(Failure::Invalid(DESTINATION_RETIRED)) if login => { + let control = service.control.clone(); + service + .operations + .admit(async move { control.cancel(ended).await }) + .await + .map_err(|f| errors::operation(&f))?; + Outcome::Refusal(ClaimRefusal::DestinationRetired.into()) + } Err(Failure::Invalid(ALREADY_OWNED)) => Outcome::Refusal(ClaimRefusal::AlreadyConnected.into()), Err(failure) => return Err(errors::operation(&failure)), }; diff --git a/crates/chunk-proxy/src/server/managed.rs b/crates/chunk-proxy/src/server/managed.rs index aae17ff8..252e9673 100644 --- a/crates/chunk-proxy/src/server/managed.rs +++ b/crates/chunk-proxy/src/server/managed.rs @@ -20,7 +20,7 @@ use chunk_proto::sync::v1::{ use tokio::{ io::{AsyncRead, AsyncWrite}, sync::{Notify, oneshot}, - time::{Instant, sleep, timeout}, + time::{Instant, sleep, timeout, timeout_at}, }; use super::{ @@ -129,7 +129,8 @@ impl Assignment { /// Claims `guard`'s login or queued move. `None` means the deployment that admitted it is no longer current, so it must /// be admitted again; it reserved nothing. -async fn claim(guard: &ClaimGuard) -> io::Result> { +/// Asks core to place `guard`'s claim. +async fn claim(guard: &ClaimGuard) -> io::Result { let arguments = ClaimArguments { login: guard.claim.source.is_none().then(|| login(&guard.claim)), deployment: guard.claim.deployment.clone(), @@ -137,9 +138,12 @@ async fn claim(guard: &ClaimGuard) -> io::Result> { let operation = &guard.claim.operation_id; let (result, _): (ClaimResult, _) = guard.platform.call("claim", operation, &arguments, WAIT_TIMEOUT).await?; match result.outcome { - Some(Outcome::Assignment(assigned)) => Assignment::new(assigned, &guard.claim).map(Some), + Some(Outcome::Assignment(assigned)) => { + Assignment::new(assigned, &guard.claim).map(|assignment| Claimed::Placed(assignment.into())) + } Some(Outcome::Refusal(refusal)) => match ClaimRefusal::try_from(refusal) { - Ok(ClaimRefusal::RouteAgain) => Ok(None), + Ok(ClaimRefusal::RouteAgain) => Ok(Claimed::RouteAgain), + Ok(ClaimRefusal::DestinationRetired) => Ok(Claimed::Retired), Ok(ClaimRefusal::AlreadyConnected) => { Err(io::Error::new(io::ErrorKind::PermissionDenied, "You are already connected.")) } @@ -149,6 +153,15 @@ async fn claim(guard: &ClaimGuard) -> io::Result> { } } +/// What core made of a claim. +enum Claimed { + Placed(Box), + /// The deployment that admitted the claim no longer accepts it; it is admitted again under the same operation ID. + RouteAgain, + /// The claim's session retired and core ended the claim; its player is routed again under a new operation ID. + Retired, +} + fn login(claim: &Claim) -> GatewayLogin { GatewayLogin { connection_id: claim.connection_id.clone(), @@ -192,17 +205,13 @@ pub(super) async fn serve( deadline.min(PACK_TIMEOUT).min(remaining), ) .await?; - let opened = timeout(deadline.min(WAIT_TIMEOUT), open(&assignment, &guard, &authenticated, &settings)) + // Activation and, when the session retired first, routing the player again share this one bound. + let budget = Instant::now() + deadline.min(WAIT_TIMEOUT); + let opened = timeout_at(budget, open(&assignment, &guard, &authenticated, &settings)) .await .map_err(io::Error::other)??; let Some(mut internal) = opened else { - // The client is configuring and its player holds no other claim, so they are routed as a login, once. - let routed = if rerouted { - Err(io::Error::other("destination session retired after routing again")) - } else { - timeout(deadline.min(WAIT_TIMEOUT), reroute(guard, current)).await.map_err(io::Error::other).flatten() - }; - match routed { + match reroute_once(rerouted, guard, current, budget).await { Ok(next) => (guard, assignment) = next, Err(error) => { let _ = @@ -240,15 +249,15 @@ pub(super) async fn serve( commands.arrived(); rerouted = false; tracing::info!(operation = %guard.claim.operation_id, player = %guard.claim.player.uuid, "player arrived in managed session"); - let next = relay::until( + let next = live_move( &mut authenticated.transport, &mut internal, &mut settings, - Box::pin(next_move(&guard, &identity, authenticated.protocol_version, current)), - true, Some(&mut commands), + (&guard, &identity, authenticated.protocol_version), + (current, &platform), ) - .await??; + .await?; commands.configuration(); timeout( Duration::from_secs(10), @@ -256,7 +265,12 @@ pub(super) async fn serve( ) .await .map_err(io::Error::other)??; - check_move(&platform, &next.0.claim).await?; + // A destination that retired while the client configured is routed again once the source is withdrawn. + if let Err(error) = check_move(&platform, &next.0.claim).await + && !retired(&next.0, deadline).await + { + return Err(error); + } // The client's acknowledgment fences all remaining source PLAY input. if let Err(error) = withdraw(&guard).await { let _ = configuration::disconnect(&mut authenticated.transport, 0x02, "Session move unavailable").await; @@ -269,6 +283,53 @@ pub(super) async fn serve( } } +/// Whether core reports that `destination`'s session retired. +async fn retired(destination: &ClaimGuard, deadline: Duration) -> bool { + matches!(timeout(deadline.min(WAIT_TIMEOUT), activate(destination)).await, Ok(Ok(false))) +} + +/// Routes the player of `retired` as a login again within `budget`, unless that already happened since they arrived. +/// The client is configuring and their player holds no other claim. +async fn reroute_once( + rerouted: bool, + retired: ClaimGuard, + current: &Retarget, + budget: Instant, +) -> io::Result<(ClaimGuard, Assignment)> { + if rerouted { + return Err(io::Error::other("destination session retired after routing again")); + } + timeout_at(budget, reroute(retired, current)).await.map_err(io::Error::other).flatten() +} + +/// Waits for a move from `source` whose claim core still holds, relaying the delivery meanwhile. A move that ended after +/// it was prepared, as when its destination retired, is abandoned and the player stays in play. +async fn live_move( + public: &mut Transport, + internal: &mut Transport, + settings: &mut chunk_protocol::versions::v26_2::ConfigurationClientInformation, + mut commands: Option<&mut commands::Commands>, + (source, identity, protocol): (&ClaimGuard, &ClaimIdentity, i32), + (current, platform): (&Retarget, &Platform), +) -> io::Result<(ClaimGuard, Assignment)> +where + S: AsyncRead + AsyncWrite + Unpin, + I: AsyncRead + AsyncWrite + Unpin, +{ + loop { + let ready = Box::pin(next_move(source, identity, protocol, current)); + let mut next = relay::until(public, internal, settings, ready, true, commands.as_deref_mut()).await??; + if check_move(platform, &next.0.claim).await.is_ok() { + return Ok(next); + } + tracing::warn!(operation = %next.0.claim.operation_id, "move ended before configuration; source remains active"); + next.0.failure = Some("move ended before configuration".into()); + if next.0.cancel().await.is_ok() { + next.0.armed = false; + } + } +} + /// Routes the player of `retired`, a claim whose session retired before admitting them, as a login again. Core admits /// that login once `retired` is withdrawn, as the player then holds no claim: a move fenced its source before /// activating. A move is first recorded as failed. @@ -317,12 +378,13 @@ async fn claim_destination(login: &Claim, current: &Retarget) -> io::Result<(Cla ClaimGuard { platform: platform.clone(), claim, armed: true, failure: None } } }; - let Some(assignment) = self::claim(&guard).await? else { - guard.armed = false; - sleep(Duration::from_millis(100)).await; - continue; - }; - return Ok((guard, assignment)); + match self::claim(&guard).await? { + Claimed::Placed(assignment) => return Ok((guard, *assignment)), + Claimed::RouteAgain => {} + Claimed::Retired => login.operation_id = uuid::Uuid::new_v4().to_string(), + } + guard.armed = false; + sleep(Duration::from_millis(100)).await; } } diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture.rs b/crates/chunk-proxy/src/server/managed/commands/fixture.rs index a09ec93a..87518a29 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture.rs @@ -44,7 +44,7 @@ pub(super) struct Service { /// Activations answered as waiting for the rest of a roster before one succeeds. pub roster_waits: Arc, pub activations: Arc, - /// The next activation finds its claim's session retired. + /// The next activation or login claim finds its session retired. pub retiring: Arc, /// Counts claim-state changes; open streams send a new snapshot for each. pub published: Arc>, diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs index 12db5331..617ebdd3 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs @@ -61,6 +61,8 @@ impl Service { destination: Some(destination), ..reservation() }) + } else if self.retiring.swap(false, Ordering::SeqCst) { + claim_result::Outcome::Refusal(ClaimRefusal::DestinationRetired.into()) } else if logins.retired.as_ref() == Some(&login.deployment) || placement.refuses(&login.deployment) { claim_result::Outcome::Refusal(ClaimRefusal::RouteAgain.into()) } else { diff --git a/crates/chunk-proxy/src/server/managed/commands/tests.rs b/crates/chunk-proxy/src/server/managed/commands/tests.rs index 9ff03a7c..a9390f36 100644 --- a/crates/chunk-proxy/src/server/managed/commands/tests.rs +++ b/crates/chunk-proxy/src/server/managed/commands/tests.rs @@ -14,6 +14,7 @@ async fn failed_move_stops_preparation_and_reports_the_reason_despite_a_lost_rep use chunk_proto::sync::v1::{Error, GatewayMove, SessionDemand, error::Code}; for (code, message) in [ (Code::Invalid, "destination configuration rejected"), + (Code::Invalid, "destination session retired"), (Code::Stopped, "runtime stopped"), (Code::Unavailable, "app did not become ready within 35 seconds"), ] { diff --git a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs index d77da7e6..fd1f0b94 100644 --- a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs +++ b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs @@ -215,6 +215,22 @@ async fn a_reserved_move_completes_in_its_deployment_though_the_current_one_drop fixture.close().await; } +#[tokio::test] +async fn a_login_whose_session_retired_while_claiming_is_claimed_again_under_a_new_operation() { + let fixture = Fixture::new().await; + deployments(&fixture, "b"); + fixture.service.retiring.store(true, Ordering::SeqCst); + let login = Claim { operation_id: "login".into(), ..fixture.claim.clone() }; + let (mut guard, _) = claim_destination(&login, &retarget(&fixture, "b")).await.unwrap(); + guard.armed = false; + let claims = fixture.service.logins.lock().unwrap().claims.clone(); + assert_eq!(claims.len(), 2); + assert_eq!(claims[0].0, "login"); + assert_ne!(claims[1].0, "login"); + assert_eq!(guard.claim.operation_id, claims[1].0); + fixture.close().await; +} + #[tokio::test] async fn a_move_whose_session_retired_before_activation_is_recorded_failed_and_its_player_routed_as_a_login() { let fixture = Fixture::new().await; diff --git a/crates/chunk-proxy/src/server/managed/moves.rs b/crates/chunk-proxy/src/server/managed/moves.rs index f1bc9c5c..e34570fc 100644 --- a/crates/chunk-proxy/src/server/managed/moves.rs +++ b/crates/chunk-proxy/src/server/managed/moves.rs @@ -1,4 +1,6 @@ -use super::{Assignment, Claim, ClaimGuard, ClaimIdentity, Platform, Retarget, WAIT_TIMEOUT, claim, invalid_data}; +use super::{ + Assignment, Claim, ClaimGuard, ClaimIdentity, Claimed, Platform, Retarget, WAIT_TIMEOUT, claim, invalid_data, +}; use crate::server::platform::{RPC_TIMEOUT, failure}; use chunk_proto::sync::v1::{Position, ReservationResult, SessionDemand, error::Code}; use std::{io, time::Duration}; @@ -118,7 +120,8 @@ async fn attempt(source: &ClaimGuard, destination: &mut ClaimGuard, protocol: i3 let platform = destination.platform.clone(); platform.approve_move(&source.platform, &source.claim, &destination.claim).await?; check_move(&platform, &destination.claim).await?; - let Some(assignment) = claim(destination).await? else { return Ok(None) }; + let Claimed::Placed(assignment) = claim(destination).await? else { return Ok(None) }; + let assignment = *assignment; if !assignment.deployment.is_empty() && assignment.deployment != destination.claim.deployment { bind(destination, &assignment.deployment, assignment.destination.clone()); destination.platform.approve_move(&source.platform, &source.claim, &destination.claim).await?; diff --git a/crates/chunk-proxy/src/server/managed/testing.rs b/crates/chunk-proxy/src/server/managed/testing.rs index a30f6f90..e6e57706 100644 --- a/crates/chunk-proxy/src/server/managed/testing.rs +++ b/crates/chunk-proxy/src/server/managed/testing.rs @@ -4,7 +4,7 @@ use std::io; use chunk_proto::sync::v1::{PlayerIdentity, SessionDemand}; -use super::{Assignment, Claim, ClaimGuard, activate, arrive, claim}; +use super::{Assignment, Claim, ClaimGuard, Claimed, activate, arrive, claim}; use crate::{PlatformTarget, Retarget, server::platform::Platform}; /// Claims a login of player `uuid` into `demand` through `target` as a managed connection does, activates it, waits @@ -44,10 +44,12 @@ async fn arrived( ..Claim::default() }; let guard = ClaimGuard { platform: platform.clone(), claim: login, armed: true, failure: None }; - let assignment = claim(&guard).await?.ok_or_else(|| io::Error::other("core asked to route the login again"))?; + let Claimed::Placed(assignment) = claim(&guard).await? else { + return Err(io::Error::other("core asked to route the login again")); + }; if !activate(&guard).await? { return Err(io::Error::other("the claim's session retired before activation")); } arrive(&guard, assignment.identity.clone()).await?; - Ok((guard, assignment)) + Ok((guard, *assignment)) } diff --git a/proto/chunk/sync/v1/gateway.proto b/proto/chunk/sync/v1/gateway.proto index 8568872f..b7f11aa7 100644 --- a/proto/chunk/sync/v1/gateway.proto +++ b/proto/chunk/sync/v1/gateway.proto @@ -213,6 +213,9 @@ enum ClaimRefusal { CLAIM_REFUSAL_ROUTE_AGAIN = 1; // The player already holds a claim, such as through another connection. CLAIM_REFUSAL_ALREADY_CONNECTED = 2; + // The login's session retired before control prepared it, and control ended the claim. Route the login again under a + // new operation ID. + CLAIM_REFUSAL_DESTINATION_RETIRED = 3; } // The result of `chunk:activate`. From 97334dcf3fb362edd467a0ca10eaea59796184b5 Mon Sep 17 00:00:00 2001 From: oglassdev <23040613+oglassdev@users.noreply.github.com> Date: Mon, 12 Oct 2026 20:03:22 -0400 Subject: [PATCH 3/5] fix(control): tell retirement from cancellation by what control canceled, not by when a claim ended --- crates/chunk-control/src/delivery.rs | 36 +++++++++------------- crates/chunk-control/src/placement.rs | 33 ++++++++++++++------ crates/chunk-control/src/reconcile.rs | 10 ------ crates/chunk-control/src/recovery.rs | 2 +- crates/chunk-control/src/state/entities.rs | 4 +-- crates/chunk-control/src/state/store.rs | 2 +- crates/chunk-control/src/tests/gateway.rs | 2 +- crates/chunk-control/src/tests/moves.rs | 21 +++++-------- crates/chunk-control/src/tests/recovery.rs | 2 +- 9 files changed, 51 insertions(+), 61 deletions(-) diff --git a/crates/chunk-control/src/delivery.rs b/crates/chunk-control/src/delivery.rs index e8955867..427ca951 100644 --- a/crates/chunk-control/src/delivery.rs +++ b/crates/chunk-control/src/delivery.rs @@ -14,8 +14,8 @@ 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. Rejects with - /// [`crate::DESTINATION_RETIRED`] a claim that ended unactivated because its session retired, unless its move was - /// abandoned or a newer claim owns its player. Reports `Busy` until recovery has fenced surviving JVMs. + /// [`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 { self.admit()?; let operation = self.operation(&identity.operation_id)?; @@ -86,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 { @@ -223,9 +228,6 @@ pub(crate) fn apply(state: &mut State, host: &str, status: &JvmDeliveryStatus) - if phase <= claim.phase { return Ok(()); } - if matches!(phase, Phase::Withdrawing | Phase::Released) { - note_retirement(state, operation); - } if phase == Phase::Released { release(state, operation) } else { @@ -233,32 +235,22 @@ pub(crate) fn apply(state: &mut State, host: &str, status: &JvmDeliveryStatus) - } } -/// Whether the claim `identity` names ended, or is ending, unactivated because its session retired: not because it was -/// cancelled, its move abandoned or a newer claim took its player. +/// 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 ended = claim.retired - || (claim.phase == Phase::Reserved && state.sessions.get(&claim.session).is_some_and(|s| s.retired)); - let abandoned = state.moves.get(operation).is_some_and(|intent| intent.canceled || intent.failure.is_some()); 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 && !claim.activated && ended && !abandoned && !superseded + claim.identity(operation) == *identity && retired(state, claim) && !superseded } -/// Records that `operation` is ending because its session retired, if it is still reserved there. -pub(crate) fn note_retirement(state: &mut State, operation: &str) { - let retired = state.claims.get(operation).is_some_and(|claim| { - claim.phase == Phase::Reserved - && !claim.activated - && state.sessions.get(&claim.session).is_some_and(|s| s.retired) - }); - if retired && let Some(claim) = state.claims.get_mut(operation) { - claim.retired = true; - } +/// 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. +pub(crate) fn retired(state: &State, claim: &Claim) -> bool { + !claim.activated && !claim.canceled && state.sessions.get(&claim.session).is_some_and(|session| session.retired) } pub(crate) fn release(state: &mut State, operation: &str) -> Result<()> { diff --git a/crates/chunk-control/src/placement.rs b/crates/chunk-control/src/placement.rs index 0a56ae89..f87b204d 100644 --- a/crates/chunk-control/src/placement.rs +++ b/crates/chunk-control/src/placement.rs @@ -52,13 +52,13 @@ 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")); } let session = state.sessions.get(&claim.session).ok_or(Error::Invalid("missing session"))?; - if claim.phase == Phase::Reserved && session.retired { - return Err(Error::Invalid(crate::DESTINATION_RETIRED)); - } let deployment = &state.host_release(&session.host)?.deployment; let runtime = self.runtime(&session.host).await?; let config = self.jvm_configuration(deployment, &runtime)?; @@ -72,9 +72,23 @@ 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. + Err(error) => { + 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 { @@ -91,13 +105,14 @@ 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")); } - if state.sessions.get(&claim.session).is_some_and(|session| session.retired) { - return Err(Error::Invalid(crate::DESTINATION_RETIRED)); - } let mut persisted = assignment.clone(); persisted.configuration = None; persisted.destination = None; @@ -313,7 +328,7 @@ pub(crate) fn insert_claim( assigned_at_ms: None, released_at_ms: None, roster, - retired: false, + canceled: false, }, ); Ok(()) diff --git a/crates/chunk-control/src/reconcile.rs b/crates/chunk-control/src/reconcile.rs index 516c50ee..bea0c1e3 100644 --- a/crates/chunk-control/src/reconcile.rs +++ b/crates/chunk-control/src/reconcile.rs @@ -30,22 +30,12 @@ impl Control { if !cancel { continue; } - let retired = claim.phase == Phase::Reserved && state.sessions[&claim.session].retired; let control = self.clone(); let permits = permits.clone(); tasks.spawn(async move { let Ok(_permit) = permits.acquire_owned().await else { return; }; - if retired - && let Err(error) = control.update(|state| { - crate::delivery::note_retirement(state, &request.operation_id); - Ok(()) - }) - { - tracing::debug!(%error, "control reconciliation could not record a retirement"); - return; - } if let Err(error) = control.cancel(request).await { tracing::debug!(%error, "control reconciliation retains unresolved claim"); } diff --git a/crates/chunk-control/src/recovery.rs b/crates/chunk-control/src/recovery.rs index 265c29d3..6271b328 100644 --- a/crates/chunk-control/src/recovery.rs +++ b/crates/chunk-control/src/recovery.rs @@ -319,6 +319,6 @@ fn tombstone(status: &JvmDeliveryStatus) -> Claim { assigned_at_ms: None, released_at_ms: Some(now), roster: None, - retired: false, + canceled: false, } } diff --git a/crates/chunk-control/src/state/entities.rs b/crates/chunk-control/src/state/entities.rs index 0c91883f..953ec58c 100644 --- a/crates/chunk-control/src/state/entities.rs +++ b/crates/chunk-control/src/state/entities.rs @@ -350,8 +350,8 @@ pub(crate) struct Claim { pub released_at_ms: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub roster: Option, - /// Whether the claim ended unactivated because its session retired, rather than by cancellation. - pub retired: bool, + /// Whether control began withdrawing the claim while its session still ran. + pub canceled: bool, } impl Claim { diff --git a/crates/chunk-control/src/state/store.rs b/crates/chunk-control/src/state/store.rs index b6d42973..9ab3ea80 100644 --- a/crates/chunk-control/src/state/store.rs +++ b/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, "retired": boolean, + "roster?": string, "canceled": boolean, }, MOVES: { "request": string, "canceled": boolean, "sequence": integer, diff --git a/crates/chunk-control/src/tests/gateway.rs b/crates/chunk-control/src/tests/gateway.rs index 27077b0c..cd313074 100644 --- a/crates/chunk-control/src/tests/gateway.rs +++ b/crates/chunk-control/src/tests/gateway.rs @@ -22,7 +22,7 @@ fn add(control: &Control, gateway: &str, operations: impl IntoIterator Date: Mon, 12 Oct 2026 20:15:41 -0400 Subject: [PATCH 4/5] test(control,proxy): cover moves and logins whose session retires in one report, in a roster, mid-claim and mid-configuration --- crates/chunk-control/src/tests/jvm.rs | 10 ++ crates/chunk-control/src/tests/moves.rs | 39 +++++ crates/chunk-control/src/tests/roster.rs | 27 ++++ .../src/server/managed/commands/fixture.rs | 6 + .../managed/commands/fixture/core_service.rs | 19 ++- .../managed/commands/tests/placement.rs | 148 ++++++++++++++++++ 6 files changed, 242 insertions(+), 7 deletions(-) diff --git a/crates/chunk-control/src/tests/jvm.rs b/crates/chunk-control/src/tests/jvm.rs index 63345f78..f47bec26 100644 --- a/crates/chunk-control/src/tests/jvm.rs +++ b/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(); diff --git a/crates/chunk-control/src/tests/moves.rs b/crates/chunk-control/src/tests/moves.rs index 4b5263cf..79c48166 100644 --- a/crates/chunk-control/src/tests/moves.rs +++ b/crates/chunk-control/src/tests/moves.rs @@ -132,3 +132,42 @@ async fn reconciliation_ending_a_claim_in_a_retired_session_is_reported_until_a_ 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).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; +} diff --git a/crates/chunk-control/src/tests/roster.rs b/crates/chunk-control/src/tests/roster.rs index 20ca3411..e1e8df09 100644 --- a/crates/chunk-control/src/tests/roster.rs +++ b/crates/chunk-control/src/tests/roster.rs @@ -243,3 +243,30 @@ async fn moves_to_undeclared_destinations_are_unknown() { assert!(matches!(moved, Err(Error::Refused(chunk_contract::MoveRefusal::UnknownDestination))), "{moved:?}"); fixture.close().await; } + +#[tokio::test] +async fn members_of_a_roster_whose_session_retires_are_told_so_rather_than_that_their_move_was_abandoned() { + let fixture = fixture(); + let control = fixture.control().await; + let members = [arrive(&fixture, &control, "a").await, arrive(&fixture, &control, "b").await]; + let destinations = control.move_roster(&roster("party", 1, &members)).unwrap(); + let mut activations = Vec::new(); + for (destination, (player, source)) in destinations.iter().zip(&members) { + let prepared = control.claim(destination.clone()).await.unwrap(); + control.cancel(request(&source.operation_id, player)).await.unwrap(); + activations.push(prepared.claim.unwrap()); + } + let (first, second) = (&destinations[0].operation_id, &destinations[1].operation_id); + let session = control.state().unwrap().claims[first].session.clone(); + assert_eq!(control.state().unwrap().claims[second].session, session); + + // The first member's claim releasing fails the second member's move. + fixture.runtime.finish(&session, &[first]); + eventually(|| control.state().unwrap().moves[second].failure.is_some()).await; + assert_eq!(control.state().unwrap().moves[second].failure.as_ref().unwrap().reason, "roster member left"); + for activation in activations { + let result = control.activate(activation).await; + assert!(matches!(result, Err(Error::Invalid(crate::DESTINATION_RETIRED))), "{result:?}"); + } + fixture.close().await; +} diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture.rs b/crates/chunk-proxy/src/server/managed/commands/fixture.rs index 87518a29..a1d075aa 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture.rs @@ -73,6 +73,8 @@ pub(super) struct Logins { pub cancels: Vec, /// Logins routed with this deployment are rejected for routing again. pub retired: Option, + /// Activating any of these operations finds its session retired, every time. + pub retired_operations: BTreeSet, /// The next routing waits for this, then fails. pub unroutable: Option>, } @@ -92,6 +94,8 @@ pub(super) struct Placement { pub returns: Option<(String, sync::SessionDemand)>, /// The deployment and destination a move's operation already holds a reservation in. pub reserved: Option<(String, sync::SessionDemand)>, + /// The player listener every reservation names, where the fixture's own is not one. + pub endpoint: Option, } #[derive(Default)] @@ -102,6 +106,8 @@ pub(super) struct Movement { pub reports: usize, pub lose_report: bool, pub stall_retries: bool, + /// The operation ID of a move core ended without recording a failure; the gateway's abandonment records it. + pub vanished: Option, /// The operation ID and reason of the move abandoned. pub failure: Option<(String, String)>, } diff --git a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs index 617ebdd3..155dc188 100644 --- a/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs +++ b/crates/chunk-proxy/src/server/managed/commands/fixture/core_service.rs @@ -66,7 +66,7 @@ impl Service { } else if logins.retired.as_ref() == Some(&login.deployment) || placement.refuses(&login.deployment) { claim_result::Outcome::Refusal(ClaimRefusal::RouteAgain.into()) } else { - claim_result::Outcome::Assignment(placed(&login.deployment)) + claim_result::Outcome::Assignment(placement.placed(&login.deployment)) }; ClaimResult { outcome: Some(outcome) }.encode_to_vec() } @@ -85,7 +85,8 @@ impl Service { self.activations.fetch_add(1, Ordering::SeqCst); let waits = self.roster_waits.fetch_update(Ordering::SeqCst, Ordering::SeqCst, |waits| waits.checked_sub(1)); - let retired = self.retiring.swap(false, Ordering::SeqCst); + let retired = self.retiring.swap(false, Ordering::SeqCst) + || self.logins.lock().unwrap().retired_operations.contains(&operation); ActivateResult { waiting: waits.is_ok(), retired }.encode_to_vec() } "chunk:withdraw" => { @@ -100,6 +101,7 @@ impl Service { } let expected = movement.pending.as_ref().map(|pending| &pending.operation_id); let expected = expected.or_else(|| movement.failure.as_ref().map(|(operation, _)| operation)); + let expected = expected.or(movement.vanished.as_ref()); assert_eq!(expected, Some(&operation)); let reason = sync::AbandonMoveArguments::decode(call.arguments.as_slice()).unwrap().reason; movement.pending = None; @@ -186,7 +188,7 @@ impl Service { } else if placement.refuses(approved) { claim_result::Outcome::Refusal(ClaimRefusal::RouteAgain.into()) } else { - claim_result::Outcome::Assignment(placed(approved)) + claim_result::Outcome::Assignment(placement.placed(approved)) }; Ok(ClaimResult { outcome: Some(outcome) }.encode_to_vec()) } @@ -199,10 +201,13 @@ impl super::Placement { } } -/// A reservation in `deployment`, or in the fixture's own when it is empty. -fn placed(deployment: &str) -> ClaimAssignment { - let deployment = if deployment.is_empty() { "deployment" } else { deployment }; - ClaimAssignment { deployment: deployment.into(), ..reservation() } +impl super::Placement { + /// A reservation in `deployment`, or in the fixture's own when it is empty. + fn placed(&self, deployment: &str) -> ClaimAssignment { + let deployment = if deployment.is_empty() { "deployment" } else { deployment }; + let endpoint = self.endpoint.clone().unwrap_or_else(|| reservation().endpoint); + ClaimAssignment { deployment: deployment.into(), endpoint, ..reservation() } + } } /// The claim the gateway's topic holds. diff --git a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs index fd1f0b94..38cfad2b 100644 --- a/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs +++ b/crates/chunk-proxy/src/server/managed/commands/tests/placement.rs @@ -2,6 +2,7 @@ //! activation. use super::*; +use crate::server::{authentication::Authenticated, transport::Transport}; use crate::{ PlatformTarget, server::{ @@ -10,11 +11,20 @@ use crate::{ }, }; use chunk_proto::sync::v1::{GatewayMove, SessionDemand}; +use chunk_protocol::{ + RemainingBytes, + versions::v26_2::{ + AcknowledgeConfiguration, ConfigurationAcknowledged, ConfigurationKeepAlive, ConfigurationKeepAliveResponse, + FinishConfiguration, Handshake, LoginAcknowledged, LoginPluginRequest, LoginPluginResponse, LoginStart, + LoginSuccess, StartConfiguration, + }, +}; use serde_json::{Value, json}; use std::{ io, sync::{RwLock, atomic::Ordering}, }; +use tokio::net::TcpStream; fn manifest(app: &str) -> Value { let hook = |event: &str, export: &str| json!({"domain": "", "event": event, "export": export}); @@ -256,3 +266,141 @@ async fn a_move_whose_session_retired_before_activation_is_recorded_failed_and_i } fixture.close().await; } + +/// Plays a JVM's player listener for one connection: its delivery login, then, when `finish`, the end of configuration. +/// Returns the connection, which stays open until dropped. +async fn jvm_accepts(listener: &tokio::net::TcpListener, profile: &LoginSuccess, finish: bool) -> Transport { + let (socket, _) = listener.accept().await.unwrap(); + let mut jvm = Transport::new(socket); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + let challenge = LoginPluginRequest { + message_id: VarInt(1), + channel: McString::new("chunk:delivery").unwrap(), + data: RemainingBytes::new(vec![]).unwrap(), + }; + jvm.write_packet(&challenge).await.unwrap(); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + jvm.write_packet(profile).await.unwrap(); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + if finish { + jvm.write_packet(&FinishConfiguration).await.unwrap(); + decode_packet::(&jvm.read_frame(4096).await.unwrap()).unwrap(); + } + jvm +} + +/// Reads frames from `client`, answering configuration keepalives, until one is a `P`. +async fn client_reads(client: &mut Transport) { + loop { + let frame = client.read_frame(16384).await.unwrap(); + match VarInt::decode(&mut frame.as_ref()).unwrap().0 { + id if id == P::ID => return, + id if id == ConfigurationKeepAlive::ID => { + let keep_alive_id = decode_packet::(&frame).unwrap().keep_alive_id; + client.write_packet(&ConfigurationKeepAliveResponse { keep_alive_id }).await.unwrap(); + } + _ => {} + } + } +} + +#[tokio::test] +async fn a_move_whose_destination_retires_during_configuration_keeps_the_player_connected_on_a_new_login() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fixture = Fixture::new().await; + { + let mut placement = fixture.service.placement.lock().unwrap(); + placement.manifests.insert("deployment".into(), manifest("hub")); + placement.endpoint = Some(listener.local_addr().unwrap().to_string()); + } + fixture.service.logins.lock().unwrap().retired_operations.insert("move".into()); + let current = retarget(&fixture, "deployment"); + let (client, public) = tokio::io::duplex(65_536); + let profile = LoginSuccess { + session_id: chunk_protocol::Uuid([2; 16]), + uuid: chunk_protocol::Uuid([1; 16]), + username: McString::new("Alex").unwrap(), + properties: chunk_protocol::BoundedArray::new(vec![]).unwrap(), + }; + // Reservations name protocol 0. + let authenticated = + Authenticated { protocol_version: 0, profile: profile.clone(), transport: Transport::new(public) }; + let serving = { + let current = current.clone(); + tokio::spawn(async move { + Box::pin(crate::server::managed::serve(authenticated, ¤t, Duration::from_secs(30))).await + }) + }; + let mut client = Transport::new(client); + let settings = ConfigurationClientInformation { + locale: McString::new("en_US").unwrap(), + view_distance: 8, + chat_flags: VarInt(0), + chat_colors: true, + skin_parts: 127, + main_hand: VarInt(1), + enable_text_filtering: false, + enable_server_listing: true, + particle_status: ConfigurationClientInformationParticleStatus::All, + }; + client.write_packet(&settings).await.unwrap(); + + // Core's topic shows the player's login once it is claimed. + let login = loop { + let claimed = fixture.service.logins.lock().unwrap().claims.first().cloned(); + if let Some(login) = claimed { + break login; + } + tokio::time::sleep(Duration::from_millis(5)).await; + }; + fixture.service.claim.lock().unwrap().operation.clone_from(&login.0); + fixture.service.publish(); + let ((), source) = tokio::join!( + async { + client_reads::(&mut client).await; + client.write_packet(&AcknowledgeConfiguration).await.unwrap(); + }, + jvm_accepts(&listener, &profile, true) + ); + + // Core queues a move, which the gateway prepares, and the client starts configuring for it. + fixture.service.movement.lock().unwrap().pending = + Some(GatewayMove { operation_id: "move".into(), destination: Some(demand("hub")) }); + fixture.service.publish(); + client_reads::(&mut client).await; + + // The destination's session retires while the client configures: core ends the move. + { + let mut movement = fixture.service.movement.lock().unwrap(); + movement.pending = None; + movement.vanished = Some("move".into()); + } + fixture.sync().await; + client.write_packet(&ConfigurationAcknowledged).await.unwrap(); + + // The player's new login reaches a JVM on the connection that was moving. + let replacement = + tokio::time::timeout(Duration::from_secs(10), jvm_accepts(&listener, &profile, false)).await.unwrap(); + let (claims, cancels) = { + let logins = fixture.service.logins.lock().unwrap(); + (logins.claims.clone(), logins.cancels.clone()) + }; + assert_eq!(claims.len(), 2, "one login for the player, and one fresh login after the retirement"); + assert_ne!(claims[1].0, login.0); + assert_eq!(claims[1].1.connection_id, login.1.connection_id); + assert_eq!(claims[1].1.player, login.1.player); + assert_eq!(cancels, [login.0, "move".to_owned()], "the source, then the retired destination"); + { + let movement = fixture.service.movement.lock().unwrap(); + assert_eq!(movement.failure, Some(("move".into(), "destination session retired".into()))); + assert_eq!(movement.reports, 1); + } + // The login, the move twice (before cutover and after it), and the new login. + assert_eq!(fixture.service.activations.load(Ordering::SeqCst), 4); + assert!(!serving.is_finished()); + serving.abort(); + drop((source, replacement, client)); + fixture.close().await; +} From 4797a37c3c4e8df7b3af9c30c7edbd7190e9cf36 Mon Sep 17 00:00:00 2001 From: oglassdev <23040613+oglassdev@users.noreply.github.com> Date: Mon, 12 Oct 2026 20:37:43 -0400 Subject: [PATCH 5/5] fix(control): keep a claim retired after its session's row is gone, and read it after the report commits --- crates/chunk-control/src/delivery.rs | 4 ++-- crates/chunk-control/src/placement.rs | 4 +++- crates/chunk-control/src/tests/moves.rs | 4 ++++ 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/crates/chunk-control/src/delivery.rs b/crates/chunk-control/src/delivery.rs index 427ca951..d7c62b97 100644 --- a/crates/chunk-control/src/delivery.rs +++ b/crates/chunk-control/src/delivery.rs @@ -248,9 +248,9 @@ fn lost_to_retirement(state: &State, identity: &ClaimIdentity) -> bool { } /// 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. +/// 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_some_and(|session| session.retired) + !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<()> { diff --git a/crates/chunk-control/src/placement.rs b/crates/chunk-control/src/placement.rs index f87b204d..9c095e52 100644 --- a/crates/chunk-control/src/placement.rs +++ b/crates/chunk-control/src/placement.rs @@ -78,8 +78,10 @@ impl Control { }; let preparation = match prepared.await { Ok(preparation) => preparation, - // The session's retirement is what stopped or closed the delivery. + // 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 diff --git a/crates/chunk-control/src/tests/moves.rs b/crates/chunk-control/src/tests/moves.rs index 79c48166..56046de9 100644 --- a/crates/chunk-control/src/tests/moves.rs +++ b/crates/chunk-control/src/tests/moves.rs @@ -148,6 +148,10 @@ async fn a_report_closing_a_delivery_and_ending_its_session_together_leaves_the_ 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; }