amadan

carlos / platform Public

edge asset cache task 8 fix round 1: no decision across a running pass

Records and alias passes now mark themselves running (passBegin and
passEnd move a sequence and an in-flight count). An asset decision
copies the clocks and the sequence, reads its registry tables, and
trusts them only if no pass was running at the copy and none started
or finished since. Otherwise the writer republishes the blocks this
process last wrote, with their old at, and retries on the next tick;
For, Publish and Declared answer false. Deterministic tests run a real
alias pass and a real records pass between a decision's reads.

The publication tests read the written object through
internal/edgesfile, which gains Host.Assets; tests that feed the clocks
or test one unit say "not acceptance".
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude cd20a0b2c9d7a26e95a71bcb57567385c325714e parent 68b93e5
6 files changed, +434 −44
  • internal/edgesfile/edgesfile.go +15 −0
  • internal/host/agent.go +7 −3
  • internal/host/assetauthority.go +45 −3
  • internal/host/assetpolicy.go +50 −15
  • internal/host/assetpublish_test.go +296 −21
  • internal/host/edgesfile.go +21 −2
diff --git a/internal/edgesfile/edgesfile.go b/internal/edgesfile/edgesfile.go
index 10346696d..fb2cf2cf6 100644
--- a/internal/edgesfile/edgesfile.go
+++ b/internal/edgesfile/edgesfile.go
@@ -62,6 +62,21 @@ type Host struct {
// #305): the key is "*.<parent>". With Role RoleHome the box also serves
// the catch-all; with any other role it is the cert half alone.
Wildcard bool `json:"wildcard"`
+ // Assets is the host's asset-cache block (edge asset cache design §4.1),
+ // in the shape internal/host's edgeAssets writes, pinned by
+ // testdata/assets-host.golden.json. Additive: nil on a file without it,
+ // which includes every file an unarmed box writes.
+ Assets *HostAssets `json:"assets,omitempty"`
+}
+
+// HostAssets is the decoded assets block: the path prefixes the host's
+// serving builds declare, the instance's asset generation, and the host's
+// authority time in Unix seconds. A reader serves nothing for a host whose
+// At is over an hour old (design §4.2).
+type HostAssets struct {
+ Paths []string `json:"paths"`
+ Gen string `json:"gen"`
+ At int64 `json:"at"`
}
// File is one box's published file.
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 161cb439f..e26574cbf 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -1032,12 +1032,15 @@ var aliasPassMu sync.Mutex
// 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
+// domain confirmed time when the report says complete. The pass is marked
+// running for its whole length (passBegin, passEnd), so an asset decision
+// that overlaps it withholds rather than mixing its writes. 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()
+ start := auth.passBegin()
+ defer auth.passEnd()
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)
@@ -1656,7 +1659,8 @@ func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry
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()
+ start := auth.passBegin()
+ defer auth.passEnd() // after the sidecar pass's writes too
// 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).
diff --git a/internal/host/assetauthority.go b/internal/host/assetauthority.go
index 1734217ea..af1ce6fcd 100644
--- a/internal/host/assetauthority.go
+++ b/internal/host/assetauthority.go
@@ -42,6 +42,16 @@ type assetAuthority struct {
// publishable after a restart is published within seconds, not at the
// five-minute timer.
known uint64
+ // seq moves at the start and at the end of every records and alias
+ // pass, and inflight counts the passes running now. A decision that
+ // reads several registry tables (routes, paths, aliases) is consistent
+ // only when no pass was running when it began and seq has not moved by
+ // the time it finished (settled): the passes write those tables one
+ // statement at a time, so a read that overlaps one can join a target
+ // from after it with a path set from before it, a combination neither
+ // the pass's start nor its end allowed.
+ seq uint64
+ inflight int
}
func newAssetAuthority(now func() time.Time) *assetAuthority {
@@ -56,8 +66,24 @@ func newAssetAuthority(now func() time.Time) *assetAuthority {
// 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() }
+// passBegin marks a records or alias pass as running and returns its start
+// time, read once before the pass reads anything. Every passBegin is paired
+// with a passEnd after the pass's last registry write and its report.
+func (a *assetAuthority) passBegin() time.Time {
+ a.mu.Lock()
+ a.seq++
+ a.inflight++
+ a.mu.Unlock()
+ return a.now()
+}
+
+// passEnd marks the pass passBegin started as finished.
+func (a *assetAuthority) passEnd() {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ a.seq++
+ a.inflight--
+}
// recordsPass records a reconcile pass that started at start. Each app the
// report names complete has its records confirmed time moved to start; every
@@ -126,17 +152,23 @@ func (a *assetAuthority) knownCount() uint64 {
// 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.
+//
+// The copy also carries the pass sequence it was taken at; settled says
+// whether the registry reads that followed it can be trusted together.
type assetClocks struct {
records map[string]time.Time
domain time.Time
domainOK bool
known uint64
+ seq uint64
+ quiet bool // no pass was running when the copy was taken
}
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}
+ c := assetClocks{records: make(map[string]time.Time, len(a.records)), domain: a.domain, domainOK: a.domainOK, known: a.known,
+ seq: a.seq, quiet: a.inflight == 0}
for k, v := range a.records {
c.records[k] = v
}
@@ -154,3 +186,13 @@ func (c assetClocks) hostAt(acct, app string) (time.Time, bool) {
}
return rec, true
}
+
+// settled reports whether no records or alias pass ran at any point between
+// the copy c and now: none was running when c was taken, and none has
+// started or finished since. Registry reads made in between then show one
+// state the passes left, not a mixture. Called after the last read.
+func (a *assetAuthority) settled(c assetClocks) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return c.quiet && a.seq == c.seq
+}
diff --git a/internal/host/assetpolicy.go b/internal/host/assetpolicy.go
index bfb99b5c6..19651224e 100644
--- a/internal/host/assetpolicy.go
+++ b/internal/host/assetpolicy.go
@@ -96,18 +96,27 @@ type assetPolicy struct {
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 is the block this process last wrote for each host, with the
+ // owner it was decided for: rewriteDue's input, and what a pass that
+ // overlapped a records or alias pass publishes instead of deciding.
written map[string]writtenAssets
- // knownAt is auth.known as of the last successful pass.
+ // knownAt is auth.known as of the last settled pass.
knownAt uint64
+ // torn is set when the last writer pass overlapped a records or alias
+ // pass and so decided nothing; the loop runs another pass on the next
+ // tick (rewriteDue).
+ torn bool
// lastDigest is the serving-build digest of the last read that worked.
lastDigest string
+
+ // betweenReads, when set, runs between a decision's path-rows read and
+ // its aliases read, the window a concurrent pass can tear. Tests only.
+ betweenReads func()
}
type writtenAssets struct {
account, app string
- at int64
+ block edgeAssets
}
var _ edge.LocalAssets = (*assetPolicy)(nil)
@@ -167,6 +176,9 @@ func (p *assetPolicy) readAssetInputs() (assetInputs, error) {
if err != nil {
return assetInputs{}, err
}
+ if p.betweenReads != nil {
+ p.betweenReads()
+ }
aliases, err := p.reg.ListAliases()
if err != nil {
return assetInputs{}, err
@@ -272,7 +284,7 @@ func (p *assetPolicy) Publish(host string) (edgeAssets, bool) {
}
clocks := p.auth.snapshot() // before the registry reads: see assetClocks
in, err := p.readAssetInputs()
- if err != nil {
+ if err != nil || !p.auth.settled(clocks) {
return edgeAssets{}, false
}
a, ok := p.decide(context.Background(), in, clocks, host, false, false)
@@ -284,14 +296,16 @@ func (p *assetPolicy) Publish(host string) (edgeAssets, bool) {
// 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.
+// assetFreshFor. It reads the registry and never the bucket. A read that
+// overlapped a records or alias pass answers false: the request is served
+// uncached rather than from a mixture of the pass's before and after.
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 {
+ if err != nil || !p.auth.settled(clocks) {
return edge.HostAssets{}, false
}
a, ok := p.decide(context.Background(), in, clocks, host, true, true)
@@ -308,8 +322,9 @@ func (p *assetPolicy) Declared(host, build, path string) bool {
if p == nil || build == "" {
return false
}
+ clocks := p.auth.snapshot()
in, err := p.readAssetInputs()
- if err != nil {
+ if err != nil || !p.auth.settled(clocks) {
return false
}
r, ok := in.primary(host)
@@ -359,28 +374,48 @@ func (p *assetPolicy) rewriteDue() bool {
if p == nil {
return false
}
- if p.auth.knownCount() != p.knownAt {
+ if p.torn || 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) {
+ if at, ok := p.auth.hostAt(w.account, w.app); ok && at.Unix()-w.block.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
+// withheld is what a writer pass that overlapped a records or alias pass
+// publishes for host: the block this process last wrote for it, byte for
+// byte with its old at, when the host's entry still names the owner it was
+// decided for. Nothing otherwise. The old at only ages; the next settled
+// pass decides afresh.
+func (p *assetPolicy) withheld(host string, h edgeHost) (*edgeAssets, bool) {
+ w, ok := p.written[host]
+ if !ok || w.account != h.Account || w.app != h.App {
+ return nil, false
+ }
+ b := w.block
+ b.Paths = slices.Clone(b.Paths)
+ return &b, true
+}
+
+// passDone records a successful writer pass: whether it was settled, the
+// clocks it used, and, when it wrote, what it published.
+func (p *assetPolicy) passDone(clocks assetClocks, settled, wrote bool, hosts map[string]edgeHost) {
+ p.torn = !settled
+ if settled {
+ 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}
+ b := *h.Assets
+ b.Paths = slices.Clone(b.Paths)
+ p.written[host] = writtenAssets{account: h.Account, app: h.App, block: b}
}
}
}
diff --git a/internal/host/assetpublish_test.go b/internal/host/assetpublish_test.go
index befc34a3e..d06440caf 100644
--- a/internal/host/assetpublish_test.go
+++ b/internal/host/assetpublish_test.go
@@ -225,26 +225,33 @@ func (w *assetWorld) raw() map[string]json.RawMessage {
}
// hosts runs one pass and reads the written hosts back.
-func (w *assetWorld) hosts() map[string]edgeHost {
+func (w *assetWorld) hosts() map[string]edgesfile.Host {
w.t.Helper()
w.pass()
return w.written()
}
-func (w *assetWorld) written() map[string]edgeHost {
+// written reads the written object back through the production reader,
+// internal/edgesfile, never the writer's own type.
+func (w *assetWorld) written() map[string]edgesfile.Host {
w.t.Helper()
- body, _, err := w.writes.Get(context.Background(), edgesFileKey("annie"))
+ return parseWrittenHosts(w.t, w.writes)
+}
+
+func parseWrittenHosts(t *testing.T, s release.Store) map[string]edgesfile.Host {
+ t.Helper()
+ body, _, err := s.Get(context.Background(), edgesfile.Key("annie"))
if err != nil {
- w.t.Fatal(err)
+ t.Fatal(err)
}
- var f edgesFile
- if err := json.Unmarshal(body, &f); err != nil {
- w.t.Fatal(err)
+ f, err := edgesfile.Parse(body)
+ if err != nil {
+ t.Fatalf("the written file does not parse: %v", err)
}
return f.Hosts
}
-func wantNoAssets(t *testing.T, hosts map[string]edgeHost, host string) {
+func wantNoAssets(t *testing.T, hosts map[string]edgesfile.Host, host string) {
t.Helper()
h, ok := hosts[host]
if !ok {
@@ -255,7 +262,7 @@ func wantNoAssets(t *testing.T, hosts map[string]edgeHost, host string) {
}
}
-func wantAssets(t *testing.T, hosts map[string]edgeHost, host string, paths []string, gen string, at time.Time) {
+func wantAssets(t *testing.T, hosts map[string]edgesfile.Host, host string, paths []string, gen string, at time.Time) {
t.Helper()
h, ok := hosts[host]
if !ok {
@@ -271,6 +278,7 @@ func wantAssets(t *testing.T, hosts map[string]edgeHost, host string, paths []st
// ---- the clocks -----------------------------------------------------------
+// Not acceptance: a unit test of the clocks, with no written object.
func TestAssetAuthorityHostAtIsTheOlderClock(t *testing.T) {
t.Parallel()
a := newAssetAuthority(nil)
@@ -286,6 +294,7 @@ func TestAssetAuthorityHostAtIsTheOlderClock(t *testing.T) {
}
}
+// Not acceptance: a unit test of the clocks, with no written object.
func TestAssetAuthorityIncompletePassDoesNotAdvance(t *testing.T) {
t.Parallel()
a := newAssetAuthority(nil)
@@ -312,6 +321,7 @@ func TestAssetAuthorityIncompletePassDoesNotAdvance(t *testing.T) {
}
}
+// Not acceptance: a unit test of the clocks, with no written object.
func TestAssetAuthorityUnknownUntilBothClocks(t *testing.T) {
t.Parallel()
a := newAssetAuthority(nil)
@@ -340,6 +350,9 @@ func TestAssetAuthorityUnknownUntilBothClocks(t *testing.T) {
// 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.
+//
+// Not acceptance: a unit test of the clocks over the real pass wiring, with
+// no written object.
func TestAssetAuthorityRecordsTheStartTime(t *testing.T) {
t.Parallel()
b := newAuthorityBox(t)
@@ -378,6 +391,7 @@ func TestAssetAuthorityRecordsTheStartTime(t *testing.T) {
// ---- the manifest cache ---------------------------------------------------
+// Not acceptance: a unit test of the manifest cache, with no written object.
func TestAssetManifestCacheReadsTheBucketOnce(t *testing.T) {
t.Parallel()
s := newAssetFaultStore(release.NewDirStore(t.TempDir()))
@@ -404,6 +418,7 @@ func TestAssetManifestCacheReadsTheBucketOnce(t *testing.T) {
}
}
+// Not acceptance: a unit test of the manifest cache, with no written object.
func TestAssetManifestCacheDoesNotCacheAFailure(t *testing.T) {
t.Parallel()
s := newAssetFaultStore(release.NewDirStore(t.TempDir()))
@@ -432,6 +447,9 @@ func TestAssetManifestCacheDoesNotCacheAFailure(t *testing.T) {
// 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.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestAssetManifestCacheStaysBoundedAcrossDeploys(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -471,10 +489,13 @@ func TestAssetManifestCacheStaysBoundedAcrossDeploys(t *testing.T) {
// ---- publication ----------------------------------------------------------
-// TestPublishesAssetsForAnArmedHome is the acceptance case: an armed home's
+// TestPublishesAssetsForAnArmedHome: 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.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestPublishesAssetsForAnArmedHome(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -508,6 +529,8 @@ func TestPublishesAssetsForAnArmedHome(t *testing.T) {
}
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
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} {
@@ -540,6 +563,9 @@ func TestUnarmedBoxPublishesNoAssets(t *testing.T) {
}
// Each §3.3 exclusion, alone, on a host that otherwise publishes.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestPublishesNothingUnderAnExclusion(t *testing.T) {
t.Parallel()
cases := []struct {
@@ -636,6 +662,9 @@ func TestPublishesNothingUnderAnExclusion(t *testing.T) {
// 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.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestPooledRouteCatchesLocallyOnly(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -649,6 +678,8 @@ func TestPooledRouteCatchesLocallyOnly(t *testing.T) {
}
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestPublishesTheUnionOfInstalledAndServingBuilds(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -688,6 +719,9 @@ func TestPublishesTheUnionOfInstalledAndServingBuilds(t *testing.T) {
// A sleeping instance has no serving generation; it keeps the installed
// build's paths, the build a wake would start.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestSleepingInstanceKeepsTheInstalledBuildsPaths(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -697,6 +731,8 @@ func TestSleepingInstanceKeepsTheInstalledBuildsPaths(t *testing.T) {
wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestAliasPublishesItsPrimarysBlock(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -734,6 +770,8 @@ func TestAliasPublishesItsPrimarysBlock(t *testing.T) {
}
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestCatchallAliasNeverPublishes(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -764,6 +802,9 @@ func TestCatchallAliasNeverPublishes(t *testing.T) {
// Review Focus 4: the console box homes the console's own route beside
// instances; only the instance route publishes.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestConsoleClassRouteDoesNotPublish(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -783,6 +824,8 @@ func TestConsoleClassRouteDoesNotPublish(t *testing.T) {
}
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestRewritesWhenAtAdvancesTwentyMinutes(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -811,6 +854,8 @@ func TestRewritesWhenAtAdvancesTwentyMinutes(t *testing.T) {
}
}
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestDoesNotRewriteForASmallerAdvance(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -846,6 +891,9 @@ func TestDoesNotRewriteForASmallerAdvance(t *testing.T) {
// A host that becomes publishable when the clocks first become known (after
// a restart) does not wait for the timer.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestRewriteDueWhenTheAuthorityFirstBecomesKnown(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -868,6 +916,9 @@ func TestRewriteDueWhenTheAuthorityFirstBecomesKnown(t *testing.T) {
// 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.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestServingBuildSwitchRepublishesWithAnUnchangedFingerprint(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -930,6 +981,9 @@ func TestServingBuildSwitchRepublishesWithAnUnchangedFingerprint(t *testing.T) {
// 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.
+//
+// Not acceptance: the clocks are fed reports directly (the authority table
+// runs the real passes); the writer and the reader are the production ones.
func TestFailedRecordReadsStopAtFromAdvancing(t *testing.T) {
t.Parallel()
w := newAssetWorld(t)
@@ -1114,19 +1168,13 @@ func (b *authorityBox) run(d time.Duration) {
}
}
-func (b *authorityBox) published(host string) (edgeAssets, bool) {
+// published is host's block in the written object, read through the
+// production reader (internal/edgesfile).
+func (b *authorityBox) published(host string) (edgesfile.HostAssets, 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]
+ h, ok := parseWrittenHosts(b.t, b.base)[host]
if !ok || h.Assets == nil {
- return edgeAssets{}, false
+ return edgesfile.HostAssets{}, false
}
return *h.Assets, true
}
@@ -1325,5 +1373,232 @@ func TestPathRowAddedUnderAPrimaryStopsIt(t *testing.T) {
}
}
+// ---- overlapping passes ----------------------------------------------------
+
+// tearBox is an authority box with an alias that can move between two
+// primaries, each of which gets path rows when the alias points at it:
+// www.example.com -> blog.example.com with blog's path rows, or
+// www.example.com -> shop.example.com with shop's. Both settled states
+// exclude the alias; only a read joining one state's alias with the other's
+// path set lets it catch.
+func newTearBox(t *testing.T) *authorityBox {
+ t.Helper()
+ b := newAuthorityBox(t)
+ b.route("blog.example.com", func(r *registry.Route) { r.AssetGen = assetGen2; r.Version = "v1" })
+ b.route("api.example.com")
+ b.tearTo("blog.example.com")
+ b.cycle()
+ if err := b.reg.SetAliasDNSStatus("www.example.com", "pointing", assetT0.Unix()); err != nil {
+ t.Fatal(err)
+ }
+ b.cycle()
+ return b
+}
+
+// tearTo rewrites the claims so the alias and the path rows sit on primary
+// (no pass is run).
+func (b *authorityBox) tearTo(primary string) {
+ b.t.Helper()
+ for _, h := range []string{"www.example.com", "blog.example.com", "shop.example.com"} {
+ if _, _, err := b.base.Get(context.Background(), domains.ClaimKey(h)); err == nil {
+ b.unclaim(h)
+ }
+ }
+ b.claim(domains.Claim{Host: "www.example.com", App: "shop", Route: primary})
+ b.claim(domains.Claim{Host: primary, App: "shop", Route: primary, Paths: map[string]string{"api": "api.example.com"}})
+}
+
+// once runs fn the first time the policy's between-reads hook fires.
+func (b *authorityBox) once(fn func()) {
+ var done bool
+ b.policy.betweenReads = func() {
+ if !done {
+ done = true
+ fn()
+ }
+ }
+}
+
+func (b *authorityBox) write() {
+ b.t.Helper()
+ if err := edgesFileWriteClaims(context.Background(), b.base, b.reg, edgesFile{Box: "annie"}, "1.2.3.4", &b.last, false, nil, b.policy); err != nil {
+ b.t.Fatal(err)
+ }
+}
+
+// TestAssetDecisionOverlappingAnAliasPassWithholds: a complete alias pass
+// runs between a decision's path-rows read and its aliases read, moving the
+// alias from blog (with path rows) to shop (with path rows). Reading the old
+// path set with the new alias would let the alias catch with shop's block.
+// The writer instead republishes the blocks it last wrote, with their old
+// at, and retries; For and Publish answer false. Run with -race too.
+func TestAssetDecisionOverlappingAnAliasPassWithholds(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ b := newTearBox(t)
+ if _, ok := b.published("www.example.com"); ok {
+ t.Fatal("the alias of a primary with path rows published")
+ }
+ before, ok := b.published("shop.example.com")
+ if !ok {
+ t.Fatal("shop.example.com did not publish before the tear")
+ }
+
+ b.tearTo("shop.example.com")
+ b.clock.advance(25 * time.Minute)
+ b.auth.recordsPass(b.clock.now(), provision.ReconcileReport{Complete: map[string]bool{b.acct + "/shop": true}})
+ b.once(func() {
+ if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
+ t.Error(err)
+ }
+ })
+ b.write()
+ if got, ok := b.published("www.example.com"); ok {
+ t.Fatalf("the torn read published the alias with %+v", got)
+ }
+ got, ok := b.published("shop.example.com")
+ if !ok || got.At != before.At || got.Gen != before.Gen || !slices.Equal(got.Paths, before.Paths) {
+ t.Fatalf("the torn pass published shop as %+v %v, want its last block %+v unchanged", got, ok, before)
+ }
+ if !b.policy.rewriteDue() {
+ t.Fatal("a torn pass did not ask for another")
+ }
+ // The next pass is settled and decides the state the alias pass left:
+ // shop now has path rows, so neither name catches.
+ b.policy.betweenReads = nil
+ b.write()
+ if _, ok := b.published("shop.example.com"); ok {
+ t.Fatal("shop published with path rows under it")
+ }
+ if _, ok := b.published("www.example.com"); ok {
+ t.Fatal("the alias published with path rows under its primary")
+ }
+ if b.policy.rewriteDue() {
+ t.Fatal("still due after a settled pass")
+ }
+
+ // For and Publish: the same tear in the other direction, back to blog.
+ // The old path set (shop's) with the new alias (to blog) would catch.
+ b.tearTo("blog.example.com")
+ b.once(func() {
+ if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
+ t.Error(err)
+ }
+ })
+ if got, ok := b.policy.For("www.example.com"); ok {
+ t.Fatalf("For answered %+v from a torn read", got)
+ }
+ aliases, err := b.reg.ListAliases()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !slices.ContainsFunc(aliases, func(a registry.RouteAlias) bool { return a.Alias == "www.example.com" && a.Host == "blog.example.com" }) {
+ t.Fatalf("the hook's pass did not move the alias: %+v", aliases)
+ }
+ b.tearTo("shop.example.com")
+ b.once(func() {
+ if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
+ t.Error(err)
+ }
+ })
+ if got, ok := b.policy.Publish("www.example.com"); ok {
+ t.Fatalf("Publish answered %+v from a torn read", got)
+ }
+ // Settled again, the controls: blog has no path rows now, so blog
+ // catches, and the alias (on shop, with path rows) does not.
+ b.policy.betweenReads = nil
+ if _, ok := b.policy.For("blog.example.com"); !ok {
+ t.Fatal("control: For refused blog on a settled read")
+ }
+ if _, ok := b.policy.For("www.example.com"); ok {
+ t.Fatal("For answered for the alias of a primary with path rows")
+ }
+}
+
+// TestAssetDecisionOverlappingARecordsPassWithholds: the records pass (route
+// rows, gates, generations) tears the same way and is guarded the same way.
+func TestAssetDecisionOverlappingARecordsPassWithholds(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ b := newAuthorityBox(t)
+ b.cycle()
+ before, ok := b.published("shop.example.com")
+ if !ok {
+ t.Fatal("no block before the overlap")
+ }
+ b.clock.advance(25 * time.Minute)
+ b.auth.aliasPass(b.clock.now(), domains.AliasReport{Complete: true})
+ b.once(func() {
+ if err := reconcileSerializedAuth(ctx, b.recs, b.reg, b.reconcileConfig(), nil, b.auth); err != nil {
+ t.Error(err)
+ }
+ })
+ b.write()
+ if got, ok := b.published("shop.example.com"); !ok || got.At != before.At {
+ t.Fatalf("an overlapped pass published %+v %v, want the last block at %d", got, ok, before.At)
+ }
+ b.once(func() {
+ if err := reconcileSerializedAuth(ctx, b.recs, b.reg, b.reconcileConfig(), nil, b.auth); err != nil {
+ t.Error(err)
+ }
+ })
+ if _, ok := b.policy.For("shop.example.com"); ok {
+ t.Fatal("For answered across a records pass")
+ }
+ b.policy.betweenReads = nil
+ b.write()
+ if got, ok := b.published("shop.example.com"); !ok || got.At != b.clock.now().Unix() {
+ t.Fatalf("the settled pass published %+v %v, want at %d", got, ok, b.clock.now().Unix())
+ }
+ if _, ok := b.policy.For("shop.example.com"); !ok {
+ t.Fatal("For refused on a settled read")
+ }
+}
+
+// TestAssetDecisionWithholdsWhileAPassRuns: a pass running across the whole
+// decision moves no sequence number between its reads, and is caught by the
+// in-flight count instead.
+//
+// Not acceptance: the clocks are fed reports directly; the writer and the
+// reader are the production ones.
+func TestAssetDecisionWithholdsWhileAPassRuns(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ w.clock.advance(25 * time.Minute)
+ w.confirm()
+ w.auth.passBegin()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
+ if _, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatal("For answered while a pass ran")
+ }
+ if _, ok := w.policy.Publish("shop.example.com"); ok {
+ t.Fatal("Publish answered while a pass ran")
+ }
+ if w.policy.Declared("shop.example.com", "v1", "/assets/a.js") {
+ t.Fatal("Declared answered while a pass ran")
+ }
+ if !w.policy.rewriteDue() {
+ t.Fatal("no retry asked for")
+ }
+ w.auth.passEnd()
+ wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, w.clock.now())
+ if _, ok := w.policy.For("shop.example.com"); !ok {
+ t.Fatal("For refused once the pass ended")
+ }
+ // A restart's first pass has written nothing: an overlapped pass then
+ // publishes no block at all, never one from the file it found.
+ w2 := newAssetWorld(t)
+ w2.ship("v1", "/assets/")
+ w2.instance("shop.example.com", assetGen1, "v1")
+ w2.last = w.last // the previous process's file, as seedEdgesFileGate reads it
+ w2.confirm()
+ w2.auth.passBegin()
+ wantNoAssets(t, w2.hosts(), "shop.example.com")
+}
+
// Compile-time: the policy is what plan 3's edge consumes.
var _ edge.LocalAssets = (*assetPolicy)(nil)
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index 28b314a8f..85c185371 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -499,6 +499,9 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
if err != nil {
return fmt.Errorf("edges: list paths: %w", err)
}
+ if assets != nil && assets.betweenReads != nil {
+ assets.betweenReads()
+ }
owners := servedOwners(routes, hosts, paths)
for host, h := range hosts {
h.Account, h.App = owners[host].owner()
@@ -654,9 +657,25 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
// 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.
+ //
+ // The rows above came from separate reads. If a records or alias pass
+ // was running, or started or finished, between the clocks copy and the
+ // last read (settled), they may join one table's state from before the
+ // pass with another's from after it, which neither allowed; such a pass
+ // decides nothing and republishes the blocks this process last wrote,
+ // with their old at, and the loop tries again on the next tick.
+ settled := false
if assets != nil {
+ settled = assets.auth.settled(clocks)
in := newAssetInputs(routes, hosts, paths, aliases, owners)
for host, h := range hosts {
+ if !settled {
+ if b, ok := assets.withheld(host, h); ok {
+ h.Assets = b
+ hosts[host] = h
+ }
+ continue
+ }
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
@@ -710,7 +729,7 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
}
if bytes.Equal(cmp, *last) && !(refresh && f.Headroom != nil) {
if assets != nil {
- assets.passDone(clocks, false, nil)
+ assets.passDone(clocks, settled, false, nil)
}
return nil
}
@@ -725,7 +744,7 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
}
*last = gate
if assets != nil {
- assets.passDone(clocks, true, f.Hosts)
+ assets.passDone(clocks, settled, true, f.Hosts)
}
return nil
}