Replace compute-billing push with the pull API

CLAUDE.md

@@ -78,7 +78,7 @@ Key CLI flags for control-plane mode:
   - Config store: `--config-store`, `--config-poll-interval`, `--internal-secret`
   - K8s pool: `--k8s-worker-image`, `--k8s-worker-namespace`, `--k8s-control-plane-id`, `--k8s-worker-port`, `--k8s-worker-secret`, `--k8s-worker-configmap`, `--k8s-worker-image-pull-policy`, `--k8s-worker-service-account` (no global worker cap — per-org `Org.MaxWorkers`, 0=unbounded, is the only cap)
   - AWS / STS: `--aws-region`
-  - Compute-usage billing (emit side): `--billing-ingest-url` / `--billing-ingest-token` (env `DUCKGRES_BILLING_INGEST_URL` / `DUCKGRES_BILLING_INGEST_TOKEN`). Both must be set to enable per-org compute-usage metering+emit; either unset disables it (usage ships nowhere, queries never fail). See `docs/design/billing-compute-seconds-plan.md` and "Compute-Usage Billing" below.
+  - Compute-usage billing needs no config: metering is always on for the remote backend and billing PULLS usage over the internal-secret-authed HTTP API (`GET /api/v1/billing/usage` + `POST /api/v1/billing/ack`). See `docs/design/billing-pull-api.md` and "Compute-Usage Billing" below.
   - Pod scheduling knobs (CPU/memory requests, node selector, tolerations) are env-only — see `config_resolution.go`.
 
 Key CLI flags for duckdb-service mode:
@@ -438,58 +438,73 @@ impersonation, audit log; sliceable by org + user). Design + decisions:
 
 ## Compute-Usage Billing (managed-warehouse, remote backend only)
 
-duckgres meters per-org compute usage of worker pods and ships it to PostHog's
-public ingestion as `"managed warehouse compute usage"` capture events (billing
-sums the two raw metrics externally). Full design + decisions:
-`docs/design/billing-compute-seconds-plan.md`. Scope is the **emit side**, and
-**only** the remote/k8s backend (per-org worker pod with a known
-`WorkerProfile` size). Pipeline:
+duckgres meters per-org compute usage of worker pods into 60s buckets in the
+config store; the billing service **pulls** the accumulated usage over an HTTP
+API and acks a watermark, at which point duckgres deletes the acked buckets.
+Full design + decisions: `docs/design/billing-pull-api.md` (supersedes the
+push/capture reporting hop of `billing-compute-seconds-plan.md`; the metering
+side of that doc still applies). Scope is **only** the remote/k8s backend
+(per-org worker pod with a known `WorkerProfile` size). Pipeline:
 
 ```
-conn end → in-proc per-org counter (best-effort; never fails teardown)
+conn end → in-proc counter keyed (org, default team, query_source, worker size)
         │  flusher (~15s) UPSERT-increment → config-store buffer (cross-CP sum)
-        ▼  duckgres_org_compute_usage / duckgres_org_compute_drain_state
-   leader drain (~60s) → POST {DUCKGRES_BILLING_INGEST_URL}/capture (ship-then-delete)
+        ▼  duckgres_org_compute_usage (+ duckgres_compute_billing_cursor)
+billing: GET /api/v1/billing/usage (aggregated per key per UTC day, watermarks)
+       → POST /api/v1/billing/ack {watermark_high} → cursor advance + delete ≤ it
+safety:  leader-only GC hard-deletes buckets older than 30 days (WARN, alertable)
 ```
 
 Two raw metrics per connection over its full lifetime, using the **provisioned**
 worker size: `cpu_seconds = vCPU × ceil(conn_secs)`, `memory_seconds = GiB ×
 ceil(conn_secs)`. Counted internally in integer **millicore-seconds** /
 **MiB-seconds** (`compute_meter.go`) to avoid truncating a fractional-core /
