amadan

carlos / platform Public

WIP files seed checkpoint 13 (not for merge)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude bdd69cca151f038539f1a5973097ef576296fb31 parent f156851
3 files changed, +504 −3
  • internal/host/claimservinghandover_linux_test.go +162 −0
  • internal/host/claimservingpeer_rootrunner_linux_test.go +22 −3
  • internal/host/claimservingwriter_rootrunner_linux_test.go +320 −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	348.677s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.163s
ok  	github.com/carlosframework/platform/cmd/moveprobe	92.218s
ok  	github.com/carlosframework/platform/cmd/zddprobe	0.710s
ok  	github.com/carlosframework/platform/internal/activator	233.283s
ok  	github.com/carlosframework/platform/internal/agentsock	0.017s
ok  	github.com/carlosframework/platform/internal/alerting	0.004s
ok  	github.com/carlosframework/platform/internal/analytics	0.023s
ok  	github.com/carlosframework/platform/internal/appconfig	2.365s
ok  	github.com/carlosframework/platform/internal/apperr	0.076s
ok  	github.com/carlosframework/platform/internal/applog	0.134s
ok  	github.com/carlosframework/platform/internal/awsinv	0.003s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.004s
ok  	github.com/carlosframework/platform/internal/boxid	0.895s
?   	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.720s
ok  	github.com/carlosframework/platform/internal/changelog	0.294s
ok  	github.com/carlosframework/platform/internal/clamd	0.055s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	366.469s
?   	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.049s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.009s
ok  	github.com/carlosframework/platform/internal/dbverify	0.005s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.005s
ok  	github.com/carlosframework/platform/internal/docsite	0.005s
ok  	github.com/carlosframework/platform/internal/domains	2.142s
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.320s
?   	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.133s
ok  	github.com/carlosframework/platform/internal/gate	1.664s
ok  	github.com/carlosframework/platform/internal/genselect	0.017s
ok  	github.com/carlosframework/platform/internal/geo	0.017s
ok  	github.com/carlosframework/platform/internal/geoip	0.037s
ok  	github.com/carlosframework/platform/internal/host	148.662s
ok  	github.com/carlosframework/platform/internal/hostactors	0.438s
ok  	github.com/carlosframework/platform/internal/hostbox	2.371s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.548s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.007s
ok  	github.com/carlosframework/platform/internal/hostexec	4.783s
ok  	github.com/carlosframework/platform/internal/hostfence	0.041s
ok  	github.com/carlosframework/platform/internal/hostfiles	10.482s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.012s
ok  	github.com/carlosframework/platform/internal/hostguard	0.977s
ok  	github.com/carlosframework/platform/internal/hostops	1.467s
?   	github.com/carlosframework/platform/internal/hostops/fencerelease	[no test files]
ok  	github.com/carlosframework/platform/internal/hostops/internal/fencerelease	0.002s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.002s
ok  	github.com/carlosframework/platform/internal/hostrelocate	0.463s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.142s
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.819s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.728s
ok  	github.com/carlosframework/platform/internal/instancefs	0.006s
ok  	github.com/carlosframework/platform/internal/instancekey	0.002s
ok  	github.com/carlosframework/platform/internal/instanceupdates	0.007s
ok  	github.com/carlosframework/platform/internal/layout	1.433s
ok  	github.com/carlosframework/platform/internal/lease	9.451s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.008s
ok  	github.com/carlosframework/platform/internal/ledger	0.020s
ok  	github.com/carlosframework/platform/internal/mail	0.161s
ok  	github.com/carlosframework/platform/internal/maintenance	0.086s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.548s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.074s
ok  	github.com/carlosframework/platform/internal/metrics	0.006s
ok  	github.com/carlosframework/platform/internal/moves	21.070s
ok  	github.com/carlosframework/platform/internal/moves/archive	0.768s
ok  	github.com/carlosframework/platform/internal/objstore	0.199s
ok  	github.com/carlosframework/platform/internal/ops	0.166s
ok  	github.com/carlosframework/platform/internal/park	0.210s
ok  	github.com/carlosframework/platform/internal/privatefile	0.008s
ok  	github.com/carlosframework/platform/internal/provision	12.864s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.009s
?   	github.com/carlosframework/platform/internal/regionname	[no test files]
ok  	github.com/carlosframework/platform/internal/registry	4.023s
ok  	github.com/carlosframework/platform/internal/relay	7.225s
ok  	github.com/carlosframework/platform/internal/release	19.316s
ok  	github.com/carlosframework/platform/internal/rollout	3.711s
ok  	github.com/carlosframework/platform/internal/rolloutwords	0.005s
ok  	github.com/carlosframework/platform/internal/rootaws	0.007s
ok  	github.com/carlosframework/platform/internal/rootparticipation	0.007s
ok  	github.com/carlosframework/platform/internal/rootstate	0.025s
ok  	github.com/carlosframework/platform/internal/s3compat	0.005s
ok  	github.com/carlosframework/platform/internal/safefile	0.004s
ok  	github.com/carlosframework/platform/internal/schedule	2.297s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	9.624s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	3.989s
ok  	github.com/carlosframework/platform/internal/status	0.039s
ok  	github.com/carlosframework/platform/internal/steering	0.007s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.263s
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.004s
ok  	github.com/carlosframework/platform/internal/version	0.002s
ok  	github.com/carlosframework/platform/internal/vet	1.559s
=== 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/14 19:16:04 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DE7B62589E8447, 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

