amadan
hotfix #281: the edge listens before the startup sweep and recovery finish
On a claims-off box the agent binds its listeners and answers the update gate's loopback probe before opening the registry or touching instance control. Sweep, layout recovery and Start run in the background; instance admission (wakes, workers, sidecars) and scheduled duties wait for Start, bounded at 30 s, then answer as a wake timeout. Non-instance routes serve once the edge handlers are configured. The instance-key pass takes the agent's lifetime and is joined on shutdown. Claim boxes keep Start alone. v0.27.0 on annie-welding (142 instances) failed the 20 s gate both ways: the cold instance-control sweep held the edge for about 22 s.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
14 files changed,
+962
−255
internal/activator/activator.go+92 −35internal/activator/startup_test.go+54 −0internal/edge/door.go+16 −5internal/edge/early.go+120 −0internal/edge/early_test.go+62 −0internal/edge/edge.go+73 −32internal/host/agent.go+193 −135internal/host/instancekeys.go+3 −0internal/host/instancekeys_test.go+9 −4internal/host/layoutleg.go+45 −44internal/host/startup.go+24 −0internal/host/startup_test.go+241 −0internal/provision/plan14_root_reads_test.go+7 −0internal/provision/testdata/plan14-root-reads-hotfix281.tsv+23 −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 341.771s
? github.com/carlosframework/platform/cmd/changelog [no test files]
ok github.com/carlosframework/platform/cmd/console 3.083s
ok github.com/carlosframework/platform/cmd/moveprobe 92.266s
ok github.com/carlosframework/platform/cmd/zddprobe 0.384s
ok github.com/carlosframework/platform/internal/activator 226.689s
ok github.com/carlosframework/platform/internal/agentsock 0.016s
ok github.com/carlosframework/platform/internal/alerting 0.004s
ok github.com/carlosframework/platform/internal/analytics 0.020s
ok github.com/carlosframework/platform/internal/appconfig 2.314s
ok github.com/carlosframework/platform/internal/apperr 0.073s
ok github.com/carlosframework/platform/internal/applog 0.137s
ok github.com/carlosframework/platform/internal/awsinv 0.004s
ok github.com/carlosframework/platform/internal/boxchoice 0.004s
ok github.com/carlosframework/platform/internal/boxid 0.925s
? 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.701s
ok github.com/carlosframework/platform/internal/changelog 0.296s
ok github.com/carlosframework/platform/internal/clamd 0.058s
? github.com/carlosframework/platform/internal/cliflag [no test files]
ok github.com/carlosframework/platform/internal/console 363.802s
? 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 5.662s
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.147s
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.023s
? 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.120s
ok github.com/carlosframework/platform/internal/gate 1.654s
ok github.com/carlosframework/platform/internal/genselect 0.014s
ok github.com/carlosframework/platform/internal/geo 0.014s
ok github.com/carlosframework/platform/internal/geoip 0.029s
ok github.com/carlosframework/platform/internal/host 96.026s
ok github.com/carlosframework/platform/internal/hostactors 0.451s
ok github.com/carlosframework/platform/internal/hostbox 2.159s
ok github.com/carlosframework/platform/internal/hostclaim 0.491s
ok github.com/carlosframework/platform/internal/hostevidence 0.014s
ok github.com/carlosframework/platform/internal/hostexec 4.893s
ok github.com/carlosframework/platform/internal/hostfence 0.044s
ok github.com/carlosframework/platform/internal/hostfiles 10.581s
ok github.com/carlosframework/platform/internal/hostfloor 0.014s
ok github.com/carlosframework/platform/internal/hostguard 1.004s
ok github.com/carlosframework/platform/internal/hostops 1.432s
? github.com/carlosframework/platform/internal/hostops/fencerelease [no test files]
ok github.com/carlosframework/platform/internal/hostops/internal/fencerelease 0.014s
ok github.com/carlosframework/platform/internal/hostquarantine 0.015s
ok github.com/carlosframework/platform/internal/hostrelocate 0.470s
ok github.com/carlosframework/platform/internal/hostresolve 0.140s
ok github.com/carlosframework/platform/internal/hostwrite 0.014s
ok github.com/carlosframework/platform/internal/infraview 0.014s
ok github.com/carlosframework/platform/internal/instancectl 1.812s
ok github.com/carlosframework/platform/internal/instanceexec 2.714s
ok github.com/carlosframework/platform/internal/instancefs 0.014s
ok github.com/carlosframework/platform/internal/instancekey 0.015s
ok github.com/carlosframework/platform/internal/instanceupdates 0.023s
ok github.com/carlosframework/platform/internal/layout 1.395s
ok github.com/carlosframework/platform/internal/lease 9.473s
ok github.com/carlosframework/platform/internal/lease/leasetest 0.005s
ok github.com/carlosframework/platform/internal/ledger 0.011s
ok github.com/carlosframework/platform/internal/mail 0.154s
ok github.com/carlosframework/platform/internal/maintenance 0.079s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.558s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.080s
ok github.com/carlosframework/platform/internal/metrics 0.005s
ok github.com/carlosframework/platform/internal/moves 21.062s
ok github.com/carlosframework/platform/internal/moves/archive 0.773s
ok github.com/carlosframework/platform/internal/objstore 0.166s
ok github.com/carlosframework/platform/internal/ops 0.163s
ok github.com/carlosframework/platform/internal/park 0.209s
ok github.com/carlosframework/platform/internal/privatefile 0.003s
ok github.com/carlosframework/platform/internal/provision 13.270s
ok github.com/carlosframework/platform/internal/publicrelease 0.006s
? github.com/carlosframework/platform/internal/regionname [no test files]
ok github.com/carlosframework/platform/internal/registry 4.057s
ok github.com/carlosframework/platform/internal/relay 7.236s
ok github.com/carlosframework/platform/internal/release 19.400s
ok github.com/carlosframework/platform/internal/rollout 3.755s
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.007s
ok github.com/carlosframework/platform/internal/rootstate 0.025s
ok github.com/carlosframework/platform/internal/s3compat 0.005s
ok github.com/carlosframework/platform/internal/safefile 0.004s
ok github.com/carlosframework/platform/internal/schedule 2.275s
ok github.com/carlosframework/platform/internal/sim/hostlife 9.602s
ok github.com/carlosframework/platform/internal/sim/litestreamfacts 4.082s
ok github.com/carlosframework/platform/internal/status 0.038s
ok github.com/carlosframework/platform/internal/steering 0.006s
ok github.com/carlosframework/platform/internal/stopevidence 0.189s
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.003s
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.564s
=== 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.023s
=== RUN TestT4StorageProbeMinio
2026/10/13 20:09:47 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE2FBC5DB3BA47, 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..baadd327a 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -1014,6 +1014,9 @@ 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{} |
| + |
| cfg Config |
| backend Backend |
| due DueStore |
| @@ -1272,6 +1275,25 @@ func New(cfg Config, backend Backend, due DueStore) *Service { |
| return a |
| } |
| +func (a *Service) awaitStartup(ctx context.Context, deadline <-chan time.Time) error { |
| + if a.Startup == nil { |
| + return nil |
| + } |
| + select { |
| + case <-a.Startup: |
| + if err := a.lifeCtx.Err(); err != nil { |
| + return err |
| + } |
| + return ctx.Err() |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + case <-a.lifeCtx.Done(): |
| + return context.Canceled |
| + 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 +1306,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 +1449,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 +2950,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 +4148,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 +4166,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 +4826,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 +4841,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 +4981,15 @@ 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) |
| + } |
| + } |
| + |
| a.mu.Lock() |
| fenced := a.fences[rt.Host] != 0 |
| a.mu.Unlock() |
| @@ -4995,40 +5048,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..a4ddf36bc |
| --- /dev/null |
| +++ b/internal/activator/startup_test.go |
| @@ -0,0 +1,54 @@ |
| +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") |
| + } |
| +} |
diff --git a/internal/edge/door.go b/internal/edge/door.go| index c9438f5bb..fd3c6e0c2 100644 |
| --- a/internal/edge/door.go |
| +++ b/internal/edge/door.go |
| @@ -236,14 +236,25 @@ func (e *Edge) armDoor(ctx context.Context) error { |
| return why |
| } |
| keys := newDoorKeyMirror(e.staticStore, e.staticPub) |
| - rctx, cancel := context.WithTimeout(ctx, doorKeyTimeout) |
| - if err := keys.Refresh(rctx); err != nil { |
| - log.Printf("edge: first door-key refresh failed: %v. Every door key is refused until a later poll lands", err) |
| + refresh := func() { |
| + rctx, cancel := context.WithTimeout(ctx, doorKeyTimeout) |
| + if err := keys.Refresh(rctx); err != nil { |
| + log.Printf("edge: first door-key refresh failed: %v. Every door key is refused until a later poll lands", err) |
| + } |
| + cancel() |
| + } |
| + if e.early == nil { |
| + refresh() |
| } |
| - cancel() |
| + |
| e.doorKeys = keys |
| stop := make(chan struct{}) |
| - go keys.run(stop) |
| + go func() { |
| + if e.early != nil { |
| + refresh() |
| + } |
| + keys.run(stop) |
| + }() |
| closeSrv, err := e.openDoor(e.doorPath) |
| if err != nil { |
| close(stop) |
diff --git a/internal/edge/early.go b/internal/edge/early.go| new file mode 100644 |
| index 000000000..05aa55e1c |
| --- /dev/null |
| +++ b/internal/edge/early.go |
| @@ -0,0 +1,120 @@ |
| +package edge |
| + |
| +import ( |
| + "context" |
| + "crypto/tls" |
| + "log" |
| + "net" |
| + "net/http" |
| + "sync" |
| + "time" |
| +) |
| + |
| +// EarlyListeners binds before agent setup, which can scan installations and |
| +// contact other services. Only the local update probe bypasses setup. |
| +type EarlyListeners struct { |
| + http, https *http.Server |
| + httpReady, httpsReady chan struct{} |
| + httpHandler, httpsHandler http.Handler |
| + tlsConfig *tls.Config |
| + once sync.Once |
| +} |
| + |
| +func ListenEarly(args []string) (*EarlyListeners, error) { |
| + f := parseRunFlags(args) |
| + return listenEarly(*f.httpAddr, *f.httpsAddr, *f.dev) |
| +} |
| + |
| +func listenEarly(httpAddr, httpsAddr string, dev bool) (*EarlyListeners, error) { |
| + l := &EarlyListeners{httpReady: make(chan struct{}), httpsReady: make(chan struct{})} |
| + httpListener, err := net.Listen("tcp", httpAddr) |
| + if err != nil { |
| + return nil, err |
| + } |
| + var httpsListener net.Listener |
| + if !dev { |
| + httpsListener, err = net.Listen("tcp", httpsAddr) |
| + if err != nil { |
| + httpListener.Close() |
| + return nil, err |
| + } |
| + } |
| + l.http = &http.Server{Addr: httpListener.Addr().String(), Handler: http.HandlerFunc(l.serveHTTP), ReadHeaderTimeout: 10 * time.Second} |
| + go func() { |
| + if err := l.http.Serve(httpListener); err != nil && err != http.ErrServerClosed { |
| + log.Fatalf("edge: http listener %s: %v", httpAddr, err) |
| + } |
| + }() |
| + if httpsListener != nil { |
| + l.https = &http.Server{ |
| + Addr: httpsListener.Addr().String(), |
| + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + l.httpsHandler.ServeHTTP(w, r) |
| + }), |
| + ReadHeaderTimeout: 10 * time.Second, |
| + TLSConfig: &tls.Config{GetConfigForClient: func(hello *tls.ClientHelloInfo) (*tls.Config, error) { |
| + ctx, cancel := context.WithTimeout(hello.Context(), 30*time.Second) |
| + defer cancel() |
| + select { |
| + case <-l.httpsReady: |
| + return l.tlsConfig, nil |
| + case <-ctx.Done(): |
| + return nil, ctx.Err() |
| + } |
| + }}, |
| + } |
| + go func() { |
| + if err := l.https.ServeTLS(httpsListener, "", ""); err != nil && err != http.ErrServerClosed { |
| + log.Fatalf("edge: https listener %s: %v", httpsAddr, err) |
| + } |
| + }() |
| + } |
| + return l, nil |
| +} |
| + |
| +func (l *EarlyListeners) Option() RunOption { return func(e *Edge) { e.early = l } } |
| + |
| +func (l *EarlyListeners) serveHTTP(w http.ResponseWriter, r *http.Request) { |
| + select { |
| + case <-l.httpReady: |
| + l.httpHandler.ServeHTTP(w, r) |
| + return |
| + default: |
| + } |
| + host, _, err := net.SplitHostPort(r.Host) |
| + if err != nil { |
| + host = r.Host |
| + } |
| + peer, _, _ := net.SplitHostPort(r.RemoteAddr) |
| + if net.ParseIP(host).IsLoopback() && net.ParseIP(peer).IsLoopback() && r.URL.Path == "/" && r.Method == http.MethodGet { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) |
| + defer cancel() |
| + select { |
| + case <-l.httpReady: |
| + l.httpHandler.ServeHTTP(w, r) |
| + case <-ctx.Done(): |
| + http.Error(w, http.StatusText(http.StatusServiceUnavailable), http.StatusServiceUnavailable) |
| + } |
| +} |
| + |
| +func (l *EarlyListeners) publish(h, s http.Handler, cfg *tls.Config) { |
| + l.httpHandler, l.httpsHandler, l.tlsConfig = h, s, cfg |
| + close(l.httpReady) |
| + close(l.httpsReady) |
| +} |
| + |
| +func (l *EarlyListeners) Close() { |
| + l.once.Do(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + defer cancel() |
| + if l.https != nil { |
| + _ = l.https.Shutdown(ctx) |
| + _ = l.https.Close() |
| + } |
| + _ = l.http.Shutdown(ctx) |
| + _ = l.http.Close() |
| + }) |
| +} |
diff --git a/internal/edge/early_test.go b/internal/edge/early_test.go| new file mode 100644 |
| index 000000000..bd94f0a2c |
| --- /dev/null |
| +++ b/internal/edge/early_test.go |
| @@ -0,0 +1,62 @@ |
| +package edge |
| + |
| +import ( |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func TestEarlyListenersAnswerProbeBeforeSetupAndPublishHTTPAndTLS(t *testing.T) { |
| + l, err := listenEarly("127.0.0.1:0", "127.0.0.1:0", false) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close() |
| + client := &http.Client{Timeout: time.Second} |
| + resp, err := client.Get("http://" + l.http.Addr + "/") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + resp.Body.Close() |
| + if resp.StatusCode != http.StatusNotFound { |
| + t.Fatalf("gate: %d", resp.StatusCode) |
| + } |
| + |
| + result := make(chan int, 1) |
| + go func() { |
| + req, _ := http.NewRequest(http.MethodGet, "http://"+l.http.Addr+"/", nil) |
| + req.Host = "console.example.com" |
| + resp, err := client.Do(req) |
| + if err != nil { |
| + result <- 0 |
| + return |
| + } |
| + resp.Body.Close() |
| + result <- resp.StatusCode |
| + }() |
| + select { |
| + case status := <-result: |
| + t.Fatalf("normal request bypassed setup: %d", status) |
| + case <-time.After(25 * time.Millisecond): |
| + } |
| + cert := httptest.NewTLSServer(http.NotFoundHandler()) |
| + defer cert.Close() |
| + h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) }) |
| + l.publish(h, h, cert.TLS.Clone()) |
| + if status := <-result; status != http.StatusNoContent { |
| + t.Fatalf("held HTTP: %d", status) |
| + } |
| + tlsClient := cert.Client() |
| + tlsClient.Timeout = time.Second |
| + resp, err = tlsClient.Get("https://" + l.https.Addr + "/") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + io.Copy(io.Discard, resp.Body) |
| + resp.Body.Close() |
| + if resp.StatusCode != http.StatusNoContent { |
| + t.Fatalf("TLS: %d", resp.StatusCode) |
| + } |
| +} |
diff --git a/internal/edge/edge.go b/internal/edge/edge.go| index ce690d055..6a82106c8 100644 |
| --- a/internal/edge/edge.go |
| +++ b/internal/edge/edge.go |
| @@ -92,6 +92,7 @@ type Activator interface { |
| // its backend. The host→backend map is swapped atomically, so reloads and |
| // zero-downtime deploys never drop a live connection. |
| type Edge struct { |
| + early *EarlyListeners |
| reg *registry.Registry |
| upstreams atomic.Pointer[map[string]upstream] |
| // act, when non-nil, is the hibernation activator: hibernate-flagged |
| @@ -3046,6 +3047,26 @@ func RunWithActivator(args []string, act Activator, cache *Cache) { |
| RunWithActivatorReady(args, act, cache, nil) |
| } |
| +type runFlags struct { |
| + registryPath, acmeCache, httpsAddr, httpAddr, fallbackURL *string |
| + dev *bool |
| +} |
| + |
| +func parseRunFlags(args []string) runFlags { |
| + dataDir := envOr("CARLOS_DATA_DIR", "/data") |
| + fs := flag.NewFlagSet("edge", flag.ExitOnError) |
| + registryPath := fs.String("registry", filepath.Join(dataDir, "registry.db"), "registry SQLite path (default $CARLOS_DATA_DIR/registry.db)") |
| + acmeCache := fs.String("acme-cache", filepath.Join(dataDir, "acme"), "autocert certificate cache dir (default $CARLOS_DATA_DIR/acme)") |
| + httpsAddr := fs.String("https", ":443", "HTTPS listen address") |
| + httpAddr := fs.String("http", ":80", "HTTP listen address (ACME HTTP-01 challenge + redirect to https; in --dev mode, the only listener)") |
| + fallbackURL := fs.String("fallback", "", "redirect requests for unknown hosts here, e.g. an apex marketing site; empty keeps the plain 404") |
| + dev := fs.Bool("dev", false, "plain-HTTP mode: serve routes on --http with no TLS/ACME (local development only)") |
| + fs.Usage = cliflag.Usage(fs, "") |
| + _ = fs.Parse(args) |
| + |
| + return runFlags{registryPath, acmeCache, httpsAddr, httpAddr, fallbackURL, dev} |
| +} |
| + |
| // RunWithActivatorReady is RunWithActivator plus a ready hook: ready (when |
| // non-nil) is called with the fully configured Edge — every EnableXxx done, |
| // tunnel hub armed if configured — immediately before the process blocks on |
| @@ -3060,16 +3081,9 @@ func RunWithActivator(args []string, act Activator, cache *Cache) { |
| // The ready hook is the wrong home for those: it fires once, at the end, for |
| // state a caller wants to REACH. |
| func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready func(*Edge), opts ...RunOption) { |
| + f := parseRunFlags(args) |
| + registryPath, acmeCache, httpsAddr, httpAddr, fallbackURL, dev := f.registryPath, f.acmeCache, f.httpsAddr, f.httpAddr, f.fallbackURL, f.dev |
| dataDir := envOr("CARLOS_DATA_DIR", "/data") |
| - fs := flag.NewFlagSet("edge", flag.ExitOnError) |
| - registryPath := fs.String("registry", filepath.Join(dataDir, "registry.db"), "registry SQLite path (default $CARLOS_DATA_DIR/registry.db)") |
| - acmeCache := fs.String("acme-cache", filepath.Join(dataDir, "acme"), "autocert certificate cache dir (default $CARLOS_DATA_DIR/acme)") |
| - httpsAddr := fs.String("https", ":443", "HTTPS listen address") |
| - httpAddr := fs.String("http", ":80", "HTTP listen address (ACME HTTP-01 challenge + redirect to https; in --dev mode, the only listener)") |
| - fallbackURL := fs.String("fallback", "", "redirect requests for unknown hosts here, e.g. an apex marketing site; empty keeps the plain 404") |
| - dev := fs.Bool("dev", false, "plain-HTTP mode: serve routes on --http with no TLS/ACME (local development only)") |
| - fs.Usage = cliflag.Usage(fs, "") |
| - _ = fs.Parse(args) |
| reg, err := registry.Open(*registryPath) |
| if err != nil { |
| @@ -3144,8 +3158,8 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun |
| stop := make(chan os.Signal, 1) |
| signal.Notify(stop, syscall.SIGTERM, syscall.SIGINT) |
| - // The Edge is fully configured here — every EnableXxx applied and the |
| - // route table rebuilt behind them — and nothing is listening yet. |
| + // The route table and handlers are configured here. An early listener |
| + // still holds normal requests until publish below. |
| if ready != nil { |
| ready(e) |
| } |
| @@ -3157,6 +3171,12 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun |
| // ReadHeaderTimeout guards against Slowloris-style slow-header |
| // connection exhaustion. No ReadTimeout/WriteTimeout — those would |
| // break long-lived proxied WebSockets. |
| + if e.early != nil { |
| + e.early.publish(e, nil, nil) |
| + <-stop |
| + e.early.Close() |
| + return |
| + } |
| srv := &http.Server{Addr: *httpAddr, Handler: e, ReadHeaderTimeout: 10 * time.Second} |
| go func() { |
| log.Printf("edge: DEV MODE — plain HTTP, serving %d route(s) on %s (registry %s)", |
| @@ -3194,37 +3214,45 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun |
| // control/certs/, which is what makes the gate safe to shut. nil when |
| // there is no deployment store — armCertIssuance has then cleared the |
| // gate too, and this box issues and serves entirely for itself. |
| + var afterListening []func() |
| + boot := func(f func()) { |
| + if e.early != nil { |
| + afterListening = append(afterListening, f) |
| + } else { |
| + f() |
| + } |
| + } |
| mirror := e.newCertMirror(localCerts) |
| if mirror != nil { |
| - // One synchronous pass before anything listens: a box that has just |
| - // booted must not spend its first 30 seconds handshaking with nothing, |
| - // and the wildcard manager's first pass (below) reads what this lands |
| - // in the local cache. |
| - ctx, cancel := context.WithTimeout(context.Background(), certMirrorBootTimeout) |
| - if err := mirror.Refresh(ctx); err != nil { |
| - log.Printf("edge: first cert mirror refresh failed: %v — serving from this box's own cache until a later poll lands", err) |
| - } |
| - cancel() |
| + // Legacy agents refresh after publishing the handlers and can serve |
| + // locally cached certificates meanwhile. Other callers keep their |
| + // synchronous first refresh. |
| + boot(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), certMirrorBootTimeout) |
| + if err := mirror.Refresh(ctx); err != nil { |
| + log.Printf("edge: first cert mirror refresh failed: %v — serving from this box's own cache until a later poll lands", err) |
| + } |
| + cancel() |
| + }) |
| } |
| // The home map and the mesh keys (homemap.go), armed beside the cert |
| - // mirror and on the same terms: nil when there is no deployment store to |
| - // mirror from, one synchronous refresh before anything listens so the |
| - // first request does not meet an empty map, then the 30s loop. Stored |
| - // before either listener binds, so ServeHTTP never sees a half-armed edge. |
| + // mirror and on the same terms. Publish the handle before the HTTP |
| + // handler; the first refresh is asynchronous for legacy agents. |
| if hm := e.newHomeMap(); hm != nil { |
| - ctx, cancel := context.WithTimeout(context.Background(), homeMapBootTimeout) |
| - if err := hm.Refresh(ctx); err != nil { |
| - log.Printf("edge: first home map refresh failed: %v — hosts homed on another box are unreachable from here until a later poll lands", err) |
| - } |
| - cancel() |
| + boot(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), homeMapBootTimeout) |
| + if err := hm.Refresh(ctx); err != nil { |
| + log.Printf("edge: first home map refresh failed: %v — hosts homed on another box are unreachable from here until a later poll lands", err) |
| + } |
| + cancel() |
| + }) |
| e.homes.Store(hm) |
| go hm.run(make(chan struct{})) // process-lifetime; never stopped, like the cert mirror's |
| } |
| // The edge-local door (door.go, #175 P7), armed right after the home map |
| - // it dials through: its own key mirror, one synchronous refresh, its |
| - // loop, then the socket. Off, with the reason in the journal, when there |
| + // it dials through: its own key mirror, its loop, then the socket. Off, with the reason in the journal, when there |
| // is no store, no release key or no public address. Closed in the |
| // shutdown below, ahead of the public listeners. |
| _ = e.armDoor(context.Background()) |
| @@ -3264,7 +3292,7 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun |
| // A remote box's edges file landing (or leaving) can add or drop a |
| // parent (#300): re-derive on the home map's poll, not the 12h tick. |
| e.homes.Load().setHostsWake(w.refresh) |
| - w.refresh() |
| + boot(w.refresh) |
| go w.run() |
| log.Printf("edge: wildcard certs on (static %v, apps domain %q, delegation zone %q)", |
| splitDomains(os.Getenv("CARLOS_WILDCARD_DOMAINS")), os.Getenv("PLATFORM_APPS_DOMAIN"), w.delegationZone) |
| @@ -3320,6 +3348,19 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun |
| acmeAndFallback = mirror.challengeHandler(acmeAndFallback) |
| } |
| + if e.early != nil { |
| + e.early.publish(acmeAndFallback, e, tlsCfg) |
| + for _, f := range afterListening { |
| + go f() |
| + } |
| + <-stop |
| + if e.closeDoor != nil { |
| + e.closeDoor() |
| + } |
| + e.early.Close() |
| + return |
| + } |
| + |
| // ReadHeaderTimeout guards the public front door against Slowloris-style |
| // slow-header connection exhaustion. No ReadTimeout/WriteTimeout — those |
| // would break long-lived proxied WebSockets. |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 3d8fdf552..8620001c5 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -213,6 +213,15 @@ func RunAgent(args []string) { runAgent(args, productionClaimFactory) } |
| func runAgent(args []string, buildClaims ClaimFactory) { |
| agentCtx, cancelAgent := context.WithCancel(context.Background()) |
| defer cancelAgent() |
| + var listeners *edge.EarlyListeners |
| + if !hostwrite.Enabled(os.Getenv) { |
| + var err error |
| + listeners, err = edge.ListenEarly(args) |
| + if err != nil { |
| + log.Fatal(err) |
| + } |
| + defer listeners.Close() |
| + } |
| claimCtx, err := claimEntryContext(agentCtx, os.Getenv, openDeployStore) |
| if err != nil { |
| log.Fatalf("agent: host claims: %v", err) |
| @@ -375,6 +384,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 +533,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 +547,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(ctx 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(ctx, 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(ctx, 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 +595,8 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| Slot: svc.PrepareScheduledToken, |
| Publish: armedLayoutPass.Store, |
| Start: svc.Start, |
| - }.Run(context.Background()) |
| + }.Run(agentCtx) |
| 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 +627,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 <-agentCtx.Done(): |
| + return |
| + } |
| routes, err := reg.List() |
| if err != nil { |
| log.Printf("agent: worker boot pass: list routes: %v", err) |
| @@ -618,68 +641,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 +653,145 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| log.Printf("agent: regional claims shutdown: %v", err) |
| } |
| }() |
| + var stopHousekeeper func() |
| + dutiesDone := make(chan struct{}) |
| + runDuties := func() { |
| + defer close(dutiesDone) |
| + select { |
| + case <-started: |
| + case <-agentCtx.Done(): |
| + return |
| + } |
| + if agentCtx.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(agentCtx) // 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. |
| + 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 publishing the handler, 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{ |
| @@ -778,6 +825,9 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| if tunnelEnvComplete(os.Getenv) { |
| edgeOpts = append(edgeOpts, edge.WithGeoTunnelClient()) |
| } |
| + if listeners != nil { |
| + edgeOpts = append(edgeOpts, listeners.Option()) |
| + } |
| edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) { |
| go tunnelLoop(e) |
| // The edge owns the health probers, so its state table is only |
| @@ -812,9 +862,15 @@ func runAgent(args []string, buildClaims ClaimFactory) { |
| go usageLoop(m.Drain, am) |
| }, edgeOpts...) |
| + cancelAgent() |
| + joinCtx, joinCancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer joinCancel() |
| + joinAgentStartup(joinCtx, startupDone) |
| + dutiesJoined := 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 { |
| + if svc != nil && startupFinished(started) { |
| ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| defer cancel() |
| svc.Shutdown(ctx) |
| @@ -846,7 +902,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 dutiesJoined && 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..111cfe54c 100644 |
| --- a/internal/host/instancekeys_test.go |
| +++ b/internal/host/instancekeys_test.go |
| @@ -304,7 +304,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 +313,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..60ec2ee66 |
| --- /dev/null |
| +++ b/internal/host/startup.go |
| @@ -0,0 +1,24 @@ |
| +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 |
| + } |
| +} |
| + |
| +func startupFinished(done <-chan struct{}) bool { |
| + select { |
| + case <-done: |
| + return true |
| + default: |
| + return false |
| + } |
| +} |
diff --git a/internal/host/startup_test.go b/internal/host/startup_test.go| new file mode 100644 |
| index 000000000..63875ea44 |
| --- /dev/null |
| +++ b/internal/host/startup_test.go |
| @@ -0,0 +1,241 @@ |
| +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 |
| + 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() { |
| + return nil, errors.New("wake preceded recovery") |
| + } |
| + 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 atomic.Bool |
| + backend := &startupBackend{fakeWakeBackend: fakeWakeBackend{inst: inst}, recovered: &recovered} |
| + 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: svc.Start, 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") |
| + } |
| +} |
diff --git a/internal/provision/plan14_root_reads_test.go b/internal/provision/plan14_root_reads_test.go| index e51bd4792..5c8cb566f 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 = "48e04b1e94edbd7c05ed68b0e6f8f060934365ff7fcc8c91bbbd183da69be0c2" |
| 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..48803e471 |
| --- /dev/null |
| +++ b/internal/provision/testdata/plan14-root-reads-hotfix281.tsv |
| @@ -0,0 +1,23 @@ |
| +# Hotfix 281: reviewed read-preserving startup changes. Earlier overlays remain unchanged. |
| +@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 aa4915f25d4d2cb1bc7f896b9808ecfd25fe9c4b5cbb050976e896947a1476aa 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. The same route fields and sources are read. registry:Hibernate@151,registry:Hibernate@788,registry:Host@397,registry:Kind@126,registry:Kind@762,whole:Route@1907,whole:Route@2876,whole:Route@2936,whole:Route@347 44becc931aec04495ff7daae2154fbbf500c4e6d4df65b328a790e33cfa3a0ac |
| +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. The same route fields and sources are read. registry:Hibernate@151,registry:Hibernate@788,registry:Host@397,registry:Kind@126,registry:Kind@762,whole:Route@1907,whole:Route@2876,whole:Route@2936,whole:Route@347 44becc931aec04495ff7daae2154fbbf500c4e6d4df65b328a790e33cfa3a0ac |
| +@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 |
| +@legacy 910d75544b6bbcfffefa1958ca9170599ddec521faacf0ab1ac4d71fe6e345dd internal/edge/edge.go RunWithActivatorReady legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Flag parsing moved to parseRunFlags; the same registry and data-directory sources are used. Initial mirrors run asynchronously only with early listeners. data-literal@297,data-literal@316,file:Open@339 f061377870ded091f7c0de63f17733e48edd9c9a063025898521497ad13bdc42 |
| +internal/edge/edge.go RunWithActivatorReady legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Flag parsing moved to parseRunFlags; the same registry and data-directory sources are used. Initial mirrors run asynchronously only with early listeners. data-literal@297,data-literal@316,file:Open@339 f061377870ded091f7c0de63f17733e48edd9c9a063025898521497ad13bdc42 |
| +@legacy df3dc08211aba4445f61ccaf6fda66ff445ef32a31bffbeec8536a84c11f238d internal/edge/edge.go parseRunFlags legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Extracted unchanged flags and data-directory defaults from RunWithActivatorReady; no new source or authority. data-literal@218,data-literal@351,data-literal@63,data-literal@82 f0109557c4e84d4b52c3dc73ef31a7da34ecbbe57fbddb2f3ae6358557acb5e2 |
| +internal/edge/edge.go parseRunFlags legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Extracted unchanged flags and data-directory defaults from RunWithActivatorReady; no new source or authority. data-literal@218,data-literal@351,data-literal@63,data-literal@82 f0109557c4e84d4b52c3dc73ef31a7da34ecbbe57fbddb2f3ae6358557acb5e2 |
| +internal/host/agent.go runAgent authority B,C,D,F rev3=census-ruling-hotfix281;basis=Startup ordering and lifetime changed; registry, route, data-directory and backend inputs are unchanged. Existing B,C,D,F authority classification is retained. file:Exec@13711,file:Open@768,registry:App@10102,whole:Route@10094,whole:Route@21421 8fb88c131c968692ad2bf2967e9968aff19dc7caa195e0a56df67b2fc4eb85ce |
| +@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 8f772c7c43dec1102a01dbe4a945b8eb0c751f0e2f6efe05dc437e13b21daede internal/host/layoutleg.go activatorStart.Run legacy accepted-legacy accepted-legacy;rev3=census-ruling-hotfix281;basis=Legacy sweep, recovery and key pass run under the agent lifetime in the background. 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 under the agent lifetime in the background. The same route list and callbacks are used; claim boxes retain Start alone. whole:Route@803 e93d0c33f6ac39f9c644b7bfa2463c30a9980dfd0e23a515ba87f356e8899c57 |
| +@coverage internal/activator ccf2519fb636a45e909e302662a619d14a032a5e191df87971b413fecb87aabb |
| +@coverage internal/edge d5d3002fd6f1f920aeb983234d7ca8c4f83ad353fdeae76e37fc375ea8def822 |
| +@coverage internal/host 2aaede4445eb11839322b809d81e5b34eb1f610debf57db03a4e7027de95185a |