amadan

carlos / platform Public

edge asset cache task 8: the box's authority clocks and publication

The home box keeps two authority clocks (assetauthority.go): a per-app
records confirmed time fed by each reconcile pass's ReconcileReport, and
a domain confirmed time fed by each complete ReconcileAliases pass, both
at the pass's start time. A host's at is the older of the two.

An armed box (CARLOS_ASSET_CACHE=1) publishes an assets block for each
host the rule allows (assetpolicy.go): not opted out, a generation, no
gate or import gate, not pooled, no path rows, one owner, a known
authority, and the union of the installed and serving builds' declared
paths, read once per build from the bucket manifest (assetmanifests.go)
and pruned to the builds local routes reference. Aliases take their
primary's block. The writer copies the clocks before it reads the
registry, rewrites when an at would advance 20 minutes or more, and the
loop also runs a pass on a serving-build change or a due rewrite.

assetPolicy implements edge.LocalAssets (For, Declared) for plan 3.
Both alias legs go through aliasPassReporting, serialized. The writer
test's host entry is the golden file Task 9 reads.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 68b93e5e171480d9a638b251b60db8317b2d2c31 parent 60f91da
12 files changed, +2191 −35
  • internal/edge/assettypes.go +34 −0
  • internal/edgesfile/testdata/assets-host.golden.json +11 −0
  • internal/host/agent.go +62 −10
  • internal/host/assetauthority.go +156 −0
  • internal/host/assetmanifests.go +87 −0
  • internal/host/assetpolicy.go +412 −0
  • internal/host/assetpublish_test.go +1329 −0
  • internal/host/boxaddress_test.go +7 −7
  • internal/host/changeindex.go +5 −9
  • internal/host/edgesclaims.go +12 −2
  • internal/host/edgesclaims_test.go +1 −1
  • internal/host/edgesfile.go +75 −6
