feat: add volume commitment feature and analytics

config/development.toml

@@ -119,6 +119,21 @@ aws_region = ""
 # `openssl rand -hex 32`. On rotation add e.g. `v2 = "..."` and set current = "v2" above.
 v1 = "0000000000000000000000000000000000000000000000000000000000000000"
 
+[volume_commitment]
+# Expose the scheduler on its own port in this process. Prod: only the deployment that owns it.
+enabled = true
+# How often a merchant's plan is rebuilt — the only scheduled job. prod default: 3600 (1 h)
+default_forecast_interval_secs = 60
+# Where the scheduler calls when a forecast comes due. Normally this same process's [server].
+main_server_url = "http://127.0.0.1:8080"
+# How often the scheduler wakes. Keep well under the shortest cadence above.
+tick_secs = 10
+
+# The scheduler's own listener. GET /schedule shows the timing; GET /health is a liveness probe.
+[volume_commitment.server]
+host = "127.0.0.1"
+port = 9095
+
 [hypersense]
 base_url = "https://eu.hyperswitch.io/cost-observability/router"
 username = "hypersense_user"

src/analytics/clickhouse/common.rs

@@ -43,6 +43,11 @@ pub const PAYMENT_AUDIT_DYNAMIC_FLOW_TYPES: &[FlowType] = &[
     FlowType::UpdateScoreLegacyError,
 ];
 
+/// Payment amount on a decide event, inside the `details` JSON. Shared by every metric that
+/// sums volume so the request shape has one place to move.
+pub const PAYMENT_AMOUNT_EXPR: &str =
+    "JSONExtractFloat(assumeNotNull(details), 'request', 'paymentInfo', 'amount')";
+
 pub async fn fetch_all<T>(query: Query) -> Result<Vec<T>, ApiError>
 where
     T: Row + for<'de> Deserialize<'de>,

src/analytics/clickhouse/metrics/cost_savings.rs

@@ -17,8 +17,7 @@ const COST_SAVED_BPS_EXPR: &str =
     "JSONExtractFloat(assumeNotNull(details), 'response', 'multi_objective_info', 'costSavedBps')";
 const MO_OUTCOME_EXPR: &str =
     "JSONExtractString(assumeNotNull(details), 'response', 'multi_objective_info', 'outcome')";
-const PAYMENT_AMOUNT_EXPR: &str =
-    "JSONExtractFloat(assumeNotNull(details), 'request', 'paymentInfo', 'amount')";
+use crate::analytics::clickhouse::common::PAYMENT_AMOUNT_EXPR;
 // Currency lives in the request JSON, not the top-level `currency` column (which is NULL for
 // /decide-gateway events).
 const PAYMENT_CURRENCY_EXPR: &str =

src/analytics/flow.rs

@@ -91,6 +91,8 @@ pub enum FlowType {
     RoutingCreateVolumeContract,
     RoutingEvaluateVolumeContract,
     AutopilotCalibration,
+    /// A volume-commitment forecast run: the plan snapshot, including eliminations.
+    VolumeCommitmentForecast,
 }
 
 impl FlowType {
@@ -160,6 +162,7 @@ impl FlowType {
             Self::RoutingCreateVolumeContract => "routing_create_volume_contract",
             Self::RoutingEvaluateVolumeContract => "routing_evaluate_volume_contract",
             Self::AutopilotCalibration => "autopilot_calibration",
+            Self::VolumeCommitmentForecast => "volume_commitment_forecast",
         }
     }
 }

src/app.rs

@@ -547,6 +547,30 @@ where
             "/merchant-account/:merchant-id/cost-coverage",
             get(routes::cost_coverage::get_cost_coverage),
         )
