amadan

carlos / platform Public

host: large-box startup test, review round 1 (#281 ask 2)

Astra's findings on 671a338f, all in the test:

- TestRunAgentStartupWiring reads runAgent's source and fails if its
  wiring drifts from what the child rebuilds: the latch set, Run given
  startupCtx, the edge, then cancel, join and Shutdown in that order; the
  activatorStart fields the child copies; Recover's three calls in order;
  the boot pass gated on started or startupCtx. It also fails if anything
  before the edge receives from a channel, joins startup, or calls a
  closure that waits on the latch outside the claim-box branch.
  Sharing one orchestration function is left as a follow-up.
- Every admission kind the latch gates now arrives during startup and is
  asserted to complete only after it opens: a sidecar-marked request, a
  worker (runAgent's boot pass and a direct WakeWorker, coalesced on one
  child that reports ready), and a scheduled delivery through the real
  schedule.Runner and activatorDeliverer.
- A checkpoint in Start asserts every held host is already fenced again
  and every journal complete when the activator starts.
- The TLS listener is awaited on its own, boundedly, after the gate probe
  on the plain-HTTP listener.
- Ports: the edge cannot report a :0 port, so a collision is detected
  from the agent's log and the agent restarted on fresh ports.
- The socket root tries TMPDIR, then its own directory under /tmp, and
  skips with the reason when neither is short enough.
- The agent's log file is closed by a cleanup registered as it is made.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 7e16dd018f673a1afc6d5d4df55a060803faa1e6 parent 671a338
1 file changed, +569 −94
  • internal/host/startup_largebox_linux_test.go +569 −94

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	333.773s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	4.537s
ok  	github.com/carlosframework/platform/cmd/moveprobe	92.218s
ok  	github.com/carlosframework/platform/cmd/zddprobe	0.381s
ok  	github.com/carlosframework/platform/internal/activator	228.469s
ok  	github.com/carlosframework/platform/internal/agentsock	0.020s
ok  	github.com/carlosframework/platform/internal/alerting	0.004s
ok  	github.com/carlosframework/platform/internal/analytics	0.018s
ok  	github.com/carlosframework/platform/internal/appconfig	2.285s
ok  	github.com/carlosframework/platform/internal/apperr	0.081s
ok  	github.com/carlosframework/platform/internal/applog	0.132s
ok  	github.com/carlosframework/platform/internal/awsinv	0.007s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.004s
ok  	github.com/carlosframework/platform/internal/boxid	0.626s
?   	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.012s
ok  	github.com/carlosframework/platform/internal/certwarm	0.002s
ok  	github.com/carlosframework/platform/internal/cgroup	0.697s
ok  	github.com/carlosframework/platform/internal/changelog	0.310s
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	361.712s
?   	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.098s
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.003s
ok  	github.com/carlosframework/platform/internal/docsite	0.005s
ok  	github.com/carlosframework/platform/internal/domains	2.093s
ok  	github.com/carlosframework/platform/internal/doorkey	0.006s
ok  	github.com/carlosframework/platform/internal/economics	0.003s
ok  	github.com/carlosframework/platform/internal/edge	26.217s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.009s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.138s
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.017s
ok  	github.com/carlosframework/platform/internal/geoip	0.030s
ok  	github.com/carlosframework/platform/internal/host	125.177s
ok  	github.com/carlosframework/platform/internal/hostactors	0.374s
ok  	github.com/carlosframework/platform/internal/hostbox	2.114s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.542s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.007s
ok  	github.com/carlosframework/platform/internal/hostexec	4.672s
ok  	github.com/carlosframework/platform/internal/hostfence	0.030s
ok  	github.com/carlosframework/platform/internal/hostfiles	10.436s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.012s
ok  	github.com/carlosframework/platform/internal/hostguard	0.966s
ok  	github.com/carlosframework/platform/internal/hostops	1.447s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.002s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.005s
ok  	github.com/carlosframework/platform/internal/hostrelocate	0.469s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.132s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.002s
ok  	github.com/carlosframework/platform/internal/infraview	0.002s
ok  	github.com/carlosframework/platform/internal/instancectl	1.812s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.701s
ok  	github.com/carlosframework/platform/internal/instancefs	0.023s
ok  	github.com/carlosframework/platform/internal/instancekey	0.016s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.023s
ok  	github.com/carlosframework/platform/internal/layout	1.444s
ok  	github.com/carlosframework/platform/internal/lease	9.448s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.007s
ok  	github.com/carlosframework/platform/internal/ledger	0.027s
ok  	github.com/carlosframework/platform/internal/mail	0.152s
ok  	github.com/carlosframework/platform/internal/maintenance	0.077s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.510s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.061s
ok  	github.com/carlosframework/platform/internal/metrics	0.005s
ok  	github.com/carlosframework/platform/internal/moves	21.086s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.761s
ok  	github.com/carlosframework/platform/internal/objstore	0.174s
ok  	github.com/carlosframework/platform/internal/ops	0.161s
ok  	github.com/carlosframework/platform/internal/park	0.203s
ok  	github.com/carlosframework/platform/internal/privatefile	0.004s
ok  	github.com/carlosframework/platform/internal/provision	7.375s
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.007s
ok  	github.com/carlosframework/platform/internal/relay	7.224s
ok  	github.com/carlosframework/platform/internal/release	19.453s
ok  	github.com/carlosframework/platform/internal/rollout	3.616s
ok  	github.com/carlosframework/platform/internal/rolloutwords	0.004s
ok  	github.com/carlosframework/platform/internal/rootaws	0.008s
ok  	github.com/carlosframework/platform/internal/rootparticipation	0.009s
ok  	github.com/carlosframework/platform/internal/rootstate	0.026s
ok  	github.com/carlosframework/platform/internal/s3compat	0.005s
ok  	github.com/carlosframework/platform/internal/safefile	0.006s
ok  	github.com/carlosframework/platform/internal/schedule	2.287s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	8.340s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	4.248s
ok  	github.com/carlosframework/platform/internal/status	0.041s
ok  	github.com/carlosframework/platform/internal/steering	0.006s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.319s
ok  	github.com/carlosframework/platform/internal/svcrole	0.002s
ok  	github.com/carlosframework/platform/internal/telemetry	0.004s
ok  	github.com/carlosframework/platform/internal/termfence	0.004s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.004s
ok  	github.com/carlosframework/platform/internal/unixdial	0.002s
ok  	github.com/carlosframework/platform/internal/updatewords	0.004s
ok  	github.com/carlosframework/platform/internal/usage	0.004s
ok  	github.com/carlosframework/platform/internal/version	0.001s
ok  	github.com/carlosframework/platform/internal/vet	1.548s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
  minio: cached, digest verified
  mc: cached, digest verified
  minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
  buckets and scoped user ready
=== RUN   TestDeliveredContractMinio
=== RUN   TestDeliveredContractMinio/put
=== RUN   TestDeliveredContractMinio/get_round-trips_the_body
=== RUN   TestDeliveredContractMinio/list_shows_the_key
=== RUN   TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN   TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN   TestDeliveredContractMinio/delete_removes_it
=== RUN   TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
    --- PASS: TestDeliveredContractMinio/put (0.00s)
    --- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
    --- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
    --- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
    --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
    --- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
    --- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok  	github.com/carlosframework/platform/internal/objstore	0.022s
=== RUN   TestT4StorageProbeMinio
2026/10/13 22:17:25 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE36B340F4B381, 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.020s
minio contract: PASS
diff --git a/internal/host/startup_largebox_linux_test.go b/internal/host/startup_largebox_linux_test.go
index 7e893e684..b23550fa0 100644
--- a/internal/host/startup_largebox_linux_test.go
+++ b/internal/host/startup_largebox_linux_test.go
@@ -24,17 +24,25 @@ package host
// 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;
+// - every kind of admission the latch gates arrives during startup: an
+// instance request, a sidecar-marked request, a worker wake (from
+// runAgent's boot pass and from a direct call), and a scheduled
+// delivery through the real schedule runner and deliverer;
// - 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.
+// socket and starts workers that report ready. 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.
+// same pieces in the same order, and TestRunAgentStartupWiring reads
+// runAgent's source to fail if the two drift apart. Sharing one
+// orchestration function between them is a follow-up (#281). 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
@@ -54,6 +62,10 @@ import (
"encoding/pem"
"errors"
"fmt"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "go/types"
"io"
"log"
"math/big"
@@ -79,6 +91,7 @@ import (
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
"github.com/carlosframework/platform/internal/version"
)
@@ -87,10 +100,14 @@ const (
// 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
+ largeBoxRoutes = 300 // instance routes; 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, woken by a request
+ largeBoxDueHost = 13 // a sleeping instance with a scheduled delivery due
+ largeBoxSidecarHost = 14 // a sleeping instance its sidecar calls
+ largeBoxWorker = "worker-000.example.com"
+ largeBoxDuePath = "/jobs/tick"
// 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
@@ -110,6 +127,9 @@ const (
// requests before it returns.
largeBoxJoinBound = 5 * time.Second
largeBoxDrain = 10 * time.Second
+ // largeBoxBindAttempts is how often the agent is started on fresh ports
+ // when one it was handed is taken before it binds (see startAgent).
+ largeBoxBindAttempts = 3
)
func largeBoxHost(i int) string { return fmt.Sprintf("app-%03d.example.com", i) }
@@ -140,38 +160,45 @@ func TestStartupLargeBoxEndToEnd(t *testing.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.
+// probe while the sweep waits, hold every admission 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)
+ t.Logf("gate probe answered %s after the agent started, %d routes, instance control cold", a.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)
+ wake, held, moved, due, sidecar := largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxJournals), largeBoxHost(0), largeBoxHost(largeBoxDueHost), largeBoxHost(largeBoxSidecarHost)
flatMoved := filepath.Join(b.socketDir, moved+".sock")
- wakeAns, heldAns, movedAns := b.fire(wake), b.fire(held), b.fire(moved)
+ wakeAns, heldAns, movedAns := b.fire(wake, false), b.fire(held, false), b.fire(moved, false)
+ sidecarAns := b.fire(sidecar, true)
enterWake := a.waitEvent(t, 10*time.Second, "enter", wake)
a.waitEvent(t, 10*time.Second, "enter", held)
+ enterSidecar := a.waitEvent(t, 10*time.Second, "enter", sidecar)
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.
+ // The agent's own admissions: a direct worker wake and the scheduled
+ // tick, both started before the edge listened.
+ workerAsked := a.waitEvent(t, 10*time.Second, "worker-direct-begin")
+ ticked := a.waitEvent(t, 10*time.Second, "tick-done")
+ // Every admission is now waiting, and startup is not done.
time.Sleep(200 * time.Millisecond)
- for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns} {
+ for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns, sidecar: sidecarAns} {
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)
+ for _, kind := range []string{"acquire", "start", "worker-start", "served", "worker-direct", "worker-boot", "activator-start"} {
+ if evs := a.events(kind); len(evs) != 0 {
+ t.Fatalf("%s while instance control was cold: %v", kind, evs)
+ }
}
// Instance control warms: the sweep stops the 300 leftovers, then
@@ -185,8 +212,17 @@ func testLargeBoxColdControl(t *testing.T) {
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")
+ for what, at := range map[string]time.Time{"wake": enterWake.at, "sidecar request": enterSidecar.at, "worker wake": workerAsked.at, "scheduled tick": ticked.at} {
+ if !at.Before(ready.at) {
+ t.Fatalf("the %s did not arrive before the latch opened; the test proves nothing about its latch", what)
+ }
+ }
+
+ // Start ran only once the held hosts were fenced again and every
+ // journal was complete.
+ st := a.waitEvent(t, time.Second, "activator-start")
+ if st.arg(0) != "held="+strconv.Itoa(largeBoxHolds) || st.arg(1) != "fenced="+strconv.Itoa(largeBoxHolds) || st.arg(2) != "journals=0" || st.at.After(ready.at) {
+ t.Fatalf("at Start: %v", st)
}
// The wake was held, then served.
@@ -194,11 +230,27 @@ func testLargeBoxColdControl(t *testing.T) {
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))
+ a.assertAfter(t, ready, "acquire", wake)
+
+ // The sidecar's call was held too, then answered as the activator
+ // answers a sidecar whose instance sleeps: it does not wake it.
+ got = b.answer(t, sidecarAns)
+ if got.status != http.StatusServiceUnavailable || got.body != "instance is asleep; sidecar requests do not wake it\n" || got.at.Before(ready.at) {
+ t.Fatalf("sidecar request held through startup: %d %q %v at %s, latch at %s", got.status, got.body, got.err, got.at, ready.at)
+ }
+
+ // The worker: both wakes waited, then coalesced on one child.
+ a.assertAfter(t, ready, "worker-direct", largeBoxWorker, "true")
+ a.assertAfter(t, ready, "worker-boot", largeBoxWorker, "true")
+ a.assertAfter(t, ready, "worker-start", largeBoxWorker)
+ if n := len(a.events("worker-start")); n != 1 {
+ t.Errorf("worker started %d times, want once", n)
}
+ // The scheduled delivery waited, then woke its instance and ran.
+ a.assertAfter(t, ready, "served", due, largeBoxDuePath)
+ a.assertAfter(t, ready, "acquire", due)
+
// 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" {
@@ -216,11 +268,11 @@ func testLargeBoxColdControl(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- got = b.get(moved)
+ got = b.get(moved, false)
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)
+ got = b.get(held, false)
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)
}
@@ -252,12 +304,11 @@ func testLargeBoxColdControl(t *testing.T) {
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)
+ t.Logf("gate probe answered %s after the agent started, instance control never warms", a.listen)
wake := largeBoxHost(largeBoxWakeHost)
began := time.Now()
- got := b.get(wake)
+ got := b.get(wake, false)
took := time.Since(began)
t.Logf("wake held %s by a startup that does not finish", took)
if got.err != nil {
@@ -269,13 +320,8 @@ func testLargeBoxOverrun(t *testing.T) {
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")
- }
+ a.assertNeverStarted(t)
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.
@@ -283,6 +329,7 @@ func testLargeBoxOverrun(t *testing.T) {
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.assertNeverStarted(t)
a.assertCleanShutdown(t)
}
@@ -293,10 +340,9 @@ func testLargeBoxOverrun(t *testing.T) {
func testLargeBoxShutdownMidStartup(t *testing.T) {
b := newLargeBox(t)
a := b.startAgent(t)
- a.waitListening(t)
wake := largeBoxHost(largeBoxWakeHost)
- ans := b.fire(wake)
+ ans := b.fire(wake, false)
a.waitEvent(t, 10*time.Second, "enter", wake)
at := time.Now()
@@ -310,16 +356,262 @@ func testLargeBoxShutdownMidStartup(t *testing.T) {
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")
- }
+ a.assertNeverStarted(t)
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)
}
+// --- runAgent's wiring ------------------------------------------------------------
+
+// TestRunAgentStartupWiring reads runAgent's source and fails if its
+// startup wiring drifts from what runLargeBoxAgent rebuilds, or if
+// anything on runAgent's path to the edge waits for startup to finish.
+// It checks structure, not text, so comments, formatting and unrelated
+// statements can change freely.
+func TestRunAgentStartupWiring(t *testing.T) {
+ fset := token.NewFileSet()
+ var fn *ast.FuncDecl
+ files, err := filepath.Glob("*.go")
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, name := range files {
+ if strings.HasSuffix(name, "_test.go") {
+ continue
+ }
+ f, err := parser.ParseFile(fset, name, nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, d := range f.Decls {
+ if fd, ok := d.(*ast.FuncDecl); ok && fd.Recv == nil && fd.Name.Name == "runAgent" {
+ fn = fd
+ }
+ }
+ }
+ if fn == nil {
+ t.Fatal("runAgent not found in package host")
+ }
+ str := types.ExprString
+
+ // The calls and statements this test anchors on, by position.
+ var (
+ run *ast.CallExpr // activatorStart{...}.Run(startupCtx)
+ lit *ast.CompositeLit // its activatorStart literal
+ edgeCall *ast.CallExpr // edge.RunWithActivatorReady(...)
+ latch token.Pos // svc.Startup = started
+ cancels []token.Pos // cancelStartup() (a defer of it comes first)
+ joins []token.Pos // joinAgentStartup(..., startupDone)
+ stops []token.Pos // svc.Shutdown(...)
+ bootPass *ast.GoStmt // the go statement running runWorkerBootPass
+ waiters = map[string]bool{} // local closures that wait on the latch
+ )
+ ast.Inspect(fn.Body, func(n ast.Node) bool {
+ switch n := n.(type) {
+ case *ast.CallExpr:
+ switch s := str(n.Fun); {
+ case s == "edge.RunWithActivatorReady":
+ edgeCall = n
+ case s == "cancelStartup":
+ cancels = append(cancels, n.Pos())
+ case s == "joinAgentStartup" && len(n.Args) == 2 && str(n.Args[1]) == "startupDone":
+ joins = append(joins, n.Pos())
+ case s == "svc.Shutdown":
+ stops = append(stops, n.Pos())
+ }
+ if sel, ok := n.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" {
+ if cl, ok := sel.X.(*ast.CompositeLit); ok && str(cl.Type) == "activatorStart" {
+ run, lit = n, cl
+ }
+ }
+ case *ast.AssignStmt:
+ if len(n.Lhs) == 1 && len(n.Rhs) == 1 && str(n.Lhs[0]) == "svc.Startup" && str(n.Rhs[0]) == "started" {
+ latch = n.Pos()
+ }
+ if len(n.Lhs) == 1 && len(n.Rhs) == 1 {
+ if fl, ok := n.Rhs[0].(*ast.FuncLit); ok && receivesFrom(fl.Body, "started") {
+ waiters[str(n.Lhs[0])] = true
+ }
+ }
+ case *ast.GoStmt:
+ if fl, ok := n.Call.Fun.(*ast.FuncLit); ok && callsNamed(fl.Body, "runWorkerBootPass") {
+ bootPass = n
+ }
+ }
+ return true
+ })
+ if run == nil || edgeCall == nil {
+ t.Fatalf("runAgent: activatorStart{...}.Run found %v, edge.RunWithActivatorReady found %v", run != nil, edgeCall != nil)
+ }
+
+ // The order the child copies: the latch set, the start sequence handed
+ // the startup context, the edge, then cancel, join, Shutdown (the first
+ // of each after the edge returns).
+ afterEdge := func(ps []token.Pos) token.Pos {
+ for _, p := range ps {
+ if p > edgeCall.Pos() {
+ return p
+ }
+ }
+ return token.NoPos
+ }
+ cancel, join, shutdown := afterEdge(cancels), afterEdge(joins), afterEdge(stops)
+ if len(run.Args) != 1 || str(run.Args[0]) != "startupCtx" {
+ t.Errorf("activatorStart.Run is given %v, want startupCtx", run.Args)
+ }
+ order := []struct {
+ name string
+ pos token.Pos
+ }{
+ {"svc.Startup = started", latch}, {"activatorStart{...}.Run", run.Pos()}, {"edge.RunWithActivatorReady", edgeCall.Pos()},
+ {"cancelStartup()", cancel}, {"joinAgentStartup(_, startupDone)", join}, {"svc.Shutdown", shutdown},
+ }
+ for i, o := range order {
+ if o.pos == token.NoPos {
+ t.Errorf("runAgent: %s not found", o.name)
+ } else if i > 0 && order[i-1].pos != token.NoPos && o.pos < order[i-1].pos {
+ t.Errorf("runAgent: %s comes before %s", o.name, order[i-1].name)
+ }
+ }
+
+ // The activatorStart fields the child copies.
+ fields := map[string]ast.Expr{}
+ for _, el := range lit.Elts {
+ if kv, ok := el.(*ast.KeyValueExpr); ok {
+ fields[str(kv.Key)] = kv.Value
+ }
+ }
+ for field, want := range map[string]string{
+ "Start": "svc.Start", "Publish": "armedLayoutPass.Store", "Slot": "svc.PrepareScheduledToken",
+ "Routes": "reg.List", "Route": "reg.Get", "Claims": "claims.Writes.Enabled",
+ } {
+ if got := fields[field]; got == nil || str(got) != want {
+ t.Errorf("activatorStart.%s is %v, want %s", field, exprOrNil(got), want)
+ }
+ }
+ funcField := func(name string) *ast.FuncLit {
+ fl, _ := fields[name].(*ast.FuncLit)
+ if fl == nil {
+ t.Errorf("activatorStart.%s is not a func literal", name)
+ }
+ return fl
+ }
+ if fl := funcField("Ready"); fl != nil && !callsWith(fl.Body, "close", "started") {
+ t.Error("activatorStart.Ready does not close started")
+ }
+ if fl := funcField("Sweep"); fl != nil && !callsNamed(fl.Body, "eb.Control.Sweep") {
+ t.Error("activatorStart.Sweep does not call eb.Control.Sweep")
+ }
+ if fl := funcField("Recover"); fl != nil {
+ var seq []string
+ ast.Inspect(fl.Body, func(n ast.Node) bool {
+ if c, ok := n.(*ast.CallExpr); ok {
+ switch s := str(c.Fun); s {
+ case "newLayoutPass", "startupRecoveryIntent", "recoverLayoutAtStart":
+ seq = append(seq, s)
+ }
+ }
+ return true
+ })
+ if strings.Join(seq, ",") != "newLayoutPass,startupRecoveryIntent,recoverLayoutAtStart" {
+ t.Errorf("activatorStart.Recover calls %v, want newLayoutPass, startupRecoveryIntent, recoverLayoutAtStart in that order", seq)
+ }
+ }
+
+ // The worker boot pass waits on the latch or the startup context, then
+ // lists and wakes.
+ if bootPass == nil {
+ t.Error("runAgent: no goroutine runs runWorkerBootPass")
+ } else if body := bootPass.Call.Fun.(*ast.FuncLit).Body; !receivesFrom(body, "started") || !callsNamed(body, "startupCtx.Done") {
+ t.Error("runAgent: the worker boot pass does not wait on started and startupCtx.Done")
+ }
+
+ // Nothing before the edge waits for startup. Outside func literals and
+ // go statements: no receive at all, no join, and a closure that waits
+ // on the latch is only called directly on a claim box, where Run has
+ // already started the activator before it returns.
+ var walk func(n ast.Node, claimBranch bool)
+ walk = func(n ast.Node, claimBranch bool) {
+ ast.Inspect(n, func(n ast.Node) bool {
+ if n == nil || n.Pos() >= edgeCall.Pos() {
+ return false
+ }
+ switch n := n.(type) {
+ case *ast.FuncLit, *ast.GoStmt:
+ return false
+ case *ast.IfStmt:
+ if n.Init != nil {
+ walk(n.Init, claimBranch)
+ }
+ walk(n.Cond, claimBranch)
+ walk(n.Body, claimBranch || str(n.Cond) == "claims.Writes.Enabled")
+ if n.Else != nil {
+ walk(n.Else, claimBranch)
+ }
+ return false
+ case *ast.UnaryExpr:
+ if n.Op == token.ARROW {
+ t.Errorf("runAgent waits on %s at %s, before the edge listens", str(n.X), fset.Position(n.Pos()))
+ }
+ case *ast.CallExpr:
+ s := str(n.Fun)
+ if s == "joinAgentStartup" {
+ t.Errorf("runAgent joins startup at %s, before the edge listens", fset.Position(n.Pos()))
+ }
+ if waiters[s] && !claimBranch {
+ t.Errorf("runAgent calls %s, which waits on the latch, at %s, before the edge listens", s, fset.Position(n.Pos()))
+ }
+ }
+ return true
+ })
+ }
+ walk(fn.Body, false)
+}
+
+// receivesFrom reports whether body receives from the channel named ch
+// (in a select case or a bare receive).
+func receivesFrom(body ast.Node, ch string) bool {
+ found := false
+ ast.Inspect(body, func(n ast.Node) bool {
+ if u, ok := n.(*ast.UnaryExpr); ok && u.Op == token.ARROW && types.ExprString(u.X) == ch {
+ found = true
+ }
+ return !found
+ })
+ return found
+}
+
+// callsNamed reports whether body calls a function whose expression is name.
+func callsNamed(body ast.Node, name string) bool {
+ found := false
+ ast.Inspect(body, func(n ast.Node) bool {
+ if c, ok := n.(*ast.CallExpr); ok && types.ExprString(c.Fun) == name {
+ found = true
+ }
+ return !found
+ })
+ return found
+}
+
+// callsWith reports whether body calls name with arg as its only argument.
+func callsWith(body ast.Node, name, arg string) bool {
+ found := false
+ ast.Inspect(body, func(n ast.Node) bool {
+ if c, ok := n.(*ast.CallExpr); ok && types.ExprString(c.Fun) == name && len(c.Args) == 1 && types.ExprString(c.Args[0]) == arg {
+ found = true
+ }
+ return !found
+ })
+ return found
+}
+
+func exprOrNil(e ast.Expr) string {
+ if e == nil {
+ return "unset"
+ }
+ return types.ExprString(e)
+}
+
// --- the box ------------------------------------------------------------------
type largeBox struct {
@@ -330,28 +622,50 @@ type largeBox struct {
client *http.Client
}
+// largeBoxSocketDir makes the box's run directory. It must be short: a
+// nested instance's socket is <dir>/instances/<instance id>/app.sock, the
+// activator binds a ".control" socket beside it, and the whole path must
+// fit a unix socket address (provision.MaxUnixPath). The default temp dir
+// is tried first, then a directory of the test's own directly under /tmp;
+// when neither is short enough the test is skipped, not failed, since
+// nothing about the code under test is wrong.
+func largeBoxSocketDir(t *testing.T) string {
+ t.Helper()
+ fits := func(dir string) bool {
+ return len(filepath.Join(dir, "instances", provision.InstanceID(largeBoxHost(0)), "app.sock.control")) <= provision.MaxUnixPath
+ }
+ var tried []string
+ for _, base := range []string{os.TempDir(), "/tmp"} {
+ dir, err := os.MkdirTemp(base, "lb")
+ if err != nil {
+ tried = append(tried, base+": "+err.Error())
+ continue
+ }
+ if fits(dir) {
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+ }
+ os.RemoveAll(dir)
+ tried = append(tried, dir+": too long")
+ }
+ t.Skipf("no directory short enough for a nested instance's control socket (%d bytes at most): %s", provision.MaxUnixPath, strings.Join(tried, "; "))
+ return ""
+}
+
// 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.
+// a scheduled delivery due on one instance, an always-on worker the boot
+// pass starts, 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)
- }
+ sd := largeBoxSocketDir(t)
+ b := &largeBox{dataDir: t.TempDir(), storeDir: t.TempDir(), acmeDir: t.TempDir(), socketDir: sd}
+ var err error
b.reg, err = registry.Open(filepath.Join(b.dataDir, "registry.db"))
if err != nil {
t.Fatal(err)
@@ -368,6 +682,18 @@ func newLargeBox(t *testing.T) *largeBox {
t.Fatal(err)
}
}
+ if err := b.reg.Upsert(registry.Route{
+ Host: largeBoxWorker, Kind: "sidecar", Account: "acct", App: "shop", Status: "ready", Args: `["serve"]`,
+ }); err != nil {
+ t.Fatal(err)
+ }
+ due := time.Now().Unix() - 5
+ if err := b.reg.UpsertTimer(registry.Timer{
+ Host: largeBoxHost(largeBoxDueHost), Name: "tick", Kind: schedule.KindSchedule, Spec: "every 6h0m0s",
+ Path: largeBoxDuePath, NextAt: due, DueAt: due,
+ }); err != nil {
+ t.Fatal(err)
+ }
for i := 0; i < largeBoxJournals; i++ {
crash := "after-journal"
switch {
@@ -383,14 +709,13 @@ func newLargeBox(t *testing.T) *largeBox {
}
appendIntent(t, release.NewDirStore(b.storeDir), release.LayoutModeNested, "")
- served := []string{largeBoxHost(largeBoxWakeHost)}
+ served := []string{largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxSidecarHost)}
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{
@@ -542,21 +867,31 @@ type largeBoxAnswer struct {
body string
header http.Header
err error
+ at time.Time // when the answer was read
}
-func (b *largeBox) get(host string) largeBoxAnswer {
- resp, err := b.client.Get("https://" + host + "/")
+// get asks the edge for host's root over TLS. sidecar marks the request
+// as the host's sidecar's (activator.SidecarHeader).
+func (b *largeBox) get(host string, sidecar bool) largeBoxAnswer {
+ req, err := http.NewRequest(http.MethodGet, "https://"+host+"/", nil)
if err != nil {
- return largeBoxAnswer{err: err}
+ return largeBoxAnswer{err: err, at: time.Now()}
+ }
+ if sidecar {
+ req.Header.Set(activator.SidecarHeader, "1")
+ }
+ resp, err := b.client.Do(req)
+ if err != nil {
+ return largeBoxAnswer{err: err, at: time.Now()}
}
defer resp.Body.Close()
body, err := io.ReadAll(resp.Body)
- return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err}
+ return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err, at: time.Now()}
}
-func (b *largeBox) fire(host string) chan largeBoxAnswer {
+func (b *largeBox) fire(host string, sidecar bool) chan largeBoxAnswer {
ch := make(chan largeBoxAnswer, 1)
- go func() { ch <- b.get(host) }()
+ go func() { ch <- b.get(host, sidecar) }()
return ch
}
@@ -686,6 +1021,7 @@ func (c *largeBoxControl) systemctl(ctx context.Context, args ...string) (string
type largeBoxAgent struct {
cmd *exec.Cmd
began time.Time
+ listen time.Duration // process start to the gate probe's first answer
eventsPath string
log string
exited chan struct{} // closed once the process has been waited for
@@ -693,7 +1029,34 @@ type largeBoxAgent struct {
b *largeBox
}
+// startAgent starts the agent and returns it once the gate's probe is
+// answered and its TLS listener accepts. The edge's -http and -https take
+// an address, and :0 would bind, but the edge never says which port it
+// got, so the test picks free ports and hands them over. A port taken in
+// between makes the agent's listener fail and the agent exit, which is
+// told apart from any other failure by its log, and the agent is started
+// again on fresh ports.
func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
+ t.Helper()
+ for attempt := 1; ; attempt++ {
+ b.httpAddr, b.httpsAddr = freeLoopbackAddr(t), freeLoopbackAddr(t)
+ a := b.launchAgent(t)
+ err := a.awaitListening()
+ if err == nil {
+ err = a.awaitTLS()
+ }
+ if err == nil {
+ return a
+ }
+ if a.collided() && attempt < largeBoxBindAttempts {
+ t.Logf("attempt %d: a port was taken before the agent bound it; starting again on fresh ports", attempt)
+ continue
+ }
+ t.Fatal(err)
+ }
+}
+
+func (b *largeBox) launchAgent(t *testing.T) *largeBoxAgent {
t.Helper()
cfgDir := t.TempDir()
cfg := largeBoxConfig{
@@ -718,6 +1081,7 @@ func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
if err != nil {
t.Fatal(err)
}
+ t.Cleanup(func() { logFile.Close() })
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.
@@ -729,7 +1093,6 @@ func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
}
go func() {
a.exitErr = a.cmd.Wait()
- logFile.Close()
close(a.exited)
}()
t.Cleanup(func() {
@@ -751,13 +1114,12 @@ func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent {
return a
}
-// waitListening probes as the update gate does (systemapply.go
+// awaitListening 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
+// passes. It records 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()
+func (a *largeBoxAgent) awaitListening() error {
probe := &http.Client{
Timeout: 3 * time.Second,
Transport: &http.Transport{Proxy: nil, DisableKeepAlives: true},
@@ -768,24 +1130,64 @@ func (a *largeBoxAgent) waitListening(t *testing.T) time.Duration {
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)
+ a.listen = time.Since(a.began)
+ if a.listen > largeBoxListenBound {
+ return fmt.Errorf("gate probe answered %s after the agent started, over %s", a.listen, largeBoxListenBound)
}
- return took
+ return nil
}
select {
case <-a.exited:
- t.Fatalf("agent exited before listening: %v", a.exitErr)
+ return fmt.Errorf("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)
+ return fmt.Errorf("gate probe unanswered %s after the agent started: %v", time.Since(a.began), err)
}
time.Sleep(20 * time.Millisecond)
}
}
+// awaitTLS waits, boundedly, for the edge's TLS listener: the edge binds
+// it in its own goroutine, beside the plain-HTTP one the gate probes, so
+// it can come up a moment later. Not part of the gate's measurement.
+func (a *largeBoxAgent) awaitTLS() error {
+ deadline := time.Now().Add(largeBoxListenBound)
+ for {
+ c, err := net.DialTimeout("tcp", a.b.httpsAddr, time.Second)
+ if err == nil {
+ c.Close()
+ select {
+ case <-a.exited:
+ return fmt.Errorf("agent exited after binding: %v", a.exitErr)
+ default:
+ return nil
+ }
+ }
+ select {
+ case <-a.exited:
+ return fmt.Errorf("agent exited before its TLS listener came up: %v", a.exitErr)
+ default:
+ }
+ if time.Now().After(deadline) {
+ return fmt.Errorf("TLS listener not up %s after the gate probe: %v", largeBoxListenBound, err)
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+}
+
+// collided reports whether the agent exited because a port it was handed
+// had been taken.
+func (a *largeBoxAgent) collided() bool {
+ select {
+ case <-a.exited:
+ case <-time.After(time.Second):
+ return false
+ }
+ out, _ := os.ReadFile(a.log)
+ return strings.Contains(string(out), "address already in use")
+}
+
func (a *largeBoxAgent) signal(t *testing.T, sig os.Signal) {
t.Helper()
if err := a.cmd.Process.Signal(sig); err != nil {
@@ -805,6 +1207,27 @@ func (a *largeBoxAgent) waitExit(t *testing.T, bound time.Duration) {
}
}
+// assertNeverStarted: startup never finished, so the activator never
+// started and nothing reached the backend or ran.
+func (a *largeBoxAgent) assertNeverStarted(t *testing.T) {
+ t.Helper()
+ for _, kind := range []string{"ready", "activator-start", "acquire", "start", "worker-start", "served"} {
+ if evs := a.events(kind); len(evs) != 0 {
+ t.Errorf("%s with startup unfinished: %v", kind, evs)
+ }
+ }
+}
+
+// assertAfter waits briefly for the event kind with args and fails unless
+// it came after ref.
+func (a *largeBoxAgent) assertAfter(t *testing.T, ref largeBoxEvent, kind string, args ...string) {
+ t.Helper()
+ e := a.waitEvent(t, 10*time.Second, kind, args...)
+ if e.at.Before(ref.at) {
+ t.Errorf("%s %v came %s before %s", kind, args, ref.at.Sub(e.at), ref.kind)
+ }
+}
+
// 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.
@@ -956,7 +1379,9 @@ func runLargeBoxAgent(cfgPath string) int {
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)
+ backend := &largeBoxBackend{ev: ev, started: started}
+ svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg)
+ backend.svc = svc
svc.Routes = reg
svc.Startup = started
openStore := func(context.Context) (release.Store, error) { return release.NewDirStore(cfg.StoreDir), nil }
@@ -989,7 +1414,20 @@ func runLargeBoxAgent(cfgPath string) int {
Keys: largeBoxKeys{ev},
Slot: svc.PrepareScheduledToken,
Publish: armedLayoutPass.Store,
- Start: svc.Start,
+ // The checkpoint: when Start runs, every hold marker's host must
+ // already be fenced again and every journal complete.
+ Start: func() {
+ held, _ := layout.HoldMarkers(dd)
+ fenced := 0
+ for _, h := range held {
+ if svc.FenceGeneration(h) != 0 {
+ fenced++
+ }
+ }
+ journals, _ := layout.JournalHosts(dd)
+ ev.log("activator-start", "held="+strconv.Itoa(len(held)), "fenced="+strconv.Itoa(fenced), "journals="+strconv.Itoa(len(journals)))
+ svc.Start()
+ },
}.Run(startupCtx)
go func() {
<-startupDone
@@ -1006,9 +1444,32 @@ func runLargeBoxAgent(cfgPath string) int {
if err != nil {
return
}
- runWorkerBootPass(routes, svc.WakeWorker)
+ runWorkerBootPass(routes, func(rt registry.Route) error {
+ err := svc.WakeWorker(rt)
+ ev.log("worker-boot", rt.Host, err == nil)
+ return err
+ })
}()
+ // Admissions from inside the agent that do not pass runAgent's own
+ // gates, so they meet the activator's latch on their own: a worker
+ // wake as an operator command or the sidecar-wake leg makes it, and
+ // one scheduler tick through the real runner and deliverer with a
+ // delivery already due.
+ if rt, err := reg.Get(largeBoxWorker); err == nil && rt != nil {
+ go func() {
+ ev.log("worker-direct-begin", rt.Host)
+ err := svc.WakeWorker(*rt)
+ ev.log("worker-direct", rt.Host, err == nil)
+ }()
+ }
+ runner := &schedule.Runner{
+ Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg),
+ Now: time.Now, Timeout: largeBoxLatchBound + 15*time.Second, Box: "largebox", Build: "test",
+ }
+ runner.Tick(agentCtx) // delivers in the background, and returns
+ ev.log("tick-done")
+
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) {
@@ -1046,11 +1507,11 @@ func runLargeBoxAgent(cfgPath string) int {
// 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.
+// client, the activator, the scheduler, 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)
@@ -1058,8 +1519,8 @@ func largeBoxStartupGoroutines() []string {
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)",
+ "host.runLargeBoxAgent", "host.(*largeBoxBackend)", "host.largeBoxActivator", "host.(*activatorDeliverer)",
+ "layout.(*Pass)", "instancectl.(*Client)", "activator.(*Service)", "schedule.(*Runner)",
"net/http.(*conn).serve",
}
var left []string
@@ -1090,11 +1551,13 @@ func (a largeBoxActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt
// 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.
+// was started at, and a worker reports ready at once, as a child does
+// over the agent socket. It records every acquire, start and request
+// served, and any acquire the startup latch should have prevented.
type largeBoxBackend struct {
ev *largeBoxEvents
started <-chan struct{}
+ svc *activator.Service
}
func (b *largeBoxBackend) Acquire(_ context.Context, rt registry.Route) (activator.Lease, error) {
@@ -1120,7 +1583,11 @@ func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator
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{})}
+ p := &largeBoxProc{done: make(chan struct{})}
+ p.srv = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ b.ev.log("served", rt.Host, r.URL.Path)
+ io.WriteString(w, body)
+ })}
go func() { defer close(p.done); _ = p.srv.Serve(ln) }()
return p, "", nil
}
@@ -1128,10 +1595,13 @@ func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator
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")
+ return nil, errors.New("largeBoxBackend: no co-spawned 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")
+
+func (b *largeBoxBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) {
+ b.ev.log("worker-start", rt.Host)
+ go b.svc.NoteWorkerReady(rt.Host, "")
+ return &largeBoxWorkerProc{}, "", nil
}
type largeBoxProc struct {
@@ -1149,6 +1619,11 @@ func (p *largeBoxProc) Stop(context.Context) error {
func (p *largeBoxProc) Exited() bool { return p.exited.Load() }
+type largeBoxWorkerProc struct{ exited atomic.Bool }
+
+func (p *largeBoxWorkerProc) Stop(context.Context) error { p.exited.Store(true); return nil }
+func (p *largeBoxWorkerProc) 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 }