amadan
WIP claim Plan 14 track D checkpoint 1 (not for merge): broker recovery, root reader
Co-Authored-By: Codex <noreply@openai.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
20 files changed,
+758
−131
internal/host/claimlaunch_authority.go+6 −17internal/host/claimlaunch_execpermit.go+4 −4internal/host/claimlaunch_rev2_linux_test.go+9 −28internal/host/claimlaunch_root.go+2 −0internal/host/claimlaunch_root_test.go+12 −3internal/host/claimlaunch_transport_test.go+1 −1internal/host/claimparticipation.go+20 −39internal/host/claimproduction_admintoken_test.go+1 −1internal/host/plan9_production_test.go+3 −3internal/host/relocationattempts_linux_test.go+10 −20internal/host/rootexecution_launch_linux_test.go+253 −0internal/host/rootexecution_migration.go+28 −0internal/host/rootexecution_migration_test.go+76 −0internal/host/rootexecution_read.go+35 −5internal/hostexec/rootbroker.go+31 −3internal/hostexec/rootbroker_recovery_test.go+133 −0internal/hostexec/rootclient.go+35 −2internal/hostexec/rootclient_test.go+85 −0internal/hostexec/rootread.go+3 −0internal/provision/testdata/plan14-root-reads-merge.tsv+11 −5
CI passed — run details
CI log
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok amadan.net/carlos/platform.git/launchctx 0.012s
? 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 365.605s
? github.com/carlosframework/platform/cmd/changelog [no test files]
ok github.com/carlosframework/platform/cmd/console 3.218s
ok github.com/carlosframework/platform/cmd/moveprobe 93.397s
ok github.com/carlosframework/platform/internal/activator 234.351s
ok github.com/carlosframework/platform/internal/agentsock 0.017s
ok github.com/carlosframework/platform/internal/alerting 0.004s
ok github.com/carlosframework/platform/internal/analytics 0.022s
ok github.com/carlosframework/platform/internal/appconfig 2.372s
ok github.com/carlosframework/platform/internal/apperr 0.073s
ok github.com/carlosframework/platform/internal/applog 0.137s
ok github.com/carlosframework/platform/internal/awsinv 0.004s
ok github.com/carlosframework/platform/internal/boxchoice 0.004s
ok github.com/carlosframework/platform/internal/boxid 0.797s
? 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.012s
ok github.com/carlosframework/platform/internal/certwarm 0.003s
ok github.com/carlosframework/platform/internal/cgroup 0.198s
ok github.com/carlosframework/platform/internal/changelog 0.249s
ok github.com/carlosframework/platform/internal/clamd 0.057s
? github.com/carlosframework/platform/internal/cliflag [no test files]
ok github.com/carlosframework/platform/internal/console 370.328s
? 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.173s
ok github.com/carlosframework/platform/internal/dbdoor/dbdoortest 0.010s
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.187s
ok github.com/carlosframework/platform/internal/doorkey 0.009s
ok github.com/carlosframework/platform/internal/economics 0.004s
ok github.com/carlosframework/platform/internal/edge 26.331s
? github.com/carlosframework/platform/internal/edgeproof [no test files]
ok github.com/carlosframework/platform/internal/edgesfile 0.010s
ok github.com/carlosframework/platform/internal/fleetapi 0.125s
ok github.com/carlosframework/platform/internal/gate 1.655s
ok github.com/carlosframework/platform/internal/genselect 0.016s
ok github.com/carlosframework/platform/internal/geo 0.015s
ok github.com/carlosframework/platform/internal/geoip 0.030s
ok github.com/carlosframework/platform/internal/host 86.331s
ok github.com/carlosframework/platform/internal/hostactors 0.430s
ok github.com/carlosframework/platform/internal/hostbox 2.377s
ok github.com/carlosframework/platform/internal/hostclaim 0.542s
ok github.com/carlosframework/platform/internal/hostevidence 0.017s
ok github.com/carlosframework/platform/internal/hostexec 0.517s
ok github.com/carlosframework/platform/internal/hostfence 0.033s
ok github.com/carlosframework/platform/internal/hostfiles 11.064s
ok github.com/carlosframework/platform/internal/hostfloor 0.017s
ok github.com/carlosframework/platform/internal/hostguard 0.390s
ok github.com/carlosframework/platform/internal/hostops 1.507s
? github.com/carlosframework/platform/internal/hostops/fencerelease [no test files]
ok github.com/carlosframework/platform/internal/hostops/internal/fencerelease 0.017s
ok github.com/carlosframework/platform/internal/hostquarantine 0.017s
ok github.com/carlosframework/platform/internal/hostrelocate 0.469s
ok github.com/carlosframework/platform/internal/hostresolve 0.147s
ok github.com/carlosframework/platform/internal/hostwrite 0.017s
ok github.com/carlosframework/platform/internal/infraview 0.019s
ok github.com/carlosframework/platform/internal/instancectl 1.722s
ok github.com/carlosframework/platform/internal/instanceexec 2.758s
ok github.com/carlosframework/platform/internal/instancefs 0.020s
ok github.com/carlosframework/platform/internal/instancekey 0.018s
ok github.com/carlosframework/platform/internal/instanceupdates 0.022s
ok github.com/carlosframework/platform/internal/layout 1.472s
ok github.com/carlosframework/platform/internal/lease 9.487s
ok github.com/carlosframework/platform/internal/lease/leasetest 0.006s
ok github.com/carlosframework/platform/internal/ledger 0.012s
ok github.com/carlosframework/platform/internal/mail 0.173s
ok github.com/carlosframework/platform/internal/maintenance 0.085s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.619s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.077s
ok github.com/carlosframework/platform/internal/metrics 0.007s
ok github.com/carlosframework/platform/internal/moves 21.890s
ok github.com/carlosframework/platform/internal/moves/archive 0.765s
ok github.com/carlosframework/platform/internal/objstore 0.184s
ok github.com/carlosframework/platform/internal/ops 0.180s
ok github.com/carlosframework/platform/internal/park 0.218s
ok github.com/carlosframework/platform/internal/privatefile 0.004s
ok github.com/carlosframework/platform/internal/provision 33.454s
ok github.com/carlosframework/platform/internal/publicrelease 0.009s
? github.com/carlosframework/platform/internal/regionname [no test files]
ok github.com/carlosframework/platform/internal/registry 4.411s
ok github.com/carlosframework/platform/internal/relay 7.275s
ok github.com/carlosframework/platform/internal/release 20.971s
ok github.com/carlosframework/platform/internal/rollout 3.762s
ok github.com/carlosframework/platform/internal/rolloutwords 0.003s
ok github.com/carlosframework/platform/internal/rootaws 0.007s
ok github.com/carlosframework/platform/internal/rootparticipation 0.002s
ok github.com/carlosframework/platform/internal/rootstate 0.025s
ok github.com/carlosframework/platform/internal/s3compat 0.004s
ok github.com/carlosframework/platform/internal/safefile 0.004s
ok github.com/carlosframework/platform/internal/schedule 2.632s
ok github.com/carlosframework/platform/internal/sim/hostlife 12.157s
ok github.com/carlosframework/platform/internal/sim/litestreamfacts 4.064s
ok github.com/carlosframework/platform/internal/status 0.042s
ok github.com/carlosframework/platform/internal/steering 0.006s
ok github.com/carlosframework/platform/internal/stopevidence 0.216s
ok github.com/carlosframework/platform/internal/svcrole 0.002s
ok github.com/carlosframework/platform/internal/telemetry 0.004s
ok github.com/carlosframework/platform/internal/termfence 0.004s
ok github.com/carlosframework/platform/internal/tunnelclient 0.004s
ok github.com/carlosframework/platform/internal/unixdial 0.003s
ok github.com/carlosframework/platform/internal/updatewords 0.004s
ok github.com/carlosframework/platform/internal/usage 0.005s
ok github.com/carlosframework/platform/internal/version 0.002s
ok github.com/carlosframework/platform/internal/vet 1.573s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
minio: downloading https://github.com/minio/minio/releases/download/RELEASE.2025-09-07T16-13-09Z/minio.linux-arm64.RELEASE.2025-09-07T16-13-09Z
minio: downloaded, digest verified
mc: downloading https://github.com/minio/mc/releases/download/RELEASE.2025-08-13T08-35-41Z/mc.linux-arm64.RELEASE.2025-08-13T08-35-41Z
mc: downloaded, digest verified
minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
buckets and scoped user ready
=== RUN TestDeliveredContractMinio
=== RUN TestDeliveredContractMinio/put
=== RUN TestDeliveredContractMinio/get_round-trips_the_body
=== RUN TestDeliveredContractMinio/list_shows_the_key
=== RUN TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN TestDeliveredContractMinio/delete_removes_it
=== RUN TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.03s)
--- 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.02s)
--- 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.01s)
--- 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.041s
=== RUN TestT4StorageProbeMinio
2026/10/13 09:30:36 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE0CDB006CF87C, HostID: 7c420053e0fbcd7f41653fdd92ffa2802c903d05303b1aa15adc2e8bb65d183f, api error SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your key and signing method.
--- PASS: TestT4StorageProbeMinio (0.01s)
PASS
ok github.com/carlosframework/platform/internal/host 0.027s
minio contract: PASS
diff --git a/internal/host/claimlaunch_authority.go b/internal/host/claimlaunch_authority.go| index e7258733b..222db2b17 100644 |
| --- a/internal/host/claimlaunch_authority.go |
| +++ b/internal/host/claimlaunch_authority.go |
| @@ -4,7 +4,6 @@ import ( |
| "context" |
| "errors" |
| "github.com/carlosframework/platform/internal/park" |
| - "os" |
| "path/filepath" |
| "strconv" |
| "time" |
| @@ -110,24 +109,14 @@ func (h *ClaimLaunchRoot) authorize(ctx context.Context, peer StopPeer, q hostbo |
| if err := h.Local.Check(ctx, actor, c, selection); err != nil { |
| return err |
| } |
| - // Serving has no execution lease; a leftover operation state denies it. |
| - state, localErr := execlease.InspectLocal(h.Guard.DataDir, t.Host) |
| - if c.Op.Kind == hostclaim.KindServing { |
| - if !errors.Is(localErr, os.ErrNotExist) { |
| - return errors.Join(execlease.ErrLost, localErr) |
| - } |
| - } else { |
| - if localErr != nil { |
| - return localErr |
| - } |
| - // The operation lease has its own controller ID. Match its process to |
| - // the authenticated agent, and its binding to the current claim/boot. |
| - if state.Host != t.Host || state.Op != c.Op.ID || state.Incarnation != t.Incarnation || state.Box != t.Box || state.Enrollment != t.Enrollment || state.BootID != peer.BootID || state.Process.PID != peer.PID || state.Process.Start != peer.ProcessStart || state.Process.KernelBootID != peer.BootID { |
| + state, err := h.readExecution(ctx, c) |
| + if err != nil { |
| + return err |
| + } |
| + if c.Op.Kind != hostclaim.KindServing { |
| + if state.Authority.Box != t.Box || state.Authority.Enrollment != t.Enrollment || state.Peer.KernelBootID != peer.BootID || state.Peer.PID != peer.PID || state.Peer.Start != peer.ProcessStart { |
| return execlease.ErrLost |
| } |
| - if err := execlease.CheckLocal(h.Guard.DataDir, state.Binding, h.Clock, h.ExecLockDir); err != nil { |
| - return err |
| - } |
| } |
| now, boot, err := h.Clock.Now() |
| if err != nil { |
diff --git a/internal/host/claimlaunch_execpermit.go b/internal/host/claimlaunch_execpermit.go| index b6d805ae8..098aa03ed 100644 |
| --- a/internal/host/claimlaunch_execpermit.go |
| +++ b/internal/host/claimlaunch_execpermit.go |
| @@ -25,11 +25,11 @@ func (h *ClaimLaunchRoot) execPermit(ctx context.Context, q hostbox.RoleLaunchRe |
| // Serving also gets a finite start window. An operation's lease may only |
| // shorten it; a later renewal cannot extend this reservation's permit. |
| until := boot + int64(30*time.Second) |
| + state, err := h.readExecution(ctx, o.Claim) |
| + if err != nil { |
| + return instancectl.ClaimExecPermit{}, err |
| + } |
| if o.Claim.Op.Kind != hostclaim.KindServing { |
| - state, err := execlease.InspectLocal(h.Guard.DataDir, r.Target.Host) |
| - if err != nil { |
| - return instancectl.ClaimExecPermit{}, err |
| - } |
| if state.SafeUntil < until { |
| until = state.SafeUntil |
| } |
diff --git a/internal/host/claimlaunch_rev2_linux_test.go b/internal/host/claimlaunch_rev2_linux_test.go| index 5a40fcad6..c47e64e83 100644 |
| --- a/internal/host/claimlaunch_rev2_linux_test.go |
| +++ b/internal/host/claimlaunch_rev2_linux_test.go |
| @@ -19,7 +19,6 @@ import ( |
| "github.com/carlosframework/platform/internal/hostwrite" |
| "github.com/carlosframework/platform/internal/instancectl" |
| "github.com/carlosframework/platform/internal/provision" |
| - "golang.org/x/sys/unix" |
| ) |
| type rev2Local struct{ err error } |
| @@ -71,37 +70,19 @@ func configureRev2Launch(t *testing.T, f *launchFixture) { |
| }) |
| } else { |
| f.root.Attempts = map[string]claimLaunchAttemptReader{f.claims.obs.Claim.Op.Kind: &rev2Attempts{attempt: a, op: f.claims.obs.Claim.Op, stamp: f.target.Stamp, session: f.target.Session}} |
| - writeRev2Local(t, f, "active", 1000) |
| - for _, suffix := range []string{".lock", ".live"} { |
| - file, err := os.OpenFile(filepath.Join(f.root.ExecLockDir, f.target.Host+suffix), os.O_CREATE|os.O_RDWR, 0660) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := file.Chmod(0660); err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil { |
| - t.Fatal(err) |
| - } |
| - t.Cleanup(func() { file.Close() }) |
| - } |
| } |
| + configureLaunchParticipation(t, f) |
| } |
| -func writeRev2Local(t *testing.T, f *launchFixture, state string, deadline int64) { |
| +func writeRev2Local(t *testing.T, f *launchFixture, status string, deadline int64) { |
| t.Helper() |
| - s := execlease.State{V: 1, Binding: execlease.Binding{Host: f.target.Host, Incarnation: f.target.Incarnation, Op: f.target.Op, Box: f.target.Box, Enrollment: f.target.Enrollment, BootID: f.peer.BootID, ControllerID: "lease-controller"}, Process: execlease.ControllerProcess{PID: f.peer.PID, Start: f.peer.ProcessStart, KernelBootID: f.peer.BootID}, State: state, LeaseETag: "lease-etag", Expires: time.Now().Add(time.Hour), SafeUntil: deadline} |
| - dir := filepath.Join(f.root.Guard.DataDir, ".hostexec") |
| - if err := os.MkdirAll(dir, 0700); err != nil { |
| - t.Fatal(err) |
| - } |
| - raw, err := json.Marshal(s) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := os.WriteFile(filepath.Join(dir, f.target.Host+".json"), raw, 0600); err != nil { |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "begin", Operation: "executionBegin", Host: f.target.Host, Incarnation: f.target.Incarnation, Stamp: f.target.Stamp, Session: f.target.Session, Op: f.target.Op} |
| + r := execlease.ExecutionRecord{Schema: 1, Request: q, Peer: execlease.ControllerProcess{KernelBootID: f.peer.BootID, PID: f.peer.PID, Start: f.peer.ProcessStart}, Authority: execlease.ExecutionAuthority{Box: f.target.Box, Enrollment: f.target.Enrollment, Migration: 1, Installation: "root-installation"}, Broker: "root", Token: "root-token", Status: status, SafeUntil: deadline, LeaseETag: "root-etag", Expires: time.Now().Add(time.Minute), Replay: map[string]execlease.ExecutionReplay{}} |
| + raw, _ := json.Marshal(r) |
| + if err := f.root.executionState.Write(f.target.Host+".json", raw); err != nil { |
| t.Fatal(err) |
| } |
| } |
| + |
| func rev2Operation() hostclaim.Op { |
| return hostclaim.Op{Kind: hostclaim.KindRelocating, ID: "operation", Phase: "probing", Exec: "box-1", Journal: map[string]string{"install_stamp": "11"}} |
| } |
| @@ -126,11 +107,11 @@ func TestPlan11Rev2OperationVerifierAndLocalState(t *testing.T) { |
| case "admission closed": |
| f.root.Local.(*rev2Local).err = execlease.ErrLost |
| case "released live lock": |
| - if err := os.Remove(filepath.Join(f.root.ExecLockDir, f.target.Host+".live")); err != nil { |
| + if err := os.Remove(filepath.Join(f.root.executionState.(*launchParticipationDisk).dir, f.target.Host+".live")); err != nil { |
| t.Fatal(err) |
| } |
| case "missing state": |
| - if err := os.Remove(filepath.Join(f.root.Guard.DataDir, ".hostexec", f.target.Host+".json")); err != nil { |
| + if err := os.Remove(filepath.Join(f.root.executionState.(*launchParticipationDisk).dir, f.target.Host+".json")); err != nil { |
| t.Fatal(err) |
| } |
| } |
diff --git a/internal/host/claimlaunch_root.go b/internal/host/claimlaunch_root.go| index 980349cb1..1d5b20e31 100644 |
| --- a/internal/host/claimlaunch_root.go |
| +++ b/internal/host/claimlaunch_root.go |
| @@ -18,6 +18,7 @@ import ( |
| execlease "github.com/carlosframework/platform/internal/hostexec" |
| "github.com/carlosframework/platform/internal/hostguard" |
| "github.com/carlosframework/platform/internal/instancectl" |
| + "github.com/carlosframework/platform/internal/rootstate" |
| ) |
| // ErrClaimLaunchUnavailable refuses every launch while a root provider of |
| @@ -58,6 +59,7 @@ type ClaimLaunchObserved struct { |
| // roster identity, its local launch journal and its own view of the kernel. |
| // A request only names which reservation to run. |
| type ClaimLaunchRoot struct { |
| + executionState rootstate.Store |
| ValidateExecutable func(host, path, digest string) error |
| Local hostbox.LocalAuthority |
| Clock execlease.Clock |
diff --git a/internal/host/claimlaunch_root_test.go b/internal/host/claimlaunch_root_test.go| index 9a0d8c712..95bbcd0f0 100644 |
| --- a/internal/host/claimlaunch_root_test.go |
| +++ b/internal/host/claimlaunch_root_test.go |
| @@ -13,6 +13,7 @@ import ( |
| "github.com/carlosframework/platform/internal/cgroup" |
| "github.com/carlosframework/platform/internal/hostbox" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| + execlease "github.com/carlosframework/platform/internal/hostexec" |
| "github.com/carlosframework/platform/internal/hostguard" |
| "github.com/carlosframework/platform/internal/instancectl" |
| ) |
| @@ -133,6 +134,14 @@ type launchFixture struct { |
| func newLaunchFixture(t *testing.T, op hostclaim.Op) *launchFixture { |
| t.Helper() |
| host := "shop.example" |
| + boot, err := execlease.BootID() |
| + if err != nil { |
| + t.Skip("root launch requires Linux") |
| + } |
| + start, err := claimConstructionProcessStart() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| data := t.TempDir() |
| if err := os.Mkdir(filepath.Join(data, "hostguard"), 0o770); err != nil { |
| t.Fatal(err) |
| @@ -151,15 +160,15 @@ func newLaunchFixture(t *testing.T, op hostclaim.Op) *launchFixture { |
| previousAgentRoot := claimContainmentAgentRoot |
| claimContainmentAgentRoot = func(context.Context) (string, error) { return "/sys/fs/cgroup/system.slice/carlos-edge.service", nil } |
| t.Cleanup(func() { claimContainmentAgentRoot = previousAgentRoot }) |
| - target := hostguard.FenceTarget{Region: "eu", Host: host, Incarnation: "inc-1", Stamp: 11, Session: 13, Op: op.ID, Phase: op.Phase, Box: "box-1", Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1", ClaimETag: "etag-1", DirectoryRevision: "7"} |
| - if err = journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 42, ProcessStart: "start-42", Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: filepath.Join("system.slice/carlos-edge.service", cgroup.ExecName(claimContainmentName(target))), Device: 1, Inode: 9}}, NoState: true}); err != nil { |
| + target := hostguard.FenceTarget{Region: "eu", Host: host, Incarnation: "inc-1", Stamp: 11, Session: 13, Op: op.ID, Phase: op.Phase, Box: "box-1", Enrollment: "enrollment-1", BootID: boot, ControllerID: "controller-1", ClaimETag: "etag-1", DirectoryRevision: "7"} |
| + if err = journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: os.Getpid(), ProcessStart: start, Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: filepath.Join("system.slice/carlos-edge.service", cgroup.ExecName(claimContainmentName(target))), Device: 1, Inode: 9}}, NoState: true}); err != nil { |
| t.Fatal(err) |
| } |
| actor := &hostbox.Actor{Box: "box-1", Enrollment: "enrollment-1", Revision: 7, Region: "eu", Eligible: true} |
| exe := hostbox.LaunchFile{Device: 5, Inode: 6} |
| units := &launchUnits{started: instancectl.ClaimUnitStarted{Unit: instancectl.UnitName(instancectl.RoleRun, host), InvocationID: "unit-inv-1", ControlGroup: instancectl.ClaimUnitControlGroup(host), MainPID: 4242}} |
| kernel := &launchKernel{path: "/opt/carlos/versions/v1/app", want: exe, observed: ClaimLaunchObserved{ProcessStart: "ps-4242", TenantPID: 4243, TenantStart: "ps-4243", Group: hostbox.LaunchFile{Device: 3, Inode: 4}, Executable: exe}} |
| - peer := StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"} |
| + peer := StopPeer{UID: 1001, PID: os.Getpid(), ProcessStart: start, BootID: boot, ControllerID: "controller-1"} |
| root := &ClaimLaunchRoot{ValidateExecutable: func(string, string, string) error { return nil }, Guard: guard, Journal: journal, Units: units, Kernel: kernel, Actor: func(context.Context) (hostbox.Actor, error) { return *actor, nil }} |
| reservation := hostbox.LaunchReservation{ID: "attempt-1:main", Role: "main", InvocationID: "inv-1", Inputs: "inputs-1", Target: target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(host)}, Executable: exe} |
| req := hostbox.RoleLaunchRequest{V: hostbox.RoleLaunchV, Op: hostbox.RoleLaunchStart, Reservation: &reservation, Executable: kernel.path, MemoryMax: cgroup.TierDefault.MemoryMax} |
diff --git a/internal/host/claimlaunch_transport_test.go b/internal/host/claimlaunch_transport_test.go| index 53f11f41c..80547849c 100644 |
| --- a/internal/host/claimlaunch_transport_test.go |
| +++ b/internal/host/claimlaunch_transport_test.go |
| @@ -149,7 +149,7 @@ func TestClaimRoleLaunchEndToEndThroughTheRootProtocol(t *testing.T) { |
| t.Fatal(err) |
| } |
| launch := activator.RoleLaunch{Work: work, Target: f.target, Root: NewClaimRoleRootClient(sock), LaunchDir: launchDir, RunDir: runDir, LogSocket: filepath.Join(base, "l.sock"), MemoryMax: cgroup.TierDefault.MemoryMax, Executable: func(context.Context, *exec.Cmd) (*os.File, error) { return os.Open(exe) }} |
| - attempt := hostclaim.WakeRetryAttempt{ID: "attempt-1", InvocationID: "inv-1", LaunchInputs: "inputs-1", Session: 13, Box: "box-1", Enrollment: "enrollment-1", BootID: "boot-1", ControllerID: "controller-1"} |
| + attempt := hostclaim.WakeRetryAttempt{ID: "attempt-1", InvocationID: "inv-1", LaunchInputs: "inputs-1", Session: 13, Box: "box-1", Enrollment: "enrollment-1", BootID: f.peer.BootID, ControllerID: "controller-1"} |
| build := func() *exec.Cmd { |
| cmd := exec.Command("/opt/carlos/live/shop.example", "--socket", filepath.Join(runDir, "app.sock"), "--db", filepath.Join(launchDir, "app.db")) |
| cmd.Env = []string{"CARLOS_ADMIN_TOKEN=test"} |
diff --git a/internal/host/claimparticipation.go b/internal/host/claimparticipation.go| index b1eec455c..c7ddfc046 100644 |
| --- a/internal/host/claimparticipation.go |
| +++ b/internal/host/claimparticipation.go |
| @@ -30,18 +30,19 @@ type claimParticipation struct { |
| CleanupSucceeded bool |
| } |
| type productionClaimLocal struct { |
| - Resources ClaimRuntimeResources |
| - Identity hostbox.IdentityProvider |
| - Directory hostactors.Directory |
| - Store release.Store |
| - mu sync.Mutex |
| - slots map[string]*claimParticipation |
| - assembly *ClaimAssembly |
| - lifetime context.Context |
| - closed bool |
| - loopCancel context.CancelFunc |
| - loopDone chan struct{} |
| - stopIdentity *claimStopLocal |
| + executionRead func(context.Context, *hostclaim.Claim) (execlease.ExecutionRecord, error) |
| + Resources ClaimRuntimeResources |
| + Identity hostbox.IdentityProvider |
| + Directory hostactors.Directory |
| + Store release.Store |
| + mu sync.Mutex |
| + slots map[string]*claimParticipation |
| + assembly *ClaimAssembly |
| + lifetime context.Context |
| + closed bool |
| + loopCancel context.CancelFunc |
| + loopDone chan struct{} |
| + stopIdentity *claimStopLocal |
| // lockRoot is a test seam (region HA N12). Set, the participation's |
| // hostguard and hostexec locks live under it and its controllers skip |
| // the update-floor check (their UpdateGuardDir is empty; main's floor |
| @@ -109,22 +110,11 @@ func (p *productionClaimLocal) prepareWith(ctx context.Context, actor hostbox.Ac |
| if closed || a == nil || lifetime == nil { |
| return hostwrite.ErrUnavailable |
| } |
| + if p.Resources.Mode != "agent" { |
| + return p.checkRootParticipation(ctx, actor, c) |
| + } |
| if c.Op.Kind == hostclaim.KindServing || c.Free() || !slices.Contains(c.Op.Participants, actor.Box) { |
| - if p.Resources.Mode != "agent" { |
| - if old != nil { |
| - return execlease.ErrInactive |
| - } |
| - _, err := execlease.InspectLocal(p.Resources.DataDir, o.Host) |
| - if errors.Is(err, os.ErrNotExist) { |
| - return nil |
| - } |
| - if err != nil { |
| - return err |
| - } |
| - // Root observers must not acquire CC9 locks. Only the owning agent can |
| - // clear stale state while excluding its next Open across the claim read. |
| - return execlease.ErrInactive |
| - } |
| + |
| if old == nil { |
| state, err := execlease.InspectLocal(p.Resources.DataDir, o.Host) |
| if errors.Is(err, os.ErrNotExist) { |
| @@ -181,18 +171,6 @@ func (p *productionClaimLocal) prepareWith(ctx context.Context, actor hostbox.Ac |
| _, err = old.Controller.Ready(ctx) |
| return err |
| } |
| - if p.Resources.Mode != "agent" { |
| - state, err := execlease.InspectLocal(p.Resources.DataDir, o.Host) |
| - if err != nil { |
| - return err |
| - } |
| - if state.Op != c.Op.ID || state.Incarnation != c.Holder.Incarnation || state.Box != actor.Box || state.Enrollment != actor.Enrollment { |
| - return execlease.ErrState |
| - } |
| - return a.Guard.WithLocal(ctx, o.Host, func() error { |
| - return execlease.CheckLocal(p.Resources.DataDir, state.Binding, execlease.SystemClock{}, "") |
| - }) |
| - } |
| boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id") |
| if err != nil { |
| return err |
| @@ -286,6 +264,9 @@ func (p *productionClaimLocal) Check(ctx context.Context, a hostbox.Actor, c *ho |
| if closed { |
| return execlease.ErrLost |
| } |
| + if p.Resources.Mode != "agent" { |
| + return p.checkRootParticipation(ctx, a, c) |
| + } |
| if c.Op.Kind == hostclaim.KindServing { |
| if slot != nil { |
| return execlease.ErrLost |
diff --git a/internal/host/claimproduction_admintoken_test.go b/internal/host/claimproduction_admintoken_test.go| index 2af3216b0..2f15eb1ed 100644 |
| --- a/internal/host/claimproduction_admintoken_test.go |
| +++ b/internal/host/claimproduction_admintoken_test.go |
| @@ -72,7 +72,7 @@ func newAdminTokenFixture(t *testing.T, execBox string) *adminTokenFixture { |
| t.Cleanup(func() { reg.Close() }) |
| f := &adminTokenFixture{t: t, claims: st, data: data, backend: &activator.ExecBackend{DataDir: data, OptDir: opt}, actor: hostbox.Actor{Box: "box-1", Region: "ie", Enrollment: "enr-1"}} |
| f.rt = registry.Route{Host: adminTokenHost, Socket: filepath.Join(root, "run", "app.sock"), DBPath: filepath.Join(data, adminTokenHost+".db")} |
| - local := &productionClaimLocal{Resources: ClaimRuntimeResources{DataDir: data}, slots: map[string]*claimParticipation{}} |
| + local := &productionClaimLocal{Resources: ClaimRuntimeResources{Mode: "agent", DataDir: data}, slots: map[string]*claimParticipation{}} |
| f.tokens = claimAdminTokens{ |
| Guard: hostguard.Guard{DataDir: data, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: st}}, |
| Claims: st, |
diff --git a/internal/host/plan9_production_test.go b/internal/host/plan9_production_test.go| index 9bc3377e4..cec6ac601 100644 |
| --- a/internal/host/plan9_production_test.go |
| +++ b/internal/host/plan9_production_test.go |
| @@ -78,7 +78,7 @@ func TestPlan9ProductionLocalServingDoesNotCreateLease(t *testing.T) { |
| if err := os.Mkdir(filepath.Join(dir, ".hostexec"), 0700); err != nil { |
| t.Fatal(err) |
| } |
| - p := productionClaimLocal{Resources: ClaimRuntimeResources{DataDir: dir}, slots: map[string]*claimParticipation{}} |
| + p := productionClaimLocal{Resources: ClaimRuntimeResources{Mode: "agent", DataDir: dir}, slots: map[string]*claimParticipation{}} |
| c := &hostclaim.Claim{Host: "x.example", Holder: &hostclaim.Holder{Incarnation: "inc"}, Op: hostclaim.Op{Kind: hostclaim.KindServing}} |
| actor := hostbox.Actor{Box: "box", Region: "ie"} |
| selection := hostbox.Selection{Box: "box", Region: "ie", Captured: hostguard.Captured{Incarnation: "inc"}} |
| @@ -199,8 +199,8 @@ func TestPlan9RootAndAdoptLeaveStaleParticipationForAgent(t *testing.T) { |
| // No cleanup dependencies: a root observer must refuse before touching |
| // the journal, root socket or lifetime locks. |
| p := productionClaimLocal{Resources: ClaimRuntimeResources{Mode: mode, DataDir: f.guard.DataDir}, assembly: &ClaimAssembly{}, lifetime: t.Context(), slots: map[string]*claimParticipation{}} |
| - if err = p.prepareWith(t.Context(), actor, o); !errors.Is(err, execlease.ErrInactive) { |
| - t.Fatalf("root attempted stale cleanup: %v", err) |
| + if err = p.prepareWith(t.Context(), actor, o); err == nil { |
| + t.Fatal("root admitted without root participation") |
| } |
| after, err := os.ReadFile(path) |
| if err != nil { |
diff --git a/internal/host/relocationattempts_linux_test.go b/internal/host/relocationattempts_linux_test.go| index cc61728e9..a5f713f88 100644 |
| --- a/internal/host/relocationattempts_linux_test.go |
| +++ b/internal/host/relocationattempts_linux_test.go |
| @@ -167,7 +167,15 @@ func TestRootAuthorizesAReservationTheRelocationJournalHolds(t *testing.T) { |
| admitted := f.admit(f.read(), in) |
| h := admitted.Claim.Holder |
| enrollment := strings.Repeat("a", 32) |
| - peer := StopPeer{UID: 1001, PID: 42, ProcessStart: "start-42", BootID: "boot-1", ControllerID: "controller-1"} |
| + boot, err := execlease.BootID() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + start, err := claimConstructionProcessStart() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + peer := StopPeer{UID: 1001, PID: os.Getpid(), ProcessStart: start, BootID: boot, ControllerID: "controller-1"} |
| target := hostguard.FenceTarget{Region: "ie", Host: roHost, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Op: roOp, Phase: hostclaim.PhaseRelocatingProbing, Box: "dst", Enrollment: enrollment, BootID: peer.BootID, ControllerID: peer.ControllerID, ClaimETag: admitted.ETag, DirectoryRevision: "7"} |
| // The agent's side: the journal Reserve writes for the admitted session. |
| @@ -191,25 +199,7 @@ func TestRootAuthorizesAReservationTheRelocationJournalHolds(t *testing.T) { |
| return hostbox.Actor{Box: "dst", Enrollment: enrollment, Revision: 7, Region: "ie", Eligible: true}, nil |
| }, |
| Kernel: kernel} |
| - state := execlease.State{V: 1, Binding: execlease.Binding{Host: roHost, Incarnation: h.Incarnation, Op: roOp, Box: "dst", Enrollment: enrollment, BootID: peer.BootID, ControllerID: "lease-controller"}, |
| - Process: execlease.ControllerProcess{PID: peer.PID, Start: peer.ProcessStart, KernelBootID: peer.BootID}, State: "active", LeaseETag: "lease-etag", Expires: time.Now().Add(time.Hour), SafeUntil: 1000} |
| - if err := os.MkdirAll(filepath.Join(dataDir, ".hostexec"), 0o700); err != nil { |
| - t.Fatal(err) |
| - } |
| - raw, _ := json.Marshal(state) |
| - if err := os.WriteFile(filepath.Join(dataDir, ".hostexec", roHost+".json"), raw, 0o600); err != nil { |
| - t.Fatal(err) |
| - } |
| - for _, suffix := range []string{".lock", ".live"} { |
| - file, err := os.OpenFile(filepath.Join(root.ExecLockDir, roHost+suffix), os.O_CREATE|os.O_RDWR, 0o660) |
| - if err != nil { |
| - t.Fatal(err) |
| - } |
| - if err := unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil { |
| - t.Fatal(err) |
| - } |
| - t.Cleanup(func() { file.Close() }) |
| - } |
| + configureLaunchParticipation(t, &launchFixture{root: root, peer: peer, target: target, claims: &launchClaims{obs: admitted}}) |
| spec := &hostbox.LaunchSpec{Identity: in.Identity, Executable: in.Executable, Digest: in.ExecutableSHA256, Args: in.Args, Env: in.Env} |
| reservation := hostbox.LaunchReservation{ID: a.ID + ":main", Role: "main", InvocationID: a.InvocationID, Inputs: a.LaunchInputs, Target: target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(roHost)}, Executable: exe} |
| q := hostbox.RoleLaunchRequest{V: hostbox.RoleLaunchV, Op: hostbox.RoleLaunchStart, Spec: spec, Reservation: &reservation, Executable: in.Executable, MemoryMax: cgroup.TierDefault.MemoryMax} |
diff --git a/internal/host/rootexecution_launch_linux_test.go b/internal/host/rootexecution_launch_linux_test.go| new file mode 100644 |
| index 000000000..98ff5227f |
| --- /dev/null |
| +++ b/internal/host/rootexecution_launch_linux_test.go |
| @@ -0,0 +1,253 @@ |
| +//go:build linux |
| + |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + execlease "github.com/carlosframework/platform/internal/hostexec" |
| + "github.com/carlosframework/platform/internal/hostguard" |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| + "golang.org/x/sys/unix" |
| +) |
| + |
| +type launchParticipationDisk struct{ *rollbackDisk } |
| + |
| +func (d *launchParticipationDisk) Lock(ctx context.Context, name string) (func(), error) { |
| + f, err := os.OpenFile(filepath.Join(d.dir, name), os.O_CREATE|os.O_RDWR, 0600) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for { |
| + if err = ctx.Err(); err != nil { |
| + f.Close() |
| + return nil, err |
| + } |
| + err = unix.Flock(int(f.Fd()), unix.LOCK_EX|unix.LOCK_NB) |
| + if err == nil { |
| + return func() { _ = unix.Flock(int(f.Fd()), unix.LOCK_UN); _ = f.Close() }, nil |
| + } |
| + if !errors.Is(err, unix.EAGAIN) && !errors.Is(err, unix.EWOULDBLOCK) { |
| + f.Close() |
| + return nil, err |
| + } |
| + select { |
| + case <-ctx.Done(): |
| + f.Close() |
| + return nil, ctx.Err() |
| + case <-time.After(time.Millisecond): |
| + } |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionFinalLaunchUsesRootParticipation(t *testing.T) { |
| + for _, atFinal := range []bool{false, true} { |
| + for _, attack := range []string{"none", "lost", "deadline", "boot", "process", "missing", "participation-only", "census", "update"} { |
| + t.Run(attack+"/"+map[bool]string{true: "final", false: "initial"}[atFinal], func(t *testing.T) { |
| + f := newLaunchFixture(t, rev2Operation()) |
| + disk := f.root.executionState.(*launchParticipationDisk) |
| + mutate := func() { |
| + raw, err := disk.Read(f.target.Host + ".json") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var r execlease.ExecutionRecord |
| + if err = json.Unmarshal(raw, &r); err != nil { |
| + t.Fatal(err) |
| + } |
| + switch attack { |
| + case "lost": |
| + r.Status = "lost" |
| + case "deadline": |
| + r.SafeUntil = 100 |
| + case "boot": |
| + r.Peer.KernelBootID = "other-boot" |
| + case "process": |
| + r.Peer.Start = "other-start" |
| + case "participation-only": |
| + r.Authority.ParticipationOnly = true |
| + case "missing": |
| + if err = disk.Remove(f.target.Host + ".json"); err != nil { |
| + t.Fatal(err) |
| + } |
| + return |
| + case "census": |
| + if err = disk.Remove("census.json"); err != nil { |
| + t.Fatal(err) |
| + } |
| + return |
| + case "update": |
| + if err = disk.Write("update.json", []byte(`{"Schema":1,"ID":"update","Revision":1}`)); err != nil { |
| + t.Fatal(err) |
| + } |
| + return |
| + } |
| + raw, _ = json.Marshal(r) |
| + if err = disk.Write(f.target.Host+".json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + // A real held service lock and a forged future deadline cannot grant root work. |
| + mirror := filepath.Join(f.root.Guard.DataDir, ".hostexec") |
| + if err := os.MkdirAll(mirror, 0700); err != nil { |
| + t.Fatal(err) |
| + } |
| + raw, _ := json.Marshal(execlease.State{V: 1, Binding: execlease.Binding{Host: f.target.Host, Incarnation: f.target.Incarnation, Op: f.target.Op, Box: f.target.Box, Enrollment: f.target.Enrollment, BootID: f.peer.BootID, ControllerID: "agent"}, Process: execlease.ControllerProcess{KernelBootID: f.peer.BootID, PID: f.peer.PID, Start: f.peer.ProcessStart}, State: "active", SafeUntil: 1 << 62, LeaseETag: "forged", Expires: time.Now().Add(365 * 24 * time.Hour)}) |
| + if err := os.WriteFile(filepath.Join(mirror, f.target.Host+".json"), raw, 0600); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, suffix := range []string{".lock", ".live"} { |
| + lock, err := os.OpenFile(filepath.Join(f.root.ExecLockDir, f.target.Host+suffix), os.O_CREATE|os.O_RDWR, 0660) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err = lock.Chmod(0660); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err = unix.Flock(int(lock.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { lock.Close() }) |
| + } |
| + if atFinal { |
| + f.units.beforeCheck = mutate |
| + } else { |
| + mutate() |
| + } |
| + _, err := f.root.Serve(t.Context(), f.peer, f.req) |
| + starts, _ := f.units.counts() |
| + if attack == "none" { |
| + if err != nil || starts != 1 { |
| + t.Fatalf("positive start=%d err=%v", starts, err) |
| + } |
| + } else if err == nil || starts != 0 { |
| + t.Fatalf("root refusal bypassed: start=%d err=%v", starts, err) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func configureLaunchParticipation(t *testing.T, f *launchFixture) { |
| + t.Helper() |
| + disk := &launchParticipationDisk{&rollbackDisk{dir: t.TempDir()}} |
| + f.root.executionState = disk |
| + raw, _ := json.Marshal(rootparticipation.Census{Schema: 1, Box: f.target.Box, Enrollment: f.target.Enrollment, Migration: 1, Revision: 1, LaunchRevision: 1, Complete: true}) |
| + if err := disk.Write("census.json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f.claims.obs.Claim.Op.Kind == hostclaim.KindServing { |
| + return |
| + } |
| + writeRev2Local(t, f, "active", 1000) |
| + unlock, err := disk.Lock(t.Context(), f.target.Host+".live") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(unlock) |
| +} |
| + |
| +func TestPlan14ExecutionRootLocalIgnoresAgentAbsence(t *testing.T) { |
| + for _, serving := range []bool{false, true} { |
| + for _, lost := range []bool{false, true} { |
| + t.Run(map[bool]string{true: "serving", false: "operation"}[serving]+"/"+map[bool]string{true: "lost", false: "valid"}[lost], func(t *testing.T) { |
| + op := rev2Operation() |
| + if serving { |
| + op = servingOp() |
| + } |
| + f := newLaunchFixture(t, op) |
| + disk := f.root.executionState.(*launchParticipationDisk) |
| + if lost { |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "old", Operation: "executionBegin", Host: f.target.Host, Incarnation: f.target.Incarnation, Stamp: f.target.Stamp, Session: f.target.Session, Op: "operation"} |
| + raw, _ := json.Marshal(execlease.ExecutionRecord{Schema: 1, Request: q, Token: "lost", Status: "lost", Authority: execlease.ExecutionAuthority{Migration: 1, Box: f.actor.Box, Enrollment: f.actor.Enrollment}, Replay: map[string]execlease.ExecutionReplay{}}) |
| + if err := disk.Write(f.target.Host+".json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + p := &productionClaimLocal{Resources: ClaimRuntimeResources{Mode: "instance-control", DataDir: f.root.Guard.DataDir}, assembly: &ClaimAssembly{}, lifetime: t.Context(), executionRead: f.root.readExecution} |
| + obs, _ := f.claims.Read(t.Context(), f.target.Host) |
| + selection := hostbox.Selection{Box: f.actor.Box, Region: f.actor.Region, Captured: hostguard.Captured{Incarnation: f.target.Incarnation, Stamp: f.target.Stamp}} |
| + for _, err := range []error{p.prepareWith(t.Context(), *f.actor, obs), p.Check(t.Context(), *f.actor, obs.Claim, selection)} { |
| + if lost && err == nil { |
| + t.Fatal("missing agent mirror cleared root loss") |
| + } |
| + if !lost && err != nil { |
| + t.Fatal("valid root participation refused", err) |
| + } |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionServingRequiresRootSettlement(t *testing.T) { |
| + for _, status := range []string{"settled", "lost", "unknown"} { |
| + for _, attack := range []string{"erase", "rename", "replace locks"} { |
| + t.Run(status+"/"+attack, func(t *testing.T) { |
| + f := newLaunchFixture(t, servingOp()) |
| + disk := f.root.executionState.(*launchParticipationDisk) |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "old", Operation: "executionBegin", Host: f.target.Host, Incarnation: f.target.Incarnation, Stamp: f.target.Stamp, Session: f.target.Session, Op: "prior-operation"} |
| + raw, _ := json.Marshal(execlease.ExecutionRecord{Schema: 1, Request: q, Token: "old", Status: status, Authority: execlease.ExecutionAuthority{Migration: 1, Box: f.actor.Box, Enrollment: f.actor.Enrollment}, Replay: map[string]execlease.ExecutionReplay{}}) |
| + if err := disk.Write(f.target.Host+".json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + mirror := filepath.Join(f.root.Guard.DataDir, ".hostexec") |
| + if err := os.Mkdir(mirror, 0700); err != nil { |
| + t.Fatal(err) |
| + } |
| + name := filepath.Join(mirror, f.target.Host+".json") |
| + if err := os.WriteFile(name, []byte(`{"state":"lost"}`), 0600); err != nil { |
| + t.Fatal(err) |
| + } |
| + if attack == "rename" { |
| + if err := os.Rename(mirror, mirror+".old"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.Mkdir(mirror, 0700); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else { |
| + if err := os.Remove(name); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if attack == "replace locks" { |
| + for _, suffix := range []string{".lock", ".live"} { |
| + name := filepath.Join(f.root.ExecLockDir, f.target.Host+suffix) |
| + file, err := os.OpenFile(name, os.O_CREATE|os.O_RDWR, 0660) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err = unix.Flock(int(file.Fd()), unix.LOCK_EX|unix.LOCK_NB); err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { file.Close() }) |
| + if err = os.Rename(name, name+".old"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err = os.WriteFile(name, nil, 0660); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + } |
| + _, err := f.root.Serve(t.Context(), f.peer, f.req) |
| + starts, _ := f.units.counts() |
| + if status == "settled" { |
| + if err != nil || starts != 1 { |
| + t.Fatalf("settled serving refused: starts=%d err=%v", starts, err) |
| + } |
| + } else if err == nil || starts != 0 { |
| + t.Fatalf("agent absence granted serving: starts=%d err=%v", starts, err) |
| + } |
| + }) |
| + } |
| + } |
| +} |
diff --git a/internal/host/rootexecution_migration.go b/internal/host/rootexecution_migration.go| new file mode 100644 |
| index 000000000..29d73ab58 |
| --- /dev/null |
| +++ b/internal/host/rootexecution_migration.go |
| @@ -0,0 +1,28 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + |
| + execlease "github.com/carlosframework/platform/internal/hostexec" |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| + "github.com/carlosframework/platform/internal/rootstate" |
| +) |
| + |
| +func (h ClaimRootHTTP) executionAuthorityWithMigration(ctx context.Context, state rootstate.Store, q execlease.ExecutionRequest, peer execlease.ControllerProcess) (execlease.ExecutionAuthority, error) { |
| + census, err := (rootparticipation.Ledger{State: state}).Census() |
| + if err != nil { |
| + return execlease.ExecutionAuthority{}, err |
| + } |
| + if census.NeverEnrolled { |
| + return execlease.ExecutionAuthority{}, execlease.ErrState |
| + } |
| + a, err := h.executionAuthority(ctx, q, peer) |
| + if err != nil { |
| + return a, err |
| + } |
| + if a.Box != census.Box || a.Enrollment != census.Enrollment || (a.Migration != 0 && a.Migration != census.Migration) { |
| + return execlease.ExecutionAuthority{}, execlease.ErrState |
| + } |
| + a.Migration = census.Migration |
| + return a, nil |
| +} |
diff --git a/internal/host/rootexecution_migration_test.go b/internal/host/rootexecution_migration_test.go| new file mode 100644 |
| index 000000000..a9860e07d |
| --- /dev/null |
| +++ b/internal/host/rootexecution_migration_test.go |
| @@ -0,0 +1,76 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "os" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + execlease "github.com/carlosframework/platform/internal/hostexec" |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| +) |
| + |
| +func TestPlan14ExecutionMigrationIsIndependent(t *testing.T) { |
| + for _, attack := range []string{"none", "missing", "legacy marker", "incomplete", "pending", "never enrolled", "enrollment", "box", "generation"} { |
| + t.Run(attack, func(t *testing.T) { |
| + _, root, iq := plan14InstallRegistration(t, hostclaim.KindLanding, "checking") |
| + if _, err := root.commitRootInventory(t.Context(), iq); err != nil { |
| + t.Fatal(err) |
| + } |
| + disk := &rollbackDisk{dir: t.TempDir()} |
| + census := rootparticipation.Census{Schema: 1, Box: "box-1", Enrollment: "en", Migration: 1, Revision: 1, LaunchRevision: 1, Complete: true} |
| + switch attack { |
| + case "incomplete": |
| + census.Complete = false |
| + case "pending": |
| + census.Pending = "acquisition" |
| + case "never enrolled": |
| + census.NeverEnrolled = true |
| + case "enrollment": |
| + census.Enrollment = "other" |
| + case "box": |
| + census.Box = "other" |
| + case "generation": |
| + census.Migration++ |
| + } |
| + if attack != "missing" && attack != "legacy marker" { |
| + raw, _ := json.Marshal(census) |
| + if err := disk.Write("census.json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if attack == "legacy marker" { |
| + if err := disk.Write("migration.json", []byte(`{"complete":true}`)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "begin", Operation: "executionBegin", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op} |
| + a, err := root.executionAuthorityWithMigration(t.Context(), disk, q, execlease.ControllerProcess{}) |
| + if attack == "none" { |
| + if err != nil || a.Migration != 1 { |
| + t.Fatalf("valid migration: %+v %v", a, err) |
| + } |
| + } else if err == nil { |
| + t.Fatal("invalid migration admitted", attack) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionPreinstallationBindsMigration(t *testing.T) { |
| + _, root, iq := plan14InstallRegistration(t, hostclaim.KindLanding, "checking") |
| + disk := &rollbackDisk{dir: t.TempDir()} |
| + raw, _ := json.Marshal(rootparticipation.Census{Schema: 1, Box: "box-1", Enrollment: "en", Migration: 1, Revision: 1, LaunchRevision: 1, Complete: true}) |
| + if err := disk.Write("census.json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "begin", Operation: "executionBegin", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op} |
| + a, err := root.executionAuthorityWithMigration(context.Background(), disk, q, execlease.ControllerProcess{}) |
| + if err != nil || !a.ParticipationOnly || a.Migration != 1 || a.Installation != "" { |
| + t.Fatalf("participation before registration: %+v %v", a, err) |
| + } |
| + if _, err := disk.Read(q.Host + ".json"); !os.IsNotExist(err) { |
| + t.Fatal("authority read created participation", err) |
| + } |
| +} |
diff --git a/internal/host/rootexecution_read.go b/internal/host/rootexecution_read.go| index 10a41d606..f6733add8 100644 |
| --- a/internal/host/rootexecution_read.go |
| +++ b/internal/host/rootexecution_read.go |
| @@ -3,6 +3,7 @@ package host |
| import ( |
| "context" |
| + "github.com/carlosframework/platform/internal/hostbox" |
| "github.com/carlosframework/platform/internal/hostclaim" |
| execlease "github.com/carlosframework/platform/internal/hostexec" |
| "github.com/carlosframework/platform/internal/rootparticipation" |
| @@ -10,14 +11,43 @@ import ( |
| ) |
| func rootExecutionRead(ctx context.Context, c *hostclaim.Claim) (execlease.ExecutionRecord, error) { |
| + return readRootExecution(ctx, nil, execlease.SystemClock{}, c) |
| +} |
| + |
| +func (h *ClaimLaunchRoot) readExecution(ctx context.Context, c *hostclaim.Claim) (execlease.ExecutionRecord, error) { |
| + return readRootExecution(ctx, h.executionState, h.Clock, c) |
| +} |
| + |
| +func readRootExecution(ctx context.Context, store rootstate.Store, clock execlease.Clock, c *hostclaim.Claim) (execlease.ExecutionRecord, error) { |
| if c == nil || c.Holder == nil { |
| return execlease.ExecutionRecord{}, execlease.ErrState |
| } |
| - store, err := rootstate.Open(rootparticipation.Namespace, false, false) |
| - if err != nil { |
| - return execlease.ExecutionRecord{}, err |
| + if store == nil { |
| + var err error |
| + store, err = rootstate.Open(rootparticipation.Namespace, false, false) |
| + if err != nil { |
| + return execlease.ExecutionRecord{}, err |
| + } |
| + defer store.Close() |
| } |
| - defer store.Close() |
| q := execlease.ExecutionRequest{Host: c.Host, Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Session: c.Holder.Session, Op: c.Op.ID} |
| - return execlease.ReadRoot(ctx, store, execlease.SystemClock{}, q, c.Op.Kind == hostclaim.KindServing) |
| + return execlease.ReadRoot(ctx, store, clock, q, c.Op.Kind == hostclaim.KindServing) |
| +} |
| + |
| +func (p *productionClaimLocal) checkRootParticipation(ctx context.Context, actor hostbox.Actor, c *hostclaim.Claim) error { |
| + read := p.executionRead |
| + if read == nil { |
| + read = rootExecutionRead |
| + } |
| + r, err := read(ctx, c) |
| + if err != nil { |
| + return err |
| + } |
| + if c == nil || c.Holder == nil { |
| + return execlease.ErrState |
| + } |
| + if c.Op.Kind != hostclaim.KindServing && (r.Authority.Box != actor.Box || r.Authority.Enrollment != actor.Enrollment) { |
| + return execlease.ErrLost |
| + } |
| + return nil |
| } |
diff --git a/internal/hostexec/rootbroker.go b/internal/hostexec/rootbroker.go| index 68d520ecb..51f67dfed 100644 |
| --- a/internal/hostexec/rootbroker.go |
| +++ b/internal/hostexec/rootbroker.go |
| @@ -437,7 +437,7 @@ func (b *RootBroker) acquire(ctx context.Context, peer ControllerProcess, q Exec |
| func (b *RootBroker) settle(ctx context.Context, q ExecutionRequest, r ExecutionRecord, digest string) (ExecutionResponse, error) { |
| if r.Status == "settled" { |
| - return ExecutionResponse{}, ErrLost |
| + return executionResponse(r), nil |
| } |
| if err := b.deny(ctx, r, nil); err != nil { |
| return ExecutionResponse{}, err |
| @@ -508,7 +508,11 @@ func (b *RootBroker) Monitor(ctx context.Context) error { |
| errs = append(errs, err) |
| continue |
| } |
| - if r.Status == "settled" || r.Status == "lost" { |
| + if r.Status == "settled" { |
| + continue |
| + } |
| + if r.Status == "lost" { |
| + errs = append(errs, b.recoverSettlement(ctx, host)) |
| continue |
| } |
| _, now, clockErr := b.cfg.Clock.Now() |
| @@ -624,12 +628,36 @@ func (b *RootBroker) Recover(ctx context.Context) error { |
| continue |
| } |
| if r.Status != "settled" { |
| - errs = append(errs, b.deny(ctx, r, nil)) |
| + errs = append(errs, b.recoverSettlement(ctx, host)) |
| } |
| } |
| return errors.Join(errs...) |
| } |
| +func (b *RootBroker) recoverSettlement(ctx context.Context, host string) error { |
| + b.mu.Lock() |
| + gate := b.calls[host] |
| + if gate == nil { |
| + gate = make(chan struct{}, 1) |
| + b.calls[host] = gate |
| + } |
| + b.mu.Unlock() |
| + select { |
| + case gate <- struct{}{}: |
| + default: |
| + return nil |
| + } |
| + defer func() { <-gate }() |
| + r, err := b.read(host) |
| + if err != nil || r.Status == "settled" { |
| + return err |
| + } |
| + q := r.Request |
| + q.Operation, q.Token, q.ID = "executionSettle", r.Token, "recovery-"+r.Token |
| + _, err = b.settle(ctx, q, r, executionDigest(q, r.Peer)) |
| + return err |
| +} |
| + |
| func (b *RootBroker) Run(ctx context.Context) { |
| ctx, leave, err := b.enter(ctx) |
| if err != nil { |
diff --git a/internal/hostexec/rootbroker_recovery_test.go b/internal/hostexec/rootbroker_recovery_test.go| new file mode 100644 |
| index 000000000..bc9a16bed |
| --- /dev/null |
| +++ b/internal/hostexec/rootbroker_recovery_test.go |
| @@ -0,0 +1,133 @@ |
| +//go:build linux |
| + |
| +package hostexec |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "testing" |
| +) |
| + |
| +func TestPlan14ExecutionRecoverySettlesWithoutControllerToken(t *testing.T) { |
| + for _, fail := range []string{"", "cleanup", "cc9"} { |
| + t.Run(fail, func(t *testing.T) { |
| + b, _, remote, _, q, peer := brokerFixture(t) |
| + if _, err := b.Execute(t.Context(), peer, q); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := b.Close(t.Context()); err != nil { |
| + t.Fatal(err) |
| + } |
| + cfg := b.cfg |
| + cfg.Identity = "replacement" |
| + refused := errors.New("unresolved") |
| + if fail == "cleanup" { |
| + cfg.Cleanup = func(context.Context, ExecutionRecord) error { return refused } |
| + } |
| + if fail == "cc9" { |
| + cfg.Settle = func(context.Context, ExecutionRecord) error { return refused } |
| + } |
| + next, err := NewRootBroker(cfg) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer next.Close(t.Context()) |
| + err = next.Recover(t.Context()) |
| + r, readErr := next.read(q.Host) |
| + if readErr != nil { |
| + t.Fatal(readErr) |
| + } |
| + if fail != "" { |
| + if !errors.Is(err, refused) || r.Status == "settled" { |
| + t.Fatalf("unproven recovery: %+v %v", r, err) |
| + } |
| + next.cfg.Cleanup = func(context.Context, ExecutionRecord) error { return nil } |
| + next.cfg.Settle = func(context.Context, ExecutionRecord) error { return nil } |
| + if err = next.Monitor(t.Context()); err != nil { |
| + t.Fatal("retry", err) |
| + } |
| + r, err = next.read(q.Host) |
| + } |
| + if err != nil || r.Status != "settled" { |
| + t.Fatalf("recovery left an ownerless obligation: %+v %v", r, err) |
| + } |
| + if remote.calls != 1 { |
| + t.Fatal("recovery acquired a lease") |
| + } |
| + if _, err = next.Execute(t.Context(), peer, q); err == nil { |
| + t.Fatal("recovered begin resurrected") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionRecoveryDoesNotReleaseNewOwner(t *testing.T) { |
| + b, _, remote, _, q, peer := brokerFixture(t) |
| + if _, err := b.Execute(t.Context(), peer, q); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := b.Close(t.Context()); err != nil { |
| + t.Fatal(err) |
| + } |
| + key, err := Key(q.Host, "box") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + _, etag, err := remote.Get(t.Context(), key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + newer := []byte(`{"op":"another-owner","expires":"2099-01-01T00:00:00Z"}`) |
| + nextETag, err := remote.Put(t.Context(), key, newer, etag) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + cfg := b.cfg |
| + cfg.Identity = "replacement" |
| + next, err := NewRootBroker(cfg) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer next.Close(t.Context()) |
| + if err = next.Recover(t.Context()); err == nil { |
| + t.Fatal("replacement lease counted as our release") |
| + } |
| + body, got, err := remote.Get(t.Context(), key) |
| + if err != nil || got != nextETag || string(body) != string(newer) { |
| + t.Fatalf("recovery deleted a newer owner: %s %s %v", body, got, err) |
| + } |
| + r, err := next.read(q.Host) |
| + if err != nil || r.Status == "settled" { |
| + t.Fatalf("unresolved release settled: %+v %v", r, err) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionRecoveryJoinsRemoteWrite(t *testing.T) { |
| + b, _, remote, clock, q, peer := brokerFixture(t) |
| + out, err := b.Execute(t.Context(), peer, q) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + entered, resume := make(chan struct{}), make(chan struct{}) |
| + remote.onPut = func(int) { close(entered); <-resume } |
| + q.Operation, q.ID, q.Token = "executionRenew", "renew-in-flight", out.Token |
| + done := make(chan error, 1) |
| + go func() { _, err := b.Execute(t.Context(), peer, q); done <- err }() |
| + <-entered |
| + clock.advance(TTL) |
| + _ = b.Monitor(t.Context()) |
| + err = b.Monitor(t.Context()) |
| + r, readErr := b.read(q.Host) |
| + close(resume) |
| + writeErr := <-done |
| + if err != nil || readErr != nil || r.Status != "lost" || writeErr == nil { |
| + t.Fatalf("settled before remote write joined: %+v monitor=%v read=%v write=%v", r, err, readErr, writeErr) |
| + } |
| + if err = b.Monitor(t.Context()); err != nil { |
| + t.Fatal(err) |
| + } |
| + r, err = b.read(q.Host) |
| + if err != nil || r.Status != "settled" { |
| + t.Fatalf("joined write cannot settle: %+v %v", r, err) |
| + } |
| +} |
diff --git a/internal/hostexec/rootclient.go b/internal/hostexec/rootclient.go| index a4ff8c5ae..565b6966c 100644 |
| --- a/internal/hostexec/rootclient.go |
| +++ b/internal/hostexec/rootclient.go |
| @@ -23,6 +23,7 @@ type RootClient struct { |
| mu sync.Mutex |
| gate chan struct{} |
| request ExecutionRequest |
| + uncertain bool |
| state State |
| err error |
| next int64 |
| @@ -154,6 +155,10 @@ func (c *RootClient) refresh(ctx context.Context, begin bool) error { |
| if !begin && (old.Stamp != q.Stamp || old.Session != q.Session) { |
| q.Operation = "executionRenew" |
| } |
| + c.mu.Lock() |
| + c.request = q |
| + c.uncertain = true |
| + c.mu.Unlock() |
| out, err := c.cfg.Call(ctx, q) |
| if err != nil { |
| c.lose(err) |
| @@ -173,6 +178,7 @@ func (c *RootClient) refresh(ctx context.Context, begin bool) error { |
| } |
| c.state = State{V: 1, Binding: binding, State: out.Status, SafeUntil: out.SafeUntil, Expires: out.Expires, LeaseETag: out.LeaseETag} |
| c.request = q |
| + c.uncertain = false |
| if q.Operation != "executionView" { |
| c.next = start + int64(Renewal) |
| } |
| @@ -225,6 +231,12 @@ func (c *RootClient) close(ctx context.Context) error { |
| case <-ctx.Done(): |
| return ctx.Err() |
| } |
| + select { |
| + case c.gate <- struct{}{}: |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + } |
| + defer func() { <-c.gate }() |
| if !c.cleaned { |
| if err := c.cfg.Cleanup(ctx); err != nil { |
| return err |
| @@ -232,8 +244,25 @@ func (c *RootClient) close(ctx context.Context) error { |
| c.cleaned = true |
| } |
| c.mu.Lock() |
| - q, token := c.request, c.state.ControllerID |
| + q, token, uncertain := c.request, c.state.ControllerID, c.uncertain |
| c.mu.Unlock() |
| + if uncertain { |
| + if !q.Valid() { |
| + return ErrLost |
| + } |
| + out, err := c.cfg.Call(ctx, q) |
| + if err != nil { |
| + return err |
| + } |
| + if !executionReplyMatches(q, c.cfg.Binding.BootID, out) || out.Status != "active" { |
| + return ErrLost |
| + } |
| + token = out.Token |
| + c.mu.Lock() |
| + c.state.ControllerID = token |
| + c.uncertain = false |
| + c.mu.Unlock() |
| + } |
| if token == "" { |
| return ErrLost |
| } |
| @@ -250,9 +279,13 @@ func (c *RootClient) close(ctx context.Context) error { |
| if err != nil { |
| return err |
| } |
| - if settle.Status != "settled" || settle.Token != token { |
| + if !executionReplyMatches(q, c.cfg.Binding.BootID, settle) || settle.Status != "settled" || settle.Token != token { |
| return ErrLost |
| } |
| c.settled = true |
| return nil |
| } |
| + |
| +func executionReplyMatches(q ExecutionRequest, boot string, out ExecutionResponse) bool { |
| + return out.Schema == 1 && out.Host == q.Host && out.Incarnation == q.Incarnation && out.Stamp == q.Stamp && out.Session == q.Session && out.Op == q.Op && out.BootID == boot && out.Token != "" |
| +} |
diff --git a/internal/hostexec/rootclient_test.go b/internal/hostexec/rootclient_test.go| index b3f9178af..4e5dff928 100644 |
| --- a/internal/hostexec/rootclient_test.go |
| +++ b/internal/hostexec/rootclient_test.go |
| @@ -4,6 +4,7 @@ package hostexec |
| import ( |
| "context" |
| + "errors" |
| "sync/atomic" |
| "testing" |
| "time" |
| @@ -53,3 +54,87 @@ func TestPlan14ExecutionClientHasOneRenewalOwner(t *testing.T) { |
| t.Fatalf("cleanup=%d", cleanup.Load()) |
| } |
| } |
| + |
| +func TestPlan14ExecutionClientLostBeginResponseSettles(t *testing.T) { |
| + b, _, remote, clock, q, peer := brokerFixture(t) |
| + var begin ExecutionRequest |
| + var lost bool |
| + c, err := OpenRootClient(t.Context(), RootClientConfig{ |
| + Binding: Binding{Host: q.Host, Incarnation: q.Incarnation, Op: q.Op, Box: "box", Enrollment: "en", BootID: peer.KernelBootID}, Clock: clock, |
| + Observe: func(context.Context) (ExecutionRequest, error) { return q, nil }, |
| + Call: func(ctx context.Context, request ExecutionRequest) (ExecutionResponse, error) { |
| + if request.Operation == "executionBegin" { |
| + if begin.ID != "" && begin != request { |
| + t.Error("begin replay changed identity") |
| + } |
| + begin = request |
| + if _, err := b.Execute(ctx, peer, request); err != nil { |
| + return ExecutionResponse{}, err |
| + } |
| + if !lost { |
| + lost = true |
| + return ExecutionResponse{}, context.DeadlineExceeded |
| + } |
| + } |
| + if request.Host != begin.Host || request.Incarnation != begin.Incarnation || request.Stamp != begin.Stamp || request.Session != begin.Session || request.Op != begin.Op { |
| + t.Error("cleanup lost the captured begin tuple") |
| + } |
| + return b.Execute(ctx, peer, request) |
| + }, CloseAdmission: func() {}, Cleanup: func(context.Context) error { return nil }, |
| + }) |
| + if !errors.Is(err, context.DeadlineExceeded) { |
| + t.Fatal(err) |
| + } |
| + if err = c.Close(t.Context()); err != nil { |
| + t.Fatal("lost begin cannot settle", err) |
| + } |
| + record, err := b.read(q.Host) |
| + if err != nil || record.Status != "settled" || remote.calls != 1 { |
| + t.Fatalf("record=%+v err=%v calls=%d", record, err, remote.calls) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionClientLostTransitionResponseSettles(t *testing.T) { |
| + b, _, remote, clock, q, peer := brokerFixture(t) |
| + initial := q |
| + original := b.cfg.Authority |
| + b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) { |
| + got.Session = initial.Session |
| + return original(ctx, got, peer) |
| + } |
| + b.cfg.Transition = func(context.Context, ExecutionRecord, ExecutionRequest, ExecutionAuthority) error { return nil } |
| + var uncertain ExecutionRequest |
| + lost := false |
| + c, err := OpenRootClient(t.Context(), RootClientConfig{ |
| + Binding: Binding{Host: q.Host, Incarnation: q.Incarnation, Op: q.Op, Box: "box", Enrollment: "en", BootID: peer.KernelBootID}, Clock: clock, |
| + Observe: func(context.Context) (ExecutionRequest, error) { return q, nil }, |
| + Call: func(ctx context.Context, request ExecutionRequest) (ExecutionResponse, error) { |
| + out, err := b.Execute(ctx, peer, request) |
| + if err == nil && request.Operation == "executionRenew" { |
| + if uncertain.ID != "" && uncertain != request { |
| + t.Error("transition replay changed request") |
| + } |
| + uncertain = request |
| + if !lost { |
| + lost = true |
| + return ExecutionResponse{}, context.DeadlineExceeded |
| + } |
| + } |
| + return out, err |
| + }, CloseAdmission: func() {}, Cleanup: func(context.Context) error { return nil }, |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + q.Session++ |
| + if err = c.Refresh(t.Context()); !errors.Is(err, context.DeadlineExceeded) { |
| + t.Fatal(err) |
| + } |
| + if err = c.Close(t.Context()); err != nil { |
| + t.Fatal("lost transition cannot settle", err) |
| + } |
| + r, err := b.read(q.Host) |
| + if err != nil || r.Status != "settled" || r.Request.Session != q.Session || remote.calls != 2 { |
| + t.Fatalf("transition recovery: %+v %v writes=%d", r, err, remote.calls) |
| + } |
| +} |
diff --git a/internal/hostexec/rootread.go b/internal/hostexec/rootread.go| index 0af38a207..7b1b5b292 100644 |
| --- a/internal/hostexec/rootread.go |
| +++ b/internal/hostexec/rootread.go |
| @@ -16,6 +16,9 @@ func ReadRoot(ctx context.Context, store rootstate.Store, clock Clock, q Executi |
| if store == nil || clock == nil { |
| return r, ErrState |
| } |
| + if _, err := store.Read("update.json"); !errors.Is(err, os.ErrNotExist) { |
| + return r, ErrLost |
| + } |
| ledger := rootparticipation.Ledger{State: store} |
| census, err := ledger.Census() |
| if err != nil || census.NeverEnrolled { |
diff --git a/internal/provision/testdata/plan14-root-reads-merge.tsv b/internal/provision/testdata/plan14-root-reads-merge.tsv| index 883b9672b..fb43ac37a 100644 |
| --- a/internal/provision/testdata/plan14-root-reads-merge.tsv |
| +++ b/internal/provision/testdata/plan14-root-reads-merge.tsv |
| @@ -36,7 +36,7 @@ internal/host/claimhandover_crossbox.go relocationBefore authority C Before-admi |
| internal/host/claimhandover_crossbox.go wireClaimAdmission authority C,D Admission constructor adds completed-relocation recovery owners while retaining typed root recovery rebind. c8a2f03e2937b72d3e38cea02aa5e49ebfe8a38638d97c64dc65db5ddc9955a0 |
| internal/host/claimlaunch_production.go claimLaunchFor authority C,D Root launch keeps canonical installation attribution and adds relocation attempt reader under region-HA switch. data-literal@414,file:CommandContext@605 76cfc428c7385d1278a5d3842396a6f7b343f03547fdbfc51088af262f3980ef |
| internal/host/claimmoveparks.go openProductionClaimMoveWake authority C,G Moves owner construction wraps every bucket request in bound identity check and preserves wake refusal. file:Open@521 bff34938c41a8c054e98744c3262b182437231ac6dc361c1a5ba15a39b8b7505 |
| -internal/host/claimparticipation.go productionClaimLocal.prepareWith authority D Participation slots retain binding through cleanup, lock paths and update guards; no local deadline becomes root authority. file:Open@4044,file:ReadFile@2934 bcd59a6266bf4e75639b90beb2a3649299466adfcd7a79bc9c6eda7ef518b8e6 |
| +internal/host/claimparticipation.go productionClaimLocal.prepareWith authority D Root observers read root participation; agent controller remains legacy pending D1 production selection. TestPlan14ExecutionRootLocalIgnoresAgentAbsence. file:Open@3220,file:ReadFile@2110 e6d57a481beadde9ce567cc816007ee1e1ce0c5c43c654d65b84958bafbfa780 |
| internal/host/claimparticipation_completed.go ClaimAssembly.completedRelocationDischarge authority C,D Completed-relocation discharge requires exact claim completion and verified retirement receipt; unread evidence refuses cleanup. file:Read@227 e061f95cf3d974fb253c59540cee70ca942a8d73042679aa7f071ada90f9b69d |
| internal/host/claimparticipation_relocation.go productionClaimLocal.Acknowledged authority D Relocation acknowledgement waits for bound controller readiness; participates in execution admission. file:Prepare@268 022a1bbe9f6479e37ad7afd4a9b9201915f729953742f3306cd3f01cd79cc1d5 |
| internal/host/claimparticipation_stale.go productionClaimLocal.clearUntracked authority D Untracked participation cleanup now carries exact controller binding into retirement before stale-state clearing. file:Read@1004 067dcf7f62940af8030315834087853273a039a56e2095ae15c8b2f3342e8fb3 |
| @@ -251,7 +251,7 @@ internal/sim/hostlife/mutate.sh external legacy accepted-legacy accepted-legacy; |
| @coverage internal/console 41714f3779b343460800cb25b31782dc7d5c1e83b2aacf7d8d0a28f39c47bf76 |
| @coverage internal/console/templates 6492b7455fa3315105339cca03edb4372ac0d9bf746c018e13a308cf0b74cc05 |
| @coverage internal/edgesfile 553af147ca4882b8425afddf1ed250ee40a6cac92361d667cb4a690393be5272 |
| -@coverage internal/host d9e1b4aecc47f93e592ba94f5e51214592636cef8a677792b3985a8311c4e0b5 |
| +@coverage internal/host 6212a55d3866480b2f7b24119e5e62c03991cfd377e2d011d1c316f568fea98b |
| @coverage internal/hostbox afb08c287eaccb43adde1c488bd122482711ce4e29ecb675c9428f27130c3d93 |
| @coverage internal/hostclaim feaf701db4b49ce92733b9d6c1fa1c0ae8efc6e13439ce201fca1a9e24dcf927 |
| @coverage internal/hostfiles 554285c123758375c13388b2fa42f654819392dac0b027816a713548ec17dcc6 |
| @@ -268,15 +268,15 @@ internal/sim/hostlife/mutate.sh external legacy accepted-legacy accepted-legacy; |
| @coverage internal/sim/hostlife eaac101ea7970c74fee621e5cd7515ec7e4907bc5874c65eab86666af0f7dc61 |
| @coverage internal/status d5ef42fe4dc2cb72d1ce5e133e2c95a3e85960a222eb424f00b259b961b8d976 |
| @coverage internal/updatewords 93bb83474bad5d20281c34e72fadf56e2847e9a7eee2385a78f46a11fc39a3b6 |
| -@coverage internal/hostexec 899bdcd987f69f09cc5fcd89949a69387827ec48bc23e5d54abac96c7a335e85 |
| +@coverage internal/hostexec 4727c69c222b0b35066fbd3df941c40d78b92e311dfa3a928ebc6d84bc9bee42 |
| @coverage internal/rootparticipation db8fb3a896a400cfe6396f436625732502a83af1618be7fc3df7b198a8c1bc32 |
| internal/host/rootexecution_authority.go ClaimRootHTTP.executionAuthority authority D D1 reads the fresh claim, deployment record and root inventory, including the serving writer box; candidate admission only. Production broker construction is pending. file:Read@553 90b20d5af85ba9ef95bb8487797a715af7d43a1b3350058173079aece3bdea47 |
| -internal/host/rootexecution_read.go rootExecutionRead authority D D1/D2 open the descriptor-checked root participation namespace; no service mirror grants authority. Production final-effect wiring is pending. file:Open@210 87ad4f6167cf63d0364c12a48ab0ef7279fed6fa0920c012ae7738295496bbdc |
| +internal/host/rootexecution_read.go rootExecutionRead authority D Production root observer selects the descriptor-checked participation reader. c70216cd8e62430b43669d6cea5bcb9101fcb31c04aea178a951a620c369b1e5 |
| internal/hostexec/participation.go PreinstallationParticipant authority D D1 selects only the fresh claim executor or named target in its allowed phase; Exec is a claim field, not a process invocation. No installation authority. file:Exec@1035,file:Exec@1307,file:Exec@276 f14568ea08c9d439d7239ac9d5d596a134a17627e1eac878c074629ca2ba7304 |
| internal/hostexec/rootbroker.go RootBroker.Execute authority D D1/D2 register in-flight work before acquiring root lifetime locks, dispatch a tokenless serving view without acquisition, and retain conditional-write and replay evidence. file:Read@3101,file:Read@4519 10b222fb88be75fc062d1a45c86037580eaab725b1bffe5d4848b5881d538e30 |
| internal/hostexec/rootbroker.go RootBroker.deny authority D D2 compares the current root token before durable loss and live-lock revocation; cleanup is bounded independently of request cancellation. file:Read@117 e29e9132d0ceb540000a1836249fda71c4a75b4746d6037a42cbf6e2e2752b9d |
| internal/hostexec/rootbroker.go RootBroker.read authority D D1/D2 decode the retained root participation record; missing or malformed state grants nothing. file:Read@113 8a2fb0134e9c24bbe448efc35f193e60d01e909a3129b465aa16f682eb4114f1 |
| -internal/hostexec/rootread.go ReadRoot authority D D1/D2 require migration completeness, exact root record, kernel peer identity, request-start cutoff and held root live lock. file:Read@1915,file:Read@392 6e9e0b87e455815edca1c2e1ea43d4e0001bef5f5eeb2d5c0c6171a4a86dbb62 |
| +internal/hostexec/rootread.go ReadRoot authority D Root census, update marker, tuple, kernel identity, deadline and live lock; final launch and serving-settlement witnesses exercise the production reader. file:Read@2013,file:Read@228,file:Read@490 0151eab12e55ebf3674f66c9a48d6f7f991a6b5dc529491a25adcbfa77e4101c |
| internal/rootparticipation/ledger.go Ledger.Acquire authority D D3 excludes updates and persists the root acquisition census intent before the reservation callback. Production wiring is pending. file:Read@345 a0800db3005f69f6379e4483e596bd880c501077591a3c748ab2f52407af7d9c |
| internal/rootparticipation/ledger.go Ledger.BeginUpdate authority D D3 serializes the root update marker and complete participation/launch census with acquisition. file:Read@289 01bee25b1e5271d2a3ebd64562e3b7246becf0be05072c87afc291c0800cdd78 |
| internal/rootparticipation/ledger.go Ledger.Census authority D D3 requires explicit root migration, enrollment and completeness; missing history is unavailable. file:Read@119 ff240c4ba7e395688a83c0ffc652ce0e337c1723519a4bd5bbbbd858a2edb005 |
| @@ -284,3 +284,9 @@ internal/rootparticipation/ledger.go Ledger.EndUpdate authority D D3 removes onl |
| internal/rootparticipation/ledger.go Ledger.idle authority D D3 enumerates root records and requires settled obligations plus an independently supplied launch/kernel census. file:Read@429 b7b27590041bac6bec3de245b7fe20da7965cb2a9e52725c888f8a83313ff9f3 |
| internal/hostexec/rootclient.go requestIdentity authority D D1 supplies collision-resistant request correlation only; root authenticates peer and authority independently. Production client integration pending. file:Read@73 2acec351fdc4ab671e65b4a9bb5017aeb094fbf78fad8e34d713e48a2846c158 |
| internal/hostexec/rootserving.go RootBroker.servingView authority D D1 reads only completed root census and settled root participation under transaction exclusion, rechecks claim authority and peer identity, and returns no lease or token. Production client selection is pending. file:Read@441 9a57720834b3d7c661854196cf0ed1261c643673b5ba900c38987a162f5de0bb |
| +internal/host/claimlaunch_execpermit.go ClaimLaunchRoot.execPermit authority D Root participation bounds the executable permit; TestPlan14ExecutionFinalLaunchUsesRootParticipation covers initial and final checks. file:Read@157 795354a21d94cfe2eaf955dc913e88079426947e9547ebe83b79332bfdd7f3f1 |
| +internal/host/rootexecution_read.go readRootExecution authority D Descriptor-checked root opener shared by root Prepare/Check, initial launch authorization and final executable checks; no agent fallback. file:Open@291 2f70b4203a803b3fe0cd1e57c21106433a5af6a14dad8217e67bd24a28ad2b2f |
| +internal/host/rootexecution_read.go productionClaimLocal.checkRootParticipation authority D Root observer binds participation to enrolled actor; TestPlan14ExecutionRootLocalIgnoresAgentAbsence. 9cb9f9b98b9f143c7ce9225dd8c6b7798ce17683ffa85dded31a9bf90bcaeab9 |
| +internal/host/claimlaunch_authority.go ClaimLaunchRoot.authorize authority D Both root launch authorizations read root participation and match authenticated peer; TestPlan14ExecutionFinalLaunchUsesRootParticipation. de7e7f2ffebfc7d7e26e0c72492b3d9fb4b8766cefcd2648931f3f0ac975cf45 |
| +internal/host/rootexecution_migration.go ClaimRootHTTP.executionAuthorityWithMigration authority D Root participation census supplies migration generation; TestPlan14ExecutionMigrationIsIndependent and TestPlan14ExecutionPreinstallationBindsMigration. Production construction remains pending. 1c7f21b4296de5284a89ad4fa97b352cc6a59998156a22b07b6798291b778f7d |
| +internal/hostexec/rootbroker.go RootBroker.recoverSettlement authority D Recovery serializes settlement with in-flight requests and requires cleanup, exact conditional release and CC9. TestPlan14ExecutionRecoverySettlesWithoutControllerToken and TestPlan14ExecutionRecoveryJoinsRemoteWrite. 88c7b2d1a863590e549a2477732badcb6e003289da8db46aee16a1d777a8f77d |