+        .route(
+            "/merchant-account/:merchant-id/volume-commitment",
+            get(routes::volume_commitment::get_volume_commitment),
+        )
+        .route(
+            "/merchant-account/:merchant-id/volume-commitment/series",
+            get(routes::volume_commitment::get_series),
+        )
+        .route(
+            "/merchant-account/:merchant-id/volume-commitment/audit",
+            get(routes::volume_commitment::get_audit),
+        )
+        .route(
+            "/merchant-account/:merchant-id/volume-commitment/impact",
+            get(routes::volume_commitment::get_impact),
+        )
+        // Called by the volume-commitment scheduler when a merchant's forecast comes due. Behind
+        // the same auth as any other write, and the handler narrows further: the admin secret
+        // (which the scheduler presents) may run anything, a session or api key only its own
+        // merchant, and the every-merchant sweep is admin-only.
+        .route(
+            "/volume-commitment/run-forecast",
+            post(routes::volume_commitment::run_forecast),
+        )
         .route(
             "/merchant-account/:merchant-id/cost-ingestions",
             get(routes::report_upload::list_ingestions),

src/bin/open_router.rs

@@ -1,6 +1,7 @@
 #![allow(clippy::unwrap_in_result)]
 
 use masking::PeekInterface;
+use open_router::decider::gatewaydecider::volume_commitment;
 use open_router::{logger, tenant::GlobalAppState};
 
 #[allow(clippy::expect_used)]
@@ -32,7 +33,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await
         .expect("Failed while configuring global application state");
 
-    // Run both servers concurrently using tokio::spawn
+    // Volume commitment routing: install the shared dependencies before anything binds, so the
+    // routing path and the controller's own port see the same plan and the same counters.
+    let volume_commitment_admin_secret = global_config.admin_secret.secret.peek().clone();
+    let volume_commitment_deps = std::sync::Arc::new(
+        volume_commitment::build_deps(
+            &global_config.volume_commitment,
+            &global_config.analytics.clickhouse,
+        )
+        .await,
+    );
+    volume_commitment::init_deps(volume_commitment_deps.clone());
+
+    // Run the servers concurrently using tokio::spawn
     let main_server_handle = tokio::spawn(async move {
         open_router::app::server_builder(global_app_state)
             .await
@@ -45,8 +58,23 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             .expect("Failed while building the metrics server")
     });
 
-    // Wait for both servers to complete (they should run indefinitely)
-    tokio::try_join!(main_server_handle, metrics_server_handle)?;
+    // The pacing scheduler, on a port of its own. It owns the clock: when a merchant's forecast
+    // comes due it calls the main server, which does the work.
+    let volume_commitment_server_handle = tokio::spawn(async move {
+        volume_commitment::server::volume_commitment_server_builder(
+            volume_commitment_deps,
+            volume_commitment_admin_secret,
+        )
+        .await
+        .expect("Failed while building the volume commitment scheduler server")
+    });
+
+    // Wait for the servers to complete (they should run indefinitely)
+    tokio::try_join!(
+        main_server_handle,
+        metrics_server_handle,
+        volume_commitment_server_handle
+    )?;
 
     Ok(())
 }

src/config.rs