-sub-GiB worker. Invariants for anyone touching this path:
+sub-GiB worker; worker size is stored in the bucket key as exact NUMERIC
+decimals (vCPU / GiB). `team_id` is the org\'s `default_team_id` (resolved from
+the config snapshot at connection end); `query_source` is the
+`duckgres.query_source` session GUC (`standard` unless set; a mid-connection
+change bills the whole connection under the final value). Invariants for anyone
+touching this path:
 
 - **Metering is strictly best-effort and off the hot path.** A metering error
-  (counter, flush, drain) must NEVER block or fail a query or connection
-  teardown. The connection-end record is added to an in-process per-org counter
-  (map+mutex, microseconds, no I/O); everything downstream is async/retried.
-  `cp.computeMeter` is nil unless the remote backend is configured with both
-  ingest URL+token — every call site is nil-safe.
-- **Enable gate = both `DUCKGRES_BILLING_INGEST_URL` and
-  `DUCKGRES_BILLING_INGEST_TOKEN` set** (`ControlPlaneConfig.BillingMeteringEnabled`).
-  Either unset → metering disabled (logged once at startup, ships nowhere). The
-  env names are fixed (infra wires them); keep them exact.
+  (counter, flush) must NEVER block or fail a query or connection teardown. The
+  connection-end record is added to an in-process counter (map+mutex,
+  microseconds, no I/O); the flush is async. `cp.computeMeter` is nil outside
+  the remote backend — every call site is nil-safe. There is no enable knob:
+  the remote backend always meters.
 - **Worker size is plumbed onto the connection** (`server.SetConnectionWorkerSize`
   → `clientConn.workerMillicores/workerMiB`, set in `control.go::handleConnection`
   from `workerBillingSize(workerProfile)`, remote-only). `workerMillicores==0`
   (non-remote / unknown) → metering skipped. The metric is computed once at the
-  SAME teardown point as `CloseConnectionMetrics` (the `#841` lifetime defer).
+  SAME teardown point as `CloseConnectionMetrics` (the `#841` lifetime defer),
+  via `server.ConnectionBilling` (which also carries the query source).
 - **Bucket = connection-end time floored to 60s.** Flush carries the sub-unit
   remainder forward so rounding never loses counts across flushes. Buffer flush
