amadan

carlos / platform Public

WIP claim Plan 14 batch 3 checkpoint 1 (not for merge): execution lease broker core, root deadlines and loss latches, acquisition/update exclusion ledger; production wiring and E1-E4 remain

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 02d951ffb2107f705c735df2991abd782508c446 parent 1920027
26 files changed, +2752 −9
  • approved_moves_copy_test.go +1 −0
  • internal/host/claimretire_transport.go +7 −1
  • internal/host/plan14_execution_authority_test.go +119 −0
  • internal/host/plan14_root_participation_test.go +46 −0
  • internal/host/rootexecution_authority.go +132 −0
  • internal/host/rootexecution_read.go +23 −0
  • internal/host/rootexecution_transport.go +81 −0
  • internal/host/rootexecution_transport_test.go +87 −0
  • internal/hostexec/participation.go +48 −0
  • internal/hostexec/plan14_root_participation_test.go +723 −0
  • internal/hostexec/record.go +2 −5
  • internal/hostexec/rootbroker.go +573 −0
  • internal/hostexec/rootclient.go +258 −0
  • internal/hostexec/rootclient_test.go +55 −0
  • internal/hostexec/rootprocess_linux.go +23 −0
  • internal/hostexec/rootprocess_other.go +7 −0
  • internal/hostexec/rootread.go +77 −0
  • internal/hostguard/root.go +14 −0
  • internal/hostrelocate/plan14_participation_executor_test.go +81 −0
  • internal/instancectl/server.go +1 −0
  • internal/provision/plan9_b8_census_test.go +4 −0
  • internal/provision/testdata/plan14-root-reads-merge.tsv +18 −3
  • internal/rootparticipation/ledger.go +190 −0
  • internal/rootparticipation/ledger_test.go +102 −0
  • internal/rootparticipation/record.go +57 −0
  • internal/rootparticipation/request.go +23 −0

CI passed — run details

