From b6cd9bf5ef1ab2d6f27c7445f162b7174e7f0891 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 5 Nov 2026 10:02:10 +0100 Subject: [PATCH 1/2] fix(scoring)!: panic when a projection emits more than MAX_EMITS Projection::MAX_EMITS was documented as the most rows one project call emits, but nothing read it at runtime, so a projection emitting more was accepted silently. A single-source projection now runs through a bounded sink that counts each input's rows and panics on the first row past the bound, naming the projection and both counts, on both the full and the incremental path. BREAKING CHANGE: a Projection that emits more rows for one input than its MAX_EMITS now panics; raise MAX_EMITS to the real maximum. --- crates/solverforge-scoring/WIREFRAME.md | 2 +- .../src/constraint/tests/projected.rs | 1 + .../constraint/tests/projected/emit_bound.rs | 75 +++++++++++++++++++ .../src/stream/projected_stream/source.rs | 50 +++++++++++-- .../stream/projected_stream/source/single.rs | 26 ++----- crates/solverforge/WIREFRAME.md | 2 +- 6 files changed, 129 insertions(+), 27 deletions(-) create mode 100644 crates/solverforge-scoring/src/constraint/tests/projected/emit_bound.rs diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index b78e1b488..b26d5af96 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -754,7 +754,7 @@ Dynamic closure weights are non-hard metadata by default, even when their score **`UniConstraintBuilder`** — `named()` → `IncrementalUniConstraint` -**`Projection`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink)`. Projection implementations emit bounded scoring rows into the sink; Vec-returning closures are not part of the API. `Out` does not need `Clone`. +**`Projection`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink)`. Projection implementations emit bounded scoring rows into the sink: emitting more than `MAX_EMITS` rows for one input panics, naming the projection and both counts; Vec-returning closures are not part of the API. `Out` does not need `Clone`. The fluent projected stream keeps its concrete source type in the return type: `SingleSource` for one projection, `FilteredSource` after a projected filter, diff --git a/crates/solverforge-scoring/src/constraint/tests/projected.rs b/crates/solverforge-scoring/src/constraint/tests/projected.rs index fd4da8d23..f5b1517f4 100644 --- a/crates/solverforge-scoring/src/constraint/tests/projected.rs +++ b/crates/solverforge-scoring/src/constraint/tests/projected.rs @@ -1,3 +1,4 @@ +mod emit_bound; mod joined; mod localization; mod self_join; diff --git a/crates/solverforge-scoring/src/constraint/tests/projected/emit_bound.rs b/crates/solverforge-scoring/src/constraint/tests/projected/emit_bound.rs new file mode 100644 index 000000000..0571e2954 --- /dev/null +++ b/crates/solverforge-scoring/src/constraint/tests/projected/emit_bound.rs @@ -0,0 +1,75 @@ +use super::support::*; + +/// Emits one entry per unit of demand while declaring at most two. +struct EntryPerDemand; + +impl Projection for EntryPerDemand { + type Out = Entry; + const MAX_EMITS: usize = 2; + + fn project(&self, work: &Work, out: &mut Sink) + where + Sink: ProjectionSink, + { + for bucket in 0..work.demand as usize { + out.emit(Entry { bucket, delta: 1 }); + } + } +} + +fn entries_per_demand() -> impl IncrementalConstraint { + ConstraintFactory::::new() + .for_each(source( + work as fn(&Plan) -> &[Work], + ChangeSource::Descriptor(0), + )) + .project(EntryPerDemand) + .group_by( + |entry: &Entry| entry.bucket, + sum(|entry: &Entry| entry.delta), + ) + .penalize(|_bucket: &usize, delta: &i64| SoftScore::of(*delta)) + .named("entries") +} + +fn plan(demand: i64) -> Plan { + Plan { + work: vec![Work { + bucket: 0, + demand, + enabled: true, + }], + capacity: Vec::new(), + } +} + +#[test] +fn a_projection_may_emit_up_to_its_bound() { + let mut constraint = entries_per_demand(); + let mut plan = plan(1); + let mut total = constraint.initialize(&plan); + assert_eq!(total, SoftScore::of(-1)); + + total = total + constraint.on_retract(&plan, 0, 0); + plan.work[0].demand = 2; + total = total + constraint.on_insert(&plan, 0, 0); + assert_eq!(total, SoftScore::of(-2)); + assert_eq!(total, constraint.evaluate(&plan)); +} + +#[test] +#[should_panic(expected = "EntryPerDemand` emitted 3 rows for one input; its MAX_EMITS is 2")] +fn emitting_past_the_bound_panics_on_initialization() { + entries_per_demand().initialize(&plan(3)); +} + +#[test] +#[should_panic(expected = "EntryPerDemand` emitted 3 rows for one input; its MAX_EMITS is 2")] +fn emitting_past_the_bound_panics_on_an_incremental_insert() { + let mut constraint = entries_per_demand(); + let mut plan = plan(2); + constraint.initialize(&plan); + constraint.on_retract(&plan, 0, 0); + plan.work[0].demand = 3; + constraint.on_insert(&plan, 0, 0); +} diff --git a/crates/solverforge-scoring/src/stream/projected_stream/source.rs b/crates/solverforge-scoring/src/stream/projected_stream/source.rs index 302824117..00725d371 100644 --- a/crates/solverforge-scoring/src/stream/projected_stream/source.rs +++ b/crates/solverforge-scoring/src/stream/projected_stream/source.rs @@ -25,8 +25,9 @@ pub trait ProjectionSink { pub trait Projection: Send + Sync { /// The emitted row type; it need not be `Clone`. type Out: Send + Sync + 'static; - /// Declared maximum number of rows one `project` call emits. Sources combine - /// it (a merge adds both bounds); nothing reads it at runtime. + /// Maximum number of rows one `project` call emits. Emitting more panics, + /// naming the projection and both counts. Sources combine it (a merge adds + /// both bounds). const MAX_EMITS: usize; /// Emits the rows for `input` into `sink`. @@ -35,20 +36,57 @@ pub trait Projection: Send + Sync { Sink: ProjectionSink; } -struct VisitSink { +/// Runs `projection` on `input`, passing each row with its emit index to +/// `visit`. +/// +/// # Panics +/// +/// When the projection emits more than `P::MAX_EMITS` rows. +fn project_bounded(projection: &P, input: &A, visit: V) +where + P: Projection, + V: FnMut(usize, P::Out), +{ + let mut sink = BoundedSink { + visit, + emitted: 0, + max_emits: P::MAX_EMITS, + projection: std::any::type_name::