-  is UPSERT-increment so all CP pods sum into one row.
-- **Drain is leader-only** (co-located under the janitor lease via
-  `JanitorLeaderManager.AttachLeaderLoop`), **ship-then-delete / at-least-once**:
-  only buckets `bucket_start ≤ now-60s-30s grace AND > last_drained_bucket` ship;
-  on HTTP 2xx → TXN advance high-water + DELETE row; on failure → keep + retry.
-  `event_uuid = hash(org_id, bucket_start)` + `timestamp = bucket_start` are
-  deterministic so a re-ship collapses onto the same ClickHouse row.
+  is UPSERT-increment so all CP pods sum into one row per key.
+- **Serve only closed buckets.** `watermark_high` = the newest bucket with
+  `bucket_start ≤ now − 60s − 30s grace` (grace > flush interval, so every
+  CP\'s contribution has landed before a minute is served). The GET aggregates
+  the window `(cursor, watermark_high]` into one row per
+  `(org, team, query_source, cpu, mem_gib)` per **UTC day** — response size is
+  bounded by active keys × days, so billing downtime can\'t make it explode.
+- **Ack is the only deletion path (plus the 30d GC).** `POST /billing/ack`
+  advances the single global cursor monotonically and deletes buckets
+  `≤ watermark_high` in one TXN (`AckComputeUsage`). Idempotent — re-acks and
+  stale acks are no-ops. An ack beyond the latest closed bucket is rejected
+  (400) so it can never delete buckets that were never served. Auth is the
+  admin internal secret (`RequireAdmin` on both routes, registered inside the
+  audited `/api/v1` group in `multitenant.go`).
+- **Safety GC is leader-only** (`runComputeUsageGC`, attached under the janitor
+  lease): hard-deletes buckets older than 30 days regardless of ack and logs a
+  WARN with the dropped count — nonzero means billing stopped pulling (alert).
 - **Graceful shutdown does a final flush** after connections drain to their
   natural end (`shutdown`/`drainAndShutdown`), so a departing CP pod lands its
   last interval before exit.
-- Touching the meter/flush/drain/capture, the worker-size plumbing, or the
-  config knobs → update `controlplane/compute_meter_test.go`,
-  `compute_drain_test.go`, `compute_size_test.go`, the migration assertion in
-  `tests/configstore/migrations_postgres_test.go`, and the
-  `compute_usage_metering_wired` assertion in `tests/e2e-mw-dev/harness.sh`.
+- Touching the meter/flush/API/GC, the worker-size or query-source plumbing, or
+  the bucket key → update `controlplane/compute_meter_test.go`,
+  `compute_billing_api_test.go`, `compute_size_test.go`, the migration assertion
+  in `tests/configstore/migrations_postgres_test.go`, and the
+  `compute_usage_pull_api` assertion in `tests/e2e-mw-dev/harness.sh`.
 
 ## TODO Reference
 

README.md

@@ -280,8 +280,6 @@ Run with config file:
 | `DUCKGRES_QUERY_LOG_KAFKA_TOPIC` | Kafka topic for query-log events; required for Kafka mode | - |
 | `DUCKGRES_QUERY_LOG_KAFKA_CLIENT_ID` | Kafka client ID used by the query-log producer | `duckgres-query-log` |
 | `DUCKGRES_QUERY_LOG_KAFKA_GROUP_ID` | Kafka consumer group used by `duckgres --mode query-log-writer` | `duckgres-query-log-writer` |
-| `DUCKGRES_BILLING_INGEST_URL` | PostHog public ingestion base URL for managed-warehouse compute-usage events (e.g. `https://us.i.posthog.com`). Remote backend only. Unset (or token unset) disables metering — usage ships nowhere and a query is never failed on its account. | - |
-| `DUCKGRES_BILLING_INGEST_TOKEN` | PostHog project API token stamped on the compute-usage capture event. Remote backend only; both URL and token must be set to enable metering. | - |
 | `POSTHOG_API_KEY` | PostHog project API key (`phc_...`); enables log export **and product-analytics events** | - |
 | `POSTHOG_HOST` | PostHog ingest host | `us.i.posthog.com` |
 | `ADDITIONAL_POSTHOG_API_KEYS` | **(Experimental)** Comma-separated list of additional PostHog API keys to publish logs to. Requires `POSTHOG_API_KEY` to be set. | - |

configresolve/cliflags.go

@@ -72,8 +72,6 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 	k8sWorkerServiceAccount := fs.String("k8s-worker-service-account", "", "Neutral ServiceAccount name for K8s worker pods (default: duckgres-worker) (env: DUCKGRES_K8S_WORKER_SERVICE_ACCOUNT)")
 	awsRegion := fs.String("aws-region", "", "AWS region for STS client (env: DUCKGRES_AWS_REGION)")
 	queryLog := fs.Bool("query-log", true, "Enable/disable DuckLake query log (use --query-log=false to disable; env: DUCKGRES_QUERY_LOG_ENABLED)")
-	billingIngestURL := fs.String("billing-ingest-url", "", "PostHog public ingestion base URL for managed-warehouse compute-usage events, e.g. https://us.i.posthog.com (remote backend only; unset disables metering) (env: DUCKGRES_BILLING_INGEST_URL)")
-	billingIngestToken := fs.String("billing-ingest-token", "", "PostHog project API token for managed-warehouse compute-usage events (remote backend only; unset disables metering) (env: DUCKGRES_BILLING_INGEST_TOKEN)")
 
 	return func() CLIInputs {
 		cli := CLIInputs{Set: map[string]bool{}}
@@ -130,8 +128,6 @@ func RegisterCLIInputsFlags(fs *flag.FlagSet) func() CLIInputs {
 		cli.K8sWorkerServiceAccount = *k8sWorkerServiceAccount
 		cli.AWSRegion = *awsRegion
 		cli.QueryLog = *queryLog
-		cli.BillingIngestURL = *billingIngestURL
-		cli.BillingIngestToken = *billingIngestToken
 		return cli
 	}
 }

configresolve/cliflags_test.go

@@ -51,8 +51,6 @@ func fieldNameToFlagName(name string) string {
 		{"AWSRegion", "aws-region"},
 		{"SNIRoutingMode", "sni-routing-mode"},
 		{"ManagedHostnameSuffixes", "managed-hostname-suffixes"},
-		{"BillingIngestURL", "billing-ingest-url"},
-		{"BillingIngestToken", "billing-ingest-token"},
 		{"ConfigStoreConn", "config-store"},
 		{"ConfigPollInterval", "config-poll-interval"},
 		{"InternalSecretFallbacks", "internal-secret-fallbacks"},

configresolve/resolve.go

@@ -82,8 +82,6 @@ type CLIInputs struct {
 	K8sWorkerTolerationValue    string
 	AWSRegion                   string
 	QueryLog                    bool
-	BillingIngestURL            string
-	BillingIngestToken          string
 }
 
 type Resolved struct {
@@ -131,8 +129,6 @@ type Resolved struct {
 	ManagedHostnameSuffixes         []string
 	DucklingBucketSuffix            string
 	DuckLakeDefaultSpecVersion      string
-	BillingIngestURL                string
-	BillingIngestToken              string
 }
 
 func intPtr(n int) *int    { return &n }
@@ -222,7 +218,6 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 	var sniRoutingMode string
 	var managedHostnameSuffixes []string
 	var ducklingBucketSuffix string
-	var billingIngestURL, billingIngestToken string
 
 	if fileCfg != nil {
 		if fileCfg.Host != "" {
@@ -792,15 +787,6 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 	if v := getenv("DUCKGRES_DUCKLING_BUCKET_SUFFIX"); v != "" {
 		ducklingBucketSuffix = v
 	}
-	// Managed-warehouse compute-usage billing ingest (remote backend only). Both
-	// must be set for metering to ship anything; either unset disables metering.
-	// Env names are fixed — infra wires these exact keys.
-	if v := getenv("DUCKGRES_BILLING_INGEST_URL"); v != "" {
-		billingIngestURL = v
-	}
-	if v := getenv("DUCKGRES_BILLING_INGEST_TOKEN"); v != "" {
-		billingIngestToken = v
-	}
 	if v := getenv("DUCKGRES_WORKER_BACKEND"); v != "" {
 		workerBackend = v
 	}
@@ -1112,12 +1098,6 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 		// JWT_SIGNING_KEY_FALLBACKS.
 		internalSecretFallbacks = splitAndTrim(cli.InternalSecretFallbacks, ",")
 	}
-	if cli.Set["billing-ingest-url"] {
-		billingIngestURL = cli.BillingIngestURL
-	}
-	if cli.Set["billing-ingest-token"] {
-		billingIngestToken = cli.BillingIngestToken
-	}
 	if cli.Set["sni-routing-mode"] {
 		sniRoutingMode = cli.SNIRoutingMode
 	}
@@ -1273,8 +1253,6 @@ func ResolveEffective(fileCfg *configloader.FileConfig, cli CLIInputs, getenv fu
 		ManagedHostnameSuffixes:         managedHostnameSuffixes,
 		DucklingBucketSuffix:            ducklingBucketSuffix,
 		DuckLakeDefaultSpecVersion:      cfg.DuckLake.SpecVersion,
-		BillingIngestURL:                billingIngestURL,
-		BillingIngestToken:              billingIngestToken,
 	}
 }
 

controlplane/compute_billing_api.go

@@ -0,0 +1,117 @@
+package controlplane
+
+import (
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+
+	"github.com/posthog/duckgres/controlplane/configstore"
+)
+
+// billingUsageStore is the config-store surface the billing pull API needs.
+type billingUsageStore interface {
+	AggregateComputeUsage(low, high time.Time) ([]configstore.ComputeUsageRow, error)
+	ComputeBillingCursor() (time.Time, bool, error)
+	AckComputeUsage(watermarkHigh time.Time) (int64, error)
+}
+
+// billingAPIHandler serves the pull-based compute-billing API
+// (docs/design/billing-pull-api.md): billing GETs the usage accumulated since
+// its last ack, processes it, and POSTs the watermark back; duckgres advances
+// the cursor and deletes everything at or below it. Reads/deletes on the
+// config store only — never on the query hot path.
+type billingAPIHandler struct {
+	store billingUsageStore
+	now   func() time.Time
+}
+
+// registerBillingAPI mounts the billing pull API on an authenticated route
+// group. The caller passes the admin-gating middleware (the billing service
+// authenticates with the internal secret, which resolves to admin).
+func registerBillingAPI(r gin.IRouter, store billingUsageStore, requireAdmin gin.HandlerFunc) {
+	h := &billingAPIHandler{store: store, now: time.Now}
+	r.GET("/billing/usage", requireAdmin, h.getUsage)
+	r.POST("/billing/ack", requireAdmin, h.postAck)
+}
+
+// latestClosedBucket returns the newest bucket_start that is fully closed:
+// every contribution for it has landed (grace exceeds the in-process flush
+// interval), so it is safe to serve — and later delete on ack.
+func (h *billingAPIHandler) latestClosedBucket() time.Time {
+	return h.now().UTC().Add(-computeBucketWidth - computeBucketGrace).Truncate(computeBucketWidth)
+}
+
+// getUsage returns usage aggregated since the last ack — one row per key
+// (org, team, query_source, worker size) per UTC day — plus the watermark
+// window. watermark_low is the server cursor (what billing last acked;
+// billing should cross-check it against its own record), watermark_high is
+// what billing acks after processing.
+func (h *billingAPIHandler) getUsage(c *gin.Context) {
+	low, _, err := h.store.ComputeBillingCursor()
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "read billing cursor: " + err.Error()})
+		return
+	}
+	low = low.UTC() // zero time (never acked) serves everything buffered
+
+	high := h.latestClosedBucket()
+	if !high.After(low) {
+		// Nothing closed beyond the cursor yet (fresh deploy or a pull racing
+		// right behind an ack). An empty window with high == low is a valid
+		// response: billing acks it as a no-op.
+		c.JSON(http.StatusOK, gin.H{
+			"watermark_low":  low.Format(time.RFC3339),
+			"watermark_high": low.Format(time.RFC3339),
+			"usage":          []configstore.ComputeUsageRow{},
+		})
+		return
+	}
+
+	rows, err := h.store.AggregateComputeUsage(low, high)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "aggregate usage: " + err.Error()})
+		return
+	}
+	if rows == nil {
+		rows = []configstore.ComputeUsageRow{}
+	}
+	c.JSON(http.StatusOK, gin.H{
+		"watermark_low":  low.Format(time.RFC3339),
+		"watermark_high": high.Format(time.RFC3339),
+		"usage":          rows,
+	})
+}
+
+type billingAckRequest struct {
+	WatermarkHigh time.Time `json:"watermark_high" binding:"required"`
+}
+
+// postAck advances the cursor to watermark_high and deletes every buffered
+// bucket at or below it. Idempotent: re-acking an already-acked (or older)
+// watermark is a no-op. The watermark must be a closed bucket boundary the
+// server could have served — acking into the still-accumulating present is
+// rejected so an ack can never delete buckets that were never returned.
+func (h *billingAPIHandler) postAck(c *gin.Context) {
+	var req billingAckRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid ack body (want {\"watermark_high\": RFC3339}): " + err.Error()})
+		return
+	}
+	high := req.WatermarkHigh.UTC()
+	if latest := h.latestClosedBucket(); high.After(latest) {
+		c.JSON(http.StatusBadRequest, gin.H{
+			"error": "watermark_high is beyond the latest closed bucket; ack exactly the watermark_high returned by GET /billing/usage",
+		})
+		return
+	}
+	deleted, err := h.store.AckComputeUsage(high)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "ack usage: " + err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{
+		"acked":   high.Format(time.RFC3339),
+		"deleted": deleted,
+	})
+}