WIP claim Plan 14 batch 3 checkpoint 2 (not for merge): broker shutdown, serving view, cleanup bounds
Co-Authored-By: Codex <noreply@openai.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
11 files changed,
+478
−23
internal/host/plan14_execution_authority_test.go+28 −0internal/host/rootexecution_authority.go+3 −0internal/host/rootexecution_transport.go+5 −1internal/host/rootexecution_transport_test.go+75 −0internal/hostexec/rootbroker.go+90 −14internal/hostexec/rootbroker_lifetime_test.go+118 −0internal/hostexec/rootbroker_serving_test.go+101 −0internal/hostexec/rootserving.go+43 −0internal/provision/plan9_b8_census_test.go+1 −1internal/provision/testdata/plan14-root-reads-merge.tsv+7 −6internal/rootparticipation/request.go+7 −1
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 336.020s
? github.com/carlosframework/platform/cmd/changelog [no test files]
ok github.com/carlosframework/platform/cmd/console 3.047s
ok github.com/carlosframework/platform/cmd/moveprobe 92.170s
ok github.com/carlosframework/platform/internal/activator 230.623s
ok github.com/carlosframework/platform/internal/agentsock 0.022s
ok github.com/carlosframework/platform/internal/alerting 0.004s
ok github.com/carlosframework/platform/internal/analytics 0.020s
ok github.com/carlosframework/platform/internal/appconfig 2.316s
ok github.com/carlosframework/platform/internal/apperr 0.079s
ok github.com/carlosframework/platform/internal/applog 0.134s
ok github.com/carlosframework/platform/internal/awsinv 0.005s
ok github.com/carlosframework/platform/internal/boxchoice 0.005s
ok github.com/carlosframework/platform/internal/boxid 0.742s
? github.com/carlosframework/platform/internal/boxid/boxidtest [no test files]
? github.com/carlosframework/platform/internal/boxid/internal/signtest [no test files]
ok github.com/carlosframework/platform/internal/boxvolumes 0.009s
ok github.com/carlosframework/platform/internal/caps 0.022s
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.241s
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.475s
? github.com/carlosframework/platform/internal/controlname [no test files]
? github.com/carlosframework/platform/internal/controlsock [no test files]
ok github.com/carlosframework/platform/internal/dbdoor 6.516s
ok github.com/carlosframework/platform/internal/dbdoor/dbdoortest 0.012s
ok github.com/carlosframework/platform/internal/dbverify 0.005s
ok github.com/carlosframework/platform/internal/dnsrec 0.004s
ok github.com/carlosframework/platform/internal/docsite 0.004s
ok github.com/carlosframework/platform/internal/domains 2.118s
ok github.com/carlosframework/platform/internal/doorkey 0.009s
ok github.com/carlosframework/platform/internal/economics 0.004s
ok github.com/carlosframework/platform/internal/edge 26.195s
? github.com/carlosframework/platform/internal/edgeproof [no test files]
ok github.com/carlosframework/platform/internal/edgesfile 0.009s
ok github.com/carlosframework/platform/internal/fleetapi 0.125s
ok github.com/carlosframework/platform/internal/gate 1.659s
ok github.com/carlosframework/platform/internal/genselect 0.021s
ok github.com/carlosframework/platform/internal/geo 0.014s
ok github.com/carlosframework/platform/internal/geoip 0.031s
ok github.com/carlosframework/platform/internal/host 83.583s
ok github.com/carlosframework/platform/internal/hostactors 0.325s
ok github.com/carlosframework/platform/internal/hostbox 2.110s
ok github.com/carlosframework/platform/internal/hostclaim 0.472s
ok github.com/carlosframework/platform/internal/hostevidence 0.019s
ok github.com/carlosframework/platform/internal/hostexec 0.493s
ok github.com/carlosframework/platform/internal/hostfence 0.029s
ok github.com/carlosframework/platform/internal/hostfiles 10.484s
ok github.com/carlosframework/platform/internal/hostfloor 0.019s
ok github.com/carlosframework/platform/internal/hostguard 0.380s
ok github.com/carlosframework/platform/internal/hostops 1.392s
? github.com/carlosframework/platform/internal/hostops/fencerelease [no test files]
ok github.com/carlosframework/platform/internal/hostops/internal/fencerelease 0.020s
ok github.com/carlosframework/platform/internal/hostquarantine 0.019s
ok github.com/carlosframework/platform/internal/hostrelocate 0.420s
ok github.com/carlosframework/platform/internal/hostresolve 0.150s
ok github.com/carlosframework/platform/internal/hostwrite 0.020s
ok github.com/carlosframework/platform/internal/infraview 0.021s
ok github.com/carlosframework/platform/internal/instancectl 1.727s
ok github.com/carlosframework/platform/internal/instanceexec 2.704s
ok github.com/carlosframework/platform/internal/instancefs 0.024s
ok github.com/carlosframework/platform/internal/instancekey 0.023s
ok github.com/carlosframework/platform/internal/instanceupdates 0.028s
ok github.com/carlosframework/platform/internal/layout 1.406s
ok github.com/carlosframework/platform/internal/lease 9.462s
ok github.com/carlosframework/platform/internal/lease/leasetest 0.024s
ok github.com/carlosframework/platform/internal/ledger 0.030s
ok github.com/carlosframework/platform/internal/mail 0.151s
ok github.com/carlosframework/platform/internal/maintenance 0.097s
ok github.com/carlosframework/platform/internal/maintenanceexec 5.495s
ok github.com/carlosframework/platform/internal/maintenanceguard 0.066s
ok github.com/carlosframework/platform/internal/metrics 0.004s
ok github.com/carlosframework/platform/internal/moves 21.014s
ok github.com/carlosframework/platform/internal/moves/archive 0.728s
ok github.com/carlosframework/platform/internal/objstore 0.165s
ok github.com/carlosframework/platform/internal/ops 0.157s
ok github.com/carlosframework/platform/internal/park 0.191s
ok github.com/carlosframework/platform/internal/privatefile 0.005s
ok github.com/carlosframework/platform/internal/provision 10.967s
ok github.com/carlosframework/platform/internal/publicrelease 0.008s
? 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.227s
ok github.com/carlosframework/platform/internal/release 19.010s
ok github.com/carlosframework/platform/internal/rollout 3.256s
ok github.com/carlosframework/platform/internal/rolloutwords 0.003s
ok github.com/carlosframework/platform/internal/rootaws 0.006s
ok github.com/carlosframework/platform/internal/rootparticipation 0.005s
ok github.com/carlosframework/platform/internal/rootstate 0.024s
ok github.com/carlosframework/platform/internal/s3compat 0.004s
ok github.com/carlosframework/platform/internal/safefile 0.004s
ok github.com/carlosframework/platform/internal/schedule 2.258s
ok github.com/carlosframework/platform/internal/sim/hostlife 8.566s
ok github.com/carlosframework/platform/internal/sim/litestreamfacts 3.820s
ok github.com/carlosframework/platform/internal/status 0.038s
ok github.com/carlosframework/platform/internal/steering 0.005s
ok github.com/carlosframework/platform/internal/stopevidence 0.262s
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.003s
ok github.com/carlosframework/platform/internal/tunnelclient 0.004s
ok github.com/carlosframework/platform/internal/unixdial 0.002s
ok github.com/carlosframework/platform/internal/updatewords 0.004s
ok github.com/carlosframework/platform/internal/usage 0.004s
ok github.com/carlosframework/platform/internal/version 0.002s
ok github.com/carlosframework/platform/internal/vet 1.527s
=== cross ===
--- shard 0 of 5 ---
cross-compiling darwin/arm64
--- shard 1 of 5 ---
cross-compiling darwin/amd64
--- shard 2 of 5 ---
cross-compiling linux/arm64
--- shard 3 of 5 ---
cross-compiling windows/amd64
--- shard 4 of 5 ---
cross-compiling windows/arm64
=== minio-contract ===
minio contract: arm64, minio RELEASE.2025-09-07T16-13-09Z, mc RELEASE.2025-08-13T08-35-41Z
minio: cached, digest verified
mc: cached, digest verified
minio healthy on 127.0.0.1:20999
Added `local` successfully.
Bucket created successfully `local/contract-bucket`.
Bucket created successfully `local/denied-bucket`.
Created policy `contract-only` successfully.
Added user `scopeduser` successfully.
Attached Policies: [contract-only]
To User: scopeduser
buckets and scoped user ready
=== RUN TestDeliveredContractMinio
=== RUN TestDeliveredContractMinio/put
=== RUN TestDeliveredContractMinio/get_round-trips_the_body
=== RUN TestDeliveredContractMinio/list_shows_the_key
=== RUN TestDeliveredContractMinio/head_and_list_agree_on_the_version
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict
=== RUN TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed
=== RUN TestDeliveredContractMinio/delete_removes_it
=== RUN TestDeliveredContractMinio/cross-bucket_denial
--- PASS: TestDeliveredContractMinio (0.02s)
--- PASS: TestDeliveredContractMinio/put (0.00s)
--- PASS: TestDeliveredContractMinio/get_round-trips_the_body (0.00s)
--- PASS: TestDeliveredContractMinio/list_shows_the_key (0.00s)
--- PASS: TestDeliveredContractMinio/head_and_list_agree_on_the_version (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/not_by_PutObjectTagging (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_its_own_object/and_not_by_x-amz-tagging_on_upload (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route (0.01s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.00s)
--- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_an_ordinary_tagged-with-something-else_upload_is_allowed (0.00s)
--- PASS: TestDeliveredContractMinio/delete_removes_it (0.00s)
--- PASS: TestDeliveredContractMinio/cross-bucket_denial (0.00s)
PASS
ok github.com/carlosframework/platform/internal/objstore 0.022s
=== RUN TestT4StorageProbeMinio
2026/10/13 08:32:28 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE09AEC68BD89A, HostID: 7c420053e0fbcd7f41653fdd92ffa2802c903d05303b1aa15adc2e8bb65d183f, api error SignatureDoesNotMatch: The request signature we calculated does not match the signature you provided. Check your key and signing method.
--- PASS: TestT4StorageProbeMinio (0.00s)
PASS
ok github.com/carlosframework/platform/internal/host 0.021s
minio contract: PASS
diff --git a/internal/host/plan14_execution_authority_test.go b/internal/host/plan14_execution_authority_test.go| index e395cc4f3..77a27884a 100644 |
| --- a/internal/host/plan14_execution_authority_test.go |
| +++ b/internal/host/plan14_execution_authority_test.go |
| @@ -2,6 +2,7 @@ package host |
| import ( |
| "context" |
| + "fmt" |
| "testing" |
| "github.com/carlosframework/platform/internal/hostbox" |
| @@ -117,3 +118,30 @@ func TestPlan14ExecutionPendingRegistration(t *testing.T) { |
| }) |
| } |
| } |
| + |
| +func TestPlan14ExecutionServingNamesLocalWriter(t *testing.T) { |
| + for _, foreign := range []bool{false, true} { |
| + t.Run(fmt.Sprint(foreign), func(t *testing.T) { |
| + f, root, iq := plan14InstallRegistration(t, hostclaim.KindServing, "") |
| + if _, err := root.commitRootInventory(t.Context(), iq); err != nil { |
| + t.Fatal(err) |
| + } |
| + if foreign { |
| + if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error { |
| + c.Op.Exec, c.Holder.WriterBox = "box-0", "box-0" |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "serving", Operation: "executionView", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op} |
| + _, err := root.executionAuthority(t.Context(), q, execlease.ControllerProcess{}) |
| + if foreign && err == nil { |
| + t.Fatal("retained local installation granted another box's serving session") |
| + } |
| + if !foreign && err != nil { |
| + t.Fatal(err) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/host/rootexecution_authority.go b/internal/host/rootexecution_authority.go| index 947eabf82..e65d1a9a6 100644 |
| --- a/internal/host/rootexecution_authority.go |
| +++ b/internal/host/rootexecution_authority.go |
| @@ -41,6 +41,9 @@ func (h ClaimRootHTTP) executionAuthority(ctx context.Context, q execlease.Execu |
| if c.Op.Kind != hostclaim.KindServing && !slices.Contains(c.Op.Participants, actor.Box) { |
| return out, hostbox.ErrChanged |
| } |
| + if c.Op.Kind == hostclaim.KindServing && execlease.PreinstallationParticipant(c, actor.Box, actor.Enrollment) != nil { |
| + return out, hostbox.ErrChanged |
| + } |
| _, etag, err := rootClaimRecord(ctx, h.Stop.Coordinator.Evidence, c, h.Roots["data"]) |
| if err != nil { |
| return out, err |
diff --git a/internal/host/rootexecution_transport.go b/internal/host/rootexecution_transport.go| index a3c710587..8c8f722f6 100644 |
| --- a/internal/host/rootexecution_transport.go |
| +++ b/internal/host/rootexecution_transport.go |
| @@ -71,7 +71,11 @@ func rootExecutionClient(path string) func(context.Context, execlease.ExecutionR |
| 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 == "" { |
| + serving := q.Operation == "executionView" && q.Token == "" && out.Status == "serving" && out.Token == "" && out.SafeUntil == 0 && out.LeaseETag == "" && out.Expires.IsZero() |
| + if q.Operation == "executionView" && q.Token == "" && !serving { |
| + return execlease.ExecutionResponse{}, execlease.ErrState |
| + } |
| + if out.Schema != 1 || out.Host != q.Host || out.Incarnation != q.Incarnation || out.Stamp != q.Stamp || out.Session != q.Session || out.Op != q.Op || (!serving && out.Token == "") || out.BootID == "" { |
| return execlease.ExecutionResponse{}, execlease.ErrState |
| } |
| return out, nil |
diff --git a/internal/host/rootexecution_transport_test.go b/internal/host/rootexecution_transport_test.go| index f97df4cab..9da19e71b 100644 |
| --- a/internal/host/rootexecution_transport_test.go |
| +++ b/internal/host/rootexecution_transport_test.go |
| @@ -13,6 +13,7 @@ import ( |
| "github.com/carlosframework/platform/internal/hostclaim" |
| execlease "github.com/carlosframework/platform/internal/hostexec" |
| "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| ) |
| func TestPlan14ExecutionBrokerTransport(t *testing.T) { |
| @@ -85,3 +86,77 @@ func TestPlan14ExecutionBrokerTransport(t *testing.T) { |
| t.Fatal("PID reuse borrowed execution") |
| } |
| } |
| + |
| +func TestPlan14ExecutionServingTransport(t *testing.T) { |
| + f, root, iq := plan14InstallRegistration(t, hostclaim.KindServing, "") |
| + if _, err := root.commitRootInventory(t.Context(), iq); err != nil { |
| + t.Fatal(err) |
| + } |
| + 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 } |
| + disk := &rollbackDisk{dir: t.TempDir()} |
| + raw, _ := json.Marshal(rootparticipation.Census{Schema: 1, Box: "box-1", Enrollment: "en", Migration: 1, Revision: 1, LaunchRevision: 1, Complete: true}) |
| + if err := disk.Write("census.json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + b, err := execlease.NewRootBroker(execlease.RootBrokerConfig{State: disk, Remote: f.claims, Clock: execlease.SystemClock{}, BootID: peer.BootID, Identity: "broker", Authority: root.executionAuthority, 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()) }) |
| + call := rootExecutionClient(serveClaimRoot(t, root)) |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "serving-view", Operation: "executionView", Host: iq.Target.Host, Incarnation: iq.Target.Incarnation, Stamp: iq.Target.Stamp, Session: iq.Target.Session, Op: iq.Target.Op} |
| + for _, mismatch := range []string{"none", "stamp", "session", "incarnation", "op", "token"} { |
| + t.Run(mismatch, func(t *testing.T) { |
| + bad := q |
| + switch mismatch { |
| + case "stamp": |
| + bad.Stamp++ |
| + case "session": |
| + bad.Session++ |
| + case "incarnation": |
| + bad.Incarnation = "other" |
| + case "op": |
| + bad.Op = "other" |
| + case "token": |
| + bad.Token = "other" |
| + } |
| + out, err := call(t.Context(), bad) |
| + if mismatch == "none" { |
| + if err != nil || out.Status != "serving" || out.Token != "" || out.SafeUntil != 0 { |
| + t.Fatalf("serving response: %+v %v", out, err) |
| + } |
| + } else if err == nil { |
| + t.Fatalf("mismatched %s admitted: %+v", mismatch, out) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionServingResponseShape(t *testing.T) { |
| + q := execlease.ExecutionRequest{Schema: 1, ID: "view", Operation: "executionView", Host: "shop.example.com", Incarnation: "inc", Stamp: 1, Session: 1} |
| + for _, attack := range []string{"none", "active token", "deadline", "lease", "wrong session"} { |
| + t.Run(attack, func(t *testing.T) { |
| + out := execlease.ExecutionResponse{Schema: 1, Host: q.Host, Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, BootID: "boot", Status: "serving"} |
| + switch attack { |
| + case "active token": |
| + out.Status, out.Token = "active", "unexpected-token" |
| + case "deadline": |
| + out.SafeUntil = 1234 |
| + case "lease": |
| + out.LeaseETag = "lease" |
| + case "wrong session": |
| + out.Session++ |
| + } |
| + socket := serveClaimRoot(t, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { _ = json.NewEncoder(w).Encode(out) })) |
| + _, err := rootExecutionClient(socket)(t.Context(), q) |
| + if attack == "none" && err != nil { |
| + t.Fatal(err) |
| + } |
| + if attack != "none" && err == nil { |
| + t.Fatalf("accepted %s response", attack) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/hostexec/rootbroker.go b/internal/hostexec/rootbroker.go| index 02f3c245b..68d520ecb 100644 |
| --- a/internal/hostexec/rootbroker.go |
| +++ b/internal/hostexec/rootbroker.go |
| @@ -37,20 +37,42 @@ type RootBrokerConfig struct { |
| } |
| 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() |
| + cfg RootBrokerConfig |
| + mu sync.Mutex |
| + workers sync.WaitGroup |
| + stopping bool |
| + disposed bool |
| + joined chan struct{} |
| + closeGate chan struct{} |
| + lifetimeCtx context.Context |
| + cancel context.CancelFunc |
| + closeErr error |
| + 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 |
| + lifetimeCtx, cancel := context.WithCancel(context.Background()) |
| + return &RootBroker{cfg: cfg, lifetimeCtx: lifetimeCtx, cancel: cancel, closeGate: make(chan struct{}, 1), denied: make(map[string]bool), calls: make(map[string]chan struct{}), lifetime: make(map[string]func()), live: make(map[string]func())}, nil |
| +} |
| + |
| +func (b *RootBroker) enter(ctx context.Context) (context.Context, func(), error) { |
| + b.mu.Lock() |
| + if b.stopping { |
| + b.mu.Unlock() |
| + return nil, nil, ErrLost |
| + } |
| + b.workers.Add(1) |
| + b.mu.Unlock() |
| + work, cancel := context.WithCancel(ctx) |
| + stop := context.AfterFunc(b.lifetimeCtx, cancel) |
| + return work, func() { stop(); cancel(); b.workers.Done() }, nil |
| } |
| func executionDigest(q ExecutionRequest, p ControllerProcess) string { |
| @@ -166,7 +188,9 @@ func (b *RootBroker) deny(ctx context.Context, r ExecutionRecord, cause error) e |
| } |
| } |
| b.mu.Unlock() |
| - clean := b.cfg.Cleanup(context.WithoutCancel(ctx), r) |
| + cleanup, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second) |
| + defer cancel() |
| + clean := b.cfg.Cleanup(cleanup, r) |
| return errors.Join(cause, persist, clean) |
| } |
| @@ -174,6 +198,11 @@ func (b *RootBroker) Execute(ctx context.Context, peer ControllerProcess, q Exec |
| if !q.Valid() || peer.PID <= 0 || !validID(peer.Start) || peer.KernelBootID != b.cfg.BootID { |
| return ExecutionResponse{}, ErrState |
| } |
| + ctx, leave, enterErr := b.enter(ctx) |
| + if enterErr != nil { |
| + return ExecutionResponse{}, enterErr |
| + } |
| + defer leave() |
| b.mu.Lock() |
| gate := b.calls[q.Host] |
| if gate == nil { |
| @@ -220,6 +249,12 @@ func (b *RootBroker) Execute(ctx context.Context, peer ControllerProcess, q Exec |
| } |
| return ExecutionResponse{}, err |
| } |
| + if q.Operation == "executionView" && q.Token == "" { |
| + return b.servingView(ctx, peer, q, authority) |
| + } |
| + if authority.Kind == "serving" { |
| + return ExecutionResponse{}, ErrInactive |
| + } |
| } |
| r, err := b.read(q.Host) |
| missing := errors.Is(err, os.ErrNotExist) |
| @@ -447,6 +482,11 @@ func (b *RootBroker) settle(ctx context.Context, q ExecutionRequest, r Execution |
| } |
| func (b *RootBroker) Monitor(ctx context.Context) error { |
| + ctx, leave, err := b.enter(ctx) |
| + if err != nil { |
| + return err |
| + } |
| + defer leave() |
| names, err := b.cfg.State.Names() |
| if err != nil { |
| return err |
| @@ -498,12 +538,34 @@ func executionInstallationBinding(old, next ExecutionAuthority) bool { |
| func (b *RootBroker) Close(ctx context.Context) error { |
| b.mu.Lock() |
| b.closed = true |
| + if !b.stopping { |
| + b.stopping = true |
| + b.cancel() |
| + for host, unlock := range b.live { |
| + unlock() |
| + delete(b.live, host) |
| + } |
| + b.joined = make(chan struct{}) |
| + go func() { b.workers.Wait(); close(b.joined) }() |
| + } |
| + joined := b.joined |
| b.mu.Unlock() |
| - names, err := b.cfg.State.Names() |
| - if err != nil { |
| - return err |
| + select { |
| + case <-joined: |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| } |
| - var errs []error |
| + select { |
| + case b.closeGate <- struct{}{}: |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + } |
| + defer func() { <-b.closeGate }() |
| + if b.disposed { |
| + return b.closeErr |
| + } |
| + names, err := b.cfg.State.Names() |
| + errs := []error{err} |
| for _, name := range names { |
| if name == "census.json" || name == "update.json" || !strings.HasSuffix(name, ".json") { |
| continue |
| @@ -523,10 +585,17 @@ func (b *RootBroker) Close(ctx context.Context) error { |
| delete(b.lifetime, host) |
| } |
| b.mu.Unlock() |
| - return errors.Join(errs...) |
| + b.disposed = true |
| + b.closeErr = errors.Join(errs...) |
| + return b.closeErr |
| } |
| func (b *RootBroker) Recover(ctx context.Context) error { |
| + ctx, leave, err := b.enter(ctx) |
| + if err != nil { |
| + return err |
| + } |
| + defer leave() |
| names, err := b.cfg.State.Names() |
| if err != nil { |
| return err |
| @@ -562,10 +631,17 @@ func (b *RootBroker) Recover(ctx context.Context) error { |
| } |
| func (b *RootBroker) Run(ctx context.Context) { |
| + ctx, leave, err := b.enter(ctx) |
| + if err != nil { |
| + return |
| + } |
| + defer leave() |
| for { |
| select { |
| case <-ctx.Done(): |
| return |
| + case <-b.lifetimeCtx.Done(): |
| + return |
| case <-b.cfg.Clock.After(100 * time.Millisecond): |
| } |
| _ = b.Monitor(ctx) |
diff --git a/internal/hostexec/rootbroker_lifetime_test.go b/internal/hostexec/rootbroker_lifetime_test.go| new file mode 100644 |
| index 000000000..a58a0c333 |
| --- /dev/null |
| +++ b/internal/hostexec/rootbroker_lifetime_test.go |
| @@ -0,0 +1,118 @@ |
| +//go:build linux |
| + |
| +package hostexec |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func TestPlan14ExecutionShutdownJoinsAcquisition(t *testing.T) { |
| + b, disk, remote, _, q, peer := brokerFixture(t) |
| + entered, resume := make(chan struct{}), make(chan struct{}) |
| + remote.onPut = func(int) { close(entered); <-resume } |
| + result := make(chan error, 1) |
| + go func() { _, err := b.Execute(t.Context(), peer, q); result <- err }() |
| + <-entered |
| + closing := make(chan error, 1) |
| + go func() { closing <- b.Close(t.Context()) }() |
| + var early bool |
| + select { |
| + case <-closing: |
| + early = true |
| + case <-time.After(50 * time.Millisecond): |
| + } |
| + probe, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) |
| + unlock, lockErr := disk.Lock(probe, q.Host+".lock") |
| + cancel() |
| + if lockErr == nil { |
| + unlock() |
| + } |
| + close(resume) |
| + if err := <-result; err == nil { |
| + t.Error("acquisition returned a permit during shutdown") |
| + } |
| + if !early { |
| + if err := <-closing; err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if early || !errors.Is(lockErr, context.DeadlineExceeded) { |
| + t.Fatalf("shutdown released ownership before acquisition joined: early=%v lock=%v", early, lockErr) |
| + } |
| + r, err := b.read(q.Host) |
| + if err != nil || r.Status != "lost" || r.LeaseETag == "" { |
| + t.Fatalf("late acquisition evidence was lost: %+v %v", r, err) |
| + } |
| + probe, cancel = context.WithTimeout(t.Context(), time.Second) |
| + defer cancel() |
| + unlock, err = disk.Lock(probe, q.Host+".lock") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + unlock() |
| +} |
| + |
| +func TestPlan14ExecutionShutdownRevokesBeforeCleanup(t *testing.T) { |
| + b, disk, _, _, q, peer := brokerFixture(t) |
| + if _, err := b.Execute(t.Context(), peer, q); err != nil { |
| + t.Fatal(err) |
| + } |
| + entered, resume := make(chan struct{}), make(chan struct{}) |
| + b.cfg.Cleanup = func(context.Context, ExecutionRecord) error { close(entered); <-resume; return nil } |
| + closing := make(chan error, 1) |
| + go func() { closing <- b.Close(t.Context()) }() |
| + <-entered |
| + probe, cancel := context.WithTimeout(t.Context(), time.Second) |
| + unlock, err := disk.Lock(probe, q.Host+".live") |
| + cancel() |
| + if err == nil { |
| + unlock() |
| + } |
| + close(resume) |
| + if closeErr := <-closing; closeErr != nil { |
| + t.Fatal(closeErr) |
| + } |
| + if err != nil { |
| + t.Fatalf("cleanup began before live permission was withdrawn: %v", err) |
| + } |
| + q.ID = "after-shutdown" |
| + if _, err := b.Execute(t.Context(), peer, q); err == nil { |
| + t.Fatal("closed broker accepted new work") |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionShutdownTimeoutRetainsOwnership(t *testing.T) { |
| + b, disk, remote, _, q, peer := brokerFixture(t) |
| + entered, resume := make(chan struct{}), make(chan struct{}) |
| + remote.onPut = func(int) { close(entered); <-resume } |
| + result := make(chan error, 1) |
| + go func() { _, err := b.Execute(t.Context(), peer, q); result <- err }() |
| + <-entered |
| + bounded, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) |
| + err := b.Close(bounded) |
| + cancel() |
| + if !errors.Is(err, context.DeadlineExceeded) { |
| + close(resume) |
| + <-result |
| + t.Fatalf("shutdown did not retain the outstanding acquisition: %v", err) |
| + } |
| + probe, cancel := context.WithTimeout(t.Context(), 20*time.Millisecond) |
| + unlock, lockErr := disk.Lock(probe, q.Host+".lock") |
| + cancel() |
| + if lockErr == nil { |
| + unlock() |
| + } |
| + close(resume) |
| + if err := <-result; err == nil { |
| + t.Fatal("late acquisition returned permission") |
| + } |
| + if err := b.Close(t.Context()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !errors.Is(lockErr, context.DeadlineExceeded) { |
| + t.Fatalf("timeout abandoned ownership: %v", lockErr) |
| + } |
| +} |
diff --git a/internal/hostexec/rootbroker_serving_test.go b/internal/hostexec/rootbroker_serving_test.go| new file mode 100644 |
| index 000000000..456fb02d0 |
| --- /dev/null |
| +++ b/internal/hostexec/rootbroker_serving_test.go |
| @@ -0,0 +1,101 @@ |
| +//go:build linux |
| + |
| +package hostexec |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "os" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| +) |
| + |
| +func TestPlan14ExecutionServingView(t *testing.T) { |
| + for _, scenario := range []string{"empty", "settled", "active", "lost", "missing census", "incomplete census", "wrong migration", "wrong enrollment", "update", "wrong kind", "participation only", "dead peer"} { |
| + t.Run(scenario, func(t *testing.T) { |
| + b, disk, remote, _, q, peer := brokerFixture(t) |
| + if scenario == "settled" || scenario == "active" || scenario == "lost" { |
| + out, err := b.Execute(t.Context(), peer, q) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if scenario == "settled" { |
| + settle := q |
| + settle.Operation, settle.ID, settle.Token = "executionSettle", "settle", out.Token |
| + if _, err := b.Execute(t.Context(), peer, settle); err != nil { |
| + t.Fatal(err) |
| + } |
| + } else if scenario == "lost" { |
| + r, _ := b.read(q.Host) |
| + if err := b.deny(t.Context(), r, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + } |
| + census := rootparticipation.Census{Schema: 1, Box: "box", Enrollment: "en", Migration: 1, Revision: 1, LaunchRevision: 1, Complete: true} |
| + if scenario == "incomplete census" { |
| + census.Complete = false |
| + } |
| + if scenario == "wrong migration" { |
| + census.Migration++ |
| + } |
| + if scenario == "wrong enrollment" { |
| + census.Enrollment = "other" |
| + } |
| + if scenario != "missing census" { |
| + raw, _ := json.Marshal(census) |
| + if err := disk.Write("census.json", raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if scenario == "update" { |
| + if err := disk.Write("update.json", []byte(`{}`)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + original := b.cfg.Authority |
| + originalQ := q |
| + b.cfg.Authority = func(ctx context.Context, _ ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) { |
| + a, err := original(ctx, originalQ, p) |
| + a.Kind, a.Phase = "serving", "" |
| + if scenario == "wrong kind" { |
| + a.Kind = "landing" |
| + } |
| + if scenario == "participation only" { |
| + a.ParticipationOnly = true |
| + } |
| + return a, err |
| + } |
| + if scenario == "dead peer" { |
| + b.cfg.Alive = func(context.Context, ControllerProcess) error { return os.ErrNotExist } |
| + } |
| + q.Operation, q.ID, q.Token, q.Op = "executionView", "serving-view", "", "" |
| + writes, acquisitions := disk.writes, remote.calls |
| + out, err := b.Execute(t.Context(), peer, q) |
| + if scenario == "empty" || scenario == "settled" { |
| + if err != nil || out.Status != "serving" || out.Host != q.Host || out.Stamp != q.Stamp || out.Session != q.Session || out.Token != "" || out.SafeUntil != 0 || out.LeaseETag != "" || !out.Expires.IsZero() { |
| + t.Fatalf("serving view: %+v %v", out, err) |
| + } |
| + } else if err == nil { |
| + t.Fatalf("%s granted serving: %+v", scenario, out) |
| + } |
| + if disk.writes != writes || remote.calls != acquisitions { |
| + t.Fatal("serving view wrote participation or acquired a lease") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestPlan14ExecutionServingCannotAcquire(t *testing.T) { |
| + b, _, remote, _, q, peer := brokerFixture(t) |
| + original := b.cfg.Authority |
| + b.cfg.Authority = func(ctx context.Context, q ExecutionRequest, p ControllerProcess) (ExecutionAuthority, error) { |
| + a, err := original(ctx, q, p) |
| + a.Kind = "serving" |
| + return a, err |
| + } |
| + if _, err := b.Execute(t.Context(), peer, q); err == nil || remote.calls != 0 { |
| + t.Fatalf("serving acquired an operation lease: err=%v acquisitions=%d", err, remote.calls) |
| + } |
| +} |
diff --git a/internal/hostexec/rootserving.go b/internal/hostexec/rootserving.go| new file mode 100644 |
| index 000000000..ac3c6053d |
| --- /dev/null |
| +++ b/internal/hostexec/rootserving.go |
| @@ -0,0 +1,43 @@ |
| +package hostexec |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + |
| + "github.com/carlosframework/platform/internal/rootparticipation" |
| +) |
| + |
| +func (b *RootBroker) servingView(ctx context.Context, peer ControllerProcess, q ExecutionRequest, authority ExecutionAuthority) (ExecutionResponse, error) { |
| + if authority.Kind != "serving" || authority.ParticipationOnly || authority.Installation == "" { |
| + return ExecutionResponse{}, ErrInactive |
| + } |
| + unlock, err := b.cfg.State.Lock(ctx, "transaction.lock") |
| + if err != nil { |
| + return ExecutionResponse{}, err |
| + } |
| + defer unlock() |
| + if _, err := b.cfg.State.Read("update.json"); !errors.Is(err, os.ErrNotExist) { |
| + return ExecutionResponse{}, ErrInactive |
| + } |
| + census, err := (rootparticipation.Ledger{State: b.cfg.State, Box: authority.Box, Enrollment: authority.Enrollment}).Census() |
| + if err != nil || census.Migration != authority.Migration { |
| + return ExecutionResponse{}, errors.Join(err, ErrState) |
| + } |
| + if _, err := ReadRoot(ctx, b.cfg.State, b.cfg.Clock, q, true); err != nil { |
| + return ExecutionResponse{}, err |
| + } |
| + fresh, err := b.cfg.Authority(ctx, q, peer) |
| + if err != nil || fresh != authority { |
| + return ExecutionResponse{}, errors.Join(err, ErrLost) |
| + } |
| + if err := b.cfg.Alive(ctx, peer); err != nil { |
| + return ExecutionResponse{}, err |
| + } |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + if b.closed || ctx.Err() != nil { |
| + return ExecutionResponse{}, ErrLost |
| + } |
| + return ExecutionResponse{Schema: 1, Host: q.Host, Incarnation: q.Incarnation, Stamp: q.Stamp, Session: q.Session, Op: q.Op, BootID: peer.KernelBootID, Status: "serving"}, nil |
| +} |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index 3a55ade04..cf1020b8d 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -734,7 +734,7 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"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/rootexecution_authority.go", "ClaimRootHTTP.executionAuthority", "key-site", "key:RecordKey", "90b20d5af85ba9ef95bb8487797a715af7d43a1b3350058173079aece3bdea47"}, |
| {"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"}, |
diff --git a/internal/provision/testdata/plan14-root-reads-merge.tsv b/internal/provision/testdata/plan14-root-reads-merge.tsv| index 305d4a417..883b9672b 100644 |
| --- a/internal/provision/testdata/plan14-root-reads-merge.tsv |
| +++ b/internal/provision/testdata/plan14-root-reads-merge.tsv |
| @@ -251,7 +251,7 @@ internal/sim/hostlife/mutate.sh external legacy accepted-legacy accepted-legacy; |
| @coverage internal/console 41714f3779b343460800cb25b31782dc7d5c1e83b2aacf7d8d0a28f39c47bf76 |
| @coverage internal/console/templates 6492b7455fa3315105339cca03edb4372ac0d9bf746c018e13a308cf0b74cc05 |
| @coverage internal/edgesfile 553af147ca4882b8425afddf1ed250ee40a6cac92361d667cb4a690393be5272 |
| -@coverage internal/host 4b6fccbf2d3d6c0c9696563d4deefebfbe0d35b24f75bfca493acd6f5185c272 |
| +@coverage internal/host d9e1b4aecc47f93e592ba94f5e51214592636cef8a677792b3985a8311c4e0b5 |
| @coverage internal/hostbox afb08c287eaccb43adde1c488bd122482711ce4e29ecb675c9428f27130c3d93 |
| @coverage internal/hostclaim feaf701db4b49ce92733b9d6c1fa1c0ae8efc6e13439ce201fca1a9e24dcf927 |
| @coverage internal/hostfiles 554285c123758375c13388b2fa42f654819392dac0b027816a713548ec17dcc6 |
| @@ -268,13 +268,13 @@ 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 |
| +@coverage internal/hostexec 899bdcd987f69f09cc5fcd89949a69387827ec48bc23e5d54abac96c7a335e85 |
| +@coverage internal/rootparticipation db8fb3a896a400cfe6396f436625732502a83af1618be7fc3df7b198a8c1bc32 |
| +internal/host/rootexecution_authority.go ClaimRootHTTP.executionAuthority authority D D1 reads the fresh claim, deployment record and root inventory, including the serving writer box; candidate admission only. Production broker construction is pending. file:Read@553 90b20d5af85ba9ef95bb8487797a715af7d43a1b3350058173079aece3bdea47 |
| internal/host/rootexecution_read.go rootExecutionRead authority D D1/D2 open the descriptor-checked root participation namespace; no service mirror grants authority. Production final-effect wiring is pending. file:Open@210 87ad4f6167cf63d0364c12a48ab0ef7279fed6fa0920c012ae7738295496bbdc |
| internal/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.Execute authority D D1/D2 register in-flight work before acquiring root lifetime locks, dispatch a tokenless serving view without acquisition, and retain conditional-write and replay evidence. file:Read@3101,file:Read@4519 10b222fb88be75fc062d1a45c86037580eaab725b1bffe5d4848b5881d538e30 |
| +internal/hostexec/rootbroker.go RootBroker.deny authority D D2 compares the current root token before durable loss and live-lock revocation; cleanup is bounded independently of request cancellation. file:Read@117 e29e9132d0ceb540000a1836249fda71c4a75b4746d6037a42cbf6e2e2752b9d |
| internal/hostexec/rootbroker.go RootBroker.read authority D D1/D2 decode the retained root participation record; missing or malformed state grants nothing. file:Read@113 8a2fb0134e9c24bbe448efc35f193e60d01e909a3129b465aa16f682eb4114f1 |
| internal/hostexec/rootread.go ReadRoot authority D D1/D2 require migration completeness, exact root record, kernel peer identity, request-start cutoff and held root live lock. file:Read@1915,file:Read@392 6e9e0b87e455815edca1c2e1ea43d4e0001bef5f5eeb2d5c0c6171a4a86dbb62 |
| internal/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 |
| @@ -283,3 +283,4 @@ internal/rootparticipation/ledger.go Ledger.Census authority D D3 requires expli |
| 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 |
| +internal/hostexec/rootserving.go RootBroker.servingView authority D D1 reads only completed root census and settled root participation under transaction exclusion, rechecks claim authority and peer identity, and returns no lease or token. Production client selection is pending. file:Read@441 9a57720834b3d7c661854196cf0ed1261c643673b5ba900c38987a162f5de0bb |
diff --git a/internal/rootparticipation/request.go b/internal/rootparticipation/request.go| index 07b975b64..a1eadd61e 100644 |
| --- a/internal/rootparticipation/request.go |
| +++ b/internal/rootparticipation/request.go |
| @@ -6,7 +6,13 @@ import ( |
| ) |
| 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) { |
| + if q.Schema != 1 || !validID(q.ID) || len(q.ID) > 128 || !controlname.ValidHostShape(q.Host) || !validID(q.Incarnation) || q.Stamp == 0 { |
| + return false |
| + } |
| + if q.Operation == "executionView" && q.Token == "" { |
| + return q.Op == "" || validID(q.Op) |
| + } |
| + if !validID(q.Op) { |
| return false |
| } |
| switch q.Operation { |