1use std::collections::{HashMap, VecDeque};2use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};34use anyhow::{Context, Result, bail, ensure};5use sigil_protocol::{6 ControllerSlot, FocusProposalV2, FocusStateV2, FocusTransitionReasonV2, ViewerPresenceId,7};89const FOCUS_PROPOSAL_TTL: Duration = Duration::from_secs(15);10const FOCUS_ACTIVATION_TTL: Duration = Duration::from_secs(10);11const FOCUS_COMMAND_RATE_WINDOW: Duration = Duration::from_secs(2);12// Eight commands per two-second window permits one complete request/approve/13// activate/release exchange with retry headroom for each of the hard maximum14// eight viewers, while each viewer retains only its own bounded timestamps.15const MAX_FOCUS_COMMANDS_PER_WINDOW: usize = 8;161718pub(super) struct FocusCandidate {19 pub(super) presence_id: ViewerPresenceId,20 pub(super) session_id: u64,21}222324pub(crate) struct FocusNeutralization {25 pub(crate) transition_id: u64,26 pub(crate) former_session_id: u64,27}282930pub(super) struct FocusMutation {31 pub(super) changed: bool,32 pub(super) neutralization: Option<FocusNeutralization>,33}343536struct FocusSuccessor {37 candidate: FocusCandidate,38}394041struct ActivationDeadline {42 presence_id: ViewerPresenceId,43 session_id: u64,44 focus_generation: u64,45 deadline: Instant,46}474849pub(super) struct FocusArbiter {50 state: FocusStateV2,51 proposal: Option<FocusProposalV2>,52 proposal_deadline: Option<Instant>,53 transition_reason: FocusTransitionReasonV2,54 configured_owner: Option<ViewerPresenceId>,55 successor: Option<FocusSuccessor>,56 activation: Option<ActivationDeadline>,57 next_focus_generation: u64,58 next_transition_id: u64,59 next_proposal_id: u64,60 command_times: HashMap<(ViewerPresenceId, u64), VecDeque<Instant>>,61}6263impl FocusArbiter {64 pub(super) fn new(configured_owner: Option<ViewerPresenceId>) -> Self {65 Self {66 state: FocusStateV2::Vacant {67 slot: ControllerSlot::ZERO,68 },69 proposal: None,70 proposal_deadline: None,71 transition_reason: FocusTransitionReasonV2::Initial,72 configured_owner,73 successor: None,74 activation: None,75 next_focus_generation: 0,76 next_transition_id: 0,77 next_proposal_id: 0,78 command_times: HashMap::new(),79 }80 }8182 pub(super) fn state(&self) -> &FocusStateV2 {83 &self.state84 }8586 pub(super) fn proposal(&self) -> Option<&FocusProposalV2> {87 self.proposal.as_ref()88 }8990 pub(super) fn transition_reason(&self) -> FocusTransitionReasonV2 {91 self.transition_reason92 }9394 pub(super) fn is_configured_owner(&self, presence_id: &ViewerPresenceId) -> bool {95 self.configured_owner.as_ref() == Some(presence_id)96 }9798 pub(super) fn check_rate_limit(99 &mut self,100 requester: &FocusCandidate,101 now: Instant,102 ) -> Result<()> {103 let key = (requester.presence_id.clone(), requester.session_id);104 let entries = self.command_times.entry(key).or_default();105 while entries106 .front()107 .is_some_and(|time| now.duration_since(*time) >= FOCUS_COMMAND_RATE_WINDOW)108 {109 entries.pop_front();110 }111 ensure!(112 entries.len() < MAX_FOCUS_COMMANDS_PER_WINDOW,113 "focus command rate limit exceeded"114 );115 entries.push_back(now);116 Ok(())117 }118119 120 pub(super) fn tracked_command_candidates(&self) -> usize {121 self.command_times.len()122 }123124 pub(super) fn retire_candidate(&mut self, candidate: &FocusCandidate) {125 self.command_times126 .remove(&(candidate.presence_id.clone(), candidate.session_id));127 }128129 pub(super) fn expire_proposal(&mut self, now: Instant) -> bool {130 if self.proposal_deadline.is_none_or(|deadline| deadline > now) {131 return false;132 }133 self.clear_proposal();134 self.transition_reason = FocusTransitionReasonV2::ProposalExpired;135 true136 }137138 pub(super) fn request(139 &mut self,140 requester: FocusCandidate,141 now: Instant,142 ) -> Result<FocusMutation> {143 self.expire_proposal(now);144 match &self.state {145 FocusStateV2::Vacant { slot } => {146 let slot = *slot;147 self.grant(slot, requester, now)?;148 self.transition_reason = FocusTransitionReasonV2::Requested;149 Ok(FocusMutation {150 changed: true,151 neutralization: None,152 })153 }154 FocusStateV2::Held {155 slot,156 holder,157 session_id,158 ..159 } => {160 ensure!(161 holder != &requester.presence_id || *session_id != requester.session_id,162 "current holder cannot request focus from itself"163 );164 ensure!(165 self.proposal.is_none(),166 "controller slot 0 already has a pending handoff proposal"167 );168 self.next_proposal_id = next_nonzero(self.next_proposal_id, "focus proposal id")?;169 let expires_at_unix_ms = unix_millis_after(FOCUS_PROPOSAL_TTL)?;170 self.proposal = Some(FocusProposalV2 {171 proposal_id: self.next_proposal_id,172 slot: *slot,173 requester: requester.presence_id,174 requester_session_id: requester.session_id,175 holder: holder.clone(),176 holder_session_id: *session_id,177 expires_at_unix_ms,178 });179 self.proposal_deadline = Some(now + FOCUS_PROPOSAL_TTL);180 self.transition_reason = FocusTransitionReasonV2::HandoffRequested;181 Ok(FocusMutation {182 changed: true,183 neutralization: None,184 })185 }186 FocusStateV2::Neutralizing { .. } => {187 bail!("controller slot 0 is neutralizing")188 }189 }190 }191192 pub(super) fn approve(193 &mut self,194 holder: &FocusCandidate,195 expected_focus_generation: u64,196 expected_proposal_id: u64,197 ) -> Result<FocusMutation> {198 self.ensure_holder(holder, expected_focus_generation)?;199 let proposal = self.ensure_proposal(holder, expected_proposal_id)?.clone();200 self.begin_neutralization(201 FocusTransitionReasonV2::HandoffApproved,202 Some(FocusCandidate {203 presence_id: proposal.requester,204 session_id: proposal.requester_session_id,205 }),206 )207 }208209 pub(super) fn deny(210 &mut self,211 holder: &FocusCandidate,212 expected_focus_generation: u64,213 expected_proposal_id: u64,214 ) -> Result<FocusMutation> {215 self.ensure_holder(holder, expected_focus_generation)?;216 self.ensure_proposal(holder, expected_proposal_id)?;217 self.clear_proposal();218 self.transition_reason = FocusTransitionReasonV2::HandoffDenied;219 Ok(FocusMutation {220 changed: true,221 neutralization: None,222 })223 }224225 pub(super) fn release(226 &mut self,227 holder: &FocusCandidate,228 expected_focus_generation: u64,229 ) -> Result<FocusMutation> {230 self.ensure_holder(holder, expected_focus_generation)?;231 let successor = self.proposal.as_ref().map(|proposal| FocusCandidate {232 presence_id: proposal.requester.clone(),233 session_id: proposal.requester_session_id,234 });235 self.begin_neutralization(FocusTransitionReasonV2::Released, successor)236 }237238 pub(super) fn preempt(239 &mut self,240 owner: FocusCandidate,241 expected_focus_generation: u64,242 ) -> Result<FocusMutation> {243 ensure!(244 self.is_configured_owner(&owner.presence_id),245 "viewer is not the configured focus owner"246 );247 let FocusStateV2::Held {248 holder,249 session_id,250 focus_generation,251 ..252 } = &self.state253 else {254 bail!("controller slot 0 is not held")255 };256 ensure!(257 *focus_generation == expected_focus_generation,258 "preemption expected a stale focus generation"259 );260 ensure!(261 holder != &owner.presence_id || *session_id != owner.session_id,262 "configured owner already holds focus"263 );264 self.begin_neutralization(FocusTransitionReasonV2::Preempted, Some(owner))265 }266267 pub(super) fn begin_invalidation(268 &mut self,269 candidate: &FocusCandidate,270 reason: FocusTransitionReasonV2,271 ) -> Result<FocusMutation> {272 let proposal_involved = self.proposal.as_ref().is_some_and(|proposal| {273 proposal.requester == candidate.presence_id274 && proposal.requester_session_id == candidate.session_id275 || proposal.holder == candidate.presence_id276 && proposal.holder_session_id == candidate.session_id277 });278 if proposal_involved {279 self.clear_proposal();280 }281 let held = matches!(282 &self.state,283 FocusStateV2::Held { holder, session_id, .. }284 if holder == &candidate.presence_id && *session_id == candidate.session_id285 );286 if held {287 return self.begin_neutralization(reason, None);288 }289 if proposal_involved {290 self.transition_reason = reason;291 }292 Ok(FocusMutation {293 changed: proposal_involved,294 neutralization: None,295 })296 }297298 pub(super) fn mark_activated(&mut self, candidate: &FocusCandidate, focus_generation: u64) {299 if self.activation.as_ref().is_some_and(|activation| {300 activation.presence_id == candidate.presence_id301 && activation.session_id == candidate.session_id302 && activation.focus_generation == focus_generation303 }) {304 self.activation = None;305 }306 }307308 pub(super) fn begin_activation_expiry(&mut self, now: Instant) -> Result<FocusMutation> {309 let Some(activation) = self.activation.as_ref() else {310 return Ok(FocusMutation::default());311 };312 if activation.deadline > now {313 return Ok(FocusMutation::default());314 }315 let candidate = FocusCandidate {316 presence_id: activation.presence_id.clone(),317 session_id: activation.session_id,318 };319 let focus_generation = activation.focus_generation;320 self.ensure_holder(&candidate, focus_generation)?;321 self.begin_neutralization(FocusTransitionReasonV2::ActivationExpired, None)322 }323324 pub(super) fn transition_successor(&self, transition_id: u64) -> Option<&FocusCandidate> {325 match &self.state {326 FocusStateV2::Neutralizing {327 transition_id: current,328 ..329 } if *current == transition_id => self330 .successor331 .as_ref()332 .map(|successor| &successor.candidate),333 _ => None,334 }335 }336337 pub(super) fn complete_transition(338 &mut self,339 transition_id: u64,340 successor_is_valid: bool,341 now: Instant,342 ) -> Result<bool> {343 let FocusStateV2::Neutralizing {344 slot,345 transition_id: current,346 ..347 } = self.state348 else {349 return Ok(false);350 };351 if current != transition_id {352 return Ok(false);353 }354 let successor = self.successor.take().filter(|_| successor_is_valid);355 self.clear_proposal();356 if let Some(successor) = successor {357 self.grant(slot, successor.candidate, now)?;358 if self.transition_reason == FocusTransitionReasonV2::Released {359 self.transition_reason = FocusTransitionReasonV2::HandoffApproved;360 }361 } else {362 self.state = FocusStateV2::Vacant { slot };363 self.activation = None;364 }365 Ok(true)366 }367368 fn ensure_holder(369 &self,370 candidate: &FocusCandidate,371 expected_focus_generation: u64,372 ) -> Result<()> {373 ensure!(374 matches!(375 &self.state,376 FocusStateV2::Held {377 holder,378 session_id,379 focus_generation,380 ..381 } if holder == &candidate.presence_id382 && *session_id == candidate.session_id383 && *focus_generation == expected_focus_generation384 ),385 "focus command does not match the current holder generation"386 );387 Ok(())388 }389390 fn ensure_proposal(391 &self,392 holder: &FocusCandidate,393 expected_proposal_id: u64,394 ) -> Result<&FocusProposalV2> {395 self.proposal396 .as_ref()397 .filter(|proposal| {398 proposal.proposal_id == expected_proposal_id399 && proposal.holder == holder.presence_id400 && proposal.holder_session_id == holder.session_id401 })402 .context("handoff decision does not match the pending proposal")403 }404405 fn begin_neutralization(406 &mut self,407 reason: FocusTransitionReasonV2,408 successor: Option<FocusCandidate>,409 ) -> Result<FocusMutation> {410 let FocusStateV2::Held {411 slot,412 holder,413 session_id,414 focus_generation,415 } = self.state.clone()416 else {417 bail!("controller slot 0 is not held")418 };419 self.next_transition_id = next_nonzero(self.next_transition_id, "focus transition id")?;420 self.state = FocusStateV2::Neutralizing {421 slot,422 former_holder: holder,423 former_session_id: session_id,424 former_focus_generation: focus_generation,425 transition_id: self.next_transition_id,426 };427 self.transition_reason = reason;428 self.successor = successor.map(|candidate| FocusSuccessor { candidate });429 self.activation = None;430 self.clear_proposal();431 Ok(FocusMutation {432 changed: true,433 neutralization: Some(FocusNeutralization {434 transition_id: self.next_transition_id,435 former_session_id: session_id,436 }),437 })438 }439440 fn grant(441 &mut self,442 slot: ControllerSlot,443 candidate: FocusCandidate,444 now: Instant,445 ) -> Result<()> {446 self.next_focus_generation = next_nonzero(self.next_focus_generation, "focus generation")?;447 self.state = FocusStateV2::Held {448 slot,449 holder: candidate.presence_id.clone(),450 session_id: candidate.session_id,451 focus_generation: self.next_focus_generation,452 };453 self.activation = Some(ActivationDeadline {454 presence_id: candidate.presence_id,455 session_id: candidate.session_id,456 focus_generation: self.next_focus_generation,457 deadline: now + FOCUS_ACTIVATION_TTL,458 });459 Ok(())460 }461462 fn clear_proposal(&mut self) {463 self.proposal = None;464 self.proposal_deadline = None;465 }466}467468fn next_nonzero(current: u64, name: &'static str) -> Result<u64> {469 current470 .checked_add(1)471 .with_context(|| format!("{name} exhausted"))472}473474fn unix_millis_after(duration: Duration) -> Result<u64> {475 let now = SystemTime::now()476 .duration_since(UNIX_EPOCH)477 .context("system clock is before the Unix epoch")?;478 u64::try_from((now + duration).as_millis()).context("focus proposal expiry overflowed")479}480481482mod tests {483 use super::*;484485 fn candidate(handle: &str, session_id: u64) -> FocusCandidate {486 FocusCandidate {487 presence_id: ViewerPresenceId::new(handle).unwrap(),488 session_id,489 }490 }491492 493 fn handoff_publishes_neutralizing_before_successor_generation() {494 let now = Instant::now();495 let first = candidate("viewer-0000000000000001", 1);496 let second = candidate("viewer-0000000000000002", 2);497 let mut focus = FocusArbiter::new(None);498 focus.request(first.clone(), now).unwrap();499 let generation = match focus.state() {500 FocusStateV2::Held {501 focus_generation, ..502 } => *focus_generation,503 _ => panic!("focus was not granted"),504 };505 focus.request(second.clone(), now).unwrap();506 let proposal_id = focus.proposal().unwrap().proposal_id;507 let mutation = focus.approve(&first, generation, proposal_id).unwrap();508 let transition = mutation.neutralization.unwrap();509 assert!(matches!(focus.state(), FocusStateV2::Neutralizing { .. }));510 assert_eq!(511 focus.transition_successor(transition.transition_id),512 Some(&second)513 );514 focus515 .complete_transition(transition.transition_id, true, now)516 .unwrap();517 assert!(matches!(518 focus.state(),519 FocusStateV2::Held { holder, session_id: 2, focus_generation, .. }520 if holder == &second.presence_id && *focus_generation > generation521 ));522 }523524 525 fn only_configured_owner_can_preempt_and_commands_are_rate_limited() {526 let now = Instant::now();527 let owner = candidate("viewer-0000000000000001", 1);528 let holder = candidate("viewer-0000000000000002", 2);529 let mut focus = FocusArbiter::new(Some(owner.presence_id.clone()));530 focus.request(holder.clone(), now).unwrap();531 let generation = match focus.state() {532 FocusStateV2::Held {533 focus_generation, ..534 } => *focus_generation,535 _ => unreachable!(),536 };537 assert!(538 focus539 .preempt(candidate("viewer-0000000000000003", 3), generation)540 .is_err()541 );542 assert!(focus.preempt(owner.clone(), generation).is_ok());543544 let mut limited = FocusArbiter::new(None);545 for _ in 0..MAX_FOCUS_COMMANDS_PER_WINDOW {546 limited.check_rate_limit(&owner, now).unwrap();547 }548 assert!(limited.check_rate_limit(&owner, now).is_err());549 }550551 552 fn proposals_expire_without_creating_a_queue() {553 let now = Instant::now();554 let first = candidate("viewer-0000000000000001", 1);555 let second = candidate("viewer-0000000000000002", 2);556 let third = candidate("viewer-0000000000000003", 3);557 let mut focus = FocusArbiter::new(None);558 focus.request(first, now).unwrap();559 focus.request(second, now).unwrap();560 assert!(focus.request(third.clone(), now).is_err());561 assert!(focus.expire_proposal(now + FOCUS_PROPOSAL_TTL));562 assert!(focus.request(third, now + FOCUS_PROPOSAL_TTL).is_ok());563 }564}