Server P10b-ii: shared checkpoint results (G27); Live pins its inputs once per activation (D56)

Cargo.toml

@@ -72,7 +72,7 @@ convex-server = { path = "crates/convex-server" }
 chrono = { version = "0.4", features = ["serde"] }
 rust_decimal = { version = "1.42", features = ["serde", "serde-float", "maths", "serde-with-str"] }
 rust_decimal_macros = "1.40"
-serde = { version = "1.0", features = ["derive"] }
+serde = { version = "1.0", features = ["derive", "rc"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }
 thiserror = "2.0"
 anyhow = "1.0"

crates/convex-engine/CONFIGURATION.md

@@ -234,7 +234,9 @@ activation reports `restart_required: true` unless the runtime accepted that exa
 revision. A `hot` activation of a restart-required change is
 refused with `CONFIGURATION_RESTART_REQUIRED`. Its outcome includes the committed record and optional accepted
 generation/activation revision. A full queue or application failure leaves a
-committed selection for reconciliation; it does not guess a rollback. A later
+committed selection for reconciliation; it does not guess a rollback. A Live
+runtime's quotes do not read the active pointer, so a selection it did not accept
+waits for a retry of the same activation or a restart. A later
 concurrent activation can supersede the command before runtime acceptance.
 Readiness closes while a notified desired revision has not been accepted.
 New work freezes one coherent selection into its checkpoint before receipt;

crates/convex-engine/RUNTIME.md

@@ -70,12 +70,24 @@ instrument publishes a failed result with no retained last-good value.
 
 Live ticks conflate quotes. After the actor takes one quote, it also admits the
 quotes already queued behind it, up to 1,024 per tick in all (the HTTP ingress
-queue's capacity, so a full queue fits in one tick), then assembles and values once. Each quote is still validated and admitted in order and
-advances the generation and cutoff, so a tick's results carry the generation of its
-newest quote. A quote superseded within one tick gets no result of its own. Under
-load a quote waits for at most the running tick and its own, not for every queued
-tick before it. A tick still values every instrument. Replay and Captured value
-every event.
+queue's capacity, so a full queue fits in one tick), then assembles and values
+once. Each quote is still validated and admitted in order and advances the
+generation and cutoff, so a tick's results carry the generation of its newest
+quote. A quote superseded within one tick gets no result of its own. Under load a
+quote waits for at most the running tick and its own, not for every queued tick
+before it. A tick still values every instrument. Replay and Captured value every
+event.
+
+A Live runtime resolves its pricing configuration and pinned references, with
+their issuer, calendar, convention and rate-index records, once per accepted
+activation, at the first snapshot that needs them, and reuses them at every tick.
+All are exact pinned versions, so none changes while that selection holds, and each
+snapshot still checks them at its own cutoff. Quotes, fixings and stored curve
+revisions are read at every snapshot's cutoff. A Live quote does not read the active
+configuration: the runtime keeps the selection it accepted until the engine's
+activation refresh or a restart applies another. An activation committed without
+the engine therefore waits for one of those. Captured and Replay read the active
+selection at each event.
 
 A Live plan that pins no curve data reads its required curves from the engine's
 curve store each time it builds a snapshot, at that snapshot's cutoff. A re-marked
@@ -188,7 +200,8 @@ The additive `pricing_runtime_v1` redb table contains one CAS checkpoint per run
 It is covered by the same database backup as authoritative inputs. Immutable
 snapshot files and result archives must be retained/backed up with that database.
 The application schema remains version 1; checkpoint records have their own schema.
-Older code does not process the new checkpoint/outbox and must not run its old
+In memory a checkpoint shares its results and outbox, so copying it costs the same
+at any width; they are stored as plain arrays. Older code does not process the new checkpoint/outbox and must not run its old
 pricing lifecycle against this state. Keep a closed pre-upgrade backup for rollback;
 forward recovery uses the same plan and verified assets. A changed plan digest,
 corrupt checkpoint, missing last-good snapshot or conflicting replay prefix refuses

crates/convex-engine/src/runtime/actor.rs

@@ -11,6 +11,8 @@ struct LiveState {
     quotes: crate::snapshot::IngestedQuotes,
     last_receive_ms: Option<i64>,
     reserved: RuntimeCheckpoint,
+    /// The selection's configuration and references, resolved once per activation (D56).
+    pinned: Option<crate::snapshot::PinnedInputs>,
 }
 pub(super) struct Actor {
     services: RuntimeServices,
@@ -21,7 +23,7 @@ pub(super) struct Actor {
     checkpoint: RuntimeCheckpoint,
     live: Option<StreamReceiver<QuoteEvent>>,
     health: Arc<parking_lot::RwLock<RuntimeHealth>>,
-    results: Arc<parking_lot::RwLock<Vec<PublishedValuation>>>,
+    results: Arc<parking_lot::RwLock<Arc<[PublishedValuation]>>>,
     updates: broadcast::Sender<PublishedValuation>,
     live_source: Option<Arc<dyn convex_ports::market_data::QuoteSource>>,
     data_deadline: Option<tokio::time::Instant>,
@@ -39,7 +41,7 @@ impl Actor {
         services: RuntimeServices,
         engine: &crate::PricingEngine,
         health: Arc<parking_lot::RwLock<RuntimeHealth>>,
-        results: Arc<parking_lot::RwLock<Vec<PublishedValuation>>>,
+        results: Arc<parking_lot::RwLock<Arc<[PublishedValuation]>>>,
         updates: broadcast::Sender<PublishedValuation>,
     ) -> Result<Self, RuntimeError> {
         let digest = services.plan.digest()?;
@@ -73,8 +75,8 @@ impl Actor {
             snapshot_id: None,
             pending: true,
             result_revision: 0,
-            results: vec![],
-            outbox: vec![],
+            results: Arc::default(),
+            outbox: Arc::default(),
             delivery_lost: false,
         });
         checkpoint.validate()?;
@@ -158,8 +160,8 @@ impl Actor {
             if !absent {
                 // Nothing stored describes the market. Recalculate from no quotes, above the ceilings.
                 checkpoint.events.clear();
-                checkpoint.results.clear();
-                checkpoint.outbox.clear();
+                checkpoint.results = Arc::default();
+                checkpoint.outbox = Arc::default();
                 checkpoint.snapshot_id = None;
                 checkpoint.pending = true;
                 checkpoint.generation = Version::new(
@@ -175,6 +177,7 @@ impl Actor {
                 quotes: crate::snapshot::IngestedQuotes::new()?,
                 last_receive_ms: None,
                 reserved,
+                pinned: None,
             })
         } else {
             None
@@ -367,8 +370,22 @@ impl Actor {
             ))
             .await?;
         }
-        let snapshot = match &self.live_state {
-            Some(live) => bounded(self.assembler.resolve_ingested(&request, &live.quotes)).await?,
+        let snapshot = match self.live_state.as_mut() {
+            Some(live) => {
+                let pinned = match live.pinned.take() {
+                    Some(pinned) if pinned.serves(&request) => pinned,
+                    // A refresh or restart selects another activation and re-resolves.
+                    _ => bounded(self.assembler.pin(&request)).await?,
+                };
+                let snapshot = bounded(self.assembler.resolve_ingested(
+                    &request,
+                    &pinned,
+                    &live.quotes,
+                ))
+                .await;
+                live.pinned = Some(pinned);
+                snapshot?
+            }
             None => bounded(self.assembler.assemble(&request, &self.checkpoint.events)).await?,
         };
         Ok(Arc::new(snapshot))
@@ -389,7 +406,7 @@ impl Actor {
             }
         }
         let mut retained = std::collections::BTreeMap::new();
-        for record in &self.checkpoint.results {
+        for record in self.checkpoint.results.iter() {
             if let Some(good) = &record.last_good {
                 if good.provenance().snapshot_id != *snapshot.id() {
                     let old = bounded(self.services.snapshots.get(&good.provenance().snapshot_id))
@@ -466,7 +483,14 @@ impl Actor {
             return Err(RuntimeError::CapacityExceeded);
         }
         let mut next = self.checkpoint.clone();
-        let selected = self.current_configuration().await?;
+        // Live keeps the selection it holds until a refresh or restart applies another
+        // (D56); the other modes read the active selection at each event.
+        let reread = self.live_state.is_none();
+        let selected = if reread {
+            self.current_configuration().await?
+        } else {
+            None
+        };
         if let Some(selected) = &selected {
             selected.capture(as_of, known_at, next.generation)?;
             next.configuration_activation = Some(selected.activation.clone());
@@ -499,7 +523,9 @@ impl Actor {
             live.quotes
                 .ingest(event, as_of.max(known_at).datetime().timestamp_millis());
         }
-        self.selected = selected;
+        if reread {
+            self.selected = selected;
+        }
         {
             let mut health = self.health.write();
             health.ready = false;
@@ -548,7 +574,7 @@ impl Actor {
         )
         .await;
         crate::telemetry::observe(crate::telemetry::CALCULATE_SECONDS, started);
-        let produced = produced?;
+        let produced: Arc<[PublishedValuation]> = produced?.into();
         let mut next = self.checkpoint.clone();
         next.result_revision = produced.last().map_or(next.result_revision, |p| p.revision);
         next.results = produced.clone();
@@ -561,7 +587,7 @@ impl Actor {
     }
     async fn flush_outbox(&mut self) -> Result<(), RuntimeError> {
         // Whole bounded batch retains its original delivery keys on retry; consumers deduplicate.
-        for record in &self.checkpoint.outbox {
+        for record in self.checkpoint.outbox.iter() {
             let started = std::time::Instant::now();
             let published =
                 tokio::time::timeout(IO_BUDGET, self.services.publisher.publish(record)).await;
@@ -573,11 +599,11 @@ impl Actor {
         if !self.checkpoint.outbox.is_empty() {
             let delivered = self.checkpoint.outbox.clone();
             let mut next = self.checkpoint.clone();
-            next.outbox.clear();
+            next.outbox = Arc::default();
             self.save(next).await?;
             self.refresh_health();
-            for record in delivered {
-                let _ = self.updates.send(record);
+            for record in delivered.iter() {
+                let _ = self.updates.send(record.clone());
             }
         }
         Ok(())
@@ -636,13 +662,14 @@ impl Actor {
         let mut next = self.checkpoint.clone();
         next.delivery_lost = true;
         let mut records = Vec::with_capacity(next.results.len());
-        for prior in &next.results {
+        for prior in next.results.iter() {
             next.result_revision = next
                 .result_revision
                 .checked_add(1)
                 .ok_or(RuntimeError::CapacityExceeded)?;
             records.push(worker::lost_record(prior, next.result_revision)?);
         }
+        let records: Arc<[PublishedValuation]> = records.into();
         next.results = records.clone();
         next.outbox = records;
         self.save(next).await?;
@@ -1033,8 +1060,8 @@ impl Actor {
 fn reservation(next: &RuntimeCheckpoint) -> Result<RuntimeCheckpoint, RuntimeError> {
     let mut image = next.clone();
     image.events.clear();
-    image.results.clear();
-    image.outbox.clear();
+    image.results = Arc::default();
+    image.outbox = Arc::default();
     image.snapshot_id = None;
     image.pending = true;
     image.result_revision = next

crates/convex-engine/src/runtime/mod.rs

@@ -309,7 +309,7 @@ struct Lifecycle {
 pub(crate) struct RuntimeController {
     lifecycle: Mutex<Lifecycle>,
     health: Arc<parking_lot::RwLock<RuntimeHealth>>,
-    results: Arc<parking_lot::RwLock<Vec<PublishedValuation>>>,
+    results: Arc<parking_lot::RwLock<Arc<[PublishedValuation]>>>,
     updates: broadcast::Sender<PublishedValuation>,
     demo: bool,
     stopped: watch::Sender<bool>,
@@ -336,7 +336,7 @@ impl RuntimeController {
                 interactive_jobs: Vec::new(),
             }),
             health: Arc::new(parking_lot::RwLock::new(RuntimeHealth::default())),
-            results: Arc::new(parking_lot::RwLock::new(vec![])),
+            results: Arc::new(parking_lot::RwLock::new(Arc::default())),
             updates,
             demo,
             stopped,
@@ -416,7 +416,7 @@ impl RuntimeController {
         self.health.read().clone()
     }
     pub(crate) fn results(&self) -> Vec<PublishedValuation> {
-        self.results.read().clone()
+        self.results.read().to_vec()
     }
     pub(crate) fn settlement_date(&self) -> Option<BusinessDate> {
         self.settlement_date

crates/convex-engine/src/snapshot.rs

@@ -3,6 +3,7 @@
 
 use convex_core::contracts::*;
 use convex_ports::{
+    configuration::ConfigurationPin,
     fixing::FixingRepository,
     quote_event::{QuoteEvent, QuoteField, QuoteKey},
     reference_master::{ReferenceLookup, ResolvedBondReference},
@@ -222,18 +223,42 @@ impl SnapshotAssembler {
         }
         Ok(references)
     }
-    /// Resolve over quote state the caller already ingested, without publishing.
-    /// Live mode keeps that state in memory; market data is not stored.
+    /// Resolve the exact configuration and the pinned references `request` selects, at its
+    /// cutoff, for reuse by later snapshots of the same selection (D56).
+    pub(crate) async fn pin(&self, request: &SnapshotAssembly) -> Result<PinnedInputs, TraitError> {
+        validate_request(request, &[])?;
+        let configuration = self.configuration(request).await?;
+        let references = self.references(request, &configuration).await?;
+        Ok(PinnedInputs {
+            activation: activation_revision(request),
+            configuration_pin: request.configuration.clone(),
+            instruments: request.instruments.clone(),
+            configuration,
+            references,
+        })
+    }
+    /// Resolve over quote state the caller already ingested and inputs it pinned, without
+    /// publishing. Live mode keeps both in memory; market data is not stored. Fixings are
+    /// read at this request's cutoff.
     pub(crate) async fn resolve_ingested(
         &self,
         request: &SnapshotAssembly,
+        pinned: &PinnedInputs,
         quotes: &IngestedQuotes,
     ) -> Result<MarketSnapshot, TraitError> {
         validate_request(request, &[])?;
-        let configuration = self.configuration(request).await?;
-        let references = self.references(request, &configuration).await?;
-        self.build(request, configuration, references, quotes.quotes(request))
-            .await
+        if !pinned.serves(request) {
+            return Err(TraitError::Conflict(
+                "pinned inputs belong to another selection",
+            ));
+        }
+        self.build(
+            request,
+            pinned.configuration.clone(),
+            pinned.references.clone(),
+            quotes.quotes(request),
+        )
+        .await
     }
     async fn build(
         &self,
@@ -316,6 +341,34 @@ impl IngestedQuotes {
     }
 }
 
+/// The pricing configuration and pinned references of one selection, resolved once at
+/// the cutoff of its first snapshot (D56). Every record is an exact pinned version, so
+/// none changes while the selection holds; each snapshot still checks them at its own
+/// cutoff (`MarketSnapshot::new`).
+pub(crate) struct PinnedInputs {
+    activation: Option<Version>,
+    configuration_pin: ConfigurationPin,
+    instruments: Vec<InstrumentPin>,
+    configuration: convex_ports::storage::BondPricingConfig,
+    references: Vec<ResolvedBondReference>,
+}
+impl PinnedInputs {
+    /// Whether these inputs were resolved for `request`'s activation, configuration and
+    /// instrument pins.
+    pub(crate) fn serves(&self, request: &SnapshotAssembly) -> bool {
+        self.activation == activation_revision(request)
+            && self.configuration_pin == request.configuration
+            && self.instruments == request.instruments
+    }
+}
+
+fn activation_revision(request: &SnapshotAssembly) -> Option<Version> {
+    request
+        .configuration_capture
+        .as_ref()
+        .map(|capture| capture.activation.revision)
+}
+
 fn capture_ms(request: &SnapshotAssembly) -> i64 {
     request
         .as_of