amadan

carlos / platform Public

console: publish account limits from the plan catalog

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 20e92283fb59c2ef0007f5a01809a426ee8036c9 parent 2c32794
7 files changed, +398 −0
  • cmd/console/main.go +17 −0
  • cmd/console/plancatalog_test.go +28 −0
  • internal/console/claim_workers.go +1 −0
  • internal/console/planlimits.go +127 −0
  • internal/console/planlimits_test.go +219 −0
  • internal/console/server.go +5 −0
  • internal/console/testdata/claim-routes.txt +1 −0
diff --git a/cmd/console/main.go b/cmd/console/main.go
index 192e023d4..d1f6ae81f 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -111,6 +111,7 @@ import (
"github.com/carlosframework/platform/internal/mail"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/objstore"
+ "github.com/carlosframework/platform/internal/plans"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/rootaws"
@@ -292,6 +293,11 @@ func serve() {
st := openServeDB(hostClaim)
storeProvisioner, storeDeployment := storeBackend()
storeScanner, seedAntivirus := storeScannerFor(storeDeployment)
+ catalog, err := planCatalog()
+ if err != nil {
+ // A broken catalog must never quietly mean "no limits".
+ log.Fatalf("console: CARLOS_PLAN_CATALOG: %v", err)
+ }
mailDep := mailDeployment()
mailProvisioner, dnsWriter := mailBackends(mailDep, mail.NewAWSProvisioner, dnsrec.NewRoute53Writer)
// One release.Store, named, because the senders sweep below needs the
@@ -391,6 +397,7 @@ func serve() {
StoreDeployment: storeDeployment,
StoreMaxPerApp: envCount("CARLOS_STORE_MAX_PER_APP", 0),
StoreOperators: storeOperators(),
+ PlanCatalog: catalog,
TemporaryApps: os.Getenv("CARLOS_TEMPORARY_APPS") == "1",
// CARLOS_ACME_DELEGATION_ZONE — the same variable the edge reads
// (internal/edge/wildcard.go's newWildcardFromEnv) for the same
@@ -1504,3 +1511,13 @@ func regionHAEnabled(getenv func(string) string) bool {
on, err := strconv.ParseBool(strings.TrimSpace(getenv("CARLOS_REGION_HA")))
return err == nil && on
}
+
+// planCatalog loads the plan catalog named by CARLOS_PLAN_CATALOG
+// (plans-caps-packs spec §4.1). Unset means plans are off: nil, nil.
+func planCatalog() (*plans.Catalog, error) {
+ path := strings.TrimSpace(os.Getenv("CARLOS_PLAN_CATALOG"))
+ if path == "" {
+ return nil, nil
+ }
+ return plans.LoadCatalog(path)
+}
diff --git a/cmd/console/plancatalog_test.go b/cmd/console/plancatalog_test.go
new file mode 100644
index 000000000..e3a2a0ec7
--- /dev/null
+++ b/cmd/console/plancatalog_test.go
@@ -0,0 +1,28 @@
+package main
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func TestLoadPlanCatalogEnv(t *testing.T) {
+ t.Setenv("CARLOS_PLAN_CATALOG", "")
+ if c, err := planCatalog(); c != nil || err != nil {
+ t.Fatalf("unset: %v, %v", c, err)
+ }
+ t.Setenv("CARLOS_PLAN_CATALOG", "../../internal/plans/testdata/example-catalog.json")
+ if c, err := planCatalog(); c == nil || err != nil {
+ t.Fatalf("example: %v, %v", c, err)
+ }
+ bad := filepath.Join(t.TempDir(), "bad.json")
+ if err := os.WriteFile(bad, []byte(`{"version": ""}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ for _, p := range []string{bad, filepath.Join(t.TempDir(), "missing.json")} {
+ t.Setenv("CARLOS_PLAN_CATALOG", p)
+ if c, err := planCatalog(); c != nil || err == nil {
+ t.Fatalf("%s: %v, %v; want an error", p, c, err)
+ }
+ }
+}
diff --git a/internal/console/claim_workers.go b/internal/console/claim_workers.go
index 21983223d..c0df46a7d 100644
--- a/internal/console/claim_workers.go
+++ b/internal/console/claim_workers.go
@@ -13,6 +13,7 @@ func (s *Server) registerClaimWorkers() {
"StartInfraPass": s.StartInfraPass,
"StartScanPass": s.StartScanPass,
"StartExpiryPass": s.StartExpiryPass,
+ "StartPlanLimitsPass": s.StartPlanLimitsPass,
"StartRunSweep": s.StartRunSweep,
"StartSeedsPass": s.StartSeedsPass,
"StartRelay": s.StartRelay,
diff --git a/internal/console/planlimits.go b/internal/console/planlimits.go
new file mode 100644
index 000000000..d5e510f47
--- /dev/null
+++ b/internal/console/planlimits.go
@@ -0,0 +1,127 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/plans"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const (
+ planLimitsPassInterval = time.Minute
+ planLimitsPassBatch = 200
+)
+
+// StartPlanLimitsPass keeps every account's limits record in the bucket in
+// step with its plan row and the configured catalog: now, then every
+// minute (StartExpiryPass's shape). With no catalog it does nothing at all.
+func (s *Server) StartPlanLimitsPass(ctx context.Context) {
+ if s.cfg.Releases == nil || s.cfg.PlanCatalog == nil {
+ return
+ }
+ t := time.NewTicker(planLimitsPassInterval)
+ defer t.Stop()
+ for {
+ s.planLimitsPass(ctx, time.Now())
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ }
+ }
+}
+
+// planLimitsPass is one pass at the given clock: give unassigned accounts
+// the default plan, then publish every row the bucket is behind on. One
+// account's failure never stops the others.
+func (s *Server) planLimitsPass(ctx context.Context, now time.Time) {
+ cat := s.cfg.PlanCatalog
+ if s.cfg.Releases == nil || cat == nil {
+ return
+ }
+ if _, err := s.store.EnsureDefaultPlans(cat.DefaultPlan, now); err != nil {
+ log.Printf("console: plan limits: default plans: %v", err)
+ return
+ }
+ rows, err := s.store.PlansToPublish(cat.ID(), planLimitsPassBatch)
+ if err != nil {
+ log.Printf("console: plan limits: list: %v", err)
+ return
+ }
+ for _, row := range rows {
+ if ctx.Err() != nil {
+ return
+ }
+ if err := s.publishAccountLimits(ctx, row, now); err != nil {
+ log.Printf("console: plan limits: %s: %v", row.Sqid, err)
+ }
+ }
+}
+
+// publishAccountLimits writes row's effective limits to its account's
+// record. An assignment the catalog cannot price (a plan or pack it no
+// longer has) writes nothing, so boxes keep the last good record. A record
+// already at this revision or newer, under this catalog, is left alone.
+func (s *Server) publishAccountLimits(ctx context.Context, row AccountPlanRow, now time.Time) error {
+ cat := s.cfg.PlanCatalog
+ if cat == nil {
+ return errors.New("no plan catalog")
+ }
+ limits, err := cat.Effective(row.Plan, row.Packs, row.Sponsored)
+ if err != nil {
+ return err
+ }
+ rec := &plans.Record{
+ Account: row.Sqid,
+ CatalogID: cat.ID(),
+ Plan: row.Plan,
+ Source: row.Source,
+ Revision: row.Revision,
+ Limits: limits,
+ IssuedAt: now.UTC().Truncate(time.Second),
+ }
+ for attempt := 0; attempt < 2; attempt++ {
+ cur, etag, err := plans.GetRecord(ctx, s.cfg.Releases, row.Sqid)
+ if err != nil {
+ return err
+ }
+ if cur != nil && cur.CatalogID == rec.CatalogID && cur.Revision >= rec.Revision {
+ return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID())
+ }
+ _, err = plans.PutRecord(ctx, s.cfg.Releases, rec, etag)
+ if errors.Is(err, release.ErrCASConflict) {
+ continue
+ }
+ if err != nil {
+ return err
+ }
+ return s.store.MarkPlanPublished(row.AccountID, row.Revision, cat.ID())
+ }
+ return errors.New("lost two write races; the next pass retries")
+}
+
+// EffectiveLimits is acct's limits under the configured catalog, or nil
+// when plans are off. An account with no row is on the default plan.
+func (s *Server) EffectiveLimits(acct *Account) (*plans.Limits, error) {
+ cat := s.cfg.PlanCatalog
+ if cat == nil {
+ return nil, nil
+ }
+ row, err := s.store.AccountPlan(acct.ID)
+ if err != nil {
+ return nil, err
+ }
+ plan, packs := cat.DefaultPlan, map[string]int(nil)
+ var sponsored *plans.Limits
+ if row != nil {
+ plan, packs, sponsored = row.Plan, row.Packs, row.Sponsored
+ }
+ l, err := cat.Effective(plan, packs, sponsored)
+ if err != nil {
+ return nil, err
+ }
+ return &l, nil
+}
diff --git a/internal/console/planlimits_test.go b/internal/console/planlimits_test.go
new file mode 100644
index 000000000..1bbdcedd1
--- /dev/null
+++ b/internal/console/planlimits_test.go
@@ -0,0 +1,219 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "os"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/plans"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const exampleCatalogPath = "../plans/testdata/example-catalog.json"
+
+func examplePlanCatalog(t *testing.T) *plans.Catalog {
+ t.Helper()
+ c, err := plans.LoadCatalog(exampleCatalogPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return c
+}
+
+// bumpedCatalog is the example catalog with free's running_things set to n
+// under a new version.
+func bumpedCatalog(t *testing.T, version string, n string) *plans.Catalog {
+ t.Helper()
+ b, err := os.ReadFile(exampleCatalogPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ s := strings.Replace(string(b), `"version": "2026-10-20-v1"`, `"version": "`+version+`"`, 1)
+ s = strings.Replace(s, `"running_things": 3,`, `"running_things": `+n+`,`, 1)
+ c, err := plans.ParseCatalog([]byte(s))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return c
+}
+
+// planAccount adds a console account with a valid sqid and returns it.
+func planAccount(t *testing.T, fx *consoleFixture, id int64) *Account {
+ t.Helper()
+ sqid := domains.Sqid(id)
+ if err := fx.st.CreateAccount(id, sqid, "acct", "owner"+sqid+"@example.com", time.Unix(1_700_000_000, 0)); err != nil {
+ t.Fatal(err)
+ }
+ acct, err := fx.st.AccountBySqid(sqid)
+ if err != nil || acct == nil {
+ t.Fatalf("account %s: %v", sqid, err)
+ }
+ return acct
+}
+
+func readLimits(t *testing.T, fx *consoleFixture, sqid string) (*plans.Record, string) {
+ t.Helper()
+ r, etag, err := plans.GetRecord(context.Background(), fx.rel, sqid)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return r, etag
+}
+
+func TestPlanLimitsPassOffWithoutACatalog(t *testing.T) {
+ fx := newConsoleFixture(t)
+ acct := planAccount(t, fx, 501)
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ if r, _ := readLimits(t, fx, acct.Sqid); r != nil {
+ t.Fatalf("wrote a record with no catalog: %+v", r)
+ }
+ if row, _ := fx.st.AccountPlan(acct.ID); row != nil {
+ t.Fatalf("inserted a plan row with no catalog: %+v", row)
+ }
+}
+
+func TestPlanLimitsPassPublishesDefaults(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
+ acct := planAccount(t, fx, 501)
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ r, _ := readLimits(t, fx, acct.Sqid)
+ if r == nil {
+ t.Fatal("no record published")
+ }
+ if r.Plan != "free" || r.Source != PlanSourceDefault || r.Limits.RunningThings != 3 || r.CatalogID != fx.srv.cfg.PlanCatalog.ID() {
+ t.Fatalf("record %+v", r)
+ }
+ row, _ := fx.st.AccountPlan(acct.ID)
+ if row.PublishedRevision != row.Revision || row.PublishedCatalog != fx.srv.cfg.PlanCatalog.ID() {
+ t.Fatalf("row not marked published: %+v", row)
+ }
+}
+
+func TestPlanLimitsPassRepublishesOnCatalogChange(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
+ acct := planAccount(t, fx, 501)
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ next := bumpedCatalog(t, "2026-10-21-v2", "4")
+ fx.srv.cfg.PlanCatalog = next
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ r, etag := readLimits(t, fx, acct.Sqid)
+ if r.CatalogID != next.ID() || r.Limits.RunningThings != 4 {
+ t.Fatalf("not republished: %+v", r)
+ }
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ if _, again := readLimits(t, fx, acct.Sqid); again != etag {
+ t.Fatal("an up-to-date record was rewritten")
+ }
+}
+
+func TestPlanLimitsPassSkipsAnOrphanedAssignment(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
+ orphan := planAccount(t, fx, 501)
+ other := planAccount(t, fx, 502)
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ _, before := readLimits(t, fx, orphan.Sqid)
+ if _, err := fx.st.SetAccountPlan(orphan.ID, "gold", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := fx.st.SetAccountPlan(other.ID, "personal", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ if _, after := readLimits(t, fx, orphan.Sqid); after != before {
+ t.Fatal("the orphan's last good record was replaced")
+ }
+ if row, _ := fx.st.AccountPlan(orphan.ID); row.PublishedRevision == row.Revision {
+ t.Fatal("the orphan was marked published")
+ }
+ if r, _ := readLimits(t, fx, other.Sqid); r.Plan != "personal" {
+ t.Fatalf("the other account was not published: %+v", r)
+ }
+}
+
+func TestPublishNeverOverwritesANewerRecord(t *testing.T) {
+ fx := newConsoleFixture(t)
+ cat := examplePlanCatalog(t)
+ fx.srv.cfg.PlanCatalog = cat
+ acct := planAccount(t, fx, 501)
+ newer := &plans.Record{Account: acct.Sqid, CatalogID: cat.ID(), Plan: "personal", Source: PlanSourceOperator, Revision: 5}
+ if _, err := plans.PutRecord(context.Background(), fx.rel, newer, ""); err != nil {
+ t.Fatal(err)
+ }
+ for range 3 {
+ if _, err := fx.st.SetAccountPlan(acct.ID, "free", nil, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ }
+ row, _ := fx.st.AccountPlan(acct.ID)
+ if err := fx.srv.publishAccountLimits(context.Background(), *row, planNow); err != nil {
+ t.Fatal(err)
+ }
+ if r, _ := readLimits(t, fx, acct.Sqid); r.Revision != 5 || r.Plan != "personal" {
+ t.Fatalf("newer record overwritten: %+v", r)
+ }
+ if row, _ := fx.st.AccountPlan(acct.ID); row.PublishedRevision != 3 {
+ t.Fatalf("published_revision %d, want 3", row.PublishedRevision)
+ }
+}
+
+// planFailingPuts is a bucket whose next n Puts fail.
+type planFailingPuts struct {
+ release.Store
+ n atomic.Int32
+}
+
+func (f *planFailingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if f.n.Add(-1) >= 0 {
+ return "", errors.New("bucket unavailable")
+ }
+ return f.Store.Put(ctx, key, body, ifMatch)
+}
+
+func TestPublishRetriesAfterABucketFailure(t *testing.T) {
+ fx := newConsoleFixture(t)
+ fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
+ acct := planAccount(t, fx, 501)
+ fp := &planFailingPuts{Store: fx.rel}
+ fx.srv.cfg.Releases = fp
+ if _, err := fx.st.EnsureDefaultPlans("free", planNow); err != nil {
+ t.Fatal(err)
+ }
+ fp.n.Store(1000) // every Put fails during the first pass
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ if row, _ := fx.st.AccountPlan(acct.ID); row.PublishedRevision != 0 {
+ t.Fatal("marked published although the write failed")
+ }
+ fp.n.Store(0)
+ fx.srv.planLimitsPass(context.Background(), planNow)
+ if r, _ := readLimits(t, fx, acct.Sqid); r == nil {
+ t.Fatal("not published on the next pass")
+ }
+}
+
+func TestEffectiveLimits(t *testing.T) {
+ fx := newConsoleFixture(t)
+ acct := planAccount(t, fx, 501)
+ if l, err := fx.srv.EffectiveLimits(acct); l != nil || err != nil {
+ t.Fatalf("no catalog: %+v, %v", l, err)
+ }
+ fx.srv.cfg.PlanCatalog = examplePlanCatalog(t)
+ l, err := fx.srv.EffectiveLimits(acct)
+ if err != nil || l == nil || l.RunningThings != 3 {
+ t.Fatalf("default: %+v, %v", l, err)
+ }
+ if _, err := fx.st.SetAccountPlan(acct.ID, "personal", map[string]int{"apps": 1}, PlanSourceOperator, nil, "r", "ops@example.com", planNow); err != nil {
+ t.Fatal(err)
+ }
+ l, err = fx.srv.EffectiveLimits(acct)
+ if err != nil || l.RunningThings != 30 {
+ t.Fatalf("personal + apps: %+v, %v", l, err)
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index a54fa938f..2d8953324 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -47,6 +47,7 @@ import (
"github.com/carlosframework/platform/internal/mail"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/objstore"
+ "github.com/carlosframework/platform/internal/plans"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/publicrelease"
"github.com/carlosframework/platform/internal/registry"
@@ -715,6 +716,10 @@ type Config struct {
// (handleAPIStoreCreate) — a ceiling only the automatic path respected
// would be a ceiling anyone could step over by hand.
StoreMaxPerApp int
+ // PlanCatalog is the deployment's plan catalog (CARLOS_PLAN_CATALOG,
+ // plans-caps-packs spec §4.1). nil means plans are off: no account has
+ // limits, no limits record is published, and the plan door answers 409.
+ PlanCatalog *plans.Catalog
// TemporaryApps turns on POST /api/cli/apps/{account}/new-temporary
// (temporary static sites spec §3). Off by default, and only switched on
// after every box runs a binary that carries the tmp- reservation and
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt
index a29e86d85..614166d94 100644
--- a/internal/console/testdata/claim-routes.txt
+++ b/internal/console/testdata/claim-routes.txt
@@ -286,6 +286,7 @@ worker StartGeoipJobs
worker StartHostReports
worker StartInfraPass
worker StartMailEventPush
+worker StartPlanLimitsPass
worker StartRelay
worker StartRunSweep
worker StartScanPass