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            slot.pause_requested.store(false, Ordering::SeqCst);
233            Ok(())
234        })?;
235        slot.pause_condvar.notify_one();
236        Ok(())
237    }
238
239    pub fn cancel(&self, job_id: usize) -> Result<(), SolverManagerError> {
240        let slot = self.slot(job_id)?;
241        slot.with_publication(|_, _record| {
242            let state = slot
243                .public_state()
244                .ok_or(SolverManagerError::JobNotFound { job_id })?;
245            if !matches!(
246                state,
247                SolverLifecycleState::Solving
248                    | SolverLifecycleState::PauseRequested
249                    | SolverLifecycleState::Paused
250            ) {
251                return Err(SolverManagerError::InvalidStateTransition {
252                    job_id,
253                    action: "cancel",
254                    state,
255                });
256            }
257
258            slot.terminate.store(true, Ordering::SeqCst);
259            slot.pause_requested.store(false, Ordering::SeqCst);
260            Ok(())
261        })?;
262        slot.pause_condvar.notify_one();
263        Ok(())
264    }
265
266    pub fn delete(&self, job_id: usize) -> Result<(), SolverManagerError> {
267        let slot = self.slot(job_id)?;
268        match slot.delete_terminal() {
269            Ok(()) => {}
270            Err(None) => return Err(SolverManagerError::JobNotFound { job_id }),
271            Err(Some(state)) => {
272                return Err(SolverManagerError::InvalidStateTransition {
273                    job_id,
274                    action: "delete",
275                    state,
276                });
277            }
278        }
279        slot.try_reset_deleted();
280        Ok(())
281    }
282
283    pub fn get_snapshot(
284        &self,
285        job_id: usize,
286        snapshot_revision: Option<u64>,
287    ) -> Result<SolverSnapshot<S>, SolverManagerError> {
288        let slot = self.slot(job_id)?;
289        slot.with_publication(|_, record| {
290            if slot.public_state().is_none() {
291                return Err(SolverManagerError::JobNotFound { job_id });
292            }
293
294            if record.snapshots.is_empty() {
295                return Err(SolverManagerError::NoSnapshotAvailable { job_id });
296            }
297
298            match snapshot_revision {
299                Some(revision) => record
300                    .snapshots
301                    .iter()
302                    .find(|snapshot| snapshot.snapshot_revision == revision)
303                    .cloned()
304                    .ok_or(SolverManagerError::SnapshotNotFound {
305                        job_id,
306                        snapshot_revision: revision,
307                    }),
308                None => Ok(record
309                    .snapshots
310                    .last()
311                    .expect("checked non-empty snapshots")
312                    .clone()),
313            }
314        })
315    }
316
317    pub fn analyze_snapshot(
318        &self,
319        job_id: usize,
320        snapshot_revision: Option<u64>,
321    ) -> Result<SolverSnapshotAnalysis<S::Score>, SolverManagerError>
322    where
323        S: Analyzable,
324    {
325        let snapshot = self.get_snapshot(job_id, snapshot_revision)?;
326        Ok(SolverSnapshotAnalysis {
327            job_id,
328            lifecycle_state: snapshot.lifecycle_state,
329            terminal_reason: snapshot.terminal_reason,
330            snapshot_revision: snapshot.snapshot_revision,
331            analysis: snapshot.solution.analyze(),
332        })
333    }
334
335    pub fn active_job_count(&self) -> usize {
336        self.slots
337            .iter()
338            .filter(|slot| slot.public_state().is_some())
339            .count()
340    }
341
342    #[cfg(test)]
343    pub(crate) fn slot_is_free_for_test(&self, job_id: usize) -> bool {
344        self.slots
345            .get(job_id)
346            .is_some_and(|slot| slot.state.load(Ordering::Acquire) == SLOT_FREE)
347    }
348
349    fn slot(&self, job_id: usize) -> Result<&JobSlot<S>, SolverManagerError> {
350        self.slots
351            .get(job_id)
352            .ok_or(SolverManagerError::JobNotFound { job_id })
353    }
354}
solverforge_solver/manager/solver_manager/manager.rs

solverforge_solver/manager/solver_manager/
manager.rs
