carlos / platform Public

activator: the instance key's adoption verb, bound to the launch's writer and published on the session (L21)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Paul Campbell pushed by claude 1 week ago 1fbc04cf8a1234075b17017fd1aca3c0f37d4063 parent 4a7ee0e

6 files changed, +593 −0

internal/activator/keyadoption.go +160 −0
internal/activator/keyadoption_linux_test.go +44 −0
internal/activator/keyadoption_test.go +122 −0
internal/host/instancekey_adoption_test.go +103 −0
internal/hostfiles/relay.go +88 −0
internal/hostfiles/relay_test.go +76 −0

CI passed — run details

CI log

=== gate ===
cd launchctx && go vet ./... && go test ./... -count=1
ok  	amadan.net/carlos/platform.git/launchctx	0.014s
?   	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	331.414s
?   	github.com/carlosframework/platform/cmd/changelog	[no test files]
ok  	github.com/carlosframework/platform/cmd/console	3.836s
ok  	github.com/carlosframework/platform/internal/activator	201.178s
ok  	github.com/carlosframework/platform/internal/agentsock	0.035s
ok  	github.com/carlosframework/platform/internal/alerting	0.006s
ok  	github.com/carlosframework/platform/internal/analytics	0.067s
ok  	github.com/carlosframework/platform/internal/appconfig	2.582s
ok  	github.com/carlosframework/platform/internal/apperr	0.162s
ok  	github.com/carlosframework/platform/internal/applog	0.326s
ok  	github.com/carlosframework/platform/internal/awsinv	0.006s
ok  	github.com/carlosframework/platform/internal/boxchoice	0.010s
ok  	github.com/carlosframework/platform/internal/boxid	0.925s
?   	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.011s
ok  	github.com/carlosframework/platform/internal/caps	0.031s
ok  	github.com/carlosframework/platform/internal/certwarm	0.003s
ok  	github.com/carlosframework/platform/internal/cgroup	0.196s
ok  	github.com/carlosframework/platform/internal/changelog	0.457s
ok  	github.com/carlosframework/platform/internal/clamd	0.074s
?   	github.com/carlosframework/platform/internal/cliflag	[no test files]
ok  	github.com/carlosframework/platform/internal/console	324.060s
?   	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.197s
ok  	github.com/carlosframework/platform/internal/dbdoor/dbdoortest	0.018s
ok  	github.com/carlosframework/platform/internal/dbverify	0.006s
ok  	github.com/carlosframework/platform/internal/dnsrec	0.004s
ok  	github.com/carlosframework/platform/internal/docsite	0.007s
ok  	github.com/carlosframework/platform/internal/domains	2.287s
ok  	github.com/carlosframework/platform/internal/doorkey	0.014s
ok  	github.com/carlosframework/platform/internal/economics	0.008s
ok  	github.com/carlosframework/platform/internal/edge	26.080s
?   	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.222s
ok  	github.com/carlosframework/platform/internal/gate	1.795s
ok  	github.com/carlosframework/platform/internal/genselect	0.046s
ok  	github.com/carlosframework/platform/internal/geo	0.019s
ok  	github.com/carlosframework/platform/internal/geoip	0.071s
ok  	github.com/carlosframework/platform/internal/host	45.833s
ok  	github.com/carlosframework/platform/internal/hostactors	0.606s
ok  	github.com/carlosframework/platform/internal/hostbox	1.494s
ok  	github.com/carlosframework/platform/internal/hostclaim	0.449s
ok  	github.com/carlosframework/platform/internal/hostevidence	0.013s
ok  	github.com/carlosframework/platform/internal/hostexec	0.326s
ok  	github.com/carlosframework/platform/internal/hostfence	0.049s
ok  	github.com/carlosframework/platform/internal/hostfiles	4.147s
ok  	github.com/carlosframework/platform/internal/hostfloor	0.024s
ok  	github.com/carlosframework/platform/internal/hostguard	0.173s
ok  	github.com/carlosframework/platform/internal/hostops	1.792s
ok  	github.com/carlosframework/platform/internal/hostquarantine	0.017s
ok  	github.com/carlosframework/platform/internal/hostresolve	0.270s
ok  	github.com/carlosframework/platform/internal/hostwrite	0.002s
ok  	github.com/carlosframework/platform/internal/infraview	0.004s
ok  	github.com/carlosframework/platform/internal/instancectl	1.713s
ok  	github.com/carlosframework/platform/internal/instanceexec	2.831s
ok  	github.com/carlosframework/platform/internal/instancefs	0.009s
ok  	github.com/carlosframework/platform/internal/instancekey	0.002s
ok  	github.com/carlosframework/platform/internal/layout	1.598s
ok  	github.com/carlosframework/platform/internal/lease	9.547s
ok  	github.com/carlosframework/platform/internal/lease/leasetest	0.007s
ok  	github.com/carlosframework/platform/internal/ledger	0.046s
ok  	github.com/carlosframework/platform/internal/mail	0.316s
ok  	github.com/carlosframework/platform/internal/maintenance	0.116s
ok  	github.com/carlosframework/platform/internal/maintenanceexec	5.603s
ok  	github.com/carlosframework/platform/internal/maintenanceguard	0.167s
ok  	github.com/carlosframework/platform/internal/metrics	0.016s
ok  	github.com/carlosframework/platform/internal/moves	21.442s
ok  	github.com/carlosframework/platform/internal/moves/archive	1.033s
ok  	github.com/carlosframework/platform/internal/objstore	0.293s
ok  	github.com/carlosframework/platform/internal/ops	0.205s
ok  	github.com/carlosframework/platform/internal/park	0.315s
ok  	github.com/carlosframework/platform/internal/privatefile	0.010s
ok  	github.com/carlosframework/platform/internal/provision	3.547s
ok  	github.com/carlosframework/platform/internal/publicrelease	0.013s
ok  	github.com/carlosframework/platform/internal/registry	3.641s
ok  	github.com/carlosframework/platform/internal/relay	7.254s
ok  	github.com/carlosframework/platform/internal/release	19.075s
ok  	github.com/carlosframework/platform/internal/rollout	1.725s
ok  	github.com/carlosframework/platform/internal/rootaws	0.014s
ok  	github.com/carlosframework/platform/internal/s3compat	0.006s
ok  	github.com/carlosframework/platform/internal/safefile	0.007s
ok  	github.com/carlosframework/platform/internal/schedule	2.433s
ok  	github.com/carlosframework/platform/internal/sim/hostlife	3.429s
ok  	github.com/carlosframework/platform/internal/sim/litestreamfacts	3.999s
ok  	github.com/carlosframework/platform/internal/status	0.061s
ok  	github.com/carlosframework/platform/internal/steering	0.006s
ok  	github.com/carlosframework/platform/internal/stopevidence	0.144s
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.006s
ok  	github.com/carlosframework/platform/internal/tunnelclient	0.004s
ok  	github.com/carlosframework/platform/internal/unixdial	0.004s
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.003s
ok  	github.com/carlosframework/platform/internal/vet	1.692s
=== 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.02s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CreateMultipartUpload (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/not_by_CopyObject_with_a_REPLACE_tagging_directive (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/but_multipart_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/and_copy_itself_is_allowed (0.00s)
        --- PASS: TestDeliveredContractMinio/the_app_cannot_tag_by_any_other_route/a_POST-policy_forged_tag_with_a_matching_version_is_still_not_a_verdict (0.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.030s
=== RUN   TestT4StorageProbeMinio
2026/10/11 17:09:16 moves: T4: STORE: list: operation error S3: ListObjectsV2, https response error StatusCode: 403, RequestID: 18DD88B94B0AF8BC, 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/internal/activator/keyadoption.go b/internal/activator/keyadoption.go

new file mode 100644

index 000000000..ce63f7c24

--- /dev/null

+++ b/internal/activator/keyadoption.go

@@ -0,0 +1,160 @@

+package activator

+// The instance key's adoption record (instance-key plan rev 3, rulings 19

+// to 22). A key in <host>.env or <host>.config.env proves only that one

+// was delivered: an app with a key of its own keeps it (rastrillo's step

+// 1), and its data is sealed under that key. So relocation accepts a

+// delivered key only when the app's helper, having resolved

+// $CARLOS_INSTANCE_KEY (step 2), told the agent so, and the agent recorded

+// it for the writer the process was admitted as. The record is a

+// trusted-helper assertion: it stops accidental loss, and a tenant that

+// forges one can lose only its own data (ruling 19).

+//

+// The verb is mounted only where a launch has a writer session: claim

+// launches, once they bind a control socket (obligation O6). A legacy

+// launch has no session to bind a record to, so nothing mounts it there and

+// the helper's call is a 404 it ignores.

+import (

+ "crypto/sha256"

+ "crypto/subtle"

+ "encoding/json"

+ "errors"

+ "net/http"

+ "os"

+ "path/filepath"

+ "github.com/carlosframework/platform/internal/instancefs"

+ "github.com/carlosframework/platform/internal/instancekey"

+ "github.com/carlosframework/platform/internal/registry"

+ "github.com/carlosframework/platform/internal/safefile"

+// instanceKeyAdoptionPath is the verb's path on the control socket.

+const instanceKeyAdoptionPath = "/instance-key/adoption"

+// AdoptionLaunch is one process's launch state as the agent holds it: the

+// writer it was admitted as, the key its env delivered (by digest) and the

+// file that supplied it, where its <host>.env is, and the session's files

+// publisher. O6 builds one per claim launch from the admission (holder

+// incarnation, session, generation) and ClaimLaunchSnapshot's

+// InstanceKeySHA256 and InstanceKeySource. Nothing in it comes from a

+// request.

+type AdoptionLaunch struct {

+ Writer instancekey.Binding

+ KeySHA256 string

+ Source string // instancekey.SourcePlatform or SourceConfig

+ Dir string // the directory that holds <host>.env

+ Beneath bool // Dir is tenant-writable: write through instancefs

+ // Publish hands the record to the session's files publisher as the

+ // platform item instancekey.AdoptionItem (hostfiles.PlatformRelay). It

+ // must not block: publication follows the publisher's continuous bound.

+ Publish func(record []byte)

+// MountInstanceKeyAdoption adds PUT /instance-key/adoption to an

+// instance's control socket. host is the listener's, never the request's,

+// and the bearer is the host's own CARLOS_ADMIN_TOKEN, as for the schedule

+// verbs. launch is the calling process's launch state; false (a launch with

+// no writer session) answers 404, as an agent without the verb does.

+//

+// The body is {"sha256": "<hex>", "source": "platform"|"config"}: the

+// SHA-256 of the key the helper resolved at step 2 and where the helper

+// says it came from. The answer:

+// - 204 once the record for the launch's writer is on disk, 0600, file

+// and directory fsynced, and handed to the session's publisher;

+// - 409 when the digest or the source is not the launch's (the app holds

+// some other key), or for a replica row (ruling 26); nothing recorded;

+// - 404 with no launch state, 400 for a malformed body (a session or

+// generation in it included), 401 without the bearer, 500 when the

+// durable write fails (nothing published).

+//

+// Bodies are http.StatusText only: the helper treats every failure as

+// non-fatal, and the status is all it reads.

+func (b *ExecBackend) MountInstanceKeyAdoption(mux *http.ServeMux, rt registry.Route, token func() (string, error), launch func() (AdoptionLaunch, bool)) {

+ fail := func(w http.ResponseWriter, code int) { http.Error(w, http.StatusText(code), code) }

+ mux.HandleFunc("PUT "+instanceKeyAdoptionPath, func(w http.ResponseWriter, r *http.Request) {

+ want, err := token()

+ if err != nil || want == "" {

+ fail(w, http.StatusUnauthorized)

+ return

+ }

+ got := sha256.Sum256([]byte(r.Header.Get("Authorization")))

+ exp := sha256.Sum256([]byte("Bearer " + want))

+ if subtle.ConstantTimeCompare(got[:], exp[:]) != 1 {

+ fail(w, http.StatusUnauthorized)

+ return

+ }

+ var body struct {

+ SHA256 string `json:"sha256"`

+ Source string `json:"source"`

+ }

+ dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1024))

+ dec.DisallowUnknownFields()

+ if err := dec.Decode(&body); err != nil || dec.More() || !instancekey.ValidDigest(body.SHA256) ||

+ (body.Source != instancekey.SourcePlatform && body.Source != instancekey.SourceConfig) {

+ fail(w, http.StatusBadRequest)

+ return

+ }

+ l, ok := launch()

+ switch {

+ case !ok:

+ fail(w, http.StatusNotFound)

+ return

+ case rt.Replica:

+ fail(w, http.StatusConflict)

+ return

+ case l.Writer.Host != rt.Host || l.Publish == nil:

+ fail(w, http.StatusInternalServerError)

+ return

+ case body.SHA256 != l.KeySHA256 || body.Source != l.Source:

+ fail(w, http.StatusConflict)

+ return

+ }

+ record := instancekey.AdoptionRecord(l.Writer, l.Source, l.KeySHA256)

+ if err := writeAdoptionRecord(l, rt.Host, record); err != nil {

+ fail(w, http.StatusInternalServerError)

+ return

+ }

+ l.Publish(record)

+ w.WriteHeader(http.StatusNoContent)

+ })

+// adoptionRecordSync is the nested record's directory sync, a seam for

+// tests. Read and replaced only under adminTokenMu.

+var adoptionRecordSync = syncDir

+// writeAdoptionRecord writes record beside <host>.env: atomically, 0600,

+// the file and its directory fsynced before it returns. An identical

+// record already there is made durable and not rewritten. It holds

+// adminTokenMu, the lock every write beside <host>.env takes.

+func writeAdoptionRecord(l AdoptionLaunch, host string, record []byte) error {

+ name := host + instancekey.AdoptionSuffix

+ adminTokenMu.Lock()

+ defer adminTokenMu.Unlock()

+ var cur []byte

+ var err error

+ if l.Beneath {

+ cur, err = instancefs.ReadBeneathLimit(l.Dir, name, 1<<16)

+ } else {

+ cur, err = os.ReadFile(filepath.Join(l.Dir, name))

+ }

+ switch {

+ case err == nil && string(cur) == string(record):

+ return adoptionRecordSync(l.Dir)

+ case err != nil && !errors.Is(err, os.ErrNotExist):

+ return err

+ }

+ if l.Beneath {

+ // WriteAtomic fsyncs the file before its rename; the directory

+ // sync makes the rename durable.

+ if err := instancefs.WriteAtomic(l.Dir, name, record, 0o600); err != nil {

+ return err

+ }

+ return adoptionRecordSync(l.Dir)

+ }

+ // safefile.WriteFile fsyncs the file, renames it into place and fsyncs

+ // the directory.

+ return safefile.WriteFile(filepath.Join(l.Dir, name), record, 0o600)

diff --git a/internal/activator/keyadoption_linux_test.go b/internal/activator/keyadoption_linux_test.go

new file mode 100644

index 000000000..a891e0913

--- /dev/null

+++ b/internal/activator/keyadoption_linux_test.go

@@ -0,0 +1,44 @@

+package activator

+import (

+ "errors"

+ "net/http"

+ "os"

+ "path/filepath"

+ "strings"

+ "testing"

+ "github.com/carlosframework/platform/internal/instancekey"

+ "github.com/carlosframework/platform/internal/registry"

+// Ruling 22 (R-C): 204 means the record is durable on disk. A nested

+// record whose directory sync fails answers 500 and publishes nothing; the

+// next report writes it and publishes it.

+func TestInstanceKeyAdoptionIsDurableBeforeItAnswers(t *testing.T) {

+ tok := strings.Repeat("f", 64)

+ key := strings.Repeat("K", 43) + "="

+ rt := registry.Route{Host: "shop.example.com", Kind: "instance"}

+ f := newAdoptionFixture(t, rt, tok, key, instancekey.SourcePlatform)

+ f.launch.Beneath = true

+ adminTokenMu.Lock()

+ old := adoptionRecordSync

+ adoptionRecordSync = func(string) error { return errors.New("injected: EIO") }

+ adminTokenMu.Unlock()

+ if got := f.put(tok, adoptionBody(key, "platform")); got != http.StatusInternalServerError {

+ t.Fatalf("a failing directory sync: %d, want 500", got)

+ }

+ if len(f.published) != 0 {

+ t.Fatal("a record that is not durable was published")

+ }

+ adminTokenMu.Lock()

+ adoptionRecordSync = old

+ adminTokenMu.Unlock()

+ if got := f.put(tok, adoptionBody(key, "platform")); got != http.StatusNoContent {

+ t.Fatalf("retry: %d, want 204", got)

+ }

+ rec, err := os.ReadFile(filepath.Join(f.launch.Dir, rt.Host+instancekey.AdoptionSuffix))

+ if err != nil || len(f.published) != 1 || string(rec) != string(f.published[0]) {

+ t.Fatalf("after the retry: record %q (%v), %d published", rec, err, len(f.published))

+ }

diff --git a/internal/activator/keyadoption_test.go b/internal/activator/keyadoption_test.go

new file mode 100644

index 000000000..7013744b0

--- /dev/null

+++ b/internal/activator/keyadoption_test.go

@@ -0,0 +1,122 @@

+package activator

+import (

+ "net/http"

+ "net/http/httptest"

+ "os"

+ "path/filepath"

+ "strings"

+ "testing"

+ "github.com/carlosframework/platform/internal/instancekey"

+ "github.com/carlosframework/platform/internal/registry"

+// adoptionFixture is one claim-launched process's control socket, served

+// in memory: its launch state, and every record handed to the publisher.

+type adoptionFixture struct {

+ mux *http.ServeMux

+ launch AdoptionLaunch

+ hasLaunch bool

+ published [][]byte

+func newAdoptionFixture(t *testing.T, rt registry.Route, tok, key, source string) *adoptionFixture {

+ t.Helper()

+ f := &adoptionFixture{hasLaunch: true}

+ f.launch = AdoptionLaunch{

+ Writer: instancekey.Binding{Host: rt.Host, Incarnation: "inc-1", Session: 3, Generation: "g3"},

+ KeySHA256: instancekey.Digest(key), Source: source, Dir: t.TempDir(),

+ Publish: func(rec []byte) { f.published = append(f.published, rec) },

+ }

+ f.mux = http.NewServeMux()

+ (&ExecBackend{}).MountInstanceKeyAdoption(f.mux, rt, func() (string, error) { return tok, nil },

+ func() (AdoptionLaunch, bool) { return f.launch, f.hasLaunch })

+ return f

+func (f *adoptionFixture) put(token, body string) int {

+ req := httptest.NewRequest(http.MethodPut, instanceKeyAdoptionPath, strings.NewReader(body))

+ if token != "" {

+ req.Header.Set("Authorization", "Bearer "+token)

+ }

+ w := httptest.NewRecorder()

+ f.mux.ServeHTTP(w, req)

+ return w.Code

+func adoptionBody(key, source string) string {

+ return `{"sha256":"` + instancekey.Digest(key) + `","source":"` + source + `"}`

+// Rulings 20 and 21: the agent records the launch's own writer and the key

+// the launch delivered, from its own launch state. A request cannot name a

+// writer, a digest or a source the launch did not have.

+func TestInstanceKeyAdoptionRecordsTheLaunchesWriter(t *testing.T) {

+ tok := strings.Repeat("f", 64)

+ key := strings.Repeat("K", 43) + "="

+ rt := registry.Route{Host: "shop.example.com", Kind: "instance"}

+ f := newAdoptionFixture(t, rt, tok, key, instancekey.SourcePlatform)

+ recPath := filepath.Join(f.launch.Dir, rt.Host+instancekey.AdoptionSuffix)

+ for _, tc := range []struct {

+ name, token, body string

+ want int

+ }{

+ {"no bearer", "", adoptionBody(key, "platform"), http.StatusUnauthorized},

+ {"another host's bearer", strings.Repeat("0", 64), adoptionBody(key, "platform"), http.StatusUnauthorized},

+ {"not a digest", tok, `{"sha256":"abc","source":"platform"}`, http.StatusBadRequest},

+ {"an unknown source", tok, adoptionBody(key, "own"), http.StatusBadRequest},

+ {"a session in the request", tok, `{"sha256":"` + instancekey.Digest(key) + `","source":"platform","session":9}`, http.StatusBadRequest},

+ {"another key", tok, adoptionBody("some-other-key", "platform"), http.StatusConflict},

+ {"another source", tok, adoptionBody(key, "config"), http.StatusConflict},

+ } {

+ if got := f.put(tc.token, tc.body); got != tc.want {

+ t.Fatalf("%s: %d, want %d", tc.name, got, tc.want)

+ }

+ if _, err := os.Stat(recPath); !os.IsNotExist(err) || len(f.published) != 0 {

+ t.Fatalf("%s: a record was written or published", tc.name)

+ }

+ want := instancekey.AdoptionRecord(f.launch.Writer, instancekey.SourcePlatform, instancekey.Digest(key))

+ for i := 0; i < 2; i++ { // the helper reports on every process start

+ if got := f.put(tok, adoptionBody(key, "platform")); got != http.StatusNoContent {

+ t.Fatalf("report %d: %d, want 204", i, got)

+ }

+ rec, err := os.ReadFile(recPath)

+ if err != nil || string(rec) != string(want) {

+ t.Fatalf("report %d: record = %q, %v", i, rec, err)

+ }

+ if st, _ := os.Stat(recPath); st.Mode().Perm() != 0o600 {

+ t.Fatalf("record mode = %o, want 600", st.Mode().Perm())

+ }

+ if len(f.published) != i+1 || string(f.published[i]) != string(want) {

+ t.Fatalf("report %d: published %d records", i, len(f.published))

+ }

+ // A configured key is recorded the same way, with its source (ruling 20).

+ cfg := newAdoptionFixture(t, rt, tok, "from-config", instancekey.SourceConfig)

+ if got := cfg.put(tok, adoptionBody("from-config", "config")); got != http.StatusNoContent {

+ t.Fatalf("configured key: %d, want 204", got)

+ }

+ if a, err := instancekey.ParseAdoption(cfg.published[0]); err != nil || a.Source != instancekey.SourceConfig || !a.Binds(cfg.launch.Writer) {

+ t.Fatalf("configured key's record = %+v, %v", a, err)

+ }

+ // No launch state (a legacy launch) is the 404 an agent without the

+ // verb answers; a replica refuses (ruling 26).

+ none := newAdoptionFixture(t, rt, tok, key, instancekey.SourcePlatform)

+ none.hasLaunch = false

+ replicaRT := rt

+ replicaRT.Replica = true

+ replica := newAdoptionFixture(t, replicaRT, tok, key, instancekey.SourcePlatform)

+ if got := none.put(tok, adoptionBody(key, "platform")); got != http.StatusNotFound {

+ t.Fatalf("no launch state: %d, want 404", got)

+ }

+ if got := replica.put(tok, adoptionBody(key, "platform")); got != http.StatusConflict {

+ t.Fatalf("replica: %d, want 409", got)

+ }

+ if len(none.published)+len(replica.published) != 0 {

+ t.Fatal("a record was published without a writer")

+ }

diff --git a/internal/host/instancekey_adoption_test.go b/internal/host/instancekey_adoption_test.go

new file mode 100644

index 000000000..92741c836

--- /dev/null

+++ b/internal/host/instancekey_adoption_test.go

@@ -0,0 +1,103 @@

+package host

+import (

+ "context"

+ "encoding/json"

+ "net/http"

+ "net/http/httptest"

+ "os"

+ "path/filepath"

+ "strings"

+ "testing"

+ "time"

+ "github.com/carlosframework/platform/internal/activator"

+ "github.com/carlosframework/platform/internal/hostfiles"

+ "github.com/carlosframework/platform/internal/instancekey"

+ "github.com/carlosframework/platform/internal/registry"

+ "github.com/carlosframework/platform/internal/release"

+// Ruling 22 (R-C): an adoption is published by the running session's

+// publisher, with no admission or release after it. The box is then lost:

+// every process and every local file is gone. The data bucket's manifest

+// for the session's generation still holds the record, bound to the

+// session's writer, which is what O2' pins.

+func TestAdoptionSurvivesBoxLossWithoutAnotherAdmissionOrRelease(t *testing.T) {

+ ctx, cancel := context.WithCancel(context.Background())

+ defer cancel()

+ data := release.NewDirStore(t.TempDir())

+ writer := instancekey.Binding{Host: "shop.example.com", Incarnation: "inc-1", Session: 3, Generation: "g3"}

+ prefix := "instances/" + writer.Host + "/" + writer.Incarnation + "/files/" + writer.Generation + "/"

+ pub, err := hostfiles.Open(ctx, data, prefix)

+ if err != nil {

+ t.Fatal(err)

+ }

+ box := t.TempDir()

+ filesDir, envDir := filepath.Join(box, "files"), filepath.Join(box, "env")

+ for _, d := range []string{filesDir, envDir} {

+ if err := os.MkdirAll(d, 0o700); err != nil {

+ t.Fatal(err)

+ }

+ // The session's continuous publisher and the adoption relay.

+ running := make(chan error, 2)

+ go func() {

+ running <- hostfiles.Run(ctx, pub, filesDir, hostfiles.Options{Deadline: 40 * time.Millisecond, Scan: time.Minute, Poll: 40 * time.Millisecond})

+ }()

+ relay := &hostfiles.PlatformRelay{Name: instancekey.AdoptionItem, Stage: pub, Retry: 10 * time.Millisecond}

+ go func() { running <- relay.Run(ctx) }()

+ key := strings.Repeat("K", 43) + "="

+ tok := strings.Repeat("f", 64)

+ launch := activator.AdoptionLaunch{Writer: writer, KeySHA256: instancekey.Digest(key), Source: instancekey.SourcePlatform, Dir: envDir, Publish: relay.Publish}

+ mux := http.NewServeMux()

+ (&activator.ExecBackend{}).MountInstanceKeyAdoption(mux, registry.Route{Host: writer.Host, Kind: "instance"},

+ func() (string, error) { return tok, nil }, func() (activator.AdoptionLaunch, bool) { return launch, true })

+ req := httptest.NewRequest(http.MethodPut, "/instance-key/adoption", strings.NewReader(`{"sha256":"`+instancekey.Digest(key)+`","source":"platform"}`))

+ req.Header.Set("Authorization", "Bearer "+tok)

+ w := httptest.NewRecorder()

+ mux.ServeHTTP(w, req)

+ if w.Code != http.StatusNoContent {

+ t.Fatalf("adoption = %d, want 204", w.Code)

+ }

+ // Within the publisher's continuous bound the record is committed.

+ readRecord := func() ([]byte, error) {

+ raw, _, err := data.Get(context.Background(), hostfiles.ManifestKey(prefix))

+ if err != nil {

+ return nil, err

+ }

+ var m hostfiles.Manifest

+ if err := json.Unmarshal(raw, &m); err != nil {

+ return nil, err

+ }

+ return hostfiles.ReadPlatform(context.Background(), data, prefix, m, instancekey.AdoptionItem, nil)

+ }

+ deadline := time.Now().Add(2 * time.Second)

+ for {

+ if _, err := readRecord(); err == nil {

+ break

+ }

+ if time.Now().After(deadline) {

+ t.Fatal("the adoption was not published within the continuous bound")

+ }

+ time.Sleep(10 * time.Millisecond)

+ }

+ // Box loss: the processes stop and the disk is gone.

+ cancel()

+ <-running

+ if err := os.RemoveAll(box); err != nil {

+ t.Fatal(err)

+ }

+ rec, err := readRecord()

+ if err != nil {

+ t.Fatalf("after box loss: %v", err)

+ }

+ a, err := instancekey.ParseAdoption(rec)

+ if err != nil || !a.Binds(writer) || a.SHA256 != instancekey.Digest(key) || a.Source != instancekey.SourcePlatform {

+ t.Fatalf("after box loss the bucket holds %+v (%v), want the session's record", a, err)

+ }

diff --git a/internal/hostfiles/relay.go b/internal/hostfiles/relay.go

new file mode 100644

index 000000000..cd10b7cb7

--- /dev/null

+++ b/internal/hostfiles/relay.go

@@ -0,0 +1,88 @@

+package hostfiles

+import (

+ "context"

+ "log"

+ "sync"

+ "time"

+// PlatformStager is the part of a session's Publisher a PlatformRelay

+// needs. *Publisher implements it.

+type PlatformStager interface {

+ StagePlatform(ctx context.Context, name string, body []byte) error

+// PlatformRelay hands one platform item to a running session's publisher

+// without making its caller wait on the store (instance-key plan ruling

+// 22). Publish records the latest body and returns at once; Run stages it

+// with StagePlatform, retrying until it is staged, and the publisher's own

+// continuous loop (Run in run.go) commits it within its deadline, as it

+// commits every staged change. A body staged once is not staged again

+// unless Publish is called with it again.

+type PlatformRelay struct {

+ Name string

+ Stage PlatformStager

+ Retry time.Duration // between failed stages; 0 is one second

+ mu sync.Mutex

+ pending []byte

+ wake chan struct{}

+func (r *PlatformRelay) signal() chan struct{} {

+ r.mu.Lock()

+ defer r.mu.Unlock()

+ if r.wake == nil {

+ r.wake = make(chan struct{}, 1)

+ }

+ return r.wake

+// Publish queues body as the item's next value. It never blocks.

+func (r *PlatformRelay) Publish(body []byte) {

+ w := r.signal()

+ r.mu.Lock()

+ r.pending = append([]byte(nil), body...)

+ r.mu.Unlock()

+ select {

+ case w <- struct{}{}:

+ default:

+ }

+// Run stages queued bodies until ctx ends, and returns ctx.Err().

+func (r *PlatformRelay) Run(ctx context.Context) error {

+ retry := r.Retry

+ if retry <= 0 {

+ retry = time.Second

+ }

+ w := r.signal()

+ for {

+ r.mu.Lock()

+ body := r.pending

+ r.mu.Unlock()

+ if body != nil {

+ if err := r.Stage.StagePlatform(ctx, r.Name, body); err != nil {

+ log.Printf("hostfiles: stage platform item %s: %v (retrying)", r.Name, err)

+ select {

+ case <-ctx.Done():

+ return ctx.Err()

+ case <-time.After(retry):

+ }

+ continue

+ }

+ r.mu.Lock()

+ if string(r.pending) == string(body) {

+ r.pending = nil

+ }

+ r.mu.Unlock()

+ continue

+ }

+ select {

+ case <-ctx.Done():

+ return ctx.Err()

+ case <-w:

+ }

diff --git a/internal/hostfiles/relay_test.go b/internal/hostfiles/relay_test.go

new file mode 100644

index 000000000..73a886c99

--- /dev/null

+++ b/internal/hostfiles/relay_test.go

@@ -0,0 +1,76 @@

+package hostfiles

+import (

+ "context"

+ "errors"

+ "sync"

+ "testing"

+ "time"

+type flakyStager struct {

+ mu sync.Mutex

+ fail int

+ staged [][]byte

+ block chan struct{}

+func (f *flakyStager) StagePlatform(_ context.Context, name string, body []byte) error {

+ if f.block != nil {

+ <-f.block

+ }

+ f.mu.Lock()

+ defer f.mu.Unlock()

+ if f.fail > 0 {

+ f.fail--

+ return errors.New("store unreachable")

+ }

+ f.staged = append(f.staged, append([]byte(nil), body...))

+ return nil

+func (f *flakyStager) last() []byte {

+ f.mu.Lock()

+ defer f.mu.Unlock()

+ if len(f.staged) == 0 {

+ return nil

+ }

+ return f.staged[len(f.staged)-1]

+// Ruling 22: Publish never waits on the store; Run stages the latest body,

+// retrying a failed stage, and stages nothing again once it has.

+func TestPlatformRelayStagesTheLatestBodyWithoutBlockingItsCaller(t *testing.T) {

+ st := &flakyStager{fail: 2, block: make(chan struct{})}

+ r := &PlatformRelay{Name: "instance-key-adoption.json", Stage: st, Retry: 10 * time.Millisecond}

+ ctx, cancel := context.WithCancel(context.Background())

+ defer cancel()

+ done := make(chan error, 1)

+ go func() { done <- r.Run(ctx) }()

+ published := make(chan struct{})

+ go func() { r.Publish([]byte("one")); r.Publish([]byte("two")); close(published) }()

+ select {

+ case <-published:

+ case <-time.After(time.Second):

+ t.Fatal("Publish waited on a store that does not answer")

+ }

+ close(st.block)

+ deadline := time.Now().Add(2 * time.Second)

+ for string(st.last()) != "two" {

+ if time.Now().After(deadline) {

+ t.Fatalf("staged %q, want the latest body after two failed stages", st.last())

+ }

+ time.Sleep(5 * time.Millisecond)

+ }

+ time.Sleep(50 * time.Millisecond)

+ st.mu.Lock()

+ n := len(st.staged)

+ st.mu.Unlock()

+ if n != 1 {

+ t.Fatalf("staged %d times, want once", n)

+ }

+ cancel()

+ if err := <-done; !errors.Is(err, context.Canceled) {

+ t.Fatalf("Run = %v", err)

+ }