@@ -65,6 +65,8 @@ pub struct GlobalConfig {
     #[serde(default)]
     pub cost_ingestion: CostIngestionConfig,
     #[serde(default)]
+    pub volume_commitment: VolumeCommitmentConfig,
+    #[serde(default)]
     pub sr_auto_calibration: SrAutoCalibrationConfig,
     #[serde(default)]
     pub card_info_service: CardInfoServiceConfig,
@@ -394,9 +396,43 @@ pub struct TenantConfig {
     pub cache_config: CacheConfig,
     pub hypersense: HypersenseConfig,
     pub cost_ingestion: CostIngestionConfig,
+    pub volume_commitment: VolumeCommitmentConfig,
     pub card_info_service: CardInfoServiceConfig,
 }
 
+/// Deployment-level mechanics for volume commitment routing; everything per-merchant arrives
+/// from the contract source instead. Nothing here is a contract term.
+#[derive(Clone, Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct VolumeCommitmentConfig {
+    /// Expose the scheduler in this process, on its own port.
+    pub enabled: bool,
+    /// Where that port binds — its own listener, exactly as `[metrics]` is.
+    pub server: Server,
+    /// Where the scheduler sends run calls; normally this same process's `[server]`.
+    pub main_server_url: String,
+    /// How often the scheduler wakes. Bounds how late a due job fires; keep well under the
+    /// shortest cadence.
+    pub tick_secs: u64,
+    /// How often a merchant's plan is rebuilt — the only scheduled job. Per-merchant overridable.
+    pub default_forecast_interval_secs: u64,
+}
+
+impl Default for VolumeCommitmentConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            server: Server {
+                host: "127.0.0.1".to_string(),
+                port: 9095,
+            },
+            main_server_url: "http://127.0.0.1:8080".to_string(),
+            tick_secs: 30,
+            default_forecast_interval_secs: 3600,
+        }
+    }
+}
+
 /// Configuration for the in-house cost-estimation settlement ingestion pipeline
 /// (see `scratch/inhouse-cost-architecture.md` §7).
 #[derive(Clone, Debug, serde::Deserialize)]
@@ -553,6 +589,7 @@ impl TenantConfig {
             cache_config: global_config.cache_config.clone(),
             hypersense: global_config.hypersense.clone(),
             cost_ingestion: global_config.cost_ingestion.clone(),
+            volume_commitment: global_config.volume_commitment.clone(),
             card_info_service: global_config.card_info_service.clone(),
         }
     }

src/decider/gatewaydecider.rs

@@ -12,3 +12,4 @@ pub mod runner;
 pub mod types;
 pub mod utils;
 pub mod validators;
+pub mod volume_commitment;

src/decider/gatewaydecider/ab_test/interceptor.rs

@@ -162,6 +162,7 @@ pub async fn intercept(dreq: &DomainDeciderRequestForApiCallV2) -> AbTestInterce
                     is_rust_based_decider: true,
                     latency: None,
                     multi_objective_info: None,
+                    volume_steer_info: None,
                 }),
                 experiment_id,
                 variant_arm: arm.to_string(),

src/decider/gatewaydecider/flow_new.rs

@@ -20,6 +20,7 @@ use super::runner::handle_fallback_logic;
 use super::types as T;
 use super::types::PriorityLogicFailure;
 use super::utils as Utils;
+use super::volume_commitment;
 // use optics_core::{preview, review};
 use crate::decider::gatewaydecider::constants as C;
 use crate::feedback::constants::kvRedis;