Intent

full ledger: feat/claim-files-backfill

diff --git a/internal/host/claimservinghandover_linux_test.go b/internal/host/claimservinghandover_linux_test.go
index 5e3cf4fa3..30bba16f1 100644
--- a/internal/host/claimservinghandover_linux_test.go
+++ b/internal/host/claimservinghandover_linux_test.go
@@ -10,10 +10,13 @@ import (
"reflect"
"strconv"
"testing"
+ "time"
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
)
@@ -306,3 +309,162 @@ func TestServingHandoverRefusesForeignAuthority(t *testing.T) {
})
}
}
+
+// The stop coordinator writes the receipt used by the next handover. Process
+// and authentication observations remain fixtures; the rootrunner covers them.
+func TestServingStoppedWriterSecondHandover(t *testing.T) {
+ f := newServingRootFixture(t, true)
+ root := servingHandoverRoot(t, f)
+ f.f.peer = StopPeer{UID: 1001, PID: 77, ProcessStart: "edge-start", BootID: "boot", ControllerID: "controller"}
+ coordinator := f.f.root(nil)
+ coordinator.Evidence, coordinator.Inventory = f.f.claims, root.Journals.Inventory
+ root.Stop.Coordinator = coordinator
+ f.p.root.Stop.Coordinator = coordinator
+ socket := servingRootSocket(t, root)
+ transfer := coldHandoverWith(handoverRecoveryIdentity(t, f.actor), socket, func(context.Context, hostclaim.Observed) (hostguard.FenceReceipt, error) {
+ return f.receipt, nil
+ })
+ if err := transfer(t.Context(), f.predecessor, f.f.read()); err != nil {
+ t.Fatal(err)
+ }
+ target := recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor, BootID: "boot", ControllerID: "controller"})
+ if err := f.f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "edge-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/next", Device: 1, Inode: 99}}}); err != nil {
+ t.Fatal(err)
+ }
+ client := hostbox.ServingParkClient{Call: rootJournalClient(socket)}
+ first, err := client.Park(t.Context(), f.current, f.prior)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.tags(f.current.Stamp)
+ if err := client.Complete(t.Context(), first.ID, f.current); err != nil {
+ t.Fatal(err)
+ }
+ for suffix, bytes := range map[string]string{"": "second writer bytes", ".gen": f.current.Generation, "-wal": "second WAL", "-shm": "second SHM"} {
+ if err := os.WriteFile(f.db+suffix, []byte(bytes), 0600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ r := hostbox.LaunchReservation{ID: "second:main", Role: "main", InvocationID: "second", Inputs: "digest", Target: target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(target.Host)}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}}
+ if err := f.f.journal.IntendUnit(t.Context(), r); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.f.journal.StartedUnit(t.Context(), r, hostbox.LaunchGroup{RelativePath: r.Group.RelativePath, Device: 1, Inode: 4, InvocationID: "second-unit"}, 99, "tenant-start"); err != nil {
+ t.Fatal(err)
+ }
+ o := f.f.read()
+ req := StopRequest{Region: f.actor.Region, Host: o.Host, Incarnation: f.current.Incarnation, InstallationStamp: f.current.Stamp, WriterSession: f.current.Session, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: "boot", ControllerID: "controller", ExpectedClaimETag: o.ETag, ExpectedDirectoryRevision: "7"}
+ drain := stopDrainFunc(func(_ context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
+ return DrainClaimAcknowledgement(fence, req.ControllerID, "edge-start", nil), nil
+ })
+ if _, err := CollectClaimStop(t.Context(), socket, req, drain); err == nil {
+ t.Fatal("accepted an unsettled unit")
+ }
+ if err := f.f.journal.SettleUnit(t.Context(), target, r.ID); err != nil {
+ t.Fatal(err)
+ }
+ stopped, err := CollectClaimStop(t.Context(), socket, req, drain)
+ if err != nil {
+ t.Fatal("stop through transport", err)
+ }
+ f.receipt, err = f.f.guard.FenceReceiptFor(t.Context(), stopped.Target)
+ if err != nil || string(f.receipt.Bytes) != string(stopped.Bytes) {
+ t.Fatal("durable receipt", err)
+ }
+ previous := f.f.read()
+ h := previous.Claim.Holder
+ admitted, err := hostclaim.Update(t.Context(), f.f.claims, previous, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G3"}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ prior := f.current
+ f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := transfer(t.Context(), previous, admitted); err != nil {
+ t.Fatal("second handover", err)
+ }
+ parked, err := client.Park(t.Context(), f.current, prior)
+ if err != nil {
+ t.Fatal("second park", err)
+ }
+ f.tags(f.current.Stamp)
+ if err := client.Complete(t.Context(), parked.ID, f.current); err != nil {
+ t.Fatal(err)
+ }
+ record, err := f.p.ledger.read(parked.ID)
+ if err != nil || !record.Complete || record.Disposition != servingParkRetain || len(record.Receipts) != 1 || string(record.Receipts[0].Bytes) != string(stopped.Bytes) {
+ t.Fatal("second park did not consume exact transport stop receipt", err)
+ }
+ old, err := f.p.ledger.read(first.ID)
+ if err != nil || !old.Closed || old.Cause != "replacement" {
+ t.Fatal("previous park was not replaced", err)
+ }
+ for suffix, want := range map[string]string{"": "second writer bytes", ".gen": prior.Generation, "-wal": "second WAL", "-shm": "second SHM"} {
+ body, err := os.ReadFile(filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", parked.ID, filepath.Base(f.db)+suffix))
+ if err != nil || string(body) != want {
+ t.Fatal("second park changed bytes", suffix, err)
+ }
+ }
+ if replay, err := client.Park(t.Context(), f.current, prior); err != nil || replay != parked {
+ t.Fatal("second park replay", err)
+ }
+ if after, err := f.p.ledger.read(parked.ID); err != nil || !reflect.DeepEqual(record, after) {
+ t.Fatal("second park replay changed record", err)
+ }
+ servingRestoreAfterStoppedWriter(t, f, client)
+ if after, err := f.p.ledger.read(parked.ID); err != nil || !reflect.DeepEqual(record, after) {
+ t.Fatal("restore changed retained park", err)
+ }
+}
+
+// This connects production ColdRestore and retirement enrollment to the stopped
+// writer's park. The exact-version LTX baseline and local policy are fixtures;
+// full wake preparation and C8 remain separate acceptance gates.
+func servingRestoreAfterStoppedWriter(t *testing.T, f *servingRootFixture, parker servingDatabaseParker) {
+ t.Helper()
+ plan := plan12RestoreBaseline(t, f.f)
+ before, err := f.f.data.GetVersion(t.Context(), f.f.bucket, plan.Files[0].Key, plan.Files[0].VersionID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ admitted := f.f.read()
+ in := ClaimLaunchInputs{DBPath: f.db}
+ r := &claimServingArtifactRetirement{Claims: f.f.claims, Guard: f.f.guard, DataDir: f.f.guard.DataDir, Box: f.actor.Box, Region: f.actor.Region, Root: parker}
+ source := hostbox.WorkSource{Guard: f.f.guard, Box: f.actor.Box, Region: f.actor.Region, Now: func() time.Time { return f.f.now }, Policy: func(_ context.Context, c *hostclaim.Claim, selected hostbox.Selection, _ hostwrite.Identity) error {
+ return hostbox.Serving(c, selected)
+ }, Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
+ work, err := source.Select(t.Context(), admitted.Host, "restore")
+ if err != nil {
+ t.Fatal("select restore work", err)
+ }
+ if err := r.Recover(t.Context(), admitted, in); err != nil {
+ t.Fatal("recover completed retirement", err)
+ }
+ enrolled := 0
+ restore := hostbox.ColdRestore{Reader: f.f.data, Bucket: f.f.bucket, Generation: f.current.Generation, Work: work,
+ BeforeLegacyInstall: func(ctx context.Context, _, _ string) error { return r.BeforeLegacyInstall(ctx, admitted, in, work) },
+ BeforeInstallLocked: func(ctx context.Context, _, _ string) error {
+ if err := r.EnrollLocked(ctx, admitted, in); err != nil {
+ return err
+ }
+ enrolled++
+ return nil
+ },
+ RetireArtifacts: func(ctx context.Context, generation string) error { return r.Retire(ctx, admitted, in, generation) },
+ }
+ if err := restore.Install(t.Context(), f.db, plan); err != nil {
+ t.Fatal("restore after stopped writer", err)
+ }
+ if enrolled != 1 || mwReadDB(t, f.db) != "selected recovery baseline" {
+ t.Fatal("restore missed enrollment or restored the wrong baseline")
+ }
+ if marker, err := hostbox.ReadGenerationMarker(f.db + ".gen"); err != nil || marker != f.current.Generation {
+ t.Fatal("restored generation", marker, err)
+ }
+ after, err := f.f.data.GetVersion(t.Context(), f.f.bucket, plan.Files[0].Key, plan.Files[0].VersionID)
+ if err != nil || string(after) != string(before) {
+ t.Fatal("restore changed remote baseline", err)
+ }
+}
diff --git a/internal/host/claimservingpeer_rootrunner_linux_test.go b/internal/host/claimservingpeer_rootrunner_linux_test.go
index c2e348617..f825123c2 100644
--- a/internal/host/claimservingpeer_rootrunner_linux_test.go
+++ b/internal/host/claimservingpeer_rootrunner_linux_test.go
@@ -38,12 +38,14 @@ type servingPeerCommand struct {
Previous hostclaim.Observed
Admitted hostclaim.Observed
Actor hostbox.Actor
+ Stop StopRequest
}
type servingPeerReply struct {
Peer StopPeer
Authorization servingParkAuthorization
Launch hostbox.LaunchInventory
+ Stopped StopResult
Error string
}
@@ -101,6 +103,10 @@ func TestRootServingParkPeerPayload(t *testing.T) {
var reply servingPeerReply
var callErr error
switch q.Action {
+ case "stop":
+ reply.Stopped, callErr = CollectClaimStop(t.Context(), socket, q.Stop, stopDrainFunc(func(_ context.Context, req StopRequest, fence string) (DrainAcknowledgement, error) {
+ return DrainClaimAcknowledgement(fence, req.ControllerID, peer.ProcessStart, nil), nil
+ }))
case "handover":
hook := coldHandoverWith(handoverRecoveryIdentity(t, q.Actor), socket, func(context.Context, hostclaim.Observed) (hostguard.FenceReceipt, error) {
return hostguard.FenceReceipt{Target: q.Target}, nil
@@ -184,14 +190,14 @@ func TestRootServingParkPeerPayload(t *testing.T) {
// Claim, installation, predecessor stop/start and roster/census data are fixtures;
// peer authentication, prelaunch containment, park authority and IO are real.
func TestRootServingParkAuthenticatedPeer(t *testing.T) {
- runServingParkAuthenticatedPeer(t, false)
+ runServingParkAuthenticatedPeer(t, false, false)
}
func TestRootServingParkAuthenticatedHandover(t *testing.T) {
- runServingParkAuthenticatedPeer(t, true)
+ runServingParkAuthenticatedPeer(t, true, false)
}
-func runServingParkAuthenticatedPeer(t *testing.T, handover bool) {
+func runServingParkAuthenticatedPeer(t *testing.T, handover, writer bool) {
t.Helper()
servingRootRunner(t)
pid1, err := os.ReadFile("/proc/1/comm")
@@ -281,11 +287,21 @@ func runServingParkAuthenticatedPeer(t *testing.T, handover bool) {
f.p.ledger.fs = newServingParkFS(f.f.guard.DataDir, state)
coordinator := f.f.root(nil)
coordinator.Evidence = f.f.claims
+ coordinator.Inventory = f.p.root.Journals.Inventory
production, err := NewProductionClaimStopHandler(coordinator, f.f.journal, productionStopControllers{}, f.p.root.Roots)
if err != nil {
t.Fatal(err)
}
f.p.root.Stop = production.(ClaimStopHTTP)
+ kernelResults := make(chan error, 8)
+ if writer {
+ kernel := f.p.root.Stop.Coordinator.Kernel
+ f.p.root.Stop.Coordinator.Kernel = stopKernelFunc(func(ctx context.Context, target hostguard.FenceTarget, inventory StopInventory, ack DrainAcknowledgement) (StopKernelEvidence, error) {
+ evidence, err := kernel.VerifyStopped(ctx, target, inventory, ack)
+ kernelResults <- err
+ return evidence, err
+ })
+ }
f.p.root.Reconcile.Box, f.p.root.Reconcile.Enrollment = f.actor.Box, f.actor.Enrollment
var admitted atomic.Bool
admitted.Store(true)
@@ -504,6 +520,9 @@ func runServingParkAuthenticatedPeer(t *testing.T, handover bool) {
t.Fatal("retirement changed preserved state", name, err)
}
}
+ if writer {
+ runServingPeerStoppedWriter(t, f, installed, pre.Launch, next, command, kernelResults)
+ }
}
// Transfer the serving installation on the nested mount without moving its
diff --git a/internal/host/claimservingwriter_rootrunner_linux_test.go b/internal/host/claimservingwriter_rootrunner_linux_test.go
new file mode 100644
index 000000000..bbc0add8d
--- /dev/null
+++ b/internal/host/claimservingwriter_rootrunner_linux_test.go
@@ -0,0 +1,320 @@
+//go:build linux && rootrunner
+
+package host
+
+import (
+ "bytes"
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "os/signal"
+ "path/filepath"
+ "reflect"
+ "strconv"
+ "strings"
+ "syscall"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/instancectl"
+)
+
+func TestRootServingParkStoppedWriterHandover(t *testing.T) {
+ runServingParkAuthenticatedPeer(t, true, true)
+}
+
+// Only the fixture unit starts this helper. Keep SQLite open until SIGTERM;
+// the root stop verifier must independently observe descriptor loss.
+func TestRootServingParkWriterPayload(t *testing.T) {
+ path, readyPath := os.Getenv("SERVING_WRITER_DB"), os.Getenv("SERVING_WRITER_READY")
+ if path == "" || readyPath == "" {
+ t.Fatal("invoke through TestRootServingParkStoppedWriterHandover")
+ }
+ stop := make(chan os.Signal, 1)
+ signal.Notify(stop, syscall.SIGTERM)
+ defer signal.Stop(stop)
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer db.Close()
+ conn, err := db.Conn(t.Context())
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer conn.Close()
+ for _, query := range []string{"PRAGMA journal_mode=WAL", "CREATE TABLE serving_witness(value TEXT)", "INSERT INTO serving_witness VALUES ('tenant committed bytes')"} {
+ if _, err := conn.ExecContext(t.Context(), query); err != nil {
+ t.Fatal(err)
+ }
+ }
+ start, err := claimConstructionProcessStart()
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := json.Marshal(StopPeer{PID: os.Getpid(), ProcessStart: start})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(readyPath, body, 0600); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-stop:
+ case <-t.Context().Done():
+ t.Fatal(t.Context().Err())
+ }
+}
+
+func runServingPeerStoppedWriter(t *testing.T, f *servingRootFixture, binary string, launch hostbox.LaunchInventory, peer StopPeer, command func(servingPeerCommand, bool) servingPeerReply, kernelResults <-chan error) {
+ t.Helper()
+ ctx, cancel := context.WithTimeout(t.Context(), 45*time.Second)
+ defer cancel()
+ unitName := instancectl.UnitName(instancectl.RoleRun, f.current.Host)
+ slices := []string{instancectl.SliceName(f.current.Host), "carlos-instances.slice", "carlos.slice"}
+ probe, probeErr := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", "LoadState", "--value").Output()
+ if strings.TrimSpace(string(probe)) != "not-found" {
+ t.Fatalf("refusing existing/unreadable writer unit: %q %v", probe, probeErr)
+ }
+ for _, name := range slices {
+ out, err := exec.CommandContext(ctx, "systemctl", "show", name, "-p", "ActiveState", "--value").Output()
+ if strings.TrimSpace(string(out)) != "inactive" {
+ t.Fatalf("refusing existing/unreadable fixture unit %s: %q %v", name, out, err)
+ }
+ }
+ executable, err := stopIdentity(binary)
+ if err != nil {
+ t.Fatal(err)
+ }
+ reservation := hostbox.LaunchReservation{ID: "writer:main", Role: "main", InvocationID: "writer", Inputs: "sqlite-witness", Target: launch.Target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(f.current.Host)}, Executable: hostbox.LaunchFile(executable)}
+ // Fixture unit creation and launch journal writes are explicit seams. The
+ // identity observations, stop verifier and stop transport are production.
+ if err := f.f.journal.IntendUnit(ctx, reservation); err != nil {
+ t.Fatal(err)
+ }
+ readyPath := filepath.Join(filepath.Dir(binary), "writer-ready")
+ content := fmt.Sprintf("[Unit]\nDescription=Serving park SQLite witness\n[Service]\nType=simple\nUser=root\nSlice=%s\nExecStart=%s -test.run=^TestRootServingParkWriterPayload$ -test.v -test.timeout=60s\nEnvironment=SERVING_WRITER_DB=%s SERVING_WRITER_READY=%s\nIPAddressDeny=any\nTimeoutStopSec=10s\n", slices[0], binary, f.db, readyPath)
+ unitPath := filepath.Join("/run/systemd/system", unitName)
+ unit, err := os.OpenFile(unitPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0644)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+ defer cancel()
+ for _, name := range append([]string{unitName}, slices...) {
+ if out, err := exec.CommandContext(ctx, "systemctl", "stop", name).CombinedOutput(); err != nil {
+ t.Errorf("stop fixture %s: %v %s", name, err, out)
+ }
+ }
+ if err := os.Remove(unitPath); err != nil {
+ t.Error(err)
+ }
+ if out, err := exec.CommandContext(ctx, "systemctl", "daemon-reload").CombinedOutput(); err != nil {
+ t.Errorf("reload fixture: %v %s", err, out)
+ }
+ })
+ _, writeErr := unit.WriteString(content)
+ if err := errors.Join(writeErr, unit.Close()); err != nil {
+ t.Fatal(err)
+ }
+ servingRunnerSystemctl(t, "daemon-reload")
+ servingRunnerSystemctl(t, "start", unitName)
+ var process StopPeer
+ for {
+ body, err := os.ReadFile(readyPath)
+ if err == nil && json.Unmarshal(body, &process) == nil {
+ break
+ }
+ select {
+ case <-ctx.Done():
+ t.Fatal("writer readiness", ctx.Err())
+ case <-time.After(20 * time.Millisecond):
+ }
+ }
+ properties := func(name string) string {
+ t.Helper()
+ out, err := exec.CommandContext(ctx, "systemctl", "show", unitName, "-p", name, "--value").Output()
+ if err != nil {
+ t.Fatal(err)
+ }
+ return strings.TrimSpace(string(out))
+ }
+ pid, err := strconv.Atoi(properties("MainPID"))
+ if err != nil || pid != process.PID || pid <= 0 {
+ t.Fatal("writer MainPID", pid, process.PID, err)
+ }
+ start, err := claimLaunchProcessStart("/proc", pid)
+ if err != nil || process.ProcessStart != start {
+ t.Fatalf("writer identity: %+v %v", process, err)
+ }
+ group := reservation.Group
+ group.InvocationID = properties("InvocationID")
+ if properties("ControlGroup") != "/"+group.RelativePath || group.InvocationID == "" {
+ t.Fatal("writer unit identity")
+ }
+ groupID, err := stopIdentity(filepath.Join("/sys/fs/cgroup", group.RelativePath))
+ if err != nil {
+ t.Fatal(err)
+ }
+ group.Device, group.Inode = groupID.Device, groupID.Inode
+ if err := f.f.journal.StartedUnit(ctx, reservation, group, pid, start); err != nil {
+ t.Fatal(err)
+ }
+ id, err := stopIdentity(f.db)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.f.journal.RecordFile(ctx, launch.Target, hostbox.LaunchFile(id)); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(f.db+".gen", []byte(f.current.Generation), 0600); err != nil {
+ t.Fatal(err)
+ }
+ current := f.f.read()
+ req := StopRequest{Region: f.actor.Region, Host: current.Host, Incarnation: f.current.Incarnation, InstallationStamp: f.current.Stamp, WriterSession: f.current.Session, Box: f.actor.Box, Enrollment: f.actor.Enrollment, BootID: peer.BootID, ControllerID: peer.ControllerID, ExpectedClaimETag: current.ETag, ExpectedDirectoryRevision: "7"}
+ stop := servingPeerCommand{Action: "stop", Stop: req}
+ checkNoReceipt := func() {
+ t.Helper()
+ receipts, err := f.f.guard.CompletedServingStops(ctx, current.Host, hostguard.Captured{Incarnation: f.current.Incarnation, Stamp: f.current.Stamp}, f.current.Session, f.actor.Box)
+ if !errors.Is(err, hostguard.ErrNoFenceReceipt) || len(receipts) != 0 {
+ t.Fatal("root trusted a drain acknowledgement while a writer remained")
+ }
+ }
+ command(stop, true)
+ select {
+ case err := <-kernelResults:
+ if err == nil {
+ t.Fatal("live writer escaped kernel verification")
+ }
+ default:
+ t.Fatal("stop refused before reaching the production kernel verifier")
+ }
+ checkNoReceipt()
+ servingRunnerSystemctl(t, "stop", unitName)
+ if properties("MainPID") != "0" || properties("ActiveState") != "inactive" || properties("Result") != "success" {
+ t.Fatal("writer unit did not stop cleanly")
+ }
+ if _, err := claimLaunchProcessStart("/proc", pid); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("writer process still observable: %v", err)
+ }
+ if err := f.f.journal.SettleUnit(ctx, launch.Target, reservation.ID); err != nil {
+ t.Fatal(err)
+ }
+ // Even an empty group and settled journal cannot hide an independent FD.
+ openDB, err := os.Open(f.db)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer openDB.Close()
+ command(stop, true)
+ select {
+ case err := <-kernelResults:
+ if err == nil || !strings.Contains(err.Error(), "descriptor remains") {
+ t.Fatalf("independent FD refusal: %v", err)
+ }
+ default:
+ t.Fatal("FD refusal did not reach the production kernel verifier")
+ }
+ checkNoReceipt()
+ if err := openDB.Close(); err != nil {
+ t.Fatal(err)
+ }
+ stopped := command(stop, false).Stopped
+ select {
+ case err := <-kernelResults:
+ if err != nil {
+ t.Fatal("stopped writer kernel verification", err)
+ }
+ default:
+ t.Fatal("stop succeeded without production kernel verification")
+ }
+ if err := verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: stopped.Target, Bytes: stopped.Bytes, EvidenceAt: stopped.EvidenceAt}); err != nil {
+ t.Fatal("independent stop receipt", err)
+ }
+ if replay := command(stop, false).Stopped; !reflect.DeepEqual(replay, stopped) {
+ t.Fatal("stop replay changed receipt")
+ }
+ before, err := os.ReadFile(f.db)
+ if err != nil {
+ t.Fatal(err)
+ }
+ previous := f.f.read()
+ h := previous.Claim.Holder
+ admitted, err := hostclaim.Update(ctx, f.f.claims, previous, hostclaim.AdmitWriter(hostclaim.Admission{ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: f.actor.Box, Region: f.actor.Region, PredecessorRegion: f.actor.Region, Mode: "cold", NewGeneration: "G3"}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ prior := f.current
+ f.current, err = servingWriterIdentity(admitted, f.actor.Box, f.actor.Region)
+ if err != nil {
+ t.Fatal(err)
+ }
+ command(servingPeerCommand{Action: "handover", Previous: previous, Admitted: admitted, Actor: f.actor, Target: stopped.Target}, false)
+ command(servingPeerCommand{Action: "prelaunch", Target: recoveryTuple(f.f.read(), claimStopLocal{Actor: f.actor})}, false)
+ q := servingPeerCommand{Action: "park", Current: f.current, Prior: prior}
+ parked := command(q, false).Authorization
+ f.tags(f.current.Stamp)
+ command(servingPeerCommand{Action: "complete", Current: f.current, ID: parked.ID}, false)
+ record, err := f.p.ledger.read(parked.ID)
+ if err != nil || !record.Complete || record.Disposition != servingParkRetain || len(record.Receipts) != 1 || !bytes.Equal(record.Receipts[0].Bytes, stopped.Bytes) {
+ t.Fatalf("park did not use independent stop: %+v %v", record, err)
+ }
+ parkPath := filepath.Join(f.f.guard.DataDir, "instances", ".serving-parked", parked.ID, filepath.Base(f.db))
+ after, err := os.ReadFile(parkPath)
+ if err != nil || !bytes.Equal(before, after) {
+ t.Fatal("park changed committed SQLite bytes", err)
+ }
+ if _, err := os.Lstat(f.db); !errors.Is(err, os.ErrNotExist) {
+ t.Fatal("park left live database", err)
+ }
+ if replay := command(q, false).Authorization; replay != parked {
+ t.Fatal("stopped-writer park replay changed authority")
+ }
+ for name, want := range map[string]string{"files/upload": "tenant upload", "env": "tenant configuration"} {
+ body, err := os.ReadFile(filepath.Join(filepath.Dir(f.db), name))
+ if err != nil || string(body) != want {
+ t.Fatal("writer retirement changed preserved state", name, err)
+ }
+ }
+ if err := f.p.ledger.fs.CheckParked(&record); err != nil {
+ t.Fatal(err)
+ }
+ servingRestoreAfterStoppedWriter(t, f, servingPeerRetirement{command: command})
+ check, err := f.p.ledger.read(parked.ID)
+ if err != nil || !reflect.DeepEqual(record, check) {
+ t.Fatal("stopped-writer park replay changed record", err)
+ }
+}
+
+// Adapt the fixture's control connection to the production client in the edge
+// process. Root never receives a retirement request from the test controller.
+type servingPeerRetirement struct {
+ command func(servingPeerCommand, bool) servingPeerReply
+}
+
+func (p servingPeerRetirement) Park(ctx context.Context, current, prior servingArtifactIdentity) (servingParkAuthorization, error) {
+ if err := ctx.Err(); err != nil {
+ return servingParkAuthorization{}, err
+ }
+ return p.command(servingPeerCommand{Action: "park", Current: current, Prior: prior}, false).Authorization, nil
+}
+
+func (p servingPeerRetirement) Complete(ctx context.Context, id string, current servingArtifactIdentity) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ p.command(servingPeerCommand{Action: "complete", Current: current, ID: id}, false)
+ return nil
+}
+
+func (servingPeerRetirement) ParkLegacy(context.Context, servingArtifactIdentity, servingLegacyPredecessor) (servingParkAuthorization, error) {
+ return servingParkAuthorization{}, hostbox.ErrUnsupported
+}