amadan

carlos / platform Public

hotfix #281: the edge no longer waits on the startup sweep and recovery

v0.27.0 failed the 20 s update gate on a box with 142 instances: the
updater restarts instance control before the edge, and the agent's
startup sweep then waited about 22 s on a cold instance control before
the edge listened. The edge's own startup is unchanged; the activator
start (sweep, layout recovery and hold restoration, Start, instance-key
pass) now runs in the background, and instance admission (wakes, workers,
sidecars, scheduled duties) waits on a startup latch for up to 30 s, then
answers with the wake-timeout 503. A request near startup whose upstream
recovery moved is answered the same way, never proxied to the old socket.
Shutdown order unchanged. Census: plan14-root-reads-hotfix281.tsv.

Prompts:

  <task-notification>
  <task-id>bgtu3i3pb</task-id>
  <tool-use-id>toolu_016Dd95E2RQQZ21WTbkSpsE4</tool-use-id>
  <output-file>/home/paulca/.cache/tmp/claude-1001/-home-paulca-amadan-net-carlos-platform/6387394d-0b76-4de7-bbb4-d725fe24b08c/tasks/bgtu3i3pb.output</output-file>
  <status>completed</status>
  <summary>Background command "Wait for hotfix fix round 1" completed (exit code 0)</summary>
  </task-notification>

Squashed from hotfix-startup-sweep-281 into main.
Paul Campbell pushed by claude 6dfbff103985e1e81899f2e611b1afff0ddb3769 parent f4cf27d
11 files changed, +965 −225
  • internal/activator/activator.go +132 −35
  • internal/activator/startup_test.go +159 −0
  • internal/edge/startup_test.go +86 −0
  • internal/host/agent.go +189 −134
  • internal/host/instancekeys.go +3 −0
  • internal/host/instancekeys_test.go +26 −12
  • internal/host/layoutleg.go +45 −44
  • internal/host/startup.go +15 −0
  • internal/host/startup_test.go +285 −0
  • internal/provision/plan14_root_reads_test.go +7 −0
  • internal/provision/testdata/plan14-root-reads-hotfix281.tsv +18 −0

CI passed — run details

CI log 
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.011s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... -count=1 -timeout 30m
ok  	github.com/carlosframework/platform	346.072s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	2.906s
ok  	github.com/carlosframework/platform/cmd/moveprobe	92.176s
ok  	github.com/carlosframework/platform/cmd/zddprobe	0.381s
ok  	github.com/carlosframework/platform/internal/activator	227.844s
ok  	github.com/carlosframework/platform/internal/agentsock	0.017s
ok  	github.com/carlosframework/platform/internal/alerting	0.003s
ok  	github.com/carlosframework/platform/internal/analytics	0.025s
ok  	github.com/carlosframework/platform/internal/appconfig	2.350s
ok  	github.com/carlosframework/platform/internal/apperr	0.090s
ok  	github.com/carlosframework/platform/internal/applog	0.136s
ok  	github.com/carlosframework/platform/internal/awsinv	0.003s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.004s
ok  	github.com/carlosframework/platform/internal/boxid	0.893s
?   	github.com/carlosframework/platform/internal/boxid/boxidtest	[no test files]
?   	github.com/carlosframework/platform/internal/boxid/internal/signtest	[no test files]
ok  	github.com/carlosframework/platform/internal/boxvolumes	0.008s
ok  	github.com/carlosframework/platform/internal/caps	0.014s
ok  	github.com/carlosframework/platform/internal/certwarm	0.002s
ok  	github.com/carlosframework/platform/internal/cgroup	0.692s
ok  	github.com/carlosframework/platform/internal/changelog	0.300s
ok  	github.com/carlosframework/platform/internal/clamd	0.067s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	362.980s
?   	github.com/carlosframework/platform/internal/controlname	[no test files]
?   	github.com/carlosframework/platform/internal/controlsock	[no test files]
ok  	github.com/carlosframework/platform/internal/dbdoor	6.468s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.009s
ok  	github.com/carlosframework/platform/internal/dbverify	0.004s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.002s
ok  	github.com/carlosframework/platform/internal/docsite	0.005s
ok  	github.com/carlosframework/platform/internal/domains	2.109s
ok  	github.com/carlosframework/platform/internal/doorkey	0.006s
ok  	github.com/carlosframework/platform/internal/economics	0.003s
ok  	github.com/carlosframework/platform/internal/edge	25.928s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.007s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.117s
ok  	github.com/carlosframework/platform/internal/gate	1.655s
ok  	github.com/carlosframework/platform/internal/genselect	0.015s
ok  	github.com/carlosframework/platform/internal/geo	0.015s
ok  	github.com/carlosframework/platform/internal/geoip	0.029s
ok  	github.com/carlosframework/platform/internal/host	94.357s
ok  	github.com/carlosframework/platform/internal/hostactors	0.345s
ok  	github.com/carlosframework/platform/internal/hostbox	2.233s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.491s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.006s
ok  	github.com/carlosframework/platform/internal/hostexec	4.835s
ok  	github.com/carlosframework/platform/internal/hostfence	0.036s
ok  	github.com/carlosframework/platform/internal/hostfiles	10.564s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.012s
ok  	github.com/carlosframework/platform/internal/hostguard	1.004s
ok  	github.com/carlosframework/platform/internal/hostops	1.421s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.002s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.002s
ok  	github.com/carlosframework/platform/internal/hostrelocate	0.429s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.139s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.002s
ok  	github.com/carlosframework/platform/internal/infraview	0.002s
ok  	github.com/carlosframework/platform/internal/instancectl	1.814s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.716s
ok  	github.com/carlosframework/platform/internal/instancefs	0.005s
ok  	github.com/carlosframework/platform/internal/instancekey	0.002s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.012s
ok  	github.com/carlosframework/platform/internal/layout	1.424s
ok  	github.com/carlosframework/platform/internal/lease	9.450s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.008s
ok  	github.com/carlosframework/platform/internal/ledger	0.014s
ok  	github.com/carlosframework/platform/internal/mail	0.150s
ok  	github.com/carlosframework/platform/internal/maintenance	0.095s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.526s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.085s
ok  	github.com/carlosframework/platform/internal/metrics	0.007s
ok  	github.com/carlosframework/platform/internal/moves	20.937s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.731s
ok  	github.com/carlosframework/platform/internal/objstore	0.171s
ok  	github.com/carlosframework/platform/internal/ops	0.163s
ok  	github.com/carlosframework/platform/internal/park	0.205s
ok  	github.com/carlosframework/platform/internal/privatefile	0.003s
ok  	github.com/carlosframework/platform/internal/provision	12.928s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.007s
?   	github.com/carlosframework/platform/internal/regionname	[no test files]
ok  	github.com/carlosframework/platform/internal/registry	4.079s
ok  	github.com/carlosframework/platform/internal/relay	7.232s
ok  	github.com/carlosframework/platform/internal/release	19.836s
ok  	github.com/carlosframework/platform/internal/rollout	3.775s
ok  	github.com/carlosframework/platform/internal/rolloutwords	0.003s
ok  	github.com/carlosframework/platform/internal/rootaws	0.007s
ok  	github.com/carlosframework/platform/internal/rootparticipation	0.007s
ok  	github.com/carlosframework/platform/internal/rootstate	0.025s
ok  	github.com/carlosframework/platform/internal/s3compat	0.004s
ok  	github.com/carlosframework/platform/internal/safefile	0.004s
ok  	github.com/carlosframework/platform/internal/schedule	2.284s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	9.324s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	4.143s
ok  	github.com/carlosframework/platform/internal/status	0.040s
ok  	github.com/carlosframework/platform/internal/steering	0.006s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.167s
ok  	github.com/carlosframework/platform/internal/svcrole	0.002s
ok  	github.com/carlosframework/platform/internal/telemetry	0.004s
ok  	github.com/carlosframework/platform/internal/termfence	0.004s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.004s
ok  	github.com/carlosframework/platform/internal/unixdial	0.002s
ok  	github.com/carlosframework/platform/internal/updatewords	0.004s
ok  	github.com/carlosframework/platform/internal/usage	0.004s
ok  	github.com/carlosframework/platform/internal/version	0.002s
ok  	github.com/carlosframework/platform/internal/vet	1.550s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
  minio: downloading https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio.linux-arm64.RELEASE.2025-09-07T16-13-09Z
  minio: downloaded, digest verified
  mc: downloading https://github.com/minio/mc/releases/download/RELEASE.2025-08-13T08-35-41Z/mc.linux-arm64.RELEASE.2025-08-13T08-35-41Z
  mc: downloaded, digest verified
  minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
  buckets and scoped user ready
=== RUN   TestDeliveredContractMinio
=== RUN   TestDeliveredContractMinio/put
=== RUN   TestDeliveredContractMinio/get_round-trips_the_body
=== RUN   TestDeliveredContractMinio/list_shows_the_key
=== RUN   TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN   TestDeliveredContractMinio/delete_removes_it
=== RUN   TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
    --- PASS: TestDeliveredContractMinio/put (0.00s)
    --- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
    --- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
    --- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
    --- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
    --- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok  	github.com/carlosframework/platform/internal/objstore	0.021s
