diff --git a/.gitignore b/.gitignore index 8ffca6f1..1ddc39fb 100644 --- a/.gitignore +++ b/.gitignore @@ -21,6 +21,11 @@ Thumbs.db # Cloudflare Worker local secrets (wrangler dev) .dev.vars .dev.vars.* +# Wrangler local state — miniflare D1/KV/R2 caches, dev sqlite DBs, etc. +# Written by `wrangler dev` and `wrangler d1 migrations list` against a +# --remote target (which still initializes a local shadow). Never commit. +.wrangler/ +**/.wrangler/ # Per-developer dev-box lifecycle state (written by deploy/*/deploy-qemu-dev.sh) deploy/**/.qemu-dev-state-* @@ -47,7 +52,13 @@ cloudflare-workers/api-edge/assets/ # Compiled binaries bin/ opensandbox-worker +opensandbox-server +# Repo-root Go build outputs — produced when someone runs `go build ./cmd/X/` +# from the repo root without `-o`. Path-anchored so we don't accidentally +# ignore legitimate `server` / `worker` / `oc` DIRECTORIES elsewhere. /worker +/server +/oc # Worker env (secrets) worker.env diff --git a/cloudflare-workers/api-edge/migrations/0004_usage_samples_disk_mb.sql b/cloudflare-workers/api-edge/migrations/0004_usage_samples_disk_mb.sql new file mode 100644 index 00000000..5e32e6d9 --- /dev/null +++ b/cloudflare-workers/api-edge/migrations/0004_usage_samples_disk_mb.sql @@ -0,0 +1,7 @@ +-- Add disk_mb to usage_samples so the autumn-meter can attribute disk-overage +-- GB-seconds against the running-sandbox billing bucket. Cell emits it in the +-- usage_tick payload, events-ingest lands it in this column, autumn_meter reads +-- it in the per-bucket aggregation. Default 0 = "no disk signal for this tick", +-- which the autumn-meter treats as no overage (matches "free 20GB is included" +-- and prevents pre-migration rows from silently accruing overage). +ALTER TABLE usage_samples ADD COLUMN disk_mb INTEGER NOT NULL DEFAULT 0; diff --git a/cloudflare-workers/api-edge/schema-snapshots/current_schema.sql b/cloudflare-workers/api-edge/schema-snapshots/current_schema.sql index 05f6d7d8..00505d44 100644 --- a/cloudflare-workers/api-edge/schema-snapshots/current_schema.sql +++ b/cloudflare-workers/api-edge/schema-snapshots/current_schema.sql @@ -243,7 +243,8 @@ CREATE TABLE usage_meter_events ( CREATE TABLE usage_samples ( id TEXT PRIMARY KEY, org_id TEXT NOT NULL, sandbox_id TEXT NOT NULL, memory_mb INTEGER NOT NULL, cpu_count INTEGER NOT NULL, interval_s INTEGER NOT NULL, - ts INTEGER NOT NULL, cell_id TEXT NOT NULL, rolled_up INTEGER NOT NULL DEFAULT 0); + ts INTEGER NOT NULL, cell_id TEXT NOT NULL, rolled_up INTEGER NOT NULL DEFAULT 0, + disk_mb INTEGER NOT NULL DEFAULT 0); CREATE TABLE usage_snapshots ( org_id TEXT NOT NULL, diff --git a/cloudflare-workers/api-edge/src/autumn_meter.ts b/cloudflare-workers/api-edge/src/autumn_meter.ts index 4c58f214..2e77043d 100644 --- a/cloudflare-workers/api-edge/src/autumn_meter.ts +++ b/cloudflare-workers/api-edge/src/autumn_meter.ts @@ -38,6 +38,16 @@ const TIER_FEATURE_BY_MEMORY_MB: Record = { 65536: "compute_64gb", }; +// Disk-overage billing. Every GB above DISK_FREE_ALLOWANCE_MB (20 GB, the +// baseline every sandbox ships with) is metered at $0.0000001/GB-second +// (~$0.26/GB-month) — see pricing.md + internal/billing/pricing.go. +// The Autumn feature is charged in whole GB-seconds; we ceil the fractional +// remainder per bucket so sub-unit accrual across ticks isn't silently +// dropped. The Autumn dashboard must have a `disk_overage_gb_seconds` +// feature attached to the pro product for track() to land. +const DISK_FREE_ALLOWANCE_MB = 20480; +const DISK_OVERAGE_FEATURE = "disk_overage_gb_seconds"; + // Globally-unique, retry-stable key. Autumn dedupes on the bare key across all // customers, so it includes the org; keyed on bucket start (not wall-clock) so a // replay reuses it. Matches autumn.UsageIdempotencyKey on the (removed) cell. @@ -122,8 +132,14 @@ async function meterOrg(env: AutumnEnv, org: AutumnOrgRow, nowSec: number): Prom return true; } -// trackBucket aggregates usage_samples in [from, to) by memory tier and tracks -// one usage event per tier to Autumn. Returns true if the balance is now <= 0. +// trackBucket aggregates usage_samples in [from, to) by memory tier + a single +// disk-overage GB-second total, tracks one usage event per dimension to Autumn, +// and returns true if the resulting balance is <= 0. +// +// The two dimensions are independent Autumn features (compute_{tier}, and +// disk_overage_gb_seconds) so a customer running an idle-but-large-disk +// sandbox accrues disk cost separately from compute cost. Both use the same +// per-org per-bucket idempotency key namespace so a retry deduplicates cleanly. async function trackBucket(env: AutumnEnv, orgID: string, fromSec: number, toSec: number): Promise { const aggRes = await env.OPENCOMPUTER_DB.prepare( `SELECT memory_mb AS memory_mb, SUM(interval_s) AS secs @@ -134,7 +150,22 @@ async function trackBucket(env: AutumnEnv, orgID: string, fromSec: number, toSec .bind(orgID, fromSec * 1000, toSec * 1000) .all(); const tiers = aggRes.results ?? []; - if (tiers.length === 0) return false; + + // Disk overage: SUM((disk_mb - 20480) * interval_s / 1024) — bytes above the + // free allowance × seconds, converted to GB-seconds. Only rows with real + // overage contribute (the WHERE clause culls 0/default rows so idle sandboxes + // at baseline size don't produce empty aggregation work). Returned as a real + // (SQLite REAL), ceiled below for the whole-GB-second Autumn feature. + const diskRes = await env.OPENCOMPUTER_DB.prepare( + `SELECT COALESCE(SUM((disk_mb - ?1) * interval_s), 0) / 1024.0 AS gb_seconds + FROM usage_samples + WHERE org_id = ?2 AND ts >= ?3 AND ts < ?4 AND disk_mb > ?1`, + ) + .bind(DISK_FREE_ALLOWANCE_MB, orgID, fromSec * 1000, toSec * 1000) + .first<{ gb_seconds: number | null }>(); + const diskGBSeconds = Math.ceil(diskRes?.gb_seconds ?? 0); + + if (tiers.length === 0 && diskGBSeconds <= 0) return false; let remaining: number | null = null; for (const t of tiers) { @@ -151,6 +182,16 @@ async function trackBucket(env: AutumnEnv, orgID: string, fromSec: number, toSec idempotencyKey: usageIdempotencyKey(orgID, fromSec, feature), }); } + + if (diskGBSeconds > 0) { + remaining = await trackAutumnUsage(env, { + customerID: orgID, + featureID: DISK_OVERAGE_FEATURE, + value: diskGBSeconds, + idempotencyKey: usageIdempotencyKey(orgID, fromSec, DISK_OVERAGE_FEATURE), + }); + } + return remaining !== null && remaining <= 0; } diff --git a/cloudflare-workers/api-edge/src/index.ts b/cloudflare-workers/api-edge/src/index.ts index a3cc23c2..a6f7fc8f 100644 --- a/cloudflare-workers/api-edge/src/index.ts +++ b/cloudflare-workers/api-edge/src/index.ts @@ -458,7 +458,8 @@ async function enforceCreatePolicy( // is_halted projection — no CreditAccount DO. On a halt, self-heal (re-check // Autumn) so a just-topped-up user isn't stuck behind a lagging webhook. The // free-tier memory/CPU/disk ceilings below are skipped for autumn orgs (they - // pay per GB-second); only the per-org max_disk_mb cap applies. + // pay per GB-second and are metered per-bucket by autumn_meter — disk overage + // via the `disk_overage_gb_seconds` feature). if (org.billing_provider === "autumn") { if (org.is_halted === 1 && (await selfHealHalt(env, orgID))) { return json({ error: "credits exhausted — top up to resume" }, 402); @@ -484,7 +485,9 @@ async function enforceCreatePolicy( // Free-tier ceilings: 4GB / 1 vCPU, 20GB disk. Legacy only — autumn (prepaid) // orgs pay per GB-second and are gated by balance/halt, so they may launch any - // size. Disk is still bounded for everyone by the per-org max_disk_mb check below. + // size up to the platform ceiling (enforced upstream). The per-org + // `max_disk_mb` column is no longer enforced now that disk overage is metered + // to Autumn end-to-end; the D1 column stays as dead legacy. if (org.billing_provider !== "autumn" && plan === "free") { if (sizes.memoryMB > 4096 || sizes.cpuCount > 1) { return json({ error: "upgrade to pro for larger instances" }, 402); @@ -494,14 +497,6 @@ async function enforceCreatePolicy( } } - // Per-org disk ceiling (all plans). 0 in D1 means "use the 20GB default". - if (sizes.diskMB > 0) { - const maxDisk = org.max_disk_mb > 0 ? org.max_disk_mb : 20480; - if (sizes.diskMB > maxDisk) { - return json({ error: `disk size ${sizes.diskMB}MB exceeds org limit of ${maxDisk}MB` }, 403); - } - } - // Concurrent-sandbox limit (all plans). Counts `running` only — hibernated // sandboxes live in S3 and don't consume worker capacity. The count spans // every cell via the global sandboxes_index, which is the whole reason it diff --git a/cloudflare-workers/events-ingest/src/index.ts b/cloudflare-workers/events-ingest/src/index.ts index 950f7e6c..9e1be3c7 100644 --- a/cloudflare-workers/events-ingest/src/index.ts +++ b/cloudflare-workers/events-ingest/src/index.ts @@ -429,8 +429,8 @@ export default { // cross-DO debit fan-out, it's a local D1 write, so it should share the // batch's retry-on-failure guarantee (503 → CP forwarder keeps the PEL). const usageSampleInsert = env.OPENCOMPUTER_DB.prepare( - `INSERT INTO usage_samples (id, org_id, sandbox_id, memory_mb, cpu_count, interval_s, ts, cell_id) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + `INSERT INTO usage_samples (id, org_id, sandbox_id, memory_mb, cpu_count, interval_s, ts, cell_id, disk_mb) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9) ON CONFLICT(id) DO NOTHING`, ); // Worker-validation guard. Edge usage is counted per-tick (one usage_sample @@ -483,6 +483,7 @@ export default { memory_mb?: number; cpu_count?: number; interval_s?: number; + disk_mb?: number; }; return usageSampleInsert.bind( e.id, @@ -493,6 +494,10 @@ export default { p.interval_s ?? 0, Date.parse(e.timestamp) || Date.now(), e.cell_id, + // disk_mb is 0 on pre-billing-cutover rows (older cells that don't + // stamp it yet). autumn_meter reads that as "no overage this tick", + // matching the free 20GB allowance and keeping backfill safe. + p.disk_mb ?? 0, ); }); diff --git a/cmd/server/main.go b/cmd/server/main.go index 66e86060..53891247 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -103,6 +103,7 @@ func main() { CFAdminSecret: cfg.CFAdminSecret, CFEventSecret: cfg.CFEventSecret, RequireCapToken: proBillingEdge, + MaxDiskMB: cfg.MaxDiskMB, } // Initialize PostgreSQL if configured @@ -269,6 +270,26 @@ func main() { _ = fwd.Stop(stopCtx) }() log.Printf("opensandbox: CF event forwarder started (endpoint=%s cell=%s)", cfg.CFEventEndpoint, cfg.CellID) + + // Hibernation-billing sweeper — periodically synthesizes usage_tick + // events for hibernated sandboxes with disk overage, so the edge + // autumn_meter bills them even though no worker's usage_ticker fires + // while the qcow2 sits in Tigris. Sends directly to events-ingest via + // cfClient (bypasses Redis — the payload is generated on the CP, not + // on a worker). Inert unless opts.Store is wired (needs the shared PG + // to enumerate hibernated sandboxes + their current disk envelope). + if opts.Store != nil { + sweeper := controlplane.NewHibernationBillingSweeper(opts.Store, cfClient, cfg.CellID, 5*time.Minute) + if sweeper != nil { + sweeper.Start(context.Background()) + defer func() { + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + _ = sweeper.Stop(stopCtx) + }() + log.Printf("opensandbox: hibernation-billing sweeper started (5 min interval, cell=%s)", cfg.CellID) + } + } } else if cfg.Mode == "server" { log.Printf("opensandbox: CF event forwarder NOT started (CFEventEndpoint/Secret/CellID unset)") } diff --git a/cmd/worker/main.go b/cmd/worker/main.go index 5838a0cf..e9fde04b 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go @@ -899,7 +899,7 @@ func main() { // Inert unless CellID is set (combined-mode dev without Redis stream // would write events to /dev/null since there's no consumer). if cfg.CellID != "" && mgr != nil { - usageTicker := worker.NewUsageTicker(mgr, sandboxDBMgr, 20*time.Second, 10) + usageTicker := worker.NewUsageTicker(mgr, sandboxDBMgr, store, 20*time.Second, 10) if usageTicker != nil { // Wire the ticker as the manager's lifecycle observer so scale, // destroy, hibernate, and wake events flush accurate final-slice diff --git a/internal/api/internal_sandbox.go b/internal/api/internal_sandbox.go index d72fbf8b..41fb8476 100644 --- a/internal/api/internal_sandbox.go +++ b/internal/api/internal_sandbox.go @@ -1,6 +1,7 @@ package api import ( + "fmt" "net/http" "strings" @@ -136,15 +137,20 @@ func (s *Server) internalCreateSandbox(c echo.Context) error { // Physical disk bounds. These are hardware/safety limits, not org policy: // diskMB=1 boots-fail opaquely, diskMB=10_000_000 would allocate 10TB per - // sandbox. Per-org disk ceilings (free-tier cap, custom max_disk_mb) are - // enforced at the edge against D1 — see enforceCreatePolicy in - // cloudflare-workers/api-edge/src/index.ts. The cell trusts the cap-token - // and only checks these physical bounds. + // sandbox. The free-tier disk ceiling is enforced at the edge against D1 + // (see enforceCreatePolicy in cloudflare-workers/api-edge/src/index.ts). + // Paying orgs are billed per GB-second above the 20GB free allowance via + // the edge autumn_meter — no per-org admission cap. The cell trusts the + // cap-token and only checks these physical bounds. if cfg.DiskMB < 20480 { return c.JSON(http.StatusBadRequest, map[string]string{"error": "diskMB must be at least 20480 (20GB)"}) } - if cfg.DiskMB > 262144 { - return c.JSON(http.StatusBadRequest, map[string]string{"error": "diskMB cannot exceed 262144 (256GB)"}) + maxDiskMB := s.maxDiskMB + if maxDiskMB <= 0 { + maxDiskMB = 262144 // default 256GB + } + if cfg.DiskMB > maxDiskMB { + return c.JSON(http.StatusBadRequest, map[string]string{"error": fmt.Sprintf("diskMB cannot exceed %d (%dGB)", maxDiskMB, maxDiskMB/1024)}) } // Declarative image or named snapshot → resolve to a checkpoint and use diff --git a/internal/api/router.go b/internal/api/router.go index ec1725c1..73f93c7e 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -51,6 +51,7 @@ type Server struct { cfAdminSecret string // HMAC shared with CreditAccount DO for /admin/halt-org and /admin/resume-org; empty disables auth (dev only) cfEventSecret string // HMAC shared with the api-edge Worker for /internal/secret-refresh and other edge-→cell push paths cellID string // this control plane's cell_id (for the cap-token cell check) + maxDiskMB int // platform per-sandbox disk ceiling; 0 = default 262144 (256GB) platformOrgID uuid.UUID // owner of the shared catalog snapshots; anchors the public-snapshot fallback + gates publish (uuid.Nil = fallback disabled) mode string // "server", "worker", "combined" workerID string // this worker's ID @@ -162,6 +163,13 @@ type ServerOpts struct { SandboxAPIProxy *proxy.SandboxAPIProxy // nil except in server mode (proxies data-plane to workers) StripeClient *billing.StripeClient // nil if Stripe not configured RedisClient *redis.Client // nil if Redis not configured (for health checks) + // MaxDiskMB caps `diskMB` at create time (both /api/sandboxes and the + // edge-fanout /internal/sandboxes/create). Not a per-org policy — a + // platform sanity ceiling protecting worker-density + hibernate/wake + // latency at very large disk sizes. Zero = code default (262144 = 256GB). + // Bump via OPENSANDBOX_MAX_DISK_MB once timeouts + worker density have + // been validated for the new ceiling. + MaxDiskMB int } // NewServer creates a new API server with all routes configured. @@ -192,6 +200,7 @@ func NewServer(mgr sandbox.Manager, ptyMgr *sandbox.PTYManager, apiKey string, o s.cfAdminSecret = opts.CFAdminSecret s.cfEventSecret = opts.CFEventSecret s.cellID = opts.CellID + s.maxDiskMB = opts.MaxDiskMB if opts.PlatformOrgID != "" { if pid, err := uuid.Parse(opts.PlatformOrgID); err == nil { s.platformOrgID = pid diff --git a/internal/api/sandbox.go b/internal/api/sandbox.go index 59454be2..af98c2bd 100644 --- a/internal/api/sandbox.go +++ b/internal/api/sandbox.go @@ -126,8 +126,8 @@ func (s *Server) createSandbox(c echo.Context) error { // Free-tier gate (legacy only): trial-credit + machine-size limit. // Autumn orgs pay per GB-second and are gated at the edge // (balance/halt), so neither the credit check nor the size ceiling - // applies — they may launch any size (disk still bounded by the - // per-org MaxDiskMB knob below). + // applies — they may launch any size up to the platform ceiling + // (currently 256GB for disk; enforced above). if effPlan == "free" && effProvider != "autumn" { if org.FreeCreditsRemainingCents <= 0 { return c.JSON(http.StatusPaymentRequired, map[string]string{ @@ -158,26 +158,33 @@ func (s *Server) createSandbox(c echo.Context) error { "error": "diskMB must be at least 20480 (20GB)", }) } - if cfg.DiskMB > 262144 { + maxDiskMB := s.maxDiskMB + if maxDiskMB <= 0 { + maxDiskMB = 262144 // default 256GB + } + if cfg.DiskMB > maxDiskMB { return c.JSON(http.StatusBadRequest, map[string]string{ - "error": "diskMB cannot exceed 262144 (256GB)", + "error": fmt.Sprintf("diskMB cannot exceed %d (%dGB)", maxDiskMB, maxDiskMB/1024), }) } if org != nil { + // Free-tier ceiling: 20GB disk. Autumn orgs pay per GB-second (metered + // by the edge autumn_meter → billed via Autumn's `disk_overage_gb_seconds` + // feature) so they may launch anything up to the platform ceiling + // (256GB by default, tunable via OPENSANDBOX_MAX_DISK_MB — see + // checked above; only free users are gated here. + // + // NOTE: the per-org `max_disk_mb` column is no longer enforced. Prior to + // disk-billing being wired end-to-end it was a manual admission cap ("has + // this customer been approved for larger disks"). Now that disk overage is + // actually billed, any paying org may launch any allowed size and get an + // accurate bill. The column stays in the schema as dead legacy; a follow-up + // migration can drop it. if effPlan == "free" && effProvider != "autumn" && cfg.DiskMB > 20480 { return c.JSON(http.StatusPaymentRequired, map[string]string{ "error": "upgrade to pro for larger disk sizes", }) } - maxDisk := org.MaxDiskMB - if maxDisk == 0 { - maxDisk = 20480 - } - if cfg.DiskMB > maxDisk { - return c.JSON(http.StatusForbidden, map[string]string{ - "error": fmt.Sprintf("disk size %dMB exceeds org limit of %dMB", cfg.DiskMB, maxDisk), - }) - } } // Declarative image or named snapshot → resolve to checkpoint and use createFromCheckpoint flow @@ -2474,7 +2481,12 @@ func (s *Server) wakeSandboxRemote(c echo.Context, sandboxID string, req types.W } } - grpcCtx, cancel := context.WithTimeout(c.Request().Context(), 60*time.Second) + // 5 min covers cross-worker wake at the platform's 256GB disk cap: the + // target worker chunk-downloads (16-way parallel, 64MB chunks — see + // storage/s3.go) + tar-extract. At the old 60s cap, any wake of a >30GB-used + // sandbox on a worker that didn't already hold the qcow2 timed out. Same- + // worker wake is unaffected (opens a local qcow2 in <1s). + grpcCtx, cancel := context.WithTimeout(c.Request().Context(), 5*time.Minute) defer cancel() grpcResp, err := grpcClient.WakeSandbox(grpcCtx, &pb.WakeSandboxRequest{ diff --git a/internal/config/config.go b/internal/config/config.go index 46c26ac4..0b8bd337 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -108,6 +108,13 @@ type Config struct { DefaultSandboxCPUs int // default vCPUs per sandbox, default 1 DefaultSandboxDiskMB int // default disk quota per sandbox (MB), 0 = no quota + // MaxDiskMB caps `diskMB` accepted by the create APIs (/api/sandboxes and + // the edge-fanout /internal/sandboxes/create). Platform sanity ceiling — + // protects worker density + hibernate/wake latency at very large disks. + // 0 = code default 262144 (256GB). Bump via OPENSANDBOX_MAX_DISK_MB once + // timeouts + worker density have been validated for the new ceiling. + MaxDiskMB int + // QEMU VM configuration (worker mode) KernelPath string // Path to vmlinux kernel ImagesDir string // Path to base rootfs images @@ -372,6 +379,7 @@ func Load() (*Config, error) { DefaultSandboxMemoryMB: envOrDefaultInt("OPENSANDBOX_DEFAULT_SANDBOX_MEMORY_MB", 256), DefaultSandboxCPUs: envOrDefaultInt("OPENSANDBOX_DEFAULT_SANDBOX_CPUS", 1), DefaultSandboxDiskMB: envOrDefaultInt("OPENSANDBOX_DEFAULT_SANDBOX_DISK_MB", 0), + MaxDiskMB: envOrDefaultInt("OPENSANDBOX_MAX_DISK_MB", 0), KernelPath: os.Getenv("OPENSANDBOX_KERNEL_PATH"), // default derived from DataDir ImagesDir: os.Getenv("OPENSANDBOX_IMAGES_DIR"), // default derived from DataDir diff --git a/internal/config/keyvault.go b/internal/config/keyvault.go index b160c06a..6a65a491 100644 --- a/internal/config/keyvault.go +++ b/internal/config/keyvault.go @@ -74,6 +74,11 @@ var kvMapping = map[string]string{ "server-cf-admin-secret": "OPENSANDBOX_CF_ADMIN_SECRET", "server-session-jwt-secret": "OPENSANDBOX_SESSION_JWT_SECRET", "server-halt-list-url": "OPENSANDBOX_HALT_LIST_URL", + // Platform per-sandbox disk ceiling enforced by the CP admission checks. + // Zero = code default 262144 (256GB). Raise via KV to allow larger sandboxes + // on this cell after validating hibernate/wake/migration timeouts at the + // new ceiling. + "server-max-disk-mb": "OPENSANDBOX_MAX_DISK_MB", // Pre-warmed sandbox pool (control-plane owned). Per-cell, per-worker target. "server-pool-target": "OPENSANDBOX_POOL_TARGET", // pooled boxes per worker (default 10; 0 disables) "server-pool-enabled": "OPENSANDBOX_POOL_ENABLED", // "0" to disable (default on) diff --git a/internal/controlplane/hibernation_billing.go b/internal/controlplane/hibernation_billing.go new file mode 100644 index 00000000..6f476471 --- /dev/null +++ b/internal/controlplane/hibernation_billing.go @@ -0,0 +1,186 @@ +package controlplane + +import ( + "context" + "encoding/json" + "fmt" + "log" + "sync" + "time" + + "github.com/opensandbox/opensandbox/internal/db" +) + +// syntheticSandboxEvent mirrors the JSON shape of +// worker.SandboxEventEnvelope (kept local here because controlplane cannot +// import worker without a cycle). Only the fields events-ingest actually reads +// are declared. +type syntheticSandboxEvent struct { + ID string `json:"id"` + Type string `json:"type"` + SandboxID string `json:"sandbox_id"` + OrgID string `json:"org_id,omitempty"` + WorkerID string `json:"worker_id"` + CellID string `json:"cell_id"` + Payload json.RawMessage `json:"payload"` + Timestamp time.Time `json:"timestamp"` +} + +// HibernationBillingSweeper closes the "billed for the lifetime of the sandbox +// — running or hibernated" gap. +// +// Running sandboxes emit a `usage_tick` every 20s from the worker's ticker; +// events-ingest lands them in `usage_samples` on the edge, autumn_meter reads +// disk_mb per row and bills overage. Hibernated sandboxes have no running VM +// on any worker so no organic tick fires, yet the qcow2 still sits in Tigris +// consuming real storage cost — this sweeper mints the missing signal. +// +// Every `interval`, query the cell's PG for hibernated sandboxes whose current +// disk_mb exceeds the 20 GB free allowance, wrap each into a synthetic +// `usage_tick` envelope (memory_mb=0, cpu_count=0, disk_mb=), batch, +// and POST to events-ingest via the same HMAC-signed path the CF forwarder +// uses. events-ingest INSERTs into `usage_samples` with the same +// `ON CONFLICT(id) DO NOTHING` dedup as organic ticks, and autumn_meter +// naturally aggregates the disk-overage column across all rows in the bucket +// — a synthetic hibernated row is indistinguishable from an organic running +// row at the aggregation layer. +// +// The synthetic event's ID is deterministic per (sandbox, bucket-start) so a +// retry after a partial batch failure deduplicates cleanly at events-ingest. +// worker_id is intentionally empty: the zombie-tick guard drops mismatches +// but treats missing worker_id as "unknown, allow" (hibernated boxes have no +// live owner to match against). +type HibernationBillingSweeper struct { + store *db.Store + client *CFEventClient + cellID string + interval time.Duration + + stopCh chan struct{} + doneCh chan struct{} + once sync.Once +} + +// NewHibernationBillingSweeper wires the sweeper. A nil store or client +// returns nil (sweeper disabled — matches the CFEventClient's own opt-in +// behavior when the events endpoint isn't configured). +func NewHibernationBillingSweeper(store *db.Store, client *CFEventClient, cellID string, interval time.Duration) *HibernationBillingSweeper { + if store == nil || client == nil || cellID == "" { + return nil + } + if interval <= 0 { + interval = 5 * time.Minute // matches autumn_meter bucket size + } + return &HibernationBillingSweeper{ + store: store, + client: client, + cellID: cellID, + interval: interval, + stopCh: make(chan struct{}), + doneCh: make(chan struct{}), + } +} + +// Start begins the sweep loop. Safe to call once. +func (s *HibernationBillingSweeper) Start(ctx context.Context) { + go s.loop(ctx) +} + +// Stop signals the loop to exit and waits for it to drain. +func (s *HibernationBillingSweeper) Stop(ctx context.Context) error { + s.once.Do(func() { close(s.stopCh) }) + select { + case <-s.doneCh: + case <-ctx.Done(): + return ctx.Err() + } + return nil +} + +func (s *HibernationBillingSweeper) loop(ctx context.Context) { + defer close(s.doneCh) + ticker := time.NewTicker(s.interval) + defer ticker.Stop() + // First sweep runs immediately so a hibernated sandbox created just before + // process start doesn't wait a full interval for its first bill. + s.safeSweep(ctx) + for { + select { + case <-ctx.Done(): + return + case <-s.stopCh: + return + case <-ticker.C: + s.safeSweep(ctx) + } + } +} + +func (s *HibernationBillingSweeper) safeSweep(ctx context.Context) { + defer func() { + if v := recover(); v != nil { + log.Printf("hibernation_billing: recovered from panic: %v", v) + } + }() + s.sweep(ctx) +} + +func (s *HibernationBillingSweeper) sweep(ctx context.Context) { + rows, err := s.store.ListHibernatedSandboxesForBilling(ctx) + if err != nil { + log.Printf("hibernation_billing: list failed: %v", err) + return + } + if len(rows) == 0 { + return + } + + // Bucket the tick to the current 5-minute wall boundary so the synthetic + // event ID is deterministic across retries. A retry within the same bucket + // dedupes at events-ingest (ON CONFLICT DO NOTHING on id). + now := time.Now() + bucketStart := now.Unix() / 300 * 300 + intervalSec := int(s.interval / time.Second) + + envelopes := make([]syntheticSandboxEvent, 0, len(rows)) + for _, r := range rows { + payload, err := json.Marshal(map[string]interface{}{ + "sandbox_id": r.SandboxID, + "cost_cents": 0, + "interval_s": intervalSec, + "memory_mb": 0, + "cpu_count": 0, + "disk_mb": r.DiskMB, + }) + if err != nil { + log.Printf("hibernation_billing: marshal payload for %s failed: %v", r.SandboxID, err) + continue + } + envelopes = append(envelopes, syntheticSandboxEvent{ + ID: fmt.Sprintf("hibernated:%s:%d", r.SandboxID, bucketStart), + Type: "usage_tick", + SandboxID: r.SandboxID, + OrgID: r.OrgID, + WorkerID: "", // intentionally empty — no live owner while hibernated + CellID: s.cellID, + Payload: payload, + Timestamp: now, + }) + } + if len(envelopes) == 0 { + return + } + + body, err := json.Marshal(envelopes) + if err != nil { + log.Printf("hibernation_billing: marshal batch failed: %v", err) + return + } + sendCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if err := s.client.SendBatch(sendCtx, body); err != nil { + log.Printf("hibernation_billing: send batch (%d envelopes) failed: %v", len(envelopes), err) + return + } + log.Printf("hibernation_billing: emitted %d disk-overage ticks for bucket=%d", len(envelopes), bucketStart) +} diff --git a/internal/db/usage.go b/internal/db/usage.go index 6ad49b33..78d1e726 100644 --- a/internal/db/usage.go +++ b/internal/db/usage.go @@ -74,6 +74,69 @@ func (s *Store) RecordScaleEvent(ctx context.Context, sandboxID, orgID string, m return tx.Commit(ctx) } +// HibernatedSandboxBilling is one row of the "billable while hibernated" set. +// Emitted by the hibernation-billing sweeper as a synthetic usage_tick so the +// edge autumn_meter attributes disk overage against the org for the whole time +// the qcow2 sits in Tigris (not just while the VM is running). +type HibernatedSandboxBilling struct { + SandboxID string + OrgID string + DiskMB int +} + +// ListHibernatedSandboxesForBilling returns every currently-hibernated sandbox +// in this cell's PG with its current disk envelope, filtered to those with +// disk_mb strictly above the 20 GB free allowance (rows at or below the free +// tier accrue nothing and would just be discarded downstream). One JOIN per +// sweep, indexed on status + started_at. +func (s *Store) ListHibernatedSandboxesForBilling(ctx context.Context) ([]HibernatedSandboxBilling, error) { + rows, err := s.pool.Query(ctx, + `SELECT ss.sandbox_id, ss.org_id::text, se.disk_mb + FROM sandbox_sessions ss + JOIN LATERAL ( + SELECT disk_mb FROM sandbox_scale_events + WHERE sandbox_id = ss.sandbox_id + ORDER BY (ended_at IS NULL) DESC, started_at DESC + LIMIT 1 + ) se ON TRUE + WHERE ss.status = 'hibernated' AND se.disk_mb > 20480`) + if err != nil { + return nil, err + } + defer rows.Close() + var out []HibernatedSandboxBilling + for rows.Next() { + var r HibernatedSandboxBilling + if err := rows.Scan(&r.SandboxID, &r.OrgID, &r.DiskMB); err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// GetCurrentDiskMB returns the disk_mb of the sandbox's most recent +// (open-preferred) scale event, falling back to the platform default 20480 if +// no scale event exists yet. Called per-sandbox per usage_ticker tick to stamp +// the current disk envelope onto usage_tick events for edge-side disk billing. +// A missing row is treated as the default (20 GB) rather than a hard error so +// a lookup failure never blocks memory/CPU billing. +func (s *Store) GetCurrentDiskMB(ctx context.Context, sandboxID string) (int, error) { + var diskMB int + err := s.pool.QueryRow(ctx, + `SELECT disk_mb FROM sandbox_scale_events + WHERE sandbox_id = $1 + ORDER BY (ended_at IS NULL) DESC, started_at DESC + LIMIT 1`, sandboxID).Scan(&diskMB) + if err != nil { + return 20480, err + } + if diskMB <= 0 { + return 20480, nil + } + return diskMB, nil +} + // GetSandboxOrgID looks up the org ID for a sandbox from the sessions table. func (s *Store) GetSandboxOrgID(ctx context.Context, sandboxID string) (string, error) { var orgID uuid.UUID diff --git a/internal/proxy/controlplane_proxy.go b/internal/proxy/controlplane_proxy.go index d621f707..e993fe41 100644 --- a/internal/proxy/controlplane_proxy.go +++ b/internal/proxy/controlplane_proxy.go @@ -246,8 +246,13 @@ func (p *ControlPlaneProxy) wakeHibernatedSandbox(ctx context.Context, sandboxID log.Printf("cp-proxy: waking sandbox %s on worker %s (region=%s)", sandboxID, worker.ID, region) - // Wake via gRPC with a generous timeout (cold boot + S3 download) - grpcCtx, cancel := context.WithTimeout(ctx, 90*time.Second) + // Wake via gRPC with a generous timeout (cold boot + S3 download). 5 min + // covers the platform's 256GB per-sandbox disk cap — a smaller ceiling + // silently strands large-disk sandboxes on any cross-worker wake because + // the target worker's chunked download + tar-extract can't complete in + // time. The client-side Cloudflare 100s edge deadline is a separate + // concern handled by the X-OSB-Async-Wake header on new SDKs. + grpcCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() _, err = grpcClient.WakeSandbox(grpcCtx, &pb.WakeSandboxRequest{ diff --git a/internal/proxy/sandbox_api_proxy.go b/internal/proxy/sandbox_api_proxy.go index c1eb0494..9d3eff69 100644 --- a/internal/proxy/sandbox_api_proxy.go +++ b/internal/proxy/sandbox_api_proxy.go @@ -311,7 +311,9 @@ func (p *SandboxAPIProxy) ProxyHandler(c echo.Context) error { } // File ops are synchronous (the caller wants the bytes/result now) so // handle+poll doesn't apply, but a cold restore can exceed Cloudflare's - // 100s → 524 (or the 90s WakeSandbox cap → 502). For clients that opt in + // 100s → 524 (WakeSandbox itself is capped at 5 min to cover the 256GB + // disk ceiling — Cloudflare's 100s edge deadline is the tighter bound + // on synchronous requests). For clients that opt in // (X-OSB-Async-Wake — the newer SDK, which retries on 503), kick the wake // to the background and return 503 "waking" + Retry-After; the warm retry // proxies normally. Older SDKs don't send the header and don't retry, so @@ -673,7 +675,10 @@ func (p *SandboxAPIProxy) wakeHibernatedSandbox(ctx context.Context, sandboxID s log.Printf("sandbox-api-proxy: waking sandbox %s on worker %s (region=%s)", sandboxID, worker.ID, region) - grpcCtx, cancel := context.WithTimeout(ctx, 90*time.Second) + // 5 min covers the platform's 256GB per-sandbox disk cap on cross-worker + // wake (chunked S3 download + tar-extract). The old 90s cap silently + // stranded any wake landing on a worker without a warm qcow2 cache. + grpcCtx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() _, err = grpcClient.WakeSandbox(grpcCtx, &pb.WakeSandboxRequest{ diff --git a/internal/qemu/snapshot.go b/internal/qemu/snapshot.go index fceac0c1..21751fd6 100644 --- a/internal/qemu/snapshot.go +++ b/internal/qemu/snapshot.go @@ -318,7 +318,16 @@ func (m *Manager) doHibernate(ctx context.Context, vm *VMInstance, checkpointSto sandboxID, time.Since(t1).Milliseconds(), float64(sizeBytes)/(1024*1024)) t2 := time.Now() - uploadCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + // 30 min ceiling covers the platform's 256GB per-sandbox disk cap + // (`OPENSANDBOX_MAX_DISK_MB` default 262144): archive size scales + // with USED bytes, and a ~200GB customer disk over Azure→Tigris at + // realistic same-region throughput (~150–350 MB/s) needs 10–20 min + // end-to-end (multipart PUT is 8MB × 4 concurrent, `blobstore/s3.go`). + // The previous 5-min cap tripped silently — the sync gRPC had already + // returned 200 so the customer saw "hibernated", but `uploaded_at` + // stayed NULL and any cross-worker wake refused with "hibernation + // upload not yet complete" (`api/sandbox.go:2470-2473`). + uploadCtx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() if _, err := checkpointStore.Upload(uploadCtx, checkpointKey, archivePath); err != nil { goroutineErr = fmt.Errorf("upload: %w", err) diff --git a/internal/worker/usage_ticker.go b/internal/worker/usage_ticker.go index 47574a44..0dbc3225 100644 --- a/internal/worker/usage_ticker.go +++ b/internal/worker/usage_ticker.go @@ -69,8 +69,16 @@ import ( // and small compared to the 300% drifts we were seeing pre-fix. If // parity post-deploy still flags meaningfully, add hook calls in // qemu.Manager.{Scale,Hibernate,Kill} next. +// scaleEventStore is the minimum store surface usage_ticker needs to stamp +// the current disk envelope onto usage_tick events. Kept as a narrow interface +// so tests can stub it without pulling in the full *db.Store. +type scaleEventStore interface { + GetCurrentDiskMB(ctx context.Context, sandboxID string) (int, error) +} + type UsageTicker struct { manager sandbox.Manager + store scaleEventStore sandboxDBs *sandbox.SandboxDBManager interval time.Duration costPerTickCs int // cents debited per FULL tick interval; scaled by actual interval below @@ -95,8 +103,10 @@ type UsageTicker struct { // costPerTickCs ≤ 0 defaults to 10 cents (so a steady-state $5 = 50 ticks // at the default interval ≈ 17 min). Cost is scaled by actual emit interval, // so short sandboxes pay proportionally less. -// nil manager or nil sandboxDBs returns nil (ticker disabled). -func NewUsageTicker(manager sandbox.Manager, sandboxDBs *sandbox.SandboxDBManager, interval time.Duration, costPerTickCs int) *UsageTicker { +// nil manager or nil sandboxDBs returns nil (ticker disabled). A nil store +// disables per-tick disk_mb stamping (payloads emit disk_mb=0 → the edge +// treats them as "no disk billing signal", falling back to the default). +func NewUsageTicker(manager sandbox.Manager, sandboxDBs *sandbox.SandboxDBManager, store scaleEventStore, interval time.Duration, costPerTickCs int) *UsageTicker { if manager == nil || sandboxDBs == nil { return nil } @@ -108,6 +118,7 @@ func NewUsageTicker(manager sandbox.Manager, sandboxDBs *sandbox.SandboxDBManage } return &UsageTicker{ manager: manager, + store: store, sandboxDBs: sandboxDBs, interval: interval, costPerTickCs: costPerTickCs, @@ -118,6 +129,23 @@ func NewUsageTicker(manager sandbox.Manager, sandboxDBs *sandbox.SandboxDBManage } } +// currentDiskMB looks up the sandbox's current disk envelope via the store. +// Fail-open: on any error we return the platform default (20480) so a store +// blip never breaks the memory/CPU emit path. The 20480 sentinel also lets +// the edge treat the tick as "no overage" (since 20480 is the free allowance). +func (t *UsageTicker) currentDiskMB(ctx context.Context, sandboxID string) int { + if t.store == nil { + return 20480 + } + q, cancel := context.WithTimeout(ctx, 500*time.Millisecond) + defer cancel() + mb, err := t.store.GetCurrentDiskMB(q, sandboxID) + if err != nil { + return 20480 + } + return mb +} + // Start begins the tick loop. Safe to call once; subsequent calls are no-ops. func (t *UsageTicker) Start(ctx context.Context) { go t.run(ctx) @@ -232,8 +260,12 @@ func (t *UsageTicker) tick(ctx context.Context) { // orgs ignore these (they debit cost_cents); pro orgs land them // in D1 usage_samples. MemoryMB/CpuCount come straight off the // running VM's tier — the worker owns the VM so these are exact. + // disk_mb comes from sandbox_scale_events (updated by scale + // events + inherited otherwise) so runtime resizes are naturally + // picked up on the next tick. "memory_mb": sb.MemoryMB, "cpu_count": sb.CpuCount, + "disk_mb": t.currentDiskMB(ctx, sb.ID), }); err != nil { log.Printf("usage_ticker: %s: LogEvent failed: %v", sb.ID, err) continue @@ -461,12 +493,16 @@ func (t *UsageTicker) flushSlice(sandboxID string, memoryMB, cpuCount int, start log.Printf("usage_ticker: flushSlice %s: Get failed: %v", sandboxID, err) return } + // flushSlice fires from lifecycle hooks (scale, destroy, hibernate, wake) + // with no request context in scope — use a fresh bounded one for the + // disk_mb lookup so a slow store call can't leak the caller's goroutine. if err := sdb.LogEvent("usage_tick", map[string]interface{}{ "sandbox_id": sandboxID, "cost_cents": t.scaledCost(intervalSec), "interval_s": intervalSec, "memory_mb": memoryMB, "cpu_count": cpuCount, + "disk_mb": t.currentDiskMB(context.Background(), sandboxID), }); err != nil { log.Printf("usage_ticker: flushSlice %s: LogEvent failed: %v", sandboxID, err) }