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 434f34c5..d7c62b97 100644 --- a/crates/chunk-control/src/delivery.rs +++ b/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 { 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)?; 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..9c095e52 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 } @@ -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(()) diff --git a/crates/chunk-control/src/recovery.rs b/crates/chunk-control/src/recovery.rs index ee5fdf20..6271b328 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, + canceled: false, } } diff --git a/crates/chunk-control/src/state/entities.rs b/crates/chunk-control/src/state/entities.rs index be49ac20..953ec58c 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 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 fbb30b97..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, + "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 fcb4791c..cd313074 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 (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 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; +} diff --git a/crates/chunk-control/src/tests/recovery.rs b/crates/chunk-control/src/tests/recovery.rs index 12cec1c7..8e486805 100644 --- a/crates/chunk-control/src/tests/recovery.rs +++ b/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); } 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-environment/src/core/sync/platform/claims.rs b/crates/chunk-environment/src/core/sync/platform/claims.rs index f0702cee..baac291d 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::{ @@ -92,11 +94,22 @@ async fn claim( (None, Some(_)) => 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)), }; @@ -108,8 +121,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..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(), @@ -174,7 +187,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 +205,25 @@ 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)) + // 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 { + match reroute_once(rerouted, guard, current, budget).await { + 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,16 +247,17 @@ 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( + 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), @@ -229,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; @@ -239,14 +280,76 @@ 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?; + } +} + +/// 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. +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 /// an earlier release is admitted there, as that session's destination, without consulting the current release's hooks, /// and claimed for exactly that session; otherwise it is routed and placed with the current release. A failed routing @@ -275,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; } } @@ -327,26 +431,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..a1d075aa 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 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>, /// Ends open streams at their next publication and refuses new ones. @@ -71,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>, } @@ -90,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)] @@ -100,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)>, } @@ -150,6 +158,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..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 @@ -61,10 +61,12 @@ 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 { - claim_result::Outcome::Assignment(placed(&login.deployment)) + claim_result::Outcome::Assignment(placement.placed(&login.deployment)) }; ClaimResult { outcome: Some(outcome) }.encode_to_vec() } @@ -83,7 +85,9 @@ 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) + || self.logins.lock().unwrap().retired_operations.contains(&operation); + ActivateResult { waiting: waits.is_ok(), retired }.encode_to_vec() } "chunk:withdraw" => { self.logins.lock().unwrap().cancels.push(operation); @@ -97,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; @@ -183,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()) } @@ -196,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.rs b/crates/chunk-proxy/src/server/managed/commands/tests.rs index fd4ab39d..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"), ] { @@ -435,7 +436,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..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,16 +2,29 @@ //! activation. use super::*; +use crate::server::{authentication::Authenticated, transport::Transport}; 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 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}; +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}); @@ -211,3 +224,183 @@ 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_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; + 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; +} + +/// 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; +} 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 1cb58136..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,8 +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"))?; - activate(&guard).await?; + 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 d3dce642..b7f11aa7 100644 --- a/proto/chunk/sync/v1/gateway.proto +++ b/proto/chunk/sync/v1/gateway.proto @@ -213,12 +213,18 @@ 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`. 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`.