Bill Pro plans on active time-series

lib/api_organizations/src/plan.rs

@@ -5,7 +5,7 @@ use bencher_endpoint::{
     CorsResponse, Delete, Endpoint, Get, Patch, Post, ResponseCreated, ResponseDeleted, ResponseOk,
 };
 use bencher_json::{
-    DateTime, MeteredPlanId, OrganizationResourceId,
+    DateTime, OrganizationResourceId, PlanStatus,
     organization::plan::{JsonNewPlan, JsonPlan, JsonUpdatePlan},
 };
 use bencher_rbac::organization::Permission;
@@ -14,7 +14,7 @@ use bencher_schema::{
     context::ApiContext,
     error::{
         BencherResource, bad_request_error, forbidden_error, issue_error, resource_conflict_err,
-        resource_conflict_error, resource_not_found_err,
+        resource_conflict_error, resource_not_found_err, service_unavailable_error,
     },
     model::{
         organization::{
@@ -235,16 +235,9 @@ async fn post_inner(
         .rbac
         .is_allowed_organization(auth_user, Permission::Manage, &query_organization)
         .map_err(forbidden_error)?;
-    // Check to make sure the organization doesn't already have a plan
-    if let Ok(query_plan) =
-        QueryPlan::belonging_to(&query_organization).first::<QueryPlan>(auth_conn!(context))
-    {
-        return Err(resource_conflict_error(
-            BencherResource::Plan,
-            (query_organization, query_plan),
-            "Organization already has a plan",
-        ));
-    }
+    // Block creating a plan when the organization still has an active plan; prune a
+    // stale lapsed metered plan row first so the organization can subscribe again.
+    prune_or_conflict_existing_plan(context, biller, &query_organization).await?;
 
     let JsonNewPlan {
         checkout,
@@ -311,10 +304,6 @@ async fn post_inner(
             .as_ref()
             .parse()
             .map_err(resource_not_found_err!(Plan, subscription_id))?;
-        // Grant the first period's included usage credit so a new metered org has
-        // its credit immediately (best-effort; the daily sweep handles later
-        // periods). A no-op for plans without a base fee.
-        grant_initial_credit(biller, &metered_plan_id).await;
         InsertPlan::metered_plan(write_conn!(context), metered_plan_id, &query_organization)?;
         QueryPlan::belonging_to(&query_organization)
             .first::<QueryPlan>(auth_conn!(context))
@@ -329,6 +318,87 @@ async fn post_inner(
     }
 }
 
+/// Block plan creation when the organization still has a plan that is active, trialing,
+/// or recoverable (a dunning `past_due`/`unpaid`/`incomplete`/`paused` subscription still
+/// exists in Stripe). Only a *terminal* metered subscription (canceled or
+/// incomplete-expired), or one Stripe reports as gone, leaves a stale local plan row safe
+/// to prune so the organization can subscribe again now that the daily reconciliation
+/// sweep is gone; pruning a still-live subscription would orphan it and let the org
+/// create a duplicate (double billing). Licensed (Self-Hosted) plan rows do not lapse on
+/// their own, so they always block.
+async fn prune_or_conflict_existing_plan(
+    context: &ApiContext,
+    biller: &Biller,
+    query_organization: &QueryOrganization,
+) -> Result<(), HttpError> {
+    let Ok(query_plan) =
+        QueryPlan::belonging_to(query_organization).first::<QueryPlan>(auth_conn!(context))
+    else {
+        return Ok(());
+    };
+
+    // A licensed (Self-Hosted) plan carries no metered subscription and always blocks.
+    let Some(metered_plan_id) = &query_plan.metered_plan else {
+        return Err(plan_conflict(query_organization, query_plan));
+    };
+
+    // Resolve the live metered subscription status (`None` => Stripe reports it gone). A
+    // transient Stripe error surfaces as 503 rather than being mistaken for "gone", so we
+    // never prune a subscription that might still be live.
+    let status = biller
+        .metered_plan_status(metered_plan_id)
+        .await
+        .map_err(service_unavailable_error)?;
+
+    match existing_plan_action(status) {
+        ExistingPlan::Conflict => Err(plan_conflict(query_organization, query_plan)),
+        ExistingPlan::Prune => {
+            diesel::delete(schema::plan::table.filter(schema::plan::id.eq(query_plan.id)))
+                .execute(write_conn!(context))
+                .map_err(resource_conflict_err!(Plan, query_plan))?;
+            Ok(())
+        },
+    }
+}
+
+/// The conflict error returned when an organization already has a plan that blocks
+/// creating a new one.
+fn plan_conflict(query_organization: &QueryOrganization, query_plan: QueryPlan) -> HttpError {
+    resource_conflict_error(
+        BencherResource::Plan,
+        (query_organization.clone(), query_plan),
+        "Organization already has a plan",
+    )
+}
+
+/// Whether an organization's existing metered plan row blocks creating a new plan, or is
+/// a stale row to prune. `status` is `None` when Stripe reports the subscription gone
+/// (404), else its live status.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum ExistingPlan {
+    Conflict,
+    Prune,
+}
+
+/// Prune (allow re-subscribe) only when the subscription is gone or *terminal* (canceled
+/// / incomplete-expired). An active, trialing, or recoverable (dunning) subscription
+/// still exists in Stripe and must block, so we never orphan a live subscription and let
+/// the org create a duplicate. Matched exhaustively so a new `PlanStatus` forces a
+/// decision here.
+fn existing_plan_action(status: Option<PlanStatus>) -> ExistingPlan {
+    match status {
+        None | Some(PlanStatus::Canceled | PlanStatus::IncompleteExpired) => ExistingPlan::Prune,
+        Some(
+            PlanStatus::Active
+            | PlanStatus::Trialing
+            | PlanStatus::PastDue
+            | PlanStatus::Unpaid
+            | PlanStatus::Incomplete
+            | PlanStatus::Paused,
+        ) => ExistingPlan::Conflict,
+    }
+}
+
 #[derive(Deserialize, JsonSchema)]
 pub struct OrgPlanQuery {
     pub remote: Option<bool>,
@@ -400,21 +470,6 @@ async fn delete_inner(
     delete_plan_result
 }
 
-/// Best-effort grant of the first period's included usage credit for a new
-/// metered subscription. Idempotent and self-guarding (a no-op for plans without
-/// a base fee); never blocks plan creation.
-async fn grant_initial_credit(biller: &Biller, metered_plan_id: &MeteredPlanId) {
-    let _credit = biller
-        .ensure_period_credit(metered_plan_id)
-        .await
-        .inspect_err(|e| {
-            #[cfg(feature = "sentry")]
-            sentry::capture_error(e);
-            #[cfg(not(feature = "sentry"))]
-            let _ = e;
-        });
-}
-
 async fn delete_plan(
     context: &ApiContext,
     biller: &Biller,
@@ -463,3 +518,36 @@ async fn delete_plan(
 
     Ok(())
 }
+
+#[cfg(test)]
+mod tests {
+    use bencher_json::PlanStatus;
+
+    use super::{ExistingPlan, existing_plan_action};
+
+    #[test]
+    fn existing_plan_action_decides() {
+        // Gone in Stripe (404) or terminal: prune the stale row.
+        assert_eq!(existing_plan_action(None), ExistingPlan::Prune);
+        assert_eq!(
+            existing_plan_action(Some(PlanStatus::Canceled)),
+            ExistingPlan::Prune,
+        );
+        assert_eq!(
+            existing_plan_action(Some(PlanStatus::IncompleteExpired)),
+            ExistingPlan::Prune,
+        );
+        // Active, trialing, or recoverable (dunning) still exists in Stripe: block, so we
+        // never orphan a live subscription and let the org create a duplicate.
+        for status in [
+            PlanStatus::Active,
+            PlanStatus::Trialing,
+            PlanStatus::PastDue,
+            PlanStatus::Unpaid,
+            PlanStatus::Incomplete,
+            PlanStatus::Paused,
+        ] {
+            assert_eq!(existing_plan_action(Some(status)), ExistingPlan::Conflict);
+        }
+    }
+}

lib/api_organizations/src/usage.rs

@@ -219,10 +219,11 @@ fn free_plan_usage(
 
 /// Estimate billable usage for a metered (Bencher Cloud) plan.
 ///
-/// On Pro, only Private Project metrics are metered; Public Project metrics are free
-/// and unlimited. On legacy Team (and metered Enterprise) plans, both Public and
-/// Private Project metrics are metered. This mirrors `PlanKind::check_usage` so the
-/// estimate matches what is actually billed. Bare metal runner minutes are metered
+/// Legacy Team (and metered Enterprise) plans are metered on both Public and Private
+/// Project metrics, so the returned metrics figure matches their bill. Pro now bills on
+/// active series rather than metrics, so for Pro this returns only its Private Project
+/// metrics as an informational figure that no longer matches the bill; surfacing the
+/// active-series count here is a follow-up. Bare metal runner minutes are metered
 /// regardless of visibility and plan level.
 fn metered_plan_usage(
     conn: &mut DbConnection,

lib/bencher_config/src/config_tx.rs

@@ -24,7 +24,7 @@ use bencher_schema::{
     context::RateLimiting,
     model::{
         runner::job::{reprocess_completed_jobs, spawn_heartbeat_timeout},
-        server::{QueryServer, spawn_credit_grants},
+        server::QueryServer,
     },
     write_conn,
 };
@@ -173,22 +173,6 @@ impl ConfigTx {
         #[cfg(feature = "plus")]
         spawn_stats(log, server.app_private()).await?;
 
-        // Bencher Cloud only (a Biller is configured): keep each metered
-        // subscription's monthly usage credit granted and reconcile lapses on a
-        // daily sweep, off the report-ingestion path.
-        #[cfg(feature = "plus")]
-        {
-            let context = server.app_private();
-            if let Some(biller) = context.biller.clone() {
-                spawn_credit_grants(
-                    log.clone(),
-                    context.database.path.clone(),
-                    context.database.busy_timeout,
-                    biller,
-                );
-            }
-        }
-
         Ok(server)
     }
 }

lib/bencher_json/src/organization/plan.rs

@@ -12,6 +12,12 @@ use crate::{BigInt, OrganizationUuid, system::payment::JsonCustomer};
 
 pub const DEFAULT_PRICE_NAME: &str = "default";
 pub const METRICS_METER_NAME: &str = "metrics";
+/// Stripe meter for monthly-active series (distinct testbed x benchmark x measure).
+/// Backs the Pro tiered price (tier 1 flat fee plus per-series step-ups) and is billed
+/// with `last` aggregation: after each report we post the org's cumulative
+/// period-to-date series count, so the final post of the period is the period total and
+/// a missed post self-heals on the next report.
+pub const ACTIVE_SERIES_METER_NAME: &str = "active_series";
 pub const RUNNER_MINUTES_METER_NAME: &str = "runner_minutes";
 
 #[typeshare::typeshare]

lib/bencher_json/src/system/config/plus/cloud/billing.rs

@@ -25,6 +25,8 @@ pub struct JsonProducts {
     // Legacy self-serve paid tier, retained for grandfathered customers.
     pub team: JsonProduct,
     pub enterprise: JsonProduct,
+    // Shared metered metrics product for legacy Team/Enterprise plans. Pro bills on
+    // its own tiered active-series price (its `metered` map), not on this meter.
     pub metrics: JsonProduct,
     pub bare_metal: JsonProduct,
 }
@@ -37,54 +39,43 @@ pub struct JsonProduct {
     pub metered: HashMap<String, String>,
     #[serde(default)]
     pub licensed: HashMap<String, String>,
-    // Flat recurring base prices.
-    // Empty for products with no flat base fee.
-    #[serde(default)]
-    pub base: HashMap<String, String>,
-    #[serde(default)]
-    pub trial_coupon: Option<String>,
 }
 
 #[cfg(test)]
 mod tests {
     use super::{JsonProduct, JsonProducts};
 
-    // Existing team/enterprise/bare_metal config has no `base` key; it must still
-    // deserialize, defaulting `base` to an empty map (backward compatibility).
+    // A product given only an `id` still deserializes, defaulting `metered` and
+    // `licensed` to empty maps (backward compatibility).
     #[test]
-    fn product_without_base_defaults_empty() {
-        let json =
-            r#"{"id":"prod_x","metered":{"default":"price_m"},"licensed":{"default":"price_l"}}"#;
+    fn product_without_pricing_defaults_empty() {
+        let json = r#"{"id":"prod_x"}"#;
         let product: JsonProduct = serde_json::from_str(json).unwrap();
-        assert!(product.base.is_empty());
-        assert!(product.trial_coupon.is_none());
-        assert_eq!(
-            product.metered.get("default").map(String::as_str),
-            Some("price_m"),
-        );
+        assert!(product.metered.is_empty());
+        assert!(product.licensed.is_empty());
     }
 
+    // Pro carries a single tiered metered price (base fee plus per-series step-ups);
+    // the shared metrics product carries the legacy Team/Enterprise metered price.
     #[test]
-    fn products_with_pro_base_and_trial_coupon() {
+    fn products_with_pro_tiered_price() {
         let json = r#"{
-            "pro": {"id":"p","base":{"default":"price_b"},"trial_coupon":"coupon_x"},
+            "pro": {"id":"p","metered":{"default":"price_pro_tiered"}},
             "team": {"id":"t","metered":{},"licensed":{}},
             "enterprise": {"id":"e","metered":{},"licensed":{}},
             "metrics": {"id":"m","metered":{"default":"price_mm"},"licensed":{"default":"price_ml"}},
             "bare_metal": {"id":"bm","metered":{},"licensed":{}}
         }"#;
         let products: JsonProducts = serde_json::from_str(json).unwrap();
         assert_eq!(
-            products.pro.base.get("default").map(String::as_str),
-            Some("price_b"),
+            products.pro.metered.get("default").map(String::as_str),
+            Some("price_pro_tiered"),
         );
-        assert_eq!(products.pro.trial_coupon.as_deref(), Some("coupon_x"));
-        assert!(products.pro.metered.is_empty());
+        assert!(products.pro.licensed.is_empty());
         assert_eq!(
             products.metrics.metered.get("default").map(String::as_str),
             Some("price_mm"),
         );
-        assert!(products.team.base.is_empty());
-        assert!(products.team.trial_coupon.is_none());
+        assert!(products.team.metered.is_empty());
     }
 }

lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS series_last_seen;

lib/bencher_schema/migrations/2026-06-24-120000_series_last_seen/up.sql

@@ -0,0 +1,53 @@
+-- Cache of each monitored series' most recent activity, for value-based billing on
+-- monthly-active series (a series is a distinct testbed x benchmark x measure) and
+-- for telemetry. `last_seen` is the greatest `report.created` (the server-side
+-- ingestion time, not the user-supplied `end_time`) ever recorded for the series, kept
+-- monotonic on ingest so reprocessing an older report cannot lower it.
+--
+-- `organization_id` and `project_id` are denormalized so a billing read is a single
+-- index range scan over (organization_id, last_seen), without joining through the
+-- entity tables. The series key (testbed_id, benchmark_id, measure_id) is globally
+-- unique on its own because each entity is per-project and id-unique, so it is the
+-- primary key.
+--
+-- Hard-deleting a testbed, benchmark, or measure cascades its series rows away and can
+-- lower the current period's active-series count; an accepted exception to monotonic
+-- billing, since entity deletion requires first deleting all of its reports.
+CREATE TABLE series_last_seen (
+    organization_id INTEGER NOT NULL,
+    project_id INTEGER NOT NULL,
+    testbed_id INTEGER NOT NULL,
+    benchmark_id INTEGER NOT NULL,
+    measure_id INTEGER NOT NULL,
+    last_seen BIGINT NOT NULL,
+    PRIMARY KEY (testbed_id, benchmark_id, measure_id),
+    FOREIGN KEY (organization_id) REFERENCES organization (id) ON DELETE CASCADE,
+    FOREIGN KEY (project_id) REFERENCES project (id) ON DELETE CASCADE,
+    FOREIGN KEY (testbed_id) REFERENCES testbed (id) ON DELETE CASCADE,
+    FOREIGN KEY (benchmark_id) REFERENCES benchmark (id) ON DELETE CASCADE,
+    FOREIGN KEY (measure_id) REFERENCES measure (id) ON DELETE CASCADE
+);
+
+CREATE INDEX index_series_last_seen_org_last_seen
+    ON series_last_seen (organization_id, last_seen);
+
+-- Index the remaining cascaded foreign keys so parent deletes do not full-scan this
+-- table. `testbed_id` is already the primary key's prefix and `organization_id` is
+-- covered by the billing index above.
+CREATE INDEX index_series_last_seen_project ON series_last_seen (project_id);
+CREATE INDEX index_series_last_seen_benchmark ON series_last_seen (benchmark_id);
+CREATE INDEX index_series_last_seen_measure ON series_last_seen (measure_id);
+
+-- Backfill from existing metrics: one row per distinct series with the latest
+-- `report.created` it has produced. Runs once, against the table just created empty
+-- above and before the server serves any request, so it needs no conflict handling
+-- (like `metric_count_by_report`). Correct on day one, so the cache equals a fresh
+-- COUNT(DISTINCT testbed, benchmark, measure) over the same metrics.
+INSERT INTO series_last_seen (organization_id, project_id, testbed_id, benchmark_id, measure_id, last_seen)
+SELECT p.organization_id, p.id, r.testbed_id, rb.benchmark_id, m.measure_id, MAX(r.created)
+FROM metric m
+INNER JOIN report_benchmark rb ON m.report_benchmark_id = rb.id
+INNER JOIN report r ON rb.report_id = r.id
+INNER JOIN benchmark b ON rb.benchmark_id = b.id
+INNER JOIN project p ON b.project_id = p.id
+GROUP BY r.testbed_id, rb.benchmark_id, m.measure_id;