amadan
provision: converge the app policy into the registry (#328, #329)
Reconcile reads each in-scope app's signed operator policy through a long-lived PolicyCache, verifies it with the release key, and converges egress_allow, size and policy_expires_at in one SetPolicy per row. Fresh rows (ready, held, and the seed swap's import row) are minted with the columns (PF-2). Refused, unverifiable and expired policies fail closed and are reported in PolicyReadStates; a transient failure keeps a row's columns only until its own recorded expiry. Sidecar members get none. The cache keeps reads inside spec §3a (ruling 11): one ETag listing of control/app-policy/ per 30 to 33 minute backstop, a GET only for a changed ETag, a moved app generation, a failed read, or an app new to the cache whose scope moved since the listing's snapshot (Astra plan R3-2). The agent builds one cache after arming the change index and hands it to both provisioning loops. Counting-store witnesses pin the budget, including a fleet box's day through the real store door.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
10 files changed,
+1654
−14
internal/edge/fleetapi_e2e_test.go+212 −1internal/host/agent.go+17 −1internal/provision/apppolicy_cache.go+270 −0internal/provision/apppolicy_cache_test.go+384 −0internal/provision/apppolicy_reconcile.go+232 −0internal/provision/apppolicy_reconcile_test.go+446 −0internal/provision/reconcile.go+86 −7internal/provision/seedjournal_test.go+1 −1internal/provision/seedswap.go+4 −2internal/provision/seedswap_test.go+2 −2
diff --git a/internal/edge/fleetapi_e2e_test.go b/internal/edge/fleetapi_e2e_test.go| index d28c4ad69..3d6c71959 100644 |
| --- a/internal/edge/fleetapi_e2e_test.go |
| +++ b/internal/edge/fleetapi_e2e_test.go |
| @@ -30,17 +30,21 @@ import ( |
| "crypto/ed25519" |
| "encoding/json" |
| "errors" |
| + "io" |
| "net/http/httptest" |
| "net/url" |
| "os" |
| "path/filepath" |
| "runtime" |
| "slices" |
| + "strconv" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| "github.com/carlosframework/platform/internal/applog" |
| + "github.com/carlosframework/platform/internal/apppolicy" |
| "github.com/carlosframework/platform/internal/domains" |
| "github.com/carlosframework/platform/internal/edge" |
| "github.com/carlosframework/platform/internal/fleetapi" |
| @@ -99,6 +103,14 @@ type fleetDoorWorld struct { |
| // channels connected — the same credential that dials the tunnel is the one |
| // their store clients authenticate with, which is F1/F2 stated as a fixture. |
| func newFleetDoorWorld(t *testing.T) *fleetDoorWorld { |
| + t.Helper() |
| + return newFleetDoorWorldWith(t, nil) |
| +} |
| + |
| +// newFleetDoorWorldWith is newFleetDoorWorld with the door's store handle |
| +// wrapped by doorStore (nil: the DirStore itself), so a test can count what |
| +// the door asks of the bucket. |
| +func newFleetDoorWorldWith(t *testing.T, doorStore func(*release.DirStore) release.Store) *fleetDoorWorld { |
| t.Helper() |
| w := &fleetDoorWorld{ |
| sqid: domains.Sqid(fapiAcctID), |
| @@ -151,7 +163,11 @@ func newFleetDoorWorld(t *testing.T) *fleetDoorWorld { |
| // reserved hostname (here the listener's own, as in production |
| // CARLOS_TUNNEL_HOST). |
| w.hub = edge.NewTunnelHub(u.Hostname(), nil) |
| - e.EnableStatic(w.store, w.pub, filepath.Join(t.TempDir(), "cache")) |
| + var door release.Store = w.store |
| + if doorStore != nil { |
| + door = doorStore(w.store) |
| + } |
| + e.EnableStatic(door, w.pub, filepath.Join(t.TempDir(), "cache")) |
| e.EnableTunnels(w.hub) |
| if err := e.Reload(); err != nil { |
| t.Fatalf("Reload: %v", err) |
| @@ -553,6 +569,201 @@ func TestFleetBoxConvergesWithoutBucketCredentials(t *testing.T) { |
| t.Log("(g) log chunks ship, list and prune through the door; neighbouring keys stay refused") |
| } |
| +// doorCountingStore is the bucket as the store door reaches it, counting |
| +// what spec §3a's budget is stated in: listings that reach |
| +// control/app-policy/ (List, ListETags) and reads of keys under it (Get, |
| +// GetIfChanged, Open). Everything else passes uncounted; every other method |
| +// is the DirStore's own. |
| +type doorCountingStore struct { |
| + *release.DirStore |
| + |
| + mu sync.Mutex |
| + lists int |
| + listETags int |
| + gets map[string]int |
| +} |
| + |
| +func doorPolicyPrefix(prefix string) bool { |
| + return strings.HasPrefix(prefix, apppolicy.Prefix) || strings.HasPrefix(apppolicy.Prefix, prefix) |
| +} |
| + |
| +func (c *doorCountingStore) countGet(key string) { |
| + if strings.HasPrefix(key, apppolicy.Prefix) { |
| + c.mu.Lock() |
| + c.gets[key]++ |
| + c.mu.Unlock() |
| + } |
| +} |
| + |
| +func (c *doorCountingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + if doorPolicyPrefix(prefix) { |
| + c.mu.Lock() |
| + c.lists++ |
| + c.mu.Unlock() |
| + } |
| + return c.DirStore.List(ctx, prefix) |
| +} |
| + |
| +func (c *doorCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + if doorPolicyPrefix(prefix) { |
| + c.mu.Lock() |
| + c.listETags++ |
| + c.mu.Unlock() |
| + } |
| + return c.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +func (c *doorCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + c.countGet(key) |
| + return c.DirStore.Get(ctx, key) |
| +} |
| + |
| +func (c *doorCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| + c.countGet(key) |
| + return c.DirStore.GetIfChanged(ctx, key, etag) |
| +} |
| + |
| +func (c *doorCountingStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) { |
| + c.countGet(key) |
| + return c.DirStore.Open(ctx, key) |
| +} |
| + |
| +// take returns the counts since the last take and resets them. |
| +func (c *doorCountingStore) take() (lists, listETags int, gets map[string]int) { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + lists, listETags, gets = c.lists, c.listETags, c.gets |
| + c.lists, c.listETags, c.gets = 0, 0, map[string]int{} |
| + return lists, listETags, gets |
| +} |
| + |
| +func sumGets(gets map[string]int) int { |
| + n := 0 |
| + for _, v := range gets { |
| + n += v |
| + } |
| + return n |
| +} |
| + |
| +// TestFleetBoxPolicyReadsKeepTheRequestBudget is the §3a box budget |
| +// (ruling 11) through the store door (Astra plan R3-4): a fleet box has no |
| +// change index, so its policy reads are the backstop's ETag listing through |
| +// the door and GETs only for policies that changed, counted where the door |
| +// reaches the bucket. Fails on any increase. |
| +func TestFleetBoxPolicyReadsKeepTheRequestBudget(t *testing.T) { |
| + ctx := context.Background() |
| + var cs *doorCountingStore |
| + w := newFleetDoorWorldWith(t, func(d *release.DirStore) release.Store { |
| + cs = &doorCountingStore{DirStore: d, gets: map[string]int{}} |
| + return cs |
| + }) |
| + now := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC) |
| + w.put(t, "accounts/"+w.sqid+"/account.json", fapiAccountManifest(fapiAcctID, w.sqid, "Orchard")) |
| + host := func(app string) string { return "a." + app + "." + fapiDomain } |
| + for _, app := range []string{"shop", "cart", "blog"} { |
| + if err := provision.PutManifest(ctx, w.store, w.sqid, app, provision.Manifest{ |
| + Schema: 1, App: app, Domain: app + "." + fapiDomain, Regions: []string{fapiRegion}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := provision.CreateRecord(ctx, w.store, provision.Record{ |
| + Schema: 1, Host: host(app), Slug: "a", App: app, Account: w.sqid, |
| + Region: fapiRegion, Channel: fapiChannel, CreatedAt: now, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // The console's write and its re-sign: the same revision, signed now. |
| + sign := func(app string) { |
| + t.Helper() |
| + _, etag, err := w.store.Get(ctx, apppolicy.Key(w.sqid, app)) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal(err) |
| + } |
| + p := apppolicy.Policy{V: 1, Account: w.sqid, App: app, Revision: 1, SignedAt: now, |
| + EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"} |
| + if err := apppolicy.PutIfMatch(ctx, w.store, w.priv, p, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + sign("shop") |
| + sign("cart") |
| + |
| + boxReg, err := registry.Open(filepath.Join(t.TempDir(), "box-registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + cfg := provision.ReconcileConfig{Region: fapiRegion, DataDir: t.TempDir(), ReleasePub: w.pub, |
| + Now: func() time.Time { return now }, Policies: provision.NewPolicyCache(nil)} |
| + pass := func() { |
| + t.Helper() |
| + if err := provision.Reconcile(ctx, w.boxA, boxReg, cfg); err != nil { |
| + t.Fatalf("provision.Reconcile through the door: %v", err) |
| + } |
| + } |
| + wantRow := func(app, egress, size string, signed time.Time) { |
| + t.Helper() |
| + rt, err := boxReg.Get(host(app)) |
| + if err != nil || rt == nil { |
| + t.Fatalf("route %s = %+v, %v", host(app), rt, err) |
| + } |
| + exp := "" |
| + if egress != "" { |
| + exp = strconv.FormatInt(signed.Add(apppolicy.Lifetime).Unix(), 10) |
| + } |
| + if rt.EgressAllow != egress || rt.Size != size || rt.PolicyExpiresAt != exp { |
| + t.Fatalf("%s: egress %q size %q expires %q, want %q %q %q", app, rt.EgressAllow, rt.Size, rt.PolicyExpiresAt, egress, size, exp) |
| + } |
| + } |
| + blogKey := apppolicy.Key(w.sqid, "blog") |
| + |
| + // Pass 1, the first backstop. |
| + signed := now |
| + pass() |
| + lists, listETags, gets := cs.take() |
| + if listETags != 1 || lists != 0 || sumGets(gets) != 2 || gets[blogKey] != 0 { |
| + t.Fatalf("first pass: %d ETag listings, %d LISTs, GETs %v; want 1, 0, one per policy", listETags, lists, gets) |
| + } |
| + wantRow("shop", "10.0.5.20", "large", signed) |
| + wantRow("cart", "10.0.5.20", "large", signed) |
| + wantRow("blog", "", "", signed) |
| + |
| + // A quiet day, passes 5 minutes apart. |
| + for i := 0; i < 288; i++ { |
| + now = now.Add(5 * time.Minute) |
| + pass() |
| + } |
| + lists, listETags, gets = cs.take() |
| + if listETags > 48 || sumGets(gets) != 0 || lists != 0 { |
| + t.Fatalf("a quiet day: %d ETag listings (at most 48), GETs %v (want none), %d LISTs (want none)", listETags, gets, lists) |
| + } |
| + if listETags < 36 { |
| + t.Fatalf("a quiet day: %d ETag listings; the 30 to 33 minute backstop should have run about 41 times", listETags) |
| + } |
| + |
| + // A day with the console's two re-signs, at hours 6 and 18. |
| + start := now |
| + resigned := map[time.Duration]bool{} |
| + for i := 0; i < 288; i++ { |
| + now = now.Add(5 * time.Minute) |
| + for _, at := range []time.Duration{6 * time.Hour, 18 * time.Hour} { |
| + if !resigned[at] && !now.Before(start.Add(at)) { |
| + resigned[at] = true |
| + sign("shop") |
| + sign("cart") |
| + signed = now |
| + } |
| + } |
| + pass() |
| + } |
| + lists, listETags, gets = cs.take() |
| + if listETags > 48 || sumGets(gets) != 4 || gets[blogKey] != 0 || lists != 0 { |
| + t.Fatalf("a day with two re-signs: %d ETag listings (at most 48), GETs %v (want 2 per live policy), %d LISTs (want none)", listETags, gets, lists) |
| + } |
| + wantRow("shop", "10.0.5.20", "large", signed) |
| + wantRow("cart", "10.0.5.20", "large", signed) |
| +} |
| + |
| // TestFleetBoxCatchallYieldsToAnotherTenantsClaim (#215 review): the store |
| // door withholds another account's claim BODY from a fleet box, but must |
| // still list its NAME. An instance catch-all derives no <label>.<parent> |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index ef5e93ddf..147304d9e 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -803,6 +803,10 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| if v := armChangeIndex(agentCtx, os.Getenv, openDeployStore); v != nil { |
| go recordViewBackstopLoop(agentCtx, v) |
| } |
| + // The app policy cache (#328, #329), built once the index is armed |
| + // so it follows the watch just armed (nil on a fleet or disarmed |
| + // box), before any loop that reconciles. |
| + armedPolicyCache.Store(provision.NewPolicyCache(armedChangeWatch.Load())) |
| go adoptLoop(reg, svc, sd, claims.Writes) |
| go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision) |
| if logSink != nil { |
| @@ -1575,6 +1579,15 @@ func configAckWriter(store release.Store, logPrefix string) func(context.Context |
| // no Reconcile and is not this mutex's problem. |
| var reconcileMu sync.Mutex |
| +// armedPolicyCache is the process's one app policy cache (#328, #329), |
| +// handed to every instances reconcile pass (ReconcileConfig.Policies), from |
| +// both provisioning loops, so the policy reads stay inside spec §3a's |
| +// budget: one backstop listing every 30 to 33 minutes, and a GET only for a |
| +// policy that may have changed. Two caches, or one per pass, would read |
| +// more. RunAgent builds it after arming the change index; until then |
| +// (tests) it is nil and each pass builds a throwaway. |
| +var armedPolicyCache atomic.Pointer[provision.PolicyCache] |
| + |
| // reconcileSerialized is the one way any agent loop runs the record-family |
| // reconcile passes — the mutex lives in a shared wrapper, not at the call |
| // sites, so a future third caller cannot forget it. Pinned by |
| @@ -3036,7 +3049,9 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu |
| legs := &passLegs{store: store, reg: reg, pub: pub, optDir: optDir, restart: restart, opts: opts, |
| priv: priv, dd: dd, svc: svc, writes: writes, bs: bs, |
| cfg: func() provision.ReconcileConfig { |
| - return configuredClaimProvision(provisionTickConfig(os.Getenv), configure) |
| + cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure) |
| + cfg.Policies = armedPolicyCache.Load() |
| + return cfg |
| }} |
| d.fullPass = legs.full |
| // The change index (S3 change-index design §4.5): an index-only change |
| @@ -3578,6 +3593,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta |
| return |
| } |
| cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure) |
| + cfg.Policies = armedPolicyCache.Load() |
| backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET")) |
| log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools) |
| // Platform-only: a fleet box (CARLOS_FLEET set) never reads other |
diff --git a/internal/provision/apppolicy_cache.go b/internal/provision/apppolicy_cache.go| new file mode 100644 |
| index 000000000..2c0e85ea6 |
| --- /dev/null |
| +++ b/internal/provision/apppolicy_cache.go |
| @@ -0,0 +1,270 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "crypto/ed25519" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "math/rand/v2" |
| + "strings" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/apppolicy" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// PolicyCache is the agent's memory of the app policies it has read, so a |
| +// reconcile pass reads the store for a policy only when something says it |
| +// may have changed. |
| +// |
| +// Ruling 11 and spec §3a: every request to the deployment bucket costs |
| +// money, and this feature adds no request a box makes per pass when nothing |
| +// changed. Steady state, a box adds at most 48 LISTs a day (the backstop's |
| +// one ListETags of control/app-policy/ every 30 to 33 minutes) and GETs |
| +// only for policies that changed. A pass reads an app's policy only when: |
| +// |
| +// 1. the backstop is due (the first pass, then 30 to 33 minutes after the |
| +// last completed one, or after a failed one). It snapshots the watch, |
| +// lists every policy's ETag once, and GETs only the pass's apps whose |
| +// listed ETag differs from the cached one. A pass app the listing does |
| +// not name is known absent with no read. A refused listing (an edge |
| +// older than the ETag listing) makes every pass app unreadable and still |
| +// completes the backstop, so an old edge costs one LIST per backstop. |
| +// A store with no ETag listing at all (test fakes) falls back to one GET |
| +// per pass app per backstop. Fleet boxes have no change index, so the |
| +// backstop is their only periodic read; |
| +// 2. the app's generation in the change index moved since its entry was |
| +// read: one GET; |
| +// 3. the app's last read failed: one GET on the next pass, until it reads; |
| +// 4. the app is new to the cache between backstops. It reuses the last |
| +// listing's absence only when that listing succeeded and did not name |
| +// it, the watch was usable when the listing was taken and is usable |
| +// now, and the app's generation still equals the one in the snapshot |
| +// taken just before that listing (Astra plan R3-2). Equal generations |
| +// mean nothing in the app's scope, its policy included, was written |
| +// after the snapshot, so the listing's absence still holds. Otherwise |
| +// (a policy or a record written since, a nil or unusable watch, a |
| +// fleet box) it gets one GET on first encounter. |
| +// |
| +// A box whose change index is disarmed (a nil or unusable watch) reads |
| +// nothing between backstops. Ruling 11 allows no reader its own timer, and |
| +// polling every pass is exactly the per-pass request §3a forbids, so a |
| +// disarmed box sees a policy change at the next backstop, up to about 33 |
| +// minutes later (§3a "stated limits"). |
| +// |
| +// Each entry's generation is the one taken before its read, so a write that |
| +// lands during the read leaves the generation moved and the app due again on |
| +// the next pass, never hidden by the read. |
| +// |
| +// A verified policy's expiry is re-checked against the pass's clock on every |
| +// pass with no read: a cached record past its lifetime is expired. |
| +// |
| +// Safe for concurrent use; a read holds the cache for its whole run, so two |
| +// passes never read the same app twice. Production builds one per agent |
| +// process and hands it to every pass (ReconcileConfig.Policies). |
| +type PolicyCache struct { |
| + watch *release.ChangeWatch |
| + |
| + mu sync.Mutex |
| + entries map[[2]string]*policyEntry |
| + // next is when the next backstop is due; zero before the first one |
| + // completes. failed says the last backstop's listing failed. |
| + next time.Time |
| + failed bool |
| + // listed is the last backstop's listing ("<sqid>/<app>" -> ETag) when |
| + // it succeeded, nil otherwise; listSnap is the watch's observation |
| + // taken immediately before it, nil when the watch was nil or unusable. |
| + listed map[string]string |
| + listSnap map[string]uint64 |
| +} |
| + |
| +// policyEntry is one app's last read: the object's ETag ("" for absent or |
| +// unread), what it established, and the app's generation taken before it. |
| +type policyEntry struct { |
| + etag string |
| + read policyRead |
| + gen uint64 |
| +} |
| + |
| +// NewPolicyCache returns an empty cache gated on watch. A nil watch (the |
| +// index is disarmed, or this is a fleet box) reads only at the backstop and |
| +// for apps new to the cache. |
| +func NewPolicyCache(watch *release.ChangeWatch) *PolicyCache { |
| + return &PolicyCache{watch: watch, entries: map[[2]string]*policyEntry{}} |
| +} |
| + |
| +// appScope is the change-index scope a policy write bumps (release.ChangeScope). |
| +func appScope(a [2]string) string { return "apps/" + a[0] + "/" + a[1] } |
| + |
| +// gen is scope's generation now, and whether the watch can vouch for it. |
| +func (c *PolicyCache) gen(scope string) (uint64, bool) { |
| + if c.watch == nil { |
| + return 0, false |
| + } |
| + return c.watch.Gen(scope) |
| +} |
| + |
| +func (c *PolicyCache) backstopDue(now time.Time) bool { |
| + return c.failed || c.next.IsZero() || !now.Before(c.next) |
| +} |
| + |
| +// completeBackstop schedules the next backstop, jittered upward only, so a |
| +// day holds at most 48. |
| +func (c *PolicyCache) completeBackstop(now time.Time) { |
| + c.failed = false |
| + c.next = now.Add(release.ChangeIndexBackstop + rand.N(release.ChangeIndexBackstop/10)) |
| +} |
| + |
| +// read is the pass's policies for apps (each {sqid, app}), reading the store |
| +// only as the rules on PolicyCache allow. Every read is apppolicy.Get with |
| +// pub and now, so nothing unverified or expired is ever returned as known. |
| +func (c *PolicyCache) read(ctx context.Context, s release.Store, pub ed25519.PublicKey, apps [][2]string, now time.Time) map[[2]string]policyRead { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + out := make(map[[2]string]policyRead, len(apps)) |
| + if len(apps) == 0 { |
| + return out |
| + } |
| + // An app that left the pass is forgotten: its entry would otherwise go |
| + // unrefreshed by the backstop and be trusted, stale, if it came back. |
| + inPass := make(map[[2]string]bool, len(apps)) |
| + for _, a := range apps { |
| + inPass[a] = true |
| + } |
| + for a := range c.entries { |
| + if !inPass[a] { |
| + delete(c.entries, a) |
| + } |
| + } |
| + |
| + due := map[[2]string]uint64{} // app -> generation taken before its read |
| + listedThisPass := false |
| + if c.backstopDue(now) { |
| + listedThisPass = c.backstop(ctx, s, apps, now, due) |
| + } |
| + if !listedThisPass { |
| + for _, a := range apps { |
| + if _, ok := due[a]; ok { |
| + continue |
| + } |
| + scope := appScope(a) |
| + gen, usable := c.gen(scope) |
| + e := c.entries[a] |
| + switch { |
| + case e == nil: |
| + if _, named := c.listed[a[0]+"/"+a[1]]; c.listed != nil && !named && c.listSnap != nil && |
| + usable && gen == c.listSnap[scope] { |
| + c.entries[a] = &policyEntry{read: policyRead{state: policyAbsentKnown}, gen: gen} |
| + continue |
| + } |
| + due[a] = gen |
| + case e.read.state == policyTransient: |
| + due[a] = gen |
| + case usable && gen != e.gen: |
| + due[a] = gen |
| + } |
| + } |
| + } |
| + c.readDue(ctx, s, pub, due, now) |
| + |
| + for _, a := range apps { |
| + e := c.entries[a] |
| + if e == nil { |
| + out[a] = policyRead{state: policyTransient, err: errors.New("the policy listing failed and the policy has not been read")} |
| + continue |
| + } |
| + if e.read.state == policyAbsentKnown && e.read.p != nil && e.read.p.Expired(now) { |
| + e.read = policyRead{state: policyExpired, |
| + err: fmt.Errorf("%w: signed %s", apppolicy.ErrExpired, e.read.p.SignedAt.UTC().Format(time.RFC3339))} |
| + } |
| + out[a] = e.read |
| + } |
| + return out |
| +} |
| + |
| +// backstop is rule 1. It fills due with the apps to read and reports |
| +// whether the listing decided every pass app (false after a failed |
| +// listing, so the pass falls back to rules 2 to 4). |
| +func (c *PolicyCache) backstop(ctx context.Context, s release.Store, apps [][2]string, now time.Time, due map[[2]string]uint64) bool { |
| + var snap map[string]uint64 |
| + if c.watch != nil { |
| + if obs, ok := c.watch.Observed(); ok { |
| + snap = obs |
| + } |
| + } |
| + listed, err := apppolicy.ListETags(ctx, s) |
| + switch { |
| + case err == nil: |
| + c.completeBackstop(now) |
| + c.listed, c.listSnap = listed, snap |
| + for _, a := range apps { |
| + scope := appScope(a) |
| + etag, named := listed[a[0]+"/"+a[1]] |
| + e := c.entries[a] |
| + switch { |
| + case !named: |
| + c.entries[a] = &policyEntry{read: policyRead{state: policyAbsentKnown}, gen: snap[scope]} |
| + case e != nil && e.etag == etag && e.read.state != policyTransient: |
| + // Unchanged since its read. The listing is newer than snap, |
| + // so the entry vouches for the app up to snap. |
| + if snap != nil { |
| + e.gen = snap[scope] |
| + } |
| + default: |
| + gen, _ := c.gen(scope) |
| + due[a] = gen |
| + } |
| + } |
| + return true |
| + case errors.Is(err, release.ErrRefused): |
| + // An edge older than the ETag listing: unreadable, and the |
| + // backstop is done, so this costs one LIST per backstop. |
| + c.completeBackstop(now) |
| + c.listed, c.listSnap = nil, nil |
| + for _, a := range apps { |
| + gen, _ := c.gen(appScope(a)) |
| + c.entries[a] = &policyEntry{read: policyRead{state: policyRefused, err: err}, gen: gen} |
| + } |
| + return true |
| + case errors.Is(err, errors.ErrUnsupported): |
| + // No ETag listing at all (test fakes): one GET per pass app. |
| + c.completeBackstop(now) |
| + c.listed, c.listSnap = nil, nil |
| + for _, a := range apps { |
| + gen, _ := c.gen(appScope(a)) |
| + due[a] = gen |
| + } |
| + return true |
| + } |
| + log.Printf("provision: app policies: listing failed, retried next pass: %v", err) |
| + c.failed = true |
| + c.listed, c.listSnap = nil, nil |
| + return false |
| +} |
| + |
| +// readDue reads each due app's policy, eight at a time (forEach), and |
| +// records what each read established under the generation taken before it. |
| +func (c *PolicyCache) readDue(ctx context.Context, s release.Store, pub ed25519.PublicKey, due map[[2]string]uint64, now time.Time) { |
| + if len(due) == 0 { |
| + return |
| + } |
| + ids := make([]string, 0, len(due)) |
| + for a := range due { |
| + ids = append(ids, a[0]+"/"+a[1]) |
| + } |
| + var mu sync.Mutex |
| + forEach(ids, func(id string) { |
| + sqid, app, _ := strings.Cut(id, "/") |
| + p, etag, err := apppolicy.Get(ctx, s, pub, sqid, app, now) |
| + r := classifyPolicyGet(p, err) |
| + if r.state == policyTransient || r.state == policyRefused { |
| + etag = "" // read again at the next backstop whatever is listed |
| + } |
| + a := [2]string{sqid, app} |
| + mu.Lock() |
| + c.entries[a] = &policyEntry{etag: etag, read: r, gen: due[a]} |
| + mu.Unlock() |
| + }) |
| +} |
diff --git a/internal/provision/apppolicy_cache_test.go b/internal/provision/apppolicy_cache_test.go| new file mode 100644 |
| index 000000000..119de4234 |
| --- /dev/null |
| +++ b/internal/provision/apppolicy_cache_test.go |
| @@ -0,0 +1,384 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "os" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/apppolicy" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// countingStore is a change-indexed DirStore that counts what the §3a |
| +// budget is stated in: listings that reach control/app-policy/ (List and |
| +// ListETags) and reads of keys under it (Get, GetIfChanged, Open). Every |
| +// other request passes uncounted. refuseList makes the policy listing |
| +// refused, as an edge older than the ETag listing answers. |
| +type countingStore struct { |
| + *release.DirStore |
| + |
| + mu sync.Mutex |
| + lists int |
| + listETags int |
| + gets map[string]int |
| + refuseList bool |
| +} |
| + |
| +func newCountingStore(t *testing.T) *countingStore { |
| + d := dirStore(t) |
| + d.EnableChangeIndex() |
| + return &countingStore{DirStore: d, gets: map[string]int{}} |
| +} |
| + |
| +// policyPrefixOverlap reports whether a listing of prefix could return a |
| +// policy key. |
| +func policyPrefixOverlap(prefix string) bool { |
| + return strings.HasPrefix(prefix, apppolicy.Prefix) || strings.HasPrefix(apppolicy.Prefix, prefix) |
| +} |
| + |
| +func (c *countingStore) countGet(key string) { |
| + if strings.HasPrefix(key, apppolicy.Prefix) { |
| + c.mu.Lock() |
| + c.gets[key]++ |
| + c.mu.Unlock() |
| + } |
| +} |
| + |
| +func (c *countingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + if policyPrefixOverlap(prefix) { |
| + c.mu.Lock() |
| + c.lists++ |
| + c.mu.Unlock() |
| + } |
| + return c.DirStore.List(ctx, prefix) |
| +} |
| + |
| +func (c *countingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + if policyPrefixOverlap(prefix) { |
| + c.mu.Lock() |
| + c.listETags++ |
| + refuse := c.refuseList |
| + c.mu.Unlock() |
| + if refuse { |
| + return nil, fmt.Errorf("x: %w", release.ErrRefused) |
| + } |
| + } |
| + return c.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +func (c *countingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + c.countGet(key) |
| + return c.DirStore.Get(ctx, key) |
| +} |
| + |
| +func (c *countingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| + c.countGet(key) |
| + return c.DirStore.GetIfChanged(ctx, key, etag) |
| +} |
| + |
| +func (c *countingStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) { |
| + c.countGet(key) |
| + return c.DirStore.Open(ctx, key) |
| +} |
| + |
| +// counts is the requests since the last reset: LISTs, ETag listings, and |
| +// GETs by key. |
| +type counts struct { |
| + lists, listETags int |
| + gets map[string]int |
| +} |
| + |
| +func (c counts) totalGets() int { |
| + n := 0 |
| + for _, v := range c.gets { |
| + n += v |
| + } |
| + return n |
| +} |
| + |
| +func (c *countingStore) take() counts { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + out := counts{lists: c.lists, listETags: c.listETags, gets: c.gets} |
| + c.lists, c.listETags, c.gets = 0, 0, map[string]int{} |
| + return out |
| +} |
| + |
| +// noETagLister hides every extension but the conditional read: the shape |
| +// of a test fake with no ETag listing (no production store is one). |
| +type noETagLister struct{ release.Store } |
| + |
| +func (s noETagLister) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| + return s.Store.(release.ConditionalGetter).GetIfChanged(ctx, key, etag) |
| +} |
| + |
| +// budgetFixture is one box's reconcile over a counting store: a fake clock, |
| +// an optional change index watch polled before each pass as the agent's |
| +// fast tick does, and one long-lived policy cache. |
| +type budgetFixture struct { |
| + t *testing.T |
| + cs *countingStore |
| + store release.Store // what Reconcile reads: cs, or a wrapper over it |
| + sqid string |
| + reg *registry.Registry |
| + cfg ReconcileConfig |
| + fx *policyFixture // the signing key and clock |
| + watch *release.ChangeWatch |
| +} |
| + |
| +// newBudgetFixture: an account; apps with one in-scope record each. |
| +func newBudgetFixture(t *testing.T, armed bool, apps ...string) *budgetFixture { |
| + t.Helper() |
| + cs := newCountingStore(t) |
| + priv, pub := releaseKeys(t) |
| + b := &budgetFixture{t: t, cs: cs, store: cs, sqid: writeAccountManifest(t, cs, acctID, "Shop"), reg: regWith(t)} |
| + b.fx = &policyFixture{t: t, priv: priv, pub: pub, now: time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)} |
| + if armed { |
| + b.watch = release.NewChangeWatch(cs) |
| + } |
| + b.cfg = ReconcileConfig{Region: "ie", DataDir: "/data", ReleasePub: pub, |
| + Now: func() time.Time { return b.fx.now }, Policies: NewPolicyCache(b.watch)} |
| + for _, app := range apps { |
| + b.addApp(app, "") |
| + } |
| + return b |
| +} |
| + |
| +func (b *budgetFixture) host(app string) string { return "a." + app + ".example.com" } |
| + |
| +// addApp writes an app's instances manifest and one record in pool. |
| +func (b *budgetFixture) addApp(app, pool string) { |
| + b.t.Helper() |
| + if err := PutManifest(ctx, b.cs, b.sqid, app, Manifest{Schema: 1, App: app, Domain: app + ".example.com", Pool: pool}); err != nil { |
| + b.t.Fatal(err) |
| + } |
| + if err := CreateRecord(ctx, b.cs, Record{Schema: 1, Host: b.host(app), Slug: "a", App: app, Account: b.sqid, |
| + Pool: pool, Region: "ie", Channel: "stable", CreatedAt: b.fx.now}); err != nil { |
| + b.t.Fatal(err) |
| + } |
| +} |
| + |
| +// putPolicy signs and writes app's policy over whatever is there, through |
| +// the store's own write path (so it bumps apps/<sqid>/<app>). |
| +func (b *budgetFixture) putPolicy(app string, p apppolicy.Policy) { |
| + b.t.Helper() |
| + p.V, p.Account, p.App = 1, b.sqid, app |
| + if p.Revision == 0 { |
| + p.Revision = 1 |
| + } |
| + if p.SignedAt.IsZero() { |
| + p.SignedAt = b.fx.now |
| + } |
| + _, etag, err := b.cs.DirStore.Get(ctx, apppolicy.Key(b.sqid, app)) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + b.t.Fatal(err) |
| + } |
| + if err := apppolicy.PutIfMatch(ctx, b.cs.DirStore, b.fx.priv, p, etag); err != nil { |
| + b.t.Fatal(err) |
| + } |
| +} |
| + |
| +func (b *budgetFixture) pass() { |
| + b.t.Helper() |
| + if b.watch != nil { |
| + if _, err := b.watch.Poll(ctx); err != nil { |
| + b.t.Fatalf("watch poll: %v", err) |
| + } |
| + } |
| + if err := Reconcile(ctx, b.store, b.reg, b.cfg); err != nil { |
| + b.t.Fatal(err) |
| + } |
| +} |
| + |
| +// passesFor runs passes every step for d, starting one step from now. |
| +func (b *budgetFixture) passesFor(d, step time.Duration) { |
| + b.t.Helper() |
| + for end := b.fx.now.Add(d); b.fx.now.Before(end); { |
| + b.fx.now = b.fx.now.Add(step) |
| + b.pass() |
| + } |
| +} |
| + |
| +func (b *budgetFixture) row(app string) registry.Route { return *mustGet(b.t, b.reg, b.host(app)) } |
| + |
| +func (b *budgetFixture) key(app string) string { return apppolicy.Key(b.sqid, app) } |
| + |
| +func (b *budgetFixture) wantNothing(c counts, what string) { |
| + b.t.Helper() |
| + if c.lists != 0 || c.listETags != 0 || c.totalGets() != 0 { |
| + b.t.Fatalf("%s: %d LISTs, %d ETag listings, GETs %v; want no policy request", what, c.lists, c.listETags, c.gets) |
| + } |
| +} |
| + |
| +var policyGrant = apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"} |
| + |
| +// The §3a budget, pinned (ruling 11). Fails on any increase. |
| +func TestPolicyReadsKeepTheRequestBudget(t *testing.T) { |
| + b := newBudgetFixture(t, true, "shop", "cart", "blog") |
| + b.putPolicy("shop", policyGrant) |
| + b.putPolicy("cart", policyGrant) |
| + // Two more apps whose records exist before the first listing; "docs" |
| + // is in a pool this box does not serve yet. |
| + b.addApp("docs", "four") |
| + |
| + // Pass 1, the first backstop: one listing, a GET per policy, none for |
| + // the app with no policy. |
| + b.pass() |
| + c := b.cs.take() |
| + if c.listETags != 1 || c.lists != 0 || c.totalGets() != 2 || c.gets[b.key("shop")] != 1 || c.gets[b.key("cart")] != 1 { |
| + t.Fatalf("first pass: %d ETag listings, %d LISTs, GETs %v; want 1, 0, one each of shop and cart", c.listETags, c.lists, c.gets) |
| + } |
| + wantCols(t, b.row("shop"), "10.0.5.20", "large", b.fx.now.Add(apppolicy.Lifetime).Unix()) |
| + wantCols(t, b.row("blog"), "", "", 0) |
| + |
| + // A day of passes 5 minutes apart with nothing written. |
| + b.passesFor(24*time.Hour, 5*time.Minute) |
| + c = b.cs.take() |
| + if c.listETags > 48 || c.totalGets() != 0 || c.lists != 0 { |
| + t.Fatalf("a quiet day: %d ETag listings (at most 48), GETs %v (want none), %d LISTs (want none)", c.listETags, c.gets, c.lists) |
| + } |
| + if c.listETags < 36 { |
| + t.Fatalf("a quiet day: %d ETag listings; the 30 to 33 minute backstop should have run about 41 times", c.listETags) |
| + } |
| + |
| + // A write to one policy: one GET of that key on the next pass. |
| + b.putPolicy("shop", apppolicy.Policy{Revision: 2, EgressAllow: []string{"10.0.5.20"}, DefaultSize: "xlarge"}) |
| + b.pass() |
| + c = b.cs.take() |
| + if c.totalGets() != 1 || c.gets[b.key("shop")] != 1 || c.listETags != 0 || c.lists != 0 { |
| + t.Fatalf("after a policy write: GETs %v, %d ETag listings, %d LISTs; want exactly one GET of shop's key", c.gets, c.listETags, c.lists) |
| + } |
| + wantCols(t, b.row("shop"), "10.0.5.20", "xlarge", b.fx.now.Add(apppolicy.Lifetime).Unix()) |
| + |
| + // An app whose records predate the last listing enters scope between |
| + // backstops, its generation unchanged since the snapshot: no read. |
| + b.cfg.Pools = []string{"", "four"} |
| + b.pass() |
| + c = b.cs.take() |
| + b.wantNothing(c, "an unmoved app entering scope") |
| + wantCols(t, b.row("docs"), "", "", 0) |
| + |
| + // An app whose records are written after the listing: one GET on its |
| + // first pass, then none. |
| + b.addApp("mail", "") |
| + b.pass() |
| + c = b.cs.take() |
| + if c.totalGets() != 1 || c.gets[b.key("mail")] != 1 || c.listETags != 0 || c.lists != 0 { |
| + t.Fatalf("a new app: GETs %v, %d ETag listings, %d LISTs; want exactly one GET of its key", c.gets, c.listETags, c.lists) |
| + } |
| + b.pass() |
| + b.wantNothing(b.cs.take(), "the new app's second pass") |
| +} |
| + |
| +// Astra plan R3-2: the exact ordering. Fails if a new app reuses the |
| +// listing's absence after its scope moved. |
| +func TestAPolicyWrittenBeforeItsAppEntersScopeIsReadOnFirstEncounter(t *testing.T) { |
| + for _, armed := range []bool{true, false} { |
| + t.Run(fmt.Sprintf("armed=%v", armed), func(t *testing.T) { |
| + b := newBudgetFixture(t, armed, "shop") |
| + b.pass() // the backstop: lists control/app-policy/ empty |
| + c := b.cs.take() |
| + if c.listETags != 1 || c.totalGets() != 0 { |
| + t.Fatalf("first pass: %d ETag listings, GETs %v", c.listETags, c.gets) |
| + } |
| + b.fx.now = b.fx.now.Add(5 * time.Minute) |
| + b.putPolicy("mail", policyGrant) |
| + b.addApp("mail", "") |
| + b.pass() |
| + c = b.cs.take() |
| + if c.totalGets() != 1 || c.gets[b.key("mail")] != 1 || c.listETags != 0 || c.lists != 0 { |
| + t.Fatalf("first encounter: GETs %v, %d ETag listings, %d LISTs; want exactly one GET of mail's key", c.gets, c.listETags, c.lists) |
| + } |
| + wantCols(t, b.row("mail"), "10.0.5.20", "large", b.fx.now.Add(apppolicy.Lifetime).Unix()) |
| + }) |
| + } |
| +} |
| + |
| +func TestPolicyReadsWithTheIndexDisarmedWaitForTheBackstop(t *testing.T) { |
| + b := newBudgetFixture(t, false, "shop") |
| + b.pass() |
| + b.cs.take() |
| + signed := b.fx.now |
| + b.putPolicy("shop", policyGrant) |
| + b.passesFor(25*time.Minute, 5*time.Minute) |
| + b.wantNothing(b.cs.take(), "passes inside the backstop") |
| + wantCols(t, b.row("shop"), "", "", 0) |
| + b.fx.now = b.fx.now.Add(10 * time.Minute) // 35 minutes on: the backstop is due |
| + b.pass() |
| + c := b.cs.take() |
| + if c.listETags != 1 || c.totalGets() != 1 || c.gets[b.key("shop")] != 1 || c.lists != 0 { |
| + t.Fatalf("the backstop pass: %d ETag listings, GETs %v, %d LISTs; want 1 listing and one GET of shop's key", c.listETags, c.gets, c.lists) |
| + } |
| + wantCols(t, b.row("shop"), "10.0.5.20", "large", signed.Add(apppolicy.Lifetime).Unix()) |
| +} |
| + |
| +func TestPolicyReadsFallBackToOneGetPerAppWithoutAListing(t *testing.T) { |
| + b := newBudgetFixture(t, false, "shop", "cart", "blog") |
| + b.store = noETagLister{b.cs} |
| + b.putPolicy("shop", policyGrant) |
| + b.putPolicy("cart", policyGrant) |
| + b.pass() |
| + c := b.cs.take() |
| + if c.totalGets() != 3 || c.lists != 0 || c.listETags != 0 { |
| + t.Fatalf("the backstop pass: GETs %v, %d LISTs, %d ETag listings; want one GET per app and no listing", c.gets, c.lists, c.listETags) |
| + } |
| + for _, app := range []string{"shop", "cart", "blog"} { |
| + if c.gets[b.key(app)] != 1 { |
| + t.Fatalf("GETs %v: want one of %s", c.gets, app) |
| + } |
| + } |
| + wantCols(t, b.row("cart"), "10.0.5.20", "large", b.fx.now.Add(apppolicy.Lifetime).Unix()) |
| + b.passesFor(25*time.Minute, 5*time.Minute) |
| + b.wantNothing(b.cs.take(), "passes inside the backstop") |
| +} |
| + |
| +func TestARefusedListingIsUnreadableAndWaitsForTheNextBackstop(t *testing.T) { |
| + b := newBudgetFixture(t, false, "shop", "cart") |
| + b.putPolicy("shop", policyGrant) |
| + b.pass() |
| + wantCols(t, b.row("shop"), "10.0.5.20", "large", b.fx.now.Add(apppolicy.Lifetime).Unix()) |
| + b.cs.take() |
| + |
| + b.cs.mu.Lock() |
| + b.cs.refuseList = true |
| + b.cs.mu.Unlock() |
| + b.fx.now = b.fx.now.Add(35 * time.Minute) |
| + b.pass() |
| + c := b.cs.take() |
| + if c.listETags != 1 || c.totalGets() != 0 || c.lists != 0 { |
| + t.Fatalf("the refused backstop: %d ETag listings, GETs %v, %d LISTs; want 1, none, 0", c.listETags, c.gets, c.lists) |
| + } |
| + states := PolicyReadStates() |
| + for _, app := range []string{"shop", "cart"} { |
| + wantCols(t, b.row(app), "", "", 0) |
| + if got := states[b.sqid+"/"+app]; got != "policy unreadable" { |
| + t.Fatalf("PolicyReadStates[%s] = %q, want policy unreadable", app, got) |
| + } |
| + } |
| + b.passesFor(25*time.Minute, 5*time.Minute) |
| + b.wantNothing(b.cs.take(), "passes after a refused listing") |
| +} |
| + |
| +func TestPolicyCacheExpiresWithoutAReading(t *testing.T) { |
| + b := newBudgetFixture(t, true, "shop") |
| + signed := b.fx.now.Add(-apppolicy.Lifetime + 10*time.Minute) |
| + b.putPolicy("shop", apppolicy.Policy{SignedAt: signed, EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"}) |
| + b.pass() |
| + wantCols(t, b.row("shop"), "10.0.5.20", "large", signed.Add(apppolicy.Lifetime).Unix()) |
| + b.cs.take() |
| + |
| + b.fx.now = b.fx.now.Add(15 * time.Minute) // past the lifetime, inside the backstop |
| + b.pass() |
| + b.wantNothing(b.cs.take(), "a cached policy expiring") |
| + wantCols(t, b.row("shop"), "", "", 0) |
| + if got := PolicyReadStates()[b.sqid+"/shop"]; got != "policy expired" { |
| + t.Fatalf("PolicyReadStates = %q, want policy expired", got) |
| + } |
| +} |
diff --git a/internal/provision/apppolicy_reconcile.go b/internal/provision/apppolicy_reconcile.go| new file mode 100644 |
| index 000000000..179564d06 |
| --- /dev/null |
| +++ b/internal/provision/apppolicy_reconcile.go |
| @@ -0,0 +1,232 @@ |
| +package provision |
| + |
| +// The agent's converge of an app's signed operator policy |
| +// (control/app-policy/<sqid>/<app>.json, #328, #329) into its instance rows' |
| +// egress_allow, size and policy_expires_at columns: spec §3 and §4.3 |
| +// "before claims". Root reads the columns (plans 2 and 3); this pass is the |
| +// only writer. |
| +// |
| +// Every column is written from a record the release key signed for this |
| +// account and app and that has not expired (apppolicy.Get). Anything else |
| +// fails closed: no allow, and sizes at the build's tier from the next start. |
| +// A read that failed for a reason that says nothing about the policy (a |
| +// network error, a store outage) leaves an existing row as it is while the |
| +// row's own recorded expiry is in the future, so a blip never revokes an |
| +// allow, and an outage longer than the lifetime still closes it. |
| +// |
| +// Instance rows only: sidecar members get no policy (spec §4.3 rev 8), and |
| +// ReconcileSidecars never writes these columns. |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "log" |
| + "maps" |
| + "os" |
| + "strconv" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/apppolicy" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// policyState is what one read of an app's policy established. |
| +type policyState int |
| + |
| +const ( |
| + // policyAbsentKnown: a verified policy, or a confirmed absence (p nil). |
| + policyAbsentKnown policyState = iota |
| + // policyRefused: the store refused the read (release.ErrRefused, e.g. a |
| + // fleet door that does not grant the key). Never read as "no policy". |
| + policyRefused |
| + // policyUnverifiable: an object that is not a policy the release key |
| + // signed for this key (apppolicy.ErrUnverifiable), or no key to check. |
| + policyUnverifiable |
| + // policyExpired: a validly signed policy past apppolicy.Lifetime. |
| + policyExpired |
| + // policyTransient: the read failed for a reason that says nothing about |
| + // the policy. Retried on the next pass. |
| + policyTransient |
| +) |
| + |
| +// word is the state as PolicyReadStates and the log line name it; "" for |
| +// the states that are not a problem to report. |
| +func (s policyState) word() string { |
| + switch s { |
| + case policyRefused: |
| + return "unreadable" |
| + case policyUnverifiable: |
| + return "unverifiable" |
| + case policyExpired: |
| + return "expired" |
| + } |
| + return "" |
| +} |
| + |
| +// policyRead is one app's policy as this pass knows it. err is why, for the |
| +// states that are not policyAbsentKnown. |
| +type policyRead struct { |
| + p *apppolicy.Policy |
| + state policyState |
| + err error |
| +} |
| + |
| +// classifyPolicyGet turns apppolicy.Get's answer into a policyRead. |
| +func classifyPolicyGet(p *apppolicy.Policy, err error) policyRead { |
| + switch { |
| + case err == nil: |
| + return policyRead{p: p, state: policyAbsentKnown} |
| + case errors.Is(err, os.ErrNotExist): |
| + return policyRead{state: policyAbsentKnown} |
| + case errors.Is(err, release.ErrRefused): |
| + return policyRead{state: policyRefused, err: err} |
| + case errors.Is(err, apppolicy.ErrExpired): |
| + return policyRead{state: policyExpired, err: err} |
| + case errors.Is(err, apppolicy.ErrUnverifiable): |
| + return policyRead{state: policyUnverifiable, err: err} |
| + } |
| + return policyRead{state: policyTransient, err: err} |
| +} |
| + |
| +// appPolicyCols is what one instance row's three policy columns should |
| +// hold. Known false means "this pass cannot say": an existing row keeps its |
| +// columns (subject to its own expiry, policyPass.forRow) and a new row is |
| +// minted with none. |
| +type appPolicyCols struct { |
| + EgressAllow, Size string |
| + ExpiresAt int64 |
| + Known bool |
| +} |
| + |
| +// expiresCol is ExpiresAt as the registry stores it: decimal unix seconds, |
| +// or "" for none (ruling PF-1, so a row with no policy reads exactly as |
| +// before the column existed). |
| +func (c appPolicyCols) expiresCol() string { |
| + if c.ExpiresAt == 0 { |
| + return "" |
| + } |
| + return strconv.FormatInt(c.ExpiresAt, 10) |
| +} |
| + |
| +// policyColsFor is rec's row's columns under r. |
| +func policyColsFor(r policyRead, rec Record) appPolicyCols { |
| + switch r.state { |
| + case policyAbsentKnown: |
| + if r.p == nil { |
| + return appPolicyCols{Known: true} |
| + } |
| + return appPolicyCols{ |
| + EgressAllow: r.p.EgressColumn(), |
| + Size: r.p.SizeFor(rec.Host, rec.CreatedAt), |
| + ExpiresAt: r.p.SignedAt.Add(apppolicy.Lifetime).Unix(), |
| + Known: true, |
| + } |
| + case policyTransient: |
| + return appPolicyCols{} |
| + } |
| + // Refused, unverifiable, expired: fail closed. |
| + return appPolicyCols{Known: true} |
| +} |
| + |
| +// rowPolicyExpiry is the row's recorded policy expiry, 0 for none or for a |
| +// value that does not parse. |
| +func rowPolicyExpiry(rt registry.Route) int64 { |
| + n, err := strconv.ParseInt(rt.PolicyExpiresAt, 10, 64) |
| + if err != nil { |
| + return 0 |
| + } |
| + return n |
| +} |
| + |
| +// policyStates is the last pass's per-app problems (PolicyReadStates). |
| +var policyStates struct { |
| + mu sync.Mutex |
| + m map[string]string |
| +} |
| + |
| +// PolicyReadStates is a copy of the last Reconcile pass's per-app policy |
| +// problems, keyed "<sqid>/<app>", valued "policy unreadable", "policy |
| +// unverifiable" or "policy expired". An app read cleanly, present or |
| +// absent, has no entry. Replaced whole at the end of each pass. |
| +func PolicyReadStates() map[string]string { |
| + policyStates.mu.Lock() |
| + defer policyStates.mu.Unlock() |
| + return maps.Clone(policyStates.m) |
| +} |
| + |
| +// policyPass is one Reconcile pass's view of its apps' policies: the reads, |
| +// the problems it has logged, and the states it will publish. |
| +type policyPass struct { |
| + reads map[[2]string]policyRead |
| + now time.Time |
| + states map[string]string |
| +} |
| + |
| +// newPolicyPass logs each app's problem once and records it. |
| +func newPolicyPass(reads map[[2]string]policyRead, now time.Time) *policyPass { |
| + pp := &policyPass{reads: reads, now: now, states: map[string]string{}} |
| + for a, r := range reads { |
| + switch r.state { |
| + case policyRefused, policyUnverifiable, policyExpired: |
| + pp.problem(a, r.state.word(), r.err) |
| + case policyTransient: |
| + log.Printf("provision: %s/%s: policy read failed: %v", a[0], a[1], r.err) |
| + } |
| + } |
| + return pp |
| +} |
| + |
| +func (pp *policyPass) problem(a [2]string, word string, err error) { |
| + id := a[0] + "/" + a[1] |
| + if _, done := pp.states[id]; done { |
| + return |
| + } |
| + log.Printf("provision: %s/%s: policy %s: %v", a[0], a[1], word, err) |
| + pp.states[id] = "policy " + word |
| +} |
| + |
| +// cols is the columns rec's row should hold this pass. An app this pass did |
| +// not read is unknown. |
| +func (pp *policyPass) cols(rec Record) appPolicyCols { |
| + r, ok := pp.reads[[2]string{rec.Account, rec.App}] |
| + if !ok { |
| + return appPolicyCols{} |
| + } |
| + return policyColsFor(r, rec) |
| +} |
| + |
| +// forRow is cols for an existing row: an unknown read keeps the row's |
| +// columns only while the row's own recorded expiry is in the future. A |
| +// row whose expiry has passed, or that carries a column with no expiry, is |
| +// cleared as an expired policy is (Astra plan R1-3): an outage longer than |
| +// the lifetime closes the allow even if root never reads the store. |
| +func (pp *policyPass) forRow(rec Record, existing registry.Route) appPolicyCols { |
| + ap := pp.cols(rec) |
| + if ap.Known { |
| + return ap |
| + } |
| + exp := rowPolicyExpiry(existing) |
| + stale := exp == 0 && (existing.EgressAllow != "" || existing.Size != "") |
| + if exp != 0 && !pp.now.Before(time.Unix(exp, 0)) { |
| + stale = true |
| + } |
| + if !stale { |
| + return ap |
| + } |
| + why := fmt.Errorf("%w: the row's policy expired at %s and the store could not be read", |
| + apppolicy.ErrExpired, time.Unix(exp, 0).UTC().Format(time.RFC3339)) |
| + if exp == 0 { |
| + why = fmt.Errorf("%w: the row carries a policy with no expiry and the store could not be read", apppolicy.ErrExpired) |
| + } |
| + pp.problem([2]string{rec.Account, rec.App}, policyExpired.word(), why) |
| + return appPolicyCols{Known: true} |
| +} |
| + |
| +// publish replaces PolicyReadStates with this pass's. |
| +func (pp *policyPass) publish() { |
| + policyStates.mu.Lock() |
| + policyStates.m = pp.states |
| + policyStates.mu.Unlock() |
| +} |
diff --git a/internal/provision/apppolicy_reconcile_test.go b/internal/provision/apppolicy_reconcile_test.go| new file mode 100644 |
| index 000000000..8563dac05 |
| --- /dev/null |
| +++ b/internal/provision/apppolicy_reconcile_test.go |
| @@ -0,0 +1,446 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "crypto/ed25519" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/apppolicy" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// policyFixture is updatesFixture's shape for the app policy: an account, |
| +// the "shop" app's instances manifest, a DirStore, a temp registry, the |
| +// release key that signs policies, and a clock the test moves. |
| +type policyFixture struct { |
| + t *testing.T |
| + s *release.DirStore |
| + sqid string |
| + reg *registry.Registry |
| + cfg ReconcileConfig |
| + priv ed25519.PrivateKey |
| + pub ed25519.PublicKey |
| + now time.Time |
| + t0 time.Time |
| + mu sync.Mutex |
| + changed []string |
| +} |
| + |
| +const policyHost = "shop-a.example.com" |
| + |
| +func newPolicyFixture(t *testing.T) *policyFixture { |
| + t.Helper() |
| + s := dirStore(t) |
| + sqid := writeAccountManifest(t, s, acctID, "Shop") |
| + if err := PutManifest(ctx, s, sqid, "shop", Manifest{Schema: 1, App: "shop", Domain: "example.com"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + priv, pub := releaseKeys(t) |
| + fx := &policyFixture{t: t, s: s, sqid: sqid, reg: regWith(t), priv: priv, pub: pub, |
| + now: time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)} |
| + fx.t0 = fx.now.Add(-24 * time.Hour) |
| + fx.cfg = ReconcileConfig{Region: "ie", DataDir: "/data", ReleasePub: pub, |
| + Now: func() time.Time { return fx.now }, |
| + PolicyChanged: func(host string) { |
| + fx.mu.Lock() |
| + fx.changed = append(fx.changed, host) |
| + fx.mu.Unlock() |
| + }} |
| + return fx |
| +} |
| + |
| +func (fx *policyFixture) record(host string) Record { |
| + slug, _, _ := strings.Cut(host, ".") |
| + return Record{Schema: 1, Host: host, Slug: slug, App: "shop", Account: fx.sqid, |
| + Region: "ie", Channel: "stable", CreatedAt: fx.t0} |
| +} |
| + |
| +func (fx *policyFixture) putRecord(rec Record) { |
| + fx.t.Helper() |
| + if err := CreateRecord(ctx, fx.s, rec); err != nil { |
| + fx.t.Fatal(err) |
| + } |
| +} |
| + |
| +// policy is the shop app's policy with the signing fields filled in. |
| +func (fx *policyFixture) policy(rev uint64, p apppolicy.Policy) apppolicy.Policy { |
| + p.V, p.Account, p.App, p.Revision = 1, fx.sqid, "shop", rev |
| + if p.SignedAt.IsZero() { |
| + p.SignedAt = fx.now |
| + } |
| + return p |
| +} |
| + |
| +// putPolicy signs p with the release key and writes it over whatever is at |
| +// the shop app's key. |
| +func (fx *policyFixture) putPolicy(p apppolicy.Policy) { |
| + fx.t.Helper() |
| + if err := apppolicy.PutIfMatch(ctx, fx.s, fx.priv, p, fx.etag()); err != nil { |
| + fx.t.Fatal(err) |
| + } |
| +} |
| + |
| +// putRaw writes bytes at the shop app's policy key. |
| +func (fx *policyFixture) putRaw(b []byte) { |
| + fx.t.Helper() |
| + if _, err := fx.s.Put(ctx, apppolicy.Key(fx.sqid, "shop"), b, fx.etag()); err != nil { |
| + fx.t.Fatal(err) |
| + } |
| +} |
| + |
| +func (fx *policyFixture) etag() string { |
| + _, etag, err := fx.s.Get(ctx, apppolicy.Key(fx.sqid, "shop")) |
| + if err != nil && !errors.Is(err, os.ErrNotExist) { |
| + fx.t.Fatal(err) |
| + } |
| + return etag |
| +} |
| + |
| +func (fx *policyFixture) run() { fx.runOn(fx.s) } |
| + |
| +func (fx *policyFixture) runOn(s release.Store) { |
| + fx.t.Helper() |
| + if err := Reconcile(ctx, s, fx.reg, fx.cfg); err != nil { |
| + fx.t.Fatal(err) |
| + } |
| +} |
| + |
| +func (fx *policyFixture) row(host string) registry.Route { return *mustGet(fx.t, fx.reg, host) } |
| + |
| +func (fx *policyFixture) changedHosts() []string { |
| + fx.mu.Lock() |
| + defer fx.mu.Unlock() |
| + return append([]string(nil), fx.changed...) |
| +} |
| + |
| +// wantCols checks a row's three policy columns; exp 0 is "" (ruling PF-1). |
| +func wantCols(t *testing.T, rt registry.Route, egress, size string, exp int64) { |
| + t.Helper() |
| + wantExp := "" |
| + if exp != 0 { |
| + wantExp = strconv.FormatInt(exp, 10) |
| + } |
| + if rt.EgressAllow != egress || rt.Size != size || rt.PolicyExpiresAt != wantExp { |
| + t.Fatalf("%s policy columns = egress %q size %q expires %q, want %q %q %q", |
| + rt.Host, rt.EgressAllow, rt.Size, rt.PolicyExpiresAt, egress, size, wantExp) |
| + } |
| +} |
| + |
| +func (fx *policyFixture) wantState(want string) { |
| + fx.t.Helper() |
| + got, ok := PolicyReadStates()[fx.sqid+"/shop"] |
| + if want == "" && ok { |
| + fx.t.Fatalf("PolicyReadStates has %q for the app, want no entry", got) |
| + } |
| + if want != "" && got != want { |
| + fx.t.Fatalf("PolicyReadStates for the app = %q, want %q", got, want) |
| + } |
| +} |
| + |
| +func TestReconcileCarriesThePolicyToTheRow(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20", "fd00:5::20"}, DefaultSize: "large"})) |
| + fx.putRecord(fx.record(policyHost)) |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20 fd00:5::20", "large", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + if got := fx.changedHosts(); len(got) != 0 { |
| + t.Fatalf("PolicyChanged on a mint: %v", got) |
| + } |
| + fx.wantState("") |
| + |
| + // An override for this instance. |
| + over := fx.policy(2, apppolicy.Policy{EgressAllow: []string{"10.0.5.20", "fd00:5::20"}, DefaultSize: "large", |
| + Sizes: map[string]apppolicy.SizeOverride{policyHost: {Size: "xlarge", CreatedAt: fx.t0}}}) |
| + fx.putPolicy(over) |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20 fd00:5::20", "xlarge", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + if got := fx.changedHosts(); len(got) != 1 || got[0] != policyHost { |
| + t.Fatalf("PolicyChanged = %v, want [%s] once", got, policyHost) |
| + } |
| + |
| + // The console's re-sign 12 hours later: same content, later expiry. |
| + fx.now = fx.now.Add(12 * time.Hour) |
| + over.SignedAt = fx.now |
| + fx.putPolicy(over) |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20 fd00:5::20", "xlarge", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + if got := fx.changedHosts(); len(got) != 1 { |
| + t.Fatalf("PolicyChanged on an expiry-only change: %v", got) |
| + } |
| + |
| + // The policy is deleted: back to nothing, byte-identical to no policy. |
| + if err := apppolicy.Delete(ctx, fx.s, fx.sqid, "shop"); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "", "", 0) |
| + fx.wantState("") |
| +} |
| + |
| +func TestReconcileVerifiesBeforeConverging(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putRecord(fx.record(policyHost)) |
| + good := func() apppolicy.Policy { |
| + return fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"}) |
| + } |
| + converge := func() { |
| + t.Helper() |
| + fx.putPolicy(good()) |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + fx.wantState("") |
| + } |
| + sign := func(priv ed25519.PrivateKey, p apppolicy.Policy) []byte { |
| + t.Helper() |
| + b, err := apppolicy.Sign(priv, p) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return b |
| + } |
| + otherPriv, _ := releaseKeys(t) |
| + grant := good() |
| + grant.EgressAllow = []string{"10.0.5.21"} |
| + other := grant |
| + other.App = "other" |
| + stale := good() |
| + stale.SignedAt = fx.now.Add(-apppolicy.Lifetime - time.Minute) |
| + |
| + for _, c := range []struct { |
| + name string |
| + body []byte |
| + state string |
| + }{ |
| + {"unsigned", mustJSON(t, grant), "policy unverifiable"}, |
| + {"another key", sign(otherPriv, grant), "policy unverifiable"}, |
| + {"another app's record", sign(fx.priv, other), "policy unverifiable"}, |
| + {"expired", sign(fx.priv, stale), "policy expired"}, |
| + } { |
| + converge() |
| + fx.putRaw(c.body) |
| + fx.run() |
| + rt := fx.row(policyHost) |
| + if rt.EgressAllow != "" || rt.Size != "" || rt.PolicyExpiresAt != "" { |
| + t.Fatalf("%s: row = egress %q size %q expires %q, want all cleared", c.name, rt.EgressAllow, rt.Size, rt.PolicyExpiresAt) |
| + } |
| + fx.wantState(c.state) |
| + } |
| + |
| + // No release key on the box: a valid record cannot be verified. |
| + converge() |
| + fx.cfg.ReleasePub = nil |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "", "", 0) |
| + fx.wantState("policy unverifiable") |
| +} |
| + |
| +func TestReconcileGivesSidecarMembersNoPolicy(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.putRecord(fx.record(policyHost)) |
| + if err := PutSidecarManifest(ctx, fx.s, fx.sqid, "shop", SidecarManifest{ |
| + Schema: 1, App: "shop", Args: []string{"sidecar", "run"}, Channel: "stable"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + member := SidecarHost("bot", "shop", fx.sqid) |
| + if err := CreateSidecarRecord(ctx, fx.s, SidecarRecord{Schema: 1, Name: "bot", Host: member, App: "shop", |
| + Account: fx.sqid, Region: "ie", Channel: "stable", CreatedAt: fx.t0}); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.run() |
| + if err := ReconcileSidecars(ctx, fx.s, fx.reg, fx.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.run() // and again, with the member's row present |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + sc := fx.row(member) |
| + if sc.Kind != SidecarKind { |
| + t.Fatalf("member row kind %q", sc.Kind) |
| + } |
| + wantCols(t, sc, "", "", 0) |
| +} |
| + |
| +func TestReconcileIgnoresAnOverrideForAnotherIncarnation(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{DefaultSize: "small", |
| + Sizes: map[string]apppolicy.SizeOverride{policyHost: {Size: "xlarge", CreatedAt: fx.t0.Add(-time.Hour)}}})) |
| + fx.putRecord(fx.record(policyHost)) |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "", "small", fx.now.Add(apppolicy.Lifetime).Unix()) |
| +} |
| + |
| +// policyFaultStore is a DirStore whose policy listing and policy reads can |
| +// be made to fail, counting the policy GETs. |
| +type policyFaultStore struct { |
| + *release.DirStore |
| + mu sync.Mutex |
| + listErr error // ListETags under the policy prefix fails with this; nil forwards |
| + getErr error // Get of a policy key fails with this; nil forwards |
| + gets int |
| +} |
| + |
| +func (s *policyFaultStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + s.mu.Lock() |
| + err := s.listErr |
| + s.mu.Unlock() |
| + if err != nil && strings.HasPrefix(prefix, apppolicy.Prefix) { |
| + return nil, err |
| + } |
| + return s.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +func (s *policyFaultStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if strings.HasPrefix(key, apppolicy.Prefix) { |
| + s.mu.Lock() |
| + s.gets++ |
| + err := s.getErr |
| + s.mu.Unlock() |
| + if err != nil { |
| + return nil, "", err |
| + } |
| + } |
| + return s.DirStore.Get(ctx, key) |
| +} |
| + |
| +func (s *policyFaultStore) set(listErr, getErr error) { |
| + s.mu.Lock() |
| + s.listErr, s.getErr, s.gets = listErr, getErr, 0 |
| + s.mu.Unlock() |
| +} |
| + |
| +func (s *policyFaultStore) policyGets() int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.gets |
| +} |
| + |
| +func TestReconcilePolicyReadFailures(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.cfg.Policies = NewPolicyCache(nil) |
| + fs := &policyFaultStore{DirStore: fx.s} |
| + // past any backstop's jitter (30 to 33 minutes) |
| + backstop := func() { fx.now = fx.now.Add(34 * time.Minute) } |
| + noList := fmt.Errorf("x: %w", errors.ErrUnsupported) |
| + blip := errors.New("connection reset by peer") |
| + |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.putRecord(fx.record(policyHost)) |
| + fx.runOn(fs) |
| + converged := fx.now.Add(apppolicy.Lifetime).Unix() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + |
| + // Transient, on a backstop pass: no listing, and the fallback GET fails. |
| + backstop() |
| + fs.set(noList, blip) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + fx.wantState("") |
| + if n := fs.policyGets(); n != 1 { |
| + t.Fatalf("transient backstop pass: %d policy GETs, want 1", n) |
| + } |
| + // The next pass retries that key, once. |
| + fs.set(noList, blip) |
| + fx.runOn(fs) |
| + if n := fs.policyGets(); n != 1 { |
| + t.Fatalf("retry pass: %d policy GETs, want exactly 1", n) |
| + } |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + |
| + // Still unreadable once the row's own expiry has passed: closed. |
| + fx.now = time.Unix(converged, 0).Add(time.Minute) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row(policyHost), "", "", 0) |
| + fx.wantState("policy expired") |
| + |
| + // A good pass, then a listing failure that is not a refusal: a failed |
| + // backstop, which reads nothing new and changes nothing. |
| + fs.set(noList, nil) |
| + fx.putPolicy(fx.policy(2, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.runOn(fs) |
| + converged = fx.now.Add(apppolicy.Lifetime).Unix() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + fx.wantState("") |
| + backstop() |
| + fs.set(blip, nil) |
| + fx.runOn(fs) |
| + if n := fs.policyGets(); n != 0 { |
| + t.Fatalf("failed listing: %d policy GETs, want 0", n) |
| + } |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + |
| + // A record minted while the policy cannot be read: no columns, then |
| + // converged on the next good pass. |
| + backstop() |
| + fs.set(noList, blip) |
| + fx.putRecord(fx.record("shop-b.example.com")) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row("shop-b.example.com"), "", "", 0) |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", converged) |
| + fs.set(noList, nil) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row("shop-b.example.com"), "10.0.5.20", "large", converged) |
| + |
| + // Refused: unreadable, never "no policy" and never kept. |
| + backstop() |
| + fs.set(noList, fmt.Errorf("x: %w", release.ErrRefused)) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row(policyHost), "", "", 0) |
| + fx.wantState("policy unreadable") |
| + |
| + // Unverifiable bytes. |
| + fs.set(noList, nil) |
| + backstop() |
| + fx.putPolicy(fx.policy(3, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", fx.now.Add(apppolicy.Lifetime).Unix()) |
| + backstop() |
| + fx.putRaw([]byte("{")) |
| + fx.runOn(fs) |
| + wantCols(t, fx.row(policyHost), "", "", 0) |
| + fx.wantState("policy unverifiable") |
| +} |
| + |
| +func TestReconcileDryRunWritesNoPolicy(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.putRecord(fx.record(policyHost)) |
| + fx.run() |
| + exp := fx.now.Add(apppolicy.Lifetime).Unix() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", exp) |
| + |
| + fx.now = fx.now.Add(time.Hour) |
| + fx.putPolicy(fx.policy(2, apppolicy.Policy{EgressAllow: []string{"10.0.5.21"}, DefaultSize: "xlarge"})) |
| + fx.putRecord(fx.record("shop-b.example.com")) |
| + fx.cfg.DryRun = true |
| + fx.run() |
| + wantCols(t, fx.row(policyHost), "10.0.5.20", "large", exp) |
| + if rt, err := fx.reg.Get("shop-b.example.com"); err != nil || rt != nil { |
| + t.Fatalf("dry run minted %+v, %v", rt, err) |
| + } |
| + if got := fx.changedHosts(); len(got) != 0 { |
| + t.Fatalf("PolicyChanged in a dry run: %v", got) |
| + } |
| +} |
| + |
| +// A move participant's row is minted held (mintHeld) with the policy's |
| +// columns, as a ready mint is. |
| +func TestReconcileHeldMintCarriesThePolicy(t *testing.T) { |
| + fx := newPolicyFixture(t) |
| + fx.putPolicy(fx.policy(1, apppolicy.Policy{EgressAllow: []string{"10.0.5.20"}, DefaultSize: "large"})) |
| + fx.putRecord(fx.record(policyHost)) |
| + fx.cfg.Hold = func(Record) (string, string, bool) { return "landing", "op-1", true } |
| + fx.run() |
| + rt := fx.row(policyHost) |
| + if rt.Status != "landing" || rt.HoldOp != "op-1" { |
| + t.Fatalf("row status %q hold %q, want a held mint", rt.Status, rt.HoldOp) |
| + } |
| + wantCols(t, rt, "10.0.5.20", "large", fx.now.Add(apppolicy.Lifetime).Unix()) |
| +} |
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go| index 31945adc1..83c3aa036 100644 |
| --- a/internal/provision/reconcile.go |
| +++ b/internal/provision/reconcile.go |
| @@ -136,6 +136,19 @@ type ReconcileConfig struct { |
| // Seeds is the seeded-instances wiring (seedreconcile.go). Nil: every |
| // seeded record is left alone. |
| Seeds *SeedHooks |
| + // Policies is the agent's cache of the apps' signed operator policies |
| + // (control/app-policy/, #328, #329; apppolicy_cache.go), which keeps the |
| + // policy reads inside spec §3a's budget: no read per pass when nothing |
| + // changed. Production builds one per process and hands the same pointer |
| + // to every pass. Nil (tests, `carlos` tools) builds a throwaway cache per |
| + // pass, which therefore lists and reads every pass: correct, never cheap. |
| + // Each policy is verified with ReleasePub; a nil ReleasePub makes every |
| + // present policy unverifiable (no allow, sizes at the build's tier). |
| + Policies *PolicyCache |
| + // PolicyChanged is called with a host after this pass changed its row's |
| + // egress allow or size (never for an expiry-only change, which the |
| + // console's 12-hour re-sign causes). Nil: nobody is told. |
| + PolicyChanged func(host string) |
| // beforeHoldWrite is the tests' pause under the host's hold lock, just |
| // before holdExisting's conditional write (TestReconcileHoldWrite...). |
| @@ -703,7 +716,11 @@ func shapeMismatch(existing, want registry.Route) string { |
| // one of the two callers is exactly the single-unguarded-write failure mode |
| // the pass exists to prevent (design §4.3). Guarding here means both call |
| // sites are safe by construction, not by each one remembering to check. |
| -func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, policy *UpdatePolicy, policyKnown bool, dryRun bool) { |
| +// |
| +// ap is the row's app policy columns (apppolicy_reconcile.go), converged in |
| +// one SetPolicy when Known. It reports whether the egress allow or size |
| +// moved, so the caller can tell cfg.PolicyChanged. |
| +func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, policy *UpdatePolicy, policyKnown bool, ap appPolicyCols, dryRun bool) (policyMoved bool) { |
| if rec.Channel != "" && existing.Channel != rec.Channel { |
| if dryRun { |
| log.Printf("provision: dry-run: would converge %s channel %s -> %s", rec.Host, existing.Channel, rec.Channel) |
| @@ -804,6 +821,30 @@ func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, |
| } |
| } |
| } |
| + // The app's signed operator policy (#328, #329): all three columns in |
| + // one write, both directions, including down to "". Left alone when the |
| + // policy is unknown this pass (Known false, which Reconcile has already |
| + // turned into a clear for a row past its own expiry). An expiry-only |
| + // change, which every re-sign causes, writes silently. |
| + if ap.Known { |
| + exp := ap.expiresCol() |
| + moved := existing.EgressAllow != ap.EgressAllow || existing.Size != ap.Size |
| + if moved || existing.PolicyExpiresAt != exp { |
| + if dryRun { |
| + if moved { |
| + log.Printf("provision: dry-run: would converge %s policy egress %q -> %q, size %q -> %q", |
| + rec.Host, existing.EgressAllow, ap.EgressAllow, existing.Size, ap.Size) |
| + } |
| + } else if err := reg.SetPolicy(rec.Host, ap.EgressAllow, ap.Size, exp); err != nil { |
| + log.Printf("provision: converge policy on %s: %v", rec.Host, err) |
| + } else if moved { |
| + log.Printf("provision: converged %s policy egress %q -> %q, size %q -> %q", |
| + rec.Host, existing.EgressAllow, ap.EgressAllow, existing.Size, ap.Size) |
| + policyMoved = true |
| + } |
| + } |
| + } |
| + return policyMoved |
| } |
| // Reconcile materializes bucket instance records into registry routes for |
| @@ -1022,6 +1063,24 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| manifests[k] = r |
| prefetchMu.Unlock() |
| }) |
| + // The apps' signed operator policies (#328, #329), for the apps this |
| + // pass mints or keeps rows for: those with an in-scope record and a |
| + // trash marker read and absent. Read through the cache, so a pass reads |
| + // the store only for what may have changed (spec §3a, ruling 11), eight |
| + // at a time, from the bucket itself, never a record view snapshot. |
| + var policyApps [][2]string |
| + for _, id := range recApps { |
| + k := appOf[id] |
| + if t := tombs[k]; t.tomb == nil && t.err == nil { |
| + policyApps = append(policyApps, [2]string{k.acct, k.app}) |
| + } |
| + } |
| + policyCache := cfg.Policies |
| + if policyCache == nil { |
| + policyCache = NewPolicyCache(nil) |
| + } |
| + policyNow := cfg.Now() |
| + policies := newPolicyPass(policyCache.read(ctx, freshStore(s), cfg.ReleasePub, policyApps, policyNow), policyNow) |
| seedCache := map[acctApp]string{} |
| seedKnown := map[acctApp]bool{} |
| @@ -1203,7 +1262,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| policyUnread(rec.Host) |
| continue |
| } |
| - swapImport(ctx, s, reg, cfg, rec, row, mintLayout, p) |
| + swapImport(ctx, s, reg, cfg, rec, row, mintLayout, p, policies.cols(rec)) |
| continue |
| case seedFinishAbort: |
| // A --from seed this box minted is still running: its row |
| @@ -1229,7 +1288,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| if status, op, held := cfg.Hold(rec); held { |
| if !ok { |
| p, known := policyFor(rec.Account, rec.App) |
| - mintHeld(ctx, reg, cfg, rec, seed, status, op, p, known) |
| + mintHeld(ctx, reg, cfg, rec, seed, status, op, p, known, policies.cols(rec)) |
| continue |
| } |
| holdExisting(ctx, s, reg, cfg, rec, existing, status, op) |
| @@ -1283,6 +1342,17 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| continue |
| } |
| nr := RouteFor(rec, cfg.DataDir, seed, nested, p) |
| + // The app policy's columns (#328, #329): a fresh row carries the |
| + // verified policy from its first pass, or none while it cannot be |
| + // read (a mint never waits for it). An existing row's are converged |
| + // below; set here only so the claim hook sees the row it would mint. |
| + var ap appPolicyCols |
| + if ok { |
| + ap = policies.forRow(rec, existing) |
| + } else { |
| + ap = policies.cols(rec) |
| + nr.EgressAllow, nr.Size, nr.PolicyExpiresAt = ap.EgressAllow, ap.Size, ap.expiresCol() |
| + } |
| // A box that cannot provide this instance's config must not |
| // admit it: it would start without its secrets, and a keyed |
| // sibling would be left with a route that lost the race to one |
| @@ -1326,7 +1396,9 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // deploy's SetVersions or the activator's SetNextDue would |
| // clobber those, and Upsert cannot express "clear Environment" |
| // at all (it preserves a non-empty value by design). |
| - convergeOwned(reg, existing, rec, p, pKnown, cfg.DryRun) |
| + if convergeOwned(reg, existing, rec, p, pKnown, ap, cfg.DryRun) && cfg.PolicyChanged != nil { |
| + cfg.PolicyChanged(rec.Host) |
| + } |
| // Absent means "this record does not speak to hibernation" — |
| // never a derivation stamped over an operator's own setting. |
| // |
| @@ -1460,7 +1532,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // already on the record's channel. Every write inside |
| // convergeOwned is dry-run guarded there, so this call |
| // writes nothing. |
| - convergeOwned(reg, existing, rec, p, pKnown, true) |
| + convergeOwned(reg, existing, rec, p, pKnown, ap, true) |
| return nil |
| } |
| if err := reg.SetProvisioned(rec.Host, true); err != nil { |
| @@ -1475,13 +1547,16 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // live process is allowed to sleep. It converges on the next |
| // pass through the owned branch above, by which point the |
| // adoption is already visible in the log. |
| - convergeOwned(reg, existing, rec, p, pKnown, cfg.DryRun) |
| + if convergeOwned(reg, existing, rec, p, pKnown, ap, cfg.DryRun) && cfg.PolicyChanged != nil { |
| + cfg.PolicyChanged(rec.Host) |
| + } |
| } |
| return nil |
| }); err != nil { |
| log.Printf("provision: claim install %s: %v", rec.Host, err) |
| } |
| } |
| + policies.publish() |
| // --- removal, only when nothing failed to read this pass ------------------ |
| // Three conditions, each load-bearing: Provisioned (the flag — not the |
| @@ -1631,7 +1706,10 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg |
| // a mint that is going to land. With claims off the hook is a pass-through. |
| // A status that is not a hold, or an empty op, mints nothing: a |
| // participant's row is never minted ready. |
| -func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, seed, status, op string, policy *UpdatePolicy, policyKnown bool) { |
| +// |
| +// ap is the app policy's columns for the row (apppolicy_reconcile.go): the |
| +// verified policy's, or none while it cannot be read. |
| +func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, rec Record, seed, status, op string, policy *UpdatePolicy, policyKnown bool, ap appPolicyCols) { |
| if !registry.IsHold(status) || op == "" { |
| log.Printf("provision: %s: not minting: the move's hold %q by %q is not a hold", rec.Host, status, op) |
| return |
| @@ -1649,6 +1727,7 @@ func mintHeld(ctx context.Context, reg *registry.Registry, cfg ReconcileConfig, |
| } |
| nr := RouteFor(rec, cfg.DataDir, seed, false, policy) |
| nr.Status, nr.HoldOp = status, op |
| + nr.EgressAllow, nr.Size, nr.PolicyExpiresAt = ap.EgressAllow, ap.Size, ap.expiresCol() |
| if cfg.ConfigGate != nil { |
| if err := cfg.ConfigGate(ctx, nr); err != nil { |
| return |
diff --git a/internal/provision/seedjournal_test.go b/internal/provision/seedjournal_test.go| index 9744cc7c2..113ab8874 100644 |
| --- a/internal/provision/seedjournal_test.go |
| +++ b/internal/provision/seedjournal_test.go |
| @@ -689,7 +689,7 @@ func TestReconcileLeavesAnUngatedRowOfAnUncommittedImport(t *testing.T) { |
| // redoes the record CAS, even with no journal pass ahead of it. |
| func TestSwapImportBackstopRedoesOnlyTheCAS(t *testing.T) { |
| x := journaledImport(t, seeds.StateReady) |
| - swapImport(ctx, x.s, x.reg, x.cfg, x.rec, &x.row, func() (bool, error) { return true, nil }, nil) |
| + swapImport(ctx, x.s, x.reg, x.cfg, x.rec, &x.row, func() (bool, error) { return true, nil }, nil, appPolicyCols{}) |
| if st := seedState(t, x.s, x.rec); st.State != seeds.StateSwapped || st.Reason != "" { |
| t.Fatalf("record = %s %q, want swapped", st.State, st.Reason) |
| } |
diff --git a/internal/provision/seedswap.go b/internal/provision/seedswap.go| index 2fc4dd352..1821ecc3d 100644 |
| --- a/internal/provision/seedswap.go |
| +++ b/internal/provision/seedswap.go |
| @@ -157,8 +157,9 @@ func importRow(rec Record, dataDir string, policy *UpdatePolicy) registry.Route |
| // swapped. Transient failures defer to the next pass and write nothing; a |
| // failed predicate is a durable refusal. policy is the app's update policy, |
| // known (Reconcile defers the swap, as it defers a mint, while it cannot be |
| -// read). |
| -func swapImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record, existing *registry.Route, nested func() (bool, error), policy *UpdatePolicy) { |
| +// read). ap is the app policy's columns for the import's row: the verified |
| +// policy's, or none while it cannot be read, as for any mint. |
| +func swapImport(ctx context.Context, s release.Store, reg *registry.Registry, cfg ReconcileConfig, rec Record, existing *registry.Route, nested func() (bool, error), policy *UpdatePolicy, ap appPolicyCols) { |
| if cfg.DryRun { |
| log.Printf("provision: dry-run: would consider swapping %s for import %s", rec.Host, rec.Seed.ID) |
| return |
| @@ -188,6 +189,7 @@ func swapImport(ctx context.Context, s release.Store, reg *registry.Registry, cf |
| } |
| next := importRow(rec, cfg.DataDir, policy) |
| + next.EgressAllow, next.Size, next.PolicyExpiresAt = ap.EgressAllow, ap.Size, ap.expiresCol() |
| if why := importUngateable(rec, next, cfg.DataDir); why != "" { |
| log.Printf("provision: %s: not swapping for import %s: %s", rec.Host, rec.Seed.ID, why) |
| return |
diff --git a/internal/provision/seedswap_test.go b/internal/provision/seedswap_test.go| index c4d9e07fa..3ac6832d4 100644 |
| --- a/internal/provision/seedswap_test.go |
| +++ b/internal/provision/seedswap_test.go |
| @@ -344,7 +344,7 @@ func TestSwapSkipsARecordItCannotGate(t *testing.T) { |
| row := mustGet(t, reg, stored.Host) |
| cfg := importCfg(t, nil) |
| got, _ := reg.Get(stored.Host) |
| - swapImport(ctx, s, reg, cfg, rec, got, mintLayoutReader(ctx, s, cfg), nil) |
| + swapImport(ctx, s, reg, cfg, rec, got, mintLayoutReader(ctx, s, cfg), nil, appPolicyCols{}) |
| wantUntouched(t, s, reg, stored, row) |
| }) |
| } |
| @@ -578,7 +578,7 @@ func TestSwapLeavesAnotherSeedsJournalAlone(t *testing.T) { |
| } |
| row := mustGet(t, reg, rec.Host) |
| j, _ := reg.ImportJournal(rec.Host) |
| - swapImport(ctx, s, reg, importCfg(t, nil), rec, row, func() (bool, error) { return true, nil }, nil) |
| + swapImport(ctx, s, reg, importCfg(t, nil), rec, row, func() (bool, error) { return true, nil }, nil, appPolicyCols{}) |
| if st := seedState(t, s, rec); st.State != seeds.StateReady || st.Reason != "" { |
| t.Fatalf("record = %s %q, want ready", st.State, st.Reason) |
| } |