amadan

carlos / platform Public

host: read account limits each pass, marker-gated, with a box off switch

plansPass rides beside telemetryPass. No marker or a disabled one: one
read, nothing applied. CARLOS_PLAN_LIMITS=off: no reads at all. Errors keep
the last good limits; an account with no record has none.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 3205be39c417ef074079978b608468462c5b032a parent 30ace3c
4 files changed, +310 −0
  • internal/host/agent.go +3 −0
  • internal/host/plans_pass.go +117 −0
  • internal/host/plans_pass_test.go +189 −0
  • internal/hostenv/hostenv.go +1 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 559c0a27c..ee1b40ad7 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -2808,6 +2808,8 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
telemetryPass(ctx, store, reg)
mark("telemetry")
+ plansPass(ctx, store, reg, os.Getenv)
+ mark("plans")
runMovesPass(ctx, store)
mark("moves")
runRunsPass(ctx)
@@ -3393,6 +3395,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta
bs.refreshAll(ctx)
}
telemetryPass(ctx, store, reg)
+ plansPass(ctx, store, reg, os.Getenv)
runMovesPass(ctx, store)
runRunsPass(ctx)
deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
diff --git a/internal/host/plans_pass.go b/internal/host/plans_pass.go
new file mode 100644
index 000000000..f2daf51aa
--- /dev/null
+++ b/internal/host/plans_pass.go
@@ -0,0 +1,117 @@
+// The plans pass (plans-caps-packs spec §7, plans 2): learn whether plans are
+// on for this deployment and, only if they are, read the limits record of
+// every account this box serves into a last-known-good cache the activator
+// asks. Rides the same passes as telemetryPass.
+//
+// Self-hosters are never interrupted: with no marker (no catalog on the
+// console) the pass reads one object and applies nothing, a box with
+// CARLOS_PLAN_LIMITS=off reads nothing at all, and an account with no record
+// runs exactly as it always has.
+package host
+
+import (
+ "context"
+ "log"
+ "os"
+ "strings"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/plans"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// planLimitsSnapshot is immutable once stored; nil in planLimits means plans
+// are off on this box.
+type planLimitsSnapshot struct {
+ byAccount map[string]plans.Limits
+}
+
+var planLimits atomic.Pointer[planLimitsSnapshot]
+
+// planLimitsFailures counts consecutive failing passes for log throttling.
+var planLimitsFailures atomic.Int64
+
+// planLimitsSwitchOn reads CARLOS_PLAN_LIMITS: off, false, 0 or no turn plans
+// off on this box; anything else, including unset, leaves them to the marker.
+func planLimitsSwitchOn(getenv func(string) string) bool {
+ switch strings.ToLower(strings.TrimSpace(getenv("CARLOS_PLAN_LIMITS"))) {
+ case "off", "false", "0", "no":
+ return false
+ }
+ return true
+}
+
+// planLimitsFor is sqid's limits, or false when plans are off or the account
+// has no record.
+func planLimitsFor(sqid string) (plans.Limits, bool) {
+ snap := planLimits.Load()
+ if snap == nil {
+ return plans.Limits{}, false
+ }
+ l, ok := snap.byAccount[sqid]
+ return l, ok
+}
+
+func plansPass(ctx context.Context, s release.Store, reg *registry.Registry, getenv func(string) string) {
+ if getenv == nil {
+ getenv = os.Getenv
+ }
+ if !planLimitsSwitchOn(getenv) {
+ planLimits.Store(nil)
+ return
+ }
+ marker, _, err := plans.GetMarker(ctx, s)
+ if err != nil {
+ planLimitsFailed("marker: %v", err)
+ return // keep the last good cache
+ }
+ if marker == nil || !marker.Enabled {
+ planLimits.Store(nil)
+ planLimitsFailures.Store(0)
+ return
+ }
+ routes, err := reg.List()
+ if err != nil {
+ planLimitsFailed("routes: %v", err)
+ return
+ }
+ prev := planLimits.Load()
+ next := &planLimitsSnapshot{byAccount: map[string]plans.Limits{}}
+ seen := map[string]bool{}
+ failed := false
+ for _, rt := range routes {
+ if rt.Kind == "status" {
+ continue
+ }
+ sqid := release.RouteAccount(rt)
+ if sqid == "" || seen[sqid] {
+ continue
+ }
+ seen[sqid] = true
+ rec, _, err := plans.GetLimits(ctx, s, sqid)
+ switch {
+ case err != nil:
+ failed = true
+ planLimitsFailed("%s: %v", sqid, err)
+ if prev != nil {
+ if l, ok := prev.byAccount[sqid]; ok {
+ next.byAccount[sqid] = l
+ }
+ }
+ case rec != nil:
+ next.byAccount[sqid] = rec.Limits
+ }
+ }
+ planLimits.Store(next)
+ if !failed {
+ planLimitsFailures.Store(0)
+ }
+}
+
+// planLimitsFailed logs the first failure of a run and every tenth after it.
+func planLimitsFailed(format string, args ...any) {
+ if n := planLimitsFailures.Add(1); n == 1 || n%10 == 0 {
+ log.Printf("agent: plans pass: "+format+" (keeping the last good limits)", args...)
+ }
+}
diff --git a/internal/host/plans_pass_test.go b/internal/host/plans_pass_test.go
new file mode 100644
index 000000000..4793a6f3c
--- /dev/null
+++ b/internal/host/plans_pass_test.go
@@ -0,0 +1,189 @@
+package host
+
+// These tests share the package-level plans cache, so they stay serial.
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/plans"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// plansCountingStore records every Get and can fail chosen keys.
+type plansCountingStore struct {
+ release.Store
+ mu sync.Mutex
+ gets []string
+ fail map[string]bool
+}
+
+func (c *plansCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ c.mu.Lock()
+ c.gets = append(c.gets, key)
+ failing := c.fail[key]
+ c.mu.Unlock()
+ if failing {
+ return nil, "", errors.New("bucket unavailable")
+ }
+ return c.Store.Get(ctx, key)
+}
+
+func (c *plansCountingStore) reset() {
+ c.mu.Lock()
+ c.gets, c.fail = nil, map[string]bool{}
+ c.mu.Unlock()
+}
+
+func plansWorld(t *testing.T) (*plansCountingStore, *registry.Registry) {
+ t.Helper()
+ t.Cleanup(func() { planLimits.Store(nil) })
+ s := &plansCountingStore{Store: release.NewDirStore(t.TempDir()), fail: map[string]bool{}}
+ reg := edgesTestRegistry(t)
+ for _, rt := range []registry.Route{
+ {Host: "a.example", Kind: "instance", Account: "bab", App: "web", Status: "ready"},
+ {Host: "b.example", Kind: "instance", Account: "bdk", App: "shop", Status: "ready"},
+ {Host: "status.example", Kind: "status", Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return s, reg
+}
+
+func putMarker(t *testing.T, s release.Store, enabled bool) {
+ t.Helper()
+ ctx := context.Background()
+ _, etag, err := plans.GetMarker(ctx, s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := plans.PutMarker(ctx, s, &plans.Marker{Enabled: enabled, CatalogID: "v1@0123456789ab"}, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func putAccountLimits(t *testing.T, s release.Store, sqid string, idle int) {
+ t.Helper()
+ ctx := context.Background()
+ _, etag, err := plans.GetLimits(ctx, s, sqid)
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := &plans.Record{Account: sqid, CatalogID: "v1@0123456789ab", Plan: "free", Revision: 1, Limits: plans.Limits{IdleSeconds: idle}}
+ if _, err := plans.PutLimits(ctx, s, r, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func plansNoEnv(string) string { return "" }
+
+func TestPlanLimitsSwitch(t *testing.T) {
+ for v, want := range map[string]bool{"": true, "on": true, "off": false, "OFF": false, "false": false, "0": false, "no": false, " off ": false} {
+ if got := planLimitsSwitchOn(func(string) string { return v }); got != want {
+ t.Errorf("CARLOS_PLAN_LIMITS=%q: on = %v, want %v", v, got, want)
+ }
+ }
+}
+
+func TestPlansPassOffReadsOnlyTheMarker(t *testing.T) {
+ s, reg := plansWorld(t)
+ putAccountLimits(t, s, "bab", 90)
+ s.reset()
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if !slices.Equal(s.gets, []string{plans.MarkerKey()}) {
+ t.Fatalf("reads %v, want only the marker", s.gets)
+ }
+ if _, ok := planLimitsFor("bab"); ok {
+ t.Fatal("limits applied with no marker")
+ }
+}
+
+func TestPlansPassBoxSwitchOffReadsNothing(t *testing.T) {
+ s, reg := plansWorld(t)
+ putMarker(t, s, true)
+ putAccountLimits(t, s, "bab", 90)
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if _, ok := planLimitsFor("bab"); !ok {
+ t.Fatal("setup: limits not loaded")
+ }
+ s.reset()
+ plansPass(context.Background(), s, reg, func(string) string { return "off" })
+ if len(s.gets) != 0 {
+ t.Fatalf("switched-off box read %v", s.gets)
+ }
+ if _, ok := planLimitsFor("bab"); ok {
+ t.Fatal("switched-off box kept limits")
+ }
+}
+
+func TestPlansPassLoadsServedAccounts(t *testing.T) {
+ s, reg := plansWorld(t)
+ putMarker(t, s, true)
+ putAccountLimits(t, s, "bab", 90)
+ putAccountLimits(t, s, "bdk", 900)
+ putAccountLimits(t, s, "zzz", 60)
+ s.reset()
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
+ t.Fatalf("bab: %+v, %v", l, ok)
+ }
+ if l, ok := planLimitsFor("bdk"); !ok || l.IdleSeconds != 900 {
+ t.Fatalf("bdk: %+v, %v", l, ok)
+ }
+ if slices.Contains(s.gets, plans.LimitsKey("zzz")) {
+ t.Fatal("read the limits of an account with no routes on this box")
+ }
+}
+
+func TestPlansPassAbsentRecordMeansNoLimits(t *testing.T) {
+ s, reg := plansWorld(t)
+ putMarker(t, s, true)
+ putAccountLimits(t, s, "bab", 90)
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if _, ok := planLimitsFor("bdk"); ok {
+ t.Fatal("an account with no record got limits")
+ }
+}
+
+func TestPlansPassKeepsLastGoodOnErrors(t *testing.T) {
+ s, reg := plansWorld(t)
+ putMarker(t, s, true)
+ putAccountLimits(t, s, "bab", 90)
+ putAccountLimits(t, s, "bdk", 900)
+ plansPass(context.Background(), s, reg, plansNoEnv)
+
+ s.fail[plans.MarkerKey()] = true
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
+ t.Fatalf("a marker read error dropped bab: %+v, %v", l, ok)
+ }
+
+ s.reset()
+ putAccountLimits(t, s, "bdk", 600)
+ s.fail[plans.LimitsKey("bab")] = true
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if l, ok := planLimitsFor("bab"); !ok || l.IdleSeconds != 90 {
+ t.Fatalf("a record read error dropped bab: %+v, %v", l, ok)
+ }
+ if l, _ := planLimitsFor("bdk"); l.IdleSeconds != 600 {
+ t.Fatalf("bdk not refreshed: %+v", l)
+ }
+}
+
+func TestPlansPassClearsWhenDisabled(t *testing.T) {
+ s, reg := plansWorld(t)
+ putMarker(t, s, true)
+ putAccountLimits(t, s, "bab", 90)
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ putMarker(t, s, false)
+ plansPass(context.Background(), s, reg, plansNoEnv)
+ if _, ok := planLimitsFor("bab"); ok {
+ t.Fatal("limits kept after plans were disabled")
+ }
+}
diff --git a/internal/hostenv/hostenv.go b/internal/hostenv/hostenv.go
index 26458e0ee..e89154183 100644
--- a/internal/hostenv/hostenv.go
+++ b/internal/hostenv/hostenv.go
@@ -84,6 +84,7 @@ var Optional = []string{
"CARLOS_LAYOUT_DEFAULT",
"CARLOS_MOVES_PHASE1",
"CARLOS_MOVE_READY_DELTA_BYTES",
+ "CARLOS_PLAN_LIMITS",
"CARLOS_REQUIRE_POINTER_DIGEST",
"CARLOS_S3_ENDPOINT",
"CARLOS_S3_QUIRKS",