@@ -457,6 +458,7 @@ pub async fn run_decider_flow(
                     is_rust_based_decider: true,
                     latency: Some(cpu_time),
                     multi_objective_info: None,
+                    volume_steer_info: None,
                 })
             } else {
                 decider_flow
@@ -719,6 +721,39 @@ pub async fn run_decider_flow(
                         decider_flow.writer.multi_objective_info = Some(outcome.info);
                     }
 
+                    // Volume-commitment nudge runs last on its own flag; fails open when flag,
+                    // deps or plan is absent. Under hedging the flag is not even read.
+                    let volume_commitment_on = !hedging_on
+                        && is_feature_enabled(
+                            volume_commitment::FEATURE_FLAG.to_string(),
+                            merchant_id_text.clone(),
+                            kvRedis(),
+                        )
+                        .await;
+                    if volume_commitment_on {
+                        if let Some(vc_deps) = volume_commitment::deps() {
+                            if let Some(plan) = vc_deps.state.load_plan(&merchant_id_text).await {
+                                // Sampling, not counting: the forecast already decided each PSP's
+                                // share of the eligible flow, so this payment only has to roll.
+                                let mut roll = || rand::random::<f64>();
+                                let outcome = volume_commitment::nudge::choose(
+                                    &currentGatewayScoreMap,
+                                    &plan,
+                                    chrono::Utc::now(),
+                                    &mut roll,
+                                );
+                                // Only an actual diversion relabels the approach.
+                                if let Some(chosen) = outcome.chosen {
+                                    decidedGateway = Some(chosen);
+                                    cost_fallbacks_override = Some(outcome.fallbacks);
+                                    decider_flow.writer.gwDeciderApproach =
+                                        T::GatewayDeciderApproach::SrSelectionVolumeCommitment;
+                                }
+                                decider_flow.writer.volume_steer_info = Some(outcome.info);
+                            }
+                        }
+                    }
+
                     let stateBindings = (
                         decider_flow.writer.srElminiationApproachInfo.clone(),
                         decider_flow.writer.isOptimizedBasedOnSRMetricEnabled,
@@ -864,6 +899,7 @@ pub async fn run_decider_flow(
                                     .writer
                                     .multi_objective_info
                                     .clone(),
+                                volume_steer_info: decider_flow.writer.volume_steer_info.clone(),
                             })
                         }
                         None => Err((

src/decider/gatewaydecider/flows.rs

@@ -404,6 +404,7 @@ pub async fn run_decider_flow(
                     is_rust_based_decider: deciderParams.dpShouldConsumeResult.unwrap_or(false),
                     latency: None,
                     multi_objective_info: None,
+                    volume_steer_info: None,
                 })
             } else {
                 decider_flow
@@ -734,6 +735,7 @@ pub async fn run_decider_flow(
                                     .unwrap_or(false),
                                 latency: None,
                                 multi_objective_info: None,
+                                volume_steer_info: None,
                             })
                         }
                         None => Err((

src/decider/gatewaydecider/types.rs

@@ -477,6 +477,8 @@ pub struct DeciderState {
     /// overrides the merchant's elimination rule. Absent for control arm and non-tuning experiments.
     pub ab_test_sr_override: Option<crate::euclid::types::SrConfigOverride>,
     pub multi_objective_info: Option<super::multi_objective::MultiObjectiveInfo>,
+    /// Why the volume-commitment nudge did or did not move this payment.
+    pub volume_steer_info: Option<super::volume_commitment::VolumeSteerInfo>,
 }
 
 pub fn initial_decider_state(date_created: String) -> DeciderState {
@@ -522,6 +524,7 @@ pub fn initial_decider_state(date_created: String) -> DeciderState {
         sr_v3_hedging_percent: None,
         gateway_reference_id: None,
         multi_objective_info: None,
+        volume_steer_info: None,
         gateway_scoring_data: GatewayScoringData {
             merchantId: String::new(),
             paymentMethodType: String::new(),
@@ -655,6 +658,9 @@ pub enum GatewayDeciderApproach {
     NtwBasedRouting,
     AbTestStaticAlgorithm,
     SrSelectionMultiObjective,
+    /// A volume-contract nudge moved the payment off the SR head — the volume-driven sibling of
+    /// [`Self::SrSelectionMultiObjective`].
+    SrSelectionVolumeCommitment,
 }
 
 #[derive(Debug, Serialize, Deserialize, PartialEq)]
@@ -1378,6 +1384,7 @@ pub struct DecidedGateway {
     pub is_rust_based_decider: bool,
     pub latency: Option<u64>,
     pub multi_objective_info: Option<super::multi_objective::MultiObjectiveInfo>,
+    pub volume_steer_info: Option<super::volume_commitment::VolumeSteerInfo>,
 }
 
 #[derive(Debug, Serialize, Clone, Deserialize)]
@@ -1593,6 +1600,9 @@ impl fmt::Display for GatewayDeciderApproach {
             Self::SrSelectionMultiObjective => {
                 write!(f, "SR_SELECTION_MULTI_OBJECTIVE")
             }
+            Self::SrSelectionVolumeCommitment => {
+                write!(f, "SR_SELECTION_VOLUME_COMMITMENT")
+            }
         }
     }
 }