amadan
host: end-to-end startup test on a 300-instance box (#281 ask 2)
The hotfix for #281 (6dfbff1) moved the sweep, layout recovery, hold restoration, Start and the key pass off the edge's listen path, behind a startup latch. Its own test stubs those steps. This one runs them for real on a box twice the size of the one that failed the gate. The agent is this test binary re-executed: runAgent's legacy-box activatorStart with runAgent's Sweep and Recover closures, then edge.RunWithActivatorReady in production (TLS) mode, then runAgent's shutdown. Instance control is the real instancectl.Server, made cold through its Systemctl seam. Recovery runs over real journals left at three crash points, hold restoration over real markers whose reason persists. The activator backend and the key writer are the only fakes. Three scenarios, run in parallel: - cold control: the gate's probe (GET / on :80, any answer) is answered in ~40 ms with 300 routes; a wake, a held host and a host mid-move are all held at the latch; then the wake is served, the held host answers maintenance and is never started, and the moved host answers 503 rather than being proxied to its old socket, and its retry is served at the new one. - overrun: control never warms; a wake gets the wake-timeout 503 at the latch's 30 s bound. - shutdown mid-startup: SIGTERM with a wake held; the request gets a 503, startup work joins in under a millisecond, and no startup, activator or fake-instance goroutine outlives shutdown. Linux only (unix sockets, the layout pass's openat2). About 31 s wall.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 file changed,
+1159
−0
internal/host/startup_largebox_linux_test.go+1159 −0
CI passed — run details
CI log
=== gate ===
go: downloading golang.org/x/crypto v0.52.0
go: downloading github.com/superfly/ltx v0.5.1
go: downloading github.com/oschwald/maxminddb-golang v1.13.1
go: downloading github.com/aws/aws-sdk-go-v2 v1.44.0
go: downloading github.com/aws/aws-sdk-go-v2/service/s3 v1.105.0
go: downloading modernc.org/sqlite v1.55.0
go: downloading github.com/aws/aws-sdk-go-v2/credentials v1.19.6
go: downloading github.com/aws/aws-sdk-go-v2/service/sts v1.45.6
go: downloading github.com/benbjohnson/litestream v0.5.14
go: downloading github.com/aws/aws-sdk-go-v2/config v1.32.6
go: downloading github.com/aws/aws-sdk-go-v2/service/route53 v1.64.1
go: downloading github.com/aws/aws-sdk-go-v2/service/ec2 v1.324.0
go: downloading github.com/aws/aws-sdk-go-v2/service/sns v1.43.0
go: downloading golang.org/x/sys v0.46.0
go: downloading github.com/aws/aws-sdk-go-v2/service/acm v1.45.0
go: downloading github.com/aws/aws-sdk-go-v2/service/cloudwatch v1.68.0
go: downloading github.com/aws/aws-sdk-go-v2/service/iam v1.58.2
go: downloading github.com/aws/smithy-go v1.28.1
go: downloading github.com/aws/aws-sdk-go-v2/service/resourcegroupstaggingapi v1.37.0
go: downloading github.com/klauspost/compress v1.18.0
go: downloading github.com/aws/aws-sdk-go-v2/service/sesv2 v1.66.6
go: downloading golang.org/x/net v0.55.0
go: downloading github.com/pion/logging v0.2.4
go: downloading github.com/pion/stun/v3 v3.0.1
go: downloading github.com/pion/turn/v4 v4.1.4
go: downloading github.com/aws/aws-sdk-go-v2/service/costexplorer v1.67.5
go: downloading github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.16
go: downloading github.com/aws/aws-sdk-go-v2/internal/ini v1.8.4
go: downloading github.com/aws/aws-sdk-go-v2/service/signin v1.0.4
go: downloading github.com/aws/aws-sdk-go-v2/service/sso v1.30.8
go: downloading github.com/aws/aws-sdk-go-v2/service/ssooidc v1.35.12
go: downloading github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.40
go: downloading github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.38
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.19
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.40
go: downloading github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23
go: downloading github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31
go: downloading github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.40
go: downloading github.com/pierrec/lz4/v4 v4.1.22
go: downloading github.com/pion/randutil v0.1.0
go: downloading github.com/pion/transport/v4 v4.0.1
go: downloading github.com/pion/transport/v3 v3.0.8
go: downloading github.com/pion/dtls/v3 v3.0.7
go: downloading github.com/prometheus/client_golang v1.17.0
go: downloading golang.org/x/sync v0.21.0
go: downloading github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.20.18
go: downloading github.com/lmittmann/tint v1.1.3
go: downloading github.com/mattn/go-isatty v0.0.20
go: downloading github.com/wlynxg/anet v0.0.5
go: downloading github.com/cespare/xxhash/v2 v2.3.0
go: downloading github.com/beorn7/perks v1.0.1
go: downloading github.com/prometheus/common v0.45.0
go: downloading google.golang.org/protobuf v1.36.10
go: downloading github.com/prometheus/client_model v0.5.0
go: downloading github.com/prometheus/procfs v0.12.0
go: downloading github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0
go: downloading golang.org/x/text v0.37.0
go: downloading modernc.org/libc v1.74.1
go: downloading modernc.org/memory v1.11.0
go: downloading github.com/google/uuid v1.6.0
go: downloading github.com/dustin/go-humanize v1.0.1
go: downloading modernc.org/mathutil v1.7.1
go: downloading github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec
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 338.240s
? github.com/carlosframework/platform/cmd/changelog [no test files]
ok github.com/carlosframework/platform/cmd/console 3.206s
ok github.com/carlosframework/platform/cmd/moveprobe 93.254s
ok github.com/carlosframework/platform/cmd/zddprobe 0.387s
ok github.com/carlosframework/platform/internal/activator 227.411s
ok github.com/carlosframework/platform/internal/agentsock 0.018s
ok github.com/carlosframework/platform/internal/alerting 0.003s
ok github.com/carlosframework/platform/internal/analytics 0.021s
ok github.com/carlosframework/platform/internal/appconfig 2.338s
ok github.com/carlosframework/platform/internal/apperr 0.076s
ok github.com/carlosframework/platform/internal/applog 0.128s
ok github.com/carlosframework/platform/internal/awsinv 0.004s
ok github.com/carlosframework/platform/internal/boxchoice 0.007s
ok github.com/carlosframework/platform/internal/boxid 0.826s
? 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.009s
ok github.com/carlosframework/platform/internal/caps 0.011s
ok github.com/carlosframework/platform/internal/certwarm 0.004s
ok github.com/carlosframework/platform/internal/cgroup 0.699s
ok github.com/carlosframework/platform/internal/changelog 0.300s
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 365.086s
? 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.257s
ok github.com/carlosframework/platform/internal/dbdoor/dbdoortest 0.009s
ok github.com/carlosframework/platform/internal/dbverify 0.004s
ok github.com/carlosframework/platform/internal/dnsrec 0.002s
ok github.com/carlosframework/platform/internal/docsite 0.005s
ok github.com/carlosframework/platform/internal/domains 2.094s
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.563s
? 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.123s
ok github.com/carlosframework/platform/internal/gate 1.655s
ok github.com/carlosframework/platform/internal/genselect 0.015s
ok github.com/carlosframework/platform/internal/geo 0.020s
ok github.com/carlosframework/platform/internal/geoip 0.037s
ok github.com/carlosframework/platform/internal/host 125.840s
ok github.com/carlosframework/platform/internal/hostactors 0.388s
ok github.com/carlosframework/platform/internal/hostbox 2.193s
ok github.com/carlosframework/platform/internal/hostclaim 0.502s
ok github.com/carlosframework/platform/internal/hostevidence 0.005s
ok github.com/carlosframework/platform/internal/hostexec 5.372s
ok github.com/carlosframework/platform/internal/hostfence 0.029s
ok github.com/carlosframework/platform/internal/hostfiles 10.923s
ok github.com/carlosframework/platform/internal/hostfloor 0.012s
ok github.com/carlosframework/platform/internal/hostguard 1.005s
ok github.com/carlosframework/platform/internal/hostops 1.425s
? github.com/carlosframework/platform/internal/hostops/fencerelease [no test files]
ok github.com/carlosframework/platform/internal/hostops/internal/fencerelease 0.010s
ok github.com/carlosframework/platform/internal/hostquarantine 0.010s
ok github.com/carlosframework/platform/internal/hostrelocate 0.436s
ok github.com/carlosframework/platform/internal/hostresolve 0.151s
ok github.com/carlosframework/platform/internal/hostwrite 0.009s
ok github.com/carlosframework/platform/internal/infraview 0.009s
ok github.com/carlosframework/platform/internal/instancectl 1.813s
ok github.com/carlosframework/platform/internal/instanceexec 2.711s
ok github.com/carlosframework/platform/internal/instancefs 0.009s
ok github.com/carlosframework/platform/internal/instancekey 0.011s
ok github.com/carlosframework/platform/internal/instanceupdates 0.010s
ok github.com/carlosframework/platform/internal/layout 1.432s
ok github.com/carlosframework/platform/internal/lease 9.518s
ok github.com/carlosframework/platform/internal/lease/leasetest 0.013s
ok github.com/carlosframework/platform/internal/ledger 0.020s
ok github.com/carlosframework/platform/internal/mail 0.158s
ok github.com/carlosframework/platform/internal/maintenance 0.083s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.536s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.069s
ok github.com/carlosframework/platform/internal/metrics 0.004s
ok github.com/carlosframework/platform/internal/moves 21.634s
ok github.com/carlosframework/platform/internal/moves/archive 0.782s
ok github.com/carlosframework/platform/internal/objstore 0.161s
ok github.com/carlosframework/platform/internal/ops 0.176s
ok github.com/carlosframework/platform/internal/park 0.189s
ok github.com/carlosframework/platform/internal/privatefile 0.003s
ok github.com/carlosframework/platform/internal/provision 29.958s
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.376s
ok github.com/carlosframework/platform/internal/relay 7.287s
ok github.com/carlosframework/platform/internal/release 19.863s
ok github.com/carlosframework/platform/internal/rollout 3.765s
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.005s
ok github.com/carlosframework/platform/internal/schedule 2.278s
ok github.com/carlosframework/platform/internal/sim/hostlife 11.216s
ok github.com/carlosframework/platform/internal/sim/litestreamfacts 3.934s
ok github.com/carlosframework/platform/internal/status 0.037s
ok github.com/carlosframework/platform/internal/steering 0.005s
ok github.com/carlosframework/platform/internal/stopevidence 0.198s
ok github.com/carlosframework/platform/internal/svcrole 0.002s
ok github.com/carlosframework/platform/internal/telemetry 0.004s
ok github.com/carlosframework/platform/internal/termfence 0.004s
ok github.com/carlosframework/platform/internal/tunnelclient 0.004s
ok github.com/carlosframework/platform/internal/unixdial 0.002s
ok github.com/carlosframework/platform/internal/updatewords 0.005s
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.565s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
go: downloading github.com/ncruces/go-strftime v1.0.0
--- shard 1 of 5 ---
cross-compiling darwin/amd64
go: downloading github.com/ncruces/go-strftime v1.0.0
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
go: downloading github.com/ncruces/go-strftime v1.0.0
--- shard 4 of 5 ---
cross-compiling windows/arm64
go: downloading github.com/ncruces/go-strftime v1.0.0
=== 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 22:03:37 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE35F2768F0053, 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.00s)
PASS
ok github.com/carlosframework/platform/internal/host 0.021s
minio contract: PASS
diff --git a/internal/host/startup_largebox_linux_test.go b/internal/host/startup_largebox_linux_test.go| new file mode 100644 |
| index 000000000..7e893e684 |
| --- /dev/null |
| +++ b/internal/host/startup_largebox_linux_test.go |
| @@ -0,0 +1,1159 @@ |
| +//go:build linux |
| + |
| +package host |
| + |
| +// The large-box startup, end to end (amadan #281, ask 2). On 2026-10-13 a |
| +// box with 142 instances failed the 20 s update gate twice: the updater |
| +// restarts carlos-instance-control just before the new edge, and the |
| +// agent's startup sweep waited on that cold control service before the |
| +// edge listened. The hotfix (6dfbff1) moved the sweep, layout recovery, |
| +// hold restoration, Start and the key pass off the listen path, behind a |
| +// startup latch that instance admission waits on. |
| +// |
| +// TestStartupServesWhileSweepAndRecoveryWait proves the latch with the |
| +// steps stubbed. This test runs them for real on a box larger than the |
| +// one that failed: |
| +// |
| +// - the agent is a child process (this test binary, re-executed), so |
| +// what is timed is a process start up to the update gate's own probe; |
| +// - instance control is the real instancectl.Server over a unix socket, |
| +// made cold by its systemctl seam: every call waits until the test |
| +// warms it, then each of the box's 300 leftover units takes a moment |
| +// to stop, as a restarted control service does; |
| +// - the sweep is the real client call, layout recovery and hold |
| +// restoration are the real pass over real journals, hold markers and |
| +// files in a temp data dir, Start is the real activator's, and the |
| +// edge is edge.RunWithActivatorReady in its production (TLS) mode; |
| +// - the activator's backend is a fake that serves an app on the row's |
| +// socket. It and the key pass's writer are the only fakes in the |
| +// agent, and neither is a startup step. |
| +// |
| +// The child's startup is runAgent's legacy-box sequence: the same |
| +// activatorStart, with runAgent's Sweep and Recover closures. runAgent |
| +// itself cannot run in a test (its control socket path is a constant, |
| +// it builds AWS clients, and it log.Fatals), so the child assembles the |
| +// same pieces in the same order. Two box facts differ from production |
| +// and are set on the pass: its socket directory is the fixture's, not |
| +// /run/carlos, and it has no carlos-sockets group. |
| +// |
| +// The child reports what it does as timestamped lines in an events file; |
| +// the test reads them to order "a request reached the activator", "the |
| +// latch opened" and "the backend acquired a host" across the two |
| +// processes. |
| + |
| +import ( |
| + "bufio" |
| + "context" |
| + "crypto/ecdsa" |
| + "crypto/elliptic" |
| + "crypto/rand" |
| + "crypto/tls" |
| + "crypto/x509" |
| + "crypto/x509/pkix" |
| + "encoding/json" |
| + "encoding/pem" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "log" |
| + "math/big" |
| + "net" |
| + "net/http" |
| + "os" |
| + "os/exec" |
| + "path/filepath" |
| + "runtime" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "syscall" |
| + "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/instancefs" |
| + "github.com/carlosframework/platform/internal/layout" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/version" |
| +) |
| + |
| +const ( |
| + // largeBoxAgentEnv names the child's config file. Set, the test binary |
| + // is the agent and nothing else. |
| + largeBoxAgentEnv = "CARLOS_TEST_LARGEBOX_AGENT" |
| + |
| + largeBoxRoutes = 300 // the box that failed had 142 |
| + largeBoxJournals = 8 // hosts 0..7: a layout move was in flight at the restart |
| + largeBoxHolds = 4 // hosts 8..11: held down after a refused move |
| + largeBoxWakeHost = 12 // an ordinary sleeping instance |
| + |
| + // largeBoxListenBound is how soon the gate's probe must be answered |
| + // after the agent process starts. The gate allows 20 s; the edge's own |
| + // startup on this box takes a fraction of a second. |
| + largeBoxListenBound = 3 * time.Second |
| + // largeBoxCold is how long instance control stays cold in the scenario |
| + // where it warms: longer than the listen bound, so the gate is answered |
| + // while the sweep is provably still waiting. |
| + largeBoxCold = 4 * time.Second |
| + // largeBoxUnitStop is one leftover unit's stop once control is warm. |
| + largeBoxUnitStop = 5 * time.Millisecond |
| + // largeBoxLatchBound is the activator's wakeWaitBound, which the |
| + // startup latch shares: a wake held longer is answered 503. |
| + largeBoxLatchBound = 30 * time.Second |
| + // largeBoxJoinBound is runAgent's join on the startup work after the |
| + // edge returns, and largeBoxDrain the edge's drain of in-flight |
| + // requests before it returns. |
| + largeBoxJoinBound = 5 * time.Second |
| + largeBoxDrain = 10 * time.Second |
| +) |
| + |
| +func largeBoxHost(i int) string { return fmt.Sprintf("app-%03d.example.com", i) } |
| + |
| +// largeBoxConfig is what the child agent is told. |
| +type largeBoxConfig struct { |
| + DataDir, SocketDir, StoreDir, AcmeDir string |
| + Control, Events, HTTP, HTTPS string |
| +} |
| + |
| +func TestStartupLargeBoxEndToEnd(t *testing.T) { |
| + if path := os.Getenv(largeBoxAgentEnv); path != "" { |
| + os.Exit(runLargeBoxAgent(path)) |
| + } |
| + t.Run("cold instance control", func(t *testing.T) { |
| + t.Parallel() |
| + testLargeBoxColdControl(t) |
| + }) |
| + t.Run("startup overruns the latch", func(t *testing.T) { |
| + t.Parallel() |
| + testLargeBoxOverrun(t) |
| + }) |
| + t.Run("shutdown mid-startup", func(t *testing.T) { |
| + t.Parallel() |
| + testLargeBoxShutdownMidStartup(t) |
| + }) |
| +} |
| + |
| +// testLargeBoxColdControl is the 2026-10-13 shape: control restarted just |
| +// before the agent, and slow to answer. The edge must answer the gate's |
| +// probe while the sweep waits, hold every instance request at the latch |
| +// until sweep, recovery and hold restoration are done, and then serve each |
| +// host only where it is safe to. |
| +func testLargeBoxColdControl(t *testing.T) { |
| + b := newLargeBox(t) |
| + a := b.startAgent(t) |
| + listen := a.waitListening(t) |
| + t.Logf("gate probe answered %s after the agent started, %d routes, instance control cold", listen, largeBoxRoutes) |
| + if _, ok := a.event("ready"); ok { |
| + t.Fatal("the startup latch opened while instance control was still cold") |
| + } |
| + |
| + wake, held, moved := largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxJournals), largeBoxHost(0) |
| + flatMoved := filepath.Join(b.socketDir, moved+".sock") |
| + wakeAns, heldAns, movedAns := b.fire(wake), b.fire(held), b.fire(moved) |
| + enterWake := a.waitEvent(t, 10*time.Second, "enter", wake) |
| + a.waitEvent(t, 10*time.Second, "enter", held) |
| + enterMoved := a.waitEvent(t, 10*time.Second, "enter", moved) |
| + if enterMoved.arg(1) != flatMoved { |
| + t.Fatalf("the edge routed %s to %q before recovery, want its pre-move socket %q", moved, enterMoved.arg(1), flatMoved) |
| + } |
| + // Every request is now inside the activator, and startup is not done. |
| + time.Sleep(200 * time.Millisecond) |
| + for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns} { |
| + select { |
| + case got := <-ch: |
| + t.Fatalf("%s answered before startup finished: %d %q %v", host, got.status, got.body, got.err) |
| + default: |
| + } |
| + } |
| + if evs := a.events("acquire"); len(evs) != 0 { |
| + t.Fatalf("a wake was admitted while instance control was cold: %v", evs) |
| + } |
| + |
| + // Instance control warms: the sweep stops the 300 leftovers, then |
| + // recovery, holds, Start, and the latch. |
| + if d := time.Until(a.began.Add(largeBoxCold)); d > 0 { |
| + time.Sleep(d) |
| + } |
| + close(b.control.warm) |
| + ready := a.waitEvent(t, 30*time.Second, "ready") |
| + t.Logf("startup latch opened %s after the agent started (sweep stopped %d units)", ready.at.Sub(a.began), b.control.stops.Load()) |
| + if got := b.control.stops.Load(); got != largeBoxRoutes { |
| + t.Errorf("the sweep stopped %d units, want %d", got, largeBoxRoutes) |
| + } |
| + if !enterWake.at.Before(ready.at) { |
| + t.Fatal("the wake did not arrive before the latch opened; the test proves nothing about the latch") |
| + } |
| + |
| + // The wake was held, then served. |
| + got := b.answer(t, wakeAns) |
| + if got.status != http.StatusOK || got.body != "host="+wake+" socket="+filepath.Join(b.socketDir, wake+".sock") { |
| + t.Fatalf("held wake after startup: %d %q %v", got.status, got.body, got.err) |
| + } |
| + acq := a.waitEvent(t, time.Second, "acquire", wake) |
| + if acq.at.Before(ready.at) { |
| + t.Fatalf("%s was acquired %s before the latch opened", wake, ready.at.Sub(acq.at)) |
| + } |
| + |
| + // The held host was not started: its restored hold answers maintenance. |
| + got = b.answer(t, heldAns) |
| + if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" || got.header.Get("Cache-Control") != "no-store" { |
| + t.Fatalf("held host after startup: %d %q %v %v", got.status, got.body, got.header, got.err) |
| + } |
| + |
| + // The moved host was not served at the socket recovery moved it from: |
| + // the request answers as a wake in progress, and its retry is routed |
| + // afresh to where recovery put it. |
| + got = b.answer(t, movedAns) |
| + if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" { |
| + t.Fatalf("host whose recovery moved it, held through startup: %d %q %v", got.status, got.body, got.err) |
| + } |
| + nested, err := layout.PlanFor(b.dataDir, b.socketDir, moved, layout.ModeNested) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got = b.get(moved) |
| + if got.status != http.StatusOK || got.body != "host="+moved+" socket="+nested.To.Socket { |
| + t.Fatalf("retry of the moved host: %d %q %v", got.status, got.body, got.err) |
| + } |
| + got = b.get(held) |
| + if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" { |
| + t.Fatalf("retry of the held host: %d %q %v", got.status, got.body, got.err) |
| + } |
| + |
| + // The background key pass ran after Start, and its slot refused the |
| + // held hosts. |
| + a.waitEvent(t, 30*time.Second, "startup-done") |
| + keys := a.events("key") |
| + if len(keys) != largeBoxRoutes-largeBoxHolds { |
| + t.Errorf("key pass wrote %d rows, want %d", len(keys), largeBoxRoutes-largeBoxHolds) |
| + } |
| + for _, e := range keys { |
| + if e.at.Before(ready.at) { |
| + t.Errorf("key pass reached %s before Start", e.arg(0)) |
| + } |
| + } |
| + |
| + b.assertNoWrongService(t, a) |
| + b.assertRecovered(t) |
| + |
| + a.signal(t, syscall.SIGTERM) |
| + a.waitExit(t, largeBoxJoinBound+3*time.Second) |
| + a.assertCleanShutdown(t) |
| +} |
| + |
| +// testLargeBoxOverrun: instance control never warms, so startup never |
| +// finishes. A wake must be answered with the wake-timeout 503 at the |
| +// latch's bound: not an error, not a hang, and not a wake. |
| +func testLargeBoxOverrun(t *testing.T) { |
| + b := newLargeBox(t) |
| + a := b.startAgent(t) |
| + listen := a.waitListening(t) |
| + t.Logf("gate probe answered %s after the agent started, instance control never warms", listen) |
| + |
| + wake := largeBoxHost(largeBoxWakeHost) |
| + began := time.Now() |
| + got := b.get(wake) |
| + took := time.Since(began) |
| + t.Logf("wake held %s by a startup that does not finish", took) |
| + if got.err != nil { |
| + t.Fatalf("wake during an overrunning startup: %v", got.err) |
| + } |
| + if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" { |
| + t.Fatalf("wake during an overrunning startup: %d %q %v", got.status, got.body, got.header) |
| + } |
| + if took < largeBoxLatchBound-time.Second || took > largeBoxLatchBound+5*time.Second { |
| + t.Fatalf("wake answered after %s, want the latch bound %s", took, largeBoxLatchBound) |
| + } |
| + if _, ok := a.event("ready"); ok { |
| + t.Fatal("the latch opened with instance control cold") |
| + } |
| + b.assertNoWrongService(t, a) |
| + if evs := a.events("acquire"); len(evs) != 0 { |
| + t.Fatalf("a wake was admitted during an overrunning startup: %v", evs) |
| + } |
| + |
| + // Shutting down with the sweep still waiting: nothing in flight, so the |
| + // edge returns at once and runAgent's join bounds the rest. |
| + at := time.Now() |
| + a.signal(t, syscall.SIGTERM) |
| + a.waitExit(t, largeBoxJoinBound+3*time.Second) |
| + t.Logf("agent exited %s after SIGTERM with the sweep still waiting", time.Since(at)) |
| + a.assertCleanShutdown(t) |
| +} |
| + |
| +// testLargeBoxShutdownMidStartup: SIGTERM with startup still waiting on |
| +// the sweep and a wake held at the latch. The held request is answered |
| +// 503 (never proxied), the background work is joined within runAgent's |
| +// bound, and nothing of startup outlives it. |
| +func testLargeBoxShutdownMidStartup(t *testing.T) { |
| + b := newLargeBox(t) |
| + a := b.startAgent(t) |
| + a.waitListening(t) |
| + |
| + wake := largeBoxHost(largeBoxWakeHost) |
| + ans := b.fire(wake) |
| + a.waitEvent(t, 10*time.Second, "enter", wake) |
| + |
| + at := time.Now() |
| + a.signal(t, syscall.SIGTERM) |
| + // The edge drains in-flight requests for up to largeBoxDrain before it |
| + // returns, and the held wake is in flight until the activator shuts |
| + // down after the join. |
| + a.waitExit(t, largeBoxDrain+largeBoxJoinBound+3*time.Second) |
| + t.Logf("agent exited %s after SIGTERM with a wake held at the latch", time.Since(at)) |
| + got := b.answer(t, ans) |
| + if got.err != nil || got.status != http.StatusServiceUnavailable { |
| + t.Fatalf("wake held through shutdown: %d %q %v", got.status, got.body, got.err) |
| + } |
| + if _, ok := a.event("ready"); ok { |
| + t.Fatal("the latch opened during shutdown") |
| + } |
| + b.assertNoWrongService(t, a) |
| + if evs := a.events("acquire"); len(evs) != 0 { |
| + t.Fatalf("a wake was admitted during shutdown: %v", evs) |
| + } |
| + a.assertCleanShutdown(t) |
| +} |
| + |
| +// --- the box ------------------------------------------------------------------ |
| + |
| +type largeBox struct { |
| + dataDir, socketDir, storeDir, acmeDir string |
| + reg *registry.Registry |
| + control *largeBoxControl |
| + httpAddr, httpsAddr string |
| + client *http.Client |
| +} |
| + |
| +// newLargeBox builds the box the agent restarts on: largeBoxRoutes |
| +// sleeping instances on flat paths, largeBoxJournals of them with a layout |
| +// move cut short at three different points, largeBoxHolds held down after |
| +// a refused move (the reason, a second name for the database, persists), |
| +// a nested layout intent in the deployment store, a cached certificate for |
| +// every host the test asks for, and a cold instance control. |
| +func newLargeBox(t *testing.T) *largeBox { |
| + t.Helper() |
| + b := &largeBox{dataDir: t.TempDir(), storeDir: t.TempDir(), acmeDir: t.TempDir()} |
| + // Short on purpose, as in the layout package's own tests: a nested |
| + // instance's socket, plus its ".control" sibling, must fit a unix |
| + // socket address. |
| + sd, err := os.MkdirTemp("", "lb") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { os.RemoveAll(sd) }) |
| + b.socketDir = sd |
| + if n := len(filepath.Join(sd, "instances", provision.InstanceID(largeBoxHost(0)), "app.sock.control")); n > provision.MaxUnixPath { |
| + t.Fatalf("socket dir %s leaves a %d-byte control socket; run with a short TMPDIR", sd, n) |
| + } |
| + |
| + b.reg, err = registry.Open(filepath.Join(b.dataDir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { b.reg.Close() }) |
| + var hosts []string |
| + for i := 0; i < largeBoxRoutes; i++ { |
| + h := largeBoxHost(i) |
| + hosts = append(hosts, h) |
| + if err := b.reg.Upsert(registry.Route{ |
| + Host: h, Kind: "instance", Account: "acct", App: "shop", Status: "ready", Hibernate: true, |
| + Socket: filepath.Join(sd, h+".sock"), DBPath: provision.DefaultDBPath(b.dataDir, h), |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + for i := 0; i < largeBoxJournals; i++ { |
| + crash := "after-journal" |
| + switch { |
| + case i >= 6: |
| + crash = "before-remove" |
| + case i >= 4: |
| + crash = "mid-files" |
| + } |
| + b.crashedMove(t, largeBoxHost(i), crash) |
| + } |
| + for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ { |
| + b.heldMove(t, largeBoxHost(i), i) |
| + } |
| + appendIntent(t, release.NewDirStore(b.storeDir), release.LayoutModeNested, "") |
| + |
| + served := []string{largeBoxHost(largeBoxWakeHost)} |
| + for i := 0; i < largeBoxJournals+largeBoxHolds; i++ { |
| + served = append(served, largeBoxHost(i)) |
| + } |
| + pool := writeLargeBoxCerts(t, b.acmeDir, served) |
| + |
| + b.control = startLargeBoxControl(t, filepath.Join(sd, "control.sock"), hosts) |
| + b.httpAddr, b.httpsAddr = freeLoopbackAddr(t), freeLoopbackAddr(t) |
| + b.client = &http.Client{ |
| + Timeout: largeBoxLatchBound + 15*time.Second, |
| + Transport: &http.Transport{ |
| + Proxy: nil, |
| + DisableKeepAlives: true, |
| + TLSClientConfig: &tls.Config{RootCAs: pool}, |
| + DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| + return (&net.Dialer{}).DialContext(ctx, "tcp", b.httpsAddr) |
| + }, |
| + }, |
| + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, |
| + } |
| + return b |
| +} |
| + |
| +func (b *largeBox) seedFlat(t *testing.T, h string) { |
| + t.Helper() |
| + db := provision.DefaultDBPath(b.dataDir, h) |
| + for path, body := range map[string]string{db: "db " + h, db + "-wal": "wal", filepath.Join(b.dataDir, h+".env"): "CARLOS_ADMIN_TOKEN=t\n"} { |
| + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +// crashedMove leaves h as an agent killed mid-migration to nested leaves |
| +// it: the instance directories made and the journal written, then |
| +// nothing moved (after-journal), the database moved (mid-files), or |
| +// everything moved and the row flipped with only the journal's removal |
| +// left (before-remove). |
| +func (b *largeBox) crashedMove(t *testing.T, h, crash string) { |
| + t.Helper() |
| + b.seedFlat(t, h) |
| + if err := layout.EnsureDirs(b.dataDir, b.socketDir, h, -1); err != nil { |
| + t.Fatal(err) |
| + } |
| + plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + present := make([]bool, len(plan.Artifacts)) |
| + for i, a := range plan.Artifacts { |
| + if a.Name == "socket" { |
| + continue |
| + } |
| + _, err := os.Lstat(a.Src) |
| + present[i] = err == nil |
| + } |
| + j := plan.Journal(present) |
| + if err := layout.WriteJournal(b.dataDir, j); err != nil { |
| + t.Fatal(err) |
| + } |
| + move := func(a layout.Artifact) { |
| + if err := os.Rename(a.Src, a.Dst); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + switch crash { |
| + case "mid-files": |
| + move(plan.Artifacts[0]) |
| + case "before-remove": |
| + for i, a := range plan.Artifacts { |
| + if present[i] { |
| + move(a) |
| + } |
| + } |
| + if err := b.reg.SetInstancePaths(h, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := layout.UpdatePhase(b.dataDir, j, layout.PhaseDone); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +// heldMove leaves h held down after a refused move to nested: its hold |
| +// marker, and the reason, which persists across the restart (a second |
| +// name for its database, which a flat tenant can plant). |
| +func (b *largeBox) heldMove(t *testing.T, h string, i int) { |
| + t.Helper() |
| + b.seedFlat(t, h) |
| + if err := os.Link(provision.DefaultDBPath(b.dataDir, h), filepath.Join(b.dataDir, ".x"+strconv.Itoa(i))); err != nil { |
| + t.Fatal(err) |
| + } |
| + dir := filepath.Join(b.dataDir, "layout") |
| + if err := os.MkdirAll(dir, 0o700); err != nil { |
| + t.Fatal(err) |
| + } |
| + marker, _ := json.Marshal(map[string]any{"host": h, "reason": "db has 2 names", "since": time.Now().UTC()}) |
| + if err := os.WriteFile(filepath.Join(dir, h+".hold"), marker, 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// assertRecovered checks the files: every journal completed and removed, |
| +// each journalled host's row and database at the nested layout, and every |
| +// hold marker kept. |
| +func (b *largeBox) assertRecovered(t *testing.T) { |
| + t.Helper() |
| + if hosts, err := layout.JournalHosts(b.dataDir); err != nil || len(hosts) != 0 { |
| + t.Errorf("journals after startup: %v %v", hosts, err) |
| + } |
| + for i := 0; i < largeBoxJournals; i++ { |
| + h := largeBoxHost(i) |
| + plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + rt, err := b.reg.Get(h) |
| + if err != nil || rt == nil || rt.DBPath != plan.To.DB || rt.Socket != plan.To.Socket { |
| + t.Errorf("%s row after recovery: %+v %v", h, rt, err) |
| + } |
| + if body, err := os.ReadFile(plan.Artifacts[0].Dst); err != nil || string(body) != "db "+h { |
| + t.Errorf("%s database after recovery: %q %v", h, body, err) |
| + } |
| + } |
| + held, err := layout.HoldMarkers(b.dataDir) |
| + if err != nil || len(held) != largeBoxHolds { |
| + t.Errorf("hold markers after startup: %v %v", held, err) |
| + } |
| +} |
| + |
| +// assertNoWrongService reads the backend's record: no wake admitted before |
| +// the latch opened, no held host started, and no journalled host started |
| +// at its pre-recovery socket. |
| +func (b *largeBox) assertNoWrongService(t *testing.T, a *largeBoxAgent) { |
| + t.Helper() |
| + if v := a.events("violation"); len(v) != 0 { |
| + t.Errorf("backend violations: %v", v) |
| + } |
| + for _, e := range append(a.events("acquire"), a.events("start")...) { |
| + for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ { |
| + if e.arg(0) == largeBoxHost(i) { |
| + t.Errorf("held host %s reached the backend: %v", e.arg(0), e) |
| + } |
| + } |
| + } |
| + for _, e := range a.events("start") { |
| + for i := 0; i < largeBoxJournals; i++ { |
| + if h := largeBoxHost(i); e.arg(0) == h && e.arg(1) == filepath.Join(b.socketDir, h+".sock") { |
| + t.Errorf("%s started at its pre-recovery socket", h) |
| + } |
| + } |
| + } |
| +} |
| + |
| +type largeBoxAnswer struct { |
| + status int |
| + body string |
| + header http.Header |
| + err error |
| +} |
| + |
| +func (b *largeBox) get(host string) largeBoxAnswer { |
| + resp, err := b.client.Get("https://" + host + "/") |
| + if err != nil { |
| + return largeBoxAnswer{err: err} |
| + } |
| + defer resp.Body.Close() |
| + body, err := io.ReadAll(resp.Body) |
| + return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err} |
| +} |
| + |
| +func (b *largeBox) fire(host string) chan largeBoxAnswer { |
| + ch := make(chan largeBoxAnswer, 1) |
| + go func() { ch <- b.get(host) }() |
| + return ch |
| +} |
| + |
| +func (b *largeBox) answer(t *testing.T, ch chan largeBoxAnswer) largeBoxAnswer { |
| + t.Helper() |
| + select { |
| + case got := <-ch: |
| + return got |
| + case <-time.After(10 * time.Second): |
| + t.Fatal("request did not finish") |
| + return largeBoxAnswer{} |
| + } |
| +} |
| + |
| +func freeLoopbackAddr(t *testing.T) string { |
| + t.Helper() |
| + ln, err := net.Listen("tcp", "127.0.0.1:0") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer ln.Close() |
| + return ln.Addr().String() |
| +} |
| + |
| +// writeLargeBoxCerts puts one self-signed certificate for hosts into the |
| +// edge's autocert directory cache, under each host's name, so the edge in |
| +// production mode serves them TLS without ordering anything. |
| +func writeLargeBoxCerts(t *testing.T, dir string, hosts []string) *x509.CertPool { |
| + t.Helper() |
| + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + tmpl := &x509.Certificate{ |
| + SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "large box"}, DNSNames: hosts, |
| + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), |
| + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, |
| + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, BasicConstraintsValid: true, IsCA: true, |
| + } |
| + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + keyDER, err := x509.MarshalECPrivateKey(key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + pemBytes := append(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...) |
| + for _, h := range hosts { |
| + if err := os.WriteFile(filepath.Join(dir, h), pemBytes, 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + cert, err := x509.ParseCertificate(der) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + pool := x509.NewCertPool() |
| + pool.AddCert(cert) |
| + return pool |
| +} |
| + |
| +// --- instance control --------------------------------------------------------- |
| + |
| +// largeBoxControl is the real instance-control server with a cold systemd |
| +// behind its Systemctl seam: every call waits for warm, then the box's |
| +// leftover units are listed active and each stop takes largeBoxUnitStop. |
| +type largeBoxControl struct { |
| + warm chan struct{} |
| + units string |
| + stops atomic.Int32 |
| +} |
| + |
| +func startLargeBoxControl(t *testing.T, path string, hosts []string) *largeBoxControl { |
| + t.Helper() |
| + c := &largeBoxControl{warm: make(chan struct{})} |
| + var units strings.Builder |
| + for _, h := range hosts { |
| + fmt.Fprintf(&units, "%s loaded active running carlos instance\n", instancectl.UnitName(instancectl.RoleRun, h)) |
| + } |
| + c.units = units.String() |
| + srv := &instancectl.Server{ |
| + Systemctl: c.systemctl, |
| + PeerOK: func(net.Conn) error { return nil }, |
| + CgroupRoot: t.TempDir(), |
| + Build: version.Build, |
| + Logf: func(string, ...any) {}, |
| + } |
| + ln, err := net.Listen("unix", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + done := make(chan struct{}) |
| + go func() { defer close(done); _ = srv.Serve(ctx, ln) }() |
| + t.Cleanup(func() { cancel(); <-done }) |
| + return c |
| +} |
| + |
| +func (c *largeBoxControl) systemctl(ctx context.Context, args ...string) (string, error) { |
| + select { |
| + case <-c.warm: |
| + case <-ctx.Done(): |
| + return "", ctx.Err() |
| + } |
| + switch args[0] { |
| + case "list-units": |
| + return c.units, nil |
| + case "list-jobs": |
| + return "", nil |
| + case "stop": |
| + select { |
| + case <-time.After(largeBoxUnitStop): |
| + case <-ctx.Done(): |
| + return "", ctx.Err() |
| + } |
| + c.stops.Add(1) |
| + return "", nil |
| + case "show": |
| + return "inactive\n", nil |
| + } |
| + return "", fmt.Errorf("systemctl %v: not part of a sweep", args) |
| +} |
| + |
| +// --- the agent process ---------------------------------------------------------- |
| + |
| +type largeBoxAgent struct { |
| + cmd *exec.Cmd |
| + began time.Time |
| + eventsPath string |
| + log string |
| + exited chan struct{} // closed once the process has been waited for |
| + exitErr error // Wait's answer, read after exited closes |
| + b *largeBox |
| +} |
| + |
| +func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent { |
| + t.Helper() |
| + cfgDir := t.TempDir() |
| + cfg := largeBoxConfig{ |
| + DataDir: b.dataDir, SocketDir: b.socketDir, StoreDir: b.storeDir, AcmeDir: b.acmeDir, |
| + Control: filepath.Join(b.socketDir, "control.sock"), Events: filepath.Join(cfgDir, "events"), |
| + HTTP: b.httpAddr, HTTPS: b.httpsAddr, |
| + } |
| + raw, err := json.Marshal(cfg) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + cfgPath := filepath.Join(cfgDir, "agent.json") |
| + if err := os.WriteFile(cfgPath, raw, 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + exe, err := os.Executable() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := &largeBoxAgent{eventsPath: cfg.Events, log: filepath.Join(cfgDir, "agent.log"), exited: make(chan struct{}), b: b} |
| + logFile, err := os.Create(a.log) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a.cmd = exec.Command(exe, "-test.run=^TestStartupLargeBoxEndToEnd$", "-test.count=1") |
| + // A clean environment: nothing of the developer's or the runner's |
| + // (a deployment bucket, AWS credentials, CARLOS_*) reaches the agent. |
| + a.cmd.Env = []string{"HOME=" + b.dataDir, "TMPDIR=" + os.TempDir(), "CARLOS_DATA_DIR=" + b.dataDir, largeBoxAgentEnv + "=" + cfgPath} |
| + a.cmd.Stdout, a.cmd.Stderr = logFile, logFile |
| + a.began = time.Now() |
| + if err := a.cmd.Start(); err != nil { |
| + t.Fatal(err) |
| + } |
| + go func() { |
| + a.exitErr = a.cmd.Wait() |
| + logFile.Close() |
| + close(a.exited) |
| + }() |
| + t.Cleanup(func() { |
| + select { |
| + case <-a.exited: |
| + default: |
| + _ = a.cmd.Process.Kill() |
| + <-a.exited |
| + } |
| + if t.Failed() { |
| + if out, err := os.ReadFile(a.log); err == nil { |
| + if len(out) > 16<<10 { |
| + out = out[len(out)-16<<10:] |
| + } |
| + t.Logf("agent log (tail):\n%s", out) |
| + } |
| + } |
| + }) |
| + return a |
| +} |
| + |
| +// waitListening probes as the update gate does (systemapply.go |
| +// productionSysOps.probeEdge): GET / on the edge's plain-HTTP listener, |
| +// three-second client timeout, no redirect followed, and any HTTP answer |
| +// passes. It returns how long after the process started the first answer |
| +// came, and fails past largeBoxListenBound. |
| +func (a *largeBoxAgent) waitListening(t *testing.T) time.Duration { |
| + t.Helper() |
| + probe := &http.Client{ |
| + Timeout: 3 * time.Second, |
| + Transport: &http.Transport{Proxy: nil, DisableKeepAlives: true}, |
| + CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, |
| + } |
| + deadline := a.began.Add(largeBoxListenBound) |
| + for { |
| + resp, err := probe.Get("http://" + a.b.httpAddr + "/") |
| + if err == nil { |
| + resp.Body.Close() |
| + took := time.Since(a.began) |
| + if took > largeBoxListenBound { |
| + t.Fatalf("gate probe answered %s after the agent started, over %s", took, largeBoxListenBound) |
| + } |
| + return took |
| + } |
| + select { |
| + case <-a.exited: |
| + t.Fatalf("agent exited before listening: %v", a.exitErr) |
| + default: |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatalf("gate probe unanswered %s after the agent started: %v", time.Since(a.began), err) |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + } |
| +} |
| + |
| +func (a *largeBoxAgent) signal(t *testing.T, sig os.Signal) { |
| + t.Helper() |
| + if err := a.cmd.Process.Signal(sig); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func (a *largeBoxAgent) waitExit(t *testing.T, bound time.Duration) { |
| + t.Helper() |
| + select { |
| + case <-a.exited: |
| + if a.exitErr != nil { |
| + t.Fatalf("agent exit: %v", a.exitErr) |
| + } |
| + case <-time.After(bound): |
| + t.Fatalf("agent did not exit within %s of the signal", bound) |
| + } |
| +} |
| + |
| +// assertCleanShutdown reads the agent's own account of its shutdown: the |
| +// startup work joined within runAgent's bound, and no goroutine of the |
| +// startup sequence, the activator or the fake instances left after it. |
| +func (a *largeBoxAgent) assertCleanShutdown(t *testing.T) { |
| + t.Helper() |
| + j, ok := a.event("joined") |
| + if !ok { |
| + t.Fatal("agent did not report its startup join") |
| + } |
| + took, err := time.ParseDuration(j.arg(1)) |
| + if j.arg(0) != "true" || err != nil || took > largeBoxJoinBound { |
| + t.Fatalf("startup join: %v", j) |
| + } |
| + t.Logf("startup work joined %s after the edge returned", took) |
| + if _, ok := a.event("leak-free"); !ok { |
| + t.Fatal("goroutines of the startup sequence outlived shutdown; see the agent log") |
| + } |
| +} |
| + |
| +type largeBoxEvent struct { |
| + at time.Time |
| + kind string |
| + args []string |
| +} |
| + |
| +func (e largeBoxEvent) arg(i int) string { |
| + if i < len(e.args) { |
| + return e.args[i] |
| + } |
| + return "" |
| +} |
| + |
| +func (a *largeBoxAgent) all() []largeBoxEvent { |
| + f, err := os.Open(a.eventsPath) |
| + if err != nil { |
| + return nil |
| + } |
| + defer f.Close() |
| + var out []largeBoxEvent |
| + sc := bufio.NewScanner(f) |
| + for sc.Scan() { |
| + fields := strings.Fields(sc.Text()) |
| + if len(fields) < 2 { |
| + continue |
| + } |
| + ns, err := strconv.ParseInt(fields[0], 10, 64) |
| + if err != nil { |
| + continue |
| + } |
| + out = append(out, largeBoxEvent{at: time.Unix(0, ns), kind: fields[1], args: fields[2:]}) |
| + } |
| + return out |
| +} |
| + |
| +func (a *largeBoxAgent) events(kind string) []largeBoxEvent { |
| + var out []largeBoxEvent |
| + for _, e := range a.all() { |
| + if e.kind == kind { |
| + out = append(out, e) |
| + } |
| + } |
| + return out |
| +} |
| + |
| +func (a *largeBoxAgent) event(kind string, args ...string) (largeBoxEvent, bool) { |
| + for _, e := range a.events(kind) { |
| + match := true |
| + for i, want := range args { |
| + if e.arg(i) != want { |
| + match = false |
| + } |
| + } |
| + if match { |
| + return e, true |
| + } |
| + } |
| + return largeBoxEvent{}, false |
| +} |
| + |
| +func (a *largeBoxAgent) waitEvent(t *testing.T, bound time.Duration, kind string, args ...string) largeBoxEvent { |
| + t.Helper() |
| + deadline := time.Now().Add(bound) |
| + for { |
| + if e, ok := a.event(kind, args...); ok { |
| + return e |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatalf("agent did not report %s %v within %s", kind, args, bound) |
| + } |
| + time.Sleep(10 * time.Millisecond) |
| + } |
| +} |
| + |
| +// --- inside the agent process --------------------------------------------------- |
| + |
| +type largeBoxEvents struct { |
| + mu sync.Mutex |
| + f *os.File |
| +} |
| + |
| +func (e *largeBoxEvents) log(kind string, args ...any) { |
| + var b strings.Builder |
| + fmt.Fprintf(&b, "%d %s", time.Now().UnixNano(), kind) |
| + for _, a := range args { |
| + fmt.Fprintf(&b, " %v", a) |
| + } |
| + b.WriteByte('\n') |
| + e.mu.Lock() |
| + defer e.mu.Unlock() |
| + _, _ = e.f.WriteString(b.String()) |
| +} |
| + |
| +// runLargeBoxAgent is the agent: runAgent's legacy-box startup over the |
| +// fixture, then the edge, then runAgent's shutdown. It returns the exit |
| +// code. |
| +func runLargeBoxAgent(cfgPath string) int { |
| + raw, err := os.ReadFile(cfgPath) |
| + if err != nil { |
| + log.Printf("largebox agent: %v", err) |
| + return 2 |
| + } |
| + var cfg largeBoxConfig |
| + if err := json.Unmarshal(raw, &cfg); err != nil { |
| + log.Printf("largebox agent: %v", err) |
| + return 2 |
| + } |
| + f, err := os.OpenFile(cfg.Events, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) |
| + if err != nil { |
| + log.Printf("largebox agent: %v", err) |
| + return 2 |
| + } |
| + defer f.Close() |
| + ev := &largeBoxEvents{f: f} |
| + |
| + agentCtx, cancelAgent := context.WithCancel(context.Background()) |
| + defer cancelAgent() |
| + startupCtx, cancelStartup := context.WithCancel(agentCtx) |
| + defer cancelStartup() |
| + |
| + dd := cfg.DataDir |
| + regPath := filepath.Join(dd, "registry.db") |
| + reg, err := registry.Open(regPath) |
| + if err != nil { |
| + log.Printf("largebox agent: %v", err) |
| + return 2 |
| + } |
| + defer reg.Close() |
| + |
| + control := instancectl.NewClient(cfg.Control) |
| + nestedOK := nestedGate(func() error { return instancefs.SelfCheck(dd) }, instancefs.MountNoSymfollow, control.Ready) |
| + started := make(chan struct{}) |
| + svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, &largeBoxBackend{ev: ev, started: started}, reg) |
| + svc.Routes = reg |
| + svc.Startup = started |
| + openStore := func(context.Context) (release.Store, error) { return release.NewDirStore(cfg.StoreDir), nil } |
| + |
| + startupDone := activatorStart{ |
| + Ready: func() { |
| + ev.log("ready") // before the close, so no acquire can be stamped earlier |
| + close(started) |
| + }, |
| + Sweep: func(ctx context.Context) error { |
| + sw, err := control.Sweep(ctx) |
| + if err != nil { |
| + log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err) |
| + } else if len(sw.Units) > 0 { |
| + log.Printf("agent: swept %d instance unit(s) left from a previous agent", len(sw.Units)) |
| + } |
| + ev.log("swept", len(sw.Units), err == nil) |
| + return err |
| + }, |
| + Recover: func(ctx context.Context, swept bool) *layout.Pass { |
| + pass := newLayoutPass(reg, svc, dd, nestedOK, swept) |
| + pass.SocketDir, pass.SocketsGID = cfg.SocketDir, -1 // the fixture's run dir; no carlos-sockets group |
| + intent := startupRecoveryIntent(ctx, dd, openStore, "largebox") |
| + recoverLayoutAtStart(ctx, pass, intent) |
| + ev.log("recovered", intent) |
| + return pass |
| + }, |
| + Routes: reg.List, |
| + Route: reg.Get, |
| + Keys: largeBoxKeys{ev}, |
| + Slot: svc.PrepareScheduledToken, |
| + Publish: armedLayoutPass.Store, |
| + Start: svc.Start, |
| + }.Run(startupCtx) |
| + go func() { |
| + <-startupDone |
| + ev.log("startup-done") |
| + }() |
| + // runAgent's worker boot pass, which waits on the same latch. |
| + go func() { |
| + select { |
| + case <-started: |
| + case <-startupCtx.Done(): |
| + return |
| + } |
| + routes, err := reg.List() |
| + if err != nil { |
| + return |
| + } |
| + runWorkerBootPass(routes, svc.WakeWorker) |
| + }() |
| + |
| + edge.RunWithActivatorReady([]string{"-registry", regPath, "-acme-cache", cfg.AcmeDir, "-http", cfg.HTTP, "-https", cfg.HTTPS}, |
| + largeBoxActivator{Service: svc, ev: ev}, edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes), |
| + func(e *edge.Edge) { |
| + reload := func() { _ = e.Reload() } |
| + layoutReload.Store(&reload) |
| + }) |
| + |
| + // runAgent's shutdown, in its order. |
| + cancelStartup() |
| + joinCtx, joinCancel := context.WithTimeout(context.Background(), largeBoxJoinBound) |
| + began := time.Now() |
| + joined := joinAgentStartup(joinCtx, startupDone) |
| + joinCancel() |
| + ev.log("joined", joined, time.Since(began)) |
| + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| + svc.Shutdown(ctx) |
| + cancel() |
| + cancelAgent() |
| + |
| + deadline := time.Now().Add(2 * time.Second) |
| + for { |
| + left := largeBoxStartupGoroutines() |
| + if len(left) == 0 { |
| + ev.log("leak-free") |
| + return 0 |
| + } |
| + if time.Now().After(deadline) { |
| + log.Printf("largebox agent: %d goroutine(s) outlived shutdown:\n\n%s", len(left), strings.Join(left, "\n\n")) |
| + ev.log("leak", len(left)) |
| + return 0 |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + } |
| +} |
| + |
| +// largeBoxStartupGoroutines lists every goroutine but this one that runs, |
| +// or was started by, the startup sequence, the layout pass, the control |
| +// client, the activator, or this agent's own fakes, and every server |
| +// connection still open (a request held at the latch is answered only |
| +// once the activator shuts down, after the edge has stopped waiting for |
| +// it). The edge's process-lifetime loops (its registry watch, its SIGHUP |
| +// reader) are not startup's and are left alone. |
| +func largeBoxStartupGoroutines() []string { |
| + buf := make([]byte, 8<<20) |
| + n := runtime.Stack(buf, true) |
| + blocks := strings.Split(string(buf[:n]), "\n\n") |
| + markers := []string{ |
| + "host.activatorStart", "host.startActivator", "host.backfillInstanceKeys", "host.keyBackfillInSlot", |
| + "host.recoverLayoutAtStart", "host.startupRecoveryIntent", "host.runWorkerBootPass", |
| + "host.runLargeBoxAgent", "host.(*largeBoxBackend)", "host.largeBoxActivator", |
| + "layout.(*Pass)", "instancectl.(*Client)", "activator.(*Service)", |
| + "net/http.(*conn).serve", |
| + } |
| + var left []string |
| + for _, g := range blocks[1:] { // [0] is the calling goroutine |
| + for _, m := range markers { |
| + if strings.Contains(g, m) { |
| + left = append(left, g) |
| + break |
| + } |
| + } |
| + } |
| + return left |
| +} |
| + |
| +// largeBoxActivator is the activator as the edge sees it, reporting each |
| +// instance request as it reaches it. |
| +type largeBoxActivator struct { |
| + *activator.Service |
| + ev *largeBoxEvents |
| +} |
| + |
| +func (a largeBoxActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool { |
| + if rt.Kind == "instance" { |
| + a.ev.log("enter", rt.Host, rt.Socket) |
| + } |
| + return a.Service.ServeHTTP(w, r, rt, proxy) |
| +} |
| + |
| +// largeBoxBackend is activator.Backend with no S3 and no processes: a wake |
| +// serves an app on the row's socket that names the host and the socket it |
| +// was started at. It records every acquire and start, and any acquire the |
| +// startup latch should have prevented. |
| +type largeBoxBackend struct { |
| + ev *largeBoxEvents |
| + started <-chan struct{} |
| +} |
| + |
| +func (b *largeBoxBackend) Acquire(_ context.Context, rt registry.Route) (activator.Lease, error) { |
| + select { |
| + case <-b.started: |
| + default: |
| + b.ev.log("violation", "acquire-before-ready", rt.Host) |
| + } |
| + b.ev.log("acquire", rt.Host) |
| + return nopLease{}, nil |
| +} |
| + |
| +func (b *largeBoxBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil } |
| +func (b *largeBoxBackend) Seed(context.Context, registry.Route) error { return nil } |
| + |
| +func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) { |
| + b.ev.log("start", rt.Host, rt.Socket) |
| + if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil { |
| + return nil, "", err |
| + } |
| + ln, err := net.Listen("unix", rt.Socket) |
| + if err != nil { |
| + return nil, "", err |
| + } |
| + body := "host=" + rt.Host + " socket=" + rt.Socket |
| + p := &largeBoxProc{srv: &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { io.WriteString(w, body) })}, done: make(chan struct{})} |
| + go func() { defer close(p.done); _ = p.srv.Serve(ln) }() |
| + return p, "", nil |
| +} |
| + |
| +func (b *largeBoxBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| +func (b *largeBoxBackend) HasSidecar(registry.Route) bool { return false } |
| +func (b *largeBoxBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) { |
| + return nil, errors.New("largeBoxBackend: no sidecars on this box") |
| +} |
| +func (b *largeBoxBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) { |
| + return nil, "", errors.New("largeBoxBackend: no workers on this box") |
| +} |
| + |
| +type largeBoxProc struct { |
| + srv *http.Server |
| + done chan struct{} |
| + exited atomic.Bool |
| +} |
| + |
| +func (p *largeBoxProc) Stop(context.Context) error { |
| + _ = p.srv.Close() |
| + <-p.done |
| + p.exited.Store(true) |
| + return nil |
| +} |
| + |
| +func (p *largeBoxProc) Exited() bool { return p.exited.Load() } |
| + |
| +// largeBoxKeys is the exec backend's key writer as the key pass sees it: |
| +// every row it is handed already has its key. |
| +type largeBoxKeys struct{ ev *largeBoxEvents } |
| + |
| +func (k largeBoxKeys) BackfillInstanceKey(_ context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) { |
| + k.ev.log("key", rt.Host) |
| + return activator.InstanceKeyPresent, nil |
| +} |