, + }; + projection.project(input, &mut sink); +} + +struct BoundedSink { visit: V, + emitted: usize, + max_emits: usize, + projection: fn() -> &'static str, } -impl ProjectionSink for VisitSink +impl ProjectionSink for BoundedSink where - V: FnMut(Out), + V: FnMut(usize, Out), { #[inline] fn emit(&mut self, output: Out) { - (self.visit)(output); + if self.emitted == self.max_emits { + emitted_past_bound((self.projection)(), self.max_emits); + } + let emit_index = self.emitted; + self.emitted += 1; + (self.visit)(emit_index, output); } } +#[cold] +#[inline(never)] +fn emitted_past_bound(projection: &str, max_emits: usize) -> ! { + panic!( + "projection `{projection}` emitted {} rows for one input; its MAX_EMITS is {max_emits}", + max_emits + 1 + ); +} + #[doc(hidden)] #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] pub struct RowOwner { diff --git a/crates/solverforge-scoring/src/stream/projected_stream/source/single.rs b/crates/solverforge-scoring/src/stream/projected_stream/source/single.rs index 57724a98b..ee4b0b344 100644 --- a/crates/solverforge-scoring/src/stream/projected_stream/source/single.rs +++ b/crates/solverforge-scoring/src/stream/projected_stream/source/single.rs @@ -3,7 +3,7 @@ use std::marker::PhantomData; use crate::stream::collection_extract::{ChangeSource, CollectionExtract}; use crate::stream::filter::UniFilter; -use super::{Projection, RowCoordinate, Source, VisitSink}; +use super::{project_bounded, Projection, RowCoordinate, Source}; pub struct SingleSource { extractor: E, @@ -58,15 +58,9 @@ where if !self.filter.test(solution, entity) { continue; } - let mut emit_index = 0; - let mut sink = VisitSink { - visit: |output| { - let coordinate = RowCoordinate::single(0, idx, emit_index); - emit_index += 1; - visit(coordinate, output); - }, - }; - self.projection.project(entity, &mut sink); + project_bounded(&self.projection, entity, |emit_index, output| { + visit(RowCoordinate::single(0, idx, emit_index), output); + }); } } @@ -90,15 +84,9 @@ where if !self.filter.test(solution, entity) { return; } - let mut emit_index = 0; - let mut sink = VisitSink { - visit: |output| { - let coordinate = RowCoordinate::single(0, entity_index, emit_index); - emit_index += 1; - visit(coordinate, output); - }, - }; - self.projection.project(entity, &mut sink); + project_bounded(&self.projection, entity, |emit_index, output| { + visit(RowCoordinate::single(0, entity_index, emit_index), output); + }); } fn insert_entity_state( diff --git a/crates/solverforge/WIREFRAME.md b/crates/solverforge/WIREFRAME.md index 2058cde4a..046dd50dc 100644 --- a/crates/solverforge/WIREFRAME.md +++ b/crates/solverforge/WIREFRAME.md @@ -373,7 +373,7 @@ ConstraintFactory::::new() Generated existence ergonomics: `#[planning_solution]` generates inherent source methods such as `Plan::assignments()` with hidden descriptor/static metadata, plus the matching `PlanConstraintStreams` convenience trait. Localized incremental callbacks use entity indexes only for the owning planning-entity collection. Raw facade `for_each(...)` extractors do not carry localized source ownership. Flattened existence targets use `.flattened(...)` and `FlattenedCollectionTarget`. -Projected scoring ergonomics: `ConstraintFactory::new().for_each(Plan::assignments()).project(entries)`, where `entries` is a value of a named `Projection` type, creates bounded scoring rows from a named `Projection` type without materializing facts or entities. Keyed cross joins can group joined pairs directly with `.group_by(|left, right| key, collector)` and complement the grouped result, or project them with `.project(|assignment, capacity| Row { ... })` to emit one scoring row per retained joined pair. Projected streams can be filtered, self-joined, merged, grouped, complemented after grouping, and weighted as retained scoring state. Single-source projection implementations emit through `ProjectionSink` and declare `MAX_EMITS`; joined-pair closures do not need a helper type. Projected output rows, projected self-join keys, and grouped collector values do not need `Clone`. Projected self-join ordering is coordinate-stable by `RowCoordinate`; directed projected self-joins use `.join(joiner::equal_bi(left_key, right_key))`; low-level pair-filter indexes are primary owner entity indexes, not sparse storage row IDs. +Projected scoring ergonomics: `ConstraintFactory::new().for_each(Plan::assignments()).project(entries)`, where `entries` is a value of a named `Projection` type, creates bounded scoring rows from a named `Projection` type without materializing facts or entities. Keyed cross joins can group joined pairs directly with `.group_by(|left, right| key, collector)` and complement the grouped result, or project them with `.project(|assignment, capacity| Row { ... })` to emit one scoring row per retained joined pair. Projected streams can be filtered, self-joined, merged, grouped, complemented after grouping, and weighted as retained scoring state. Single-source projection implementations emit through `ProjectionSink` and declare `MAX_EMITS`, the most rows one input may emit (emitting more panics); joined-pair closures do not need a helper type. Projected output rows, projected self-join keys, and grouped collector values do not need `Clone`. Projected self-join ordering is coordinate-stable by `RowCoordinate`; directed projected self-joins use `.join(joiner::equal_bi(left_key, right_key))`; low-level pair-filter indexes are primary owner entity indexes, not sparse storage row IDs. Constraint authoring: `#[solverforge_constraints]` is the canonical attribute for constraint factory functions. It preserves fluent stream syntax and gives From 3602138242d1c54ddcc75712cdba5e51dad2eedb Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 5 Nov 2026 10:02:10 +0100 Subject: [PATCH 2/2] perf(scoring): size each projected row list by MAX_EMITS A projected source's per-entity row list started empty and grew by doubling as rows arrived. It is now allocated once with MAX_EMITS capacity, which holds every row a single-source projection can emit for that entity, and the bound's documentation says so. Closes #208. --- crates/solverforge-scoring/WIREFRAME.md | 2 +- .../src/stream/projected_stream/source.rs | 5 +++-- .../src/stream/projected_stream/source/operator.rs | 7 ++++++- 3 files changed, 10 insertions(+), 4 deletions(-) diff --git a/crates/solverforge-scoring/WIREFRAME.md b/crates/solverforge-scoring/WIREFRAME.md index b26d5af96..51aae76de 100644 --- a/crates/solverforge-scoring/WIREFRAME.md +++ b/crates/solverforge-scoring/WIREFRAME.md @@ -754,7 +754,7 @@ Dynamic closure weights are non-hard metadata by default, even when their score **`UniConstraintBuilder`** — `named()` → `IncrementalUniConstraint` -**`Projection`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink)`. Projection implementations emit bounded scoring rows into the sink: emitting more than `MAX_EMITS` rows for one input panics, naming the projection and both counts; Vec-returning closures are not part of the API. `Out` does not need `Clone`. +**`Projection`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink)`. Projection implementations emit bounded scoring rows into the sink: emitting more than `MAX_EMITS` rows for one input panics, naming the projection and both counts, and retained state sizes each entity's row list by `MAX_EMITS`; Vec-returning closures are not part of the API. `Out` does not need `Clone`. The fluent projected stream keeps its concrete source type in the return type: `SingleSource` for one projection, `FilteredSource` after a projected filter, diff --git a/crates/solverforge-scoring/src/stream/projected_stream/source.rs b/crates/solverforge-scoring/src/stream/projected_stream/source.rs index 00725d371..7fb9b4ebe 100644 --- a/crates/solverforge-scoring/src/stream/projected_stream/source.rs +++ b/crates/solverforge-scoring/src/stream/projected_stream/source.rs @@ -26,8 +26,9 @@ pub trait Projection: Send + Sync { /// The emitted row type; it need not be `Clone`. type Out: Send + Sync + 'static; /// Maximum number of rows one `project` call emits. Emitting more panics, - /// naming the projection and both counts. Sources combine it (a merge adds - /// both bounds). + /// naming the projection and both counts. Retained scoring state sizes + /// each entity's row list by it, and sources combine it (a merge adds both + /// bounds). const MAX_EMITS: usize; /// Emits the rows for `input` into `sink`. diff --git a/crates/solverforge-scoring/src/stream/projected_stream/source/operator.rs b/crates/solverforge-scoring/src/stream/projected_stream/source/operator.rs index 4b0a123a9..2324fd3b9 100644 --- a/crates/solverforge-scoring/src/stream/projected_stream/source/operator.rs +++ b/crates/solverforge-scoring/src/stream/projected_stream/source/operator.rs @@ -66,7 +66,12 @@ where } let handle = self.rows.insert(StoredRow { coordinate, value }); self.by_coordinate.insert(coordinate, handle); - coordinate.for_each_owner(|owner| self.by_owner.entry(owner).or_default().push(handle)); + coordinate.for_each_owner(|owner| { + self.by_owner + .entry(owner) + .or_insert_with(|| Vec::with_capacity(Src::MAX_EMITS)) + .push(handle); + }); Some(handle) }