behavior/
pool.rs

1//! A bounded, FIFO worker pool expressed entirely as a pure behavior.
2//!
3//! Pool scheduling is a derived Bombay construction, not an actor-model
4//! primitive. Runtime installation, delivery, and observation remain effects
5//! for an interpreter; this module owns only their typed protocol and fold.
6
7use core::convert::Infallible;
8use core::marker::PhantomData;
9use std::collections::{BTreeMap, VecDeque};
10use std::time::Duration;
11
12use crate::{
13    Actions, Address, Behavior, Births, Crash, CreationRejection, Delivery, Exit, Never, Own,
14    Proxy, ProxyCommand, Recipient, RestartPolicy, SendAlgebra, SendInput, Strategy,
15    SupervisionEvent, Supervisor, SupervisorSends, User, WorkerCreationResolved, WorkerStopped,
16    delegate_transition,
17};
18
19/// Caller-chosen identity used to correlate pool responses.
20#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
21pub struct JobId(pub u64);
22
23/// Pool-owned correlation token for one exact dispatch attempt.
24///
25/// This is not an actor identity or evidence that delivery occurred.
26#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
27pub struct AssignmentId(pub u64);
28
29/// One assignment accepted by a worker behavior.
30#[derive(Debug, Clone, PartialEq, Eq)]
31pub struct PoolAssignment<J> {
32    pub assignment: AssignmentId,
33    pub job: JobId,
34    pub payload: J,
35}
36
37/// Messages accepted by a pool coordinator.
38#[derive(Clone, PartialEq, Eq)]
39pub enum PoolMessage<A, D, J, R>
40where
41    A: Address,
42    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
43{
44    Submit {
45        job: JobId,
46        payload: J,
47        reply_to: Recipient<D>,
48    },
49    Completed {
50        worker: <D::Addr as Address>::Nonce,
51        assignment: AssignmentId,
52        result: R,
53    },
54}
55
56/// Messages accepted by a key-persistent pool coordinator.
57///
58/// `Rebalance` is the only input that can change an established key binding.
59/// It affects later submissions; jobs already accepted retain their selected
60/// stable worker slot.
61#[derive(Clone, PartialEq, Eq)]
62pub enum KeyedPoolMessage<A, D, K, J, R>
63where
64    A: Address,
65    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
66{
67    Submit {
68        key: K,
69        job: JobId,
70        payload: J,
71        reply_to: Recipient<D>,
72    },
73    Completed {
74        worker: <D::Addr as Address>::Nonce,
75        assignment: AssignmentId,
76        result: R,
77    },
78    Rebalance {
79        key: K,
80        worker: <D::Addr as Address>::Nonce,
81    },
82}
83
84/// Why a submitted job was not accepted by the pool.
85#[derive(Debug, Clone, Copy, PartialEq, Eq)]
86pub enum PoolRejection {
87    BacklogFull,
88    /// The key's selected stable slot is unknown or permanently retired.
89    AffinityUnavailable,
90}
91
92/// Why an accepted assignment ended without a worker completion.
93#[derive(Clone, PartialEq, Eq)]
94pub enum PoolInterruption<A: Address> {
95    WorkerStopped {
96        worker: A::Nonce,
97        outcome: Result<Exit<A>, Crash>,
98    },
99    NoRecoverableWorkers,
100    /// The job's selected stable slot retired while the job was queued.
101    AffinityRetired {
102        worker: A::Nonce,
103        reason: WorkerRetirement,
104    },
105}
106
107/// Complete response protocol for one submitted job.
108#[derive(Clone, PartialEq, Eq)]
109pub enum PoolResponse<J, R, A: Address> {
110    Accepted {
111        job: JobId,
112    },
113    Rejected {
114        job: JobId,
115        payload: J,
116        reason: PoolRejection,
117    },
118    Completed {
119        job: JobId,
120        result: R,
121    },
122    Interrupted {
123        job: JobId,
124        payload: J,
125        reason: PoolInterruption<A>,
126    },
127}
128
129/// Bombay policy for an assigned job whose worker incarnation stops.
130#[derive(Debug, Clone, Copy, PartialEq, Eq)]
131pub enum InterruptionPolicy {
132    /// End pool ownership and report the still-owned job to its submitter.
133    Fail,
134    /// Put the job at the front of the backlog for at-least-once assignment.
135    Retry,
136}
137
138/// Public, payload-free view of one stable worker slot.
139#[derive(Debug, Clone, Copy, PartialEq, Eq)]
140pub enum WorkerPhase {
141    Installing,
142    Idle,
143    Assigned {
144        assignment: AssignmentId,
145        job: JobId,
146    },
147    Retired {
148        reason: WorkerRetirement,
149    },
150}
151
152/// Why a stable worker slot is no longer eligible for dispatch.
153#[derive(Debug, Clone, Copy, PartialEq, Eq)]
154pub enum WorkerRetirement {
155    CreationRejected(CreationRejection),
156    ReplacementUnavailable,
157}
158
159/// Invalid static pool topology.
160#[derive(Debug, Clone, Copy, PartialEq, Eq)]
161pub enum PoolConfigError<N> {
162    /// No stable worker slot exists, so accepted ownership could never end.
163    NoWorkers,
164    /// Two configured positions selected the same stable worker nonce.
165    DuplicateWorker(N),
166}
167
168/// Pure, statically dispatched policy for a previously unseen affinity key.
169pub trait AffinitySelector<K, N> {
170    /// Select the stable worker nonce for a key that has no binding yet.
171    fn select(&self, key: &K) -> N;
172}
173
174impl<K, N, F> AffinitySelector<K, N> for F
175where
176    F: Fn(&K) -> N,
177{
178    fn select(&self, key: &K) -> N {
179        self(key)
180    }
181}
182
183/// Typed rejection of an event that cannot apply to the current pool state.
184#[derive(Debug, Clone, Copy, PartialEq, Eq)]
185pub enum PoolError<N> {
186    UnknownWorker(N),
187    CompletionForUnavailableWorker {
188        worker: N,
189        phase: WorkerPhase,
190    },
191    StaleCompletion {
192        worker: N,
193        expected: AssignmentId,
194        received: AssignmentId,
195    },
196    WorkerStoppedWhileUnavailable {
197        worker: N,
198        phase: WorkerPhase,
199    },
200    CreationResolvedWhileUnavailable {
201        worker: N,
202        phase: WorkerPhase,
203    },
204    RebalanceToRetiredWorker {
205        worker: N,
206        reason: WorkerRetirement,
207    },
208}
209
210struct AcceptedJob<A: Address, D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>, J, R> {
211    id: JobId,
212    payload: J,
213    reply_to: Recipient<D>,
214    interruption: Option<PoolInterruption<A>>,
215    target: Option<A::Nonce>,
216}
217
218struct QueuedJob<A: Address, D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>, J, R> {
219    accepted: AcceptedJob<A, D, J, R>,
220    dispatch_payload: J,
221}
222
223enum SlotState<A: Address, D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>, J, R> {
224    Installing,
225    Idle,
226    Assigned {
227        assignment: AssignmentId,
228        job: AcceptedJob<A, D, J, R>,
229    },
230    Retired {
231        reason: WorkerRetirement,
232    },
233}
234
235struct Slot<A: Address, D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>, J, R> {
236    nonce: A::Nonce,
237    state: SlotState<A, D, J, R>,
238}
239
240struct PlannedDispatch {
241    slot_position: usize,
242    job_position: usize,
243}
244
245enum Admission {
246    Accepted,
247    Rejected,
248}
249
250/// The pool's concrete event sum, including existing supervision facts.
251pub type PoolEvent<A, D, J, R> = SupervisionEvent<User<A, PoolMessage<A, D, J, R>>>;
252
253/// Concrete event sum for a [`KeyedWorkerPool`].
254pub type KeyedPoolEvent<A, D, K, J, R> = SupervisionEvent<User<A, KeyedPoolMessage<A, D, K, J, R>>>;
255
256/// Named pool-owned delivery lanes.
257pub struct PoolBehaviorSends<A, D, J, R, C>
258where
259    A: Address,
260    A::Nonce: From<u64>,
261    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
262    C: Behavior<Addr = A, Ph = Never>,
263{
264    /// Admission and terminal responses addressed to submitters.
265    pub responses: Vec<Delivery<D>>,
266    /// Assignments addressed to the selected stable worker proxies.
267    pub assignments: Vec<Delivery<Proxy<C>>>,
268}
269
270impl<A, D, J, R, C> SendAlgebra for PoolBehaviorSends<A, D, J, R, C>
271where
272    A: Address,
273    A::Nonce: From<u64>,
274    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
275    C: Behavior<Addr = A, Ph = Never>,
276{
277    fn empty() -> Self {
278        Self {
279            responses: Vec::new(),
280            assignments: Vec::new(),
281        }
282    }
283
284    fn append(&mut self, mut other: Self) {
285        self.responses.append(&mut other.responses);
286        self.assignments.append(&mut other.assignments);
287    }
288}
289
290impl<A, D, J, R, C> SendInput<Delivery<D>, Own> for PoolBehaviorSends<A, D, J, R, C>
291where
292    A: Address,
293    A::Nonce: From<u64>,
294    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
295    C: Behavior<Addr = A, Ph = Never>,
296{
297    fn emit(&mut self, input: Delivery<D>) {
298        self.responses.push(input);
299    }
300}
301
302impl<A, D, J, R, C> SendInput<Delivery<Proxy<C>>, Own> for PoolBehaviorSends<A, D, J, R, C>
303where
304    A: Address,
305    A::Nonce: From<u64>,
306    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
307    C: Behavior<Addr = A, Ph = Never>,
308{
309    fn emit(&mut self, input: Delivery<Proxy<C>>) {
310        self.assignments.push(input);
311    }
312}
313
314type KernelSends<A, D, J, R, C> = PoolBehaviorSends<A, D, J, R, C>;
315
316/// Pool effects keep responses and assignments in named, independently
317/// appendable lanes within the supervised behavior send product.
318pub type PoolSends<A, D, J, R, C> = SupervisorSends<A, KernelSends<A, D, J, R, C>, C>;
319
320/// Complete action type returned by a [`WorkerPool`] transition.
321pub type PoolActions<A, D, J, R, C> = Actions<A, Never, PoolSends<A, D, J, R, C>, Births<Proxy<C>>>;
322
323#[allow(
324    clippy::type_complexity,
325    reason = "the marker retains the complete pool topology signature"
326)]
327struct PoolKernel<A: Address, D, J, R, C>(PhantomData<fn(A, D, J, R, C)>);
328
329impl<A: Address, D, J, R, C> PoolKernel<A, D, J, R, C> {
330    const fn new() -> Self {
331        Self(PhantomData)
332    }
333}
334
335impl<A, D, J, R, C> Behavior for PoolKernel<A, D, J, R, C>
336where
337    A: Address,
338    A::Nonce: From<u64>,
339    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
340    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
341{
342    type Addr = A;
343    type Msg = PoolMessage<A, D, J, R>;
344    type Event = User<A, PoolMessage<A, D, J, R>>;
345    type Sends = KernelSends<A, D, J, R, C>;
346    type Ph = Never;
347    type Error = Infallible;
348    type Birth = Births<C>;
349
350    fn init(&mut self) -> crate::BehaviorActed<Self> {
351        Ok(Actions::cont())
352    }
353
354    fn transition(&mut self, _event: Self::Event) -> crate::BehaviorActed<Self> {
355        Ok(Actions::cont())
356    }
357}
358
359type PoolSupervisor<A, D, J, R, C> = Supervisor<PoolKernel<A, D, J, R, C>, C>;
360
361/// A fixed, homogeneous, bounded FIFO worker pool.
362///
363/// Each configured nonce names one stable supervised proxy. Jobs are assigned
364/// only after a successful worker-creation result makes that slot idle. The
365/// retained state records an assignment before the corresponding delivery is
366/// returned, and a completion must carry the exact assignment token.
367///
368/// # Panics
369///
370/// Admission or retry propagates a panic from the application payload's
371/// `Clone` implementation before changing pool state. Dispatch panics at the
372/// physical assignment-counter boundary before committing its dispatch plan;
373/// the executor's poison-before-step contract makes that actor turn terminal
374/// rather than exposing partial successor state. The final counter value is
375/// deliberately reserved so every successful batch has a representable
376/// successor counter. This is a Bombay implementation boundary, not an actor
377/// model law.
378///
379/// A worker with any other message protocol cannot form a pool:
380///
381/// ```compile_fail
382/// use behavior::{Actions, Behavior, MailAddr, Never, NoBirths, PoolResponse, User, WorkerPool};
383///
384/// struct Reply;
385/// struct WrongWorker;
386/// impl Behavior for Reply {
387///     type Addr = MailAddr;
388///     type Msg = PoolResponse<String, (), MailAddr>;
389///     type Event = User<MailAddr, Self::Msg>;
390///     type Sends = Vec<Never>;
391///     type Ph = Never;
392///     type Error = Never;
393///     type Birth = NoBirths;
394///     fn init(&mut self) -> behavior::BehaviorActed<Self> { Ok(Actions::cont()) }
395///     fn transition(&mut self, _: Self::Event) -> behavior::BehaviorActed<Self> { Ok(Actions::cont()) }
396/// }
397/// impl Behavior for WrongWorker {
398///     type Addr = MailAddr;
399///     type Msg = u8;
400///     type Event = User<MailAddr, u8>;
401///     type Sends = Vec<behavior::Never>;
402///     type Ph = Never;
403///     type Error = Never;
404///     type Birth = NoBirths;
405///     fn init(&mut self) -> behavior::BehaviorActed<Self> { unimplemented!() }
406///     fn transition(&mut self, _: Self::Event) -> behavior::BehaviorActed<Self> { unimplemented!() }
407/// }
408///
409/// // `WrongWorker::Msg` is not `PoolAssignment<String>`.
410/// let _: Option<WorkerPool<MailAddr, Reply, String, (), WrongWorker>> = None;
411/// ```
412pub struct WorkerPool<A: Address, D, J, R, C>
413where
414    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
415    A::Nonce: From<u64>,
416    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
417{
418    supervisor: PoolSupervisor<A, D, J, R, C>,
419    slots: Vec<Slot<A, D, J, R>>,
420    backlog: VecDeque<QueuedJob<A, D, J, R>>,
421    backlog_capacity: usize,
422    next_assignment: u64,
423    interruption: InterruptionPolicy,
424}
425
426impl<A, D, J, R, C> WorkerPool<A, D, J, R, C>
427where
428    A: Address,
429    A::Nonce: From<u64>,
430    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
431    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
432{
433    /// Construct a pool after proving that every configured child route is
434    /// unique.
435    ///
436    /// # Errors
437    ///
438    /// Returns [`PoolConfigError::NoWorkers`] for an empty topology or
439    /// [`PoolConfigError::DuplicateWorker`] for the first repeated
440    /// creator-local nonce. No behavior or creation request is produced.
441    #[allow(
442        clippy::too_many_arguments,
443        reason = "the arguments expose the complete pool policy"
444    )]
445    pub fn new(
446        nonces: fn(usize) -> A::Nonce,
447        count: usize,
448        build: fn(usize) -> C,
449        backlog_capacity: usize,
450        interruption: InterruptionPolicy,
451        restart_policy: RestartPolicy,
452        max_restarts: u32,
453        restart_window: Duration,
454    ) -> Result<Self, PoolConfigError<A::Nonce>> {
455        if count == 0 {
456            return Err(PoolConfigError::NoWorkers);
457        }
458        let mut slots = Vec::with_capacity(count);
459        for index in 0..count {
460            let nonce = nonces(index);
461            if slots
462                .iter()
463                .any(|slot: &Slot<A, D, J, R>| slot.nonce == nonce)
464            {
465                return Err(PoolConfigError::DuplicateWorker(nonce));
466            }
467            slots.push(Slot {
468                nonce,
469                state: SlotState::Installing,
470            });
471        }
472        Ok(Self {
473            supervisor: Supervisor::new(
474                PoolKernel::new(),
475                nonces,
476                count,
477                build,
478                Strategy::OneForOne,
479                restart_policy,
480                max_restarts,
481                restart_window,
482            ),
483            slots,
484            backlog: VecDeque::new(),
485            backlog_capacity,
486            next_assignment: 0,
487            interruption,
488        })
489    }
490
491    #[must_use]
492    pub fn backlog_len(&self) -> usize {
493        self.backlog.len()
494    }
495
496    #[must_use]
497    pub fn worker_phase(&self, worker: A::Nonce) -> Option<WorkerPhase> {
498        self.slots
499            .iter()
500            .find(|slot| slot.nonce == worker)
501            .map(|slot| match &slot.state {
502                SlotState::Installing => WorkerPhase::Installing,
503                SlotState::Idle => WorkerPhase::Idle,
504                SlotState::Assigned { assignment, job } => WorkerPhase::Assigned {
505                    assignment: *assignment,
506                    job: job.id,
507                },
508                SlotState::Retired { reason } => WorkerPhase::Retired { reason: *reason },
509            })
510    }
511
512    fn slot_position(&self, worker: A::Nonce) -> Result<usize, PoolError<A::Nonce>> {
513        self.slots
514            .iter()
515            .position(|slot| slot.nonce == worker)
516            .ok_or(PoolError::UnknownWorker(worker))
517    }
518}
519
520impl<A, D, J, R, C> WorkerPool<A, D, J, R, C>
521where
522    A: Address,
523    A::Nonce: From<u64>,
524    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
525    J: Clone,
526    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
527{
528    fn supervisor_transition(
529        &mut self,
530        event: PoolEvent<A, D, J, R>,
531    ) -> PoolActions<A, D, J, R, C> {
532        match delegate_transition(&mut self.supervisor, event) {
533            Ok(actions) => actions,
534            Err(never) => match never {},
535        }
536    }
537
538    fn submit(
539        &mut self,
540        job: JobId,
541        payload: J,
542        reply_to: Recipient<D>,
543        actions: &mut PoolActions<A, D, J, R, C>,
544    ) {
545        let can_dispatch = self
546            .slots
547            .iter()
548            .any(|slot| matches!(slot.state, SlotState::Idle));
549        if !can_dispatch && self.backlog.len() == self.backlog_capacity {
550            actions
551                .sends
552                .behavior
553                .send::<Delivery<D>, Own>(Delivery::new(
554                    reply_to,
555                    PoolResponse::Rejected {
556                        job,
557                        payload,
558                        reason: PoolRejection::BacklogFull,
559                    },
560                ));
561            return;
562        }
563        let dispatch_payload = payload.clone();
564        self.backlog.push_back(QueuedJob {
565            accepted: AcceptedJob {
566                id: job,
567                payload,
568                reply_to,
569                interruption: None,
570                target: None,
571            },
572            dispatch_payload,
573        });
574        actions
575            .sends
576            .behavior
577            .send::<_, Own>(Delivery::new(reply_to, PoolResponse::Accepted { job }));
578    }
579
580    fn submit_to(
581        &mut self,
582        target: A::Nonce,
583        job: JobId,
584        payload: J,
585        reply_to: Recipient<D>,
586        actions: &mut PoolActions<A, D, J, R, C>,
587    ) -> Admission {
588        let Some(slot) = self.slots.iter().find(|slot| slot.nonce == target) else {
589            actions
590                .sends
591                .behavior
592                .send::<Delivery<D>, Own>(Delivery::new(
593                    reply_to,
594                    PoolResponse::Rejected {
595                        job,
596                        payload,
597                        reason: PoolRejection::AffinityUnavailable,
598                    },
599                ));
600            return Admission::Rejected;
601        };
602        if matches!(slot.state, SlotState::Retired { .. }) {
603            actions.sends.behavior.send::<_, Own>(Delivery::new(
604                reply_to,
605                PoolResponse::Rejected {
606                    job,
607                    payload,
608                    reason: PoolRejection::AffinityUnavailable,
609                },
610            ));
611            return Admission::Rejected;
612        }
613        let can_dispatch = matches!(slot.state, SlotState::Idle);
614        if !can_dispatch && self.backlog.len() == self.backlog_capacity {
615            actions.sends.behavior.send::<_, Own>(Delivery::new(
616                reply_to,
617                PoolResponse::Rejected {
618                    job,
619                    payload,
620                    reason: PoolRejection::BacklogFull,
621                },
622            ));
623            return Admission::Rejected;
624        }
625        let dispatch_payload = payload.clone();
626        self.backlog.push_back(QueuedJob {
627            accepted: AcceptedJob {
628                id: job,
629                payload,
630                reply_to,
631                interruption: None,
632                target: Some(target),
633            },
634            dispatch_payload,
635        });
636        actions
637            .sends
638            .behavior
639            .send::<_, Own>(Delivery::new(reply_to, PoolResponse::Accepted { job }));
640        Admission::Accepted
641    }
642
643    fn complete(
644        &mut self,
645        worker: A::Nonce,
646        assignment: AssignmentId,
647        result: R,
648        actions: &mut PoolActions<A, D, J, R, C>,
649    ) -> Result<(), PoolError<A::Nonce>> {
650        let position = self.slot_position(worker)?;
651        let phase = self
652            .worker_phase(worker)
653            .expect("position proves the slot exists");
654        let SlotState::Assigned {
655            assignment: expected,
656            ..
657        } = &self.slots[position].state
658        else {
659            return Err(PoolError::CompletionForUnavailableWorker { worker, phase });
660        };
661        if *expected != assignment {
662            return Err(PoolError::StaleCompletion {
663                worker,
664                expected: *expected,
665                received: assignment,
666            });
667        }
668        let SlotState::Assigned { job, .. } =
669            core::mem::replace(&mut self.slots[position].state, SlotState::Idle)
670        else {
671            unreachable!("the state was proven assigned")
672        };
673        actions.sends.behavior.send::<_, Own>(Delivery::new(
674            job.reply_to,
675            PoolResponse::Completed {
676                job: job.id,
677                result,
678            },
679        ));
680        Ok(())
681    }
682
683    fn worker_stopped(
684        &mut self,
685        stopped: &WorkerStopped<A>,
686        responses: &mut Vec<Delivery<D>>,
687    ) -> Result<(), PoolError<A::Nonce>> {
688        let position = self.slot_position(stopped.proxy)?;
689        let phase = self
690            .worker_phase(stopped.proxy)
691            .expect("position proves the slot exists");
692        if matches!(phase, WorkerPhase::Installing | WorkerPhase::Retired { .. }) {
693            return Err(PoolError::WorkerStoppedWhileUnavailable {
694                worker: stopped.proxy,
695                phase,
696            });
697        }
698        if self.interruption == InterruptionPolicy::Retry {
699            if let SlotState::Assigned { job, .. } = &self.slots[position].state {
700                let dispatch_payload = job.payload.clone();
701                let SlotState::Assigned { mut job, .. } =
702                    core::mem::replace(&mut self.slots[position].state, SlotState::Installing)
703                else {
704                    unreachable!("the assigned state was matched before committing retry")
705                };
706                job.interruption = Some(PoolInterruption::WorkerStopped {
707                    worker: stopped.proxy,
708                    outcome: stopped.outcome,
709                });
710                self.backlog.push_front(QueuedJob {
711                    accepted: job,
712                    dispatch_payload,
713                });
714            } else {
715                self.slots[position].state = SlotState::Installing;
716            }
717            return Ok(());
718        }
719
720        let previous = core::mem::replace(&mut self.slots[position].state, SlotState::Installing);
721        if let SlotState::Assigned { job, .. } = previous {
722            responses.push(Delivery::new(
723                job.reply_to,
724                PoolResponse::Interrupted {
725                    job: job.id,
726                    payload: job.payload,
727                    reason: PoolInterruption::WorkerStopped {
728                        worker: stopped.proxy,
729                        outcome: stopped.outcome,
730                    },
731                },
732            ));
733        }
734        Ok(())
735    }
736
737    fn fail_backlog_if_irrecoverable(&mut self, actions: &mut PoolActions<A, D, J, R, C>) {
738        if self
739            .slots
740            .iter()
741            .any(|slot| !matches!(slot.state, SlotState::Retired { .. }))
742        {
743            return;
744        }
745        for queued in self.backlog.drain(..) {
746            let job = queued.accepted;
747            actions.sends.behavior.send::<_, Own>(Delivery::new(
748                job.reply_to,
749                PoolResponse::Interrupted {
750                    job: job.id,
751                    payload: job.payload,
752                    reason: job
753                        .interruption
754                        .unwrap_or(PoolInterruption::NoRecoverableWorkers),
755                },
756            ));
757        }
758    }
759
760    fn fail_jobs_for_retired_slot(
761        &mut self,
762        worker: A::Nonce,
763        reason: WorkerRetirement,
764        actions: &mut PoolActions<A, D, J, R, C>,
765    ) {
766        let mut retained = VecDeque::with_capacity(self.backlog.len());
767        while let Some(queued) = self.backlog.pop_front() {
768            if queued.accepted.target == Some(worker) {
769                let job = queued.accepted;
770                actions.sends.behavior.send::<_, Own>(Delivery::new(
771                    job.reply_to,
772                    PoolResponse::Interrupted {
773                        job: job.id,
774                        payload: job.payload,
775                        reason: job
776                            .interruption
777                            .unwrap_or(PoolInterruption::AffinityRetired { worker, reason }),
778                    },
779                ));
780            } else {
781                retained.push_back(queued);
782            }
783        }
784        self.backlog = retained;
785    }
786
787    fn creation_resolved(
788        &mut self,
789        resolved: &WorkerCreationResolved<A::Nonce>,
790    ) -> Result<(), PoolError<A::Nonce>> {
791        let position = self.slot_position(resolved.proxy)?;
792        let phase = self
793            .worker_phase(resolved.proxy)
794            .expect("position proves the slot exists");
795        if !matches!(phase, WorkerPhase::Installing) {
796            return Err(PoolError::CreationResolvedWhileUnavailable {
797                worker: resolved.proxy,
798                phase,
799            });
800        }
801        self.slots[position].state = match resolved.result {
802            Ok(()) => SlotState::Idle,
803            Err(rejection) => SlotState::Retired {
804                reason: WorkerRetirement::CreationRejected(rejection),
805            },
806        };
807        Ok(())
808    }
809
810    fn dispatch(&mut self, actions: &mut PoolActions<A, D, J, R, C>) {
811        let mut selected_jobs = Vec::new();
812        let mut plan = Vec::new();
813        for (slot_position, slot) in self.slots.iter().enumerate() {
814            if !matches!(slot.state, SlotState::Idle) {
815                continue;
816            }
817            let Some(job_position) = self.backlog.iter().enumerate().find_map(|(position, job)| {
818                (!selected_jobs.contains(&position)
819                    && job
820                        .accepted
821                        .target
822                        .is_none_or(|target| target == slot.nonce))
823                .then_some(position)
824            }) else {
825                continue;
826            };
827            selected_jobs.push(job_position);
828            plan.push(PlannedDispatch {
829                slot_position,
830                job_position,
831            });
832        }
833        let count =
834            u64::try_from(plan.len()).expect("a pool cannot contain more than u64::MAX slots");
835        let next_assignment = self
836            .next_assignment
837            .checked_add(count)
838            .expect("pool assignment identifiers exhausted");
839
840        let mut selected_by_position = BTreeMap::new();
841        for planned in plan {
842            selected_by_position.insert(planned.job_position, planned.slot_position);
843        }
844        let mut selected_by_slot: Vec<Option<QueuedJob<A, D, J, R>>> =
845            std::iter::repeat_with(|| None)
846                .take(self.slots.len())
847                .collect();
848        let mut remaining = VecDeque::new();
849        for (position, queued) in self.backlog.drain(..).enumerate() {
850            if let Some(slot_position) = selected_by_position.remove(&position) {
851                selected_by_slot[slot_position] = Some(queued);
852            } else {
853                remaining.push_back(queued);
854            }
855        }
856        self.backlog = remaining;
857
858        for (offset, (slot_position, queued)) in selected_by_slot
859            .into_iter()
860            .enumerate()
861            .filter_map(|(slot_position, queued)| queued.map(|queued| (slot_position, queued)))
862            .enumerate()
863        {
864            let payload = queued.dispatch_payload;
865            let job = queued.accepted;
866            let assignment = AssignmentId(
867                self.next_assignment
868                    + u64::try_from(offset).expect("offset is bounded by the checked plan length"),
869            );
870            let nonce = self.slots[slot_position].nonce;
871            let job_id = job.id;
872            self.slots[slot_position].state = SlotState::Assigned { assignment, job };
873            actions
874                .sends
875                .behavior
876                .send::<Delivery<Proxy<C>>, Own>(Delivery::new(
877                    Recipient::child(nonce),
878                    ProxyCommand::Forward(PoolAssignment {
879                        assignment,
880                        job: job_id,
881                        payload,
882                    }),
883                ));
884        }
885        self.next_assignment = next_assignment;
886    }
887}
888
889impl<A, D, J, R, C> Behavior for WorkerPool<A, D, J, R, C>
890where
891    A: Address,
892    A::Nonce: From<u64>,
893    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
894    J: Clone,
895    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
896{
897    type Addr = A;
898    type Msg = PoolMessage<A, D, J, R>;
899    type Event = PoolEvent<A, D, J, R>;
900    type Sends = PoolSends<A, D, J, R, C>;
901    type Ph = Never;
902    type Error = PoolError<A::Nonce>;
903    type Birth = Births<Proxy<C>>;
904
905    fn init(&mut self) -> crate::BehaviorActed<Self> {
906        match self.supervisor.init() {
907            Ok(actions) => Ok(actions),
908            Err(never) => match never {},
909        }
910    }
911
912    fn transition(&mut self, event: Self::Event) -> crate::BehaviorActed<Self> {
913        match event {
914            SupervisionEvent::Inner(User {
915                message:
916                    PoolMessage::Submit {
917                        job,
918                        payload,
919                        reply_to,
920                    },
921                ..
922            }) => {
923                let mut actions = Actions::cont();
924                self.submit(job, payload, reply_to, &mut actions);
925                self.dispatch(&mut actions);
926                Ok(actions)
927            }
928            SupervisionEvent::Inner(User {
929                message:
930                    PoolMessage::Completed {
931                        worker,
932                        assignment,
933                        result,
934                    },
935                ..
936            }) => {
937                let mut actions = Actions::cont();
938                self.complete(worker, assignment, result, &mut actions)?;
939                self.dispatch(&mut actions);
940                Ok(actions)
941            }
942            SupervisionEvent::WorkerStopped(stopped) => {
943                let proxy = stopped.proxy;
944                let mut responses = Vec::new();
945                self.worker_stopped(&stopped, &mut responses)?;
946                let mut actions =
947                    self.supervisor_transition(SupervisionEvent::WorkerStopped(stopped));
948                actions.sends.behavior.responses.extend(responses);
949                let replacement_requested = actions
950                    .sends
951                    .replacement_commands
952                    .iter()
953                    .any(|delivery| delivery.to.is_child(proxy));
954                if !replacement_requested {
955                    let position = self.slot_position(proxy)?;
956                    let reason = WorkerRetirement::ReplacementUnavailable;
957                    self.slots[position].state = SlotState::Retired { reason };
958                    self.fail_jobs_for_retired_slot(proxy, reason, &mut actions);
959                }
960                self.dispatch(&mut actions);
961                self.fail_backlog_if_irrecoverable(&mut actions);
962                Ok(actions)
963            }
964            SupervisionEvent::WorkerCreationResolved(resolved) => {
965                let proxy = resolved.proxy;
966                self.creation_resolved(&resolved)?;
967                let mut actions =
968                    self.supervisor_transition(SupervisionEvent::WorkerCreationResolved(resolved));
969                if let Some(WorkerPhase::Retired { reason }) = self.worker_phase(proxy) {
970                    self.fail_jobs_for_retired_slot(proxy, reason, &mut actions);
971                }
972                self.dispatch(&mut actions);
973                self.fail_backlog_if_irrecoverable(&mut actions);
974                Ok(actions)
975            }
976            SupervisionEvent::ChildStopped(stopped) => {
977                Ok(self.supervisor_transition(SupervisionEvent::ChildStopped(stopped)))
978            }
979            SupervisionEvent::CreationResolved(resolved) => {
980                Ok(self.supervisor_transition(SupervisionEvent::CreationResolved(resolved)))
981            }
982        }
983    }
984}
985
986#[cfg(test)]
987#[allow(
988    clippy::items_after_test_module,
989    reason = "the local fixed-pool regression sits beside that implementation"
990)]
991mod tests {
992    use super::*;
993    use crate::{MailAddr, NoBirths};
994
995    struct TestReply;
996
997    impl Behavior for TestReply {
998        type Addr = MailAddr;
999        type Msg = PoolResponse<u8, (), MailAddr>;
1000        type Event = User<MailAddr, Self::Msg>;
1001        type Sends = Vec<Never>;
1002        type Ph = Never;
1003        type Error = Never;
1004        type Birth = NoBirths;
1005
1006        fn init(&mut self) -> crate::BehaviorActed<Self> {
1007            Ok(Actions::cont())
1008        }
1009
1010        fn transition(&mut self, _: Self::Event) -> crate::BehaviorActed<Self> {
1011            Ok(Actions::cont())
1012        }
1013    }
1014
1015    #[derive(Clone, Copy)]
1016    struct TestWorker;
1017
1018    impl Behavior for TestWorker {
1019        type Addr = MailAddr;
1020        type Msg = PoolAssignment<u8>;
1021        type Event = User<MailAddr, PoolAssignment<u8>>;
1022        type Sends = Vec<Never>;
1023        type Ph = Never;
1024        type Error = Never;
1025        type Birth = NoBirths;
1026
1027        fn init(&mut self) -> crate::BehaviorActed<Self> {
1028            Ok(Actions::cont())
1029        }
1030
1031        fn transition(&mut self, _: Self::Event) -> crate::BehaviorActed<Self> {
1032            Ok(Actions::cont())
1033        }
1034    }
1035
1036    fn test_worker(_: usize) -> TestWorker {
1037        TestWorker
1038    }
1039
1040    #[test]
1041    fn one_dispatch_batch_preserves_fifo_jobs_across_index_removal() {
1042        let mut pool = WorkerPool::new(
1043            |index| u64::try_from(index).unwrap(),
1044            2,
1045            test_worker,
1046            3,
1047            InterruptionPolicy::Fail,
1048            RestartPolicy::Permanent,
1049            1,
1050            Duration::from_secs(1),
1051        )
1052        .unwrap();
1053        pool.init().unwrap();
1054
1055        for job in 1..=3 {
1056            pool.transition(SupervisionEvent::Inner(User::new(
1057                MailAddr(90),
1058                PoolMessage::Submit {
1059                    job: JobId(job),
1060                    payload: u8::try_from(job).unwrap(),
1061                    reply_to: Recipient::global(MailAddr(91)),
1062                },
1063            )))
1064            .unwrap();
1065        }
1066        pool.slots[0].state = SlotState::Idle;
1067        pool.slots[1].state = SlotState::Idle;
1068
1069        let mut actions: PoolActions<MailAddr, TestReply, u8, (), TestWorker> = Actions::cont();
1070        pool.dispatch(&mut actions);
1071
1072        let assignments = &actions.sends.behavior.assignments;
1073        assert_eq!(assignments.len(), 2);
1074        for (index, expected_job) in [JobId(1), JobId(2)].into_iter().enumerate() {
1075            assert!(
1076                assignments[index]
1077                    .to
1078                    .is_child(u64::try_from(index).unwrap())
1079            );
1080            let ProxyCommand::Forward(assignment) = &assignments[index].message else {
1081                panic!("pool dispatches with Forward");
1082            };
1083            assert_eq!(
1084                assignment.assignment,
1085                AssignmentId(u64::try_from(index).unwrap())
1086            );
1087            assert_eq!(assignment.job, expected_job);
1088        }
1089        assert_eq!(pool.backlog.len(), 1);
1090        assert_eq!(pool.backlog[0].accepted.id, JobId(3));
1091    }
1092}
1093
1094/// A worker pool whose admitted keys remain bound to stable worker slots.
1095///
1096/// The selector chooses a stable proxy nonce only when a key is first
1097/// admitted. Replacement incarnations remain behind that proxy, so they do
1098/// not alter affinity. [`KeyedPoolMessage::Rebalance`] is the sole transition
1099/// that changes an established binding, and jobs accepted before it retain
1100/// their original target.
1101///
1102/// Keys must have a concrete equality relation; a key type without `Eq` cannot
1103/// form an affinity table:
1104///
1105/// ```compile_fail
1106/// use behavior::{Actions, Behavior, KeyedWorkerPool, MailAddr, Never, NoBirths, PoolResponse, User};
1107/// struct NonKey(f64);
1108/// struct Reply;
1109/// struct Worker;
1110/// impl Behavior for Reply {
1111///     type Addr = MailAddr;
1112///     type Msg = PoolResponse<u8, (), MailAddr>;
1113///     type Event = User<MailAddr, Self::Msg>;
1114///     type Sends = Vec<Never>;
1115///     type Ph = Never;
1116///     type Error = Never;
1117///     type Birth = NoBirths;
1118///     fn init(&mut self) -> behavior::BehaviorActed<Self> { Ok(Actions::cont()) }
1119///     fn transition(&mut self, _: Self::Event) -> behavior::BehaviorActed<Self> { Ok(Actions::cont()) }
1120/// }
1121/// #[behavior::behavior(
1122///     addr = MailAddr,
1123///     message = behavior::PoolAssignment<u8>,
1124///     sends = Vec<Never>,
1125///     births = NoBirths,
1126///     error = Never,
1127/// )]
1128/// impl Worker {
1129///     fn init(&mut self) -> behavior::Acted<MailAddr, Never, Vec<Never>, NoBirths, Never> {
1130///         Ok(Actions::cont())
1131///     }
1132///     fn receive(&mut self, _: MailAddr, _: behavior::PoolAssignment<u8>) -> behavior::Acted<MailAddr, Never, Vec<Never>, NoBirths, Never> {
1133///         Ok(Actions::cont())
1134///     }
1135/// }
1136/// let _: Option<KeyedWorkerPool<MailAddr, Reply, NonKey, u8, (), Worker, fn(&NonKey) -> u64>> = None;
1137/// ```
1138pub struct KeyedWorkerPool<A: Address, D, K, J, R, C, S>
1139where
1140    A::Nonce: From<u64>,
1141    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
1142    K: Eq,
1143    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
1144    S: AffinitySelector<K, A::Nonce>,
1145{
1146    pool: WorkerPool<A, D, J, R, C>,
1147    bindings: Vec<(K, A::Nonce)>,
1148    selector: S,
1149}
1150
1151impl<A, D, K, J, R, C, S> KeyedWorkerPool<A, D, K, J, R, C, S>
1152where
1153    A: Address,
1154    A::Nonce: From<u64>,
1155    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
1156    K: Eq,
1157    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
1158    S: AffinitySelector<K, A::Nonce>,
1159{
1160    /// Construct a key-persistent pool over the same fixed supervised slots as
1161    /// [`WorkerPool`]. The selector is pure and is consulted once per
1162    /// previously unseen key. It chooses behavior policy; runtime route
1163    /// resolution remains outside this type.
1164    ///
1165    /// # Errors
1166    ///
1167    /// Returns [`PoolConfigError::NoWorkers`] for an empty topology or
1168    /// [`PoolConfigError::DuplicateWorker`] for a repeated stable nonce.
1169    #[allow(
1170        clippy::too_many_arguments,
1171        reason = "the arguments expose the complete pool and affinity policy"
1172    )]
1173    pub fn new(
1174        nonces: fn(usize) -> A::Nonce,
1175        count: usize,
1176        build: fn(usize) -> C,
1177        backlog_capacity: usize,
1178        interruption: InterruptionPolicy,
1179        restart_policy: RestartPolicy,
1180        max_restarts: u32,
1181        restart_window: Duration,
1182        selector: S,
1183    ) -> Result<Self, PoolConfigError<A::Nonce>> {
1184        Ok(Self {
1185            pool: WorkerPool::new(
1186                nonces,
1187                count,
1188                build,
1189                backlog_capacity,
1190                interruption,
1191                restart_policy,
1192                max_restarts,
1193                restart_window,
1194            )?,
1195            bindings: Vec::new(),
1196            selector,
1197        })
1198    }
1199
1200    /// Return the stable slot currently bound to `key`.
1201    #[must_use]
1202    pub fn affinity(&self, key: &K) -> Option<A::Nonce> {
1203        self.bindings
1204            .iter()
1205            .find_map(|(bound, worker)| (bound == key).then_some(*worker))
1206    }
1207
1208    #[must_use]
1209    pub fn backlog_len(&self) -> usize {
1210        self.pool.backlog_len()
1211    }
1212
1213    #[must_use]
1214    pub fn worker_phase(&self, worker: A::Nonce) -> Option<WorkerPhase> {
1215        self.pool.worker_phase(worker)
1216    }
1217
1218    fn rebalance(&mut self, key: K, worker: A::Nonce) -> Result<(), PoolError<A::Nonce>> {
1219        let position = self.pool.slot_position(worker)?;
1220        if let SlotState::Retired { reason } = self.pool.slots[position].state {
1221            return Err(PoolError::RebalanceToRetiredWorker { worker, reason });
1222        }
1223        if let Some((_, bound)) = self.bindings.iter_mut().find(|(bound, _)| *bound == key) {
1224            *bound = worker;
1225        } else {
1226            self.bindings.push((key, worker));
1227        }
1228        Ok(())
1229    }
1230}
1231
1232impl<A, D, K, J, R, C, S> Behavior for KeyedWorkerPool<A, D, K, J, R, C, S>
1233where
1234    A: Address,
1235    A::Nonce: From<u64>,
1236    D: Behavior<Addr = A, Msg = PoolResponse<J, R, A>>,
1237    K: Eq,
1238    J: Clone,
1239    C: Behavior<Addr = A, Msg = PoolAssignment<J>, Ph = Never>,
1240    S: AffinitySelector<K, A::Nonce>,
1241{
1242    type Addr = A;
1243    type Msg = KeyedPoolMessage<A, D, K, J, R>;
1244    type Event = KeyedPoolEvent<A, D, K, J, R>;
1245    type Sends = PoolSends<A, D, J, R, C>;
1246    type Ph = Never;
1247    type Error = PoolError<A::Nonce>;
1248    type Birth = Births<Proxy<C>>;
1249
1250    fn init(&mut self) -> crate::BehaviorActed<Self> {
1251        self.pool.init()
1252    }
1253
1254    fn transition(&mut self, event: Self::Event) -> crate::BehaviorActed<Self> {
1255        match event {
1256            SupervisionEvent::Inner(User {
1257                message:
1258                    KeyedPoolMessage::Submit {
1259                        key,
1260                        job,
1261                        payload,
1262                        reply_to,
1263                    },
1264                ..
1265            }) => {
1266                let existing = self.affinity(&key);
1267                let target = existing.unwrap_or_else(|| self.selector.select(&key));
1268                let mut actions = Actions::cont();
1269                let admission = self
1270                    .pool
1271                    .submit_to(target, job, payload, reply_to, &mut actions);
1272                match (admission, existing) {
1273                    (Admission::Accepted, None) => self.bindings.push((key, target)),
1274                    (Admission::Accepted | Admission::Rejected, Some(_))
1275                    | (Admission::Rejected, None) => {}
1276                }
1277                self.pool.dispatch(&mut actions);
1278                Ok(actions)
1279            }
1280            SupervisionEvent::Inner(User {
1281                message:
1282                    KeyedPoolMessage::Completed {
1283                        worker,
1284                        assignment,
1285                        result,
1286                    },
1287                ..
1288            }) => {
1289                let mut actions = Actions::cont();
1290                self.pool
1291                    .complete(worker, assignment, result, &mut actions)?;
1292                self.pool.dispatch(&mut actions);
1293                Ok(actions)
1294            }
1295            SupervisionEvent::Inner(User {
1296                message: KeyedPoolMessage::Rebalance { key, worker },
1297                ..
1298            }) => {
1299                self.rebalance(key, worker)?;
1300                Ok(Actions::cont())
1301            }
1302            SupervisionEvent::WorkerStopped(stopped) => self
1303                .pool
1304                .transition(SupervisionEvent::WorkerStopped(stopped)),
1305            SupervisionEvent::WorkerCreationResolved(resolved) => self
1306                .pool
1307                .transition(SupervisionEvent::WorkerCreationResolved(resolved)),
1308            SupervisionEvent::ChildStopped(stopped) => self
1309                .pool
1310                .transition(SupervisionEvent::ChildStopped(stopped)),
1311            SupervisionEvent::CreationResolved(resolved) => self
1312                .pool
1313                .transition(SupervisionEvent::CreationResolved(resolved)),
1314        }
1315    }
1316}
behavior/pool.rs

behavior/
pool.rs
