1060 lines
38 KiB
Rust
1060 lines
38 KiB
Rust
//! Deterministic channel-join pending-state reducer.
|
|
//!
|
|
//! This module implements SDD-121 Phase A for the `chanora_state` crate:
|
|
//! Rust-owned authoritative current-channel membership, non-authoritative
|
|
//! pending join state, and reconnect/snapshot reconciliation.
|
|
|
|
use std::time::Instant;
|
|
|
|
/// Channel identifier at the reducer seam.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
|
pub struct ChannelId(pub u64);
|
|
|
|
/// Per-connection epoch used to reject stale outcomes and deltas.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
|
|
pub struct ConnectionEpoch(pub u64);
|
|
|
|
/// Monotonic join/leave generation within reducer state.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
|
|
pub struct JoinGeneration(pub u64);
|
|
|
|
/// Protocol request identifier unique within one connection epoch.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
|
pub struct JoinRequestId(pub u64);
|
|
|
|
/// Server-confirmed voice-channel membership.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct AuthoritativeMembership {
|
|
/// Last server-authoritative current channel, or none if not in voice.
|
|
pub current_channel: Option<ChannelId>,
|
|
/// Epoch that produced the confirmed membership value.
|
|
pub confirmed_epoch: ConnectionEpoch,
|
|
}
|
|
|
|
/// Active non-authoritative join intent awaiting server confirmation.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub struct JoinPending {
|
|
/// Requested target channel.
|
|
pub target_channel: ChannelId,
|
|
/// Previous confirmed channel to preserve on failure/timeout.
|
|
pub previous_confirmed_channel: Option<ChannelId>,
|
|
/// Protocol request id after command-send acceptance.
|
|
pub request_id: Option<JoinRequestId>,
|
|
/// Reducer generation for this intent.
|
|
pub generation: JoinGeneration,
|
|
/// Deterministic start time supplied by the caller.
|
|
pub started_at: Instant,
|
|
/// Connection epoch in which the intent was created.
|
|
pub connection_epoch: ConnectionEpoch,
|
|
}
|
|
|
|
/// Reducer-owned channel-join state.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ChannelJoinState {
|
|
/// Server-authoritative membership.
|
|
pub authoritative: AuthoritativeMembership,
|
|
/// Active pending join, if any.
|
|
pub pending: Option<JoinPending>,
|
|
/// Current snapshot/reconnect synchronization state.
|
|
pub sync_state: ChannelJoinSyncState,
|
|
/// Next generation to allocate.
|
|
pub next_generation: JoinGeneration,
|
|
/// Last sanitized join error for projection consumers.
|
|
pub last_join_error: Option<JoinErrorCode>,
|
|
}
|
|
|
|
impl ChannelJoinState {
|
|
/// Create channel-join reducer state for a connection epoch.
|
|
pub fn new(epoch: ConnectionEpoch) -> Self {
|
|
Self {
|
|
authoritative: AuthoritativeMembership {
|
|
current_channel: None,
|
|
confirmed_epoch: epoch,
|
|
},
|
|
pending: None,
|
|
sync_state: ChannelJoinSyncState::Ready,
|
|
next_generation: JoinGeneration(1),
|
|
last_join_error: None,
|
|
}
|
|
}
|
|
|
|
fn allocate_generation(&mut self) -> JoinGeneration {
|
|
let generation = self.next_generation;
|
|
self.next_generation = JoinGeneration(self.next_generation.0.saturating_add(1));
|
|
generation
|
|
}
|
|
|
|
fn bump_generation(&mut self) {
|
|
self.next_generation = JoinGeneration(self.next_generation.0.saturating_add(1));
|
|
}
|
|
|
|
fn current_epoch(&self) -> ConnectionEpoch {
|
|
match self.sync_state {
|
|
ChannelJoinSyncState::Ready => self.authoritative.confirmed_epoch,
|
|
ChannelJoinSyncState::Synchronizing { epoch, .. } => epoch,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// State-sync readiness for channel actions.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum ChannelJoinSyncState {
|
|
/// Reducer is ready to accept channel actions.
|
|
Ready,
|
|
/// Reducer is waiting for a snapshot boundary.
|
|
Synchronizing {
|
|
/// Reason for synchronization.
|
|
reason: SyncReason,
|
|
/// Epoch being synchronized.
|
|
epoch: ConnectionEpoch,
|
|
},
|
|
}
|
|
|
|
/// Reason channel-join state is synchronizing.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum SyncReason {
|
|
/// Initial snapshot is in progress.
|
|
InitialSnapshot,
|
|
/// Reconnect snapshot is in progress.
|
|
Reconnect,
|
|
}
|
|
|
|
/// Correlation key for protocol outcomes and timeout callbacks.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
|
pub struct JoinOutcomeKey {
|
|
/// Connection epoch of the pending join.
|
|
pub connection_epoch: ConnectionEpoch,
|
|
/// Generation of the pending join.
|
|
pub generation: JoinGeneration,
|
|
/// Protocol request id, once accepted for send.
|
|
pub request_id: Option<JoinRequestId>,
|
|
}
|
|
|
|
/// Reducer projection consumed by core/bridge mapping.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ChannelJoinProjection {
|
|
/// Last confirmed current channel.
|
|
pub current_channel: Option<ChannelId>,
|
|
/// Non-authoritative pending target channel.
|
|
pub pending_target: Option<ChannelId>,
|
|
/// Start time of the active pending join.
|
|
pub pending_since: Option<Instant>,
|
|
/// Generation of the active pending join.
|
|
pub pending_generation: Option<JoinGeneration>,
|
|
/// Whether a new join can be initiated.
|
|
pub can_join: bool,
|
|
/// Whether leave is possible for current/pending state.
|
|
pub can_leave: bool,
|
|
/// Current sync state.
|
|
pub sync_state: ChannelJoinSyncState,
|
|
/// Last sanitized join error.
|
|
pub last_join_error: Option<JoinErrorCode>,
|
|
}
|
|
|
|
/// Source of authoritative membership input.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum AuthoritativeSource {
|
|
/// Live protocol delta.
|
|
LiveDelta,
|
|
/// Snapshot input.
|
|
Snapshot,
|
|
}
|
|
|
|
/// Events accepted by the channel-join reducer.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum ChannelJoinEvent {
|
|
/// User requested a channel join.
|
|
UserJoinRequested {
|
|
/// Requested target channel.
|
|
target_channel: ChannelId,
|
|
/// Caller-supplied deterministic request time.
|
|
now: Instant,
|
|
},
|
|
/// Protocol command send accepted and produced a request id.
|
|
JoinCommandAccepted {
|
|
/// Pending generation accepted by the command path.
|
|
generation: JoinGeneration,
|
|
/// Protocol request id for future outcomes.
|
|
request_id: JoinRequestId,
|
|
},
|
|
/// Protocol command failed before a request id existed.
|
|
JoinCommandRejectedBeforeSend {
|
|
/// Pending generation rejected by the command path.
|
|
generation: JoinGeneration,
|
|
/// Sanitized failure kind.
|
|
error: JoinFailureKind,
|
|
},
|
|
/// Protocol command reported success; not authoritative membership.
|
|
ProtocolJoinSucceeded {
|
|
/// Outcome correlation key.
|
|
key: JoinOutcomeKey,
|
|
},
|
|
/// Protocol command reported failure.
|
|
ProtocolJoinFailed {
|
|
/// Outcome correlation key.
|
|
key: JoinOutcomeKey,
|
|
/// Sanitized failure kind.
|
|
failure: JoinFailureKind,
|
|
},
|
|
/// Join command timed out.
|
|
JoinTimeout {
|
|
/// Outcome correlation key.
|
|
key: JoinOutcomeKey,
|
|
/// Caller-supplied deterministic timeout time.
|
|
now: Instant,
|
|
},
|
|
/// Server-authoritative self membership changed.
|
|
AuthoritativeSelfMove {
|
|
/// Server-authoritative channel value.
|
|
channel: Option<ChannelId>,
|
|
/// Epoch of the live delta or snapshot source.
|
|
epoch: ConnectionEpoch,
|
|
/// Authoritative input source.
|
|
source: AuthoritativeSource,
|
|
},
|
|
/// User requested leave.
|
|
UserLeaveRequested {
|
|
/// Caller-supplied deterministic leave time.
|
|
now: Instant,
|
|
},
|
|
/// Reconnect started.
|
|
ReconnectStarted {
|
|
/// New connection epoch.
|
|
new_epoch: ConnectionEpoch,
|
|
/// Caller-supplied deterministic reconnect time.
|
|
now: Instant,
|
|
},
|
|
/// Fresh snapshot is ready.
|
|
SnapshotReady {
|
|
/// Snapshot current channel value.
|
|
current_channel: Option<ChannelId>,
|
|
/// Snapshot epoch.
|
|
epoch: ConnectionEpoch,
|
|
/// Caller-supplied deterministic snapshot time.
|
|
now: Instant,
|
|
},
|
|
}
|
|
|
|
/// Side-effect actions returned by the reducer.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum ChannelJoinAction {
|
|
/// Send a protocol join command.
|
|
SendJoinCommand {
|
|
/// Requested target channel.
|
|
target_channel: ChannelId,
|
|
/// Generation associated with the command.
|
|
generation: JoinGeneration,
|
|
/// Connection epoch associated with the command.
|
|
epoch: ConnectionEpoch,
|
|
},
|
|
/// Start a deterministic join timeout.
|
|
StartJoinTimeout {
|
|
/// Outcome key the timeout will report.
|
|
key: JoinOutcomeKey,
|
|
/// Timeout start time.
|
|
started_at: Instant,
|
|
},
|
|
/// Cancel a deterministic join timeout.
|
|
CancelJoinTimeout {
|
|
/// Outcome key to cancel.
|
|
key: JoinOutcomeKey,
|
|
},
|
|
/// Send a protocol leave command.
|
|
SendLeaveCommand {
|
|
/// Current authoritative channel at leave request time.
|
|
current_channel: Option<ChannelId>,
|
|
/// Generation associated with the leave command.
|
|
generation: JoinGeneration,
|
|
/// Connection epoch associated with the leave command.
|
|
epoch: ConnectionEpoch,
|
|
},
|
|
/// Publish the latest projection.
|
|
PublishProjection(ChannelJoinProjection),
|
|
/// Emit a sanitized diagnostic.
|
|
EmitDiagnostic {
|
|
/// Sanitized diagnostic key.
|
|
key: JoinDiagnosticKey,
|
|
/// Optional stable error code.
|
|
code: Option<JoinErrorCode>,
|
|
},
|
|
}
|
|
|
|
/// Sanitized diagnostic event key.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum JoinDiagnosticKey {
|
|
/// Join was requested.
|
|
JoinRequested,
|
|
/// Join command was accepted for send.
|
|
JoinCommandSent,
|
|
/// Duplicate target was coalesced.
|
|
JoinDuplicateCoalesced,
|
|
/// Join was confirmed authoritatively.
|
|
JoinConfirmed,
|
|
/// Join failed.
|
|
JoinFailed,
|
|
/// Join timed out.
|
|
JoinTimeout,
|
|
/// Stale outcome ignored.
|
|
JoinStaleOutcomeIgnored,
|
|
/// Pending join superseded by leave.
|
|
JoinPendingSupersededByLeave,
|
|
/// Reconnect synchronization started.
|
|
JoinReconnectSynchronizing,
|
|
/// Snapshot reconciled membership.
|
|
JoinSnapshotReconciled,
|
|
}
|
|
|
|
/// Status for a reducer transition.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum JoinReduceStatus {
|
|
/// Event was accepted.
|
|
Accepted,
|
|
/// Duplicate same-target join was coalesced.
|
|
CoalescedSameTarget,
|
|
/// Intent was rejected.
|
|
Rejected(JoinIntentRejected),
|
|
/// Stale outcome was ignored.
|
|
StaleOutcomeIgnored,
|
|
/// Join was authoritatively confirmed.
|
|
Confirmed,
|
|
/// Join failed.
|
|
Failed(JoinFailureKind),
|
|
/// Join timed out.
|
|
TimedOut,
|
|
/// Join was superseded by leave.
|
|
SupersededByLeave,
|
|
/// State was reconciled by snapshot.
|
|
ReconciledBySnapshot,
|
|
}
|
|
|
|
/// User intent rejection reason.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum JoinIntentRejected {
|
|
/// A different target is already pending.
|
|
JoinAlreadyPendingDifferentTarget,
|
|
/// Joins cannot start while synchronizing.
|
|
CannotJoinWhileSynchronizing,
|
|
}
|
|
|
|
/// Coarse failure kind without raw protocol payloads.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum JoinFailureKind {
|
|
/// Server denied the join.
|
|
Denied,
|
|
/// Network failure.
|
|
Network,
|
|
/// Protocol failure.
|
|
Protocol,
|
|
/// Timeout.
|
|
Timeout,
|
|
/// Unknown sanitized failure.
|
|
Unknown,
|
|
}
|
|
|
|
/// Stable sanitized error code consumed by bridge/UI mapping.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum JoinErrorCode {
|
|
/// Duplicate same target was coalesced.
|
|
DuplicateSameTargetCoalesced,
|
|
/// Different target requested while pending.
|
|
JoinAlreadyPendingDifferentTarget,
|
|
/// Join was denied.
|
|
JoinDenied,
|
|
/// Protocol failure.
|
|
JoinProtocolFailure,
|
|
/// Network failure.
|
|
JoinNetworkFailure,
|
|
/// Join timed out.
|
|
JoinTimeout,
|
|
/// Join superseded by leave.
|
|
JoinSupersededByLeave,
|
|
/// Stale outcome ignored.
|
|
JoinStaleOutcomeIgnored,
|
|
/// Snapshot/live delta reconciled to a different channel.
|
|
JoinReconciledDifferentChannel,
|
|
/// Command rejected before send.
|
|
JoinCommandRejectedBeforeSend,
|
|
/// Cannot start while synchronizing.
|
|
JoinCannotStartWhileSynchronizing,
|
|
}
|
|
|
|
/// Complete reducer result.
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub struct ChannelJoinReduction {
|
|
/// Transition status.
|
|
pub status: JoinReduceStatus,
|
|
/// Projection after reduction.
|
|
pub projection: ChannelJoinProjection,
|
|
/// Side effects for the owning runtime.
|
|
pub actions: Vec<ChannelJoinAction>,
|
|
}
|
|
|
|
/// Reduce one channel-join event into state and side-effect actions.
|
|
pub fn reduce(state: &mut ChannelJoinState, event: ChannelJoinEvent) -> ChannelJoinReduction {
|
|
let mut actions = Vec::new();
|
|
let status = match event {
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel,
|
|
now,
|
|
} => {
|
|
if matches!(state.sync_state, ChannelJoinSyncState::Synchronizing { .. }) {
|
|
state.last_join_error = Some(JoinErrorCode::JoinCannotStartWhileSynchronizing);
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinFailed,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::Rejected(JoinIntentRejected::CannotJoinWhileSynchronizing)
|
|
} else if let Some(pending) = state.pending {
|
|
if pending.target_channel == target_channel {
|
|
state.last_join_error = Some(JoinErrorCode::DuplicateSameTargetCoalesced);
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinDuplicateCoalesced,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::CoalescedSameTarget
|
|
} else {
|
|
state.last_join_error = Some(JoinErrorCode::JoinAlreadyPendingDifferentTarget);
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinFailed,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::Rejected(
|
|
JoinIntentRejected::JoinAlreadyPendingDifferentTarget,
|
|
)
|
|
}
|
|
} else {
|
|
let generation = state.allocate_generation();
|
|
let epoch = state.current_epoch();
|
|
state.pending = Some(JoinPending {
|
|
target_channel,
|
|
previous_confirmed_channel: state.authoritative.current_channel,
|
|
request_id: None,
|
|
generation,
|
|
started_at: now,
|
|
connection_epoch: epoch,
|
|
});
|
|
state.last_join_error = None;
|
|
let key = JoinOutcomeKey {
|
|
connection_epoch: epoch,
|
|
generation,
|
|
request_id: None,
|
|
};
|
|
actions.push(ChannelJoinAction::SendJoinCommand {
|
|
target_channel,
|
|
generation,
|
|
epoch,
|
|
});
|
|
actions.push(ChannelJoinAction::StartJoinTimeout {
|
|
key,
|
|
started_at: now,
|
|
});
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinRequested,
|
|
code: None,
|
|
});
|
|
JoinReduceStatus::Accepted
|
|
}
|
|
}
|
|
ChannelJoinEvent::JoinCommandAccepted {
|
|
generation,
|
|
request_id,
|
|
} => {
|
|
if let Some(pending) = state.pending.as_mut() {
|
|
if pending.generation == generation {
|
|
let old_key = pending_key(pending);
|
|
pending.request_id = Some(request_id);
|
|
let new_key = pending_key(pending);
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key: old_key });
|
|
actions.push(ChannelJoinAction::StartJoinTimeout {
|
|
key: new_key,
|
|
started_at: pending.started_at,
|
|
});
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinCommandSent,
|
|
code: None,
|
|
});
|
|
JoinReduceStatus::Accepted
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
}
|
|
ChannelJoinEvent::JoinCommandRejectedBeforeSend { generation, error } => {
|
|
if let Some(pending) = state.pending {
|
|
if pending.generation == generation {
|
|
let key = pending_key(&pending);
|
|
state.authoritative.current_channel = pending.previous_confirmed_channel;
|
|
state.pending = None;
|
|
state.last_join_error = Some(JoinErrorCode::JoinCommandRejectedBeforeSend);
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key });
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinFailed,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::Failed(error)
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
}
|
|
ChannelJoinEvent::ProtocolJoinSucceeded { key } => {
|
|
if key_matches(state.pending, key) {
|
|
JoinReduceStatus::Accepted
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
}
|
|
ChannelJoinEvent::ProtocolJoinFailed { key, failure } => {
|
|
if key_matches(state.pending, key) {
|
|
let pending = state.pending.expect("key match requires pending");
|
|
state.authoritative.current_channel = pending.previous_confirmed_channel;
|
|
state.pending = None;
|
|
state.last_join_error = Some(failure_error_code(failure));
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key });
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinFailed,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::Failed(failure)
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
}
|
|
ChannelJoinEvent::JoinTimeout { key, now: _ } => {
|
|
if key_matches(state.pending, key) {
|
|
let pending = state.pending.expect("key match requires pending");
|
|
state.authoritative.current_channel = pending.previous_confirmed_channel;
|
|
state.pending = None;
|
|
state.last_join_error = Some(JoinErrorCode::JoinTimeout);
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key });
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinTimeout,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::TimedOut
|
|
} else {
|
|
stale(state, &mut actions)
|
|
}
|
|
}
|
|
ChannelJoinEvent::AuthoritativeSelfMove {
|
|
channel,
|
|
epoch,
|
|
source: _,
|
|
} => {
|
|
if epoch != state.current_epoch() {
|
|
stale(state, &mut actions)
|
|
} else {
|
|
state.authoritative.current_channel = channel;
|
|
state.authoritative.confirmed_epoch = epoch;
|
|
if let Some(pending) = state.pending {
|
|
let key = pending_key(&pending);
|
|
state.pending = None;
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key });
|
|
if channel == Some(pending.target_channel) {
|
|
state.last_join_error = None;
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinConfirmed,
|
|
code: None,
|
|
});
|
|
JoinReduceStatus::Confirmed
|
|
} else {
|
|
state.last_join_error = Some(JoinErrorCode::JoinReconciledDifferentChannel);
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinSnapshotReconciled,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::ReconciledBySnapshot
|
|
}
|
|
} else {
|
|
JoinReduceStatus::Accepted
|
|
}
|
|
}
|
|
}
|
|
ChannelJoinEvent::UserLeaveRequested { now: _ } => {
|
|
let generation = state.allocate_generation();
|
|
let epoch = state.current_epoch();
|
|
if let Some(pending) = state.pending {
|
|
let key = pending_key(&pending);
|
|
state.pending = None;
|
|
state.last_join_error = Some(JoinErrorCode::JoinSupersededByLeave);
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout { key });
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinPendingSupersededByLeave,
|
|
code: state.last_join_error,
|
|
});
|
|
if state.authoritative.current_channel.is_some() {
|
|
actions.push(ChannelJoinAction::SendLeaveCommand {
|
|
current_channel: state.authoritative.current_channel,
|
|
generation,
|
|
epoch,
|
|
});
|
|
}
|
|
JoinReduceStatus::SupersededByLeave
|
|
} else {
|
|
if state.authoritative.current_channel.is_some() {
|
|
actions.push(ChannelJoinAction::SendLeaveCommand {
|
|
current_channel: state.authoritative.current_channel,
|
|
generation,
|
|
epoch,
|
|
});
|
|
}
|
|
JoinReduceStatus::Accepted
|
|
}
|
|
}
|
|
ChannelJoinEvent::ReconnectStarted { new_epoch, now: _ } => {
|
|
if let Some(pending) = state.pending {
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout {
|
|
key: pending_key(&pending),
|
|
});
|
|
}
|
|
state.pending = None;
|
|
state.sync_state = ChannelJoinSyncState::Synchronizing {
|
|
reason: SyncReason::Reconnect,
|
|
epoch: new_epoch,
|
|
};
|
|
state.authoritative.confirmed_epoch = new_epoch;
|
|
state.bump_generation();
|
|
state.last_join_error = None;
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinReconnectSynchronizing,
|
|
code: None,
|
|
});
|
|
JoinReduceStatus::Accepted
|
|
}
|
|
ChannelJoinEvent::SnapshotReady {
|
|
current_channel,
|
|
epoch,
|
|
now: _,
|
|
} => {
|
|
if epoch != state.current_epoch() {
|
|
stale(state, &mut actions)
|
|
} else {
|
|
let pending = state.pending;
|
|
if let Some(pending) = pending {
|
|
actions.push(ChannelJoinAction::CancelJoinTimeout {
|
|
key: pending_key(&pending),
|
|
});
|
|
}
|
|
state.authoritative.current_channel = current_channel;
|
|
state.authoritative.confirmed_epoch = epoch;
|
|
state.pending = None;
|
|
state.sync_state = ChannelJoinSyncState::Ready;
|
|
state.bump_generation();
|
|
state.last_join_error = match pending {
|
|
Some(pending) if current_channel != Some(pending.target_channel) => {
|
|
Some(JoinErrorCode::JoinReconciledDifferentChannel)
|
|
}
|
|
_ => None,
|
|
};
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinSnapshotReconciled,
|
|
code: state.last_join_error,
|
|
});
|
|
if pending.is_some() && state.last_join_error.is_none() {
|
|
JoinReduceStatus::Confirmed
|
|
} else {
|
|
JoinReduceStatus::ReconciledBySnapshot
|
|
}
|
|
}
|
|
}
|
|
};
|
|
|
|
finish(status, state, actions)
|
|
}
|
|
|
|
/// Build the channel-join projection for the current reducer state.
|
|
pub fn project(state: &ChannelJoinState) -> ChannelJoinProjection {
|
|
let pending = state.pending;
|
|
let synchronizing = matches!(state.sync_state, ChannelJoinSyncState::Synchronizing { .. });
|
|
ChannelJoinProjection {
|
|
current_channel: state.authoritative.current_channel,
|
|
pending_target: pending.map(|pending| pending.target_channel),
|
|
pending_since: pending.map(|pending| pending.started_at),
|
|
pending_generation: pending.map(|pending| pending.generation),
|
|
can_join: !synchronizing && pending.is_none(),
|
|
can_leave: !synchronizing
|
|
&& (state.authoritative.current_channel.is_some() || pending.is_some()),
|
|
sync_state: state.sync_state,
|
|
last_join_error: state.last_join_error,
|
|
}
|
|
}
|
|
|
|
fn finish(
|
|
status: JoinReduceStatus,
|
|
state: &ChannelJoinState,
|
|
mut actions: Vec<ChannelJoinAction>,
|
|
) -> ChannelJoinReduction {
|
|
let projection = project(state);
|
|
actions.push(ChannelJoinAction::PublishProjection(projection.clone()));
|
|
ChannelJoinReduction {
|
|
status,
|
|
projection,
|
|
actions,
|
|
}
|
|
}
|
|
|
|
fn pending_key(pending: &JoinPending) -> JoinOutcomeKey {
|
|
JoinOutcomeKey {
|
|
connection_epoch: pending.connection_epoch,
|
|
generation: pending.generation,
|
|
request_id: pending.request_id,
|
|
}
|
|
}
|
|
|
|
fn key_matches(pending: Option<JoinPending>, key: JoinOutcomeKey) -> bool {
|
|
pending.is_some_and(|pending| pending_key(&pending) == key)
|
|
}
|
|
|
|
fn stale(state: &mut ChannelJoinState, actions: &mut Vec<ChannelJoinAction>) -> JoinReduceStatus {
|
|
state.last_join_error = Some(JoinErrorCode::JoinStaleOutcomeIgnored);
|
|
actions.push(ChannelJoinAction::EmitDiagnostic {
|
|
key: JoinDiagnosticKey::JoinStaleOutcomeIgnored,
|
|
code: state.last_join_error,
|
|
});
|
|
JoinReduceStatus::StaleOutcomeIgnored
|
|
}
|
|
|
|
fn failure_error_code(failure: JoinFailureKind) -> JoinErrorCode {
|
|
match failure {
|
|
JoinFailureKind::Denied => JoinErrorCode::JoinDenied,
|
|
JoinFailureKind::Network => JoinErrorCode::JoinNetworkFailure,
|
|
JoinFailureKind::Protocol => JoinErrorCode::JoinProtocolFailure,
|
|
JoinFailureKind::Timeout => JoinErrorCode::JoinTimeout,
|
|
JoinFailureKind::Unknown => JoinErrorCode::JoinProtocolFailure,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn epoch(value: u64) -> ConnectionEpoch {
|
|
ConnectionEpoch(value)
|
|
}
|
|
|
|
fn channel(value: u64) -> ChannelId {
|
|
ChannelId(value)
|
|
}
|
|
|
|
fn request(value: u64) -> JoinRequestId {
|
|
JoinRequestId(value)
|
|
}
|
|
|
|
fn now(offset: u64) -> Instant {
|
|
Instant::now() + std::time::Duration::from_millis(offset)
|
|
}
|
|
|
|
fn accepted_pending(state: &mut ChannelJoinState, target: ChannelId) -> JoinOutcomeKey {
|
|
let start = now(1);
|
|
let reduction = reduce(
|
|
state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: target,
|
|
now: start,
|
|
},
|
|
);
|
|
let generation = reduction
|
|
.projection
|
|
.pending_generation
|
|
.expect("pending generation");
|
|
let request_id = request(7);
|
|
reduce(
|
|
state,
|
|
ChannelJoinEvent::JoinCommandAccepted {
|
|
generation,
|
|
request_id,
|
|
},
|
|
);
|
|
JoinOutcomeKey {
|
|
connection_epoch: epoch(1),
|
|
generation,
|
|
request_id: Some(request_id),
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn user_join_creates_pending_and_send_start_timeout_actions() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
let start = now(1);
|
|
|
|
let reduction = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: channel(10),
|
|
now: start,
|
|
},
|
|
);
|
|
|
|
assert_eq!(reduction.status, JoinReduceStatus::Accepted);
|
|
assert_eq!(reduction.projection.pending_target, Some(channel(10)));
|
|
assert!(matches!(
|
|
reduction.actions[0],
|
|
ChannelJoinAction::SendJoinCommand {
|
|
target_channel: ChannelId(10),
|
|
generation: JoinGeneration(1),
|
|
epoch: ConnectionEpoch(1)
|
|
}
|
|
));
|
|
assert!(
|
|
matches!(reduction.actions[1], ChannelJoinAction::StartJoinTimeout { key: JoinOutcomeKey { connection_epoch: ConnectionEpoch(1), generation: JoinGeneration(1), request_id: None }, started_at } if started_at == start)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn duplicate_same_target_coalesces() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: channel(10),
|
|
now: now(1),
|
|
},
|
|
);
|
|
|
|
let reduction = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: channel(10),
|
|
now: now(2),
|
|
},
|
|
);
|
|
|
|
assert_eq!(reduction.status, JoinReduceStatus::CoalescedSameTarget);
|
|
assert_eq!(
|
|
reduction.projection.pending_generation,
|
|
Some(JoinGeneration(1))
|
|
);
|
|
assert!(!reduction.actions.iter().any(|action| matches!(
|
|
action,
|
|
ChannelJoinAction::SendJoinCommand { .. } | ChannelJoinAction::StartJoinTimeout { .. }
|
|
)));
|
|
}
|
|
|
|
#[test]
|
|
fn different_target_while_pending_rejects_and_serializes() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: channel(10),
|
|
now: now(1),
|
|
},
|
|
);
|
|
|
|
let reduction = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserJoinRequested {
|
|
target_channel: channel(11),
|
|
now: now(2),
|
|
},
|
|
);
|
|
|
|
assert_eq!(
|
|
reduction.status,
|
|
JoinReduceStatus::Rejected(JoinIntentRejected::JoinAlreadyPendingDifferentTarget)
|
|
);
|
|
assert_eq!(reduction.projection.pending_target, Some(channel(10)));
|
|
assert_eq!(
|
|
reduction.projection.last_join_error,
|
|
Some(JoinErrorCode::JoinAlreadyPendingDifferentTarget)
|
|
);
|
|
assert!(!reduction
|
|
.actions
|
|
.iter()
|
|
.any(|action| matches!(action, ChannelJoinAction::SendJoinCommand { .. })));
|
|
}
|
|
|
|
#[test]
|
|
fn authoritative_current_channel_not_changed_until_self_move_or_snapshot() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
state.authoritative.current_channel = Some(channel(1));
|
|
let key = accepted_pending(&mut state, channel(2));
|
|
|
|
let success = reduce(&mut state, ChannelJoinEvent::ProtocolJoinSucceeded { key });
|
|
assert_eq!(success.projection.current_channel, Some(channel(1)));
|
|
assert_eq!(success.projection.pending_target, Some(channel(2)));
|
|
|
|
let confirmed = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::AuthoritativeSelfMove {
|
|
channel: Some(channel(2)),
|
|
epoch: epoch(1),
|
|
source: AuthoritativeSource::LiveDelta,
|
|
},
|
|
);
|
|
assert_eq!(confirmed.status, JoinReduceStatus::Confirmed);
|
|
assert_eq!(confirmed.projection.current_channel, Some(channel(2)));
|
|
assert_eq!(confirmed.projection.pending_target, None);
|
|
}
|
|
|
|
#[test]
|
|
fn failure_and_timeout_clear_pending_and_preserve_previous_authoritative_channel() {
|
|
let mut failed = ChannelJoinState::new(epoch(1));
|
|
failed.authoritative.current_channel = Some(channel(1));
|
|
let fail_key = accepted_pending(&mut failed, channel(2));
|
|
|
|
let fail_reduction = reduce(
|
|
&mut failed,
|
|
ChannelJoinEvent::ProtocolJoinFailed {
|
|
key: fail_key,
|
|
failure: JoinFailureKind::Denied,
|
|
},
|
|
);
|
|
assert_eq!(
|
|
fail_reduction.status,
|
|
JoinReduceStatus::Failed(JoinFailureKind::Denied)
|
|
);
|
|
assert_eq!(fail_reduction.projection.current_channel, Some(channel(1)));
|
|
assert_eq!(fail_reduction.projection.pending_target, None);
|
|
|
|
let mut timed_out = ChannelJoinState::new(epoch(1));
|
|
timed_out.authoritative.current_channel = Some(channel(1));
|
|
let timeout_key = accepted_pending(&mut timed_out, channel(2));
|
|
let timeout_reduction = reduce(
|
|
&mut timed_out,
|
|
ChannelJoinEvent::JoinTimeout {
|
|
key: timeout_key,
|
|
now: now(9),
|
|
},
|
|
);
|
|
assert_eq!(timeout_reduction.status, JoinReduceStatus::TimedOut);
|
|
assert_eq!(
|
|
timeout_reduction.projection.current_channel,
|
|
Some(channel(1))
|
|
);
|
|
assert_eq!(timeout_reduction.projection.pending_target, None);
|
|
}
|
|
|
|
#[test]
|
|
fn stale_outcomes_ignored_by_epoch_generation_or_request_mismatch() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
state.authoritative.current_channel = Some(channel(1));
|
|
let key = accepted_pending(&mut state, channel(2));
|
|
|
|
for stale_key in [
|
|
JoinOutcomeKey {
|
|
connection_epoch: epoch(2),
|
|
..key
|
|
},
|
|
JoinOutcomeKey {
|
|
generation: JoinGeneration(99),
|
|
..key
|
|
},
|
|
JoinOutcomeKey {
|
|
request_id: Some(request(99)),
|
|
..key
|
|
},
|
|
] {
|
|
let reduction = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::ProtocolJoinFailed {
|
|
key: stale_key,
|
|
failure: JoinFailureKind::Network,
|
|
},
|
|
);
|
|
assert_eq!(reduction.status, JoinReduceStatus::StaleOutcomeIgnored);
|
|
assert_eq!(reduction.projection.current_channel, Some(channel(1)));
|
|
assert_eq!(reduction.projection.pending_target, Some(channel(2)));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn leave_supersedes_pending_and_emits_leave_action_if_current_exists() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
state.authoritative.current_channel = Some(channel(1));
|
|
let key = accepted_pending(&mut state, channel(2));
|
|
|
|
let reduction = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::UserLeaveRequested { now: now(3) },
|
|
);
|
|
assert_eq!(reduction.status, JoinReduceStatus::SupersededByLeave);
|
|
assert_eq!(reduction.projection.current_channel, Some(channel(1)));
|
|
assert_eq!(reduction.projection.pending_target, None);
|
|
assert!(reduction.actions.iter().any(|action| matches!(
|
|
action,
|
|
ChannelJoinAction::SendLeaveCommand {
|
|
current_channel: Some(ChannelId(1)),
|
|
generation: JoinGeneration(2),
|
|
epoch: ConnectionEpoch(1)
|
|
}
|
|
)));
|
|
|
|
let stale_success = reduce(&mut state, ChannelJoinEvent::ProtocolJoinSucceeded { key });
|
|
assert_eq!(stale_success.status, JoinReduceStatus::StaleOutcomeIgnored);
|
|
assert_eq!(stale_success.projection.current_channel, Some(channel(1)));
|
|
}
|
|
|
|
#[test]
|
|
fn reconnect_clears_stales_pending_and_updates_sync_epoch() {
|
|
let mut state = ChannelJoinState::new(epoch(1));
|
|
state.authoritative.current_channel = Some(channel(1));
|
|
let key = accepted_pending(&mut state, channel(2));
|
|
|
|
let reconnect = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::ReconnectStarted {
|
|
new_epoch: epoch(2),
|
|
now: now(4),
|
|
},
|
|
);
|
|
assert_eq!(reconnect.projection.pending_target, None);
|
|
assert_eq!(
|
|
reconnect.projection.sync_state,
|
|
ChannelJoinSyncState::Synchronizing {
|
|
reason: SyncReason::Reconnect,
|
|
epoch: epoch(2)
|
|
}
|
|
);
|
|
assert!(!reconnect.projection.can_join);
|
|
|
|
let stale_failure = reduce(
|
|
&mut state,
|
|
ChannelJoinEvent::ProtocolJoinFailed {
|
|
key,
|
|
failure: JoinFailureKind::Network,
|
|
},
|
|
);
|
|
assert_eq!(stale_failure.status, JoinReduceStatus::StaleOutcomeIgnored);
|
|
}
|
|
|
|
#[test]
|
|
fn snapshot_replaces_authoritative_membership_and_clears_or_reconciles_pending() {
|
|
let mut confirmed = ChannelJoinState::new(epoch(1));
|
|
accepted_pending(&mut confirmed, channel(2));
|
|
let confirm = reduce(
|
|
&mut confirmed,
|
|
ChannelJoinEvent::SnapshotReady {
|
|
current_channel: Some(channel(2)),
|
|
epoch: epoch(1),
|
|
now: now(5),
|
|
},
|
|
);
|
|
assert_eq!(confirm.status, JoinReduceStatus::Confirmed);
|
|
assert_eq!(confirm.projection.current_channel, Some(channel(2)));
|
|
assert_eq!(confirm.projection.pending_target, None);
|
|
|
|
let mut reconciled = ChannelJoinState::new(epoch(1));
|
|
accepted_pending(&mut reconciled, channel(2));
|
|
let reconcile = reduce(
|
|
&mut reconciled,
|
|
ChannelJoinEvent::SnapshotReady {
|
|
current_channel: Some(channel(3)),
|
|
epoch: epoch(1),
|
|
now: now(6),
|
|
},
|
|
);
|
|
assert_eq!(reconcile.status, JoinReduceStatus::ReconciledBySnapshot);
|
|
assert_eq!(reconcile.projection.current_channel, Some(channel(3)));
|
|
assert_eq!(reconcile.projection.pending_target, None);
|
|
assert_eq!(
|
|
reconcile.projection.last_join_error,
|
|
Some(JoinErrorCode::JoinReconciledDifferentChannel)
|
|
);
|
|
}
|
|
}
|