solverforge_solver/manager/solver_manager/
manager.rs

1use std::marker::PhantomData;
2use std::panic::AssertUnwindSafe;
3use std::sync::atomic::Ordering;
4
5use solverforge_core::domain::PlanningSolution;
6use solverforge_core::score::Score;
7use tokio::sync::mpsc;
8
9use crate::stats::QualifiedCandidateTraceRunProvenance;
10
11use super::super::solution_manager::Analyzable;
12use super::runtime::{panic_payload_to_string, SolverRuntime};
13#[cfg(test)]
14use super::slot::SLOT_FREE;
15use super::slot::{JobSlot, SLOT_SOLVING};
16use super::types::{
17    SolverEvent, SolverLifecycleState, SolverManagerError, SolverSnapshot, SolverSnapshotAnalysis,
18    SolverStatus, SolverTelemetryDetail,
19};
20
21/// Maximum concurrent jobs per SolverManager instance.
22pub const MAX_JOBS: usize = 16;
23
24/// Trait for solutions that can run inside the retained lifecycle manager.
25pub trait Solvable: PlanningSolution + Send + 'static {
26    fn solve(
27        self,
28        runtime: SolverRuntime<Self>,
29        qualified_candidate_trace_provenance: Option<QualifiedCandidateTraceRunProvenance>,
30    );
31}
32
33/// Manages retained async solve jobs with lifecycle-complete event streaming.
34pub struct SolverManager<S: Solvable> {
35    slots: [JobSlot<S>; MAX_JOBS],
36    _phantom: PhantomData<fn() -> S>,
37}
38
39impl<S: Solvable> Default for SolverManager<S> {
40    fn default() -> Self {
41        Self::new()
42    }
43}
44
45impl<S: Solvable> SolverManager<S>
46where
47    S::Score: Score,
48{
49    pub const fn new() -> Self {
50        Self {
51            slots: [
52                JobSlot::new(),
53                JobSlot::new(),
54                JobSlot::new(),
55                JobSlot::new(),
56                JobSlot::new(),
57                JobSlot::new(),
58                JobSlot::new(),
59                JobSlot::new(),
60                JobSlot::new(),
61                JobSlot::new(),
62                JobSlot::new(),
63                JobSlot::new(),
64                JobSlot::new(),
65                JobSlot::new(),
66                JobSlot::new(),
67                JobSlot::new(),
68            ],
69            _phantom: PhantomData,
70        }
71    }
72
73    pub fn solve(
74        &'static self,
75        solution: S,
76    ) -> Result<(usize, mpsc::UnboundedReceiver<SolverEvent<S>>), SolverManagerError> {
77        self.submit(solution, None)
78    }
79
80    /// Submits a retained solve with externally validated candidate-trace provenance.
81    ///
82    /// This changes only the diagnostic trace header. Job allocation, worker
83    /// execution, event publication, snapshots, and terminalization use the
84    /// same retained lifecycle as [`Self::solve`].
85    pub fn solve_with_qualified_candidate_trace_provenance(
86        &'static self,
87        solution: S,
88        provenance: QualifiedCandidateTraceRunProvenance,
89    ) -> Result<(usize, mpsc::UnboundedReceiver<SolverEvent<S>>), SolverManagerError> {
90        self.submit(solution, Some(provenance))
91    }
92
93    fn submit(
94        &'static self,
95        solution: S,
96        qualified_candidate_trace_provenance: Option<QualifiedCandidateTraceRunProvenance>,
97    ) -> Result<(usize, mpsc::UnboundedReceiver<SolverEvent<S>>), SolverManagerError> {
98        let (sender, receiver) = mpsc::unbounded_channel();
99
100        let Some(slot_idx) = self
101            .slots
102            .iter()
103            .position(|slot| slot.try_initialize(sender.clone()))
104        else {
105            return Err(SolverManagerError::NoFreeJobSlots);
106        };
107
108        let slot = &self.slots[slot_idx];
109        let runtime = SolverRuntime::new(slot_idx, slot);
110
111        rayon::spawn(move || {
112            let result = std::panic::catch_unwind(AssertUnwindSafe(|| {
113                solution.solve(runtime, qualified_candidate_trace_provenance)
114            }));
115
116            match result {
117                Ok(()) => {
118                    if !runtime.is_terminal() {
119                        if runtime.is_cancel_requested() {
120                            let (current_score, best_score, telemetry) =
121                                runtime.slot.with_publication(|_, record| {
122                                    (
123                                        record.current_score,
124                                        record.best_score,
125                                        record.telemetry.clone(),
126                                    )
127                                });
128                            runtime.emit_cancelled(current_score, best_score, telemetry);
129                        } else {
130                            runtime.emit_failed(
131                                "solver returned without emitting a terminal lifecycle event"
132                                    .to_string(),
133                            );
134                        }
135                    }
136                }
137                Err(payload) => {
138                    runtime.emit_failed(panic_payload_to_string(payload));
139                }
140            }
141
142            runtime.slot.worker_exited();
143        });
144
145        Ok((slot_idx, receiver))
146    }
147
148    pub fn get_status(&self, job_id: usize) -> Result<SolverStatus<S::Score>, SolverManagerError> {
149        let slot = self.slot(job_id)?;
150        slot.with_publication(|_, record| {
151            let state = slot
152                .public_state()
153                .ok_or(SolverManagerError::JobNotFound { job_id })?;
154            Ok(record.status(job_id, state))
155        })
156    }
157
158    /// Returns compact aggregate telemetry and bounded candidate-pull detail
159    /// from one retained publication instant. The embedded status's scores and
160    /// latest snapshot revision identify that same instant.
161    ///
162    /// This is deliberately separate from [`Self::get_status`], lifecycle
163    /// events, and snapshots: those control-plane payloads remain compact as
164    /// a solve publishes progress. Detail is returned only when it belongs to
165    /// the current compact telemetry publication; it is cleared on every
166    /// later trace-free progress, resume, lifecycle-only, or terminal update.
167    pub fn get_telemetry_detail(
168        &self,
169        job_id: usize,
170    ) -> Result<SolverTelemetryDetail<S::Score>, SolverManagerError> {
171        let slot = self.slot(job_id)?;
172        slot.with_publication(|_, record| {
173            let state = slot
174                .public_state()
175                .ok_or(SolverManagerError::JobNotFound { job_id })?;
176            Ok(SolverTelemetryDetail {
177                status: record.status(job_id, state),
178                candidate_trace: record.candidate_trace_detail.clone(),
179            })
180        })
181    }
182
183    pub fn pause(&self, job_id: usize) -> Result<(), SolverManagerError> {
184        let slot = self.slot(job_id)?;
185        let paused = slot.with_publication(|sender, record| {
186            match slot.state.compare_exchange(
187                SLOT_SOLVING,
188                super::slot::SLOT_PAUSE_REQUESTED,
189                Ordering::SeqCst,
190                Ordering::SeqCst,
191            ) {
192                Ok(_) => {
193                    slot.pause_requested.store(true, Ordering::SeqCst);
194                    record.clear_candidate_trace_detail();
195                    let metadata =
196                        record.next_metadata(job_id, SolverLifecycleState::PauseRequested, None);
197                    if let Some(sender) = sender {
198                        let _ = sender.send(SolverEvent::PauseRequested { metadata });
199                    }
200                    true
201                }
202                Err(_) => false,
203            }
204        });
205        if paused {
206            Ok(())
207        } else {
208            let state = slot
209                .public_state()
210                .ok_or(SolverManagerError::JobNotFound { job_id })?;
211            Err(SolverManagerError::InvalidStateTransition {
212                job_id,
213                action: "pause",
214                state,
215            })
216        }
217    }
218
219    pub fn resume(&self, job_id: usize) -> Result<(), SolverManagerError> {
220        let slot = self.slot(job_id)?;
221        slot.with_publication(|_, _record| {
222            let state = slot
223                .public_state()
224                .ok_or(SolverManagerError::JobNotFound { job_id })?;
225            if state != SolverLifecycleState::Paused {
226                return Err(SolverManagerError::InvalidStateTransition {
227                    job_id,
228                    action: "resume",
229                    state,
230                });
231            }
232            Ok(())
233        })?;
234        slot.clear_pause_and_wake();
235        Ok(())
236    }
237
238    pub fn cancel(&self, job_id: usize) -> Result<(), SolverManagerError> {
239        let slot = self.slot(job_id)?;
240        slot.with_publication(|_, _record| {
241            let state = slot
242                .public_state()
243                .ok_or(SolverManagerError::JobNotFound { job_id })?;
244            if !matches!(
245                state,
246                SolverLifecycleState::Solving
247                    | SolverLifecycleState::PauseRequested
248                    | SolverLifecycleState::Paused
249            ) {
250                return Err(SolverManagerError::InvalidStateTransition {
251                    job_id,
252                    action: "cancel",
253                    state,
254                });
255            }
256
257            slot.terminate.store(true, Ordering::SeqCst);
258            Ok(())
259        })?;
260        slot.clear_pause_and_wake();
261        Ok(())
262    }
263
264    pub fn delete(&self, job_id: usize) -> Result<(), SolverManagerError> {
265        let slot = self.slot(job_id)?;
266        match slot.delete_terminal() {
267            Ok(()) => {}
268            Err(None) => return Err(SolverManagerError::JobNotFound { job_id }),
269            Err(Some(state)) => {
270                return Err(SolverManagerError::InvalidStateTransition {
271                    job_id,
272                    action: "delete",
273                    state,
274                });
275            }
276        }
277        slot.try_reset_deleted();
278        Ok(())
279    }
280
281    pub fn get_snapshot(
282        &self,
283        job_id: usize,
284        snapshot_revision: Option<u64>,
285    ) -> Result<SolverSnapshot<S>, SolverManagerError> {
286        let slot = self.slot(job_id)?;
287        slot.with_publication(|_, record| {
288            if slot.public_state().is_none() {
289                return Err(SolverManagerError::JobNotFound { job_id });
290            }
291
292            if record.snapshots.is_empty() {
293                return Err(SolverManagerError::NoSnapshotAvailable { job_id });
294            }
295
296            match snapshot_revision {
297                Some(revision) => record
298                    .snapshots
299                    .iter()
300                    .find(|snapshot| snapshot.snapshot_revision == revision)
301                    .cloned()
302                    .ok_or(SolverManagerError::SnapshotNotFound {
303                        job_id,
304                        snapshot_revision: revision,
305                    }),
306                None => Ok(record
307                    .snapshots
308                    .last()
309                    .expect("checked non-empty snapshots")
310                    .clone()),
311            }
312        })
313    }
314
315    pub fn analyze_snapshot(
316        &self,
317        job_id: usize,
318        snapshot_revision: Option<u64>,
319    ) -> Result<SolverSnapshotAnalysis<S::Score>, SolverManagerError>
320    where
321        S: Analyzable,
322    {
323        let snapshot = self.get_snapshot(job_id, snapshot_revision)?;
324        Ok(SolverSnapshotAnalysis {
325            job_id,
326            lifecycle_state: snapshot.lifecycle_state,
327            terminal_reason: snapshot.terminal_reason,
328            snapshot_revision: snapshot.snapshot_revision,
329            analysis: snapshot.solution.analyze(),
330        })
331    }
332
333    pub fn active_job_count(&self) -> usize {
334        self.slots
335            .iter()
336            .filter(|slot| slot.public_state().is_some())
337            .count()
338    }
339
340    #[cfg(test)]
341    pub(crate) fn slot_is_free_for_test(&self, job_id: usize) -> bool {
342        self.slots
343            .get(job_id)
344            .is_some_and(|slot| slot.state.load(Ordering::Acquire) == SLOT_FREE)
345    }
346
347    fn slot(&self, job_id: usize) -> Result<&JobSlot<S>, SolverManagerError> {
348        self.slots
349            .get(job_id)
350            .ok_or(SolverManagerError::JobNotFound { job_id })
351    }
352}
solverforge_solver/manager/solver_manager/manager.rs

solverforge_solver/manager/solver_manager/
manager.rs
