fx_durable_ga/services/optimization/
service.rs

1use super::Error;
2use super::events::{
3    GenotypeEvaluatedEvent, GenotypeGenerated, OptimizationRequestedEvent, RequestCompletedEvent,
4    RequestInterruptedEvent, RequestTerminatedEvent,
5};
6use crate::models::{
7    Breeder, Crossover, Distribution, Encodeable, Fitness, FitnessGoal, Genotype, Mutagen, Request,
8    RequestConclusion, Schedule, ScheduleDecision, Selector, Terminated,
9};
10use crate::repositories::chainable::{Chain, FromTx, ToTx};
11use crate::repositories::{genotypes, morphologies, requests};
12use crate::services::lock;
13use crate::services::optimization::models::{Terminator, TypeErasedEvaluator};
14use fx_event_bus::Publisher;
15use serde::Serialize;
16use std::collections::{HashMap, HashSet};
17use tracing::instrument;
18use uuid::Uuid;
19
20/// Genetic algorithm optimization service that manages the entire optimization lifecycle.
21/// Handles request creation, population generation, genotype evaluation, and breeding.
22pub struct Service {
23    pub(super) locking: lock::Service,
24    pub(super) requests: requests::Repository,
25    pub(super) morphologies: morphologies::Repository,
26    pub(super) genotypes: genotypes::Repository,
27    pub(super) evaluators: HashMap<i32, Box<dyn TypeErasedEvaluator + 'static>>,
28    pub(super) max_deduplication_attempts: i32,
29}
30
31impl Service {
32    /// Creates a new service builder with the required dependencies.
33    #[instrument(level = "debug", skip_all)]
34    pub(crate) fn builder(
35        locking: lock::Service,
36        requests: requests::Repository,
37        morphologies: morphologies::Repository,
38        genotypes: genotypes::Repository,
39    ) -> super::ServiceBuilder {
40        super::ServiceBuilder {
41            locking,
42            requests,
43            morphologies,
44            genotypes,
45            evaluators: HashMap::new(),
46            max_deduplication_attempts: 5, // Default value
47        }
48    }
49
50    /// Creates a new genetic algorithm optimization request.
51    ///
52    /// This is the main entry point for starting genetic algorithm optimization.
53    /// It configures and initiates an evolutionary process that will attempt
54    /// to find optimal solutions for your problem type.
55    ///
56    /// # Parameters
57    ///
58    /// * `type_name` - Human-readable name of the type being optimized (from `Encodeable::NAME`)
59    /// * `type_hash` - Unique hash identifying the type structure (from `Encodeable::HASH`)
60    /// * `goal` - Optimization objective and stopping criteria
61    /// * `schedule` - Controls generation timing and population lifecycle
62    /// * `selector` - Parent selection strategy for breeding operations
63    /// * `mutagen` - Mutation behavior including temperature and rate schedules
64    /// * `crossover` - Genetic recombination method for combining parents
65    /// * `distribution` - Initial population generation strategy
66    ///
67    /// # Optimization Goal
68    ///
69    /// The `goal` parameter defines when optimization should stop:
70    /// - `FitnessGoal::maximize(0.95)` - Stop when fitness reaches 95% or higher
71    /// - `FitnessGoal::minimize(0.1)` - Stop when fitness drops to 10% or lower
72    ///
73    /// # Population Management
74    ///
75    /// The `schedule` controls generation timing and resource allocation:
76    /// - `Schedule::generational(100, 50)` - 100 generations, 50 individuals each
77    /// - `Schedule::rolling(5000, 100, 10)` - 5000 total evaluations, 100 max population, breed 10 at a time
78    ///
79    /// # Parent Selection
80    ///
81    /// The `selector` determines how parents are chosen for breeding:
82    /// - `Selector::tournament(3, 100)` - Tournament selection (size 3) from 100 candidates
83    /// - `Selector::roulette(100)` - Fitness-proportionate selection from 100 candidates
84    ///
85    /// # Mutation Strategy
86    ///
87    /// The `mutagen` combines temperature (step size) and mutation rate (frequency):
88    /// - `Mutagen::constant(0.5, 0.3)` - Fixed 50% temperature, 30% mutation rate
89    /// - Adaptive strategies use `Temperature::linear()` and `MutationRate::exponential()`
90    ///
91    /// # Genetic Recombination
92    ///
93    /// The `crossover` method combines genetic material from parents:
94    /// - `Crossover::uniform(0.5)` - Each gene has 50% chance from first parent
95    /// - `Crossover::single_point()` - Cut genome at random point, swap tails
96    ///
97    /// # Initial Population
98    ///
99    /// The `distribution` strategy affects initial diversity and convergence:
100    /// - `Distribution::latin_hypercube(50)` - Structured sampling for better coverage
101    /// - `Distribution::random(50)` - Pure random sampling
102    ///
103    /// # Common Configurations
104    ///
105    /// ## Quick Convergence (Time-Constrained)
106    /// ```rust,no_run
107    /// use fx_durable_ga::models::*;
108    /// # use fx_durable_ga::optimization::Service;
109    /// # #[derive(Debug)] struct MyType;
110    /// # impl Encodeable for MyType {
111    /// #     const NAME: &'static str = "MyType";
112    /// #     type Phenotype = MyType;
113    /// #     fn morphology() -> Vec<GeneBounds> { vec![] }
114    /// #     fn encode(&self) -> Vec<i64> { vec![] }
115    /// #     fn decode(_: &[i64]) -> Self::Phenotype { MyType }
116    /// # }
117    /// # async fn example(service: &Service) -> Result<(), Box<dyn std::error::Error>> {
118    ///
119    /// service.new_optimization_request(
120    ///     MyType::NAME,
121    ///     MyType::HASH,
122    ///     FitnessGoal::maximize(0.90)?,                    // Modest target
123    ///     Schedule::generational(20, 30),                 // Small, fast generations
124    ///     Selector::tournament(5, 50)?,                    // Strong selection pressure
125    ///     Mutagen::new(
126    ///         Temperature::exponential(0.8, 0.1, 1.2, 3)?, // Rapid cooling
127    ///         MutationRate::exponential(0.6, 0.05, 1.1, 2)?
128    ///     ),
129    ///     Crossover::uniform(0.6)?,                       // High recombination
130    ///     Distribution::latin_hypercube(30),
131    ///     None::<()>,
132    /// ).await?;
133    /// # Ok(())
134    /// # }
135    /// ```
136    ///
137    /// ## Balanced Search (General Purpose)
138    /// ```rust,no_run
139    /// use fx_durable_ga::models::*;
140    /// # use fx_durable_ga::optimization::Service;
141    /// # #[derive(Debug)] struct MyType;
142    /// # impl Encodeable for MyType {
143    /// #     const NAME: &'static str = "MyType";
144    /// #     type Phenotype = MyType;
145    /// #     fn morphology() -> Vec<GeneBounds> { vec![] }
146    /// #     fn encode(&self) -> Vec<i64> { vec![] }
147    /// #     fn decode(_: &[i64]) -> Self::Phenotype { MyType }
148    /// # }
149    /// # async fn example(service: &Service) -> Result<(), Box<dyn std::error::Error>> {
150    ///
151    /// service.new_optimization_request(
152    ///     MyType::NAME,
153    ///     MyType::HASH,
154    ///     FitnessGoal::maximize(0.95)?,                    // High-quality target
155    ///     Schedule::generational(50, 50),                 // Moderate generations
156    ///     Selector::tournament(3, 100)?,                   // Balanced selection
157    ///     Mutagen::new(
158    ///         Temperature::linear(0.7, 0.2, 1.0)?,         // Gradual cooling
159    ///         MutationRate::linear(0.4, 0.1, 1.0)?
160    ///     ),
161    ///     Crossover::uniform(0.5)?,                       // Balanced recombination
162    ///     Distribution::latin_hypercube(50),
163    ///     None::<()>,
164    /// ).await?;
165    /// # Ok(())
166    /// # }
167    /// ```
168    ///
169    /// ## Thorough Exploration (Complex Problems)
170    /// ```rust,no_run
171    /// use fx_durable_ga::models::*;
172    /// # use fx_durable_ga::optimization::Service;
173    /// # #[derive(Debug)] struct MyType;
174    /// # impl Encodeable for MyType {
175    /// #     const NAME: &'static str = "MyType";
176    /// #     type Phenotype = MyType;
177    /// #     fn morphology() -> Vec<GeneBounds> { vec![] }
178    /// #     fn encode(&self) -> Vec<i64> { vec![] }
179    /// #     fn decode(_: &[i64]) -> Self::Phenotype { MyType }
180    /// # }
181    /// # async fn example(service: &Service) -> Result<(), Box<dyn std::error::Error>> {
182    ///
183    /// service.new_optimization_request(
184    ///     MyType::NAME,
185    ///     MyType::HASH,
186    ///     FitnessGoal::maximize(0.99)?,                    // High precision target
187    ///     Schedule::rolling(10000, 200, 20),              // Large budget
188    ///     Selector::tournament(2, 150)?,                   // Low selection pressure
189    ///     Mutagen::constant(0.6, 0.4)?,                   // Sustained exploration
190    ///     Crossover::single_point(),                      // Conservative recombination
191    ///     Distribution::latin_hypercube(100),             // High initial diversity
192    ///     None::<()>,
193    /// ).await?;
194    /// # Ok(())
195    /// # }
196    /// ```
197    ///
198    /// # Parameter Interactions
199    ///
200    /// Key relationships to consider:
201    /// - **High selection pressure + Low mutation**: Fast convergence, may get stuck
202    /// - **Low selection pressure + High mutation**: Slow convergence, good exploration
203    /// - **Large populations + Tournament selection**: Better genetic diversity
204    /// - **Small populations + Roulette selection**: Risk of premature convergence
205    /// - **Exponential decay + Generational schedule**: Classic simulated annealing
206    /// - **Linear decay + Rolling schedule**: Steady refinement over time
207    ///
208    /// # Returns
209    ///
210    /// Returns `Ok(())` if the optimization request was successfully created and scheduled.
211    /// The actual optimization runs asynchronously via the event/job system.
212    ///
213    /// # Errors
214    ///
215    /// - Returns error if the specified type hasn't been registered with an evaluator
216    /// - Returns error if database operations fail
217    /// - Parameter validation errors are caught at construction time (e.g., `FitnessGoal::maximize(1.5)` fails)
218    ///
219    /// # Example Usage
220    ///
221    /// ```rust,no_run
222    /// use fx_durable_ga::models::*;
223    /// # use fx_durable_ga::optimization::Service;
224    ///
225    /// // Define your problem type
226    /// #[derive(Debug)]
227    /// struct Point { x: f64, y: f64 }
228    ///
229    /// impl Encodeable for Point {
230    ///     const NAME: &'static str = "Point";
231    ///     type Phenotype = Point;
232    ///     fn morphology() -> Vec<GeneBounds> {
233    ///         vec![
234    ///             GeneBounds::decimal(0.0, 10.0, 1000, 3).unwrap(),
235    ///             GeneBounds::decimal(0.0, 10.0, 1000, 3).unwrap(),
236    ///         ]
237    ///     }
238    ///     fn encode(&self) -> Vec<i64> {
239    ///         let bounds = Self::morphology();
240    ///         vec![
241    ///             bounds[0].from_sample(self.x / 10.0),
242    ///             bounds[1].from_sample(self.y / 10.0),
243    ///         ]
244    ///     }
245    ///     fn decode(genes: &[i64]) -> Self::Phenotype {
246    ///         let bounds = Self::morphology();
247    ///         Point {
248    ///             x: bounds[0].decode_f64(genes[0]),
249    ///             y: bounds[1].decode_f64(genes[1]),
250    ///         }
251    ///     }
252    /// }
253    ///
254    /// # async fn example(service: &Service) -> Result<(), Box<dyn std::error::Error>> {
255    /// // Start optimization
256    /// service.new_optimization_request(
257    ///     Point::NAME,
258    ///     Point::HASH,
259    ///     FitnessGoal::maximize(0.95)?,
260    ///     Schedule::generational(50, 50),
261    ///     Selector::tournament(3, 100)?,
262    ///     Mutagen::constant(0.5, 0.3)?,
263    ///     Crossover::uniform(0.5)?,
264    ///     Distribution::latin_hypercube(50),
265    ///     None::<()>,
266    /// ).await?;
267    /// # Ok(())
268    /// # }
269    /// ```
270    #[instrument(level = "debug", skip(self, data), fields(type_name = type_name, type_hash = type_hash, goal = ?goal, mutagen = ?mutagen, crossover = ?crossover))]
271    pub async fn new_optimization_request(
272        &self,
273        type_name: &str,
274        type_hash: i32,
275        goal: FitnessGoal,
276        schedule: Schedule,
277        selector: Selector,
278        mutagen: Mutagen,
279        crossover: Crossover,
280        distribution: Distribution,
281        data: Option<impl Serialize + Send + Sync>,
282    ) -> Result<Uuid, Error> {
283        tracing::info!("Optimization request received");
284
285        let request = self
286            .requests
287            .chain(|mut tx_requests| {
288                Box::pin(async move {
289                    // Create a new optimization request with the repository
290                    let request = tx_requests
291                        .new_request(Request::new(
292                            type_name,
293                            type_hash,
294                            goal,
295                            selector,
296                            schedule,
297                            mutagen,
298                            crossover,
299                            distribution,
300                            data,
301                        )?)
302                        .await?;
303
304                    // Instantiate a publisher
305                    let mut publisher = fx_event_bus::Publisher::from_tx(tx_requests);
306
307                    // Publish an event within the same transaction
308                    publisher
309                        .publish(OptimizationRequestedEvent::new(request.id))
310                        .await?;
311
312                    Ok((publisher, request))
313                })
314            })
315            .await?;
316
317        Ok(request.id)
318    }
319
320    /// Generates the initial population of genotypes for an optimization request.
321    /// Creates genotypes based on the request's distribution and publishes generation events.
322    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
323    pub(crate) async fn generate_initial_population(&self, request_id: Uuid) -> Result<(), Error> {
324        tracing::info!("Generating initial population");
325
326        // Get the optimization request
327        let request = self.requests.get_request(request_id).await.map_err(|e| {
328            tracing::error!(
329                "Failed to fetch request in generate_initial_population with ID {}: {:?}",
330                request_id,
331                e
332            );
333            e
334        })?;
335
336        // Get the morphology of the type under optimization
337        let morphology = self.morphologies.get_morphology(request.type_hash).await?;
338
339        // Create an inital distribution of genomes
340        let genomes = request.distribution.distribute(&morphology);
341
342        let mut genotypes = Vec::with_capacity(genomes.len());
343        let mut events = Vec::with_capacity(genomes.len());
344
345        for genome in genomes {
346            let genotype =
347                Genotype::new(&request.type_name, request.type_hash, genome, request.id, 1);
348            let event = GenotypeGenerated::new(request.id, genotype.id());
349
350            genotypes.push(genotype);
351            events.push(event);
352        }
353
354        self.genotypes
355            .chain(|mut tx_genotypes| {
356                Box::pin(async move {
357                    // Create the initial generation atomically (prevents race conditions)
358                    let inserted_genotypes = tx_genotypes.new_genotypes(genotypes).await?;
359
360                    // Only proceed if genotypes were actually inserted (no race condition)
361                    if !inserted_genotypes.is_empty() {
362                        // Instantiate a publisher
363                        let mut publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
364
365                        // Publish one event for each generated phenotype
366                        publisher.publish_many(&events).await?;
367
368                        Ok((publisher, request))
369                    } else {
370                        // Race condition occurred - another worker created the generation
371                        // Create a dummy publisher to satisfy the return type
372                        let publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
373                        Ok((publisher, request))
374                    }
375                })
376            })
377            .await?;
378
379        Ok(())
380    }
381
382    /// Evaluates a genotype's fitness using the appropriate evaluator.
383    /// Records the fitness result and publishes an evaluation event.
384    #[instrument(level = "debug", skip(self), fields(request_id = %request_id, genotype_id = %genotype_id))]
385    pub(crate) async fn evaluate_genotype(
386        &self,
387        request_id: Uuid,
388        genotype_id: Uuid,
389    ) -> Result<(), Error> {
390        // Get the genotype from the database
391        let genotype = self
392            .genotypes
393            .get_genotype(&genotype_id)
394            .await
395            .map_err(|e| {
396                tracing::error!("Failed to fetch genotype with ID {}: {:?}", genotype_id, e);
397                e
398            })?;
399
400        // Get the request from the database
401        let request = self.requests.get_request(request_id).await?;
402
403        // Get the evaluator to use for this type
404        let evaluator =
405            self.evaluators
406                .get(&genotype.type_hash())
407                .ok_or(Error::UnknownTypeError {
408                    type_hash: genotype.type_hash(),
409                    type_name: genotype.type_name().to_string(),
410                })?;
411
412        // Call the evaluation function. This is a long running function!
413        let terminator: Box<dyn Terminated> =
414            Box::new(Terminator::new(self.requests.clone(), request_id));
415        let fitness = evaluator.fitness(&genotype, &request, &terminator).await?;
416
417        self.genotypes
418            .chain(|mut tx_genotypes| {
419                Box::pin(async move {
420                    // Record the fitness of the genotype
421                    tx_genotypes
422                        .record_fitness(&Fitness::new(genotype_id, fitness))
423                        .await?;
424
425                    // Instantiate a publisher
426                    let mut publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
427
428                    // Publish an event
429                    publisher
430                        .publish(GenotypeEvaluatedEvent::new(request_id, genotype_id))
431                        .await?;
432
433                    Ok((publisher, ()))
434                })
435            })
436            .await?;
437
438        Ok(())
439    }
440
441    #[instrument(level = "debug", skip(self, request), fields(request_id = %request.id, num_offspring = num_offspring, next_generation_id = next_generation_id))]
442    async fn breed_genotypes(
443        &self,
444        request: &Request,
445        num_offspring: usize,
446        next_generation_id: i32,
447    ) -> Result<(), Error> {
448        // Early return if generation already exists (prevents duplicate work)
449        let exists = self
450            .genotypes
451            .check_if_generation_exists(request.id, next_generation_id)
452            .await?;
453        if exists {
454            return Ok(());
455        }
456
457        tracing::info!("Breeding genotypes");
458        // Get candidates with fitness from populations repository
459        let candidates_with_fitness = self
460            .genotypes
461            .search_genotypes(
462                &genotypes::Filter::default()
463                    .with_request_id(request.id)
464                    .with_fitness(true)
465                    .with_order_random(),
466                request.selector.sample_size(),
467            )
468            .await?;
469
470        // Get morphology for mutation bounds
471        let morphology = self.morphologies.get_morphology(request.type_hash).await?;
472
473        // Get current population to calculate optimization progress
474        let population = self.genotypes.get_population(&request.id).await?;
475
476        let best_fitness = *request
477            .goal
478            .best_fitness(&population.min_fitness, &population.max_fitness);
479
480        // Iterative breeding with deduplication to avoid creating duplicate genomes
481        let mut final_genotypes = Vec::with_capacity(num_offspring);
482        let mut generated_hashes = HashSet::new();
483        let mut deduplication_attempts = 0;
484
485        while final_genotypes.len() < num_offspring
486            && deduplication_attempts < self.max_deduplication_attempts
487        {
488            let needed = num_offspring - final_genotypes.len();
489
490            // Re-select parents for this iteration (borrowed refs)
491            let parent_pairs = request
492                .selector
493                .select_parents(needed, &candidates_with_fitness, &request.goal)
494                .map_err(|e| Error::SelectionError(e))?;
495
496            // Breed using Breeder (before async operations)
497            let batch_genotypes = {
498                let mut rng = rand::rng();
499                Breeder::breed_batch(
500                    &request,
501                    &morphology,
502                    &parent_pairs,
503                    next_generation_id,
504                    request.goal.calculate_progress(best_fitness),
505                    &mut rng,
506                )
507            };
508
509            // Collect hashes from this batch
510            let batch_hashes: Vec<i64> = batch_genotypes.iter().map(|g| g.genome_hash()).collect();
511
512            // Check database for intersections
513            let intersecting_hashes = self
514                .genotypes
515                .get_intersection(request.id, &batch_hashes)
516                .await?
517                .into_iter()
518                .collect::<HashSet<i64>>();
519
520            // Filter out duplicates (database + already generated)
521            let mut unique_count = 0;
522            for genotype in batch_genotypes {
523                if !intersecting_hashes.contains(&genotype.genome_hash())
524                    && !generated_hashes.contains(&genotype.genome_hash())
525                {
526                    generated_hashes.insert(genotype.genome_hash());
527                    final_genotypes.push(genotype);
528                    unique_count += 1;
529                }
530            }
531
532            if unique_count < num_offspring {
533                // Inform of duplicate generation. This is an indication that request params may require tuning.
534                tracing::info!(
535                    "Breeding generated {} non-unique genomes during attempt {}",
536                    num_offspring - unique_count,
537                    deduplication_attempts
538                );
539            }
540
541            // Update deduplication_attempts counter
542            if unique_count == 0 {
543                deduplication_attempts += 1;
544            } else {
545                deduplication_attempts = 0;
546            }
547        }
548
549        // Check if we exhausted max_deduplication_attempts progress iterations
550        if deduplication_attempts >= self.max_deduplication_attempts {
551            tracing::warn!(
552                "Breeding exhausted max_deduplication_attempts ({}) for request {}. Consider tuning selection parameters or increasing diversity.",
553                self.max_deduplication_attempts,
554                request.id
555            );
556        }
557
558        // If we couldn't generate enough unique genotypes, log a warning
559        if final_genotypes.len() < num_offspring {
560            tracing::warn!(
561                "Could only generate {} unique genotypes out of {} requested for request {}",
562                final_genotypes.len(),
563                num_offspring,
564                request.id
565            );
566        }
567
568        // Update database (only if we have any genotypes)
569        if !final_genotypes.is_empty() {
570            // Create events for final genotypes
571            let events: Vec<GenotypeGenerated> = final_genotypes
572                .iter()
573                .map(|genotype| GenotypeGenerated::new(request.id, genotype.id()))
574                .collect();
575
576            self.genotypes
577                .chain(|mut tx_genotypes| {
578                    Box::pin(async move {
579                        let inserted_genotypes =
580                            tx_genotypes.new_genotypes(final_genotypes).await?;
581
582                        // Only proceed if genotypes were actually inserted (no race condition)
583                        if !inserted_genotypes.is_empty() {
584                            let mut publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
585                            publisher.publish_many(&events).await?;
586
587                            Ok((publisher, ()))
588                        } else {
589                            // Race condition occurred - another worker created the generation
590                            let publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
591                            Ok((publisher, ()))
592                        }
593                    })
594                })
595                .await?;
596        }
597
598        Ok(())
599    }
600
601    /// Maintains the population by checking completion status and scheduling breeding or termination.
602    /// Called after each genotype evaluation to determine the next optimization step.
603    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
604    pub(crate) async fn maintain_population(&self, request_id: Uuid) -> Result<(), Error> {
605        let key = format!("maintain_population_{}", request_id);
606        self.locking
607            .lock_while(&key, || async {
608                // Check if request is already concluded - skip if so
609                if let Some(_) = self.requests.get_request_conclusion(&request_id).await? {
610                    tracing::debug!("Request already concluded, skipping maintenance");
611                    return Ok(());
612                }
613
614                // Get the request
615                let request = self.requests.get_request(request_id).await.map_err(|e| {
616                    tracing::error!("Failed to fetch request with ID {}: {:?}", request_id, e);
617                    e
618                })?;
619
620                // Get the population
621                let population = self.genotypes.get_population(&request.id).await?;
622
623                let Some(best_fitness) = *request
624                    .goal
625                    .best_fitness(&population.min_fitness, &population.max_fitness)
626                else {
627                    return Ok(());
628                };
629
630                if request.is_completed(best_fitness) {
631                    // Publish RequestCompleted::new(request_id)
632                    self.genotypes
633                        .chain(|tx_genotypes| {
634                            Box::pin(async move {
635                                // Create publisher
636                                let mut publisher = fx_event_bus::Publisher::from_tx(tx_genotypes);
637
638                                // Publish completion event
639                                publisher
640                                    .publish(RequestCompletedEvent::new(request.id))
641                                    .await?;
642
643                                Ok((publisher, ()))
644                            })
645                        })
646                        .await?;
647                    return Ok(());
648                }
649
650                match request.schedule.should_breed(&population) {
651                    ScheduleDecision::Wait => Ok(()), // Do nothing
652                    ScheduleDecision::Terminate => self.publish_terminated(request.id).await,
653                    ScheduleDecision::Breed {
654                        num_offspring,
655                        next_generation_id,
656                    } => {
657                        self.breed_genotypes(&request, num_offspring, next_generation_id)
658                            .await
659                    }
660                }
661            })
662            .await?
663    }
664
665    /// Records the conclusion of an optimization request to prevent duplicate processing.
666    /// Uses locking to ensure atomicity when multiple workers might conclude the same request.
667    #[instrument(level = "debug", skip(self), fields(request_id = %request_conclusion.request_id, concluded_at = %request_conclusion.concluded_at, concluded_with = ?request_conclusion.concluded_with))]
668    pub(crate) async fn conclude_request(
669        &self,
670        request_conclusion: RequestConclusion,
671    ) -> Result<(), Error> {
672        let key = format!("conclude_request_{}", request_conclusion.request_id);
673        self.locking
674            .lock_while(&key, || async {
675                if let Some(_) = self
676                    .requests
677                    .get_request_conclusion(&request_conclusion.request_id)
678                    .await?
679                {
680                    return Ok::<(), super::Error>(());
681                }
682
683                tracing::info!("Concluding request");
684
685                self.requests
686                    .new_request_conclusion(&request_conclusion)
687                    .await?;
688
689                Ok(())
690            })
691            .await?
692    }
693
694    /// Publishes a termination event for an optimization request.
695    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
696    pub(crate) async fn publish_terminated(&self, request_id: Uuid) -> Result<(), Error> {
697        self.genotypes
698            .chain(|tx_genotypes| {
699                Box::pin(async move {
700                    let mut publisher = Publisher::new(tx_genotypes.tx());
701                    let ret = publisher
702                        .publish(RequestTerminatedEvent::new(request_id))
703                        .await?;
704
705                    Ok((publisher, ret))
706                })
707            })
708            .await?;
709
710        Ok(())
711    }
712
713    /// Interrupts a running optimization request.
714    /// Publishes an interruption event that concludes the request with the "Interrupted" reason.
715    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
716    pub async fn interrupt_request(&self, request_id: Uuid) -> Result<(), Error> {
717        self.genotypes
718            .chain(|tx_genotypes| {
719                Box::pin(async move {
720                    let mut publisher = Publisher::new(tx_genotypes.tx());
721                    let ret = publisher
722                        .publish(RequestInterruptedEvent::new(request_id))
723                        .await?;
724
725                    Ok((publisher, ret))
726                })
727            })
728            .await?;
729
730        Ok(())
731    }
732
733    /// Gets the best genotype for a request based on fitness.
734    /// Returns the genotype with the best (min or max) fitness depending on the goal.
735    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
736    pub async fn get_best_genotype(
737        &self,
738        request_id: Uuid,
739    ) -> Result<Option<(Genotype, f64)>, Error> {
740        let request = self.requests.get_request(request_id).await?;
741
742        // Determine sort order based on goal
743        let filter = match request.goal {
744            FitnessGoal::Minimize { .. } => genotypes::Filter::default()
745                .with_request_id(request_id)
746                .with_fitness(true)
747                .with_order_fitness_asc(),
748            FitnessGoal::Maximize { .. } => genotypes::Filter::default()
749                .with_request_id(request_id)
750                .with_fitness(true)
751                .with_order_fitness_desc(),
752        };
753
754        let results = self.genotypes.search_genotypes(&filter, 1).await?;
755
756        // Extract the first result if it exists and has fitness
757        let best = results
758            .into_iter()
759            .next()
760            .and_then(|(genotype, fitness_opt)| fitness_opt.map(|fitness| (genotype, fitness)));
761
762        Ok(best)
763    }
764
765    /// Checks if an optimization request has concluded.
766    /// Returns true if the request has been completed or terminated.
767    #[instrument(level = "debug", skip(self), fields(request_id = %request_id))]
768    pub async fn is_request_concluded(&self, request_id: Uuid) -> Result<bool, Error> {
769        Ok(self
770            .requests
771            .get_request_conclusion(&request_id)
772            .await?
773            .is_some())
774    }
775
776    /// Get a morphology by its type name
777    #[instrument(level = "debug", skip(self), fields(type_name))]
778    pub async fn search_genotypes(
779        &self,
780        filter: &genotypes::Filter,
781        limit: i64,
782    ) -> Result<Vec<(Genotype, Option<f64>)>, Error> {
783        let genotypes = self.genotypes.search_genotypes(filter, limit).await?;
784        Ok(genotypes)
785    }
786
787    /// Get a phenotype by its type name
788    #[instrument(level = "debug", skip(self), fields(type_name))]
789    pub async fn get_phenotype<T: Encodeable>(
790        &self,
791        genotype_id: &Uuid,
792    ) -> Result<T::Phenotype, Error> {
793        let genotype = self.genotypes.get_genotype(genotype_id).await?;
794
795        if genotype.type_hash() != T::HASH {
796            return Err(Error::UnknownPhenotype {
797                type_name: genotype.type_name().to_string(),
798                type_hash: genotype.type_hash(),
799            });
800        }
801
802        Ok(T::decode(&genotype.genome()))
803    }
804}
805
806#[cfg(test)]
807mod tests {
808    use super::*;
809    use crate::bootstrap;
810    use crate::models::{
811        Crossover, Distribution, Encodeable, Evaluator, FitnessGoal, GeneBounds, Mutagen,
812        MutationRate, Request, Schedule, Selector, Temperature, Terminated,
813    };
814    use futures::future::BoxFuture;
815
816    async fn create_test_service(pool: sqlx::PgPool) -> Service {
817        let builder = bootstrap::bootstrap(pool).await.unwrap();
818        builder.build()
819    }
820
821    #[sqlx::test(migrations = false)]
822    async fn test_new_optimization_request_happy_path(pool: sqlx::PgPool) -> anyhow::Result<()> {
823        crate::migrations::run_default_migrations(&pool).await?;
824
825        // Create service
826        let service = create_test_service(pool.clone()).await;
827
828        // Create valid request parameters
829        let type_name = "TestType";
830        let type_hash = 123;
831        let goal = FitnessGoal::maximize(0.95)?;
832        let schedule = Schedule::generational(100, 10);
833        let selector = Selector::tournament(3, 20).expect("is valid");
834        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
835        let crossover = Crossover::uniform(0.5)?;
836        let distribution = Distribution::random(50);
837
838        // Call the method
839        let result = service
840            .new_optimization_request(
841                type_name,
842                type_hash,
843                goal.clone(),
844                schedule.clone(),
845                selector.clone(),
846                mutagen.clone(),
847                crossover.clone(),
848                distribution.clone(),
849                None::<()>,
850            )
851            .await;
852
853        // Verify success
854        assert!(result.is_ok(), "new_optimization_request should succeed");
855
856        // Verify request was stored in database
857        let request_count = sqlx::query_scalar!(
858            "SELECT COUNT(*) FROM fx_durable_ga.requests WHERE type_name = $1 AND type_hash = $2",
859            type_name,
860            type_hash
861        )
862        .fetch_one(&pool)
863        .await?
864        .unwrap_or(0);
865
866        assert_eq!(
867            request_count, 1,
868            "Exactly one request should be stored in database"
869        );
870
871        // Verify an OptimizationRequestedEvent was published
872        let event_count: i64 = sqlx::query_scalar::<_, i64>(
873            "SELECT COUNT(*) FROM fx_event_bus.events_unacknowledged WHERE name = $1",
874        )
875        .bind("OptimizationRequested")
876        .fetch_one(&pool)
877        .await?;
878
879        assert_eq!(
880            event_count, 1,
881            "Exactly one OptimizationRequestedEvent should be published"
882        );
883
884        Ok(())
885    }
886
887    #[sqlx::test(migrations = false)]
888    async fn test_generate_initial_population_happy_path(pool: sqlx::PgPool) -> anyhow::Result<()> {
889        crate::migrations::run_default_migrations(&pool).await?;
890
891        // Create service
892        let service = create_test_service(pool.clone()).await;
893
894        // Create a test morphology first
895        let type_name = "TestType";
896        let type_hash = 456;
897        let morphology = crate::models::Morphology::new(
898            type_name,
899            type_hash,
900            vec![
901                crate::models::GeneBounds::integer(0, 9, 10).unwrap(), // 10 steps
902                crate::models::GeneBounds::integer(0, 4, 5).unwrap(),  // 5 steps
903            ],
904        );
905
906        // Insert morphology into database
907        service.morphologies.new_morphology(morphology).await?;
908
909        // Create a request
910        let goal = FitnessGoal::maximize(0.95)?;
911        let schedule = Schedule::generational(100, 10);
912        let selector = Selector::tournament(3, 20).expect("is valid");
913        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
914        let crossover = Crossover::uniform(0.5)?;
915        let distribution = Distribution::random(10); // Small population for testing
916
917        let request = crate::models::Request::new(
918            type_name,
919            type_hash,
920            goal,
921            selector,
922            schedule,
923            mutagen,
924            crossover,
925            distribution,
926            None::<()>,
927        )?;
928        let request_id = request.id;
929
930        // Insert request into database using chainable pattern
931        service
932            .requests
933            .chain(|mut tx_requests| {
934                Box::pin(async move {
935                    let request = tx_requests.new_request(request).await?;
936                    Ok((tx_requests, request))
937                })
938            })
939            .await?;
940
941        // Call the method under test
942        let result = service.generate_initial_population(request_id).await;
943
944        // Verify success
945        assert!(result.is_ok(), "generate_initial_population should succeed");
946
947        // Verify genotypes were created in database
948        let genotype_count = sqlx::query_scalar!(
949            "SELECT COUNT(*) FROM fx_durable_ga.genotypes WHERE request_id = $1 AND generation_id = 1",
950            request_id
951        )
952        .fetch_one(&pool)
953        .await?
954        .unwrap_or(0);
955
956        assert_eq!(
957            genotype_count, 10,
958            "Exactly 10 genotypes should be created for generation 1"
959        );
960
961        // Verify GenotypeGenerated events were published
962        let event_count: i64 = sqlx::query_scalar::<_, i64>(
963            "SELECT COUNT(*) FROM fx_event_bus.events_unacknowledged WHERE name = $1",
964        )
965        .bind("GenotypeGenerated")
966        .fetch_one(&pool)
967        .await?;
968
969        assert_eq!(
970            event_count, 10,
971            "Exactly 10 GenotypeGenerated events should be published"
972        );
973
974        Ok(())
975    }
976
977    #[sqlx::test(migrations = false)]
978    async fn test_evaluate_genotype_happy_path(pool: sqlx::PgPool) -> anyhow::Result<()> {
979        crate::migrations::run_default_migrations(&pool).await?;
980
981        // Define a simple test type
982        #[derive(Debug)]
983        struct TestType;
984
985        impl Encodeable for TestType {
986            const NAME: &'static str = "TestType";
987            type Phenotype = (i64, i64);
988
989            fn morphology() -> Vec<GeneBounds> {
990                vec![
991                    GeneBounds::integer(0, 9, 10).unwrap(),
992                    GeneBounds::integer(0, 4, 5).unwrap(),
993                ]
994            }
995
996            fn encode(&self) -> Vec<i64> {
997                vec![5, 2] // Not used in test
998            }
999
1000            fn decode(genes: &[i64]) -> Self::Phenotype {
1001                (genes[0], genes[1])
1002            }
1003        }
1004
1005        // Simple evaluator that returns constant fitness
1006        struct TestEvaluator;
1007
1008        impl Evaluator<(i64, i64)> for TestEvaluator {
1009            fn fitness<'a>(
1010                &self,
1011                _genotype_id: uuid::Uuid,
1012                _phenotype: (i64, i64),
1013                _request: &'a Request,
1014                _terminated: &'a Box<dyn Terminated>,
1015            ) -> BoxFuture<'a, Result<f64, anyhow::Error>> {
1016                Box::pin(async move { Ok(0.75) })
1017            }
1018        }
1019
1020        // Create service and register evaluator
1021        let service = bootstrap::bootstrap(pool.clone())
1022            .await?
1023            .register::<TestType, _>(TestEvaluator)
1024            .await?
1025            .build();
1026
1027        // Create a test request
1028        let goal = FitnessGoal::maximize(0.95)?;
1029        let schedule = Schedule::generational(100, 10);
1030        let selector = Selector::tournament(3, 20).expect("is valid");
1031        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
1032        let crossover = Crossover::uniform(0.5)?;
1033        let distribution = Distribution::random(1);
1034
1035        let request = crate::models::Request::new(
1036            TestType::NAME,
1037            TestType::HASH,
1038            goal,
1039            selector,
1040            schedule,
1041            mutagen,
1042            crossover,
1043            distribution,
1044            None::<()>,
1045        )?;
1046        let request_id = request.id;
1047
1048        // Insert request into database
1049        service
1050            .requests
1051            .chain(|mut tx_requests| {
1052                Box::pin(async move {
1053                    let request = tx_requests.new_request(request).await?;
1054                    Ok((tx_requests, request))
1055                })
1056            })
1057            .await?;
1058
1059        // Create a genotype manually
1060        let genotype = crate::models::Genotype::new(
1061            TestType::NAME,
1062            TestType::HASH,
1063            vec![7, 3], // Test genome
1064            request_id,
1065            1, // generation_id
1066        );
1067        let genotype_id = genotype.id();
1068
1069        // Insert genotype into database
1070        service
1071            .genotypes
1072            .chain(|mut tx_genotypes| {
1073                Box::pin(async move {
1074                    let genotypes = tx_genotypes.new_genotypes(vec![genotype]).await?;
1075                    Ok((tx_genotypes, genotypes))
1076                })
1077            })
1078            .await?;
1079
1080        // Call the method under test
1081        let result = service.evaluate_genotype(request_id, genotype_id).await;
1082
1083        // Verify success
1084        assert!(result.is_ok(), "evaluate_genotype should succeed");
1085
1086        // Verify fitness was recorded in database
1087        let fitness = sqlx::query_scalar!(
1088            "SELECT fitness FROM fx_durable_ga.fitness WHERE genotype_id = $1",
1089            genotype_id
1090        )
1091        .fetch_one(&pool)
1092        .await?;
1093
1094        assert_eq!(fitness, 0.75, "Fitness should be recorded as 0.75");
1095
1096        // Verify GenotypeEvaluatedEvent was published
1097        let event_count: i64 = sqlx::query_scalar::<_, i64>(
1098            "SELECT COUNT(*) FROM fx_event_bus.events_unacknowledged WHERE name = $1",
1099        )
1100        .bind("GenotypeEvaluated")
1101        .fetch_one(&pool)
1102        .await?;
1103
1104        assert_eq!(
1105            event_count, 1,
1106            "Exactly one GenotypeEvaluatedEvent should be published"
1107        );
1108
1109        Ok(())
1110    }
1111
1112    #[sqlx::test(migrations = false)]
1113    async fn test_maintain_population_request_completion(pool: sqlx::PgPool) -> anyhow::Result<()> {
1114        crate::migrations::run_default_migrations(&pool).await?;
1115
1116        // Create service
1117        let service = create_test_service(pool.clone()).await;
1118
1119        // Create morphology
1120        let type_name = "TestType";
1121        let type_hash = 789;
1122        let morphology = crate::models::Morphology::new(
1123            type_name,
1124            type_hash,
1125            vec![
1126                crate::models::GeneBounds::integer(0, 9, 10).unwrap(),
1127                crate::models::GeneBounds::integer(0, 4, 5).unwrap(),
1128            ],
1129        );
1130        service.morphologies.new_morphology(morphology).await?;
1131
1132        // Create request with LOW fitness goal (0.5) so it's easy to exceed
1133        let goal = FitnessGoal::maximize(0.5)?;
1134        let schedule = Schedule::generational(100, 10);
1135        let selector = Selector::tournament(3, 20).expect("is valid");
1136        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
1137        let crossover = Crossover::uniform(0.5)?;
1138        let distribution = Distribution::random(5);
1139
1140        let request = crate::models::Request::new(
1141            type_name,
1142            type_hash,
1143            goal,
1144            selector,
1145            schedule,
1146            mutagen,
1147            crossover,
1148            distribution,
1149            None::<()>,
1150        )?;
1151        let request_id = request.id;
1152
1153        // Insert request
1154        service
1155            .requests
1156            .chain(|mut tx_requests| {
1157                Box::pin(async move {
1158                    let request = tx_requests.new_request(request).await?;
1159                    Ok((tx_requests, request))
1160                })
1161            })
1162            .await?;
1163
1164        // Create genotype with fitness that EXCEEDS the goal (0.8 > 0.5)
1165        let genotype =
1166            crate::models::Genotype::new(type_name, type_hash, vec![5, 2], request_id, 1);
1167        let genotype_id = genotype.id();
1168
1169        // Insert genotype and fitness
1170        service
1171            .genotypes
1172            .chain(|mut tx_genotypes| {
1173                Box::pin(async move {
1174                    // Insert genotype
1175                    tx_genotypes.new_genotypes(vec![genotype]).await?;
1176
1177                    // Insert fitness that exceeds goal
1178                    let fitness = crate::models::Fitness::new(genotype_id, 0.8);
1179                    tx_genotypes.record_fitness(&fitness).await?;
1180
1181                    Ok((tx_genotypes, ()))
1182                })
1183            })
1184            .await?;
1185
1186        // Call the method under test
1187        let result = service.maintain_population(request_id).await;
1188
1189        // Verify success
1190        assert!(result.is_ok(), "maintain_population should succeed");
1191
1192        // Verify RequestCompletedEvent was published
1193        let event_count: i64 = sqlx::query_scalar::<_, i64>(
1194            "SELECT COUNT(*) FROM fx_event_bus.events_unacknowledged WHERE name = $1",
1195        )
1196        .bind("RequestCompleted")
1197        .fetch_one(&pool)
1198        .await?;
1199
1200        assert_eq!(
1201            event_count, 1,
1202            "Exactly one RequestCompletedEvent should be published"
1203        );
1204
1205        Ok(())
1206    }
1207
1208    #[sqlx::test(migrations = false)]
1209    async fn test_interrupt_request(pool: sqlx::PgPool) -> anyhow::Result<()> {
1210        crate::migrations::run_default_migrations(&pool).await?;
1211
1212        // Create service
1213        let service = create_test_service(pool.clone()).await;
1214
1215        // Create a test request
1216        let goal = FitnessGoal::maximize(0.95)?;
1217        let schedule = Schedule::generational(100, 10);
1218        let selector = Selector::tournament(3, 20).expect("is valid");
1219        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
1220        let crossover = Crossover::uniform(0.5)?;
1221        let distribution = Distribution::random(5);
1222
1223        let request = crate::models::Request::new(
1224            "TestType",
1225            123,
1226            goal,
1227            selector,
1228            schedule,
1229            mutagen,
1230            crossover,
1231            distribution,
1232            None::<()>,
1233        )?;
1234        let request_id = request.id;
1235
1236        // Insert request
1237        service
1238            .requests
1239            .chain(|mut tx_requests| {
1240                Box::pin(async move {
1241                    let request = tx_requests.new_request(request).await?;
1242                    Ok((tx_requests, request))
1243                })
1244            })
1245            .await?;
1246
1247        // Call the method under test
1248        let result = service.interrupt_request(request_id).await;
1249
1250        // Verify success
1251        assert!(result.is_ok(), "interrupt_request should succeed");
1252
1253        // Verify RequestInterruptedEvent was published
1254        let event_count: i64 = sqlx::query_scalar::<_, i64>(
1255            "SELECT COUNT(*) FROM fx_event_bus.events_unacknowledged WHERE name = $1",
1256        )
1257        .bind("RequestInterrupted")
1258        .fetch_one(&pool)
1259        .await?;
1260
1261        assert_eq!(
1262            event_count, 1,
1263            "Exactly one RequestInterruptedEvent should be published"
1264        );
1265
1266        Ok(())
1267    }
1268
1269    #[sqlx::test(migrations = false)]
1270    async fn test_get_phenotype(pool: sqlx::PgPool) -> anyhow::Result<()> {
1271        crate::migrations::run_default_migrations(&pool).await?;
1272
1273        #[derive(Debug, PartialEq)]
1274        struct TestPoint {
1275            x: i64,
1276            y: i64,
1277        }
1278
1279        impl Encodeable for TestPoint {
1280            const NAME: &'static str = "TestPoint";
1281            type Phenotype = TestPoint;
1282
1283            fn morphology() -> Vec<GeneBounds> {
1284                vec![
1285                    GeneBounds::integer(0, 9, 10).unwrap(),
1286                    GeneBounds::integer(0, 4, 5).unwrap(),
1287                ]
1288            }
1289
1290            fn encode(&self) -> Vec<i64> {
1291                vec![self.x, self.y]
1292            }
1293
1294            fn decode(genes: &[i64]) -> Self::Phenotype {
1295                TestPoint {
1296                    x: genes[0],
1297                    y: genes[1],
1298                }
1299            }
1300        }
1301
1302        let service = create_test_service(pool.clone()).await;
1303
1304        let goal = FitnessGoal::maximize(0.95)?;
1305        let schedule = Schedule::generational(100, 10);
1306        let selector = Selector::tournament(3, 20).expect("is valid");
1307        let mutagen = Mutagen::new(Temperature::constant(0.5)?, MutationRate::constant(0.3)?);
1308        let crossover = Crossover::uniform(0.5)?;
1309        let distribution = Distribution::random(10);
1310
1311        let request = crate::models::Request::new(
1312            TestPoint::NAME,
1313            TestPoint::HASH,
1314            goal,
1315            selector,
1316            schedule,
1317            mutagen,
1318            crossover,
1319            distribution,
1320            None::<()>,
1321        )?;
1322        let request_id = request.id;
1323
1324        service
1325            .requests
1326            .chain(|mut tx_requests| {
1327                Box::pin(async move {
1328                    let request = tx_requests.new_request(request).await?;
1329                    Ok((tx_requests, request))
1330                })
1331            })
1332            .await?;
1333
1334        let genotype = crate::models::Genotype::new(
1335            TestPoint::NAME,
1336            TestPoint::HASH,
1337            vec![7, 3],
1338            request_id,
1339            1,
1340        );
1341        let genotype_id = genotype.id();
1342
1343        service
1344            .genotypes
1345            .chain(|mut tx_genotypes| {
1346                Box::pin(async move {
1347                    tx_genotypes.new_genotypes(vec![genotype]).await?;
1348                    Ok((tx_genotypes, ()))
1349                })
1350            })
1351            .await?;
1352
1353        let result: TestPoint = service.get_phenotype::<TestPoint>(&genotype_id).await?;
1354
1355        assert_eq!(result.x, 7);
1356        assert_eq!(result.y, 3);
1357
1358        Ok(())
1359    }
1360}
fx_durable_ga/services/optimization/service.rs

fx_durable_ga/services/optimization/
service.rs