=== RUN   TestT4StorageProbeMinio
2026/10/13 22:28:12 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE3749E89E9E15, HostID: 7c420053e0fbcd7f41653fdd92ffa2802c903d05303b1aa15adc2e8bb65d183f, api error SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your key and signing method.
--- PASS: TestT4StorageProbeMinio (0.01s)
PASS
ok  	github.com/carlosframework/platform/internal/host	0.021s
minio contract: PASS

landed from hotfix-startup-sweep-281

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index e5392cabe..a3d9866a0 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -67,6 +67,7 @@ import (
"strconv"
"strings"
"sync"
+ "sync/atomic"
"time"
"github.com/carlosframework/platform/internal/cgroup"
@@ -1007,6 +1008,7 @@ type configAdmitter interface {
const configRefusedRetryAfter = 30
var errWakeTimeout = errors.New("timed out waiting for instance wake")
+var errStartupStopped = errors.New("activator stopped during startup")
// Service is the activator (Eleven's activatorSvc). It structurally satisfies
// the edge package's Activator interface — the edge never imports this
@@ -1014,6 +1016,13 @@ var errWakeTimeout = errors.New("timed out waiting for instance wake")
// edge.Cache and edge.ListenControl; serve calls edge.MarkAnswered), and the
// seam back is the interface the edge declares for itself.
type Service struct {
+ // Startup gates admission while the agent recovers installations. Set before publication.
+ Startup <-chan struct{}
+ // startupSeenAt is when serve first found Startup closed (unix nanos,
+ // 0 until then): requests in the window after it may carry a proxy the
+ // edge chose before recovery moved their upstream.
+ startupSeenAt atomic.Int64
+
cfg Config
backend Backend
due DueStore
@@ -1272,6 +1281,46 @@ func New(cfg Config, backend Backend, due DueStore) *Service {
return a
}
+// startupFreshWindow bounds how long after startup completes a request's
+// route is re-checked against the registry (serve). A proxy the edge chose
+// before Startup closed reaches serve within milliseconds; the window is
+// generous, and costs one registry read per hibernating request inside it.
+const startupFreshWindow = 2 * time.Minute
+
+// nearStartup reports whether this request is within startupFreshWindow of
+// the first time serve found Startup closed. The first observation is at or
+// after the real close, so a request racing the close is always inside it.
+func (a *Service) nearStartup() bool {
+ now := time.Now().UnixNano()
+ at := a.startupSeenAt.Load()
+ if at == 0 {
+ if a.startupSeenAt.CompareAndSwap(0, now) {
+ return true
+ }
+ at = a.startupSeenAt.Load()
+ }
+ return now-at < int64(startupFreshWindow)
+}
+
+func (a *Service) awaitStartup(ctx context.Context, deadline <-chan time.Time) error {
+ if a.Startup == nil {
+ return nil
+ }
+ select {
+ case <-a.Startup:
+ if a.lifeCtx.Err() != nil {
+ return errStartupStopped
+ }
+ return ctx.Err()
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-a.lifeCtx.Done():
+ return errStartupStopped
+ case <-deadline:
+ return errWakeTimeout
+ }
+}
+
// begin registers a request against rt's instance, waking it if asleep.
//
// stamp is what separates a visitor from the platform's own clock. Both take
@@ -1284,6 +1333,9 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel
deadline := time.NewTimer(wakeWaitBound)
defer deadline.Stop()
arrival := time.Now()
+ if err := a.awaitStartup(ctx, deadline.C); err != nil {
+ return nil, err
+ }
extended := false
for {
a.mu.Lock()
@@ -1424,6 +1476,9 @@ var errSidecarAsleep = errors.New("instance asleep; sidecar requests do not wake
func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) error {
deadline := time.NewTimer(wakeWaitBound)
defer deadline.Stop()
+ if err := a.awaitStartup(ctx, deadline.C); err != nil {
+ return err
+ }
for {
a.mu.Lock()
if a.fences[rt.Host] != 0 {
@@ -2922,6 +2977,9 @@ func (a *Service) awaitWorkerSettle(ctx context.Context, e *actEntry, proc Proc)
func (a *Service) WakeWorker(rt registry.Route) error {
deadline := time.NewTimer(wakeWaitBound)
defer deadline.Stop()
+ if err := a.awaitStartup(a.lifeCtx, deadline.C); err != nil {
+ return err
+ }
arrival := time.Now()
extended := false
for {
@@ -4117,6 +4175,11 @@ func (a *Service) Start() {
// marks the publication dirty for the sweep to retry; it never fails or
// blocks Start.
_ = a.persistGenerations()
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if a.shuttingDown {
+ return
+ }
a.sweepStop = make(chan struct{})
a.sweepDone = make(chan struct{})
// Sweep well inside the idle window so a sleep happens promptly after it
@@ -4130,6 +4193,13 @@ func (a *Service) Start() {
}
go func() {
defer close(a.sweepDone)
+ if a.Startup != nil {
+ select {
+ case <-a.Startup:
+ case <-a.sweepStop:
+ return
+ }
+ }
t := time.NewTicker(interval)
defer t.Stop()
for {
@@ -4783,6 +4853,7 @@ func (a *Service) Shutdown(ctx context.Context) {
// below cover the rest.)
a.mu.Lock()
a.shuttingDown = true
+ sweepStop, sweepDone := a.sweepStop, a.sweepDone
a.mu.Unlock()
// …and cancel the service lifetime, which TEARS DOWN any worker spawn
// already past that check rather than letting the stop below wait it out.
@@ -4797,9 +4868,9 @@ func (a *Service) Shutdown(ctx context.Context) {
if a.lifeStop != nil {
a.lifeStop()
}
- if a.sweepStop != nil {
- close(a.sweepStop)
- <-a.sweepDone
+ if sweepStop != nil {
+ close(sweepStop)
+ <-sweepDone
}
a.mu.Lock()
// The sweeper is stopped as of the line above, so nothing else will ever
@@ -4937,6 +5008,28 @@ func (a *Service) ServeScheduled(w http.ResponseWriter, r *http.Request, rt regi
}
func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler, stamp bool) bool {
+ if rt.Kind == "instance" && rt.Hibernate && a.Startup != nil {
+ deadline := time.NewTimer(wakeWaitBound)
+ err := a.awaitStartup(r.Context(), deadline.C)
+ deadline.Stop()
+ if err != nil {
+ return a.serveWakeError(w, r, rt, err)
+ }
+ // The edge may have chosen rt and its proxy before startup recovery
+ // ran, whether this request waited here or arrived just as Startup
+ // closed. If recovery moved this host's upstream (a layout journal
+ // finished), that proxy points at the old socket: answer as a wake
+ // still in progress, and the retry is routed afresh. Only requests
+ // near startup are checked; after the window the edge's own
+ // selection is current.
+ if a.nearStartup() && a.Routes != nil {
+ if fresh, ferr := a.Routes.Get(rt.Host); ferr == nil && fresh != nil &&
+ (fresh.Socket != rt.Socket || fresh.Addr != rt.Addr) {
+ return a.serveWakeError(w, r, rt, errWakeTimeout)
+ }
+ }
+ }
+
a.mu.Lock()
fenced := a.fences[rt.Host] != 0
a.mu.Unlock()
@@ -4995,40 +5088,44 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout
release, err := a.begin(r.Context(), rt, stamp)
if err != nil {
- var held *errLeaseHeld
- var refused *errConfigRefused
- switch {
- case errors.Is(err, ErrFenced):
- maintenanceResponse(w, r)
- case errors.As(err, &refused):
- // Quiet by design: the config guard logs once per host and
- // publishes the refusal in the box's status snapshot. A line per
- // request here would bury that one.
- w.Header().Set("Retry-After", strconv.Itoa(configRefusedRetryAfter))
- edge.MarkAnswered(r)
- http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
- case errors.As(err, &held):
- retry := int(held.retryAfter/time.Second) + 1
- w.Header().Set("Retry-After", strconv.Itoa(retry))
- edge.MarkAnswered(r)
- http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
- case errors.Is(err, errWakeTimeout):
- w.Header().Set("Retry-After", "5")
- edge.MarkAnswered(r)
- http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
- case errors.Is(err, context.Canceled):
- // The client hung up while queued; nothing to write.
- default:
- // The wake failure is already the member's platform log error
- // (this log line, shipped by logsink.go); marking it keeps the
- // edge from filing the same failure a second time.
- log.Printf("activator: wake %s: %v", rt.Host, err)
- edge.MarkAnswered(r)
- http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
- }
- return true
+ return a.serveWakeError(w, r, rt, err)
}
defer release()
proxy.ServeHTTP(w, r)
return true
}
+
+func (a *Service) serveWakeError(w http.ResponseWriter, r *http.Request, rt registry.Route, err error) bool {
+ var held *errLeaseHeld
+ var refused *errConfigRefused
+ switch {
+ case errors.Is(err, ErrFenced):
+ maintenanceResponse(w, r)
+ case errors.As(err, &refused):
+ // Quiet by design: the config guard logs once per host and
+ // publishes the refusal in the box's status snapshot. A line per
+ // request here would bury that one.
+ w.Header().Set("Retry-After", strconv.Itoa(configRefusedRetryAfter))
+ edge.MarkAnswered(r)
+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ case errors.As(err, &held):
+ retry := int(held.retryAfter/time.Second) + 1
+ w.Header().Set("Retry-After", strconv.Itoa(retry))
+ edge.MarkAnswered(r)
+ http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
+ case errors.Is(err, errWakeTimeout):
+ w.Header().Set("Retry-After", "5")
+ edge.MarkAnswered(r)
+ http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
+ case errors.Is(err, context.Canceled):
+ // The client hung up while queued; nothing to write.
+ default:
+ // The wake failure is already the member's platform log error
+ // (this log line, shipped by logsink.go); marking it keeps the
+ // edge from filing the same failure a second time.
+ log.Printf("activator: wake %s: %v", rt.Host, err)
+ edge.MarkAnswered(r)
+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ }
+ return true
+}
diff --git a/internal/activator/startup_test.go b/internal/activator/startup_test.go
new file mode 100644
index 000000000..8a36e7674
--- /dev/null
+++ b/internal/activator/startup_test.go
@@ -0,0 +1,159 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestStartupWaitUsesWakeTimeoutResponse(t *testing.T) {
+ old := wakeWaitBound
+ wakeWaitBound = 20 * time.Millisecond
+ t.Cleanup(func() { wakeWaitBound = old })
+ a := New(Config{}, &fakeBackend{}, nil)
+ a.Startup = make(chan struct{})
+ defer a.lifeStop()
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true}
+ w := httptest.NewRecorder()
+ a.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "http://shop.example.com/", nil), rt, http.NotFoundHandler())
+ if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") != "5" || w.Body.String() != "instance is waking; retry shortly\n" {
+ t.Fatalf("startup timeout: %d, %v, %q", w.Code, w.Header(), w.Body.String())
+ }
+ if a.Owns(rt.Host) {
+ t.Fatal("timed-out request created a wake")
+ }
+}
+
+func TestStartupBlocksWorkerAndSidecarAdmission(t *testing.T) {
+ old := wakeWaitBound
+ wakeWaitBound = 20 * time.Millisecond
+ t.Cleanup(func() { wakeWaitBound = old })
+ a := New(Config{}, &fakeBackend{}, nil)
+ a.Startup = make(chan struct{})
+ defer a.lifeStop()
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true}
+ if err := a.WakeWorker(rt); !errors.Is(err, errWakeTimeout) {
+ t.Fatalf("worker: %v", err)
+ }
+ if err := a.BeginSidecarRequest(t.Context(), rt); !errors.Is(err, errWakeTimeout) {
+ t.Fatalf("sidecar: %v", err)
+ }
+ ctx, cancel := context.WithCancel(t.Context())
+ cancel()
+ if _, err := a.begin(ctx, rt, true); !errors.Is(err, context.Canceled) {
+ t.Fatalf("cancelled request: %v", err)
+ }
+ if a.Owns(rt.Host) {
+ t.Fatal("startup admission created a wake")
+ }
+}
+
+func TestStartupShutdownAnswersServiceUnavailable(t *testing.T) {
+ for _, ready := range []bool{false, true} {
+ t.Run(map[bool]string{false: "held", true: "released"}[ready], func(t *testing.T) {
+ a := New(Config{}, &fakeBackend{}, nil)
+ started := make(chan struct{})
+ a.Startup = started
+ if ready {
+ close(started)
+ }
+ a.Shutdown(t.Context())
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true}
+ w := httptest.NewRecorder()
+ a.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "http://shop.example.com/", nil), rt, http.NotFoundHandler())
+ if w.Code != http.StatusServiceUnavailable {
+ t.Fatalf("service shutdown answered %d: %q", w.Code, w.Body.String())
+ }
+ if a.Owns(rt.Host) {
+ t.Fatal("shutdown request created a wake")
+ }
+ })
+ }
+}
+
+type startupRoutes map[string]registry.Route
+
+func (s startupRoutes) Get(host string) (*registry.Route, error) {
+ rt, ok := s[host]
+ if !ok {
+ return nil, nil
+ }
+ return &rt, nil
+}
+
+// A request held through startup carries the proxy the edge chose before
+// recovery ran. When recovery moved the host's socket, it is answered as a
+// wake in progress, so the retry is routed afresh, never sent to the old
+// socket.
+func TestStartupHeldRequestRefusesStaleUpstream(t *testing.T) {
+ for _, moved := range []bool{false, true} {
+ a := New(Config{}, &fakeBackend{}, nil)
+ ready := make(chan struct{})
+ a.Startup = ready
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/shop.example.com.sock"}
+ fresh := rt
+ if moved {
+ fresh.Socket = "/run/carlos/instances/shop.example.com/app.sock"
+ }
+ a.Routes = startupRoutes{rt.Host: fresh}
+ proxied := false
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { proxied = true })
+ w := httptest.NewRecorder()
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ a.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "http://shop.example.com/", nil).WithContext(ctx), rt, proxy)
+ }()
+ time.Sleep(20 * time.Millisecond)
+ close(ready)
+ <-done
+ a.lifeStop()
+ if moved {
+ if proxied || w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") != "5" {
+ t.Fatalf("moved socket: proxied=%v code=%d", proxied, w.Code)
+ }
+ } else if w.Code == http.StatusServiceUnavailable && w.Body.String() == "instance is waking; retry shortly\n" {
+ t.Fatalf("unchanged socket was refused as stale")
+ }
+ }
+}
+
+// The race the waiting check misses: the edge chose its proxy, recovery
+// moved the socket and Startup closed before serve looked. A request near
+// startup is still checked; one long after it is not.
+func TestStartupClosedNearStartupRefusesStaleUpstream(t *testing.T) {
+ for _, late := range []bool{false, true} {
+ a := New(Config{}, &fakeBackend{}, nil)
+ ready := make(chan struct{})
+ close(ready)
+ a.Startup = ready
+ if late {
+ a.startupSeenAt.Store(time.Now().Add(-2 * startupFreshWindow).UnixNano())
+ }
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true, Socket: "/run/carlos/shop.example.com.sock"}
+ fresh := rt
+ fresh.Socket = "/run/carlos/instances/shop.example.com/app.sock"
+ a.Routes = startupRoutes{rt.Host: fresh}
+ proxied := false
+ proxy := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { proxied = true })
+ w := httptest.NewRecorder()
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ a.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "http://shop.example.com/", nil).WithContext(ctx), rt, proxy)
+ cancel()
+ a.lifeStop()
+ stale := !proxied && w.Code == http.StatusServiceUnavailable && w.Body.String() == "instance is waking; retry shortly\n"
+ if !late && !stale {
+ t.Fatalf("near startup: moved socket not refused: proxied=%v code=%d", proxied, w.Code)
+ }
+ if late && stale {
+ t.Fatal("long after startup: the route was re-checked")
+ }
+ }
+}
diff --git a/internal/edge/startup_test.go b/internal/edge/startup_test.go
new file mode 100644
index 000000000..81fb78612
--- /dev/null
+++ b/internal/edge/startup_test.go
@@ -0,0 +1,86 @@
+package edge
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "io"
+ "net"
+ "net/http"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestStartupGateFailsWhenHandlerSetupFails(t *testing.T) {
+ if os.Getenv("CARLOS_TEST_SETUP_FAILURE") == "1" {
+ RunWithActivatorReady([]string{"-dev", "-registry", filepath.Join(os.Getenv("CARLOS_DATA_DIR"), "registry.db"), "-http", os.Getenv("CARLOS_TEST_HTTP_ADDR")}, nil, nil, nil, func(*Edge) {
+ _, _ = os.Stdout.WriteString("configuring\n")
+ _, _ = io.CopyN(io.Discard, os.Stdin, 1)
+ })
+ os.Exit(0)
+ }
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ addr := listener.Addr().String()
+ listener.Close()
+ exe, err := os.Executable()
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+ defer cancel()
+ cmd := exec.CommandContext(ctx, exe, "-test.run=^TestStartupGateFailsWhenHandlerSetupFails$")
+ dd := t.TempDir()
+ cmd.Dir = dd
+ cmd.Env = []string{
+ "HOME=" + dd, "TMPDIR=" + os.TempDir(),
+ "CARLOS_DATA_DIR=" + dd, "CARLOS_TEST_SETUP_FAILURE=1", "CARLOS_TEST_HTTP_ADDR=" + addr,
+ "CARLOS_TUNNEL_HOST=tunnel.example.com", "CARLOS_TUNNEL_CLIENTS=invalid",
+ }
+ var stderr bytes.Buffer
+ cmd.Stderr = &stderr
+ stdout, err := cmd.StdoutPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+ stdin, err := cmd.StdinPipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := cmd.Start(); err != nil {
+ t.Fatal(err)
+ }
+ defer func() { cancel(); _ = cmd.Wait() }()
+ line, err := bufio.NewReader(stdout).ReadString('\n')
+ if err != nil || line != "configuring\n" {
+ t.Fatalf("setup barrier: %q, %v", line, err)
+ }
+ client := &http.Client{Transport: &http.Transport{Proxy: nil}, Timeout: 100 * time.Millisecond}
+ defer client.CloseIdleConnections()
+ probeMustFail := func() {
+ t.Helper()
+ resp, err := client.Get("http://" + addr + "/")
+ if err == nil {
+ resp.Body.Close()
+ t.Fatalf("update probe succeeded before handler setup: %d", resp.StatusCode)
+ }
+ }
+ probeMustFail()
+ if _, err := stdin.Write([]byte("x")); err != nil {
+ t.Fatal(err)
+ }
+ stdin.Close()
+ if err := cmd.Wait(); err == nil {
+ t.Fatal("invalid tunnel configuration did not fail setup")
+ }
+ if !strings.Contains(stderr.String(), "edge: tunnels:") {
+ t.Fatalf("did not reach handler setup failure: %s", stderr.String())
+ }
+ probeMustFail()
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 3d8fdf552..350bb75a8 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -213,6 +213,9 @@ func RunAgent(args []string) { runAgent(args, productionClaimFactory) }
func runAgent(args []string, buildClaims ClaimFactory) {
agentCtx, cancelAgent := context.WithCancel(context.Background())
defer cancelAgent()
+ // Startup can stop before children drain; rollout keeps agentCtx until Close.
+ startupCtx, cancelStartup := context.WithCancel(agentCtx)
+ defer cancelStartup()
claimCtx, err := claimEntryContext(agentCtx, os.Getenv, openDeployStore)
if err != nil {
log.Fatalf("agent: host claims: %v", err)
@@ -375,6 +378,8 @@ func runAgent(args []string, buildClaims ClaimFactory) {
var act edge.Activator
var svc *activator.Service
+ started := make(chan struct{})
+ var startupDone <-chan struct{}
if cfg != nil {
// Exec children and carlos-app@ unit tenants share one state machine;
// the split backend dispatches per route on Backing. Note the arming
@@ -522,24 +527,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
if logSink != nil {
svc.Logs = logSink
}
- armAlways(svc, reg, dd) // flag on or off
- var rolloutSink activator.EventSink
- if logSink != nil { // a typed nil in the interface would not be nil
- rolloutSink = logSink
- }
- armRolloutAndRelay(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
- // The residence-move machinery (issue #331), armed with the
- // activator: the staging-fallback restore needs a same-region
- // mirror client, and the move worker needs the service's quiesce
- // answer. Both are best-effort — a mirror client that cannot
- // build logs and leaves the fallback dark, exactly like every
- // other optional duty here.
- if m, err := lease.NewS3Mirror(context.Background(), cfg.Region, cfg.Region); err != nil {
- log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err)
- } else {
- eb.Mirror = m
- }
- armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr))
+
// The startup sweep (design §6.7): a leftover instance unit from a
// previous agent process, one this box's registry no longer holds
// a live wake for, is stopped before this process starts spawning
@@ -553,7 +541,32 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// it, which this does not wait for (the edge below must listen
// within the update gate, whatever the box's size); on a claim
// box, Start alone. The order and the branch are activatorStart's.
- activatorStart{
+ svc.Startup = started
+ startupDone = activatorStart{
+ Configure: func(_ context.Context) {
+ armAlways(svc, reg, dd) // flag on or off
+ var rolloutSink activator.EventSink
+ if logSink != nil { // a typed nil in the interface would not be nil
+ rolloutSink = logSink
+ }
+ armRolloutAndRelay(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
+ // The residence-move machinery (issue #331), armed with the
+ // activator: the staging-fallback restore needs a same-region
+ // mirror client, and the move worker needs the service's quiesce
+ // answer. Both are best-effort — a mirror client that cannot
+ // build logs and leaves the fallback dark, exactly like every
+ // other optional duty here.
+ if m, err := lease.NewS3Mirror(agentCtx, cfg.Region, cfg.Region); err != nil {
+ log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err)
+ } else {
+ eb.Mirror = m
+ }
+ armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr))
+ },
+ Ready: func() {
+ close(started)
+ log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
+ },
Claims: claims.Writes.Enabled,
Sweep: func(ctx context.Context) error {
sw, err := eb.Control.Sweep(ctx)
@@ -576,9 +589,8 @@ func runAgent(args []string, buildClaims ClaimFactory) {
Slot: svc.PrepareScheduledToken,
Publish: armedLayoutPass.Store,
Start: svc.Start,
- }.Run(context.Background())
+ }.Run(startupCtx)
act = svc
- log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
// The agent socket: the one channel a sidecar dials back over
// (internal/agentsock, Task 7). Constructed only in this branch —
@@ -609,6 +621,11 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// never delay the edge coming up; runWorkerBootPass itself wakes
// its candidates concurrently, bounded, rather than serially.
go func() {
+ select {
+ case <-started:
+ case <-startupCtx.Done():
+ return
+ }
routes, err := reg.List()
if err != nil {
log.Printf("agent: worker boot pass: list routes: %v", err)
@@ -618,68 +635,10 @@ func runAgent(args []string, buildClaims ClaimFactory) {
}()
}
- // Fleet box binding publish, startup half (customer-fleets plan, Task 6):
- // a one-shot pass over every route already resident in this box's local
- // registry, covering an agent restart (activator.Service's in-memory
- // entries are gone, but an already-running child isn't) and routes that
- // never go through activator.wake at all. Non-authoritative
- // (publishResidentFleetBindings calls WriteFleetBinding with
- // takeover=false) — see fleetbind.go's doc comments for why a bare
- // "this box has this route resident" pass must never steal a binding a
- // wake-success write (lease-backed, authoritative) already holds for a
- // DIFFERENT box.
- if fleetStore != nil {
- if svc == nil {
- log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", identity.Fleet)
- }
- go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
+ if svc == nil {
+ close(started)
}
- // Scheduled work (schedule spec §6.2/§6.3), built ONCE here and shared by
- // all three legs below: the seed and the fast leg in fastConvergeLoop, the
- // hourly leg in adoptLoop, and the runner started here. One converger is a
- // correctness requirement, not tidiness — its etag memory is what makes a
- // declaration change cheap and a run signal exactly-once, and a second
- // instance elsewhere would replay or swallow one. The three legs run in
- // two different goroutines and do overlap (an hourly pass mid-flight when
- // a console bump fires the fast one); Converger.Pass serializes itself on
- // its own mutex, so no call site here needs one.
- //
- // The two halves arm on different conditions — armSchedule
- // (schedule_wiring.go) is that switch, and carries the why. In short:
- // declared schedules live in the bucket, so the converger needs the
- // deploy store; the RUNNER needs only the registry, because once-timers
- // are registered by instances on their control socket (mounted above,
- // Timers: reg, whenever the activator is armed) straight into this box's
- // rows, and a box that mounts the verb must also fire what it accepts.
- //
- // Guarded assignment, not a direct pass: logSink is a *applog.Sink, and
- // a nil one handed to an interface parameter is NOT a nil interface —
- // the same trap the activator's Logs assignment above avoids.
- var sink activator.EventSink
- if logSink != nil {
- sink = logSink
- }
- schedStore, schedErr := openDeployStore(context.Background())
- sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
- // Move holds (moves spec §9.4): one gate for the runner, the converger
- // and a cut's drain, and the converger's freeze. Nothing with the
- // Phase 1 switch off.
- armScheduleHolds(sd, armedHoldView())
- if err := configureScheduleClaims(claims, sd); err != nil {
- log.Fatalf("agent: host claims: %v", err)
- }
- if sd != nil {
- sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
- }
- serialClaims := make(chan struct{}, 1)
- regionalReport := func(err error) { log.Printf("agent: regional claims: %v", err) }
- if hostwrite.Enabled(claimEntryEnv(claimCtx)) && moves.Enabled(os.Getenv) {
- regionalReport = movesRefusalLog.report
- }
- if err := claims.startRegional(agentCtx, serialClaims, regionalReport); err != nil {
- log.Fatalf("agent: regional claims: %v", err)
- }
defer func() {
cancelAgent()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -688,63 +647,151 @@ func runAgent(args []string, buildClaims ClaimFactory) {
log.Printf("agent: regional claims shutdown: %v", err)
}
}()
+ var stopHousekeeper func()
+ if !claims.Writes.Enabled {
+ // Certificate issuance must not wait for instance recovery.
+ stopHousekeeper = startHousekeeper()
+ }
+ dutiesDone := make(chan struct{})
+ runDuties := func() {
+ defer close(dutiesDone)
+ select {
+ case <-started:
+ case <-startupCtx.Done():
+ return
+ }
+ if startupCtx.Err() != nil {
+ return
+ }
- // The relocation pass (stage B Task 10): it says itself what it started.
- startRelocationPass(agentCtx, claimCtx, claims, reg, svc)
- go adoptLoop(reg, svc, sd, claims.Writes)
- go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)
- if logSink != nil {
- go logshipLoop(logSink, status.Label(readMachineID()))
- }
- if errSink != nil {
- go errshipLoop(errSink, status.Label(readMachineID()))
- }
- // The build-state report (settings spec rev 11 §4.4) names the rows
- // keepsBuildState selects, with host claims as the build-state store
- // sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).
- buildStateClaims := cfg != nil && cfg.HostWrites.Enabled
- collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }
- go statusLoop(reg, svc, collectBuildStates)
- // The memory early-warning (memory-pressure spec §2.4): per box, gated
- // by CARLOS_HOST_METRICS — one log line and nothing else where unset.
- go memwatchLoop()
- go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)
- go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)
- // The deployment-global election, armed BEFORE the loops it gates so a
- // single-box deployment is holding well before their first tick (a
- // minute away) — see housekeeper.go. Not a `go` of its own: it publishes
- // the process's housekeeper and spawns its loop itself. Its stand-down
- // runs in the post-drain block at the end of this function.
- stopHousekeeper := startHousekeeper()
- go retentionLoop(reg, collectBuildStates)
- go registrySnapshotLoop(reg)
- go edgesFileLoop(reg)
- // Region HA enrollment evidence (spec 1 §5.1): once, at start, and only
- // with CARLOS_REGION_HA on. See boxrecord.go.
- go boxRecordLoop()
- // The mesh key's own loop (multi-region backhaul spec, Part 3 Task 2) —
- // deliberately NOT a duty riding retentionLoop's tick: that loop returns
- // early, never arming at all, on a box missing CARLOS_RELEASE_PUBKEY,
- // and such a box still WINS the housekeeper election. A mint duty living
- // only inside retentionLoop would then silently never run fleet-wide.
- // See mintMeshKeyLoop's own comment (housekeeper.go).
- go mintMeshKeyLoop()
- // The DNS steering converger (multi-region ingress spec, Part 4) — its
- // own loop for mintMeshKeyLoop's exact reason, and dormant on every box
- // today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and
- // nothing else, ever. See steering_loop.go.
- go steeringLoop()
- // The sweep reads the registry only through this closure — the stamped
- // bespoke units to sample beside the carlos- ones, re-read each tick.
- go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)
- go configSyncLoop(reg, svc, dd, claims.Writes)
- go configGenLoop(reg, svc, dd, claims.Writes)
- go socketAdvisoryLoop(reg, svc)
+ // Fleet box binding publish, startup half (customer-fleets plan, Task 6):
+ // a one-shot pass over every route already resident in this box's local
+ // registry, covering an agent restart (activator.Service's in-memory
+ // entries are gone, but an already-running child isn't) and routes that
+ // never go through activator.wake at all. Non-authoritative
+ // (publishResidentFleetBindings calls WriteFleetBinding with
+ // takeover=false) — see fleetbind.go's doc comments for why a bare
+ // "this box has this route resident" pass must never steal a binding a
+ // wake-success write (lease-backed, authoritative) already holds for a
+ // DIFFERENT box.
+ if fleetStore != nil {
+ if svc == nil {
+ log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", identity.Fleet)
+ }
+ go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
+ }
+
+ // Scheduled work (schedule spec §6.2/§6.3), built ONCE here and shared by
+ // all three legs below: the seed and the fast leg in fastConvergeLoop, the
+ // hourly leg in adoptLoop, and the runner started here. One converger is a
+ // correctness requirement, not tidiness — its etag memory is what makes a
+ // declaration change cheap and a run signal exactly-once, and a second
+ // instance elsewhere would replay or swallow one. The three legs run in
+ // two different goroutines and do overlap (an hourly pass mid-flight when
+ // a console bump fires the fast one); Converger.Pass serializes itself on
+ // its own mutex, so no call site here needs one.
+ //
+ // The two halves arm on different conditions — armSchedule
+ // (schedule_wiring.go) is that switch, and carries the why. In short:
+ // declared schedules live in the bucket, so the converger needs the
+ // deploy store; the RUNNER needs only the registry, because once-timers
+ // are registered by instances on their control socket (mounted above,
+ // Timers: reg, whenever the activator is armed) straight into this box's
+ // rows, and a box that mounts the verb must also fire what it accepts.
+ //
+ // Guarded assignment, not a direct pass: logSink is a *applog.Sink, and
+ // a nil one handed to an interface parameter is NOT a nil interface —
+ // the same trap the activator's Logs assignment above avoids.
+ var sink activator.EventSink
+ if logSink != nil {
+ sink = logSink
+ }
+ schedStore, schedErr := openDeployStore(agentCtx)
+ sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
+ // Move holds (moves spec §9.4): one gate for the runner, the converger
+ // and a cut's drain, and the converger's freeze. Nothing with the
+ // Phase 1 switch off.
+ armScheduleHolds(sd, armedHoldView())
+ if err := configureScheduleClaims(claims, sd); err != nil {
+ log.Fatalf("agent: host claims: %v", err)
+ }
+ if sd != nil {
+ sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
+ }
+ serialClaims := make(chan struct{}, 1)
+ regionalReport := func(err error) { log.Printf("agent: regional claims: %v", err) }
+ if hostwrite.Enabled(claimEntryEnv(claimCtx)) && moves.Enabled(os.Getenv) {
+ regionalReport = movesRefusalLog.report
+ }
+ if err := claims.startRegional(agentCtx, serialClaims, regionalReport); err != nil {
+ log.Fatalf("agent: regional claims: %v", err)
+ }
+
+ // The relocation pass (stage B Task 10): it says itself what it started.
+ startRelocationPass(agentCtx, claimCtx, claims, reg, svc)
+ go adoptLoop(reg, svc, sd, claims.Writes)
+ go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)
+ if logSink != nil {
+ go logshipLoop(logSink, status.Label(readMachineID()))
+ }
+ if errSink != nil {
+ go errshipLoop(errSink, status.Label(readMachineID()))
+ }
+ // The build-state report (settings spec rev 11 §4.4) names the rows
+ // keepsBuildState selects, with host claims as the build-state store
+ // sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).
+ buildStateClaims := cfg != nil && cfg.HostWrites.Enabled
+ collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }
+ go statusLoop(reg, svc, collectBuildStates)
+ // The memory early-warning (memory-pressure spec §2.4): per box, gated
+ // by CARLOS_HOST_METRICS — one log line and nothing else where unset.
+ go memwatchLoop()
+ go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)
+ go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)
+ // The deployment-global election, armed BEFORE the loops it gates so a
+ // single-box deployment is holding well before their first tick (a
+ // minute away) — see housekeeper.go. Not a `go` of its own: it publishes
+ // the process's housekeeper and spawns its loop itself. Its stand-down
+ // runs in the post-drain block at the end of this function.
+ if claims.Writes.Enabled {
+ stopHousekeeper = startHousekeeper()
+ }
+ go retentionLoop(reg, collectBuildStates)
+ go registrySnapshotLoop(reg)
+ go edgesFileLoop(reg)
+ // Region HA enrollment evidence (spec 1 §5.1): once, at start, and only
+ // with CARLOS_REGION_HA on. See boxrecord.go.
+ go boxRecordLoop()
+ // The mesh key's own loop (multi-region backhaul spec, Part 3 Task 2) —
+ // deliberately NOT a duty riding retentionLoop's tick: that loop returns
+ // early, never arming at all, on a box missing CARLOS_RELEASE_PUBKEY,
+ // and such a box still WINS the housekeeper election. A mint duty living
+ // only inside retentionLoop would then silently never run fleet-wide.
+ // See mintMeshKeyLoop's own comment (housekeeper.go).
+ go mintMeshKeyLoop()
+ // The DNS steering converger (multi-region ingress spec, Part 4) — its
+ // own loop for mintMeshKeyLoop's exact reason, and dormant on every box
+ // today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and
+ // nothing else, ever. See steering_loop.go.
+ go steeringLoop()
+ // The sweep reads the registry only through this closure — the stamped
+ // bespoke units to sample beside the carlos- ones, re-read each tick.
+ go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)
+ go configSyncLoop(reg, svc, dd, claims.Writes)
+ go configGenLoop(reg, svc, dd, claims.Writes)
+ go socketAdvisoryLoop(reg, svc)
+
+ }
+ if claims.Writes.Enabled {
+ runDuties()
+ } else {
+ go runDuties()
+ }
// The same serving loop as `carlos edge`, with the activator injected.
// Caching applies here too: cached hits still avoid the activator, which
// is the entire point of checking the cache before the hibernation branch.
- // The ready hook fires synchronously, just before the blocking serve, so
+ // The ready hook fires synchronously before listening, so
// tunnelLoop is handed off to its own goroutine rather than called here —
// calling it directly would block the front door forever.
edgeOpts := append([]edge.RunOption{
@@ -812,6 +859,12 @@ func runAgent(args []string, buildClaims ClaimFactory) {
go usageLoop(m.Drain, am)
}, edgeOpts...)
+ cancelStartup()
+ joinCtx, joinCancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer joinCancel()
+ joinAgentStartup(joinCtx, startupDone)
+ joinAgentStartup(joinCtx, dutiesDone)
+
// The edge has drained; take the children down too (they're this process's
// children — orphaning them would strand their leases until TTL expiry).
if svc != nil {
@@ -846,7 +899,9 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// roll of a single-box deployment. Stopping cancels every duty in flight
// and releases the lock, so the next box (or this one, a second later)
// takes over immediately.
- stopHousekeeper()
+ if stopHousekeeper != nil {
+ stopHousekeeper()
+ }
}
// statusSocketProbe is the kernel-table probe the status loop hands
diff --git a/internal/host/instancekeys.go b/internal/host/instancekeys.go
index 74b19c6c0..c4ece4f21 100644
--- a/internal/host/instancekeys.go
+++ b/internal/host/instancekeys.go
@@ -128,6 +128,9 @@ func backfillInstanceKeys(ctx context.Context, routes []registry.Route, b instan
defer cancel()
var st instanceKeyBackfill
for _, rt := range routes {
+ if ctx.Err() != nil {
+ break
+ }
if !instanceKeyRoute(rt) {
st.Skipped++
continue
diff --git a/internal/host/instancekeys_test.go b/internal/host/instancekeys_test.go
index f09c8ac12..8babc8ff4 100644
--- a/internal/host/instancekeys_test.go
+++ b/internal/host/instancekeys_test.go
@@ -153,20 +153,22 @@ func TestActivatorStartBackfillsOnlyALegacyBoxAfterRecoveryAndStart(t *testing.T
listErr error
want []string
}{
- {"legacy box", false, nil, []string{"sweep", "recover", "publish", "start", "backfill"}},
- {"legacy box, routes unreadable", false, errors.New("db locked"), []string{"sweep", "recover", "publish", "start"}},
- {"claim box", true, nil, []string{"start"}},
+ {"legacy box", false, nil, []string{"configure", "sweep", "recover", "publish", "start", "ready", "backfill"}},
+ {"legacy box, routes unreadable", false, errors.New("db locked"), []string{"configure", "sweep", "recover", "publish", "start", "ready"}},
+ {"claim box", true, nil, []string{"configure", "start", "ready"}},
} {
t.Run(tc.name, func(t *testing.T) {
var mu sync.Mutex
var events []string
note := func(e string) { mu.Lock(); events = append(events, e); mu.Unlock() }
done := activatorStart{
- Claims: tc.claims,
- Sweep: func(context.Context) error { note("sweep"); return nil },
- Recover: func(context.Context, bool) *layout.Pass { note("recover"); return nil },
- Routes: func() ([]registry.Route, error) { return []registry.Route{ready}, tc.listErr },
- Route: routeOf(ready),
+ Configure: func(context.Context) { note("configure") },
+ Ready: func() { note("ready") },
+ Claims: tc.claims,
+ Sweep: func(context.Context) error { note("sweep"); return nil },
+ Recover: func(context.Context, bool) *layout.Pass { note("recover"); return nil },
+ Routes: func() ([]registry.Route, error) { return []registry.Route{ready}, tc.listErr },
+ Route: routeOf(ready),
Keys: keyBackfillFunc(func(context.Context, registry.Route) (activator.InstanceKeyOutcome, error) {
note("backfill")
return activator.InstanceKeyMinted, nil
@@ -175,6 +177,13 @@ func TestActivatorStartBackfillsOnlyALegacyBoxAfterRecoveryAndStart(t *testing.T
Publish: func(*layout.Pass) { note("publish") },
Start: func() { note("start") },
}.Run(context.Background())
+ if tc.claims {
+ select {
+ case <-done:
+ default:
+ t.Fatal("claim-box startup became asynchronous")
+ }
+ }
select {
case <-done:
case <-time.After(10 * time.Second):
@@ -304,7 +313,7 @@ func TestActivatorStartDoesNotWaitForTheBackfillOnALargeBox(t *testing.T) {
}
})
began := time.Now()
- var startedAfter time.Duration
+ started := make(chan time.Duration, 1)
done := activatorStart{
Sweep: func(context.Context) error { return nil },
Recover: func(context.Context, bool) *layout.Pass { return nil },
@@ -313,11 +322,16 @@ func TestActivatorStartDoesNotWaitForTheBackfillOnALargeBox(t *testing.T) {
Keys: slowBucket,
Slot: (&activator.Service{}).PrepareScheduledToken,
Publish: func(*layout.Pass) {},
- Start: func() { startedAfter = time.Since(began) },
+ Start: func() { started <- time.Since(began) },
}.Run(ctx)
returned := time.Since(began)
- if startedAfter == 0 || startedAfter > bound {
- t.Errorf("Start was called %v after Run began, want within %v (the backfill must not hold it)", startedAfter, bound)
+ select {
+ case startedAfter := <-started:
+ if startedAfter > bound {
+ t.Errorf("Start took %v, want within %v", startedAfter, bound)
+ }
+ case <-time.After(bound):
+ t.Fatal("Start waited for the backfill")
}
if returned > bound {
t.Errorf("Run returned after %v, want within %v: the edge waits for Run", returned, bound)
diff --git a/internal/host/layoutleg.go b/internal/host/layoutleg.go
index fabd8861d..855a0548d 100644
--- a/internal/host/layoutleg.go
+++ b/internal/host/layoutleg.go
@@ -343,61 +343,60 @@ func startActivator(ctx context.Context, sweep func(context.Context) error, reco
start()
}
-// activatorStart is the activator's start sequence, assembled in runAgent
-// and run once by Run.
-//
-// On a claim box it is Start alone: every write to a host's installation
-// runs under the claim's guard, and the claim's launch source mints the
-// instance key there before every cold admission (claim_snapshot.go
-// EnsureClaimAdminToken), so a claim box never backfills (instance-key plan
-// ruling 7).
-//
-// On a legacy box it is the sweep, then layout recovery (design §8.2,
-// "Recovery is the journal": every journal a previous agent left is
-// completed and the host converged to the fleet's intent), then the pass's
-// publication, then Start. The pass is published only after recovery, so
-// the serial leg never runs beside it.
-//
-// The instance-key backfill (L21) runs after Start, in the background, and
-// Run does not wait for it (the v0.26.1 hotfix). Before, it ran between
-// recovery and the publication, one bucket read per keyless row, and on a
-// box with 127 instances the edge, which comes up after Run returns, did
-// not listen within the update gate's 20 seconds. The backfill still runs
-// after recovery has settled every journal. What kept anything from moving
-// a file under it while it ran first, nothing else running yet, is now
-// the activator's per-host preparation slot (keyBackfillInSlot): each row
-// is read again and written inside it; a host that an operation holds
-// fenced (a layout pass, a move) is skipped, and a fence asked for while
-// the row is written waits until the write is done. A start's own mint
-// beside the backfill is serialized by the activator's adminTokenMu, and
-// both are create-only, so the host ends with one key.
+// activatorStart keeps sweep, recovery and Start ordered. Legacy boxes run
+// the sequence in the background; claim boxes still call Start alone.
+// Ready releases requests and agent duties only after Start returns. The
+// returned channel also joins the instance-key pass on shutdown.
type activatorStart struct {
- Claims bool // claims.Writes.Enabled: a claim box
- Sweep func(context.Context) error // the instance-unit sweep
- Recover func(context.Context, bool) *layout.Pass // builds the pass and recovers under it; the bool is whether the sweep succeeded
- Routes func() ([]registry.Route, error) // the registry's List
- Route func(string) (*registry.Route, error) // the registry's Get: each row is read again in its host's slot
- Keys instanceKeyBackfiller // the exec backend
- Slot hostPreparationSlot // the service's PrepareScheduledToken
- Publish func(*layout.Pass) // armedLayoutPass.Store
- Start func() // the service's Start
+ Configure func(context.Context)
+ Ready func()
+ Claims bool // claims.Writes.Enabled: a claim box
+ Sweep func(context.Context) error // the instance-unit sweep
+ Recover func(context.Context, bool) *layout.Pass // builds the pass and recovers under it; the bool is whether the sweep succeeded
+ Routes func() ([]registry.Route, error) // the registry's List
+ Route func(string) (*registry.Route, error) // the registry's Get: each row is read again in its host's slot
+ Keys instanceKeyBackfiller // the exec backend
+ Slot hostPreparationSlot // the service's PrepareScheduledToken
+ Publish func(*layout.Pass) // armedLayoutPass.Store
+ Start func() // the service's Start
}
-// Run starts the activator and returns. The channel closes when the
-// instance-key backfill has finished; on a claim box it is closed already.
-// runAgent does not wait on it.
+// Run returns immediately on legacy boxes. Cancellation prevents a late
+// Start and reaches the key pass through the agent lifetime context.
func (s activatorStart) Run(ctx context.Context) <-chan struct{} {
done := make(chan struct{})
+ start := func() {
+ if ctx.Err() == nil {
+ s.Start()
+ if s.Ready != nil {
+ s.Ready()
+ }
+ }
+ }
if s.Claims {
- s.Start()
+ if s.Configure != nil {
+ s.Configure(ctx)
+ }
+ start()
close(done)
return done
}
- startActivator(ctx, s.Sweep, func(ctx context.Context, swept bool) {
- s.Publish(s.Recover(ctx, swept))
- }, s.Start)
go func() {
defer close(done)
+ if s.Configure != nil {
+ s.Configure(ctx)
+ }
+ if ctx.Err() != nil {
+ return
+ }
+ startActivator(ctx, s.Sweep, func(ctx context.Context, swept bool) {
+ if ctx.Err() == nil {
+ s.Publish(s.Recover(ctx, swept))
+ }
+ }, start)
+ if ctx.Err() != nil {
+ return
+ }
routes, err := s.Routes()
if err != nil {
log.Printf("agent: instance keys: list routes: %v (each host's next start mints its key)", err)
@@ -483,6 +482,8 @@ func startupRecoveryIntent(ctx context.Context, dataDir string, open func(contex
if err != nil {
log.Printf("layout: list journals at start: %v; reading the intent anyway", err)
}
+ ctx, cancel := context.WithTimeout(ctx, layoutIntentTimeout)
+ defer cancel()
store, err := open(ctx)
return startupLayoutIntent(ctx, store, err, boxLabel)
}
diff --git a/internal/host/startup.go b/internal/host/startup.go
new file mode 100644
index 000000000..d1de4d560
--- /dev/null
+++ b/internal/host/startup.go
@@ -0,0 +1,15 @@
+package host
+
+import "context"
+
+func joinAgentStartup(ctx context.Context, done <-chan struct{}) bool {
+ if done == nil {
+ return true
+ }
+ select {
+ case <-done:
+ return true
+ case <-ctx.Done():
+ return false
+ }
+}
diff --git a/internal/host/startup_test.go b/internal/host/startup_test.go
new file mode 100644
index 000000000..bb07584b0
--- /dev/null
+++ b/internal/host/startup_test.go
@@ -0,0 +1,285 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/layout"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type startupBackend struct {
+ fakeWakeBackend
+ recovered *atomic.Bool
+ started *atomic.Bool
+ attempts atomic.Int32
+ gate func() error
+}
+
+func (b *startupBackend) Acquire(ctx context.Context, rt registry.Route) (activator.Lease, error) {
+ b.attempts.Add(1)
+ if !b.recovered.Load() || !b.started.Load() {
+ return nil, errors.New("wake preceded recovery or Start")
+ }
+ if b.gate != nil {
+ if err := b.gate(); err != nil {
+ return nil, err
+ }
+ }
+ return b.fakeWakeBackend.Acquire(ctx, rt)
+}
+
+type observingActivator struct {
+ *activator.Service
+ entered chan struct{}
+}
+
+func (a observingActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
+ a.entered <- struct{}{}
+ return a.Service.ServeHTTP(w, r, rt, proxy)
+}
+
+func TestStartupServesWhileSweepAndRecoveryWait(t *testing.T) {
+ t.Setenv("CARLOS_STEERING_ZONE_ID", "Z1234567890")
+ t.Setenv("PLATFORM_APPS_DOMAIN", "apps.example.com")
+ for _, fail := range []bool{false, true} {
+ t.Run(map[bool]string{false: "slow sweep and recovery", true: "unreachable instance control"}[fail], func(t *testing.T) {
+ dd := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dd, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ inst := &fakeInstance{socket: filepath.Join(dd, "app.sock"), handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { io.WriteString(w, "awake") })}
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Hibernate: true, Status: "ready", Socket: inst.socket, DBPath: filepath.Join(dd, "app.db")}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }))
+ defer console.Close()
+ if err := reg.Upsert(registry.Route{Host: "console.example.com", Addr: console.Listener.Addr().String()}); err != nil {
+ t.Fatal(err)
+ }
+ var recovered, startedService atomic.Bool
+ backend := &startupBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, recovered: &recovered, started: &startedService}
+ control := instancectl.NewClient(filepath.Join(dd, "unreachable.sock"))
+ if fail {
+ backend.gate = nestedGate(func() error { return nil }, func(string) (bool, error) { return true, nil }, control.Ready)
+ }
+ svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
+ ready := make(chan struct{})
+ svc.Startup = ready
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ defer func() { svc.Shutdown(context.Background()) }()
+ sweepEntered, recoverEntered := make(chan struct{}), make(chan struct{})
+ sweepRelease, recoverRelease := make(chan struct{}), make(chan struct{})
+ done := activatorStart{
+ Sweep: func(ctx context.Context) error {
+ close(sweepEntered)
+ select {
+ case <-sweepRelease:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ if fail {
+ _, err := control.Sweep(ctx)
+ return err
+ }
+ return nil
+ },
+ Recover: func(ctx context.Context, swept bool) *layout.Pass {
+ if swept == fail {
+ t.Error("recovery received the wrong sweep result")
+ }
+ close(recoverEntered)
+ select {
+ case <-recoverRelease:
+ case <-ctx.Done():
+ return nil
+ }
+ recovered.Store(true)
+ return nil
+ },
+ Publish: func(*layout.Pass) {}, Start: func() { svc.Start(); startedService.Store(true) }, Ready: func() { close(ready) },
+ Routes: func() ([]registry.Route, error) { return nil, nil },
+ }.Run(ctx)
+ defer func() { cancel(); <-done }()
+ <-sweepEntered
+ entered := make(chan struct{}, 1)
+ e, err := edge.New(reg, observingActivator{svc, entered})
+ if err != nil {
+ t.Fatal(err)
+ }
+ server := httptest.NewServer(e)
+ defer server.Close()
+ client := &http.Client{Timeout: time.Second}
+ get := func(host string) (int, error) {
+ r, _ := http.NewRequest(http.MethodGet, server.URL+"/", nil)
+ r.Host = host
+ resp, err := client.Do(r)
+ if err != nil {
+ return 0, err
+ }
+ io.Copy(io.Discard, resp.Body)
+ resp.Body.Close()
+ return resp.StatusCode, nil
+ }
+ for host, want := range map[string]int{"127.0.0.1": http.StatusNotFound, "console.example.com": http.StatusNoContent, "edge.apps.example.com": http.StatusOK} {
+ began := time.Now()
+ if got, err := get(host); err != nil || got != want {
+ t.Fatalf("%s during sweep: %d, %v", host, got, err)
+ }
+ t.Logf("%s answered in %s while sweep was blocked", host, time.Since(began))
+ }
+ answer := make(chan int, 1)
+ go func() { status, _ := get(rt.Host); answer <- status }()
+ select {
+ case <-entered:
+ case <-time.After(time.Second):
+ t.Fatal("request never reached the activator")
+ }
+ assertHeld := func() {
+ t.Helper()
+ select {
+ case status := <-answer:
+ t.Fatalf("request answered before startup: %d", status)
+ case <-time.After(25 * time.Millisecond):
+ }
+ if backend.attempts.Load() != 0 {
+ t.Fatal("wake attempted before sweep and recovery finished")
+ }
+ }
+ assertHeld()
+ close(sweepRelease)
+ <-recoverEntered
+ assertHeld()
+ if got, err := get("console.example.com"); err != nil || got != http.StatusNoContent {
+ t.Fatalf("console during recovery: %d, %v", got, err)
+ }
+ close(recoverRelease)
+ want := http.StatusOK
+ if fail {
+ want = http.StatusServiceUnavailable
+ }
+ select {
+ case got := <-answer:
+ if got != want {
+ t.Fatalf("instance after Start: %d, want %d", got, want)
+ }
+ case <-time.After(time.Second):
+ t.Fatal("request did not finish after Start")
+ }
+ if backend.attempts.Load() != 1 || inst.running() == fail {
+ t.Fatal("unexpected wake outcome")
+ }
+ })
+ }
+}
+
+func TestStartupCancellationAndBoundedJoin(t *testing.T) {
+ ctx, cancel := context.WithCancel(context.Background())
+ entered := make(chan struct{})
+ var started atomic.Bool
+ done := activatorStart{
+ Sweep: func(ctx context.Context) error { close(entered); <-ctx.Done(); return ctx.Err() },
+ Recover: func(context.Context, bool) *layout.Pass { t.Error("recovery after cancellation"); return nil },
+ Start: func() { started.Store(true) },
+ }.Run(ctx)
+ <-entered
+ cancel()
+ join, stop := context.WithTimeout(t.Context(), time.Second)
+ defer stop()
+ if !joinAgentStartup(join, done) || started.Load() {
+ t.Fatal("cancelled startup was not joined, or started")
+ }
+ blocked := make(chan struct{})
+ expired, expire := context.WithTimeout(t.Context(), 20*time.Millisecond)
+ defer expire()
+ if joinAgentStartup(expired, blocked) {
+ t.Fatal("join ignored its bound")
+ }
+}
+
+func TestStartupJoinsKeyPassOnAgentCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ entered := make(chan struct{})
+ rt := registry.Route{Host: "shop.example.com", Kind: "instance", Socket: "/unused.sock", Status: "ready"}
+ done := activatorStart{
+ Sweep: func(context.Context) error { return nil },
+ Recover: func(context.Context, bool) *layout.Pass { return nil },
+ Publish: func(*layout.Pass) {}, Start: func() {},
+ Routes: func() ([]registry.Route, error) { return []registry.Route{rt}, nil },
+ Route: routeOf(rt), Slot: openSlot,
+ Keys: keyBackfillFunc(func(ctx context.Context, _ registry.Route) (activator.InstanceKeyOutcome, error) {
+ close(entered)
+ <-ctx.Done()
+ return activator.InstanceKeyFailed, ctx.Err()
+ }),
+ }.Run(ctx)
+ select {
+ case <-entered:
+ case <-time.After(time.Second):
+ t.Fatal("key pass did not start")
+ }
+ cancel()
+ join, stop := context.WithTimeout(t.Context(), time.Second)
+ defer stop()
+ if !joinAgentStartup(join, done) {
+ t.Fatal("key pass outlived agent cancellation")
+ }
+}
+
+func TestStartupCancellationDuringRecoveryPreservesAgentLifetime(t *testing.T) {
+ agentCtx, cancelAgent := context.WithCancel(t.Context())
+ defer cancelAgent()
+ startupCtx, cancelStartup := context.WithCancel(agentCtx)
+ defer cancelStartup()
+ recovering := make(chan struct{})
+ svc := activator.New(activator.Config{}, &fakeWakeBackend{}, nil)
+ ready := make(chan struct{})
+ svc.Startup = ready
+ done := activatorStart{
+ Configure: func(context.Context) {
+ if agentCtx.Err() != nil {
+ t.Error("agent lifetime ended before configuration")
+ }
+ },
+ Sweep: func(context.Context) error { return nil },
+ Recover: func(ctx context.Context, _ bool) *layout.Pass {
+ close(recovering)
+ <-ctx.Done()
+ return nil
+ },
+ Publish: func(*layout.Pass) {},
+ Start: func() { t.Error("Start after recovery cancellation") },
+ Ready: func() { close(ready) },
+ }.Run(startupCtx)
+ <-recovering
+ cancelStartup()
+ join, stop := context.WithTimeout(t.Context(), time.Second)
+ defer stop()
+ if !joinAgentStartup(join, done) {
+ t.Fatal("recovery did not join")
+ }
+ svc.Shutdown(join)
+ if agentCtx.Err() != nil {
+ t.Fatal("startup cancellation stopped rollout's agent lifetime before children drained")
+ }
+ select {
+ case <-ready:
+ t.Fatal("cancelled recovery released admission")
+ default:
+ }
+}
diff --git a/internal/provision/plan14_root_reads_test.go b/internal/provision/plan14_root_reads_test.go
index e51bd4792..0033203f3 100644
--- a/internal/provision/plan14_root_reads_test.go
+++ b/internal/provision/plan14_root_reads_test.go
@@ -651,6 +651,13 @@ func TestPlan14RootReadCensus(t *testing.T) {
// Census ruling zddr3 (#272): the exec shim's accepted-legacy rows and the
// three refreshed legacy rows add thirteen frozen declarations.
seal = "c748e02bfe907a009ef42eb544b1a44fe813f4f7c13e454be3980aec4fade568"
+ if errs := checkRootReadLegacy(inv, seal); len(errs) != 0 {
+ t.Fatal(errs)
+ }
+ if err = loadRootReadInventory(filepath.Join(root, "internal/provision/testdata/plan14-root-reads-hotfix281.tsv"), &inv, true); err != nil {
+ t.Fatal(err)
+ }
+ seal = "6da50872e15f7f396d23e5b1b2e044a73cd9e5c07d85c8e22252b0a1a5304bbd"
errs := append(compareRootReads(got, inv), checkRootReadLegacy(inv, seal)...)
errs = append(errs, checkRootReadDeclarations(got, inv)...)
if len(errs) > 0 {
diff --git a/internal/provision/testdata/plan14-root-reads-hotfix281.tsv b/internal/provision/testdata/plan14-root-reads-hotfix281.tsv
new file mode 100644
index 000000000..42b73f8c2
--- /dev/null
+++ b/internal/provision/testdata/plan14-root-reads-hotfix281.tsv
@@ -0,0 +1,18 @@
+# Hotfix 281: changed units only; read-preserving startup changes under census-ruling-hotfix281.
+@legacy f28afc22c409e85b3c186db8947c41e30c541bbd007b1b374e56e2c98cc0a74c internal/activator/activator.go Service.BeginSidecarRequest legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host reads follow a bounded startup wait; no new source or authority. registry:Host@259,registry:Host@333 2f3da0a2d25f3c75ed9b89dd361e28a00410843fe0f80cc3995e8f1f1a1ae994
+internal/activator/activator.go Service.BeginSidecarRequest legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host reads follow a bounded startup wait; no new source or authority. registry:Host@259,registry:Host@333 2f3da0a2d25f3c75ed9b89dd361e28a00410843fe0f80cc3995e8f1f1a1ae994
+@legacy ed600af90f562c710f917d1d0a5ebc0fae5c9f0851a69f23395a18bb83901ff3 internal/activator/activator.go Service.WakeWorker legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host and route reads follow a bounded startup wait; no new source or authority. registry:Host@1242,registry:Host@1652,registry:Host@834,registry:Host@864,whole:Route@1225 bda1e33466a92d540eeb724284aa431550a323e7dbb432cc7b8872498929bec9
+internal/activator/activator.go Service.WakeWorker legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host and route reads follow a bounded startup wait; no new source or authority. registry:Host@1242,registry:Host@1652,registry:Host@834,registry:Host@864,whole:Route@1225 bda1e33466a92d540eeb724284aa431550a323e7dbb432cc7b8872498929bec9
+@legacy b3e3b36908d945c5c486c57073c911b113966f1cc2fb93347e0057059b4750be internal/activator/activator.go Service.begin legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host and route reads follow a bounded startup wait; no new source or authority. registry:Host@1197,registry:Host@1276,registry:Host@326,registry:Host@405,whole:Route@1206,whole:Route@1213 642ece8b785a5654dc04d4461995b77dffa2073bcad44897581cfa0a390b6016
+internal/activator/activator.go Service.begin legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=The same host and route reads follow a bounded startup wait; no new source or authority. registry:Host@1197,registry:Host@1276,registry:Host@326,registry:Host@405,whole:Route@1206,whole:Route@1213 642ece8b785a5654dc04d4461995b77dffa2073bcad44897581cfa0a390b6016
+@legacy cbe493713bb70584550a72e54677628ff5ca7af7c4cd1e7a7e4b0a98302de1eb internal/activator/activator.go Service.serve legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Startup repeats the existing Kind and Hibernate checks; error logging moved to serveWakeError. A request held through startup re-reads its own route's Socket and Addr and can only refuse (wake-timeout 503) when recovery moved them; agent-side, deny-only, no root decision. registry:Addr@952,registry:Addr@966,registry:Hibernate@1468,registry:Hibernate@151,registry:Host@1077,registry:Host@877,registry:Kind@126,registry:Kind@1442,registry:Socket@923,registry:Socket@939,whole:Route@1011,whole:Route@2587,whole:Route@347,whole:Route@3556,whole:Route@3616,whole:Route@902 75fc203da1dfc4c88db100c634a370777928a441480bfc9547cb0fb19a21c9fe
+internal/activator/activator.go Service.serve legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Startup repeats the existing Kind and Hibernate checks; error logging moved to serveWakeError. A request held through startup re-reads its own route's Socket and Addr and can only refuse (wake-timeout 503) when recovery moved them; agent-side, deny-only, no root decision. registry:Addr@952,registry:Addr@966,registry:Hibernate@1468,registry:Hibernate@151,registry:Host@1077,registry:Host@877,registry:Kind@126,registry:Kind@1442,registry:Socket@923,registry:Socket@939,whole:Route@1011,whole:Route@2587,whole:Route@347,whole:Route@3556,whole:Route@3616,whole:Route@902 75fc203da1dfc4c88db100c634a370777928a441480bfc9547cb0fb19a21c9fe
+@legacy bf4ca1db43d8023db08187eb3afdcfbd4599b3f6966c81274b60e2752b6c5fac internal/activator/activator.go Service.serveWakeError legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Extracted unchanged error mapping and Host log read from Service.serve; no new source or authority. registry:Host@1364 e4ad730f8b463cf112c59bd061c7384ff4afe626d2bedd7f90ccd3df791333ba
+internal/activator/activator.go Service.serveWakeError legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Extracted unchanged error mapping and Host log read from Service.serve; no new source or authority. registry:Host@1364 e4ad730f8b463cf112c59bd061c7384ff4afe626d2bedd7f90ccd3df791333ba
+internal/host/agent.go runAgent authority B,C,D,F rev3=census-ruling-hotfix281;basis=Only activator startup and dependent duties run asynchronously; shutdown cancels their child context before the join, preserving the agent lifetime through child and rollout drain. Registry, route, data-directory and backend inputs are unchanged; B,C,D,F authority is retained. file:Exec@13674,file:Open@731,registry:App@10065,whole:Route@10057,whole:Route@21396 5d8ca6215a4f7eddc85fd05a6696494d22922e8c23eab4d4eb91605377d2277c
+@legacy daa120f77654cd1238b494d92b7a4db9ee115f4f56945243b26a7ebe02e584db internal/host/instancekeys.go backfillInstanceKeys legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Cancellation stops the loop early; route reads, per-host preparation and key sources are unchanged. registry:Host@553,registry:Host@701,registry:Host@867,whole:Route@251,whole:Route@320,whole:Route@399 b4d7ba9a5dc32cb04c214adc69184d91bc65d2487c661150fed899f1deba676d
+internal/host/instancekeys.go backfillInstanceKeys legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Cancellation stops the loop early; route reads, per-host preparation and key sources are unchanged. registry:Host@553,registry:Host@701,registry:Host@867,whole:Route@251,whole:Route@320,whole:Route@399 b4d7ba9a5dc32cb04c214adc69184d91bc65d2487c661150fed899f1deba676d
+@legacy 8884dca0e95b3cdd016d6c150db8f74a6ad64cc8aa4f0d2fdbfc935fa4947ed2 internal/host/layoutleg.go activatorStart.Run legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Legacy sweep, recovery and key pass run in the background under a startup context that is a child of the agent lifetime. The same route list and callbacks are used; claim boxes retain Start alone. whole:Route@803 e93d0c33f6ac39f9c644b7bfa2463c30a9980dfd0e23a515ba87f356e8899c57
+internal/host/layoutleg.go activatorStart.Run legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Legacy sweep, recovery and key pass run in the background under a startup context that is a child of the agent lifetime. The same route list and callbacks are used; claim boxes retain Start alone. whole:Route@803 e93d0c33f6ac39f9c644b7bfa2463c30a9980dfd0e23a515ba87f356e8899c57
+@coverage internal/activator d2f7f2c588ae1acd8166c366638f254631c3806dae4b6f523c27b15ddeb3a418
+@coverage internal/host ee4e46a4efcbea38cbc81fcdcab7ac4dc5eac2eeecb9ed879aecc4968315cf74