CI log 
=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.011s
?   	amadan.net/carlos/platform.git/launchctx/controlname	[no test files]
?   	amadan.net/carlos/platform.git/launchctx/wireid	[no test files]
go vet ./...
gofmt -l .: clean
go test ./... -count=1 -timeout 30m
ok  	github.com/carlosframework/platform	346.885s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.055s
ok  	github.com/carlosframework/platform/cmd/moveprobe	93.244s
ok  	github.com/carlosframework/platform/internal/activator	234.483s
ok  	github.com/carlosframework/platform/internal/agentsock	0.019s
ok  	github.com/carlosframework/platform/internal/alerting	0.003s
ok  	github.com/carlosframework/platform/internal/analytics	0.020s
ok  	github.com/carlosframework/platform/internal/appconfig	2.339s
ok  	github.com/carlosframework/platform/internal/apperr	0.078s
ok  	github.com/carlosframework/platform/internal/applog	0.143s
ok  	github.com/carlosframework/platform/internal/awsinv	0.003s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.005s
ok  	github.com/carlosframework/platform/internal/boxid	0.525s
?   	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.189s
ok  	github.com/carlosframework/platform/internal/changelog	0.242s
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	364.637s
?   	github.com/carlosframework/platform/internal/controlname	[no test files]
?   	github.com/carlosframework/platform/internal/controlsock	[no test files]
ok  	github.com/carlosframework/platform/internal/dbdoor	5.948s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.009s
ok  	github.com/carlosframework/platform/internal/dbverify	0.004s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.002s
ok  	github.com/carlosframework/platform/internal/docsite	0.005s
ok  	github.com/carlosframework/platform/internal/domains	2.129s
ok  	github.com/carlosframework/platform/internal/doorkey	0.006s
ok  	github.com/carlosframework/platform/internal/economics	0.004s
ok  	github.com/carlosframework/platform/internal/edge	26.071s
?   	github.com/carlosframework/platform/internal/edgeproof	[no test files]
ok  	github.com/carlosframework/platform/internal/edgesfile	0.008s
ok  	github.com/carlosframework/platform/internal/fleetapi	0.124s
ok  	github.com/carlosframework/platform/internal/gate	1.655s
ok  	github.com/carlosframework/platform/internal/genselect	0.014s
ok  	github.com/carlosframework/platform/internal/geo	0.021s
ok  	github.com/carlosframework/platform/internal/geoip	0.032s
ok  	github.com/carlosframework/platform/internal/host	85.813s
ok  	github.com/carlosframework/platform/internal/hostactors	0.386s
ok  	github.com/carlosframework/platform/internal/hostbox	2.145s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.484s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.005s
ok  	github.com/carlosframework/platform/internal/hostexec	0.370s
ok  	github.com/carlosframework/platform/internal/hostfence	0.031s
ok  	github.com/carlosframework/platform/internal/hostfiles	10.470s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.009s
ok  	github.com/carlosframework/platform/internal/hostguard	0.389s
ok  	github.com/carlosframework/platform/internal/hostops	1.436s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.004s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.005s
ok  	github.com/carlosframework/platform/internal/hostrelocate	0.447s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.144s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.001s
ok  	github.com/carlosframework/platform/internal/infraview	0.002s
ok  	github.com/carlosframework/platform/internal/instancectl	1.713s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.706s
ok  	github.com/carlosframework/platform/internal/instancefs	0.004s
ok  	github.com/carlosframework/platform/internal/instancekey	0.013s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.019s
ok  	github.com/carlosframework/platform/internal/layout	1.423s
ok  	github.com/carlosframework/platform/internal/lease	9.456s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.018s
ok  	github.com/carlosframework/platform/internal/ledger	0.011s
ok  	github.com/carlosframework/platform/internal/mail	0.162s
ok  	github.com/carlosframework/platform/internal/maintenance	0.083s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.507s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.070s
ok  	github.com/carlosframework/platform/internal/metrics	0.006s
ok  	github.com/carlosframework/platform/internal/moves	21.071s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.740s
ok  	github.com/carlosframework/platform/internal/objstore	0.158s
ok  	github.com/carlosframework/platform/internal/ops	0.167s
ok  	github.com/carlosframework/platform/internal/park	0.199s
ok  	github.com/carlosframework/platform/internal/privatefile	0.003s
ok  	github.com/carlosframework/platform/internal/provision	12.333s
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.014s
ok  	github.com/carlosframework/platform/internal/relay	7.231s
ok  	github.com/carlosframework/platform/internal/release	19.777s
ok  	github.com/carlosframework/platform/internal/rollout	3.363s
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.003s
ok  	github.com/carlosframework/platform/internal/rootstate	0.024s
ok  	github.com/carlosframework/platform/internal/s3compat	0.005s
ok  	github.com/carlosframework/platform/internal/safefile	0.003s
ok  	github.com/carlosframework/platform/internal/schedule	2.296s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	9.328s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	3.948s
ok  	github.com/carlosframework/platform/internal/status	0.039s
ok  	github.com/carlosframework/platform/internal/steering	0.005s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.192s
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.543s
=== 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.023s
=== RUN   TestT4StorageProbeMinio
2026/10/13 07:41:56 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE06ED045D72E8, 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.022s
minio contract: PASS
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 645a36fa0..a14665749 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -1155,6 +1155,7 @@ func TestMovesTextAppliedPlaceholders(t *testing.T) {
// copy" task appends the files it touched; TestMovesScanFilesCoverTheDiff
// fails when the list falls behind.
var movesScanFiles = []string{
+ "internal/host/relocationinstall.go",
"internal/host/rootfloor_reconstruct.go",
// Claim-owned pages share the alert transport; keep their drafts in the copy census.
"internal/console/claim_loss_pages.go",
diff --git a/internal/host/claimretire_transport.go b/internal/host/claimretire_transport.go
index 774e0853a..7286c9728 100644
--- a/internal/host/claimretire_transport.go
+++ b/internal/host/claimretire_transport.go
@@ -14,6 +14,7 @@ import (
"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"
"github.com/carlosframework/platform/internal/registry"
@@ -21,6 +22,7 @@ import (
)
type ClaimRootHTTP struct {
+ Execution *execlease.RootBroker
InventoryActor func(context.Context) (hostbox.Actor, error)
RequireNewWork func(context.Context) error
Journals *hostbox.JournalService
@@ -51,7 +53,7 @@ func (h ClaimRootHTTP) RecoverPending(ctx context.Context) error {
return h.recoverRootInventory(ctx)
}
func (h ClaimRootHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- if r.URL.Path != "/claim/recovery-rebind" && r.URL.Path != "/claim/retire" && r.URL.Path != "/claim/handover" && r.URL.Path != "/claim/journal" {
+ if r.URL.Path != "/claim/execution" && r.URL.Path != "/claim/recovery-rebind" && r.URL.Path != "/claim/retire" && r.URL.Path != "/claim/handover" && r.URL.Path != "/claim/journal" {
h.Stop.ServeHTTP(w, r)
return
}
@@ -73,6 +75,10 @@ func (h ClaimRootHTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
http.Error(w, "refused", http.StatusForbidden)
return
}
+ if r.URL.Path == "/claim/execution" {
+ h.serveExecution(w, r, peer)
+ return
+ }
if r.URL.Path == "/claim/recovery-rebind" {
h.serveRecoveryRebind(w, r)
return
diff --git a/internal/host/plan14_execution_authority_test.go b/internal/host/plan14_execution_authority_test.go
new file mode 100644
index 000000000..e395cc4f3
--- /dev/null
+++ b/internal/host/plan14_execution_authority_test.go
@@ -0,0 +1,119 @@
+package host
+
+import (
+ "context"
+ "testing"
+
+ "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/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestPlan14ExecutionPreinstallationAuthority(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindRelocating, hostclaim.KindLanding, hostclaim.KindMoving} {
+ for _, attack := range []string{"none", "wrong phase", "wrong box", "foreign census", "stale session", "record", "enrollment", "stamp", "incarnation"} {
+ t.Run(kind+"/"+attack, func(t *testing.T) {
+ phase := hostclaim.PhaseInstalling
+ if kind == hostclaim.KindLanding {
+ phase = provision.LandChecking
+ }
+ f, root, iq := plan14InstallRegistration(t, kind, phase)
+ actor, _ := root.InventoryActor(t.Context())
+ if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
+ c.Op.Intent["target_enrollment"] = "en"
+ if kind == hostclaim.KindMoving {
+ c.Op.Exec = "box-0"
+ c.Op.Phase = hostclaim.PhaseMoving
+ }
+ if kind == hostclaim.KindRelocating {
+ c.Op.Exec = "box-0"
+ c.Op.Phase = hostclaim.PhaseStarting
+ }
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ q := execlease.ExecutionRequest{Schema: 1, ID: "begin-1", Operation: "executionBegin", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op}
+ switch attack {
+ case "wrong phase":
+ _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error { c.Op.Phase = "probing"; c.Op.Exec = "box-1"; return nil })
+ if err != nil {
+ t.Fatal(err)
+ }
+ case "wrong box":
+ actor.Box = "unnamed"
+ case "foreign census":
+ if err := root.Reconcile.Registry.Upsert(registry.Route{Host: q.Host, Account: "foreign", App: "foreign"}); err != nil {
+ t.Fatal(err)
+ }
+ case "stale session":
+ q.Session++
+ case "stamp":
+ q.Stamp++
+ case "incarnation":
+ q.Incarnation = "other"
+ case "enrollment":
+ actor.Enrollment = "other"
+ case "record":
+ raw, etag, err := f.claims.Get(t.Context(), iq.RecordKey)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _ = raw
+ if _, err = f.claims.Put(t.Context(), iq.RecordKey, []byte(`{"schema":1,"host":"other.example.com","account":"bab","app":"jam"}`), etag); err != nil {
+ t.Fatal(err)
+ }
+ }
+ root.InventoryActor = func(context.Context) (hostbox.Actor, error) { return actor, nil }
+ a, err := root.executionAuthority(t.Context(), q, execlease.ControllerProcess{})
+ if attack != "none" {
+ if err == nil {
+ t.Fatalf("admitted %s: %+v", attack, a)
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !a.ParticipationOnly || a.Installation != "" {
+ t.Fatalf("preinstallation gained effects: %+v", a)
+ }
+ if _, err := root.Stop.Coordinator.Guard.Capture(q.Host); err == nil {
+ t.Fatal("participation registered installation")
+ }
+ })
+ }
+ }
+}
+
+func TestPlan14ExecutionPendingRegistration(t *testing.T) {
+ for _, kind := range []string{hostclaim.KindServing, hostclaim.KindRelocating, hostclaim.KindLanding} {
+ t.Run(kind, func(t *testing.T) {
+ phase := hostclaim.PhaseInstalling
+ if kind == hostclaim.KindLanding {
+ phase = provision.LandChecking
+ }
+ _, root, iq := plan14InstallRegistration(t, kind, phase)
+ store := &plan14RegistrationCrash{rollbackDisk: &rollbackDisk{dir: t.TempDir()}}
+ root.Stop.Coordinator.Guard.Root = hostguard.NewRootAuthority(store)
+ if _, err := root.commitRootInventory(t.Context(), iq); err == nil {
+ t.Fatal("missed registration interruption")
+ }
+ q := execlease.ExecutionRequest{Schema: 1, ID: "pending-registration", 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.executionAuthority(t.Context(), q, execlease.ControllerProcess{})
+ if err != nil || !a.ParticipationOnly || a.Installation == "" {
+ t.Fatalf("pending registration participation: %+v %v", a, err)
+ }
+ if _, err := root.Stop.Coordinator.Guard.Root.View(t.Context(), q.Host, hostguard.Captured{Incarnation: q.Incarnation, Stamp: q.Stamp}); err == nil {
+ t.Fatal("participation admitted installation effects")
+ }
+ q.Session++
+ if _, err := root.executionAuthority(t.Context(), q, execlease.ControllerProcess{}); err == nil {
+ t.Fatal("stale pending registration session admitted")
+ }
+ })
+ }
+}
diff --git a/internal/host/plan14_root_participation_test.go b/internal/host/plan14_root_participation_test.go
new file mode 100644
index 000000000..a7cd74116
--- /dev/null
+++ b/internal/host/plan14_root_participation_test.go
@@ -0,0 +1,46 @@
+//go:build linux
+
+package host
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+)
+
+func TestPlan14ExecutionPreinstallationSourceGap(t *testing.T) {
+ for _, phase := range []string{hostclaim.PhaseStarting, hostclaim.PhaseReleasing, hostclaim.PhasePinning} {
+ t.Run(phase, func(t *testing.T) {
+ b := newParticipationBox(t)
+ o := b.update(func(c *hostclaim.Claim) error {
+ c.Op.Phase = phase
+ c.Op.Exec = "src"
+ if phase == hostclaim.PhasePinning {
+ c.Op.Exec = "none"
+ }
+ return nil
+ })
+ root := b.guard
+ root.Root = hostguard.NewRootAuthority(&rollbackDisk{dir: t.TempDir()})
+ if _, err := root.Capture(o.Host); !errors.Is(err, hostguard.ErrNoRecord) {
+ t.Fatalf("target unexpectedly has a root installation: %v", err)
+ }
+ if err := rootFirstInstallation(o.Claim, b.actor); err == nil {
+ t.Fatal("first registration admitted before installation")
+ }
+ effects := RelocationEffects{Participation: b.p.Acknowledged}
+ ack, err := effects.Acknowledge(t.Context(), o)
+ if err != nil {
+ t.Fatal("existing participation cannot acknowledge before installation:", err)
+ }
+ if ack.Op != o.Claim.Op.ID || ack.Enrollment != b.actor.Enrollment || ack.LeaseETag == "" {
+ t.Fatalf("missing execution acknowledgement: %+v", ack)
+ }
+ if _, err := root.Capture(o.Host); !errors.Is(err, hostguard.ErrNoRecord) {
+ t.Fatalf("participation fabricated a root installation: %v", err)
+ }
+ })
+ }
+}
diff --git a/internal/host/rootexecution_authority.go b/internal/host/rootexecution_authority.go
new file mode 100644
index 000000000..947eabf82
--- /dev/null
+++ b/internal/host/rootexecution_authority.go
@@ -0,0 +1,132 @@
+package host
+
+import (
+ "context"
+ "database/sql"
+ "errors"
+ "fmt"
+ "reflect"
+ "slices"
+ "strconv"
+
+ "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"
+)
+
+func (h ClaimRootHTTP) executionAuthority(ctx context.Context, q execlease.ExecutionRequest, _ execlease.ControllerProcess) (execlease.ExecutionAuthority, error) {
+ var out execlease.ExecutionAuthority
+ if h.InventoryActor == nil || h.Stop.Coordinator.Guard.Root == nil || h.Reconcile.Census == nil || h.Journals == nil {
+ return out, hostbox.ErrUnsupported
+ }
+ actor, err := h.InventoryActor(ctx)
+ if err != nil {
+ return out, err
+ }
+ if h.RequireNewWork != nil {
+ if err = h.RequireNewWork(ctx); err != nil {
+ return out, err
+ }
+ }
+ obs, err := hostclaim.Read(ctx, h.Stop.Coordinator.Evidence, q.Host)
+ if err != nil {
+ return out, err
+ }
+ c := obs.Claim
+ if obs.Absent || c == nil || c.Holder == nil || c.Validate() != nil || c.Host != q.Host || c.Holder.Incarnation != q.Incarnation || c.Holder.Stamp != q.Stamp || c.Holder.Session != q.Session || c.Op.ID != q.Op || slices.Contains(c.Lost, actor.Box) {
+ return out, hostbox.ErrChanged
+ }
+ if c.Op.Kind != hostclaim.KindServing && !slices.Contains(c.Op.Participants, actor.Box) {
+ return out, hostbox.ErrChanged
+ }
+ _, etag, err := rootClaimRecord(ctx, h.Stop.Coordinator.Evidence, c, h.Roots["data"])
+ if err != nil {
+ return out, err
+ }
+ out = execlease.ExecutionAuthority{Box: actor.Box, Enrollment: actor.Enrollment, Account: c.Holder.Account, App: c.Holder.App, ClaimRevision: obs.ETag, RecordRevision: etag, DirectoryRevision: strconv.FormatUint(actor.Revision, 10), Kind: c.Op.Kind, Phase: c.Op.Phase}
+ captured := hostguard.Captured{Incarnation: q.Incarnation, Stamp: q.Stamp}
+ view, err := h.Stop.Coordinator.Guard.Root.View(ctx, q.Host, captured)
+ if errors.Is(err, hostguard.ErrFenced) {
+ in, raw, pendingErr := h.Stop.Coordinator.Guard.Root.PendingInstallation(ctx, q.Host)
+ if pendingErr != nil {
+ return out, pendingErr
+ }
+ var pending hostbox.JournalRequest
+ if rootparticipation.Decode(raw, &pending) != nil || pending.Op != "inventory-write" || pending.Revision != 0 || pending.RecordKey != c.Holder.Key || pending.RecordETag != etag || pending.Target.ClaimETag != obs.ETag || pending.Target.Host != q.Host || pending.Target.Incarnation != q.Incarnation || pending.Target.Stamp != q.Stamp || pending.Target.Session != q.Session || pending.Target.Op != q.Op || pending.Target.Phase != c.Op.Phase || pending.Target.Box != actor.Box || pending.Target.Enrollment != actor.Enrollment || in.Host != q.Host || in.Incarnation != q.Incarnation || in.Stamp != q.Stamp || in.Account != out.Account || in.App != out.App || in.Record != c.Holder.Key || in.Box != actor.Box || in.Enrollment != actor.Enrollment || in.Migration == 0 {
+ return out, hostbox.ErrChanged
+ }
+ if err := execlease.PreinstallationParticipant(c, actor.Box, actor.Enrollment); err != nil {
+ return out, err
+ }
+ inventory, revision, err := h.Journals.Inventory.Snapshot(ctx, q.Host, captured)
+ if err != nil {
+ return out, err
+ }
+ if revision == 0 {
+ selected := hostbox.RecordSelection{Host: q.Host, Key: c.Holder.Key, ETag: etag, Incarnation: q.Incarnation, Family: c.Holder.Family}
+ if err := h.Reconcile.Census.CheckLocal(ctx, obs, selected); err != nil {
+ return out, err
+ }
+ } else {
+ expected := pending.Inventory
+ if c.Op.Kind == hostclaim.KindLanding {
+ expected.Entries = []hostbox.StopInventoryEntry{rootLandingEntry(pending, c, actor)}
+ }
+ if revision != in.InventoryRevision || !reflect.DeepEqual(inventory, expected) {
+ return out, hostbox.ErrTypedInventory
+ }
+ }
+ out.ParticipationOnly = true
+ out.Migration = in.Migration
+ out.Installation = fmt.Sprintf("%s/%d/%s/%s", in.Incarnation, in.Stamp, in.Box, in.Enrollment)
+ return out, nil
+ }
+ if errors.Is(err, hostguard.ErrNoRecord) {
+ if err = execlease.PreinstallationParticipant(c, actor.Box, actor.Enrollment); err != nil {
+ return out, err
+ }
+ // An orphan journal is not a root registration, even when its complete bit is set.
+ if _, err = h.Journals.Inventory.ReadStopInventory(ctx, q.Host); !errors.Is(err, sql.ErrNoRows) {
+ return out, hostbox.ErrTypedInventory
+ }
+ selected := hostbox.RecordSelection{Host: q.Host, Key: c.Holder.Key, ETag: etag, Incarnation: q.Incarnation, Family: c.Holder.Family}
+ if err = h.Reconcile.Census.CheckLocal(ctx, obs, selected); err != nil {
+ return out, err
+ }
+ out.ParticipationOnly = true
+ return out, nil
+ }
+ if err != nil {
+ return out, err
+ }
+ in := view.Installation
+ if in.Host != q.Host || in.Incarnation != q.Incarnation || in.Stamp != q.Stamp || in.Account != out.Account || in.App != out.App || in.Record != c.Holder.Key || in.Box != actor.Box || in.Enrollment != actor.Enrollment || in.Migration == 0 || in.InventoryRevision == 0 {
+ return out, hostbox.ErrChanged
+ }
+ if view.Barrier != nil || slices.Contains(view.Retired, captured) {
+ return out, hostguard.ErrFenced
+ }
+ denied := func(t hostguard.FenceTarget) bool {
+ return t.Incarnation == q.Incarnation && t.Stamp == q.Stamp && t.Session == q.Session
+ }
+ if view.Pending != nil && denied(*view.Pending) {
+ return out, hostguard.ErrFenced
+ }
+ for _, receipt := range view.Receipts {
+ if denied(receipt.Target) {
+ return out, hostguard.ErrFenced
+ }
+ }
+ _, revision, err := h.Journals.Inventory.Snapshot(ctx, q.Host, captured)
+ if err != nil {
+ return out, err
+ }
+ if revision != in.InventoryRevision {
+ return out, hostbox.ErrChanged
+ }
+ out.Migration = in.Migration
+ out.Installation = fmt.Sprintf("%s/%d/%s/%s", in.Incarnation, in.Stamp, in.Box, in.Enrollment)
+ return out, nil
+}
diff --git a/internal/host/rootexecution_read.go b/internal/host/rootexecution_read.go
new file mode 100644
index 000000000..10a41d606
--- /dev/null
+++ b/internal/host/rootexecution_read.go
@@ -0,0 +1,23 @@
+package host
+
+import (
+ "context"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execlease "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/rootparticipation"
+ "github.com/carlosframework/platform/internal/rootstate"
+)
+
+func rootExecutionRead(ctx context.Context, 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
+ }
+ 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)
+}
diff --git a/internal/host/rootexecution_transport.go b/internal/host/rootexecution_transport.go
new file mode 100644
index 000000000..a3c710587
--- /dev/null
+++ b/internal/host/rootexecution_transport.go
@@ -0,0 +1,81 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "net"
+ "net/http"
+ "time"
+
+ execlease "github.com/carlosframework/platform/internal/hostexec"
+)
+
+func (h ClaimRootHTTP) serveExecution(w http.ResponseWriter, r *http.Request, peer StopPeer) {
+ var q execlease.ExecutionRequest
+ dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10))
+ dec.DisallowUnknownFields()
+ if dec.Decode(&q) != nil || dec.Decode(new(any)) != io.EOF {
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ if h.Execution == nil {
+ w.WriteHeader(http.StatusServiceUnavailable)
+ return
+ }
+ out, err := h.Execution.Execute(r.Context(), execlease.ControllerProcess{KernelBootID: peer.BootID, PID: peer.PID, Start: peer.ProcessStart}, q)
+ if err != nil {
+ w.WriteHeader(http.StatusConflict)
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(out)
+}
+
+func rootExecutionClient(path string) func(context.Context, execlease.ExecutionRequest) (execlease.ExecutionResponse, error) {
+ return func(ctx context.Context, q execlease.ExecutionRequest) (execlease.ExecutionResponse, error) {
+ var out execlease.ExecutionResponse
+ raw, err := json.Marshal(q)
+ if err != nil {
+ return out, err
+ }
+ ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+ defer cancel()
+ tr := &http.Transport{DisableKeepAlives: true, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", path)
+ }}
+ defer tr.CloseIdleConnections()
+ for attempt := 0; attempt < 2; attempt++ {
+ req, e := http.NewRequestWithContext(ctx, http.MethodPost, "http://control/claim/execution", bytes.NewReader(raw))
+ if e != nil {
+ return out, e
+ }
+ resp, e := (&http.Client{Transport: tr}).Do(req)
+ if e != nil {
+ err = e
+ continue
+ }
+ if resp.StatusCode != http.StatusOK {
+ resp.Body.Close()
+ return out, execlease.ErrLost
+ }
+ limited := &io.LimitedReader{R: resp.Body, N: (16 << 10) + 1}
+ dec := json.NewDecoder(limited)
+ dec.DisallowUnknownFields()
+ err = dec.Decode(&out)
+ if err == nil && (dec.Decode(new(any)) != io.EOF || limited.N == 0) {
+ err = execlease.ErrState
+ }
+ resp.Body.Close()
+ if err != nil {
+ continue
+ }
+ if out.Schema != 1 || out.Host != q.Host || out.Incarnation != q.Incarnation || out.Stamp != q.Stamp || out.Session != q.Session || out.Op != q.Op || out.Token == "" || out.BootID == "" {
+ return execlease.ExecutionResponse{}, execlease.ErrState
+ }
+ return out, nil
+ }
+ return out, err
+ }
+}
diff --git a/internal/host/rootexecution_transport_test.go b/internal/host/rootexecution_transport_test.go
new file mode 100644
index 000000000..f97df4cab
--- /dev/null
+++ b/internal/host/rootexecution_transport_test.go
@@ -0,0 +1,87 @@
+//go:build linux
+
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "net"
+ "net/http"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ execlease "github.com/carlosframework/platform/internal/hostexec"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+func TestPlan14ExecutionBrokerTransport(t *testing.T) {
+ f, root, iq := plan14InstallRegistration(t, hostclaim.KindLanding, provision.LandChecking)
+ q := execlease.ExecutionRequest{Schema: 1, ID: "request-1", Operation: "executionBegin", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op}
+ peer := StopPeer{UID: 1001, PID: 42, ProcessStart: "start", BootID: "boot", ControllerID: "controller"}
+ root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return peer, nil }
+ calls := 0
+ b, err := execlease.NewRootBroker(execlease.RootBrokerConfig{State: &rollbackDisk{dir: t.TempDir()}, Remote: f.claims, Clock: execlease.SystemClock{}, BootID: peer.BootID, Identity: "broker", Authority: func(ctx context.Context, q execlease.ExecutionRequest, p execlease.ControllerProcess) (execlease.ExecutionAuthority, error) {
+ calls++
+ a, e := root.executionAuthority(ctx, q, p)
+ a.Migration = 1
+ return a, e
+ }, Alive: func(context.Context, execlease.ControllerProcess) error { return nil }, Cleanup: func(context.Context, execlease.ExecutionRecord) error { return nil }, Settle: func(context.Context, execlease.ExecutionRecord) error { return nil }, Exclude: func(_ context.Context, fn func() error) error { return fn() }})
+ if err != nil {
+ t.Fatal(err)
+ }
+ root.Execution = b
+ t.Cleanup(func() { _ = b.Close(context.Background()) })
+ socket := serveClaimRoot(t, root)
+ tr := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return (&net.Dialer{}).DialContext(ctx, "unix", socket)
+ }}
+ defer tr.CloseIdleConnections()
+ raw, _ := json.Marshal(q)
+ for _, field := range []string{`"expiry":"2099-01-01"`, `"pid":42`, `"path":"/data"`, `"duration":3600`, `"safe_until":999999999999`} {
+ forged := append(append([]byte{}, raw[:len(raw)-1]...), []byte(","+field+"}")...)
+ req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://control/claim/execution", bytes.NewReader(forged))
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp, err := (&http.Client{Transport: tr}).Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusBadRequest || calls != 0 {
+ t.Fatalf("untrusted field %s reached authority: status=%d calls=%d", field, resp.StatusCode, calls)
+ }
+ }
+ oversized := append(append([]byte(nil), raw...), bytes.Repeat([]byte(" "), 16<<10)...)
+ req, err := http.NewRequestWithContext(t.Context(), http.MethodPost, "http://control/claim/execution", bytes.NewReader(oversized))
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp, err := (&http.Client{Transport: tr}).Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusBadRequest || calls != 0 {
+ t.Fatalf("oversized request reached authority: status=%d calls=%d", resp.StatusCode, calls)
+ }
+ out, err := rootExecutionClient(socket)(t.Context(), q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if out.Status != "active" || out.Token == "" {
+ t.Fatal(out)
+ }
+ again, err := rootExecutionClient(socket)(t.Context(), q)
+ if err != nil || again != out {
+ t.Fatalf("replay: %+v %v", again, err)
+ }
+ peer.ProcessStart = "reused"
+ q.Operation = "executionView"
+ q.ID = "view"
+ q.Token = out.Token
+ if _, err := rootExecutionClient(socket)(t.Context(), q); err == nil {
+ t.Fatal("PID reuse borrowed execution")
+ }
+}
diff --git a/internal/hostexec/participation.go b/internal/hostexec/participation.go
new file mode 100644
index 000000000..dc8b85af5
--- /dev/null
+++ b/internal/hostexec/participation.go
@@ -0,0 +1,48 @@
+package hostexec
+
+import (
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "slices"
+)
+
+func PreinstallationParticipant(c *hostclaim.Claim, box, enrollment string) error {
+ if c == nil || c.Holder == nil || c.Validate() != nil || box == "" || enrollment == "" || slices.Contains(c.Lost, box) {
+ return ErrInactive
+ }
+ if c.Op.Kind == hostclaim.KindServing {
+ if c.Op.Exec != box || (c.Holder.WriterBox != box && (c.Holder.WriterBox != "" || c.Holder.Session != 0)) {
+ return ErrInactive
+ }
+ return nil
+ }
+ if !slices.Contains(c.Op.Participants, box) || c.Op.ID == "" || c.Holder.Family != "instance" {
+ return ErrInactive
+ }
+ target := c.Op.Intent["target"]
+ if c.Op.Kind == hostclaim.KindRelocating {
+ target = c.Op.Intent["target_box"]
+ }
+ if target != box || (c.Op.Intent["target_enrollment"] != "" && c.Op.Intent["target_enrollment"] != enrollment) {
+ return ErrInactive
+ }
+ switch c.Op.Kind {
+ case hostclaim.KindRelocating:
+ switch c.Op.Phase {
+ case hostclaim.PhaseStarting, hostclaim.PhaseReleasing, hostclaim.PhasePinning:
+ return nil
+ case hostclaim.PhaseInstalling, hostclaim.PhaseInstalled:
+ if c.Op.Exec == box && c.Op.Journal["local_install"] == "" {
+ return nil
+ }
+ }
+ case hostclaim.KindMoving:
+ if c.Op.Phase == hostclaim.PhaseMoving || c.Op.Phase == hostclaim.PhaseSealed {
+ return nil
+ }
+ case hostclaim.KindLanding:
+ if c.Op.Phase == "checking" && c.Op.Exec == box {
+ return nil
+ }
+ }
+ return ErrInactive
+}
diff --git a/internal/hostexec/plan14_root_participation_test.go b/internal/hostexec/plan14_root_participation_test.go
new file mode 100644
index 000000000..838cda8e1
--- /dev/null
+++ b/internal/hostexec/plan14_root_participation_test.go
@@ -0,0 +1,723 @@
+//go:build linux
+
+package hostexec
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "golang.org/x/sys/unix"
+)
+
+type brokerDisk struct {
+ dir string
+ writes int
+ failAt int
+ afterRename func(string, []byte)
+}
+
+func (d *brokerDisk) Read(n string) ([]byte, error) { return os.ReadFile(filepath.Join(d.dir, n)) }
+func (d *brokerDisk) Write(n string, b []byte) error {
+ d.writes++
+ f, err := os.CreateTemp(d.dir, "pending-")
+ if err != nil {
+ return err
+ }
+ name := f.Name()
+ defer os.Remove(name)
+ if _, err = f.Write(b); err == nil {
+ err = f.Sync()
+ }
+ err = errors.Join(err, f.Close())
+ if err != nil {
+ return err
+ }
+ if err = os.Rename(name, filepath.Join(d.dir, n)); err != nil {
+ return err
+ }
+ if d.afterRename != nil {
+ d.afterRename(n, b)
+ }
+ if d.writes == d.failAt {
+ return os.ErrInvalid
+ }
+ parent, err := os.Open(d.dir)
+ if err != nil {
+ return err
+ }
+ return errors.Join(parent.Sync(), parent.Close())
+}
+func (d *brokerDisk) Remove(n string) error { return os.Remove(filepath.Join(d.dir, n)) }
+func (d *brokerDisk) Names() ([]string, error) {
+ es, e := os.ReadDir(d.dir)
+ var out []string
+ for _, v := range es {
+ out = append(out, v.Name())
+ }
+ return out, e
+}
+func (d *brokerDisk) 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.EWOULDBLOCK) && !errors.Is(err, unix.EAGAIN) {
+ f.Close()
+ return nil, err
+ }
+ select {
+ case <-ctx.Done():
+ f.Close()
+ return nil, ctx.Err()
+ case <-time.After(time.Millisecond):
+ }
+ }
+}
+func (d *brokerDisk) Close() error { return nil }
+
+func brokerFixture(t *testing.T) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
+ t.Helper()
+ return brokerFixtureAt(t, t.TempDir(), t.TempDir())
+}
+func brokerFixtureAt(t *testing.T, local, remotePath string) (*RootBroker, *brokerDisk, *delayedStore, *testClock, ExecutionRequest, ControllerProcess) {
+ t.Helper()
+ d := &brokerDisk{dir: local}
+ clock := &testClock{wall: time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC), boot: int64(100 * time.Second)}
+ remote := &delayedStore{Store: release.NewDirStore(remotePath)}
+ q := ExecutionRequest{Schema: 1, ID: "begin-1", Operation: "executionBegin", Host: "shop.example.com", Incarnation: "inc", Stamp: 3, Session: 2, Op: "move-1"}
+ p := ControllerProcess{KernelBootID: "boot", PID: 42, Start: "start-42"}
+ b, err := NewRootBroker(RootBrokerConfig{State: d, Remote: remote, Clock: clock, BootID: "boot", Identity: "broker-1",
+ Authority: func(_ context.Context, got ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
+ if got.Host != q.Host || got.Incarnation != q.Incarnation || got.Stamp != q.Stamp || got.Session != q.Session || got.Op != q.Op || peer != p {
+ return ExecutionAuthority{}, ErrInactive
+ }
+ return ExecutionAuthority{Box: "box", Enrollment: "en", Account: "bab", App: "jam", Installation: "installation-1", Migration: 1, ClaimRevision: "claim-1", RecordRevision: "record-1", DirectoryRevision: "directory-1", Kind: "landing", Phase: "checking"}, nil
+ }, Alive: func(context.Context, ControllerProcess) error { return nil }, Cleanup: func(context.Context, ExecutionRecord) error { return nil }, Settle: func(context.Context, ExecutionRecord) error { return nil },
+ Exclude: func(_ context.Context, fn func() error) error { return fn() },
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = b.Close(context.Background()) })
+ return b, d, remote, clock, q, p
+}
+
+func TestPlan14ExecutionBrokerPeerAndShape(t *testing.T) {
+ for _, mutation := range []string{"schema", "host", "incarnation", "stamp", "session", "operation", "peer-start", "peer-pid", "peer-boot"} {
+ t.Run(mutation, func(t *testing.T) {
+ b, d, remote, _, q, p := brokerFixture(t)
+ switch mutation {
+ case "schema":
+ q.Schema++
+ case "host":
+ q.Host = "other.example.com"
+ case "incarnation":
+ q.Incarnation = "other"
+ case "stamp":
+ q.Stamp++
+ case "session":
+ q.Session++
+ case "operation":
+ q.Op = "other"
+ case "peer-start":
+ p.Start = "reused"
+ case "peer-pid":
+ p.PID++
+ case "peer-boot":
+ p.KernelBootID = "reboot"
+ }
+ if _, err := b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("altered identity acquired participation")
+ }
+ if remote.calls != 0 || d.writes != 0 {
+ t.Fatalf("refusal had effects: remote=%d writes=%d", remote.calls, d.writes)
+ }
+ })
+ }
+ b, d, remote, _, q, p := brokerFixture(t)
+ first, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ writes := d.writes
+ again, err := b.Execute(t.Context(), p, q)
+ if err != nil || first != again {
+ t.Fatalf("replay: %+v %+v %v", first, again, err)
+ }
+ if remote.calls != 1 || d.writes != writes {
+ t.Fatal("replay renewed or rewrote participation")
+ }
+ q.ID = "another-controller"
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("another controller borrowed token")
+ }
+}
+
+func TestPlan14ExecutionDeadlineFromRequestStart(t *testing.T) {
+ for _, delay := range []time.Duration{30 * time.Second, 85 * time.Second, 100 * time.Second} {
+ t.Run(delay.String(), func(t *testing.T) {
+ b, _, remote, clock, q, p := brokerFixture(t)
+ remote.onPut = func(int) { clock.advance(delay) }
+ out, err := b.Execute(t.Context(), p, q)
+ if delay >= 85*time.Second {
+ if err == nil {
+ t.Fatal("late response granted participation")
+ }
+ return
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if out.SafeUntil != int64(185*time.Second) {
+ t.Fatalf("response-relative deadline: %v", out)
+ }
+ clock.advance(55 * time.Second)
+ q.Operation = "executionView"
+ q.ID = "view-1"
+ q.Token = out.Token
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("suspend beyond cutoff retained permission")
+ }
+ })
+ }
+}
+
+func TestPlan14ExecutionPendingSurvivesCommitFailure(t *testing.T) {
+ b, d, remote, _, q, p := brokerFixture(t)
+ d.failAt = 2
+ if _, err := b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("failed commit replied successfully")
+ }
+ if remote.calls != 1 {
+ t.Fatal("did not cross remote acquisition")
+ }
+ if _, err := b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("uncertain commit replay granted permission")
+ }
+ if remote.calls != 1 {
+ t.Fatal("uncertain replay acquired twice")
+ }
+ var record ExecutionRecord
+ raw, err := d.Read(q.Host + ".json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = json.Unmarshal(raw, &record); err != nil {
+ t.Fatal(err)
+ }
+ if record.Status == "settled" {
+ t.Fatal("uncertain acquisition lost obligation")
+ }
+}
+
+func TestPlan14ExecutionBrokerRestartCannotResume(t *testing.T) {
+ b, _, _, _, q, p := brokerFixture(t)
+ permit, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := b.Close(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ cfg := b.cfg
+ cfg.Identity = "broker-2"
+ restarted, err := NewRootBroker(cfg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer restarted.Close(t.Context())
+ q.Operation = "executionView"
+ q.Token = permit.Token
+ q.ID = "view-1"
+ if _, err = restarted.Execute(t.Context(), p, q); !errors.Is(err, ErrLost) {
+ t.Fatalf("restart restored prior permit: %v", err)
+ }
+}
+
+func TestPlan14ExecutionAuthorityLossIsSticky(t *testing.T) {
+ b, _, _, _, q, p := brokerFixture(t)
+ out, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ original := b.cfg.Authority
+ b.cfg.Authority = func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error) {
+ return ExecutionAuthority{}, ErrInactive
+ }
+ q.Operation = "executionView"
+ q.Token = out.Token
+ q.ID = "changed"
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("changed authority remained active")
+ }
+ b.cfg.Authority = original
+ q.ID = "restored"
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("restoring authority resurrected lost lease")
+ }
+}
+
+func TestPlan14ExecutionInvalidAuthorityAndReplayAreSticky(t *testing.T) {
+ for _, name := range []string{"invalid", "replay"} {
+ t.Run(name, func(t *testing.T) {
+ replay := name == "replay"
+ b, _, _, _, q, peer := brokerFixture(t)
+ initial, err := b.Execute(t.Context(), peer, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ original := b.cfg.Authority
+ b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
+ a, err := original(ctx, q, p)
+ if replay {
+ a.Phase = "changed"
+ } else {
+ a.Migration = 0
+ }
+ return a, err
+ }
+ if !replay {
+ q.Operation, q.ID, q.Token = "executionView", "invalid-authority", initial.Token
+ }
+ if _, err := b.Execute(t.Context(), peer, q); err == nil {
+ t.Fatal("invalid authority accepted")
+ }
+ b.cfg.Authority = original
+ q.Operation, q.ID, q.Token = "executionView", "after-repair", initial.Token
+ if _, err := b.Execute(t.Context(), peer, q); err == nil {
+ t.Fatal("authority repair resurrected denied participation", replay)
+ }
+ })
+ }
+}
+
+func TestPlan14ExecutionUncertainReleaseCanSettleAfterExpiry(t *testing.T) {
+ b, d, remote, clock, q, p := brokerFixture(t)
+ remote.onPut = func(int) { d.failAt = d.writes + 1 }
+ if _, err := b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("failed acquisition committed")
+ }
+ r, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // Simulate the crash window with no retained conditional-write response.
+ r.LeaseETag = ""
+ r.Status = "pending"
+ if err = b.write(r); err != nil {
+ t.Fatal(err)
+ }
+ q.Operation = "executionSettle"
+ q.Token = r.Token
+ q.ID = "settle"
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("unexpired uncertain lease settled")
+ }
+ clock.advance(TTL + time.Second)
+ if _, err = b.Execute(t.Context(), p, q); err != nil {
+ t.Fatal("expired exact obligation cannot settle", err)
+ }
+}
+
+func TestPlan14ExecutionMonitorIndependentOfRenewal(t *testing.T) {
+ b, _, remote, clock, q, p := brokerFixture(t)
+ out, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ entered, releasePut := make(chan struct{}), make(chan struct{})
+ remote.onPut = func(int) { close(entered); <-releasePut }
+ cleaned := make(chan struct{}, 1)
+ b.cfg.Cleanup = func(context.Context, ExecutionRecord) error {
+ select {
+ case cleaned <- struct{}{}:
+ default:
+ }
+ return nil
+ }
+ done := make(chan error, 1)
+ q.Operation = "executionRenew"
+ q.ID = "renew"
+ q.Token = out.Token
+ go func() { _, err := b.Execute(context.Background(), p, q); done <- err }()
+ <-entered
+ clock.advance(TTL)
+ monitored := make(chan error, 1)
+ go func() { monitored <- b.Monitor(context.Background()) }()
+ select {
+ case <-cleaned:
+ case <-time.After(time.Second):
+ close(releasePut)
+ <-done
+ t.Fatal("network renewal blocked deadline cleanup")
+ }
+ close(releasePut)
+ if err = <-done; err == nil {
+ t.Fatal("late renewal revived expired lease")
+ }
+ <-monitored
+}
+
+func TestPlan14ExecutionBindsRegisteredInstallation(t *testing.T) {
+ b, _, _, _, q, p := brokerFixture(t)
+ base := b.cfg.Authority
+ installed := false
+ b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
+ a, e := base(ctx, q, p)
+ if !installed {
+ a.Installation = ""
+ a.ParticipationOnly = true
+ }
+ return a, e
+ }
+ out, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ installed = true
+ q.Operation = "executionView"
+ q.ID = "registered"
+ q.Token = out.Token
+ if _, err = b.Execute(t.Context(), p, q); err != nil {
+ t.Fatal("root registration lost participation", err)
+ }
+ r, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if r.Authority.Installation == "" || r.Authority.ParticipationOnly {
+ t.Fatal("lease was not bound to root installation")
+ }
+ installed = false
+ q.ID = "removed"
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("removed root installation retained participation")
+ }
+}
+
+func TestPlan14ExecutionRootReadIgnoresMirrors(t *testing.T) {
+ b, d, _, _, q, p := brokerFixture(t)
+ original, oldPeer := b.cfg.Authority, p
+ var err error
+ p, err = currentControllerProcess()
+ if err != nil {
+ t.Fatal(err)
+ }
+ b.cfg.BootID = p.KernelBootID
+ b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
+ return original(ctx, q, oldPeer)
+ }
+ out, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
+ if err = d.Write("census.json", census); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
+ t.Fatal("valid root permit refused", err)
+ }
+ saved, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ only := saved
+ only.Authority.ParticipationOnly = true
+ if err := b.write(only); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
+ t.Fatal("participation-only record granted privileged execution")
+ }
+ if err := b.write(saved); err != nil {
+ t.Fatal(err)
+ }
+ data := t.TempDir()
+ mirror := filepath.Join(data, ".hostexec")
+ if err = os.Mkdir(mirror, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err = os.WriteFile(filepath.Join(mirror, q.Host+".json"), []byte(`{"safe_until_boottime_ns":999999999999999999,"state":"active"}`), 0600); err != nil {
+ t.Fatal(err)
+ }
+ if err = os.Rename(mirror, filepath.Join(data, "old")); err != nil {
+ t.Fatal(err)
+ }
+ if err = os.Mkdir(mirror, 0700); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err != nil {
+ t.Fatal("service replacement changed root permit", err)
+ }
+ // A failed durable loss still drops the root-owned live lock.
+ d.failAt = d.writes + 1
+ r, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = b.deny(t.Context(), r, ErrLost); err == nil {
+ t.Fatal("lost lease reported success")
+ }
+ r.Status = "active"
+ r.Token = out.Token
+ raw, _ := json.Marshal(r)
+ if err = os.WriteFile(filepath.Join(d.dir, q.Host+".json"), raw, 0600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err = ReadRoot(t.Context(), d, b.cfg.Clock, q, false); err == nil {
+ t.Fatal("old active bytes revived revoked live lock")
+ }
+}
+
+func TestPlan14ExecutionLossPersistenceRevokesAllLiveLocks(t *testing.T) {
+ b, d, _, _, q, peer := brokerFixture(t)
+ original := b.cfg.Authority
+ b.cfg.Authority = func(ctx context.Context, request ExecutionRequest, peer ControllerProcess) (ExecutionAuthority, error) {
+ request.Host = q.Host
+ return original(ctx, request, peer)
+ }
+ if _, err := b.Execute(t.Context(), peer, q); err != nil {
+ t.Fatal(err)
+ }
+ other := q
+ other.Host = "other.example.com"
+ if _, err := b.Execute(t.Context(), peer, other); err != nil {
+ t.Fatal(err)
+ }
+ r, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ d.failAt = d.writes + 1
+ if err := b.deny(t.Context(), r, ErrLost); err == nil {
+ t.Fatal("loss persistence failure accepted")
+ }
+ ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
+ defer cancel()
+ unlock, err := d.Lock(ctx, other.Host+".live")
+ if err != nil {
+ t.Fatal("independent reader can still observe another live permit", err)
+ }
+ unlock()
+}
+
+func TestPlan14ExecutionVisibleRenameCannotGrantBeforeSync(t *testing.T) {
+ b, d, _, clock, q, priorPeer := brokerFixture(t)
+ peer, err := currentControllerProcess()
+ if err != nil {
+ t.Fatal(err)
+ }
+ b.cfg.BootID = peer.KernelBootID
+ original := b.cfg.Authority
+ b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, _ ControllerProcess) (ExecutionAuthority, error) {
+ return original(ctx, q, priorPeer)
+ }
+ initial, err := b.Execute(t.Context(), peer, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ census, _ := json.Marshal(map[string]any{"Schema": 1, "Box": "box", "Enrollment": "en", "Migration": 1, "Revision": 1, "LaunchRevision": 1, "Complete": true})
+ if err := d.Write("census.json", census); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
+ t.Fatal("initial permit", err)
+ }
+ clock.advance(Renewal)
+ visible, finish := make(chan struct{}), make(chan struct{})
+ d.afterRename = func(_ string, raw []byte) {
+ var r ExecutionRecord
+ if json.Unmarshal(raw, &r) == nil && r.Status == "active" {
+ close(visible)
+ <-finish
+ }
+ }
+ q.Operation, q.ID, q.Token = "executionRenew", "renew-before-sync", initial.Token
+ done := make(chan error, 1)
+ go func() { _, err := b.Execute(t.Context(), peer, q); done <- err }()
+ select {
+ case <-visible:
+ case <-time.After(time.Second):
+ t.Fatal("renew did not reach visible rename")
+ }
+ _, readErr := ReadRoot(t.Context(), d, clock, q, false)
+ close(finish)
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+ if readErr == nil {
+ t.Fatal("visible but unsynced active bytes granted execution")
+ }
+ if _, err := ReadRoot(t.Context(), d, clock, q, false); err != nil {
+ t.Fatal("durable renewal refused", err)
+ }
+}
+
+func TestPlan14ExecutionSettledTombstoneDoesNotBlockNextOperation(t *testing.T) {
+ b, _, _, _, q, p := brokerFixture(t)
+ out, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ settle := q
+ settle.ID = "settle"
+ settle.Operation = "executionSettle"
+ settle.Token = out.Token
+ if _, err = b.Execute(t.Context(), p, settle); err != nil {
+ t.Fatal(err)
+ }
+ next := q
+ next.ID = "next-begin"
+ next.Op = "move-2"
+ original := b.cfg.Authority
+ b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
+ copy := got
+ copy.Op = q.Op
+ return original(ctx, copy, p)
+ }
+ granted, err := b.Execute(t.Context(), p, next)
+ if err != nil {
+ t.Fatal("settled operation permanently blocked host", err)
+ }
+ if granted.Token == out.Token {
+ t.Fatal("new operation reused old permission")
+ }
+ if old, err := b.Execute(t.Context(), p, settle); err != nil || old.Status != "settled" {
+ t.Fatalf("settlement tombstone lost: %+v %v", old, err)
+ }
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("old begin resurrected")
+ }
+}
+
+func TestPlan14ExecutionSessionTransitionRetainsRemoteOwner(t *testing.T) {
+ b, _, remote, _, q, p := brokerFixture(t)
+ original := b.cfg.Authority
+ changed := false
+ b.cfg.Authority = func(ctx context.Context, got ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) {
+ expected := q.Session
+ if changed {
+ expected++
+ }
+ if got.Session != expected {
+ return ExecutionAuthority{}, ErrInactive
+ }
+ copy := got
+ copy.Session = q.Session
+ a, e := original(ctx, copy, p)
+ if changed {
+ a.ClaimRevision = "claim-2"
+ a.Phase = "configured"
+ }
+ return a, e
+ }
+ b.cfg.Transition = func(_ context.Context, old ExecutionRecord, next ExecutionRequest, a ExecutionAuthority) error {
+ if old.Request.Session+1 != next.Session || old.Authority.Installation != a.Installation {
+ return ErrLost
+ }
+ return nil
+ }
+ first, err := b.Execute(t.Context(), p, q)
+ if err != nil {
+ t.Fatal(err)
+ }
+ changed = true
+ next := q
+ next.ID = "new-session"
+ next.Operation = "executionRenew"
+ next.Session++
+ next.Token = first.Token
+ second, err := b.Execute(t.Context(), p, next)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if second.Token == first.Token || second.Session != next.Session || remote.calls != 2 {
+ t.Fatalf("transition: %+v calls=%d", second, remote.calls)
+ }
+ again, err := b.Execute(t.Context(), p, next)
+ if err != nil || again != second || remote.calls != 2 {
+ t.Fatalf("lost transition response replay: %+v %v", again, err)
+ }
+ stale := next
+ stale.Operation = "executionView"
+ stale.ID = "stale-view"
+ if _, err = b.Execute(t.Context(), p, stale); err == nil {
+ t.Fatal("old token admitted new session")
+ }
+}
+
+type executionCrashStore struct{ release.Store }
+
+func (s executionCrashStore) Put(ctx context.Context, key string, raw []byte, etag string) (string, error) {
+ next, err := s.Store.Put(ctx, key, raw, etag)
+ if err == nil {
+ os.Exit(73)
+ }
+ return next, err
+}
+
+func TestPlan14ExecutionProcessCrashBeforeCommit(t *testing.T) {
+ if dir := os.Getenv("PLAN14_EXECUTION_CRASH_DIR"); dir != "" {
+ b, _, remote, _, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
+ b.cfg.Remote = executionCrashStore{remote.Store}
+ _, _ = b.Execute(context.Background(), p, q)
+ t.Fatal("crash boundary was not reached")
+ }
+ dir := t.TempDir()
+ for _, name := range []string{"local", "remote"} {
+ if err := os.Mkdir(filepath.Join(dir, name), 0700); err != nil {
+ t.Fatal(err)
+ }
+ }
+ child := exec.Command(os.Args[0], "-test.run=^TestPlan14ExecutionProcessCrashBeforeCommit$", "-test.count=1")
+ child.Dir = dir
+ child.Env = append(os.Environ(), "PLAN14_EXECUTION_CRASH_DIR="+dir)
+ raw, err := child.CombinedOutput()
+ var exited *exec.ExitError
+ if !errors.As(err, &exited) || exited.ExitCode() != 73 {
+ t.Fatalf("child: %v %s", err, raw)
+ }
+ b, _, remote, clock, q, p := brokerFixtureAt(t, filepath.Join(dir, "local"), filepath.Join(dir, "remote"))
+ b.cfg.Identity = "restarted-broker"
+ record, err := b.read(q.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if record.Status != "pending" || record.LeaseETag != "" {
+ t.Fatalf("crash did not retain uncertain pending: %+v", record)
+ }
+ key, _ := Key(q.Host, record.Authority.Box)
+ if _, _, err = remote.Get(t.Context(), key); err != nil {
+ t.Fatal("remote acquisition missing", err)
+ }
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("crashed acquisition became a permit")
+ }
+ q.Operation = "executionSettle"
+ q.ID = "settle"
+ q.Token = record.Token
+ if _, err = b.Execute(t.Context(), p, q); err == nil {
+ t.Fatal("uncertain unexpired acquisition settled")
+ }
+ clock.advance(TTL + time.Second)
+ if _, err = b.Execute(t.Context(), p, q); err != nil {
+ t.Fatal("expired acquired obligation could not settle", err)
+ }
+}
diff --git a/internal/hostexec/record.go b/internal/hostexec/record.go
index 1ad4f067c..92d096745 100644
--- a/internal/hostexec/record.go
+++ b/internal/hostexec/record.go
@@ -9,6 +9,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/rootparticipation"
)
const TTL = 90 * time.Second
@@ -37,11 +38,7 @@ type Binding struct {
ControllerID string `json:"controller_id"`
}
-type ControllerProcess struct {
- KernelBootID string `json:"kernel_boot_id"`
- PID int `json:"pid"`
- Start string `json:"start"`
-}
+type ControllerProcess = rootparticipation.ControllerProcess
type State struct {
Process ControllerProcess `json:"process,omitzero"`
diff --git a/internal/hostexec/rootbroker.go b/internal/hostexec/rootbroker.go
new file mode 100644
index 000000000..02f3c245b
--- /dev/null
+++ b/internal/hostexec/rootbroker.go
@@ -0,0 +1,573 @@
+package hostexec
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "os"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/rootparticipation"
+ "github.com/carlosframework/platform/internal/rootstate"
+)
+
+type ExecutionRequest = rootparticipation.ExecutionRequest
+type ExecutionAuthority = rootparticipation.ExecutionAuthority
+type ExecutionResponse = rootparticipation.ExecutionResponse
+type ExecutionReplay = rootparticipation.ExecutionReplay
+type ExecutionRecord = rootparticipation.ExecutionRecord
+
+type RootBrokerConfig struct {
+ Transition func(context.Context, ExecutionRecord, ExecutionRequest, ExecutionAuthority) error
+ State rootstate.Store
+ Remote release.Store
+ Clock Clock
+ BootID, Identity string
+ Authority func(context.Context, ExecutionRequest, ControllerProcess) (ExecutionAuthority, error)
+ Alive func(context.Context, ControllerProcess) error
+ Cleanup func(context.Context, ExecutionRecord) error
+ Settle func(context.Context, ExecutionRecord) error
+ Exclude func(context.Context, func() error) error
+}
+
+type RootBroker struct {
+ cfg RootBrokerConfig
+ mu sync.Mutex
+ denied map[string]bool
+ closed bool
+ calls map[string]chan struct{}
+ lifetime map[string]func()
+ live map[string]func()
+}
+
+func NewRootBroker(cfg RootBrokerConfig) (*RootBroker, error) {
+ if cfg.State == nil || cfg.Remote == nil || cfg.Clock == nil || !validID(cfg.BootID) || !validID(cfg.Identity) || cfg.Authority == nil || cfg.Alive == nil || cfg.Cleanup == nil || cfg.Settle == nil || cfg.Exclude == nil {
+ return nil, ErrState
+ }
+ return &RootBroker{cfg: cfg, denied: make(map[string]bool), calls: make(map[string]chan struct{}), lifetime: make(map[string]func()), live: make(map[string]func())}, nil
+}
+
+func executionDigest(q ExecutionRequest, p ControllerProcess) string {
+ b, _ := json.Marshal(struct {
+ Request ExecutionRequest
+ Peer ControllerProcess
+ }{q, p})
+ s := sha256.Sum256(b)
+ return hex.EncodeToString(s[:])
+}
+
+func (b *RootBroker) read(host string) (ExecutionRecord, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ raw, err := b.cfg.State.Read(host + ".json")
+ if err != nil {
+ return ExecutionRecord{}, err
+ }
+ var r ExecutionRecord
+ if decode(raw, &r) != nil || r.Schema != 1 || r.Request.Host != host || !r.Request.Valid() || !validID(r.Token) || r.Authority.Migration == 0 || r.Replay == nil {
+ return ExecutionRecord{}, ErrState
+ }
+ switch r.Status {
+ case "pending", "active", "lost", "settled":
+ default:
+ return ExecutionRecord{}, ErrState
+ }
+ return r, nil
+}
+
+func (b *RootBroker) write(r ExecutionRecord) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ if r.Status == "active" && (b.closed || b.denied[r.Request.Host]) {
+ return ErrLost
+ }
+ if release := b.live[r.Request.Host]; release != nil {
+ release()
+ delete(b.live, r.Request.Host)
+ }
+ raw, err := json.Marshal(r)
+ if err != nil {
+ return err
+ }
+ if err := b.cfg.State.Write(r.Request.Host+".json", raw); err != nil {
+ return err
+ }
+ if r.Status == "active" {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ release, err := b.cfg.State.Lock(ctx, r.Request.Host+".live")
+ if err != nil {
+ return err
+ }
+ b.live[r.Request.Host] = release
+ }
+ return nil
+}
+
+func executionResponse(r ExecutionRecord) ExecutionResponse {
+ q := r.Request
+ return ExecutionResponse{Schema: 1, Host: q.Host, Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, Op: q.Op, Token: r.Token, BootID: r.Peer.KernelBootID, Status: r.Status, SafeUntil: r.SafeUntil, LeaseETag: r.LeaseETag, Expires: r.Expires}
+}
+
+func (b *RootBroker) active(ctx context.Context, r ExecutionRecord) error {
+ b.mu.Lock()
+ denied := b.closed || b.denied[r.Request.Host]
+ b.mu.Unlock()
+ if denied || r.Status != "active" || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID {
+ return ErrLost
+ }
+ _, now, err := b.cfg.Clock.Now()
+ if err != nil {
+ return err
+ }
+ if now >= r.SafeUntil {
+ return ErrLost
+ }
+ return b.cfg.Alive(ctx, r.Peer)
+}
+
+func sameExecution(q ExecutionRequest, r ExecutionRecord) bool {
+ p := r.Request
+ return q.Host == p.Host && q.Incarnation == p.Incarnation && q.Stamp == p.Stamp && q.Session == p.Session && q.Op == p.Op
+}
+
+func (b *RootBroker) deny(ctx context.Context, r ExecutionRecord, cause error) error {
+ b.mu.Lock()
+ raw, readErr := b.cfg.State.Read(r.Request.Host + ".json")
+ var current ExecutionRecord
+ if readErr == nil && decode(raw, &current) == nil && ((current.Token != r.Token && current.Token != r.TransitionFrom) || current.Status == "settled") {
+ b.mu.Unlock()
+ return cause
+ }
+ if readErr == nil && current.Token == r.Token && current.RequestStart > r.RequestStart {
+ r = current
+ }
+ b.denied[r.Request.Host] = true
+ if release := b.live[r.Request.Host]; release != nil {
+ release()
+ delete(b.live, r.Request.Host)
+ }
+ r.Status = "lost"
+ raw, persist := json.Marshal(r)
+ if persist == nil {
+ persist = b.cfg.State.Write(r.Request.Host+".json", raw)
+ }
+ if persist != nil {
+ b.closed = true
+ for host, release := range b.live {
+ release()
+ delete(b.live, host)
+ }
+ }
+ b.mu.Unlock()
+ clean := b.cfg.Cleanup(context.WithoutCancel(ctx), r)
+ return errors.Join(cause, persist, clean)
+}
+
+func (b *RootBroker) Execute(ctx context.Context, peer ControllerProcess, q ExecutionRequest) (ExecutionResponse, error) {
+ if !q.Valid() || peer.PID <= 0 || !validID(peer.Start) || peer.KernelBootID != b.cfg.BootID {
+ return ExecutionResponse{}, ErrState
+ }
+ b.mu.Lock()
+ gate := b.calls[q.Host]
+ if gate == nil {
+ if len(b.calls) >= 4096 {
+ b.mu.Unlock()
+ return ExecutionResponse{}, ErrState
+ }
+ gate = make(chan struct{}, 1)
+ b.calls[q.Host] = gate
+ }
+ b.mu.Unlock()
+ select {
+ case gate <- struct{}{}:
+ case <-ctx.Done():
+ return ExecutionResponse{}, ctx.Err()
+ }
+ defer func() { <-gate }()
+ b.mu.Lock()
+ held := b.lifetime[q.Host] != nil
+ b.mu.Unlock()
+ if !held {
+ unlock, lockErr := b.cfg.State.Lock(ctx, q.Host+".lock")
+ if lockErr != nil {
+ return ExecutionResponse{}, lockErr
+ }
+ b.mu.Lock()
+ b.lifetime[q.Host] = unlock
+ b.mu.Unlock()
+ }
+ if err := ctx.Err(); err != nil {
+ return ExecutionResponse{}, err
+ }
+ // Settlement verifies retained obligations even after current admission ends.
+ var authority ExecutionAuthority
+ var err error
+ if q.Operation != "executionSettle" {
+ authority, err = b.cfg.Authority(ctx, q, peer)
+ if err == nil && (authority.Migration == 0 || (authority.Installation == "" && !authority.ParticipationOnly) || !validID(authority.Box) || !validID(authority.Enrollment) || authority.Account == "" || authority.App == "" || authority.ClaimRevision == "" || authority.RecordRevision == "" || authority.DirectoryRevision == "") {
+ err = ErrState
+ }
+ if err != nil {
+ if prior, readErr := b.read(q.Host); readErr == nil && prior.Peer == peer && (q.Token == prior.Token || q.ID == prior.Request.ID) && prior.Status != "settled" {
+ return ExecutionResponse{}, b.deny(ctx, prior, err)
+ }
+ return ExecutionResponse{}, err
+ }
+ }
+ r, err := b.read(q.Host)
+ missing := errors.Is(err, os.ErrNotExist)
+ if err != nil && !missing {
+ return ExecutionResponse{}, err
+ }
+ digest := executionDigest(q, peer)
+ if !missing {
+ if replay, ok := r.Replay[q.ID]; ok {
+ if replay.Digest != digest {
+ return ExecutionResponse{}, ErrState
+ }
+ if q.Operation == "executionSettle" && replay.Response.Status == "settled" {
+ return replay.Response, nil
+ }
+ if !sameExecution(q, r) || replay.Response.Token != r.Token {
+ return ExecutionResponse{}, ErrLost
+ }
+ if err = b.active(ctx, r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ if r.Peer != peer || r.Authority != authority {
+ return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
+ }
+ return replay.Response, nil
+ }
+ }
+ if q.Operation == "executionBegin" {
+ b.mu.Lock()
+ closed := b.closed
+ b.mu.Unlock()
+ if closed || (!missing && r.Status != "settled") {
+ return ExecutionResponse{}, ErrLost
+ }
+ var token [32]byte
+ if _, err = rand.Read(token[:]); err != nil {
+ return ExecutionResponse{}, err
+ }
+ history := r.Replay
+ if history == nil {
+ history = make(map[string]ExecutionReplay)
+ }
+ b.mu.Lock()
+ delete(b.denied, q.Host)
+ b.mu.Unlock()
+ r = ExecutionRecord{Schema: 1, Request: q, Peer: peer, Authority: authority, Broker: b.cfg.Identity, Token: hex.EncodeToString(token[:]), Status: "pending", Replay: history}
+ wall, start, e := b.cfg.Clock.Now()
+ if e != nil {
+ return ExecutionResponse{}, e
+ }
+ r.PriorSafeUntil = r.SafeUntil
+ r.RequestStart = start
+ r.SafeUntil = start + int64(TTL-Margin)
+ r.Expires = wall.Add(TTL)
+ r.Replay[q.ID] = ExecutionReplay{Digest: digest}
+ if err = b.cfg.Exclude(ctx, func() error { return b.write(r) }); err != nil {
+ return ExecutionResponse{}, err
+ }
+ return b.acquire(ctx, peer, q, r, digest, true)
+ }
+ if missing || r.Peer != peer || q.Token != r.Token {
+ return ExecutionResponse{}, ErrLost
+ }
+ if !sameExecution(q, r) {
+ if q.Operation != "executionRenew" || b.cfg.Transition == nil || q.Host != r.Request.Host || q.Op != r.Request.Op || q.Incarnation != r.Request.Incarnation {
+ return ExecutionResponse{}, ErrLost
+ }
+ if err = b.active(ctx, r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ if err = b.cfg.Transition(ctx, r, q, authority); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ var token [32]byte
+ if _, err = rand.Read(token[:]); err != nil {
+ return ExecutionResponse{}, err
+ }
+ r.Request = q
+ r.TransitionFrom = r.Token
+ r.Token = hex.EncodeToString(token[:])
+ r.Authority = authority
+ return b.acquire(ctx, peer, q, r, digest, false)
+ }
+ if q.Operation == "executionSettle" {
+ return b.settle(ctx, q, r, digest)
+ }
+ if err = b.active(ctx, r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ if authority != r.Authority {
+ if !executionInstallationBinding(r.Authority, authority) {
+ if (q.Operation != "executionRenew" && q.Operation != "executionView") || b.cfg.Transition == nil || b.cfg.Transition(ctx, r, q, authority) != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
+ }
+ r.Authority = authority
+ if q.Operation == "executionView" {
+ if err = b.write(r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ return executionResponse(r), nil
+ }
+ return b.acquire(ctx, peer, q, r, digest, false)
+ }
+ r.Authority = authority
+ if err = b.write(r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ }
+ if q.Operation == "executionView" {
+ return executionResponse(r), nil
+ }
+ if len(r.Replay) >= 4096 {
+ return ExecutionResponse{}, b.deny(ctx, r, ErrState)
+ }
+ return b.acquire(ctx, peer, q, r, digest, false)
+}
+
+func (b *RootBroker) acquire(ctx context.Context, peer ControllerProcess, q ExecutionRequest, r ExecutionRecord, digest string, begin bool) (ExecutionResponse, error) {
+ key, err := Key(q.Host, r.Authority.Box)
+ if err != nil {
+ return ExecutionResponse{}, err
+ }
+ etag := r.LeaseETag
+ if begin {
+ raw, existing, e := b.cfg.Remote.Get(ctx, key)
+ if e == nil {
+ var remote Record
+ wall, _, clockErr := b.cfg.Clock.Now()
+ if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) || existing == "" || clockErr != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, ErrLost)
+ }
+ etag = existing
+ } else if !errors.Is(e, os.ErrNotExist) {
+ return ExecutionResponse{}, b.deny(ctx, r, e)
+ }
+ } else {
+ wall, start, e := b.cfg.Clock.Now()
+ if e != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, e)
+ }
+ r.PriorSafeUntil = r.SafeUntil
+ r.RequestStart = start
+ r.SafeUntil = start + int64(TTL-Margin)
+ r.Expires = wall.Add(TTL)
+ r.Status = "pending"
+ r.Replay[q.ID] = ExecutionReplay{Digest: digest}
+ if e = b.cfg.Exclude(ctx, func() error { return b.write(r) }); e != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, e)
+ }
+ }
+ raw, _ := json.Marshal(Record{Op: q.Op, Expires: r.Expires})
+ next, err := b.cfg.Remote.Put(ctx, key, raw, etag)
+ if err != nil || next == "" {
+ return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
+ }
+ r.LeaseETag = next
+ authority, err := b.cfg.Authority(ctx, q, peer)
+ if err != nil || authority != r.Authority {
+ return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
+ }
+ _, now, err := b.cfg.Clock.Now()
+ if err != nil || now >= r.SafeUntil || ctx.Err() != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, errors.Join(err, ErrLost))
+ }
+ if err = b.cfg.Alive(ctx, peer); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ r.PriorSafeUntil = 0
+ r.Status = "active"
+ out := executionResponse(r)
+ r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out}
+ if err = b.write(r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ if err = b.active(ctx, r); err != nil {
+ return ExecutionResponse{}, b.deny(ctx, r, err)
+ }
+ return out, nil
+}
+
+func (b *RootBroker) settle(ctx context.Context, q ExecutionRequest, r ExecutionRecord, digest string) (ExecutionResponse, error) {
+ if r.Status == "settled" {
+ return ExecutionResponse{}, ErrLost
+ }
+ if err := b.deny(ctx, r, nil); err != nil {
+ return ExecutionResponse{}, err
+ }
+ r.Status = "lost"
+ key, err := Key(q.Host, r.Authority.Box)
+ if err != nil {
+ return ExecutionResponse{}, err
+ }
+ if r.LeaseETag != "" {
+ if err = release.DeleteIfMatch(ctx, b.cfg.Remote, key, r.LeaseETag); err != nil {
+ return ExecutionResponse{}, err
+ }
+ } else {
+ raw, _, readErr := b.cfg.Remote.Get(ctx, key)
+ wall, _, clockErr := b.cfg.Clock.Now()
+ if clockErr != nil {
+ return ExecutionResponse{}, clockErr
+ }
+ if readErr == nil {
+ var remote Record
+ if decode(raw, &remote) != nil || remote.Expires.IsZero() || !wall.After(remote.Expires) {
+ return ExecutionResponse{}, ErrLost
+ }
+ } else if !errors.Is(readErr, os.ErrNotExist) {
+ return ExecutionResponse{}, readErr
+ }
+ if !wall.After(r.Expires) {
+ return ExecutionResponse{}, ErrLost
+ }
+ }
+ if err = b.cfg.Settle(ctx, r); err != nil {
+ return ExecutionResponse{}, err
+ }
+ r.Status = "settled"
+ out := executionResponse(r)
+ r.Replay[q.ID] = ExecutionReplay{Digest: digest, Response: out}
+ if err = b.write(r); err != nil {
+ return ExecutionResponse{}, err
+ }
+ return out, nil
+}
+
+func (b *RootBroker) Monitor(ctx context.Context) error {
+ names, err := b.cfg.State.Names()
+ if err != nil {
+ return err
+ }
+ var errs []error
+ for _, name := range names {
+ if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
+ continue
+ }
+ host := strings.TrimSuffix(name, ".json")
+ b.mu.Lock()
+ owned := b.lifetime[host] != nil
+ b.mu.Unlock()
+ if !owned {
+ continue
+ }
+ r, err := b.read(host)
+ if err != nil {
+ errs = append(errs, err)
+ continue
+ }
+ if r.Status == "settled" || r.Status == "lost" {
+ continue
+ }
+ _, now, clockErr := b.cfg.Clock.Now()
+ bounded, cancel := context.WithTimeout(ctx, time.Second)
+ aliveErr := b.cfg.Alive(bounded, r.Peer)
+ cancel()
+ cutoff := r.SafeUntil
+ if r.Status == "pending" && r.PriorSafeUntil > 0 && r.PriorSafeUntil < cutoff {
+ cutoff = r.PriorSafeUntil
+ }
+ if clockErr != nil || aliveErr != nil || now >= cutoff || r.Broker != b.cfg.Identity || r.Peer.KernelBootID != b.cfg.BootID {
+ errs = append(errs, b.deny(ctx, r, ErrLost))
+ }
+ }
+ return errors.Join(errs...)
+}
+
+func executionInstallationBinding(old, next ExecutionAuthority) bool {
+ if !old.ParticipationOnly || (old.Installation != "" && (old.Installation != next.Installation || next.ParticipationOnly)) || next.Installation == "" {
+ return false
+ }
+ old.Installation = next.Installation
+ old.ParticipationOnly = next.ParticipationOnly
+ return old == next
+}
+
+func (b *RootBroker) Close(ctx context.Context) error {
+ b.mu.Lock()
+ b.closed = true
+ b.mu.Unlock()
+ names, err := b.cfg.State.Names()
+ if err != nil {
+ return err
+ }
+ var errs []error
+ for _, name := range names {
+ if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
+ continue
+ }
+ r, e := b.read(strings.TrimSuffix(name, ".json"))
+ if e != nil {
+ errs = append(errs, e)
+ continue
+ }
+ if r.Broker == b.cfg.Identity && r.Status != "settled" {
+ errs = append(errs, b.deny(ctx, r, nil))
+ }
+ }
+ b.mu.Lock()
+ for host, unlock := range b.lifetime {
+ unlock()
+ delete(b.lifetime, host)
+ }
+ b.mu.Unlock()
+ return errors.Join(errs...)
+}
+
+func (b *RootBroker) Recover(ctx context.Context) error {
+ names, err := b.cfg.State.Names()
+ if err != nil {
+ return err
+ }
+ var errs []error
+ for _, name := range names {
+ if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") {
+ continue
+ }
+ host := strings.TrimSuffix(name, ".json")
+ b.mu.Lock()
+ owned := b.lifetime[host] != nil
+ b.mu.Unlock()
+ if !owned {
+ unlock, e := b.cfg.State.Lock(ctx, host+".lock")
+ if e != nil {
+ return e
+ }
+ b.mu.Lock()
+ b.lifetime[host] = unlock
+ b.mu.Unlock()
+ }
+ r, e := b.read(host)
+ if e != nil {
+ errs = append(errs, e)
+ continue
+ }
+ if r.Status != "settled" {
+ errs = append(errs, b.deny(ctx, r, nil))
+ }
+ }
+ return errors.Join(errs...)
+}
+
+func (b *RootBroker) Run(ctx context.Context) {
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-b.cfg.Clock.After(100 * time.Millisecond):
+ }
+ _ = b.Monitor(ctx)
+ }
+}
diff --git a/internal/hostexec/rootclient.go b/internal/hostexec/rootclient.go
new file mode 100644
index 000000000..a4ff8c5ae
--- /dev/null
+++ b/internal/hostexec/rootclient.go
@@ -0,0 +1,258 @@
+package hostexec
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "errors"
+ "sync"
+ "time"
+)
+
+type RootClientConfig struct {
+ Binding Binding
+ Clock Clock
+ Observe func(context.Context) (ExecutionRequest, error)
+ Call func(context.Context, ExecutionRequest) (ExecutionResponse, error)
+ CloseAdmission func()
+ Cleanup func(context.Context) error
+}
+
+type RootClient struct {
+ cfg RootClientConfig
+ mu sync.Mutex
+ gate chan struct{}
+ request ExecutionRequest
+ state State
+ err error
+ next int64
+ cancel context.CancelFunc
+ done chan struct{}
+ watchDone chan struct{}
+ autoDone chan struct{}
+ closeMu sync.Mutex
+ cleaned, settled bool
+ settleID string
+}
+
+func requestIdentity() (string, error) {
+ var raw [24]byte
+ if _, err := rand.Read(raw[:]); err != nil {
+ return "", err
+ }
+ return hex.EncodeToString(raw[:]), nil
+}
+
+func OpenRootClient(ctx context.Context, cfg RootClientConfig) (*RootClient, error) {
+ if cfg.Clock == nil || cfg.Call == nil || cfg.Observe == nil || cfg.CloseAdmission == nil || cfg.Cleanup == nil {
+ return nil, ErrState
+ }
+ worker, cancel := context.WithCancel(ctx)
+ c := &RootClient{cfg: cfg, gate: make(chan struct{}, 1), cancel: cancel, done: make(chan struct{}), watchDone: make(chan struct{}), autoDone: make(chan struct{})}
+ if err := c.refresh(worker, true); err != nil {
+ cancel()
+ close(c.done)
+ close(c.watchDone)
+ close(c.autoDone)
+ return c, err
+ }
+ go func() {
+ defer close(c.watchDone)
+ for {
+ select {
+ case <-worker.Done():
+ return
+ case <-cfg.Clock.After(100 * time.Millisecond):
+ if _, err := c.Ready(worker); err != nil {
+ c.cancel()
+ return
+ }
+ }
+ }
+ }()
+ go func() {
+ defer close(c.done)
+ for {
+ select {
+ case <-worker.Done():
+ c.lose(worker.Err())
+ return
+ case <-cfg.Clock.After(Renewal):
+ if err := c.Refresh(worker); err != nil {
+ c.lose(err)
+ return
+ }
+ }
+ }
+ }()
+ go func() {
+ defer close(c.autoDone)
+ <-worker.Done()
+ cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+ defer cancel()
+ _ = c.close(cleanup)
+ }()
+ return c, nil
+}
+
+func (c *RootClient) lose(err error) {
+ c.mu.Lock()
+ if c.err == nil {
+ c.err = errors.Join(ErrLost, err)
+ }
+ c.mu.Unlock()
+ c.cancel()
+ c.cfg.CloseAdmission()
+}
+
+func (c *RootClient) Refresh(ctx context.Context) error { return c.refresh(ctx, false) }
+func (c *RootClient) refresh(ctx context.Context, begin bool) error {
+ select {
+ case c.gate <- struct{}{}:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ defer func() { <-c.gate }()
+ c.mu.Lock()
+ prior, next, previousErr := c.state, c.next, c.err
+ c.mu.Unlock()
+ if previousErr != nil {
+ return previousErr
+ }
+ _, start, err := c.cfg.Clock.Now()
+ if err != nil {
+ return err
+ }
+ if !begin && start >= prior.SafeUntil {
+ c.lose(ErrLost)
+ return ErrLost
+ }
+ q, err := c.cfg.Observe(ctx)
+ if err != nil {
+ c.lose(err)
+ return err
+ }
+ if q.Host != c.cfg.Binding.Host || q.Incarnation != c.cfg.Binding.Incarnation || q.Op != c.cfg.Binding.Op {
+ c.lose(ErrLost)
+ return ErrLost
+ }
+ q.Schema = 1
+ q.ID, err = requestIdentity()
+ if err != nil {
+ return err
+ }
+ q.Operation = "executionView"
+ q.Token = prior.ControllerID
+ if begin {
+ q.Operation = "executionBegin"
+ } else if start >= next {
+ q.Operation = "executionRenew"
+ }
+ c.mu.Lock()
+ old := c.request
+ c.mu.Unlock()
+ if !begin && (old.Stamp != q.Stamp || old.Session != q.Session) {
+ q.Operation = "executionRenew"
+ }
+ out, err := c.cfg.Call(ctx, q)
+ if err != nil {
+ c.lose(err)
+ return err
+ }
+ _, now, clockErr := c.cfg.Clock.Now()
+ if clockErr != nil || 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 != c.cfg.Binding.BootID || out.Token == "" || out.Status != "active" || out.SafeUntil <= now || out.LeaseETag == "" || out.Expires.IsZero() {
+ c.lose(ErrLost)
+ return ErrLost
+ }
+ binding := c.cfg.Binding
+ binding.ControllerID = out.Token
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if c.err != nil {
+ return c.err
+ }
+ c.state = State{V: 1, Binding: binding, State: out.Status, SafeUntil: out.SafeUntil, Expires: out.Expires, LeaseETag: out.LeaseETag}
+ c.request = q
+ if q.Operation != "executionView" {
+ c.next = start + int64(Renewal)
+ }
+ return nil
+}
+
+func (c *RootClient) Ready(ctx context.Context) (State, error) {
+ if ctx.Err() != nil {
+ return State{}, ctx.Err()
+ }
+ c.mu.Lock()
+ state, err := c.state, c.err
+ c.mu.Unlock()
+ if err != nil {
+ return state, err
+ }
+ _, now, err := c.cfg.Clock.Now()
+ if err != nil || state.State != "active" || now >= state.SafeUntil {
+ c.lose(ErrLost)
+ return state, ErrLost
+ }
+ return state, nil
+}
+
+func (c *RootClient) Close(ctx context.Context) error {
+ err := c.close(ctx)
+ select {
+ case <-c.autoDone:
+ return err
+ case <-ctx.Done():
+ return errors.Join(err, ctx.Err())
+ }
+}
+
+func (c *RootClient) close(ctx context.Context) error {
+ c.closeMu.Lock()
+ defer c.closeMu.Unlock()
+ if c.settled {
+ return nil
+ }
+ c.lose(ErrLost)
+ c.cancel()
+ select {
+ case <-c.done:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ select {
+ case <-c.watchDone:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ if !c.cleaned {
+ if err := c.cfg.Cleanup(ctx); err != nil {
+ return err
+ }
+ c.cleaned = true
+ }
+ c.mu.Lock()
+ q, token := c.request, c.state.ControllerID
+ c.mu.Unlock()
+ if token == "" {
+ return ErrLost
+ }
+ q.Token, q.Operation = token, "executionSettle"
+ var err error
+ if c.settleID == "" {
+ c.settleID, err = requestIdentity()
+ if err != nil {
+ return err
+ }
+ }
+ q.ID = c.settleID
+ settle, err := c.cfg.Call(ctx, q)
+ if err != nil {
+ return err
+ }
+ if settle.Status != "settled" || settle.Token != token {
+ return ErrLost
+ }
+ c.settled = true
+ return nil
+}
diff --git a/internal/hostexec/rootclient_test.go b/internal/hostexec/rootclient_test.go
new file mode 100644
index 000000000..b3f9178af
--- /dev/null
+++ b/internal/hostexec/rootclient_test.go
@@ -0,0 +1,55 @@
+//go:build linux
+
+package hostexec
+
+import (
+ "context"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func TestPlan14ExecutionClientHasOneRenewalOwner(t *testing.T) {
+ b, _, remote, clock, q, peer := brokerFixture(t)
+ var calls, cleanup atomic.Int32
+ 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, q ExecutionRequest) (ExecutionResponse, error) {
+ calls.Add(1)
+ return b.Execute(ctx, peer, q)
+ },
+ CloseAdmission: func() {}, Cleanup: func(context.Context) error { cleanup.Add(1); return nil },
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = c.Close(context.Background()) })
+ s, err := c.Ready(t.Context())
+ if err != nil || s.ControllerID == "" || s.LeaseETag == "" {
+ t.Fatalf("ready %+v %v", s, err)
+ }
+ if err = c.Refresh(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if remote.calls != 1 {
+ t.Fatalf("view renewed: %d writes", remote.calls)
+ }
+ clock.advance(20 * time.Second)
+ if err = c.Refresh(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if remote.calls != 2 || calls.Load() < 3 {
+ t.Fatalf("owners: writes=%d calls=%d", remote.calls, calls.Load())
+ }
+ clock.advance(TTL)
+ if _, err = c.Ready(t.Context()); err == nil {
+ t.Fatal("client admitted expired root response")
+ }
+ if err = c.Close(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if cleanup.Load() != 1 {
+ t.Fatalf("cleanup=%d", cleanup.Load())
+ }
+}
diff --git a/internal/hostexec/rootprocess_linux.go b/internal/hostexec/rootprocess_linux.go
new file mode 100644
index 000000000..882429745
--- /dev/null
+++ b/internal/hostexec/rootprocess_linux.go
@@ -0,0 +1,23 @@
+//go:build linux
+
+package hostexec
+
+import "context"
+
+func ControllerAlive(ctx context.Context, p ControllerProcess) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ boot, err := BootID()
+ if err != nil {
+ return err
+ }
+ start, err := controllerProcessStart(p.PID)
+ if err != nil {
+ return err
+ }
+ if boot != p.KernelBootID || start != p.Start {
+ return ErrLost
+ }
+ return nil
+}
diff --git a/internal/hostexec/rootprocess_other.go b/internal/hostexec/rootprocess_other.go
new file mode 100644
index 000000000..39f834df3
--- /dev/null
+++ b/internal/hostexec/rootprocess_other.go
@@ -0,0 +1,7 @@
+//go:build !linux
+
+package hostexec
+
+import "context"
+
+func ControllerAlive(context.Context, ControllerProcess) error { return ErrUnsupported }
diff --git a/internal/hostexec/rootread.go b/internal/hostexec/rootread.go
new file mode 100644
index 000000000..0af38a207
--- /dev/null
+++ b/internal/hostexec/rootread.go
@@ -0,0 +1,77 @@
+package hostexec
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/rootparticipation"
+ "github.com/carlosframework/platform/internal/rootstate"
+)
+
+func ReadRoot(ctx context.Context, store rootstate.Store, clock Clock, q ExecutionRequest, serving bool) (ExecutionRecord, error) {
+ var r ExecutionRecord
+ if store == nil || clock == nil {
+ return r, ErrState
+ }
+ ledger := rootparticipation.Ledger{State: store}
+ census, err := ledger.Census()
+ if err != nil || census.NeverEnrolled {
+ return r, errors.Join(err, ErrState)
+ }
+ raw, err := store.Read(q.Host + ".json")
+ if serving {
+ if errors.Is(err, os.ErrNotExist) {
+ return r, nil
+ }
+ if err != nil {
+ return r, err
+ }
+ if rootparticipation.Decode(raw, &r) != nil || r.Schema != 1 || !r.Request.Valid() || r.Token == "" || r.Replay == nil || r.Request.Host != q.Host || r.Status != "settled" || r.Authority.Migration != census.Migration || r.Authority.Box != census.Box || r.Authority.Enrollment != census.Enrollment {
+ return r, ErrLost
+ }
+ return r, nil
+ }
+ if err != nil {
+ return r, err
+ }
+ if rootparticipation.Decode(raw, &r) != nil || r.Schema != 1 || !r.Request.Valid() || !sameExecution(q, r) || r.Status != "active" || r.Authority.ParticipationOnly || r.Authority.Installation == "" || r.Authority.Migration != census.Migration || r.Authority.Box != census.Box || r.Authority.Enrollment != census.Enrollment || r.Token == "" {
+ return r, ErrLost
+ }
+ boot, err := BootID()
+ if err != nil || r.Peer.KernelBootID != boot {
+ return r, ErrLost
+ }
+ _, now, err := clock.Now()
+ if err != nil || now >= r.SafeUntil {
+ return r, ErrLost
+ }
+ if err = ControllerAlive(ctx, r.Peer); err != nil {
+ return r, err
+ }
+ // The stable root-owned live lock also carries an in-memory revocation.
+ probe, cancel := context.WithTimeout(ctx, 10*time.Millisecond)
+ unlock, err := store.Lock(probe, q.Host+".live")
+ cancel()
+ if err == nil {
+ unlock()
+ return r, ErrLost
+ }
+ if ctx.Err() != nil {
+ return r, ctx.Err()
+ }
+ if !errors.Is(err, context.DeadlineExceeded) {
+ return r, err
+ }
+ again, err := store.Read(q.Host + ".json")
+ if err != nil || !bytes.Equal(raw, again) {
+ return r, ErrLost
+ }
+ _, now, err = clock.Now()
+ if err != nil || now >= r.SafeUntil {
+ return r, ErrLost
+ }
+ return r, nil
+}
diff --git a/internal/hostguard/root.go b/internal/hostguard/root.go
index d72d9973a..579ea9da4 100644
--- a/internal/hostguard/root.go
+++ b/internal/hostguard/root.go
@@ -235,3 +235,17 @@ func (a *RootAuthority) PendingInstallations() ([][]byte, error) {
}
return out, nil
}
+
+func (a *RootAuthority) PendingInstallation(ctx context.Context, host string) (RootInstallation, []byte, error) {
+ if err := ctx.Err(); err != nil {
+ return RootInstallation{}, nil, err
+ }
+ r, err := a.read(host)
+ if err != nil {
+ return RootInstallation{}, nil, err
+ }
+ if r.pendingInstallation == nil {
+ return RootInstallation{}, nil, ErrNoRecord
+ }
+ return *r.pendingInstallation, append([]byte(nil), r.pendingInventory...), nil
+}
diff --git a/internal/hostrelocate/plan14_participation_executor_test.go b/internal/hostrelocate/plan14_participation_executor_test.go
new file mode 100644
index 000000000..6488426f5
--- /dev/null
+++ b/internal/hostrelocate/plan14_participation_executor_test.go
@@ -0,0 +1,81 @@
+package hostrelocate
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostexec"
+)
+
+type plan14TargetAcknowledgement struct{ *fakeBox }
+
+func (b plan14TargetAcknowledgement) Acknowledge(ctx context.Context, o hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) {
+ if hostexec.PreinstallationParticipant(o.Claim, b.box, enrollment(b.box)) != nil {
+ return hostclaim.ExecutionAcknowledgement{}, ErrRefused
+ }
+ return b.fakeBox.Acknowledge(ctx, o)
+}
+
+func TestPlan14PreinstallationTargetAcknowledgement(t *testing.T) {
+ for _, kind := range []string{"drain", "failover", "converted-landing"} {
+ t.Run(kind, func(t *testing.T) {
+ f := newFixture(t)
+ target, executor, next := "dst", "src", hostclaim.PhaseReleasing
+ if kind == "converted-landing" {
+ f.convertedLanding()
+ f.worker(nil)
+ target, executor, next = "spa", "none", hostclaim.PhasePinning
+ } else {
+ reason := hostclaim.ReasonDrain
+ if kind == "failover" {
+ f.terminate("src")
+ reason, executor, next = hostclaim.ReasonFailover, "none", hostclaim.PhasePinning
+ }
+ if _, err := f.w.Start(t.Context(), host, Request{Reason: reason, Source: "src", Target: target}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if kind == "drain" {
+ f.ack("src")
+ }
+ before := f.claim()
+ if before.Claim.Op.Phase != hostclaim.PhaseStarting || before.Claim.Op.Exec != executor || before.Claim.Op.Intent["target_box"] != target {
+ t.Fatalf("unexpected starting authority: %+v", before.Claim.Op)
+ }
+ box := &fakeBox{f: f, box: target}
+ p := Participant{Claims: f.s, Box: target, Enrollment: enrollment(target), Now: func() time.Time { return f.now }, Effects: plan14TargetAcknowledgement{box}}
+ for _, attack := range []string{"wrong phase", "unnamed box"} {
+ bad := *before.Claim
+ bad.Op = before.Claim.Op
+ bad.Op.Intent = map[string]string{}
+ for k, v := range before.Claim.Op.Intent {
+ bad.Op.Intent[k] = v
+ }
+ if attack == "wrong phase" {
+ bad.Op.Phase = hostclaim.PhaseRelocatingProbing
+ bad.Op.Exec = target
+ } else {
+ bad.Op.Intent["target_box"] = "elsewhere"
+ }
+ if _, err := (plan14TargetAcknowledgement{box}).Acknowledge(t.Context(), hostclaim.Observed{Host: host, Claim: &bad}); !errors.Is(err, ErrRefused) {
+ t.Fatalf("%s: %v", attack, err)
+ }
+ }
+ if after := f.claim(); after.ETag != before.ETag || box.opened != 0 || len(box.effects) != 0 || len(f.leases) != 0 {
+ t.Fatal("refusal had effects")
+ }
+ if _, err := p.Step(t.Context(), host); err != nil {
+ t.Fatal(err)
+ }
+ if o := f.worker(nil); o.Claim.Op.Phase != next || o.Claim.Op.Exec != executor {
+ t.Fatalf("named-target acknowledgement did not advance without executor handoff: %+v", o.Claim.Op)
+ }
+ if box.opened != 1 || len(box.effects) != 0 {
+ t.Fatal("acknowledgement performed installation effects")
+ }
+ })
+ }
+}
diff --git a/internal/instancectl/server.go b/internal/instancectl/server.go
index cf5531d75..5df53ae72 100644
--- a/internal/instancectl/server.go
+++ b/internal/instancectl/server.go
@@ -283,6 +283,7 @@ func (s *Server) Handler() http.Handler {
mux.Handle("POST /claim/handover", s.ClaimStop)
mux.Handle("POST /claim/recovery-rebind", s.ClaimStop)
mux.Handle("POST /claim/journal", s.ClaimStop)
+ mux.Handle("POST /claim/execution", s.ClaimStop)
mux.Handle("POST /claim/writers-stopped", s.ClaimStop)
mux.Handle("POST /claim/launch", s.ClaimStop)
}
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index a123fbfa5..3a55ade04 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -733,6 +733,8 @@ var plan9B8CensusInventory = []censusEntry{
// Region HA stage B: installs root's relocating attempt reader behind CARLOS_REGION_HA; names the instances root for the launch, writes nothing.
{"internal/host/claimlaunch_production.go", "claimLaunchFor", "key-site", "lit:instances", "76cfc428c7385d1278a5d3842396a6f7b343f03547fdbfc51088af262f3980ef"},
{"internal/host/claimretire_transport.go", "retirementAcknowledgement", "nonrecord", "sel:Put", "5c09795fbe9b952436d16bcd58e846ac27e9dbcf959fcf1cd6c432af0f3ee416"},
+ // D1 compares the retained registration's record key to the fresh claim; no record write.
+ {"internal/host/rootexecution_authority.go", "ClaimRootHTTP.executionAuthority", "key-site", "key:RecordKey", "b4dd1af9510ae7a0c71ae849709ed5ed751178aa99067c61b06e3032c2f6f97a"},
{"internal/host/errship.go", "errshipFlush", "nonrecord", "sel:Put", "82831c08563eab44de07c1c69752555eafe60c70dbf9521d9a4711bc5c104a6b"},
{"internal/host/fleetbind.go", "WriteFleetBinding", "nonrecord", "sel:Put", "44ab984c2c9b30abcc938132118f61a6922ccc0cd40ed2b6cb61c7654aa0a1c3"},
{"internal/host/logship.go", "logshipFlush", "nonrecord", "sel:Put", "a2fa594c1206028af170a2d0cbedef55ba00a2f1c698a6853e1214cc3b172dd9"},
@@ -792,6 +794,8 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/hostclaim/store.go", "Update", "nonrecord", "sel:Put", "083ddd3e076a07db2b5fca70a9b56fa022782deafb9ad56c77f8f32515633906"},
{"internal/hostevidence/signature.go", "immutable", "nonrecord", "sel:Put", "54fb2bdc230c2cb4193f3b808004b63bcfd092266846fc0c03e61617c3ad0ab5"},
{"internal/hostexec/controller.go", "Controller.run", "nonrecord", "sel:Put", "03df11b375761730700ce30c0db05c57f0828629eda0a5de126ea1a2f2f7501c"},
+ // D1: conditional execution lease write under control/hosts/<host>/exec/<box>.json; no deployment record key.
+ {"internal/hostexec/rootbroker.go", "RootBroker.acquire", "nonrecord", "sel:Put", "8999554ba85afcbee18a0a4f6691208cefb3d56f9ea24961b6f3c74c0183551e"},
{"internal/hostfence/termination.go", "TerminationWorker.persist", "nonrecord", "sel:Put", "47758114adcc3c493a9d1ead0fa7087ad0d1e49595c891bab4dfb51c9058a8a1"},
{"internal/hostfiles/contract.go", "CopyManifest", "nonrecord", "sel:Put", "9754b113ca2cad5e267402a44ac25de778e332d0c87027767f9a7daa4937d9fd"},
{"internal/hostfiles/publisher.go", "Publisher.put", "nonrecord", "sel:Put", "4bf1d0c4ab995f4e5afa8d7675cbd00446da006a82a659764f8013c011f9efaa"},
diff --git a/internal/provision/testdata/plan14-root-reads-merge.tsv b/internal/provision/testdata/plan14-root-reads-merge.tsv
index b065ce516..305d4a417 100644
--- a/internal/provision/testdata/plan14-root-reads-merge.tsv
+++ b/internal/provision/testdata/plan14-root-reads-merge.tsv
@@ -251,14 +251,14 @@ 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 6b33f71d771754381da99bc18a03155e05e8dc668e90a7ad6176fac41dbfa3d0
+@coverage internal/host 4b6fccbf2d3d6c0c9696563d4deefebfbe0d35b24f75bfca493acd6f5185c272
@coverage internal/hostbox afb08c287eaccb43adde1c488bd122482711ce4e29ecb675c9428f27130c3d93
@coverage internal/hostclaim feaf701db4b49ce92733b9d6c1fa1c0ae8efc6e13439ce201fca1a9e24dcf927
@coverage internal/hostfiles 554285c123758375c13388b2fa42f654819392dac0b027816a713548ec17dcc6
-@coverage internal/hostguard d950f2adb478d34938180109a4eaffb549f5eaeef87cef15797f22e68c394950
+@coverage internal/hostguard 9a9f6ce8dc7dd8c6fec2f52e1e928e0ae133eebdcc4b6417824d150036c60f0e
@coverage internal/hostops f1607e7a75af365c5a8977405e7df7b970614b4423593b9bd69805636a03a5a2
@coverage internal/hostrelocate b8973c260be53365767539c9e355aebf15aeae063ee8f5d4eb248d2bba5c24cc
-@coverage internal/instancectl d1c0d279937025549fbf7062e2c783437f6ac113a60e522252fa801c41ba9f91
+@coverage internal/instancectl de726ae384f2f5ad82f9ae90ac8a7c25b2db894f310e976ae9695b8704f3d1e1
@coverage internal/instancefs 41940b734d708c6a916ba001c7b3a1c903e84ea3421614071ae76e896f3c0fcc
@coverage internal/instancekey cd02b5b212e4ffbd56d36d1a68af3e6ee0a51d7155cebd1a82f9f52d5da8b118
@coverage internal/moves 5218cf9bd31f6347ae49a595a32455f441b1718ecfa0a9559f99036d32f56498
@@ -268,3 +268,18 @@ 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 1e91f8b6ffbe9f6d4512fc5d6ecfb0c4c5a438c22d63e5cca50602918070f779
+@coverage internal/rootparticipation c64cb77d4872be4d9f4e977bbbb9c2a2bd8c860cf4c313585210b3da07ad9a2d
+internal/host/rootexecution_authority.go ClaimRootHTTP.executionAuthority authority D D1 reads the fresh claim, deployment record and root inventory; candidate admission only. Production broker construction is pending. file:Read@553 b4dd1af9510ae7a0c71ae849709ed5ed751178aa99067c61b06e3032c2f6f97a
+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/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 generate root tokens and retain conditional-write, request replay and transition evidence under root lifetime locks. file:Read@2794,file:Read@4212 d8215f5c831bd08575a13d681c5501d8a2a322ea2c25841485c45f628c79d582
+internal/hostexec/rootbroker.go RootBroker.deny authority D D2 compares the current root token before durable loss and live-lock revocation; stale monitors cannot revoke successor operations. file:Read@117 d6fdcdf618287dd698ee76a3141e29088e990b34512ea5c05e8d7f60cf08ecc1
+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/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
+internal/rootparticipation/ledger.go Ledger.EndUpdate authority D D3 removes only the exact root update identity under transaction exclusion. file:Read@222 06d084eda5689aa6a85717cfcd715eb1b856e46824635236a91672f9dffd82f0
+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
diff --git a/internal/rootparticipation/ledger.go b/internal/rootparticipation/ledger.go
new file mode 100644
index 000000000..96783fa03
--- /dev/null
+++ b/internal/rootparticipation/ledger.go
@@ -0,0 +1,190 @@
+package rootparticipation
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "os"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/rootstate"
+)
+
+const Namespace = "/var/lib/carlos-claims/authority/exec"
+
+var ErrUnavailable = errors.New("root participation unavailable")
+
+type Census struct {
+ Schema int
+ Box, Enrollment string
+ Migration, Revision, LaunchRevision uint64
+ Complete, NeverEnrolled bool
+ Pending string
+}
+
+type Ledger struct {
+ State rootstate.Store
+ Box, Enrollment string
+ VerifyLaunches func(context.Context, uint64) error
+}
+
+func Decode(raw []byte, v any) error {
+ if len(raw) > rootstate.MaxRecord {
+ return ErrUnavailable
+ }
+ d := json.NewDecoder(bytes.NewReader(raw))
+ d.DisallowUnknownFields()
+ if err := d.Decode(v); err != nil {
+ return err
+ }
+ if d.Decode(new(any)) != io.EOF {
+ return ErrUnavailable
+ }
+ return nil
+}
+func (l Ledger) Census() (Census, error) {
+ var c Census
+ if l.State == nil {
+ return c, ErrUnavailable
+ }
+ raw, e := l.State.Read("census.json")
+ if e != nil {
+ return c, e
+ }
+ if Decode(raw, &c) != nil || c.Schema != 1 || c.Revision == 0 || c.Box == "" || c.Enrollment == "" || !c.Complete || c.Pending != "" || (c.Migration == 0 && !c.NeverEnrolled) || c.LaunchRevision == 0 || (l.Box != "" && c.Box != l.Box) || (l.Enrollment != "" && c.Enrollment != l.Enrollment) {
+ return c, ErrUnavailable
+ }
+ return c, nil
+}
+func (l Ledger) writeCensus(c Census) error {
+ raw, e := json.Marshal(c)
+ if e != nil {
+ return e
+ }
+ return l.State.Write("census.json", raw)
+}
+func (l Ledger) Acquire(ctx context.Context, commit func() error) error {
+ if l.State == nil || commit == nil {
+ return ErrUnavailable
+ }
+ unlock, e := l.State.Lock(ctx, "transaction.lock")
+ if e != nil {
+ return e
+ }
+ defer unlock()
+ c, e := l.Census()
+ if e != nil {
+ return e
+ }
+ if c.NeverEnrolled {
+ return ErrUnavailable
+ }
+ if _, e = l.State.Read("update.json"); !errors.Is(e, os.ErrNotExist) {
+ return ErrUnavailable
+ }
+ c.Pending = "acquisition"
+ c.Revision++
+ if e = l.writeCensus(c); e != nil {
+ return e
+ }
+ if e = commit(); e != nil {
+ return e
+ }
+ c.Pending = ""
+ return l.writeCensus(c)
+}
+func (l Ledger) idle(ctx context.Context, c Census) error {
+ if l.VerifyLaunches == nil {
+ return ErrUnavailable
+ }
+ names, e := l.State.Names()
+ if e != nil {
+ return e
+ }
+ for _, name := range names {
+ if name == "census.json" || name == "update.json" || strings.HasSuffix(name, ".lock") || strings.HasSuffix(name, ".live") {
+ continue
+ }
+ if !strings.HasSuffix(name, ".json") {
+ return ErrUnavailable
+ }
+ raw, e := l.State.Read(name)
+ if e != nil {
+ return e
+ }
+ var r ExecutionRecord
+ if Decode(raw, &r) != nil || r.Schema != 1 || !r.Request.Valid() || r.Request.Host+".json" != name || r.Authority.Box != c.Box || r.Authority.Enrollment != c.Enrollment || r.Authority.Migration != c.Migration || r.Status != "settled" || r.Token == "" || r.Replay == nil {
+ return ErrUnavailable
+ }
+ }
+ return l.VerifyLaunches(ctx, c.LaunchRevision)
+}
+func (l Ledger) Idle(ctx context.Context) error {
+ if l.State == nil {
+ return ErrUnavailable
+ }
+ unlock, e := l.State.Lock(ctx, "transaction.lock")
+ if e != nil {
+ return e
+ }
+ defer unlock()
+ c, e := l.Census()
+ if e != nil {
+ return e
+ }
+ return l.idle(ctx, c)
+}
+func (l Ledger) BeginUpdate(ctx context.Context, id string) error {
+ if l.State == nil || !validID(id) {
+ return ErrUnavailable
+ }
+ unlock, e := l.State.Lock(ctx, "transaction.lock")
+ if e != nil {
+ return e
+ }
+ defer unlock()
+ c, e := l.Census()
+ if e != nil {
+ return e
+ }
+ if _, e = l.State.Read("update.json"); !errors.Is(e, os.ErrNotExist) {
+ return ErrUnavailable
+ }
+ raw, _ := json.Marshal(struct {
+ Schema int
+ ID string
+ Revision uint64
+ }{1, id, c.Revision})
+ if e = l.State.Write("update.json", raw); e != nil {
+ return e
+ }
+ if e = l.idle(ctx, c); e != nil {
+ return errors.Join(e, l.State.Remove("update.json"))
+ }
+ return nil
+}
+func (l Ledger) EndUpdate(ctx context.Context, id string) error {
+ if l.State == nil {
+ return ErrUnavailable
+ }
+ unlock, e := l.State.Lock(ctx, "transaction.lock")
+ if e != nil {
+ return e
+ }
+ defer unlock()
+ raw, e := l.State.Read("update.json")
+ if e != nil {
+ return e
+ }
+ var u struct {
+ Schema int
+ ID string
+ Revision uint64
+ }
+ if Decode(raw, &u) != nil || u.Schema != 1 || u.ID != id {
+ return ErrUnavailable
+ }
+ return l.State.Remove("update.json")
+}
diff --git a/internal/rootparticipation/ledger_test.go b/internal/rootparticipation/ledger_test.go
new file mode 100644
index 000000000..d1cf8b3b1
--- /dev/null
+++ b/internal/rootparticipation/ledger_test.go
@@ -0,0 +1,102 @@
+package rootparticipation
+
+import (
+ "context"
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+)
+
+type testDisk struct {
+ dir string
+ mu sync.Mutex
+}
+
+func (d *testDisk) Read(n string) ([]byte, error) { return os.ReadFile(filepath.Join(d.dir, n)) }
+func (d *testDisk) Write(n string, b []byte) error {
+ return os.WriteFile(filepath.Join(d.dir, n), b, 0600)
+}
+func (d *testDisk) Remove(n string) error { return os.Remove(filepath.Join(d.dir, n)) }
+func (d *testDisk) Names() ([]string, error) {
+ es, e := os.ReadDir(d.dir)
+ var out []string
+ for _, v := range es {
+ out = append(out, v.Name())
+ }
+ return out, e
+}
+func (d *testDisk) Close() error { return nil }
+func (d *testDisk) Lock(context.Context, string) (func(), error) {
+ d.mu.Lock()
+ return d.mu.Unlock, nil
+}
+func censusFixture(t *testing.T) Ledger {
+ t.Helper()
+ s := &testDisk{dir: t.TempDir()}
+ b, _ := json.Marshal(Census{Schema: 1, Box: "box", Enrollment: "en", Migration: 1, Revision: 1, Complete: true, LaunchRevision: 1})
+ if err := s.Write("census.json", b); err != nil {
+ t.Fatal(err)
+ }
+ return Ledger{State: s, Box: "box", Enrollment: "en", VerifyLaunches: func(context.Context, uint64) error { return nil }}
+}
+
+func TestPlan14ExecutionUpdateExclusion(t *testing.T) {
+ for _, first := range []string{"acquisition", "update"} {
+ t.Run(first, func(t *testing.T) {
+ l := censusFixture(t)
+ entered, release := make(chan struct{}), make(chan struct{})
+ acquire := func() error {
+ return l.Acquire(t.Context(), func() error {
+ close(entered)
+ <-release
+ return l.State.Write("shop.example.json", []byte(`{"Schema":1,"Status":"pending"}`))
+ })
+ }
+ update := func() error { return l.BeginUpdate(t.Context(), "update-1") }
+ var winner, loser func() error
+ if first == "acquisition" {
+ winner, loser = acquire, update
+ } else {
+ l.VerifyLaunches = func(context.Context, uint64) error { close(entered); <-release; return nil }
+ winner, loser = update, acquire
+ }
+ done := make(chan error, 1)
+ go func() { done <- winner() }()
+ <-entered
+ other := make(chan error, 1)
+ go func() { other <- loser() }()
+ // Removing every service mirror cannot change the root transaction.
+ data := t.TempDir()
+ if err := os.Mkdir(filepath.Join(data, ".hostexec"), 0700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.RemoveAll(filepath.Join(data, ".hostexec")); err != nil {
+ t.Fatal(err)
+ }
+ close(release)
+ if err := <-done; err != nil {
+ t.Fatal(err)
+ }
+ if err := <-other; err == nil {
+ t.Fatal("update and acquisition both succeeded")
+ }
+ })
+ }
+}
+func TestPlan14ParticipationRequiresCompleteRootCensus(t *testing.T) {
+ l := censusFixture(t)
+ if err := l.Idle(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.State.Remove("census.json"); err != nil {
+ t.Fatal(err)
+ }
+ if err := l.Idle(t.Context()); err == nil {
+ t.Fatal("missing root census became empty")
+ }
+ if err := l.Acquire(t.Context(), func() error { t.Fatal("unknown history acquired"); return nil }); err == nil {
+ t.Fatal("unknown history admitted")
+ }
+}
diff --git a/internal/rootparticipation/record.go b/internal/rootparticipation/record.go
new file mode 100644
index 000000000..c89adb5fb
--- /dev/null
+++ b/internal/rootparticipation/record.go
@@ -0,0 +1,57 @@
+package rootparticipation
+
+import "time"
+
+type ControllerProcess struct {
+ KernelBootID string `json:"kernel_boot_id"`
+ PID int `json:"pid"`
+ Start string `json:"start"`
+}
+
+type ExecutionRequest struct {
+ Schema int `json:"schema"`
+ ID string `json:"id"`
+ Operation string `json:"operation"`
+ Host string `json:"host"`
+ Incarnation string `json:"incarnation"`
+ Stamp uint64 `json:"stamp"`
+ Session uint64 `json:"session"`
+ Op string `json:"op"`
+ Token string `json:"token,omitempty"`
+}
+
+type ExecutionAuthority struct {
+ ParticipationOnly bool
+ Box, Enrollment, Account, App, Installation string
+ Migration uint64
+ ClaimRevision, RecordRevision, DirectoryRevision string
+ Kind, Phase string
+}
+
+type ExecutionResponse struct {
+ Schema int
+ Host, Incarnation string
+ Stamp, Session uint64
+ Op, Token, BootID, Status string
+ SafeUntil int64
+ LeaseETag string
+ Expires time.Time
+}
+
+type ExecutionReplay struct {
+ Digest string
+ Response ExecutionResponse
+}
+
+type ExecutionRecord struct {
+ TransitionFrom string
+ PriorSafeUntil int64
+ Schema int
+ Request ExecutionRequest
+ Peer ControllerProcess
+ Authority ExecutionAuthority
+ Broker, Token, Status, LeaseETag string
+ RequestStart, SafeUntil int64
+ Expires time.Time
+ Replay map[string]ExecutionReplay
+}
diff --git a/internal/rootparticipation/request.go b/internal/rootparticipation/request.go
new file mode 100644
index 000000000..07b975b64
--- /dev/null
+++ b/internal/rootparticipation/request.go
@@ -0,0 +1,23 @@
+package rootparticipation
+
+import (
+ "github.com/carlosframework/platform/internal/controlname"
+ "strings"
+)
+
+func (q ExecutionRequest) Valid() bool {
+ if q.Schema != 1 || !validID(q.ID) || len(q.ID) > 128 || !controlname.ValidHostShape(q.Host) || !validID(q.Incarnation) || q.Stamp == 0 || !validID(q.Op) {
+ return false
+ }
+ switch q.Operation {
+ case "executionBegin":
+ return q.Token == ""
+ case "executionRenew", "executionSettle", "executionView":
+ return validID(q.Token)
+ }
+ return false
+}
+
+func validID(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00\r\n\t ")
+}