diff --git a/internal/edge/assettypes.go b/internal/edge/assettypes.go
new file mode 100644
index 000000000..b14f327c2
--- /dev/null
+++ b/internal/edge/assettypes.go
@@ -0,0 +1,34 @@
+package edge
+
+import "time"
+
+// HostAssets is one hostname's asset-cache answer (edge asset cache design
+// §4.1, §4.2): the one app that owns everything served under the name, the
+// URL path prefixes its serving builds declare, the instance's asset
+// generation, and the host's authority time. At is the last moment the home
+// box knew every input of the catch decision for the host was complete and
+// applied; a reader serves nothing for a host whose At is over an hour old.
+//
+// A far edge builds these from the edges files it folds; the home box builds
+// them from its own registry through LocalAssets.
+type HostAssets struct {
+ Account, App string
+ Paths []string
+ Gen string
+ At time.Time
+}
+
+// LocalAssets is the home box's own view (design §4.2: "a box serves and
+// fills its own routes' caught paths under the same rule, against its own
+// confirmed time"). It is declared here for the reason Activator is:
+// internal/host implements it and imports this package, never the reverse.
+//
+// For answers for a hostname this box serves, with the one-hour freshness
+// rule already applied, so false means "catch nothing for this host now".
+// Declared reports whether build's declared paths hold a prefix of path for
+// the app host belongs to; it never reads the bucket, and a build whose
+// declaration the box has not read answers false.
+type LocalAssets interface {
+ For(host string) (HostAssets, bool)
+ Declared(host, build, path string) bool
+}
diff --git a/internal/edgesfile/testdata/assets-host.golden.json b/internal/edgesfile/testdata/assets-host.golden.json
new file mode 100644
index 000000000..197270a1b
--- /dev/null
+++ b/internal/edgesfile/testdata/assets-host.golden.json
@@ -0,0 +1,11 @@
+{
+ "role": "home",
+ "platform": false,
+ "account": "abc",
+ "app": "shop",
+ "assets": {
+ "paths": ["/_app/immutable/", "/assets/"],
+ "gen": "0123456789abcdef",
+ "at": 1792832400
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index ee9104601..161cb439f 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -832,7 +832,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}
go retentionLoop(reg, collectBuildStates)
go registrySnapshotLoop(reg)
- go edgesFileLoop(reg)
+ go edgesFileLoop(reg, servingBuildOf(svc))
// Region HA enrollment evidence (spec 1 §5.1): once, at start, and only
// with CARLOS_REGION_HA on. See boxrecord.go.
go boxRecordLoop()
@@ -1003,6 +1003,48 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}
}
+// servingBuildOf is the edges file writer's view of the activator (edge asset
+// cache design §4.1): the build host's current generation runs, from
+// ServingGen, on a box with generations on; always false otherwise, and false
+// for a host with no live generation (asleep or waking), which then
+// publishes its installed build's paths alone.
+func servingBuildOf(svc *activator.Service) servingBuild {
+ if svc == nil || !svc.GenerationsOn() {
+ return noServingBuild
+ }
+ return func(host string) (string, bool) {
+ g, ok := svc.ServingGen(host)
+ if !ok || g.Build == "" {
+ return "", false
+ }
+ return g.Build, true
+ }
+}
+
+// aliasPassMu serializes the two alias legs (the fast converge pass and the
+// five-minute instances pass). They read and rewrite the same three tables,
+// and the domain confirmed time assumes passes finish in the order they
+// started: an older pass finishing after a newer one could write rows older
+// than the time the newer one recorded.
+var aliasPassMu sync.Mutex
+
+// aliasPassReporting is one ReconcileAliases pass, the way both alias legs
+// run it, reporting to auth (edge asset cache design §4.2). The deployment
+// route set is folded once per pass and a partial fold is carried into the
+// report; the pass's start time, taken before either read, becomes the
+// domain confirmed time when the report says complete. The error is the
+// pass's own, for the caller to log as before.
+func aliasPassReporting(ctx context.Context, store release.Store, reg *registry.Registry, cache *domains.ClaimCache, auth *assetAuthority) error {
+ aliasPassMu.Lock()
+ defer aliasPassMu.Unlock()
+ start := auth.start()
+ deploymentRoutes, routesWhole := edgesfile.DeploymentRoutesComplete(ctx, store, time.Now())
+ rep := domains.AliasReport{DeploymentRoutesPartial: deploymentRoutes != nil && !routesWhole}
+ err := domains.ReconcileAliasesReport(ctx, store, reg, time.Now, deploymentRoutes, cache, &rep)
+ auth.aliasPass(start, rep)
+ return err
+}
+
// statusSocketProbe is the kernel-table probe the status loop hands
// wrapListeningProbe: listeningProbe itself with generations off (or no
// activator), and pointed at each managed route's current generation with
@@ -1601,8 +1643,20 @@ func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Reg
// agent, a box's own worker in a test. A nil worker, or one on a box that
// cannot take part in an offline move, adds nothing.
func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig, w *movesWorker) error {
+ return reconcileSerializedAuth(ctx, s, reg, cfg, w, boxAssetAuthority)
+}
+
+// reconcileSerializedAuth is reconcileSerializedWith reporting to auth, the
+// box's asset authority (edge asset cache design §4.2): every app the
+// instance pass read completely and applied gets the pass's START time as its
+// records confirmed time, taken before the pass refreshes the record view or
+// reads anything, because a record written while the pass ran may not have
+// been seen. The report is fresh per pass (ReconcileReport is overwritten).
+// Sidecar records decide no host's catch: sidecars are never published.
+func reconcileSerializedAuth(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig, w *movesWorker, auth *assetAuthority) error {
reconcileMu.Lock()
defer reconcileMu.Unlock()
+ start := auth.start()
// With the change index armed, the record walks below read the record
// view's snapshot, refreshed here under reconcileMu so a pass's snapshot
// is never older than the previous pass's (S3 change-index design §4.3).
@@ -1644,10 +1698,12 @@ func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry
// Runs ride this pass's record reads (run spec §5.2): the worker notes
// every in-scope record with a run, and acts after the pass.
cfg = withRunsHook(cfg)
- return errors.Join(
- provision.Reconcile(ctx, rs, reg, cfg),
- provision.ReconcileSidecars(ctx, rs, reg, cfg),
- )
+ var rep provision.ReconcileReport
+ instances := cfg
+ instances.Report = &rep
+ err := provision.Reconcile(ctx, rs, reg, instances)
+ auth.recordsPass(start, rep)
+ return errors.Join(err, provision.ReconcileSidecars(ctx, rs, reg, cfg))
}
// bounce builds the restart callback SyncExec calls for a route whose
@@ -3611,11 +3667,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildSta
plansPass(ctx, store, reg, os.Getenv)
runMovesPass(ctx, store)
runRunsPass(ctx)
- deploymentRoutes, routesWhole := edgesfile.DeploymentRoutesComplete(ctx, store, time.Now())
- // The report carries "the route read was partial" into the pass;
- // Task 8 reads its Complete to feed the domain confirmed time.
- aliasRep := domains.AliasReport{DeploymentRoutesPartial: deploymentRoutes != nil && !routesWhole}
- if err := domains.ReconcileAliasesReport(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache, &aliasRep); err != nil {
+ if err := aliasPassReporting(ctx, store, reg, &aliasClaimCache, boxAssetAuthority); err != nil {
log.Printf("agent: instances alias pass: %v", err)
}
if !isFleetBox {
diff --git a/internal/host/assetauthority.go b/internal/host/assetauthority.go
new file mode 100644
index 000000000..1734217ea
--- /dev/null
+++ b/internal/host/assetauthority.go
@@ -0,0 +1,156 @@
+package host
+
+import (
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// assetAuthority holds the box's two authority clocks (edge asset cache
+// design §4.2, "The invariant").
+//
+// - The records confirmed time, per app ("<acct>/<app>"): the start time of
+// the last reconcile pass that read the app's records completely and
+// applied them (provision.ReconcileReport).
+// - The domain confirmed time, for the whole box: the start time of the
+// last ReconcileAliases pass that was complete (domains.AliasReport).
+//
+// A host's authority time is the older of the two for its app. It is the `at`
+// a home publishes in its edges file, and every edge serves a host's cached
+// assets only while that time is within the last hour. A time that is too
+// new lets a stale generation or owner be served fleet-wide, so the clocks
+// only ever move forward, only on a complete pass, and only to that pass's
+// start: a record or claim written while the pass ran may not have been
+// seen. Publishing never moves them. A restarted process starts with both
+// empty, so nothing is caught until the first complete convergence.
+//
+// Any new input a future change adds to the catch decision must bring its
+// own completeness report into this minimum.
+type assetAuthority struct {
+ // now is the clock pass start times are read from (time.Now in the
+ // agent; a fake clock in tests).
+ now func() time.Time
+
+ mu sync.Mutex
+ records map[string]time.Time
+ domain time.Time
+ domainOK bool
+ // known counts the clocks that have gone from unknown to known. The
+ // edges file loop runs a pass when it moves, so a host that becomes
+ // publishable after a restart is published within seconds, not at the
+ // five-minute timer.
+ known uint64
+}
+
+func newAssetAuthority(now func() time.Time) *assetAuthority {
+ if now == nil {
+ now = time.Now
+ }
+ return &assetAuthority{now: now, records: map[string]time.Time{}}
+}
+
+// boxAssetAuthority is this process's authority: one agent per process, and
+// the reconcile passes, both alias legs and the edges file writer must all
+// read and write the same clocks (aliasClaimCache's reason).
+var boxAssetAuthority = newAssetAuthority(time.Now)
+
+// start is a pass's start time, read once before the pass reads anything.
+func (a *assetAuthority) start() time.Time { return a.now() }
+
+// recordsPass records a reconcile pass that started at start. Each app the
+// report names complete has its records confirmed time moved to start; every
+// other app keeps the time it had. A pass that failed reports nothing.
+func (a *assetAuthority) recordsPass(start time.Time, rep provision.ReconcileReport) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ for id, ok := range rep.Complete {
+ if !ok {
+ continue
+ }
+ prev, had := a.records[id]
+ if !had {
+ a.known++
+ }
+ if !had || start.After(prev) {
+ a.records[id] = start
+ }
+ }
+}
+
+// aliasPass records a ReconcileAliases pass that started at start. Only a
+// complete pass moves the domain confirmed time.
+func (a *assetAuthority) aliasPass(start time.Time, rep domains.AliasReport) {
+ if !rep.Complete {
+ return
+ }
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if !a.domainOK {
+ a.known++
+ }
+ if !a.domainOK || start.After(a.domain) {
+ a.domain, a.domainOK = start, true
+ }
+}
+
+// hostAt is the authority time of a host owned by acct/app: the older of the
+// app's records confirmed time and the box's domain confirmed time. False
+// until both are known.
+func (a *assetAuthority) hostAt(acct, app string) (time.Time, bool) {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ rec, ok := a.records[acct+"/"+app]
+ if !ok || !a.domainOK {
+ return time.Time{}, false
+ }
+ if a.domain.Before(rec) {
+ return a.domain, true
+ }
+ return rec, true
+}
+
+// knownCount is the number of unknown-to-known transitions so far.
+func (a *assetAuthority) knownCount() uint64 {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return a.known
+}
+
+// assetClocks is a copy of the authority's clocks taken at one moment.
+//
+// A publication takes it BEFORE it reads the registry. A pass reports to the
+// authority only after its registry writes, so every pass the copy credits
+// had finished writing before the rows were read: the rows are at least as
+// new as the times. Read the other way round, a pass could finish between
+// the two reads and its newer time would be paired with the older rows it
+// replaced, which is exactly a stale generation or owner under a fresh at.
+type assetClocks struct {
+ records map[string]time.Time
+ domain time.Time
+ domainOK bool
+ known uint64
+}
+
+func (a *assetAuthority) snapshot() assetClocks {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ c := assetClocks{records: make(map[string]time.Time, len(a.records)), domain: a.domain, domainOK: a.domainOK, known: a.known}
+ for k, v := range a.records {
+ c.records[k] = v
+ }
+ return c
+}
+
+// hostAt is assetAuthority.hostAt over the copy.
+func (c assetClocks) hostAt(acct, app string) (time.Time, bool) {
+ rec, ok := c.records[acct+"/"+app]
+ if !ok || !c.domainOK {
+ return time.Time{}, false
+ }
+ if c.domain.Before(rec) {
+ return c.domain, true
+ }
+ return rec, true
+}
diff --git a/internal/host/assetmanifests.go b/internal/host/assetmanifests.go
new file mode 100644
index 000000000..be81ddeee
--- /dev/null
+++ b/internal/host/assetmanifests.go
@@ -0,0 +1,87 @@
+package host
+
+import (
+ "context"
+ "slices"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// assetManifestCache holds the asset paths each build declares (edge asset
+// cache design §4.1), read from the bucket's release manifest with
+// release.GetManifest. Never the staged copy under /opt/carlos/versions: an
+// older binary may have re-marshalled it without the assets key.
+//
+// A manifest is immutable once shipped, so a build's paths are read once and
+// kept, and only the paths are kept. A read that fails is not kept: that
+// build publishes nothing until a later read succeeds. Only the builds a
+// local route references are kept (prune, called by the edges file writer at
+// the end of each pass), so what is held is bounded by twice the box's
+// routes whatever the deploy history, and a rollback to a dropped build reads
+// its manifest again.
+type assetManifestCache struct {
+ store release.Store
+
+ mu sync.Mutex
+ byKey map[string][]string
+}
+
+func newAssetManifestCache(s release.Store) *assetManifestCache {
+ return &assetManifestCache{store: s, byKey: map[string][]string{}}
+}
+
+func assetManifestKey(acct, app, version string) string {
+ return acct + "/" + app + "/" + version
+}
+
+// paths answers build version of acct/app's declared asset paths, reading the
+// bucket if it has not been read yet. False when the read fails; a build that
+// declares nothing answers no paths and true. An empty version names no build
+// and answers no paths without a read.
+func (c *assetManifestCache) paths(ctx context.Context, acct, app, version string) ([]string, bool) {
+ if version == "" {
+ return nil, true
+ }
+ if p, ok := c.cached(acct, app, version); ok {
+ return p, true
+ }
+ m, err := release.GetManifest(ctx, c.store, acct, app, version)
+ if err != nil {
+ return nil, false
+ }
+ var p []string
+ if m.Assets != nil {
+ p = slices.Clone(m.Assets.Paths)
+ }
+ c.mu.Lock()
+ c.byKey[assetManifestKey(acct, app, version)] = p
+ c.mu.Unlock()
+ return slices.Clone(p), true
+}
+
+// cached is paths without a read: false when the build has not been read.
+func (c *assetManifestCache) cached(acct, app, version string) ([]string, bool) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ p, ok := c.byKey[assetManifestKey(acct, app, version)]
+ return slices.Clone(p), ok
+}
+
+// prune drops every build whose assetManifestKey is not in keep.
+func (c *assetManifestCache) prune(keep map[string]bool) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ for k := range c.byKey {
+ if !keep[k] {
+ delete(c.byKey, k)
+ }
+ }
+}
+
+// size is how many builds are held.
+func (c *assetManifestCache) size() int {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return len(c.byKey)
+}
diff --git a/internal/host/assetpolicy.go b/internal/host/assetpolicy.go
new file mode 100644
index 000000000..bfb99b5c6
--- /dev/null
+++ b/internal/host/assetpolicy.go
@@ -0,0 +1,412 @@
+package host
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "slices"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The home box's publication rule for the edge asset cache (design §3.3,
+// §4.1, §4.2).
+
+// edgeAssets is a host's "assets" block in the edges file:
+//
+// "assets": {"paths": ["/assets/"], "gen": "9f2c41d07a3be516", "at": 1792828800}
+//
+// At is the host's authority time in Unix seconds (assetAuthority.hostAt).
+type edgeAssets struct {
+ Paths []string `json:"paths"`
+ Gen string `json:"gen"`
+ At int64 `json:"at"`
+}
+
+const (
+ // assetFreshFor: a host whose authority time is older than this catches
+ // nothing, on any edge, the home included (design §4.2).
+ assetFreshFor = time.Hour
+ // assetRewriteAfter: the edges file is rewritten when a published at
+ // would advance by this much, so a healthy host's published at stays
+ // well inside assetFreshFor without a write on every pass.
+ assetRewriteAfter = 20 * time.Minute
+)
+
+// assetCacheArmed is the arming switch: CARLOS_ASSET_CACHE=1 exactly. Only an
+// armed box publishes assets blocks, so far edges catch a host only when its
+// home is armed (design §4.1, astra B3).
+func assetCacheArmed(getenv func(string) string) bool {
+ return getenv("CARLOS_ASSET_CACHE") == "1"
+}
+
+// servingBuild answers the build host's current generation runs, when the
+// box has generations and the host has one: the activator's
+// ServingGen(host).Build. False otherwise, which includes a sleeping
+// instance. The edges file writer has no activator handle, so the agent hands
+// it this.
+type servingBuild func(host string) (build string, ok bool)
+
+// noServingBuild is servingBuild on a box without generations.
+func noServingBuild(string) (string, bool) { return "", false }
+
+// servingBuildsDigest hashes every local route's serving build. The edges file
+// loop folds it into its change key beside the registry Fingerprint, so a
+// generation switch, which changes no registry row, republishes within one
+// watch interval (astra P1-13).
+func servingBuildsDigest(routes []registry.Route, sb servingBuild) string {
+ if sb == nil {
+ sb = noServingBuild
+ }
+ h := sha256.New()
+ hosts := make([]string, 0, len(routes))
+ for _, r := range routes {
+ hosts = append(hosts, r.Host)
+ }
+ slices.Sort(hosts)
+ for _, host := range hosts {
+ if b, ok := sb(host); ok {
+ fmt.Fprintf(h, "%s\x1f%s\x1e", host, b)
+ }
+ }
+ return hex.EncodeToString(h.Sum(nil))
+}
+
+// edgesChangeKey is the edges file loop's change key on an armed box: the
+// registry fingerprint and the serving-build digest together.
+func edgesChangeKey(fp, servingDigest string) string {
+ return fp + "\x1dserving\x1e" + servingDigest
+}
+
+// assetPolicy decides, for each hostname this box serves, whether it catches
+// assets and with what block. It holds the registry, the authority clocks,
+// the manifest cache and the serving-build function. Publish is the edges
+// file's rule; For and Declared are what plan 3's edge consumes through
+// edge.LocalAssets. The writer's own state (written, knownAt) is touched only
+// by the edges file loop's goroutine.
+type assetPolicy struct {
+ reg *registry.Registry
+ auth *assetAuthority
+ manifests *assetManifestCache
+ serving servingBuild
+
+ // written is the at the last written file published for each host, and
+ // the owner it was computed for: rewriteDue's input.
+ written map[string]writtenAssets
+ // knownAt is auth.known as of the last successful pass.
+ knownAt uint64
+ // lastDigest is the serving-build digest of the last read that worked.
+ lastDigest string
+}
+
+type writtenAssets struct {
+ account, app string
+ at int64
+}
+
+var _ edge.LocalAssets = (*assetPolicy)(nil)
+
+// armAssetPolicy builds the box's policy, or nil when the box is not armed:
+// the writer then publishes no assets blocks and reads no manifests.
+func armAssetPolicy(getenv func(string) string, reg *registry.Registry, auth *assetAuthority, manifests *assetManifestCache, sb servingBuild) *assetPolicy {
+ if !assetCacheArmed(getenv) {
+ return nil
+ }
+ if sb == nil {
+ sb = noServingBuild
+ }
+ return &assetPolicy{reg: reg, auth: auth, manifests: manifests, serving: sb, written: map[string]writtenAssets{}}
+}
+
+// assetInputs is the registry state one decision reads.
+type assetInputs struct {
+ byHost map[string]registry.Route
+ // aliasOf maps each exact alias name (never "*.") to its primary.
+ aliasOf map[string]string
+ // pathHosts holds every hostname with a route_paths row.
+ pathHosts map[string]bool
+ // owners is servedOwners over the served rows.
+ owners map[string]*hostOwner
+}
+
+func newAssetInputs(routes []registry.Route, served map[string]edgeHost, paths []registry.RoutePath, aliases []registry.RouteAlias, owners map[string]*hostOwner) assetInputs {
+ in := assetInputs{byHost: make(map[string]registry.Route, len(routes)), aliasOf: map[string]string{},
+ pathHosts: map[string]bool{}, owners: owners}
+ for _, r := range routes {
+ in.byHost[r.Host] = r
+ }
+ for _, p := range paths {
+ in.pathHosts[p.Host] = true
+ }
+ for _, a := range aliases {
+ if a.Alias == "" || strings.HasPrefix(a.Alias, "*.") {
+ continue
+ }
+ in.aliasOf[a.Alias] = a.Host
+ }
+ if in.owners == nil {
+ in.owners = servedOwners(routes, served, paths)
+ }
+ return in
+}
+
+// readAssetInputs reads the inputs fresh from the registry, with the served
+// set the writer would start from (ready, never a sidecar).
+func (p *assetPolicy) readAssetInputs() (assetInputs, error) {
+ routes, err := p.reg.List()
+ if err != nil {
+ return assetInputs{}, err
+ }
+ paths, err := p.reg.ListPaths()
+ if err != nil {
+ return assetInputs{}, err
+ }
+ aliases, err := p.reg.ListAliases()
+ if err != nil {
+ return assetInputs{}, err
+ }
+ served := make(map[string]edgeHost, len(routes))
+ for _, r := range routes {
+ if r.Status == "ready" && !r.IsSidecar() {
+ served[r.Host] = edgeHost{}
+ }
+ }
+ return newAssetInputs(routes, served, paths, aliases, nil), nil
+}
+
+// primary resolves host to the route the edge serves it with: its own row,
+// or for an exact alias its primary's (a real route beats an alias).
+func (in assetInputs) primary(host string) (registry.Route, bool) {
+ if r, ok := in.byHost[host]; ok {
+ return r, true
+ }
+ if ph, ok := in.aliasOf[host]; ok {
+ r, ok := in.byHost[ph]
+ return r, ok
+ }
+ return registry.Route{}, false
+}
+
+// decide is the publication rule for one hostname. pooledOK drops the pooled
+// exclusion (For: a pooled route still catches on the boxes that serve it).
+// cachedOnly answers from manifests already read, never the bucket (For runs
+// on the request path).
+func (p *assetPolicy) decide(ctx context.Context, in assetInputs, clocks assetClocks, host string, pooledOK, cachedOnly bool) (edge.HostAssets, bool) {
+ if p == nil || host == "" || strings.HasPrefix(host, "*.") {
+ return edge.HostAssets{}, false
+ }
+ r, ok := in.primary(host)
+ if !ok || r.Status != "ready" || r.IsSidecar() {
+ return edge.HostAssets{}, false
+ }
+ // The record's opt-out and generation, the password gate, and the import
+ // gate of the row the hostname resolves to (an alias resolves to its
+ // primary's row, and a hostname with path rows is refused below).
+ if r.AssetsOff || r.AssetGen == "" || r.Gate != "" || r.ImportGate != "" {
+ return edge.HostAssets{}, false
+ }
+ // Any path-prefix row at all, on the name or on the primary it is served
+ // with: one hostname-wide declaration cannot describe what different
+ // targets serve.
+ if in.pathHosts[host] || in.pathHosts[r.Host] {
+ return edge.HostAssets{}, false
+ }
+ o := in.owners[r.Host]
+ if o == nil {
+ return edge.HostAssets{}, false
+ }
+ acct, app := o.owner()
+ if acct == "" || app == "" {
+ return edge.HostAssets{}, false
+ }
+ at, ok := clocks.hostAt(acct, app)
+ if !ok {
+ return edge.HostAssets{}, false
+ }
+ builds := []string{r.Version}
+ if b, ok := p.serving(r.Host); ok && b != "" {
+ builds = append(builds, b)
+ }
+ var paths []string
+ for _, b := range builds {
+ var bp []string
+ var ok bool
+ if cachedOnly {
+ bp, ok = p.manifests.cached(acct, app, b)
+ if b == "" {
+ ok = true
+ }
+ } else {
+ bp, ok = p.manifests.paths(ctx, acct, app, b)
+ }
+ if !ok {
+ continue // that build publishes no paths until it can be read
+ }
+ paths = append(paths, bp...)
+ }
+ slices.Sort(paths)
+ paths = slices.Compact(paths)
+ if len(paths) == 0 {
+ return edge.HostAssets{}, false
+ }
+ // Pooled last, after the manifests are read: the writer's pass is what
+ // fills the cache For answers from, and a pooled route still catches on
+ // the boxes that serve it.
+ if r.Pool != "" && !pooledOK {
+ return edge.HostAssets{}, false
+ }
+ return edge.HostAssets{Account: acct, App: app, Paths: paths, Gen: r.AssetGen, At: at}, true
+}
+
+// Publish is the edges file's rule for host, over the registry as it is now.
+// The writer applies the same rule to the rows it has already read.
+func (p *assetPolicy) Publish(host string) (edgeAssets, bool) {
+ if p == nil {
+ return edgeAssets{}, false
+ }
+ clocks := p.auth.snapshot() // before the registry reads: see assetClocks
+ in, err := p.readAssetInputs()
+ if err != nil {
+ return edgeAssets{}, false
+ }
+ a, ok := p.decide(context.Background(), in, clocks, host, false, false)
+ if !ok {
+ return edgeAssets{}, false
+ }
+ return edgeAssets{Paths: a.Paths, Gen: a.Gen, At: a.At.Unix()}, true
+}
+
+// For is the home's own catch for host (edge.LocalAssets): the rule without
+// the pooled exclusion, answered only while the authority time is within
+// assetFreshFor. It reads the registry and never the bucket.
+func (p *assetPolicy) For(host string) (edge.HostAssets, bool) {
+ if p == nil {
+ return edge.HostAssets{}, false
+ }
+ clocks := p.auth.snapshot()
+ in, err := p.readAssetInputs()
+ if err != nil {
+ return edge.HostAssets{}, false
+ }
+ a, ok := p.decide(context.Background(), in, clocks, host, true, true)
+ if !ok || p.auth.now().Sub(a.At) > assetFreshFor {
+ return edge.HostAssets{}, false
+ }
+ return a, true
+}
+
+// Declared reports whether build, of the app host is served by, declares a
+// prefix of path (edge.LocalAssets). It answers from the manifest cache
+// alone: a build not read yet declares nothing.
+func (p *assetPolicy) Declared(host, build, path string) bool {
+ if p == nil || build == "" {
+ return false
+ }
+ in, err := p.readAssetInputs()
+ if err != nil {
+ return false
+ }
+ r, ok := in.primary(host)
+ if !ok {
+ return false
+ }
+ paths, ok := p.manifests.cached(release.RouteAccount(r), r.App, build)
+ if !ok {
+ return false
+ }
+ return slices.ContainsFunc(paths, func(prefix string) bool { return strings.HasPrefix(path, prefix) })
+}
+
+// keepBuilds is the manifest cache's keep set: the installed and serving
+// builds of every local route (design §4.1, R5-1).
+func (p *assetPolicy) keepBuilds(routes []registry.Route) map[string]bool {
+ keep := make(map[string]bool, 2*len(routes))
+ for _, r := range routes {
+ acct := release.RouteAccount(r)
+ if r.Version != "" {
+ keep[assetManifestKey(acct, r.App, r.Version)] = true
+ }
+ if b, ok := p.serving(r.Host); ok && b != "" {
+ keep[assetManifestKey(acct, r.App, b)] = true
+ }
+ }
+ return keep
+}
+
+// servingDigest is servingBuildsDigest over the registry's routes now; a
+// failed read answers the last digest, so a read error never looks like a
+// change.
+func (p *assetPolicy) servingDigest() string {
+ routes, err := p.reg.List()
+ if err != nil {
+ return p.lastDigest
+ }
+ p.lastDigest = servingBuildsDigest(routes, p.serving)
+ return p.lastDigest
+}
+
+// rewriteDue reports whether the edges file loop should run a pass for the
+// assets alone: a clock has become known since the last pass (the first
+// complete convergence after a restart), or some published at would advance
+// by assetRewriteAfter or more.
+func (p *assetPolicy) rewriteDue() bool {
+ if p == nil {
+ return false
+ }
+ if p.auth.knownCount() != p.knownAt {
+ return true
+ }
+ for _, w := range p.written {
+ if at, ok := p.auth.hostAt(w.account, w.app); ok && at.Unix()-w.at >= int64(assetRewriteAfter/time.Second) {
+ return true
+ }
+ }
+ return false
+}
+
+// passDone records a successful writer pass: the clocks it used, and, when
+// it wrote, what it published.
+func (p *assetPolicy) passDone(clocks assetClocks, wrote bool, hosts map[string]edgeHost) {
+ p.knownAt = clocks.known
+ if !wrote {
+ return
+ }
+ p.written = map[string]writtenAssets{}
+ for host, h := range hosts {
+ if h.Assets != nil {
+ p.written[host] = writtenAssets{account: h.Account, app: h.App, at: h.Assets.At}
+ }
+ }
+}
+
+// heldAssetAts is hosts as the change gate compares them: a host whose
+// published at would advance by less than assetRewriteAfter over the at in
+// the last written file (last, the gate body) is compared with that older at,
+// so the advance alone is not a reason to write. Any other change to the
+// block, or an at that went back, still differs. hosts is not modified.
+func heldAssetAts(hosts map[string]edgeHost, last []byte) map[string]edgeHost {
+ var prior struct {
+ Hosts map[string]edgeHost `json:"hosts"`
+ }
+ if len(last) == 0 || json.Unmarshal(last, &prior) != nil {
+ return hosts
+ }
+ out := make(map[string]edgeHost, len(hosts))
+ for host, h := range hosts {
+ if was, ok := prior.Hosts[host]; ok && h.Assets != nil && was.Assets != nil {
+ if d := h.Assets.At - was.Assets.At; d >= 0 && d < int64(assetRewriteAfter/time.Second) {
+ held := *h.Assets
+ held.At = was.Assets.At
+ h.Assets = &held
+ }
+ }
+ out[host] = h
+ }
+ return out
+}
diff --git a/internal/host/assetpublish_test.go b/internal/host/assetpublish_test.go
new file mode 100644
index 000000000..befc34a3e
--- /dev/null
+++ b/internal/host/assetpublish_test.go
@@ -0,0 +1,1329 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "slices"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/edgesfile"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The edge asset cache's publication on the home box (design §3.3, §4.1,
+// §4.2). The writer tests drive the real edgesFileWriteClaims over a DirStore
+// and read the written object back; the authority table drives the real
+// reconcile pass, the real alias pass and the real writer together.
+
+var assetT0 = time.Date(2026, 10, 24, 9, 0, 0, 0, time.UTC)
+
+const (
+ assetGen1 = "0123456789abcdef"
+ assetGen2 = "fedcba9876543210"
+)
+
+// assetClock is a settable clock shared by a test's authority and policy.
+type assetClock struct {
+ mu sync.Mutex
+ t time.Time
+}
+
+func (c *assetClock) now() time.Time {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.t
+}
+
+func (c *assetClock) advance(d time.Duration) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.t = c.t.Add(d)
+}
+
+func armedAssetEnv(k string) string {
+ if k == "CARLOS_ASSET_CACHE" {
+ return "1"
+ }
+ return ""
+}
+
+// assetFaultStore fails chosen reads and nothing else: GETs of named keys
+// and listings under named prefixes, with an error that is not
+// os.ErrNotExist, the way a transient bucket error reads. It counts GETs.
+type assetFaultStore struct {
+ release.Store
+ mu sync.Mutex
+ getFail map[string]bool
+ listFail []string
+ gets map[string]int
+}
+
+func newAssetFaultStore(s release.Store) *assetFaultStore {
+ return &assetFaultStore{Store: s, getFail: map[string]bool{}, gets: map[string]int{}}
+}
+
+func (f *assetFaultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ f.mu.Lock()
+ f.gets[key]++
+ fail := f.getFail[key]
+ f.mu.Unlock()
+ if fail {
+ return nil, "", errors.New("synthetic read failure")
+ }
+ return f.Store.Get(ctx, key)
+}
+
+func (f *assetFaultStore) List(ctx context.Context, prefix string) ([]string, error) {
+ f.mu.Lock()
+ fail := slices.ContainsFunc(f.listFail, func(p string) bool { return strings.HasPrefix(prefix, p) })
+ f.mu.Unlock()
+ if fail {
+ return nil, errors.New("synthetic list failure")
+ }
+ return f.Store.List(ctx, prefix)
+}
+
+func (f *assetFaultStore) failGet(key string, on bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.getFail[key] = on
+}
+
+func (f *assetFaultStore) failList(prefix string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.listFail = append(f.listFail, prefix)
+}
+
+func (f *assetFaultStore) getsOf(key string) int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.gets[key]
+}
+
+// assetShip ships app shop version v for acct into s, declaring paths.
+func assetShip(t *testing.T, s release.Store, acct, version string, paths ...string) {
+ t.Helper()
+ m := release.Manifest{App: "shop", Version: version, Kind: "binary",
+ Artifacts: map[string]release.Artifact{"linux-arm64": {}}}
+ if len(paths) > 0 {
+ m.Assets = &release.Assets{Paths: paths}
+ }
+ if err := release.Ship(context.Background(), s, acct, m, map[string][]byte{"linux-arm64": []byte("build " + version)}); err != nil {
+ t.Fatalf("ship %s: %v", version, err)
+ }
+}
+
+// ---- the asset world: registry rows written directly, account abc --------
+
+type assetWorld struct {
+ t *testing.T
+ store *assetFaultStore // manifests are read through it
+ writes *putCountingStore
+ reg *registry.Registry
+ clock *assetClock
+ auth *assetAuthority
+ cache *assetManifestCache
+ policy *assetPolicy
+ last []byte
+
+ mu sync.Mutex
+ serving map[string]string
+}
+
+func newAssetWorld(t *testing.T) *assetWorld {
+ t.Helper()
+ dir := release.NewDirStore(t.TempDir())
+ w := &assetWorld{t: t, store: newAssetFaultStore(dir), writes: &putCountingStore{Store: dir},
+ reg: edgesTestRegistry(t), clock: &assetClock{t: assetT0}, serving: map[string]string{}}
+ w.auth = newAssetAuthority(w.clock.now)
+ w.cache = newAssetManifestCache(w.store)
+ w.policy = armAssetPolicy(armedAssetEnv, w.reg, w.auth, w.cache, w.servingBuild)
+ if w.policy == nil {
+ t.Fatal("an armed environment built no policy")
+ }
+ return w
+}
+
+func (w *assetWorld) servingBuild(host string) (string, bool) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ b, ok := w.serving[host]
+ return b, ok
+}
+
+func (w *assetWorld) setServing(host, build string) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if build == "" {
+ delete(w.serving, host)
+ return
+ }
+ w.serving[host] = build
+}
+
+func (w *assetWorld) ship(version string, paths ...string) {
+ w.t.Helper()
+ assetShip(w.t, w.store.Store, "abc", version, paths...)
+}
+
+// instance upserts a ready instance route of abc/shop.
+func (w *assetWorld) instance(host, gen, version string, mod ...func(*registry.Route)) {
+ w.t.Helper()
+ r := registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/" + host + ".sock", Status: "ready",
+ Account: "abc", App: "shop", AssetGen: gen, Version: version}
+ for _, m := range mod {
+ m(&r)
+ }
+ if err := w.reg.Upsert(r); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
+// confirm records a complete records pass for abc/shop and a complete alias
+// pass, both starting now.
+func (w *assetWorld) confirm() {
+ now := w.clock.now()
+ w.auth.recordsPass(now, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ w.auth.aliasPass(now, domains.AliasReport{Complete: true})
+}
+
+func (w *assetWorld) pass() {
+ w.t.Helper()
+ if err := edgesFileWriteClaims(context.Background(), w.writes, w.reg, edgesFile{Box: "annie"}, "1.2.3.4", &w.last, false, nil, w.policy); err != nil {
+ w.t.Fatal(err)
+ }
+}
+
+// raw is the written file's host entries, unparsed.
+func (w *assetWorld) raw() map[string]json.RawMessage {
+ w.t.Helper()
+ body, _, err := w.writes.Get(context.Background(), edgesFileKey("annie"))
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ var f struct {
+ Hosts map[string]json.RawMessage `json:"hosts"`
+ }
+ if err := json.Unmarshal(body, &f); err != nil {
+ w.t.Fatal(err)
+ }
+ return f.Hosts
+}
+
+// hosts runs one pass and reads the written hosts back.
+func (w *assetWorld) hosts() map[string]edgeHost {
+ w.t.Helper()
+ w.pass()
+ return w.written()
+}
+
+func (w *assetWorld) written() map[string]edgeHost {
+ w.t.Helper()
+ body, _, err := w.writes.Get(context.Background(), edgesFileKey("annie"))
+ if err != nil {
+ w.t.Fatal(err)
+ }
+ var f edgesFile
+ if err := json.Unmarshal(body, &f); err != nil {
+ w.t.Fatal(err)
+ }
+ return f.Hosts
+}
+
+func wantNoAssets(t *testing.T, hosts map[string]edgeHost, host string) {
+ t.Helper()
+ h, ok := hosts[host]
+ if !ok {
+ t.Fatalf("%s not published at all: %+v", host, hosts)
+ }
+ if h.Assets != nil {
+ t.Fatalf("%s published assets %+v, want none", host, *h.Assets)
+ }
+}
+
+func wantAssets(t *testing.T, hosts map[string]edgeHost, host string, paths []string, gen string, at time.Time) {
+ t.Helper()
+ h, ok := hosts[host]
+ if !ok {
+ t.Fatalf("%s not published: %+v", host, hosts)
+ }
+ if h.Assets == nil {
+ t.Fatalf("%s published no assets: %+v", host, h)
+ }
+ if !slices.Equal(h.Assets.Paths, paths) || h.Assets.Gen != gen || h.Assets.At != at.Unix() {
+ t.Fatalf("%s assets = %+v, want paths %v gen %s at %d", host, *h.Assets, paths, gen, at.Unix())
+ }
+}
+
+// ---- the clocks -----------------------------------------------------------
+
+func TestAssetAuthorityHostAtIsTheOlderClock(t *testing.T) {
+ t.Parallel()
+ a := newAssetAuthority(nil)
+ t1, t2 := assetT0, assetT0.Add(10*time.Minute)
+ a.recordsPass(t2, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ a.aliasPass(t1, domains.AliasReport{Complete: true})
+ if got, ok := a.hostAt("abc", "shop"); !ok || !got.Equal(t1) {
+ t.Fatalf("hostAt = %v %v, want the domain clock %v", got, ok, t1)
+ }
+ a.aliasPass(t2.Add(5*time.Minute), domains.AliasReport{Complete: true})
+ if got, ok := a.hostAt("abc", "shop"); !ok || !got.Equal(t2) {
+ t.Fatalf("hostAt = %v %v, want the records clock %v", got, ok, t2)
+ }
+}
+
+func TestAssetAuthorityIncompletePassDoesNotAdvance(t *testing.T) {
+ t.Parallel()
+ a := newAssetAuthority(nil)
+ a.recordsPass(assetT0, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ a.aliasPass(assetT0, domains.AliasReport{Complete: true})
+ later := assetT0.Add(time.Hour)
+ // The records pass left the app out; the alias pass was incomplete.
+ a.recordsPass(later, provision.ReconcileReport{Complete: map[string]bool{"abc/other": true}})
+ a.aliasPass(later, domains.AliasReport{Complete: false})
+ if got, ok := a.hostAt("abc", "shop"); !ok || !got.Equal(assetT0) {
+ t.Fatalf("hostAt = %v %v, want it held at %v", got, ok, assetT0)
+ }
+ // Each clock alone: a complete alias pass does not move the records
+ // clock, and an app marked false does not count.
+ a.aliasPass(later, domains.AliasReport{Complete: true})
+ a.recordsPass(later, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": false}})
+ if got, _ := a.hostAt("abc", "shop"); !got.Equal(assetT0) {
+ t.Fatalf("hostAt = %v, want the records clock still at %v", got, assetT0)
+ }
+ // A pass that started earlier than the time already held moves nothing.
+ a.recordsPass(assetT0.Add(-time.Minute), provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ if got, _ := a.hostAt("abc", "shop"); !got.Equal(assetT0) {
+ t.Fatalf("hostAt = %v, want %v: the clock went backwards", got, assetT0)
+ }
+}
+
+func TestAssetAuthorityUnknownUntilBothClocks(t *testing.T) {
+ t.Parallel()
+ a := newAssetAuthority(nil)
+ if _, ok := a.hostAt("abc", "shop"); ok {
+ t.Fatal("a fresh authority answered")
+ }
+ a.recordsPass(assetT0, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ if _, ok := a.hostAt("abc", "shop"); ok {
+ t.Fatal("answered with the records clock alone")
+ }
+ b := newAssetAuthority(nil)
+ b.aliasPass(assetT0, domains.AliasReport{Complete: true})
+ if _, ok := b.hostAt("abc", "shop"); ok {
+ t.Fatal("answered with the domain clock alone")
+ }
+ a.aliasPass(assetT0, domains.AliasReport{Complete: true})
+ if _, ok := a.hostAt("abc", "shop"); !ok {
+ t.Fatal("both clocks known and still no answer")
+ }
+ if _, ok := a.hostAt("abc", "cafe"); ok {
+ t.Fatal("answered for an app no pass ever confirmed")
+ }
+}
+
+// TestAssetAuthorityRecordsTheStartTime runs the agent's own reconcile and
+// alias wiring with a clock that moves an hour on every read after the
+// first: a pass that took its time at the end, or read it twice, records a
+// later time than the one it started at.
+func TestAssetAuthorityRecordsTheStartTime(t *testing.T) {
+ t.Parallel()
+ b := newAuthorityBox(t)
+ var mu sync.Mutex
+ reads := 0
+ stepping := func() time.Time {
+ mu.Lock()
+ defer mu.Unlock()
+ reads++
+ return assetT0.Add(time.Duration(reads-1) * time.Hour)
+ }
+ auth := newAssetAuthority(stepping)
+ if err := reconcileSerializedAuth(context.Background(), b.recs, b.reg, b.reconcileConfig(), nil, auth); err != nil {
+ t.Fatal(err)
+ }
+ if err := aliasPassReporting(context.Background(), b.doms, b.reg, nil, auth); err != nil {
+ t.Fatal(err)
+ }
+ // Records started at T0 (read 1), aliases at T0+1h (read 2): hostAt is
+ // the older, T0, until the records clock is moved past the domain's.
+ got, ok := auth.hostAt(b.acct, "shop")
+ if !ok || !got.Equal(assetT0) {
+ t.Fatalf("hostAt = %v %v, want the records pass start %v", got, ok, assetT0)
+ }
+ auth.recordsPass(assetT0.Add(5*time.Hour), provision.ReconcileReport{Complete: map[string]bool{b.acct + "/shop": true}})
+ if got, _ := auth.hostAt(b.acct, "shop"); !got.Equal(assetT0.Add(time.Hour)) {
+ t.Fatalf("hostAt = %v, want the alias pass start %v", got, assetT0.Add(time.Hour))
+ }
+ mu.Lock()
+ n := reads
+ mu.Unlock()
+ if n != 2 {
+ t.Fatalf("the two passes read the clock %d times, want once each", n)
+ }
+}
+
+// ---- the manifest cache ---------------------------------------------------
+
+func TestAssetManifestCacheReadsTheBucketOnce(t *testing.T) {
+ t.Parallel()
+ s := newAssetFaultStore(release.NewDirStore(t.TempDir()))
+ assetShip(t, s.Store, "abc", "v1", "/assets/", "/static/")
+ c := newAssetManifestCache(s)
+ for range 3 {
+ got, ok := c.paths(context.Background(), "abc", "shop", "v1")
+ if !ok || !slices.Equal(got, []string{"/assets/", "/static/"}) {
+ t.Fatalf("paths = %v %v", got, ok)
+ }
+ }
+ if n := s.getsOf(release.ManifestKey("abc", "shop", "v1")); n != 1 {
+ t.Fatalf("the manifest was read %d times, want once", n)
+ }
+ // A build that declares nothing is cached too, as no paths.
+ assetShip(t, s.Store, "abc", "v2")
+ for range 2 {
+ if got, ok := c.paths(context.Background(), "abc", "shop", "v2"); !ok || len(got) != 0 {
+ t.Fatalf("v2 paths = %v %v, want none, read", got, ok)
+ }
+ }
+ if n := s.getsOf(release.ManifestKey("abc", "shop", "v2")); n != 1 {
+ t.Fatalf("the undeclared manifest was read %d times, want once", n)
+ }
+}
+
+func TestAssetManifestCacheDoesNotCacheAFailure(t *testing.T) {
+ t.Parallel()
+ s := newAssetFaultStore(release.NewDirStore(t.TempDir()))
+ assetShip(t, s.Store, "abc", "v1", "/assets/")
+ key := release.ManifestKey("abc", "shop", "v1")
+ c := newAssetManifestCache(s)
+ s.failGet(key, true)
+ if got, ok := c.paths(context.Background(), "abc", "shop", "v1"); ok || got != nil {
+ t.Fatalf("a failed read answered %v %v", got, ok)
+ }
+ if _, ok := c.cached("abc", "shop", "v1"); ok {
+ t.Fatal("a failed read was cached")
+ }
+ if got, ok := c.paths(context.Background(), "abc", "shop", "v2-never-shipped"); ok || got != nil {
+ t.Fatalf("a missing manifest answered %v %v", got, ok)
+ }
+ s.failGet(key, false)
+ if got, ok := c.paths(context.Background(), "abc", "shop", "v1"); !ok || !slices.Equal(got, []string{"/assets/"}) {
+ t.Fatalf("after recovery paths = %v %v", got, ok)
+ }
+ if n := s.getsOf(key); n != 2 {
+ t.Fatalf("reads = %d, want the failed one and one more", n)
+ }
+}
+
+// One route deployed through 50 versions, with the serving build one behind
+// the installed one, keeps at most the two builds it references, and a
+// rollback to a version the cache dropped reads the bucket again.
+func TestAssetManifestCacheStaysBoundedAcrossDeploys(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.confirm()
+ ver := func(i int) string { return fmt.Sprintf("v%d", i) }
+ for i := 1; i <= 50; i++ {
+ w.ship(ver(i), fmt.Sprintf("/b%d/", i))
+ }
+ w.instance("shop.example.com", assetGen1, ver(1))
+ w.pass()
+ for i := 2; i <= 50; i++ {
+ if err := w.reg.SetVersions("shop.example.com", ver(i), ver(i-1)); err != nil {
+ t.Fatal(err)
+ }
+ w.setServing("shop.example.com", ver(i-1))
+ hosts := w.hosts()
+ if n := w.cache.size(); n > 2 {
+ t.Fatalf("after deploy %d the cache holds %d builds, want at most 2", i, n)
+ }
+ want := []string{fmt.Sprintf("/b%d/", i-1), fmt.Sprintf("/b%d/", i)}
+ slices.Sort(want)
+ wantAssets(t, hosts, "shop.example.com", want, assetGen1, assetT0)
+ }
+ before := w.store.getsOf(release.ManifestKey("abc", "shop", ver(3)))
+ if err := w.reg.SetVersions("shop.example.com", ver(3), ver(50)); err != nil {
+ t.Fatal(err)
+ }
+ w.setServing("shop.example.com", "")
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/b3/"}, assetGen1, assetT0)
+ if after := w.store.getsOf(release.ManifestKey("abc", "shop", ver(3))); after != before+1 {
+ t.Fatalf("the rollback read v3's manifest %d more times, want 1", after-before)
+ }
+ if n := w.cache.size(); n != 1 {
+ t.Fatalf("after the rollback the cache holds %d builds, want 1", n)
+ }
+}
+
+// ---- publication ----------------------------------------------------------
+
+// TestPublishesAssetsForAnArmedHome is the acceptance case: an armed home's
+// instance publishes paths, gen and at, the file still parses through
+// internal/edgesfile, and the host entry equals the golden file Task 9's
+// readers are tested against.
+func TestPublishesAssetsForAnArmedHome(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/", "/_app/immutable/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ hosts := w.hosts()
+ wantAssets(t, hosts, "shop.example.com", []string{"/_app/immutable/", "/assets/"}, assetGen1, assetT0)
+
+ golden, err := os.ReadFile(filepath.Join("..", "edgesfile", "testdata", "assets-host.golden.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var want bytes.Buffer
+ if err := json.Compact(&want, golden); err != nil {
+ t.Fatal(err)
+ }
+ if got := w.raw()["shop.example.com"]; !bytes.Equal(got, want.Bytes()) {
+ t.Fatalf("host entry =\n%s\nwant (golden)\n%s", got, want.Bytes())
+ }
+ body, _, err := w.writes.Get(context.Background(), edgesFileKey("annie"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ f, err := edgesfile.Parse(body)
+ if err != nil {
+ t.Fatalf("the file no longer parses: %v", err)
+ }
+ if h, ok := f.Hosts["shop.example.com"]; !ok || h.Role != edgesfile.RoleHome {
+ t.Fatalf("parsed entry = %+v %v", h, ok)
+ }
+}
+
+func TestUnarmedBoxPublishesNoAssets(t *testing.T) {
+ t.Parallel()
+ for v, want := range map[string]bool{"1": true, "": false, "0": false, "true": false, " 1": false, "yes": false} {
+ if got := assetCacheArmed(func(k string) string {
+ if k == "CARLOS_ASSET_CACHE" {
+ return v
+ }
+ return ""
+ }); got != want {
+ t.Errorf("CARLOS_ASSET_CACHE=%q armed = %v, want %v", v, got, want)
+ }
+ }
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ if p := armAssetPolicy(func(string) string { return "" }, w.reg, w.auth, w.cache, w.servingBuild); p != nil {
+ t.Fatal("an unarmed environment built a policy")
+ }
+ w.policy = nil // what the loop hands the writer on an unarmed box
+ before := w.store.getsOf(release.ManifestKey("abc", "shop", "v1"))
+ hosts := w.hosts()
+ wantNoAssets(t, hosts, "shop.example.com")
+ if n := w.store.getsOf(release.ManifestKey("abc", "shop", "v1")); n != before {
+ t.Fatal("an unarmed box read a release manifest")
+ }
+ if bytes.Contains(w.raw()["shop.example.com"], []byte(`"assets"`)) {
+ t.Fatal("an unarmed box wrote an assets key")
+ }
+}
+
+// Each §3.3 exclusion, alone, on a host that otherwise publishes.
+func TestPublishesNothingUnderAnExclusion(t *testing.T) {
+ t.Parallel()
+ cases := []struct {
+ name string
+ setup func(w *assetWorld)
+ }{
+ {"opted out", func(w *assetWorld) {
+ if err := w.reg.SetAssets("shop.example.com", true, assetGen1); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ {"no generation", func(w *assetWorld) {
+ if err := w.reg.SetAssets("shop.example.com", false, ""); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ {"gated", func(w *assetWorld) {
+ if err := w.reg.SetGate("shop.example.com", "$2a$10$abcdefghijklmnopqrstuv"); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ {"import-gated", func(w *assetWorld) {
+ if err := w.reg.Delete("shop.example.com"); err != nil {
+ w.t.Fatal(err)
+ }
+ w.instance("shop.example.com", assetGen1, "v1", func(r *registry.Route) { r.ImportGate = registry.ImportGateClosed })
+ if rt, err := w.reg.Get("shop.example.com"); err != nil || rt == nil || rt.ImportGate == "" {
+ w.t.Fatalf("the fixture did not set the import gate: %+v %v", rt, err)
+ }
+ }},
+ {"pooled", func(w *assetWorld) {
+ w.instance("shop.example.com", assetGen1, "v1", func(r *registry.Route) { r.Pool = "team" })
+ }},
+ {"a path row under the host", func(w *assetWorld) {
+ w.instance("api.shop.example.com", assetGen2, "v1")
+ if err := w.reg.UpsertPath(registry.RoutePath{Host: "shop.example.com", Prefix: "api", Target: "api.shop.example.com", Account: "abc", App: "shop"}); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ {"split owners", func(w *assetWorld) {
+ w.instance("blog.shop.example.com", assetGen2, "v1", func(r *registry.Route) { r.App = "blog" })
+ if err := w.reg.UpsertPath(registry.RoutePath{Host: "shop.example.com", Prefix: "blog", Target: "blog.shop.example.com", Account: "abc", App: "blog"}); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ {"no owner", func(w *assetWorld) {
+ w.instance("shop.example.com", assetGen1, "v1", func(r *registry.Route) { r.App = "" })
+ }},
+ {"a manifest read failing", func(w *assetWorld) {
+ w.store.failGet(release.ManifestKey("abc", "shop", "v1"), true)
+ }},
+ {"no authority yet", func(w *assetWorld) {
+ w.auth = newAssetAuthority(w.clock.now)
+ w.policy = armAssetPolicy(armedAssetEnv, w.reg, w.auth, w.cache, w.servingBuild)
+ w.auth.recordsPass(assetT0, provision.ReconcileReport{Complete: map[string]bool{"abc/shop": true}})
+ }},
+ {"a build that declares nothing", func(w *assetWorld) {
+ w.ship("v2")
+ if err := w.reg.SetVersions("shop.example.com", "v2", "v1"); err != nil {
+ w.t.Fatal(err)
+ }
+ }},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ t.Parallel()
+ // Control: the same world without the exclusion publishes.
+ ctl := newAssetWorld(t)
+ ctl.ship("v1", "/assets/")
+ ctl.instance("shop.example.com", assetGen1, "v1")
+ ctl.confirm()
+ wantAssets(t, ctl.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ c.setup(w)
+ hosts := w.hosts()
+ if h, ok := hosts["shop.example.com"]; ok && h.Assets != nil {
+ t.Fatalf("published %+v under the exclusion", *h.Assets)
+ }
+ if _, ok := w.policy.Publish("shop.example.com"); ok {
+ t.Fatal("Publish answered under the exclusion")
+ }
+ if c.name != "pooled" {
+ if _, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatal("For answered under the exclusion")
+ }
+ }
+ })
+ }
+}
+
+// A pooled route publishes nothing for far edges but still catches on the
+// boxes that serve it (§3.3), so For answers and Publish does not.
+func TestPooledRouteCatchesLocallyOnly(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1", func(r *registry.Route) { r.Pool = "team" })
+ w.confirm()
+ wantNoAssets(t, w.hosts(), "shop.example.com")
+ got, ok := w.policy.For("shop.example.com")
+ if !ok || got.Gen != assetGen1 || !slices.Equal(got.Paths, []string{"/assets/"}) || got.Account != "abc" || got.App != "shop" || !got.At.Equal(assetT0) {
+ t.Fatalf("For = %+v %v", got, ok)
+ }
+}
+
+func TestPublishesTheUnionOfInstalledAndServingBuilds(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/b/")
+ w.ship("v2", "/a/")
+ w.instance("shop.example.com", assetGen1, "v2") // installed declares /a/
+ w.setServing("shop.example.com", "v1") // serving declares /b/
+ w.confirm()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/a/", "/b/"}, assetGen1, assetT0)
+ got, ok := w.policy.For("shop.example.com")
+ if !ok || !slices.Equal(got.Paths, []string{"/a/", "/b/"}) {
+ t.Fatalf("For = %+v %v", got, ok)
+ }
+ // Declared answers per build, by prefix, from the cache alone.
+ for _, c := range []struct {
+ build, path string
+ want bool
+ }{
+ {"v2", "/a/app.js", true}, {"v2", "/b/app.js", false},
+ {"v1", "/b/x/y.css", true}, {"v1", "/a/app.js", false},
+ {"v1", "/", false}, {"v9", "/a/app.js", false},
+ } {
+ if got := w.policy.Declared("shop.example.com", c.build, c.path); got != c.want {
+ t.Errorf("Declared(%s, %s) = %v, want %v", c.build, c.path, got, c.want)
+ }
+ }
+ // One build failing to read publishes the other's paths.
+ w2 := newAssetWorld(t)
+ w2.ship("v1", "/b/")
+ w2.ship("v2", "/a/")
+ w2.instance("shop.example.com", assetGen1, "v2")
+ w2.setServing("shop.example.com", "v1")
+ w2.confirm()
+ w2.store.failGet(release.ManifestKey("abc", "shop", "v1"), true)
+ wantAssets(t, w2.hosts(), "shop.example.com", []string{"/a/"}, assetGen1, assetT0)
+}
+
+// A sleeping instance has no serving generation; it keeps the installed
+// build's paths, the build a wake would start.
+func TestSleepingInstanceKeepsTheInstalledBuildsPaths(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1", func(r *registry.Route) { r.Hibernate = true })
+ w.confirm()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+}
+
+func TestAliasPublishesItsPrimarysBlock(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ for _, a := range []string{"www.shop.example.com", "pending.shop.example.com"} {
+ if err := w.reg.UpsertAlias(registry.RouteAlias{Alias: a, Host: "shop.example.com", Account: "abc", App: "shop"}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := w.reg.SetAliasDNSStatus("www.shop.example.com", "pointing", assetT0.Unix()); err != nil {
+ t.Fatal(err)
+ }
+ w.confirm()
+ hosts := w.hosts()
+ wantAssets(t, hosts, "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ wantAssets(t, hosts, "www.shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ if _, ok := hosts["pending.shop.example.com"]; ok {
+ t.Fatal("an alias whose DNS is not pointing was published")
+ }
+ for _, h := range []string{"www.shop.example.com", "pending.shop.example.com"} {
+ got, ok := w.policy.For(h)
+ if !ok || got.Gen != assetGen1 || got.App != "shop" {
+ t.Fatalf("For(%s) = %+v %v, want the primary's block", h, got, ok)
+ }
+ }
+ // The primary gated: its aliases stop with it.
+ if err := w.reg.SetGate("shop.example.com", "$2a$10$abcdefghijklmnopqrstuv"); err != nil {
+ t.Fatal(err)
+ }
+ hosts = w.hosts()
+ wantNoAssets(t, hosts, "www.shop.example.com")
+ if _, ok := w.policy.For("www.shop.example.com"); ok {
+ t.Fatal("For answered for an alias of a gated primary")
+ }
+}
+
+func TestCatchallAliasNeverPublishes(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ if err := w.reg.UpsertAlias(registry.RouteAlias{Alias: "*.shop.example.com", Host: "shop.example.com", Account: "abc", App: "shop"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := w.reg.UpsertWildcardDomain(registry.WildcardDomain{Host: "shop.example.com", Account: "abc", App: "shop", CreatedAt: assetT0.Unix()}); err != nil {
+ t.Fatal(err)
+ }
+ // A route row keyed on a catch-all name publishes nothing either.
+ w.instance("*.example.com", assetGen2, "v1")
+ w.confirm()
+ hosts := w.hosts()
+ wantAssets(t, hosts, "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ if h, ok := hosts["*.shop.example.com"]; !ok || h.Role != "home" {
+ t.Fatalf("the catch-all was not published as served: %+v", hosts)
+ }
+ wantNoAssets(t, hosts, "*.shop.example.com")
+ wantNoAssets(t, hosts, "*.example.com")
+ for _, h := range []string{"*.shop.example.com", "*.example.com", "any.shop.example.com"} {
+ if _, ok := w.policy.For(h); ok {
+ t.Fatalf("For(%s) answered", h)
+ }
+ }
+}
+
+// Review Focus 4: the console box homes the console's own route beside
+// instances; only the instance route publishes.
+func TestConsoleClassRouteDoesNotPublish(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ if err := w.reg.Upsert(registry.Route{Host: "console.example.com", Kind: "service", Addr: "127.0.0.1:8090", Status: "ready",
+ Platform: true, Account: release.OpsAccount, App: "console", Version: "v1"}); err != nil {
+ t.Fatal(err)
+ }
+ w.auth.recordsPass(assetT0, provision.ReconcileReport{Complete: map[string]bool{release.OpsAccount + "/console": true}})
+ w.confirm()
+ hosts := w.hosts()
+ wantAssets(t, hosts, "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ wantNoAssets(t, hosts, "console.example.com")
+ if _, ok := w.policy.For("console.example.com"); ok {
+ t.Fatal("For answered for the console route")
+ }
+}
+
+func TestRewritesWhenAtAdvancesTwentyMinutes(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ w.pass()
+ if w.writes.puts != 1 {
+ t.Fatalf("first pass puts = %d", w.writes.puts)
+ }
+ if w.policy.rewriteDue() {
+ t.Fatal("a rewrite was due straight after a write")
+ }
+ w.clock.advance(20 * time.Minute)
+ w.confirm()
+ if !w.policy.rewriteDue() {
+ t.Fatal("no rewrite due after a 20-minute advance")
+ }
+ w.pass()
+ if w.writes.puts != 2 {
+ t.Fatalf("puts = %d, want a rewrite at a 20-minute advance", w.writes.puts)
+ }
+ wantAssets(t, w.written(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0.Add(20*time.Minute))
+ if w.policy.rewriteDue() {
+ t.Fatal("a rewrite still due after it was written")
+ }
+}
+
+func TestDoesNotRewriteForASmallerAdvance(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ w.pass()
+ for range 3 {
+ w.clock.advance(6 * time.Minute)
+ w.confirm()
+ if w.policy.rewriteDue() {
+ t.Fatal("a rewrite was due under 20 minutes")
+ }
+ w.pass()
+ }
+ w.clock.advance(time.Minute + 59*time.Second) // 19m59s in all
+ w.confirm()
+ w.pass()
+ if w.writes.puts != 1 {
+ t.Fatalf("puts = %d, want 1: an advance under 20 minutes rewrote", w.writes.puts)
+ }
+ wantAssets(t, w.written(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ // A real change still writes at once, with the current at.
+ if err := w.reg.SetAssets("shop.example.com", false, assetGen2); err != nil {
+ t.Fatal(err)
+ }
+ w.pass()
+ if w.writes.puts != 2 {
+ t.Fatalf("puts = %d, want the generation change written", w.writes.puts)
+ }
+ wantAssets(t, w.written(), "shop.example.com", []string{"/assets/"}, assetGen2, w.clock.now())
+}
+
+// A host that becomes publishable when the clocks first become known (after
+// a restart) does not wait for the timer.
+func TestRewriteDueWhenTheAuthorityFirstBecomesKnown(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ wantNoAssets(t, w.hosts(), "shop.example.com")
+ if w.policy.rewriteDue() {
+ t.Fatal("due with nothing known")
+ }
+ w.confirm()
+ if !w.policy.rewriteDue() {
+ t.Fatal("not due when the clocks became known")
+ }
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ if w.policy.rewriteDue() {
+ t.Fatal("still due after the pass")
+ }
+}
+
+// A generation switch changes what the route answers with but not the
+// registry: the loop's change key moves with the serving build, so the
+// schedule runs a pass and the new build's paths are published.
+func TestServingBuildSwitchRepublishesWithAnUnchangedFingerprint(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/a/")
+ w.ship("v2", "/b/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.setServing("shop.example.com", "v1")
+ w.confirm()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/a/"}, assetGen1, assetT0)
+
+ fp1, err := w.reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ routes, err := w.reg.List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ d1 := servingBuildsDigest(routes, w.servingBuild)
+ state := &edgesLoopState{sched: newEdgesFileSchedule(assetT0, func() time.Duration { return 5 * time.Minute })}
+ now := assetT0.Add(2 * time.Second)
+ if state.dueWith(now, edgesChangeKey(fp1, d1), true, false) {
+ t.Fatal("the first key only baselines")
+ }
+ state.done(now, true)
+
+ w.setServing("shop.example.com", "v2") // the swap: a new generation, same row
+ fp2, err := w.reg.Fingerprint()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if fp2 != fp1 {
+ t.Fatal("the fixture changed the registry")
+ }
+ d2 := servingBuildsDigest(routes, w.servingBuild)
+ if d2 == d1 {
+ t.Fatal("the serving-build digest did not move with the generation")
+ }
+ now = now.Add(2 * time.Second)
+ if !state.dueWith(now, edgesChangeKey(fp2, d2), true, false) {
+ t.Fatal("a generation switch did not make a pass due")
+ }
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/a/", "/b/"}, assetGen1, assetT0)
+ state.done(now, true)
+ now = now.Add(2 * time.Second)
+ if state.dueWith(now, edgesChangeKey(fp2, d2), true, false) {
+ t.Fatal("due again with nothing changed")
+ }
+ // The digest covers every local route, and a route with no serving
+ // build contributes nothing.
+ if servingBuildsDigest(routes, noServingBuild) != servingBuildsDigest(nil, noServingBuild) {
+ t.Fatal("routes with no serving build moved the digest")
+ }
+ // A due rewrite runs a pass through the same schedule.
+ if !state.dueWith(now, edgesChangeKey(fp2, d2), true, true) {
+ t.Fatal("a due rewrite did not make a pass due")
+ }
+}
+
+// TestFailedRecordReadsStopAtFromAdvancing: with the reconcile report
+// leaving the app out for 61 minutes while alias passes and edges-file
+// writes succeed, the published at stays at the last complete pass.
+func TestFailedRecordReadsStopAtFromAdvancing(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ w.pass()
+ for range 61 {
+ w.clock.advance(time.Minute)
+ w.auth.recordsPass(w.clock.now(), provision.ReconcileReport{Complete: map[string]bool{}})
+ w.auth.aliasPass(w.clock.now(), domains.AliasReport{Complete: true})
+ w.pass()
+ }
+ wantAssets(t, w.written(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ if _, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatal("For answered with an authority time over an hour old")
+ }
+ got, ok := w.policy.Publish("shop.example.com")
+ if !ok || got.At != assetT0.Unix() {
+ t.Fatalf("Publish = %+v %v, want the old at still published", got, ok)
+ }
+}
+
+// ---- the authority table: real reconcile, real alias pass, real writer -----
+
+// authorityBox is one home box: its records, claims and edges file in one
+// DirStore; the reconcile pass reads through recs and the alias pass through
+// doms, so a fault can be put on exactly one input; the edges file is
+// written to the base store, healthy throughout.
+type authorityBox struct {
+ t *testing.T
+ acct string
+ base *release.DirStore
+ recs *assetFaultStore
+ doms *assetFaultStore
+ reg *registry.Registry
+ regPath string
+ dataDir string
+ clock *assetClock
+ auth *assetAuthority
+ policy *assetPolicy
+ last []byte
+}
+
+func newAuthorityBox(t *testing.T) *authorityBox {
+ t.Helper()
+ ctx := context.Background()
+ base := release.NewDirStore(t.TempDir())
+ regPath := filepath.Join(t.TempDir(), "registry.db")
+ reg, err := registry.Open(regPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ // A record-driven fixture needs an account manifest whose sqid matches
+ // its id, which "abc" cannot; Sqid(2) is the package's usual stand-in.
+ acct := domains.Sqid(2)
+ b := &authorityBox{t: t, acct: acct, base: base, recs: newAssetFaultStore(base), doms: newAssetFaultStore(base),
+ reg: reg, regPath: regPath, dataDir: t.TempDir(), clock: &assetClock{t: assetT0}}
+ b.auth = newAssetAuthority(b.clock.now)
+ b.policy = armAssetPolicy(armedAssetEnv, reg, b.auth, newAssetManifestCache(base), noServingBuild)
+ manifest, err := json.Marshal(map[string]any{"id": 2, "sqid": acct, "name": "fixture", "created_at": assetT0})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := base.Put(ctx, "accounts/"+acct+"/account.json", manifest, ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := provision.PutManifest(ctx, base, acct, "shop", provision.Manifest{Schema: 1, App: "shop", Domain: "example.com"}); err != nil {
+ t.Fatal(err)
+ }
+ b.record("shop.example.com", "shop")
+ assetShip(t, base, acct, "v1", "/assets/")
+ // The reconcile pass mints the row; adopt would install the build.
+ if err := reconcileSerializedAuth(ctx, b.recs, reg, b.reconcileConfig(), nil, newAssetAuthority(nil)); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetVersions("shop.example.com", "v1", ""); err != nil {
+ t.Fatal(err)
+ }
+ return b
+}
+
+func (b *authorityBox) record(host, slug string) {
+ b.t.Helper()
+ rec := provision.Record{Schema: 1, Host: host, Slug: slug, Account: b.acct, App: "shop", Channel: "stable",
+ CreatedAt: assetT0, AssetGen: assetGen1}
+ if err := provision.CreateRecord(context.Background(), b.base, rec); err != nil {
+ b.t.Fatal(err)
+ }
+}
+
+func (b *authorityBox) reconcileConfig() provision.ReconcileConfig {
+ return provision.ReconcileConfig{DataDir: b.dataDir, Layout: flatLayout}
+}
+
+// route upserts a plain ready route of the box's account and app, which no
+// reconcile pass owns.
+func (b *authorityBox) route(host string, mod ...func(*registry.Route)) {
+ b.t.Helper()
+ r := registry.Route{Host: host, Kind: "service", Addr: "10.0.0.1:3000", Status: "ready", Account: b.acct, App: "shop"}
+ for _, m := range mod {
+ m(&r)
+ }
+ if err := b.reg.Upsert(r); err != nil {
+ b.t.Fatal(err)
+ }
+}
+
+func (b *authorityBox) claim(c domains.Claim) {
+ b.t.Helper()
+ c.Schema = 1
+ if c.Account == "" {
+ c.Account = b.acct
+ }
+ body, err := json.Marshal(c)
+ if err != nil {
+ b.t.Fatal(err)
+ }
+ if _, err := b.base.Put(context.Background(), domains.ClaimKey(c.Host), body, ""); err != nil {
+ b.t.Fatal(err)
+ }
+}
+
+func (b *authorityBox) unclaim(host string) {
+ b.t.Helper()
+ if err := b.base.Delete(context.Background(), domains.ClaimKey(host)); err != nil {
+ b.t.Fatal(err)
+ }
+}
+
+// failWhen makes one registry write fail and nothing else: a BEFORE trigger
+// on table, created through a second handle on the registry's database file,
+// raising when cond holds. Reads still succeed.
+func (b *authorityBox) failWhen(event, table, cond string) {
+ b.t.Helper()
+ db, err := sql.Open("sqlite", b.regPath)
+ if err != nil {
+ b.t.Fatal(err)
+ }
+ defer db.Close()
+ name := fmt.Sprintf("inject_%s_%s_%d", strings.ToLower(event), table, time.Now().UnixNano())
+ for _, q := range []string{
+ "PRAGMA busy_timeout=5000",
+ fmt.Sprintf("CREATE TRIGGER %s BEFORE %s ON %s WHEN %s BEGIN SELECT RAISE(FAIL, 'injected'); END", name, event, table, cond),
+ } {
+ if _, err := db.Exec(q); err != nil {
+ b.t.Fatalf("%s: %v", q, err)
+ }
+ }
+}
+
+func (b *authorityBox) failUpsert(table, cond string) {
+ b.failWhen("INSERT", table, "NEW."+cond)
+ b.failWhen("UPDATE", table, "NEW."+cond)
+}
+
+func (b *authorityBox) failDelete(table, cond string) { b.failWhen("DELETE", table, "OLD."+cond) }
+
+// cycle is one round of the box's passes at the current fake time: records,
+// aliases, then the edges file, which must write without error.
+func (b *authorityBox) cycle() {
+ b.t.Helper()
+ ctx := context.Background()
+ if err := reconcileSerializedAuth(ctx, b.recs, b.reg, b.reconcileConfig(), nil, b.auth); err != nil {
+ b.t.Logf("records pass: %v", err)
+ }
+ if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
+ b.t.Logf("alias pass: %v", err)
+ }
+ if err := edgesFileWriteClaims(ctx, b.base, b.reg, edgesFile{Box: "annie"}, "1.2.3.4", &b.last, false, nil, b.policy); err != nil {
+ b.t.Fatalf("edges file write: %v", err)
+ }
+}
+
+// run cycles every five minutes of fake time for d.
+func (b *authorityBox) run(d time.Duration) {
+ b.t.Helper()
+ for end := b.clock.now().Add(d); b.clock.now().Before(end); {
+ b.clock.advance(5 * time.Minute)
+ b.cycle()
+ }
+}
+
+func (b *authorityBox) published(host string) (edgeAssets, bool) {
+ b.t.Helper()
+ body, _, err := b.base.Get(context.Background(), edgesFileKey("annie"))
+ if err != nil {
+ b.t.Fatal(err)
+ }
+ var f edgesFile
+ if err := json.Unmarshal(body, &f); err != nil {
+ b.t.Fatal(err)
+ }
+ h, ok := f.Hosts[host]
+ if !ok || h.Assets == nil {
+ return edgeAssets{}, false
+ }
+ return *h.Assets, true
+}
+
+// TestAuthorityTable is spec §12's authority table: each input of §4.2's
+// invariant held failing for 61 minutes of fake time, with every other input
+// and the edges-file writes healthy, stops the published at advancing (so it
+// is over an hour old) and the home's own catch (For). Each case has a
+// control on the identical fixture with nothing injected, which must keep
+// catching, so a stop is the injection's doing.
+func TestAuthorityTable(t *testing.T) {
+ t.Parallel()
+ const observed = "shop.example.com"
+ instanceCatchall := func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "example.org", App: "shop-site"})
+ b.claim(domains.Claim{Host: "*.example.org", App: "shop", Instances: true})
+ }
+ pathClaim := func(b *authorityBox, paths map[string]string) {
+ b.claim(domains.Claim{Host: "blog.example.com", App: "shop", Route: "blog.example.com", Paths: paths})
+ }
+ pathRoutes := func(b *authorityBox) {
+ b.route("blog.example.com")
+ b.route("api.example.com")
+ }
+ cases := []struct {
+ name string
+ setup func(b *authorityBox) // before the healthy hour
+ inject func(b *authorityBox) // then held for 61 minutes
+ }{
+ {"record scope incomplete", nil, func(b *authorityBox) {
+ b.recs.failGet(provision.RecordKey(b.acct, "shop", observed), true)
+ }},
+ {"claim listing failing", nil, func(b *authorityBox) { b.doms.failList(domains.ClaimsPrefix) }},
+ {"catch-all label count failing", instanceCatchall, func(b *authorityBox) {
+ b.doms.failList("apps/" + b.acct + "/shop/instances/")
+ }},
+ {"instance alias upsert failing", nil, func(b *authorityBox) {
+ b.failUpsert("route_aliases", "alias = 'shop.example.org'")
+ instanceCatchall(b)
+ }},
+ {"alias upsert failing", func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "www.example.com", App: "shop", Route: observed})
+ }, func(b *authorityBox) { b.failUpsert("route_aliases", "alias = 'www.example.com'") }},
+ {"catch-all upsert failing", func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "wild.example.com", App: "shop", Route: observed, Wildcard: true, Catchall: true})
+ }, func(b *authorityBox) { b.failUpsert("route_aliases", "alias = '*.wild.example.com'") }},
+ {"alias delete failing", func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "www.example.com", App: "shop", Route: observed})
+ }, func(b *authorityBox) {
+ b.failDelete("route_aliases", "alias = 'www.example.com'")
+ b.unclaim("www.example.com")
+ }},
+ {"wildcard upsert failing", nil, func(b *authorityBox) {
+ b.failUpsert("wildcard_domains", "host = 'wild.example.com'")
+ b.claim(domains.Claim{Host: "wild.example.com", App: "shop", Route: observed, Wildcard: true})
+ }},
+ {"wildcard delete failing", func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "wild.example.com", App: "shop", Route: observed, Wildcard: true})
+ }, func(b *authorityBox) {
+ b.failDelete("wildcard_domains", "host = 'wild.example.com'")
+ b.unclaim("wild.example.com")
+ }},
+ {"path upsert failing", func(b *authorityBox) {
+ pathRoutes(b)
+ pathClaim(b, map[string]string{"api": "api.example.com"})
+ }, func(b *authorityBox) { b.failUpsert("route_paths", "host = 'blog.example.com'") }},
+ {"path delete failing", func(b *authorityBox) {
+ pathRoutes(b)
+ pathClaim(b, map[string]string{"api": "api.example.com"})
+ }, func(b *authorityBox) {
+ b.failDelete("route_paths", "host = 'blog.example.com'")
+ b.unclaim("blog.example.com")
+ pathClaim(b, nil)
+ }},
+ {"an unreadable claim body", func(b *authorityBox) {
+ b.claim(domains.Claim{Host: "www.example.com", App: "shop", Route: observed})
+ }, func(b *authorityBox) { b.doms.failGet(domains.ClaimKey("www.example.com"), true) }},
+ {"deploymentRoutes nil", nil, func(b *authorityBox) { b.doms.failList(edgesfile.Prefix) }},
+ }
+ for _, c := range cases {
+ t.Run(c.name, func(t *testing.T) {
+ t.Parallel()
+ for _, injected := range []bool{false, true} {
+ b := newAuthorityBox(t)
+ if c.setup != nil {
+ c.setup(b)
+ }
+ b.cycle()
+ b.run(20 * time.Minute)
+ healthy, ok := b.published(observed)
+ if !ok {
+ t.Fatalf("injected=%v: the healthy box published no assets for %s", injected, observed)
+ }
+ if _, ok := b.policy.For(observed); !ok {
+ t.Fatalf("injected=%v: For refused on the healthy box", injected)
+ }
+ if injected {
+ c.inject(b)
+ }
+ b.run(61 * time.Minute)
+ now := b.clock.now()
+ got, ok := b.published(observed)
+ _, forOK := b.policy.For(observed)
+ if !injected {
+ if !ok || now.Sub(time.Unix(got.At, 0)) >= 25*time.Minute {
+ t.Fatalf("control: published %+v %v at %v: the fixture stopped catching on its own", got, ok, now)
+ }
+ if !forOK {
+ t.Fatal("control: For refused: the fixture stopped catching on its own")
+ }
+ continue
+ }
+ if ok && got.At > healthy.At {
+ t.Fatalf("the published at advanced from %d to %d with %s", healthy.At, got.At, c.name)
+ }
+ if ok && now.Sub(time.Unix(got.At, 0)) <= time.Hour {
+ t.Fatalf("the published at %d is within the hour at %v", got.At, now)
+ }
+ if forOK {
+ t.Fatalf("For still catches %s with %s", observed, c.name)
+ }
+ }
+ })
+ }
+}
+
+// An exact claim that takes over a name an instance catch-all derived stops
+// the name publishing its old primary's block: the name now resolves to the
+// claim's route, here a gated one.
+func TestExactClaimReplacingADerivedAliasStopsIt(t *testing.T) {
+ t.Parallel()
+ b := newAuthorityBox(t)
+ b.route("blog.example.com", func(r *registry.Route) {
+ r.Gate = "$2a$10$abcdefghijklmnopqrstuv"
+ r.AssetGen = assetGen2
+ r.Version = "v1"
+ })
+ b.claim(domains.Claim{Host: "example.org", App: "shop-site"})
+ b.claim(domains.Claim{Host: "*.example.org", App: "shop", Instances: true})
+ b.cycle()
+ got, ok := b.policy.Publish("shop.example.org")
+ if !ok || got.Gen != assetGen1 {
+ t.Fatalf("the derived name did not publish its primary's block: %+v %v", got, ok)
+ }
+ if _, ok := b.policy.For("shop.example.org"); !ok {
+ t.Fatal("For refused the derived name")
+ }
+ b.claim(domains.Claim{Host: "shop.example.org", App: "shop", Route: "blog.example.com"})
+ b.clock.advance(5 * time.Minute)
+ b.cycle()
+ aliases, err := b.reg.ListAliases()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !slices.ContainsFunc(aliases, func(a registry.RouteAlias) bool {
+ return a.Alias == "shop.example.org" && a.Host == "blog.example.com"
+ }) {
+ t.Fatalf("the exact claim did not take the name: %+v", aliases)
+ }
+ if got, ok := b.policy.Publish("shop.example.org"); ok {
+ t.Fatalf("the name still publishes %+v", got)
+ }
+ if _, ok := b.policy.For("shop.example.org"); ok {
+ t.Fatal("For still catches the name")
+ }
+}
+
+// A claim adding a path row under a primary, pointing at a gated target,
+// stops the primary publishing at all.
+func TestPathRowAddedUnderAPrimaryStopsIt(t *testing.T) {
+ t.Parallel()
+ b := newAuthorityBox(t)
+ b.route("admin.example.com", func(r *registry.Route) { r.Gate = "$2a$10$abcdefghijklmnopqrstuv" })
+ b.cycle()
+ if _, ok := b.policy.Publish("shop.example.com"); !ok {
+ t.Fatal("the primary did not publish before the path row")
+ }
+ b.claim(domains.Claim{Host: "shop.example.com", App: "shop", Route: "shop.example.com", Paths: map[string]string{"admin": "admin.example.com"}})
+ b.clock.advance(5 * time.Minute)
+ b.cycle()
+ paths, err := b.reg.ListPaths()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(paths) != 1 || paths[0].Host != "shop.example.com" {
+ t.Fatalf("the path row was not derived: %+v", paths)
+ }
+ if got, ok := b.policy.Publish("shop.example.com"); ok {
+ t.Fatalf("the primary still publishes %+v", got)
+ }
+ if _, ok := b.policy.For("shop.example.com"); ok {
+ t.Fatal("For still catches the primary")
+ }
+ if _, ok := b.published("shop.example.com"); ok {
+ t.Fatal("the written file still carries the primary's assets")
+ }
+}
+
+// Compile-time: the policy is what plan 3's edge consumes.
+var _ edge.LocalAssets = (*assetPolicy)(nil)
diff --git a/internal/host/boxaddress_test.go b/internal/host/boxaddress_test.go
index ec234441f..c57e12d7a 100644
--- a/internal/host/boxaddress_test.go
+++ b/internal/host/boxaddress_test.go
@@ -160,13 +160,13 @@ func TestEdgesFilePublishesOnlyWhatTheSourceReports(t *testing.T) {
var last []byte
ident := edgesFile{Box: "annie", BoxID: "annie"}
// The loop's own reading (the eip argument) is not what is published.
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "198.51.100.1", &last, false, claims); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "198.51.100.1", &last, false, claims, nil); err != nil {
t.Fatal(err)
}
if f := readEdgesFile(t, s, "annie"); f.EIP != "203.0.113.9" || f.PrivateIP != "10.0.1.5" || f.PrivateCIDR != "10.0.0.0/16" {
t.Fatalf("first publication: %+v", f)
}
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, false, claims); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, false, claims, nil); err != nil {
t.Fatal(err)
}
if f := readEdgesFile(t, s, "annie"); f.EIP != "" || f.PrivateIP != "10.0.1.5" {
@@ -184,10 +184,10 @@ func TestEdgesFileWithRegionHAOffKeepsTheEIPHold(t *testing.T) {
seedRoute(t, reg, "app.example.com")
var last []byte
ident := edgesFile{Box: "annie"}
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "203.0.113.9", &last, false, nil); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "203.0.113.9", &last, false, nil, nil); err != nil {
t.Fatal(err)
}
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, nil); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, nil, nil); err != nil {
t.Fatal(err)
}
if f := readEdgesFile(t, s, "annie"); f.EIP != "203.0.113.9" || f.PrivateIP != "" || f.PrivateCIDR != "" {
@@ -277,7 +277,7 @@ func TestEdgesFileUnknownAddressesLeaveThePublishedFileStanding(t *testing.T) {
claims.addrs = src
var last []byte
ident := edgesFile{Box: "annie", BoxID: "annie"}
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, false, claims); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, false, claims, nil); err != nil {
t.Fatal(err)
}
before, _, err := s.Get(context.Background(), edgesFileKey("annie"))
@@ -286,7 +286,7 @@ func TestEdgesFileUnknownAddressesLeaveThePublishedFileStanding(t *testing.T) {
}
lastBefore := string(last)
// A forced timed pass, which would rewrite an unchanged body.
- err = edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, claims)
+ err = edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, claims, nil)
if !errors.Is(err, errBoxAddressesUnknown) {
t.Fatalf("unknown read: err %v, want errBoxAddressesUnknown", err)
}
@@ -301,7 +301,7 @@ func TestEdgesFileUnknownAddressesLeaveThePublishedFileStanding(t *testing.T) {
t.Fatalf("published file lost its addresses: %+v", f)
}
// The retry reads again and publishes what it now reports.
- if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, claims); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), s, reg, ident, "", &last, true, claims, nil); err != nil {
t.Fatal(err)
}
if f := readEdgesFile(t, s, "annie"); f.PrivateIP != "10.0.1.6" {
diff --git a/internal/host/changeindex.go b/internal/host/changeindex.go
index 1a821de3d..37e8b0f80 100644
--- a/internal/host/changeindex.go
+++ b/internal/host/changeindex.go
@@ -20,8 +20,6 @@ import (
"time"
"github.com/carlosframework/platform/internal/activator"
- "github.com/carlosframework/platform/internal/domains"
- "github.com/carlosframework/platform/internal/edgesfile"
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -303,7 +301,8 @@ func (l *passLegs) run(ctx context.Context, what string, keep func(registry.Rout
// scoped pass leaves them out: the serial-triggered full pass, the
// instances floor and the backstop's full pass run them.
if keep == nil {
- // Folded once per pass and handed in, never re-read inside the
+ // The deployment's routes are folded once per pass
+ // (aliasPassReporting) and handed in, never re-read inside the
// loop: this leg runs on every serial bump (a seconds cadence),
// and the fold is a LoadBoxes plus a List plus one Get per edges
// file. On a fleet of under a dozen boxes that is a handful of
@@ -311,12 +310,9 @@ func (l *passLegs) run(ctx context.Context, what string, keep func(registry.Rout
// cache would need its own staleness rule, and a STALE route set
// is the one thing a security guard must not have. Aliases read
// the domain claims, not the record families, so they read the
- // bucket.
- deploymentRoutes, routesWhole := edgesfile.DeploymentRoutesComplete(ctx, l.store, time.Now())
- // See agent.go: the report carries "the route read was partial" into
- // the pass; Task 8 reads its Complete.
- aliasRep := domains.AliasReport{DeploymentRoutesPartial: deploymentRoutes != nil && !routesWhole}
- if err := domains.ReconcileAliasesReport(ctx, l.store, l.reg, time.Now, deploymentRoutes, &aliasClaimCache, &aliasRep); err != nil {
+ // bucket. The pass's report feeds the box's domain confirmed time
+ // (edge asset cache design §4.2).
+ if err := aliasPassReporting(ctx, l.store, l.reg, &aliasClaimCache, boxAssetAuthority); err != nil {
log.Printf("agent: fast converge alias pass: %v", err)
}
mark("aliases")
diff --git a/internal/host/edgesclaims.go b/internal/host/edgesclaims.go
index 96749ffdc..bd9c55690 100644
--- a/internal/host/edgesclaims.go
+++ b/internal/host/edgesclaims.go
@@ -255,13 +255,23 @@ type edgesLoopState struct {
claimsNext time.Time
schedDue bool
claimDue bool
+ assetDue bool
}
// due reports whether to run a pass on this tick.
func (s *edgesLoopState) due(now time.Time, fp string, fpOK bool) bool {
+ return s.dueWith(now, fp, fpOK, false)
+}
+
+// dueWith is due with the asset cache's own reason to publish: assetsDue is
+// assetPolicy.rewriteDue. Like a claim pass, an assets pass waits out the
+// schedule's retry backoff after a failed write and moves it with its own
+// outcome, and is never the timer's.
+func (s *edgesLoopState) dueWith(now time.Time, fp string, fpOK, assetsDue bool) bool {
s.schedDue = s.sched.due(now, fp, fpOK)
s.claimDue = s.claims != nil && !now.Before(s.claimsNext) && !now.Before(s.sched.retryAt) && s.claims.backlog(now)
- return s.schedDue || s.claimDue
+ s.assetDue = assetsDue && !now.Before(s.sched.retryAt)
+ return s.schedDue || s.claimDue || s.assetDue
}
// timed reports whether the pass due just allowed is the timer's.
@@ -272,7 +282,7 @@ func (s *edgesLoopState) timed() bool { return s.schedDue && s.sched.timed }
// clears it and records the fingerprint the pass published. A claim pass
// is never the timer's (due cleared timed), so the timer does not move.
func (s *edgesLoopState) done(now time.Time, ok bool) {
- if s.schedDue || s.claimDue {
+ if s.schedDue || s.claimDue || s.assetDue {
s.sched.done(now, ok)
}
if s.claims != nil {
diff --git a/internal/host/edgesclaims_test.go b/internal/host/edgesclaims_test.go
index a3f40a519..aed820ba6 100644
--- a/internal/host/edgesclaims_test.go
+++ b/internal/host/edgesclaims_test.go
@@ -57,7 +57,7 @@ func publishWithClaims(t *testing.T, reg *registry.Registry, claims *edgesClaims
}
st := release.NewDirStore(t.TempDir())
var last []byte
- if err := edgesFileWriteClaims(context.Background(), st, reg, edgesFile{Box: "annie", BoxID: "annie"}, "1.2.3.4", &last, false, claims); err != nil {
+ if err := edgesFileWriteClaims(context.Background(), st, reg, edgesFile{Box: "annie", BoxID: "annie"}, "1.2.3.4", &last, false, claims, nil); err != nil {
t.Fatal(err)
}
body, _, err := st.Get(context.Background(), edgesFileKey("annie"))
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index c51489b2f..28b314a8f 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -88,6 +88,12 @@ type edgeHost struct {
// that predates it. Two files claiming one host are settled by the
// higher stamp on the edge (homemap.go foldHomes).
Stamp int64 `json:"stamp,omitempty"`
+ // Assets is the host's asset-cache block (edge asset cache design §4.1):
+ // the path prefixes its serving builds declare, the instance's asset
+ // generation and the host's authority time. Published only by an armed
+ // box (CARLOS_ASSET_CACHE=1) for a host assetPolicy says catches.
+ // Additive (omitempty): a file without it parses byte-identically.
+ Assets *edgeAssets `json:"assets,omitempty"`
}
// hostOwner folds the owners of every route served under one hostname. Any
@@ -407,7 +413,7 @@ func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry,
// period stale. On every other pass a headroom change alone is not a
// reason to write.
func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool) error {
- return edgesFileWriteClaims(ctx, s, reg, ident, eip, last, refresh, nil)
+ return edgesFileWriteClaims(ctx, s, reg, ident, eip, last, refresh, nil, nil)
}
// errBoxAddressesUnknown fails an edges file pass whose address source could
@@ -417,8 +423,19 @@ var errBoxAddressesUnknown = errors.New("edges file: the box's addresses could n
// edgesFileWriteClaims is edgesFileWrite with the claim reader region HA
// adds (edgesclaims.go). claims is nil unless CARLOS_REGION_HA is on, and
// nil publishes exactly what edgesFileWrite always has.
-func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool, claims *edgesClaims) error {
+//
+// assets is the box's asset-cache policy, nil unless CARLOS_ASSET_CACHE=1;
+// nil publishes no assets blocks and reads no manifests (edge asset cache
+// design §4.1).
+func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool, claims *edgesClaims, assets *assetPolicy) error {
box := ident.Box
+ // The authority clocks are copied before any registry read (see
+ // assetClocks): the rows read below are then at least as new as the
+ // times published with them.
+ var clocks assetClocks
+ if assets != nil {
+ clocks = assets.auth.snapshot()
+ }
routes, err := reg.List()
if err != nil {
return fmt.Errorf("edges: list: %w", err)
@@ -632,6 +649,22 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
hosts[key] = edgeHost{Role: role, Wildcard: true, Alias: alias, Account: acct, App: app, Stamp: stamp}
}
+ // Asset-cache blocks (edge asset cache design §3.3, §4.1), last, over the
+ // finished map: a host the claims withheld, an alias not yet pointing and
+ // a catch-all key are not in it, or are refused by the rule. An alias
+ // takes its primary's block and at. The manifest cache keeps only the
+ // builds the local routes reference, pruned whatever this pass decides.
+ if assets != nil {
+ in := newAssetInputs(routes, hosts, paths, aliases, owners)
+ for host, h := range hosts {
+ if a, ok := assets.decide(ctx, in, clocks, host, false, false); ok {
+ h.Assets = &edgeAssets{Paths: a.Paths, Gen: a.Gen, At: a.At.Unix()}
+ hosts[host] = h
+ }
+ }
+ assets.manifests.prune(assets.keepBuilds(routes))
+ }
+
var private BoxAddresses
if claims != nil && claims.addrs != nil {
// Region HA spec 2 §6 (A1-12): publish only what the box's address
@@ -663,7 +696,22 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
if err != nil {
return fmt.Errorf("edges: marshal: %w", err)
}
- if bytes.Equal(gate, *last) && !(refresh && f.Headroom != nil) {
+ // An at that advanced by less than assetRewriteAfter is not, by itself,
+ // a reason to write (design §4.2: about three extra PUTs an hour). The
+ // comparison holds such an at at its last written value; a write carries
+ // the current one.
+ cmp := gate
+ if assets != nil {
+ held := gateCopy
+ held.Hosts = heldAssetAts(gateCopy.Hosts, *last)
+ if cmp, err = json.Marshal(held); err != nil {
+ return fmt.Errorf("edges: marshal: %w", err)
+ }
+ }
+ if bytes.Equal(cmp, *last) && !(refresh && f.Headroom != nil) {
+ if assets != nil {
+ assets.passDone(clocks, false, nil)
+ }
return nil
}
@@ -676,6 +724,9 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
return err
}
*last = gate
+ if assets != nil {
+ assets.passDone(clocks, true, f.Hosts)
+ }
return nil
}
@@ -799,7 +850,14 @@ func edgesFileSecretsFP(getenv func(string) string) string {
// Platform-box job, retentionLoop's rules: skipped on fleet boxes, disabled
// with a log line when there is no deployment bucket. Single-box deployments:
// the file is published and nothing reads it yet.
-func edgesFileLoop(reg *registry.Registry) {
+//
+// With CARLOS_ASSET_CACHE=1 the file carries each catching host's assets
+// block (edge asset cache design §4.1), and two more things make a pass due:
+// a change in any local route's serving build (sb, the activator's
+// ServingGen; a generation switch changes no registry row), and an assets
+// rewrite coming due (assetPolicy.rewriteDue: a published at would advance
+// by 20 minutes or more, or the authority clocks first became known).
+func edgesFileLoop(reg *registry.Registry, sb servingBuild) {
ctx := context.Background()
if shouldSkipPublisherElection(os.Getenv) {
log.Printf("agent: edges file loop is a platform-box job — skipped on a fleet box")
@@ -860,6 +918,13 @@ func edgesFileLoop(reg *registry.Registry) {
return hostclaim.Read(ctx, store, host)
})
}
+ // The asset cache's publication (design §4.1): nil, and so nothing at
+ // all, unless the box is armed. The manifests are read from this loop's
+ // store handle, the deployment bucket.
+ assets := armAssetPolicy(os.Getenv, reg, boxAssetAuthority, newAssetManifestCache(store), sb)
+ if assets != nil {
+ log.Printf("agent: edges file publishes asset-cache blocks (CARLOS_ASSET_CACHE=1)")
+ }
log.Printf("agent: edges file loop armed (5min + jitter, and %s after a registry change, %s)", edgesFileWatchInterval, edgesFileKey(box))
last := seedEdgesFileGate(ctx, store, box)
sched := newEdgesFileSchedule(time.Now(), func() time.Duration {
@@ -885,7 +950,11 @@ func edgesFileLoop(reg *registry.Registry) {
} else {
fpErrs = 0
}
- if !state.due(now, fp, err == nil) {
+ key := fp
+ if assets != nil && err == nil {
+ key = edgesChangeKey(fp, assets.servingDigest())
+ }
+ if !state.dueWith(now, key, err == nil, assets.rewriteDue()) {
continue
}
if state.timed() {
@@ -896,7 +965,7 @@ func edgesFileLoop(reg *registry.Registry) {
ident.Headroom = headroomFor(headroomCapabilities(ident.Capabilities), dataDir, reg)
ident = withMovesFields(ident, dataDir, optDir)
ident = withSeedsFields(ident, dataDir, optDir)
- err = edgesFileWriteClaims(ctx, store, reg, ident, eip, &last, state.timed(), claims)
+ err = edgesFileWriteClaims(ctx, store, reg, ident, eip, &last, state.timed(), claims, assets)
if err != nil {
// Throttled: a bucket that refuses every write (an IAM Deny, an
// outage) is retried on a backoff, and each retry must not be a