amadan

carlos / platform Public

registry: egress_allow, size and policy_expires_at columns (#328, #329)

The agent's converge of control/app-policy/ lands on three route columns,
written together by SetPolicy (local and over the remote registry). Upsert
keeps them for the same tenant and takes the incoming values on a tenant
change. They stay out of Fingerprint and are hashed in the snapshot only
when set.

Ruling PF-1: policy_expires_at is TEXT NOT NULL DEFAULT '' (decimal unix
seconds when converged) and Route.PolicyExpiresAt is a string, so a row
with no policy reads '' and digests byte-for-byte as on d3757ae7 (golden
test).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 56c813919727c38ffc68d916e065fb3800538947 parent d3757ae
4 files changed, +255 −22
  • internal/registry/policycols_test.go +167 −0
  • internal/registry/registry.go +73 −17
  • internal/registry/registry_test.go +8 −5
  • internal/registry/remote_dispatch.go +7 −0
diff --git a/internal/registry/policycols_test.go b/internal/registry/policycols_test.go
new file mode 100644
index 000000000..4416f31fe
--- /dev/null
+++ b/internal/registry/policycols_test.go
@@ -0,0 +1,167 @@
+package registry
+
+import (
+ "path/filepath"
+ "testing"
+)
+
+const policyHost = "shop-a.example.com"
+
+func policyRow(t *testing.T, reg *Registry) Route {
+ t.Helper()
+ got, err := reg.Get(policyHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got == nil {
+ t.Fatalf("no row for %s", policyHost)
+ }
+ return *got
+}
+
+func wantPolicy(t *testing.T, got Route, allow, size, expires string) {
+ t.Helper()
+ if got.EgressAllow != allow || got.Size != size || got.PolicyExpiresAt != expires {
+ t.Fatalf("policy = (%q, %q, %q), want (%q, %q, %q)",
+ got.EgressAllow, got.Size, got.PolicyExpiresAt, allow, size, expires)
+ }
+}
+
+func TestPolicyColumns(t *testing.T) {
+ reg := openTestRegistry(t)
+ row := Route{Host: policyHost, Kind: "instance", Account: "abc", App: "shop"}
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+ f0, s0 := claimFingerprints(t, reg)
+
+ if err := reg.SetPolicy(policyHost, "10.0.5.20 fd00:5::20", "large", "1900000000"); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "10.0.5.20 fd00:5::20", "large", "1900000000")
+ f1, s1 := claimFingerprints(t, reg)
+ if f1 != f0 {
+ t.Fatal("a policy moved Fingerprint: the edge would reload for a change that does not touch what it serves")
+ }
+ if s1 == s0 {
+ t.Fatal("a policy is invisible to SnapshotFingerprint: a restore would drop it")
+ }
+
+ // A full-row Upsert by the same tenant that carries no policy keeps it.
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "10.0.5.20 fd00:5::20", "large", "1900000000")
+
+ // An expiry-only change (the re-sign) is a different restore.
+ if err := reg.SetPolicy(policyHost, "10.0.5.20 fd00:5::20", "large", "1900043200"); err != nil {
+ t.Fatal(err)
+ }
+ if _, s2 := claimFingerprints(t, reg); s2 == s1 {
+ t.Fatal("an expiry-only change did not move SnapshotFingerprint")
+ }
+
+ // SetPolicy is authoritative in both directions, down to none.
+ if err := reg.SetPolicy(policyHost, "", "", ""); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+ if f, s := claimFingerprints(t, reg); f != f0 || s != s0 {
+ t.Fatal("a cleared policy did not restore the digests of a row that never had one")
+ }
+
+ if err := reg.SetPolicy("nobody.example.com", "10.0.5.20", "large", "1900000000"); err == nil {
+ t.Fatal("SetPolicy on an unknown host succeeded")
+ }
+}
+
+// Ruling PF-1: a row with no policy digests byte for byte as it did before
+// the three columns existed, so upgrading does not force a snapshot on every
+// box. The goldens were produced by this fixture on d3757ae7, the parent of
+// the commit that added the columns.
+func TestPolicyColumnsUnsetDigestsAsBefore(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.Upsert(Route{Host: policyHost, Kind: "instance", Socket: "/s", Account: "abc", App: "shop", CreatedAt: 1700000000}); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+ const wantF = "1:a38d2c4eaed5e64549201160a19193bcfabd1bc370b46a24eca47e51d9733b01"
+ const wantS = "1:8246d576d9fbce7acc7e07ee2ecb17e59e828a0e0313cb467b515038e5f71a87"
+ f, s := claimFingerprints(t, reg)
+ if f != wantF {
+ t.Fatalf("Fingerprint = %s, want %s", f, wantF)
+ }
+ if s != wantS {
+ t.Fatalf("SnapshotFingerprint = %s, want %s", s, wantS)
+ }
+}
+
+// A registry written before the columns existed gains them on open, and its
+// rows read as no policy (”, not "0").
+func TestPolicyColumnsMigrateOnOpen(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Upsert(Route{Host: policyHost, Kind: "instance", Account: "abc", App: "shop", CreatedAt: 1700000000}); err != nil {
+ t.Fatal(err)
+ }
+ _, before := claimFingerprints(t, r)
+ for _, c := range []string{"egress_allow", "size", "policy_expires_at"} {
+ if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN ` + c); err != nil {
+ t.Fatal(err)
+ }
+ }
+ r.Close()
+ reg, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+ if _, after := claimFingerprints(t, reg); after != before {
+ t.Fatalf("SnapshotFingerprint moved across the upgrade: %s -> %s", before, after)
+ }
+}
+
+// One tenant's allow, size or expiry never carries into another's row.
+func TestTenantChangeUpsertTakesTheNewTenantsPolicy(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.Upsert(Route{Host: policyHost, Kind: "instance", Account: "abc", App: "shop"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetPolicy(policyHost, "10.0.5.20", "xlarge", "1900000000"); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(Route{Host: policyHost, Kind: "instance", Account: "abc", App: "other"}); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+
+ // And the account half of the tenant.
+ if err := reg.SetPolicy(policyHost, "10.0.5.20", "xlarge", "1900000000"); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(Route{Host: policyHost, Kind: "instance", Account: "zzz", App: "other"}); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, reg), "", "", "")
+}
+
+// SetPolicy goes through the service-user child like every other write a
+// root or agent pass makes.
+func TestRemoteSetPolicy(t *testing.T) {
+ r, _ := remoteTestRegistry(t)
+ if err := r.Upsert(Route{Host: policyHost, Kind: "instance", Status: "ready", Account: "abc", App: "shop"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := r.SetPolicy(policyHost, "10.0.5.20 fd00:5::20", "large", "1900000000"); err != nil {
+ t.Fatal(err)
+ }
+ wantPolicy(t, policyRow(t, r), "10.0.5.20 fd00:5::20", "large", "1900000000")
+ if err := r.SetPolicy("nobody.example.com", "", "", ""); err == nil {
+ t.Fatal("remote SetPolicy on an unknown host succeeded")
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 4447cc982..4f86d76bd 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -315,6 +315,20 @@ type Route struct {
// changes what root installs, not what the edge serves.
UpdateSettings string
+ // EgressAllow, Size and PolicyExpiresAt are the agent's converge of the
+ // app's signed operator policy (control/app-policy/, #328, #329):
+ // space-separated normalized egress addresses, a size name, and the
+ // verified record's expiry as decimal unix seconds. "" in each = none
+ // (ruling PF-1: the expiry is a string so an unset row reads '' and
+ // digests exactly as before the columns existed). Root reads them
+ // (plans 2, 3) and treats an allow with no expiry, or a past one, as no
+ // allow. The registry validates nothing: the rails are at the door and
+ // in root. Written by SetPolicy. Deliberately NOT in Fingerprint(): they
+ // change what root writes, not what the edge serves.
+ EgressAllow string
+ Size string
+ PolicyExpiresAt string
+
// BuildState is the agent's copy of the build-state object (JSON of
// release.BuildStateCopy; "" = never read), refreshed at reconcile and
// before every wake. Root reads it. Deliberately NOT in Fingerprint():
@@ -609,6 +623,10 @@ CREATE TABLE IF NOT EXISTS routes (
`ALTER TABLE routes ADD COLUMN build_state TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN seed_source TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN import_gate TEXT NOT NULL DEFAULT ''`,
+ // The app policy's converge (#328, #329; Route.EgressAllow): '' = none.
+ `ALTER TABLE routes ADD COLUMN egress_allow TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN size TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN policy_expires_at TEXT NOT NULL DEFAULT ''`,
} {
if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
@@ -897,10 +915,10 @@ dns_checked_at, cert_expires_at, account, environment, caps, tunnel,
provisioned, seed, unit, fleet_box, addrs, health, steering, gate, created_at,
args, idle_seconds, max_run_seconds, replicas, replica, claim_incarnation, claim_stamp, claim_owner, claim_row_id, hold_op,
good_version, good_digest, good_demand, serving_version, update_settings, build_state,
-seed_source, import_gate`
+seed_source, import_gate, egress_allow, size, policy_expires_at`
// routeColCount is the number of names in routeCols.
-const routeColCount = 55
+const routeColCount = 58
// placeholders is n comma-joined bind parameters.
func placeholders(n int) string {
@@ -918,7 +936,7 @@ func scanRoute(scan func(...any) error) (Route, error) {
&t.Addrs, &t.Health, &t.Steering, &t.Gate, &t.CreatedAt,
&t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID, &t.HoldOp,
&t.GoodVersion, &t.GoodDigest, &t.GoodDemand, &t.ServingVersion, &t.UpdateSettings, &t.BuildState,
- &t.SeedSource, &t.ImportGate)
+ &t.SeedSource, &t.ImportGate, &t.EgressAllow, &t.Size, &t.PolicyExpiresAt)
t.Hibernate = hib != 0
t.Canary = can != 0
t.StatusNamed = sn != 0
@@ -1119,8 +1137,10 @@ func (r *Registry) Delete(host string) error {
// Route53 latency records — a real incident this repo caught in review before
// it shipped, not a hypothetical.
// `update_settings` and `build_state` keep an existing value only while the
-// row's tenant (effective account, app) stays the same. An Upsert that moves
-// the hostname to another tenant takes the incoming values, empty included:
+// row's tenant (effective account, app) stays the same; so do `egress_allow`,
+// `size` and `policy_expires_at` (#328, #329: one tenant's allow is never
+// another's). An Upsert that moves the hostname to another tenant takes the
+// incoming values, empty included:
// one tenant's manual or major hold, or its build state, is never another
// tenant's (settings-5 task 8). The account compared is the effective one
// (EffectiveAccount, as release.RouteAccount reads it), so a legacy row's
@@ -1208,7 +1228,13 @@ ON CONFLICT(host) DO UPDATE SET
WHEN excluded.update_settings = '' THEN routes.update_settings ELSE excluded.update_settings END,
build_state=CASE WHEN `+tenantChangedSQL+` THEN excluded.build_state
WHEN excluded.build_state = '' THEN routes.build_state ELSE excluded.build_state END,
- seed_source=CASE WHEN excluded.seed_source = '' THEN routes.seed_source ELSE excluded.seed_source END
+ seed_source=CASE WHEN excluded.seed_source = '' THEN routes.seed_source ELSE excluded.seed_source END,
+ egress_allow=CASE WHEN `+tenantChangedSQL+` THEN excluded.egress_allow
+ WHEN excluded.egress_allow = '' THEN routes.egress_allow ELSE excluded.egress_allow END,
+ size=CASE WHEN `+tenantChangedSQL+` THEN excluded.size
+ WHEN excluded.size = '' THEN routes.size ELSE excluded.size END,
+ policy_expires_at=CASE WHEN `+tenantChangedSQL+` THEN excluded.policy_expires_at
+ WHEN excluded.policy_expires_at = '' THEN routes.policy_expires_at ELSE excluded.policy_expires_at END
WHERE routes.claim_incarnation=excluded.claim_incarnation AND routes.claim_stamp=excluded.claim_stamp
AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.claim_row_id`,
routeArgs(normalizeRoute(t))...)
@@ -1269,7 +1295,7 @@ func routeArgs(t Route) []any {
t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds,
t.Replicas, b2iDB(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp,
t.GoodVersion, t.GoodDigest, t.GoodDemand, t.ServingVersion, t.UpdateSettings, t.BuildState,
- t.SeedSource, t.ImportGate,
+ t.SeedSource, t.ImportGate, t.EgressAllow, t.Size, t.PolicyExpiresAt,
}
}
@@ -1993,6 +2019,32 @@ func (r *Registry) SetUpdateSettings(host, col string) error {
return nil
}
+// SetPolicy converges a route's app policy columns (Route.EgressAllow, Size,
+// PolicyExpiresAt) as one narrow UPDATE: the agent's converge of
+// control/app-policy/ writes all three every pass, in both directions,
+// including down to "". It validates nothing (the rails are at the door and
+// in root). Errors on an unmatched host, like SetUpdateSettings.
+func (r *Registry) SetPolicy(host, egressAllow, size, expiresAt string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(nil, remoteRequest{Method: "SetPolicy", Host: host, EgressAllow: egressAllow, Size: size, PolicyExpiresAt: expiresAt}, nil)
+ return err
+ }
+
+ res, err := r.sql.Exec(`UPDATE routes SET egress_allow = ?, size = ?, policy_expires_at = ? WHERE host = ?`,
+ egressAllow, size, expiresAt, host)
+ if err != nil {
+ return err
+ }
+ n, err := res.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n == 0 {
+ return fmt.Errorf("no route for host %q", host)
+ }
+ return nil
+}
+
// SetBuildState stores the agent's copy of a route's build-state object as a
// narrow UPDATE: reconcile and the wake path refresh it, in both directions,
// including down to "". Errors on an unmatched host, like SetUpdateSettings.
@@ -3298,17 +3350,21 @@ var snapshotClaimCols = map[string]bool{
"claim_row_id": true,
}
-// snapshotWhenSetCols are the good-build columns (zero-downtime spec §8.5)
-// and the update-settings column. A restore has to reproduce them, so they
-// are hashed, but only when set and tagged, on the hold_op rule: a registry
-// that has never recorded a good build or set update settings digests exactly
-// as it did before the columns existed, so upgrading does not force a snapshot
-// on every box.
+// snapshotWhenSetCols are the good-build columns (zero-downtime spec §8.5),
+// the update-settings column and the app policy columns (#328, #329). A
+// restore has to reproduce them, so they are hashed, but only when set and
+// tagged, on the hold_op rule: a registry that has never recorded a good
+// build, set update settings or converged a policy digests exactly as it did
+// before the columns existed, so upgrading does not force a snapshot on every
+// box.
var snapshotWhenSetCols = map[string]bool{
- "good_version": true,
- "good_digest": true,
- "good_demand": true,
- "update_settings": true,
+ "good_version": true,
+ "good_digest": true,
+ "good_demand": true,
+ "update_settings": true,
+ "egress_allow": true,
+ "size": true,
+ "policy_expires_at": true,
}
func (r *Registry) SnapshotFingerprint() (string, error) {
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index ab3b7b1fd..7a46d012d 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -82,11 +82,14 @@ func TestUpsertRoundTripsAllColumns(t *testing.T) {
Gate: "$2a$10$distinct",
CreatedAt: 1700000000,
GoodVersion: "good789", GoodDigest: "sha256:good", GoodDemand: "swap-demand",
- ServingVersion: "serve012",
- UpdateSettings: `{"updates":"manual","channel":"stable"}`,
- BuildState: `{"known":true,"state":{"schema":1,"last_executed":"bs345"}}`,
- SeedSource: `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`,
- ImportGate: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+ ServingVersion: "serve012",
+ UpdateSettings: `{"updates":"manual","channel":"stable"}`,
+ BuildState: `{"known":true,"state":{"schema":1,"last_executed":"bs345"}}`,
+ SeedSource: `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`,
+ ImportGate: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+ EgressAllow: "10.0.5.20 fd00:5::20",
+ Size: "large",
+ PolicyExpiresAt: "1900000000",
}
if err := reg.Upsert(want); err != nil {
t.Fatal(err)
diff --git a/internal/registry/remote_dispatch.go b/internal/registry/remote_dispatch.go
index 23adccb5e..1c58ede18 100644
--- a/internal/registry/remote_dispatch.go
+++ b/internal/registry/remote_dispatch.go
@@ -92,6 +92,11 @@ type remoteRequest struct {
Keep map[string]bool
ExpectGood GoodBuild
NextGood GoodBuild
+
+ // SetPolicy's three columns (#328, #329).
+ EgressAllow string
+ Size string
+ PolicyExpiresAt string
}
type remoteResponse struct {
@@ -243,6 +248,8 @@ func (r *Registry) dispatchRemote(ctx context.Context, q remoteRequest, callback
err = r.SetSteering(q.Host, q.Steering)
case "SetUpdateSettings":
err = r.SetUpdateSettings(q.Host, q.UpdateSettings)
+ case "SetPolicy":
+ err = r.SetPolicy(q.Host, q.EgressAllow, q.Size, q.PolicyExpiresAt)
case "SetBuildState":
err = r.SetBuildState(q.Host, q.BuildState)
case "SetReplicas":