amadan
Merge remote-tracking branch 'origin/main' into feat/claim-root-trusted-state-e
# Conflicts: # internal/provision/plan14_root_reads_test.go
13 files changed,
+968
−226
internal/activator/activator.go+132 −35internal/activator/startup_test.go+159 −0internal/edge/startup_test.go+86 −0internal/host/agent.go+189 −134internal/host/instancekeys.go+3 −0internal/host/instancekeys_test.go+26 −12internal/host/layoutleg.go+45 −44internal/host/rootinventory_test.go+2 −0internal/host/startup.go+15 −0internal/host/startup_test.go+285 −0internal/provision/plan14_root_reads_test.go+7 −0internal/provision/testdata/plan14-root-reads-e.tsv+1 −1internal/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 345.580s
? github.com/carlosframework/platform/cmd/changelog [no test files]
ok github.com/carlosframework/platform/cmd/console 3.059s
ok github.com/carlosframework/platform/cmd/moveprobe 93.294s
ok github.com/carlosframework/platform/cmd/zddprobe 0.377s
ok github.com/carlosframework/platform/internal/activator 228.128s
ok github.com/carlosframework/platform/internal/agentsock 0.015s
ok github.com/carlosframework/platform/internal/alerting 0.003s
ok github.com/carlosframework/platform/internal/analytics 0.020s
ok github.com/carlosframework/platform/internal/appconfig 2.340s
ok github.com/carlosframework/platform/internal/apperr 0.072s
ok github.com/carlosframework/platform/internal/applog 0.126s
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.721s
? 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.011s
ok github.com/carlosframework/platform/internal/certwarm 0.002s
ok github.com/carlosframework/platform/internal/cgroup 0.690s
ok github.com/carlosframework/platform/internal/changelog 0.293s
ok github.com/carlosframework/platform/internal/clamd 0.054s
? github.com/carlosframework/platform/internal/cliflag [no test files]
ok github.com/carlosframework/platform/internal/console 362.763s
? 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.093s
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.004s
ok github.com/carlosframework/platform/internal/domains 2.128s
ok github.com/carlosframework/platform/internal/doorkey 0.006s
ok github.com/carlosframework/platform/internal/economics 0.004s
ok github.com/carlosframework/platform/internal/edge 26.019s
? github.com/carlosframework/platform/internal/edgeproof [no test files]
ok github.com/carlosframework/platform/internal/edgesfile 0.008s
ok github.com/carlosframework/platform/internal/fleetapi 0.125s
ok github.com/carlosframework/platform/internal/gate 1.656s
ok github.com/carlosframework/platform/internal/genselect 0.015s
ok github.com/carlosframework/platform/internal/geo 0.018s
ok github.com/carlosframework/platform/internal/geoip 0.029s
ok github.com/carlosframework/platform/internal/historyreadback 0.102s
ok github.com/carlosframework/platform/internal/host 99.677s
ok github.com/carlosframework/platform/internal/hostactors 0.553s
ok github.com/carlosframework/platform/internal/hostbox 2.196s
ok github.com/carlosframework/platform/internal/hostclaim 0.489s
ok github.com/carlosframework/platform/internal/hostevidence 0.008s
ok github.com/carlosframework/platform/internal/hostexec 4.848s
ok github.com/carlosframework/platform/internal/hostfence 0.042s
ok github.com/carlosframework/platform/internal/hostfiles 10.489s
ok github.com/carlosframework/platform/internal/hostfloor 0.014s
ok github.com/carlosframework/platform/internal/hostguard 1.006s
ok github.com/carlosframework/platform/internal/hostops 1.442s
? 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.437s
ok github.com/carlosframework/platform/internal/hostresolve 0.130s
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.813s
ok github.com/carlosframework/platform/internal/instanceexec 2.707s
ok github.com/carlosframework/platform/internal/instancefs 0.004s
ok github.com/carlosframework/platform/internal/instancekey 0.004s
ok github.com/carlosframework/platform/internal/instanceupdates 0.021s
ok github.com/carlosframework/platform/internal/layout 1.412s
ok github.com/carlosframework/platform/internal/lease 9.440s
ok github.com/carlosframework/platform/internal/lease/leasetest 0.007s
ok github.com/carlosframework/platform/internal/ledger 0.011s
ok github.com/carlosframework/platform/internal/mail 0.170s
ok github.com/carlosframework/platform/internal/maintenance 0.075s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.546s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.067s
ok github.com/carlosframework/platform/internal/metrics 0.005s
ok github.com/carlosframework/platform/internal/moves 20.905s
ok github.com/carlosframework/platform/internal/moves/archive 0.742s
ok github.com/carlosframework/platform/internal/objstore 0.172s
ok github.com/carlosframework/platform/internal/ops 0.167s
ok github.com/carlosframework/platform/internal/park 0.201s
ok github.com/carlosframework/platform/internal/privatefile 0.003s
ok github.com/carlosframework/platform/internal/provision 13.214s
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.095s
ok github.com/carlosframework/platform/internal/relay 7.237s
ok github.com/carlosframework/platform/internal/release 20.762s
ok github.com/carlosframework/platform/internal/rollout 3.722s
ok github.com/carlosframework/platform/internal/rolloutwords 0.003s
ok github.com/carlosframework/platform/internal/rootaws 0.006s
ok github.com/carlosframework/platform/internal/rootparticipation 0.008s
ok github.com/carlosframework/platform/internal/rootstate 0.035s
ok github.com/carlosframework/platform/internal/s3compat 0.004s
ok github.com/carlosframework/platform/internal/safefile 0.003s
ok github.com/carlosframework/platform/internal/schedule 2.300s
ok github.com/carlosframework/platform/internal/sim/hostlife 9.597s
ok github.com/carlosframework/platform/internal/sim/litestreamfacts 4.457s
ok github.com/carlosframework/platform/internal/status 0.036s
ok github.com/carlosframework/platform/internal/steering 0.005s
ok github.com/carlosframework/platform/internal/stopevidence 0.190s
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.006s
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.001s
ok github.com/carlosframework/platform/internal/vet 1.567s
=== 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: cached, digest verified
mc: cached, 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.022s
=== RUN TestT4StorageProbeMinio
2026/10/13 22:29:01 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE37557143D275, 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
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 a8f9e87c3..05565a072 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/rootinventory_test.go b/internal/host/rootinventory_test.go| index f714abe04..e34af4113 100644 |
| --- a/internal/host/rootinventory_test.go |
| +++ b/internal/host/rootinventory_test.go |
| @@ -1,3 +1,5 @@ |
| +//go:build linux |
| + |
| package host |
| import ( |
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 a02f43137..39802ba61 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" |
| if err = loadRootReadInventory(filepath.Join(root, "internal/provision/testdata/plan14-root-reads-e.tsv"), &inv, true); err != nil { |
| t.Fatal(err) |
| } |
diff --git a/internal/provision/testdata/plan14-root-reads-e.tsv b/internal/provision/testdata/plan14-root-reads-e.tsv| index 9b7e27b3d..cc05eeca3 100644 |
| --- a/internal/provision/testdata/plan14-root-reads-e.tsv |
| +++ b/internal/provision/testdata/plan14-root-reads-e.tsv |
| @@ -54,7 +54,7 @@ internal/release/roothistory_replace.go replacementReconstruction authority E1 S |
| internal/release/signedhead.go StageSignedHead authority E1 Canonical route is supplied by the root unit verifier. Verifies signed channel, digest ratchet and manifest/artifact before staging; unavailable pin census never authorizes restaging differing bytes. Fixtures: TestPlan14Incident20261006WithLedger and TestPlan14SignedNamedRecoveryAuthority. registry:Account@202,registry:Account@302,registry:Account@403,registry:App@214,registry:App@415,registry:Channel@222,registry:Host@434 ca02f04e55556ba0233dfa0848b0c533a9d9aa06cad6abd33cfd2968b3bd838a |
| internal/rootstate/link_linux.go ReadTrustedLink authority A,E1 Checks the full root-owned parent chain before descriptor-based live-link reading. Fixture: TestPlan14TrustedLiveLink; privileged acceptance remains required. file:Open@156 90f8e8de4e2c71f07b5d749cbe4bb1c1dfa426576a82138823449f56fc9ec5ab |
| internal/rootstate/link_linux.go directory.readLink authority A,E1 Pins the symlink inode with O_PATH and checks owner/type/link count before readlinkat. Fixture: TestPlan14TrustedLiveLink. file:Fstat@356,file:Openat@183,file:Readlinkat@554 a8559fe63dce48830916004ecfe7fbbe5d419568c7daf7a3073cf1b5bb12c3d7 |
| -@coverage internal/host 4c080abdf7dcc48daacaacaf576edd01f1d2a2db7cb41dd3a2a9b21de1039dba |
| +@coverage internal/host d2d36da953f0ca69559cbfe67b919e1cf2a66419c67b8c247177b3da8c7cdbdf |
| @coverage internal/release 2536a8c84d34aa2b798cb8af551ed488ea9ce8023d4d2383db4db5c120f08a8b |
| @coverage internal/rootstate 6338f74a6b2059cf77cf478bd6111b37b441eb9d65a309464c378359d4bdac05 |
| internal/release/roothistory_status.go RootHistory.MigrationStatus authority E1 Exact completion binds the accepted authorization digest from durable private history to enrolled-root signed readback. Pending originals, changed evidence and canonical sources cannot supply retirement completion. Fixtures: TestPlan14HistoryReconstructionCompletion, TestPlan14HistoryRetirementCompletion, TestPlan14ReadReconstructionCompletion, TestPlan14ReadRetirementCompletion, TestPlan14RecoveryRetirementProductionCallback, TestPlan14RootHistoryReconstructionCrash, TestPlan14HistoryRetirementProcessCrash. file:Read@1062,file:Read@3891 f0ab7ad2d2d70f8c22ebfd575f5a1c6931f7eba30ccdba80ad43d0efffc88154 |
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 |