d52ea9a97 · carlos/platform

diff --git a/docs/site/cli.md b/docs/site/cli.md

index cf69099d3..3c52173f3 100644

--- a/docs/site/cli.md

+++ b/docs/site/cli.md

@@ -1919,6 +1919,10 @@ image before starting `apply`, `recover`, `edge` or `console`. Once armed,

the local floor remains enforced independently of `CARLOS_HOST_CLAIM`.

The member command remains `carlos ops host resolve`.

+#### carlos system journal-migrate

+

+Before starting claim-enabled services on a box with legacy journals, stop the edge and agent. Run `carlos system journal-migrate` as root with `CARLOS_HOST_CLAIM=1` and the deployment environment loaded. The command imports journals only when signed root stop evidence verifies them. It leaves the original files in place. After migration, root ignores changed legacy files and logs an alert. Keep the migration marker.

+

#### Deployment-wide object scanning

```sh

diff --git a/internal/console/claim_mail_sql.go b/internal/console/claim_mail_sql.go

index 01b7174f5..d18b3cd03 100644

--- a/internal/console/claim_mail_sql.go

+++ b/internal/console/claim_mail_sql.go

@@ -31,6 +31,9 @@ func (s *Store) withMailAppSQL(ctx context.Context, sqid, app, action string, ca

if s.gate == nil || s.bound != nil {

return call(s)

}

+ if s.outerInsideView() {

+ return errOuterStoreInView

+ }

if !release.ValidAccount(sqid) || !release.ValidApp(app) {

return hostops.ErrIdentity

}

diff --git a/internal/console/claim_runtime.go b/internal/console/claim_runtime.go

index cc073de4f..4f1fbcd5e 100644

--- a/internal/console/claim_runtime.go

+++ b/internal/console/claim_runtime.go

@@ -218,16 +218,20 @@ func (s *Server) sweepOneClaimAdmission(ctx context.Context, app hostops.App) (r

sqlErrors = append(sqlErrors, databaseErr)

}

- if databaseErr != nil {

+ // An identity error here means the file was replaced under this

+ // running console (review m-4): no effect can settle against it, so

+ // each is alerted exactly as one bound to a restored-away database.

+ replacedUnderUs := errors.Is(databaseErr, hostops.ErrIdentity)

+ if databaseErr != nil && !replacedUnderUs {

continue

}

- if e.Inputs["sql_database"] != database {

+ if replacedUnderUs || e.Inputs["sql_database"] != database {

// Bound to a receipt database this console no longer has (a

// restore or replacement changed its identity). Never inferred:

// it stays unsettled, visible and alerted until an operator

// takes the restore runbook's break-glass path.

sqlErrors = append(sqlErrors, fmt.Errorf("%w: %s/%s effect %s", ErrSQLDatabaseReplaced, app.Account, app.Name, id))

- s.alertSQLDatabaseReplaced(app.Account, app.Name, id)

+ s.alertSQLDatabaseReplaced(ctx, app.Account, app.Name, id)

continue

}

if err := c.Control.RecoverSQLTransaction(ctx, app.Account, app.Name, appID, id, s.store); err != nil {

@@ -286,10 +290,14 @@ func sqlReplacedRule(account, app, effect string) string {

return "replaced-database " + account + "/" + app + "/" + effect

}

+// sqlAlertSendTimeout bounds all keymail sends for one effect's alert, so a

+// stalled mail service cannot hold the serial sweep for long.

+const sqlAlertSendTimeout = 15 * time.Second

+

// alertSQLDatabaseReplaced tells operators once per effect, through the same

// alert memory as box alerts: keymail where a grant exists, otherwise a log

-// line.

-func (s *Server) alertSQLDatabaseReplaced(account, app, effect string) {

+// line. Sends run under the sweep's context.

+func (s *Server) alertSQLDatabaseReplaced(ctx context.Context, account, app, effect string) {

rule := sqlReplacedRule(account, app, effect)

subject := fmt.Sprintf("[carlos] SQL effect %s on %s/%s needs the restore break-glass path", effect, account, app)

body := fmt.Sprintf("SQL effect %s on %s/%s is bound to a console database that has been replaced. It cannot be settled automatically.\nFollow the break-glass section of the console restore runbook.\n", effect, account, app)

@@ -297,10 +305,12 @@ func (s *Server) alertSQLDatabaseReplaced(account, app, effect string) {

if s.sendApp != nil {

r, err := s.alertRecipients()

if err != nil {

+ // Possibly transient: record nothing, so the next sweep tries

+ // again rather than filing the alert as undeliverable for good.

log.Printf("console: claim SQL alert: recipients: %v", err)

- } else {

- run = r

+ return

}

+ run = r

}

if run == nil || len(run.to) == 0 {

if s.alertDue(sqlReceiptAlertBox, rule, noMailboxAddress, true) {

@@ -309,14 +319,15 @@ func (s *Server) alertSQLDatabaseReplaced(account, app, effect string) {

}

return

}

- run.noteMissing()

+ ctx, cancel := context.WithTimeout(ctx, sqlAlertSendTimeout)

+ defer cancel()

for _, op := range run.to {

if !s.alertDue(sqlReceiptAlertBox, rule, op.address, true) {

continue

}

- ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)

+ // Only when there is something to deliver, as in alertPass.

+ run.noteMissing()

_, err := s.sendApp(ctx, op.token, op.address, subject, body)

- cancel()

if err != nil {

log.Printf("console: claim SQL alert: keymail to %s: %v", op.address, err)

continue

diff --git a/internal/console/claim_sql_auto.go b/internal/console/claim_sql_auto.go

index cb6ae1d02..43c13ada5 100644

--- a/internal/console/claim_sql_auto.go

+++ b/internal/console/claim_sql_auto.go

@@ -15,6 +15,9 @@ func (s *Store) withAutoAppSQL(ctx context.Context, accountID int64, app, action

if app == "" || s.gate == nil || s.bound != nil {

return call(s)

}

+ if s.outerInsideView() {

+ return errOuterStoreInView

+ }

if s.claimSQLAdmit == nil || s.claimBucket == nil {

return hostops.ErrContract

}

diff --git a/internal/console/claim_sql_grants.go b/internal/console/claim_sql_grants.go

index 85d6593f7..a00cbbc50 100644

--- a/internal/console/claim_sql_grants.go

+++ b/internal/console/claim_sql_grants.go

@@ -55,6 +55,9 @@ func (s *Store) WithAppMutation(ctx context.Context, grants []sqlGrant, fn func(

if s.bound != nil {

return fmt.Errorf("%w: nested WithAppMutation", ErrClaimSQLScope)

}

+ if s.outerInsideView() {

+ return errOuterStoreInView

+ }

byScope := map[claimSQLScope][]sqlGrant{}

scopes := make([]claimSQLScope, 0, len(grants))

for _, g := range grants {

diff --git a/internal/console/claim_sql_outer.go b/internal/console/claim_sql_outer.go

new file mode 100644

index 000000000..3e6b7dc72

--- /dev/null

+++ b/internal/console/claim_sql_outer.go

@@ -0,0 +1,97 @@

+package console

+

+import (

+ "bytes"

+ "context"

+ "database/sql"

+ "fmt"

+ "runtime"

+ "strconv"

+)

+

+// errOuterStoreInView is a call on the outer, captured Store from inside the

+// callback withClaimSQL is running (review m-5). The view holds the pool's

+// only connection, so the call could only wait for its own caller; it is

+// refused at once instead, before any remote check.

+var errOuterStoreInView = fmt.Errorf("%w: the outer Store was used inside withClaimSQL; use the view it passes", ErrClaimSQLScope)

+

+// goroutineID is the running goroutine's number. It is used only to tell

+// whether the caller is the goroutine that holds the bound view; a call from

+// any other goroutine waits for the connection as before.

+func goroutineID() uint64 {

+ var buf [64]byte

+ b := bytes.TrimPrefix(buf[:runtime.Stack(buf[:], false)], []byte("goroutine "))

+ if i := bytes.IndexByte(b, ' '); i > 0 {

+ b = b[:i]

+ }

+ id, _ := strconv.ParseUint(string(b), 10, 64)

+ return id

+}

+

+// insideView reports whether the calling goroutine holds the bound view.

+func (g *claimGate) insideView() bool {

+ h := g.viewG.Load()

+ return h != 0 && h == goroutineID()

+}

+

+// outerInsideView reports a call on the outer Store from the goroutine that

+// holds its bound view. A goroutine fn starts is not caught: it waits.

+func (s *Store) outerInsideView() bool {

+ return s.gate != nil && s.bound == nil && s.gate.insideView()

+}

+

+// claimOuterDB is the gated Store's own handle. It is the pool, except that

+// the goroutine holding the bound view is refused instead of waiting on the

+// connection it already holds.

+type claimOuterDB struct {

+ *sql.DB

+ g *claimGate

+}

+

+var _ sqlHandle = claimOuterDB{}

+

+// refusedContext is already done with errOuterStoreInView, so database/sql

+// refuses before it waits for a connection. It is how QueryRow, which has no

+// error return, carries the refusal to Scan.

+type refusedContext struct{ context.Context }

+

+var closedDone = func() chan struct{} { c := make(chan struct{}); close(c); return c }()

+

+func (refusedContext) Done() <-chan struct{} { return closedDone }

+func (refusedContext) Err() error { return errOuterStoreInView }

+

+func (o claimOuterDB) guard(ctx context.Context) context.Context {

+ if o.g.insideView() {

+ return refusedContext{ctx}

+ }

+ return ctx

+}

+

+func (o claimOuterDB) Exec(query string, args ...any) (sql.Result, error) {

+ return o.ExecContext(context.Background(), query, args...)

+}

+func (o claimOuterDB) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {

+ return o.DB.ExecContext(o.guard(ctx), query, args...)

+}

+func (o claimOuterDB) Query(query string, args ...any) (*sql.Rows, error) {

+ return o.QueryContext(context.Background(), query, args...)

+}

+func (o claimOuterDB) QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) {

+ return o.DB.QueryContext(o.guard(ctx), query, args...)

+}

+func (o claimOuterDB) QueryRow(query string, args ...any) *sql.Row {

+ return o.QueryRowContext(context.Background(), query, args...)

+}

+func (o claimOuterDB) QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row {

+ return o.DB.QueryRowContext(o.guard(ctx), query, args...)

+}

+func (o claimOuterDB) Begin() (*sql.Tx, error) { return o.BeginTx(context.Background(), nil) }

+func (o claimOuterDB) BeginTx(ctx context.Context, opts *sql.TxOptions) (*sql.Tx, error) {

+ return o.DB.BeginTx(o.guard(ctx), opts)

+}

+func (o claimOuterDB) Prepare(query string) (*sql.Stmt, error) {

+ return o.PrepareContext(context.Background(), query)

+}

+func (o claimOuterDB) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {

+ return o.DB.PrepareContext(o.guard(ctx), query)

+}

diff --git a/internal/console/claim_sql_receipts.go b/internal/console/claim_sql_receipts.go

index fcc0d1ed9..242c0520e 100644

--- a/internal/console/claim_sql_receipts.go

+++ b/internal/console/claim_sql_receipts.go

@@ -10,6 +10,7 @@ import (

"errors"

"fmt"

"github.com/carlosframework/platform/internal/hostops"

+ "log"

"time"

)

@@ -49,9 +50,25 @@ func initSQLReceipts(db *sql.DB, instance string) error {

if err = db.QueryRow(`SELECT count(*), COALESCE(MIN(instance), '') FROM claim_sql_identity`).Scan(&rows, &stored); err != nil {

return err

}

- if rows != 1 || stored != instance {

- return fmt.Errorf("claim SQL receipt database belongs to instance %q, not %q: %w", stored, instance, hostops.ErrIdentity)

+ if rows != 1 {

+ return fmt.Errorf("claim SQL receipt database has %d identity rows: %w", rows, hostops.ErrIdentity)

}

+ if stored == instance {

+ return nil

+ }

+ // Restored onto another instance (review I2-2): adopt it under a fresh

+ // nonce, so this copy has a database ID no other copy has had, even one

+ // moved back to an instance it lived on before. Copied receipts still

+ // cannot settle the original's effects, because every settlement compares

+ // the effect's database ID first; those effects stay unsettled and are

+ // alerted past sql_recover_after.

+ if _, err = rand.Read(nonce[:]); err != nil {

+ return err

+ }

+ if _, err = db.Exec(`UPDATE claim_sql_identity SET id = ?, instance = ?`, hex.EncodeToString(nonce[:]), instance); err != nil {

+ return err

+ }

+ log.Printf("console: claim SQL receipt database moved from instance %s to %s; it has a new database ID, and effects bound to the old one stay unsettled", stored, instance)

return nil

}

diff --git a/internal/console/hostreports_process_linux.go b/internal/console/hostreports_process_linux.go

index 9e41beeb6..98acad714 100644

--- a/internal/console/hostreports_process_linux.go

+++ b/internal/console/hostreports_process_linux.go

@@ -58,24 +58,57 @@ func hostReportProcessID(boot string) string {

return boot + ":" + hostReportProcessIdentity()

}

-func hostReportProcessGone(previous, next string) bool {

- old, current := strings.Split(previous, ":"), strings.Split(next, ":")

- if len(old) != 4 || len(current) != 4 || old[0] != current[0] || old[1] != current[1] {

- return false

- }

- namespace, err := hostReportPIDNamespace()

- if err != nil || namespace != old[1] {

- return false

+// hostReportRootPIDNamespace names the PID namespace of init. Main wrote its

+// two-part ProcessID from the host namespace, so a legacy report is only

+// provable here when we run in that same namespace.

+var hostReportRootPIDNamespace = func() (string, error) {

+ info, err := os.Stat("/proc/1/ns/pid")

+ if err != nil {

+ return "", err

}

- boot, err := hostReportProcReadFile("/proc/sys/kernel/random/boot_id")

- if err != nil || strings.TrimSpace(string(boot)) != old[0] {

- return false

+ st, ok := info.Sys().(*syscall.Stat_t)

+ if !ok || st.Ino == 0 {

+ return "", os.ErrInvalid

}

- start, err := strconv.ParseUint(old[3], 10, 64)

- if err != nil || start == 0 {

- return false

+ return strconv.FormatUint(st.Ino, 10), nil

+}

+

+func hostReportProcessGone(previous, next string) bool {

+ old, current := strings.Split(previous, ":"), strings.Split(next, ":")

+ switch {

+ case len(old) == 4 && len(current) == 4 && old[0] == current[0] && old[1] == current[1]:

+ namespace, err := hostReportPIDNamespace()

+ if err != nil || namespace != old[1] || !hostReportBootMatches(old[0]) {

+ return false

+ }

+ if start, err := strconv.ParseUint(old[3], 10, 64); err != nil || start == 0 {

+ return false

+ }

+ return hostReportPIDAbsent(old[2])

+ case len(old) == 2 && len(current) >= 2 && old[0] == current[0]:

+ // Main's boot:pid. It carries no namespace, but main only ran in the

+ // host namespace, so accept it only from a process in that namespace

+ // on the same boot, and only with the same absence proof.

+ namespace, err := hostReportPIDNamespace()

+ if err != nil {

+ return false

+ }

+ root, err := hostReportRootPIDNamespace()

+ if err != nil || root != namespace || !hostReportBootMatches(old[0]) {

+ return false

+ }

+ return hostReportPIDAbsent(old[1])

}

- pid, err := strconv.Atoi(old[2])

+ return false

+}

+

+func hostReportBootMatches(boot string) bool {

+ raw, err := hostReportProcReadFile("/proc/sys/kernel/random/boot_id")

+ return err == nil && strings.TrimSpace(string(raw)) == boot

+}

+

+func hostReportPIDAbsent(text string) bool {

+ pid, err := strconv.Atoi(text)

if err != nil || pid <= 0 {

return false

}

diff --git a/internal/console/plan9_fix1_hostreports_linux_test.go b/internal/console/plan9_fix1_hostreports_linux_test.go

index c24a8d59f..137aac029 100644

--- a/internal/console/plan9_fix1_hostreports_linux_test.go

+++ b/internal/console/plan9_fix1_hostreports_linux_test.go

@@ -52,7 +52,7 @@ func TestPlan9Fix1ConsoleRestartReplacesDeadProcess(t *testing.T) {

if err := hostReportCollision(old, next, time.Now()); err != nil {

t.Fatalf("restart blocked by dead predecessor: %v", err)

}

- for _, id := range []string{"other:" + strconv.Itoa(child.Process.Pid) + ":1", boot + ":" + strconv.Itoa(child.Process.Pid), boot + ":" + strconv.Itoa(child.Process.Pid) + ":bad"} {

+ for _, id := range []string{"other:" + strconv.Itoa(child.Process.Pid) + ":1", boot + ":" + strconv.Itoa(child.Process.Pid) + ":bad"} {

if err := hostReportCollision(plan9Report(id, time.Now()), next, time.Now()); err == nil {

t.Fatalf("unproven predecessor accepted: %s", id)

}

diff --git a/internal/console/plan9_fix1_sql_test.go b/internal/console/plan9_fix1_sql_test.go

index b1684b226..ad07063d7 100644

--- a/internal/console/plan9_fix1_sql_test.go

+++ b/internal/console/plan9_fix1_sql_test.go

@@ -61,7 +61,8 @@ func TestPlan9Fix1BatchRefusesTransactionLeftOpen(t *testing.T) {

}

// I-3: a cloned console (same disk image, different EC2 instance) must not

-// reuse the original's SQL receipt database identity.

+// reuse the original's SQL receipt database identity. Since fix round 2

+// (I2-2) the clone opens under an identity of its own rather than refusing.

func TestPlan9Fix1SQLIdentityRefusesClonedInstance(t *testing.T) {

path := filepath.Join(t.TempDir(), "console.db")

restore := UseSQLInstanceIdentityForTest("i-original")

@@ -79,26 +80,27 @@ func TestPlan9Fix1SQLIdentityRefusesClonedInstance(t *testing.T) {

if !strings.Contains(id, "i-original") {

t.Fatalf("database identity %q does not bind the instance", id)

}

- restore = UseSQLInstanceIdentityForTest("i-clone")

- clone, err := OpenGated(path)

- restore()

- if err == nil {

- clone.Close()

- t.Fatal("cloned instance opened the original's receipt database")

- }

- if !errors.Is(err, hostops.ErrIdentity) {

- t.Fatalf("clone refusal: %v", err)

- }

restore = UseSQLInstanceIdentityForTest("i-original")

- defer restore()

again, err := OpenGated(path)

+ restore()

if err != nil {

t.Fatalf("original instance refused its own database: %v", err)

}

- defer again.Close()

- if id2, err := again.sqlDatabaseID(); err != nil || id2 != id {

+ id2, err := again.sqlDatabaseID()

+ again.Close()

+ if err != nil || id2 != id {

t.Fatalf("identity changed across reopen: %q %q %v", id, id2, err)

}

+ restore = UseSQLInstanceIdentityForTest("i-clone")

+ defer restore()

+ clone, err := OpenGated(path)

+ if err != nil {

+ t.Fatalf("clone: %v", err)

+ }

+ defer clone.Close()

+ if cid, err := clone.sqlDatabaseID(); err != nil || cid == id {

+ t.Fatalf("cloned instance reused the original's receipt database identity: %q %v", cid, err)

+ }

}

func TestPlan9Fix1SQLIdentityRefusesMissingInstance(t *testing.T) {

diff --git a/internal/console/plan9_fix2_console_test.go b/internal/console/plan9_fix2_console_test.go

new file mode 100644

index 000000000..9d8ba13ab

--- /dev/null

+++ b/internal/console/plan9_fix2_console_test.go

@@ -0,0 +1,293 @@

+package console

+

+import (

+ "context"

+ "errors"

+ "os"

+ "path/filepath"

+ "strings"

+ "sync"

+ "testing"

+ "time"

+

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

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

+)

+

+// I2-2: a console database restored onto a replacement instance opens under

+// a new database ID. Effects bound to the original ID can never settle here.

+func TestPlan9Fix2RestoreOntoReplacementInstanceAdoptsNewIdentity(t *testing.T) {

+ path := filepath.Join(t.TempDir(), "console.db")

+ openAs := func(instance string) (*Store, string) {

+ t.Helper()

+ restore := UseSQLInstanceIdentityForTest(instance)

+ defer restore()

+ s, err := OpenGated(path)

+ if err != nil {

+ t.Fatalf("open on %s: %v", instance, err)

+ }

+ id, err := s.sqlDatabaseID()

+ if err != nil {

+ s.Close()

+ t.Fatal(err)

+ }

+ return s, id

+ }

+ s, original := openAs("i-original")

+ s.Close()

+

+ replacement, adopted := openAs("i-replacement")

+ if adopted == original || !strings.Contains(adopted, ":i-replacement:") {

+ replacement.Close()

+ t.Fatalf("replacement did not adopt a new identity: original %q, adopted %q", original, adopted)

+ }

+ nonce := func(id string) string { return strings.SplitN(id, ":", 2)[0] }

+ if nonce(adopted) == nonce(original) {

+ replacement.Close()

+ t.Fatalf("adoption kept the original nonce %q", nonce(original))

+ }

+ // The original's effects stay unsettled: a fence for them is refused.

+ sc := hostops.SQLReceiptScope{Effect: "e1", Operation: "op1", AppID: "app", Epoch: 1, Database: original}

+ if _, err := replacement.FenceSQLOutcome(t.Context(), sc); !errors.Is(err, hostops.ErrIdentity) {

+ replacement.Close()

+ t.Fatalf("replacement fenced an effect bound to the original database: %v", err)

+ }

+ replacement.Close()

+

+ // The adopted identity is stable across restarts on the replacement.

+ again, id := openAs("i-replacement")

+ again.Close()

+ if id != adopted {

+ t.Fatalf("adopted identity changed on reopen: %q then %q", adopted, id)

+ }

+ // Moving the copy back to the original instance is another adoption, not

+ // a return to the original identity.

+ back, id := openAs("i-original")

+ back.Close()

+ if id == original || id == adopted {

+ t.Fatalf("copy moved back reused an earlier identity: %q", id)

+ }

+ // m-3: the same file on the same instance keeps its ID (a block-level

+ // restore looks like this). The runbook's break-glass step copies it to

+ // a new file and moves that into place, which gives it a new ID.

+ same, sameID := openAs("i-original")

+ same.Close()

+ if sameID != id {

+ t.Fatalf("same file on the same instance changed identity: %q then %q", id, sameID)

+ }

+ b, err := os.ReadFile(path)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err = os.WriteFile(path+".copy", b, 0o600); err != nil {

+ t.Fatal(err)

+ }

+ if err = os.Rename(path+".copy", path); err != nil {

+ t.Fatal(err)

+ }

+ copied, copiedID := openAs("i-original")

+ copied.Close()

+ if copiedID == id {

+ t.Fatalf("a file copy moved into place kept the identity %q", id)

+ }

+}

+

+// replacedFixture is a gated console whose sweep sees one SQL effect past

+// sql_recover_after bound to database, with two operators of whom only

+// ops-a has a keymail grant.

+func replacedFixture(t *testing.T, st *Store, database string) (*Server, *[]string) {

+ t.Helper()

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

+ w := claimTestWriter(t, raw)

+ reg := hostops.Registration{Kind: hostops.RegistrationIngest, Owner: claimTestOwner, AppID: "app", Epoch: 1, Expires: time.Now().Add(time.Hour)}

+ past := time.Now().Add(-time.Minute).UTC().Format(time.RFC3339Nano)

+ seedClaimAdmission(t, raw, "bab", "demo", hostops.AppAdmission{V: 1, AppID: "app", Revision: 1, Epoch: 1, State: "open",

+ Registrations: map[string]hostops.Registration{"old": reg},

+ Effects: map[string]hostops.Effect{"old": {ID: "old", OperationID: "old", AppID: "app", Resource: "app-sql:old", Action: "sql-ingest", Owner: claimTestOwner, State: "uncertain",

+ Inputs: map[string]string{"sql_database": database, "sql_recover_after": past}}}})

+ for _, addr := range []string{"ops-a@keymail.dev", "ops-b@keymail.dev"} {

+ if err := st.SetOperator(addr, true); err != nil {

+ t.Fatal(err)

+ }

+ if err := st.SetGrantToken("ops-a@keymail.dev", "grant-a", time.Now()); err != nil {

+ t.Fatal(err)

+ }

+ var mu sync.Mutex

+ sent := &[]string{}

+ s := &Server{store: st, cfg: Config{HostClaimOps: &hostops.Controller{Store: raw, Control: w, Now: time.Now}}}

+ s.sendApp = func(_ context.Context, _, to, _, _ string) (string, error) {

+ mu.Lock()

+ defer mu.Unlock()

+ *sent = append(*sent, to)

+ return "msg", nil

+ }

+ return s, sent

+}

+

+// m-4: the alert send is bounded by the sweep's own context, not a fresh

+// background deadline per recipient.

+func TestPlan9Fix2ReplacedAlertSendHonoursSweepContext(t *testing.T) {

+ st := openGatedStore(t)

+ s, _ := replacedFixture(t, st, "restored-away")

+ for _, addr := range []string{"ops-b@keymail.dev", "ops-c@keymail.dev"} {

+ if err := st.SetOperator(addr, true); err != nil {

+ t.Fatal(err)

+ }

+ if err := st.SetGrantToken(addr, "grant-"+addr, time.Now()); err != nil {

+ t.Fatal(err)

+ }

+ s.sendApp = func(ctx context.Context, _, _, _, _ string) (string, error) {

+ <-ctx.Done()

+ return "", ctx.Err()

+ }

+ ctx, cancel := context.WithTimeout(t.Context(), 200*time.Millisecond)

+ defer cancel()

+ start := time.Now()

+ err := s.sweepOneClaimAdmission(ctx, hostops.App{Account: "bab", Name: "demo"})

+ if !errors.Is(err, ErrSQLDatabaseReplaced) {

+ t.Fatalf("sweep: %v", err)

+ }

+ if took := time.Since(start); took > 5*time.Second {

+ t.Fatalf("a stalled keymail send held the serial sweep for %v", took)

+ }

+}

+

+// m-4: a transient failure to read the recipients is retried on the next

+// sweep instead of being recorded as "nobody to tell".

+func TestPlan9Fix2ReplacedAlertRetriesAfterRecipientFailure(t *testing.T) {

+ st := openGatedStore(t)

+ s, sent := replacedFixture(t, st, "restored-away")

+ if _, err := st.rawBoot.Exec(`ALTER TABLE users RENAME TO users_away`); err != nil {

+ t.Fatal(err)

+ }

+ app := hostops.App{Account: "bab", Name: "demo"}

+ if err := s.sweepOneClaimAdmission(t.Context(), app); !errors.Is(err, ErrSQLDatabaseReplaced) {

+ t.Fatalf("sweep: %v", err)

+ }

+ if _, known, _ := st.AlertSeen(sqlReceiptAlertBox, sqlReplacedRule("bab", "demo", "old"), noMailboxAddress); known {

+ t.Fatal("a transient recipient failure was recorded as no mailbox")

+ }

+ if _, err := st.rawBoot.Exec(`ALTER TABLE users_away RENAME TO users`); err != nil {

+ t.Fatal(err)

+ }

+ if err := s.sweepOneClaimAdmission(t.Context(), app); !errors.Is(err, ErrSQLDatabaseReplaced) {

+ t.Fatalf("sweep: %v", err)

+ }

+ if len(*sent) != 1 || (*sent)[0] != "ops-a@keymail.dev" {

+ t.Fatalf("alert not delivered after the recipients recovered: %v", *sent)

+ }

+}

+

+// m-4: the missing-grant note is logged when an alert is delivered, not on

+// every sweep that finds nothing new to say.

+func TestPlan9Fix2ReplacedAlertMissingGrantLoggedOnce(t *testing.T) {

+ st := openGatedStore(t)

+ s, sent := replacedFixture(t, st, "restored-away")

+ buf := captureLog(t)

+ app := hostops.App{Account: "bab", Name: "demo"}

+ for range 3 {

+ if err := s.sweepOneClaimAdmission(t.Context(), app); !errors.Is(err, ErrSQLDatabaseReplaced) {

+ t.Fatalf("sweep: %v", err)

+ }

+ if len(*sent) != 1 {

+ t.Fatalf("sent %v, want one delivery", *sent)

+ }

+ if n := strings.Count(buf.String(), "no keymail grant for ops-b@keymail.dev"); n != 1 {

+ t.Fatalf("missing-grant note logged %d times over three sweeps:\n%s", n, buf.String())

+ }

+}

+

+// m-4: a database replaced under the running console alerts like a restored

+// one, rather than surfacing only as an identity error.

+func TestPlan9Fix2DatabaseReplacedUnderRunningConsoleAlerts(t *testing.T) {

+ st := openGatedStore(t)

+ id, err := st.sqlDatabaseID()

+ if err != nil {

+ t.Fatal(err)

+ }

+ s, sent := replacedFixture(t, st, id)

+ // Replace the file under the running process: same path, new inode.

+ path := st.gate.path

+ b, err := os.ReadFile(path)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err = os.WriteFile(path+".restored", b, 0o600); err != nil {

+ t.Fatal(err)

+ }

+ if err = os.Rename(path+".restored", path); err != nil {

+ t.Fatal(err)

+ }

+ err = s.sweepOneClaimAdmission(t.Context(), hostops.App{Account: "bab", Name: "demo"})

+ if !errors.Is(err, ErrSQLDatabaseReplaced) {

+ t.Fatalf("live replacement surfaced only as %v", err)

+ }

+ if len(*sent) != 1 {

+ t.Fatalf("no operator alert for a live replacement: %v", *sent)

+ }

+}

+

+// m-5: a callback that calls back into the outer, captured Store (rather

+// than the view it was given) is refused at once instead of waiting on the

+// connection the view holds, and no remote check runs first.

+func TestPlan9Fix2OuterStoreInsideViewRefused(t *testing.T) {

+ s := openGatedStore(t)

+ seedGateApp(t, s)

+ t.Run("nested mutation through the outer store", func(t *testing.T) {

+ calls := 0

+ g := sqlGrant{scope: gateScope, kind: grantEffect, check: func(context.Context) error { calls++; return nil }}

+ ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)

+ defer cancel()

+ err := s.withClaimSQL(ctx, []claimSQLScope{gateScope}, allowAllClaimSQL, func(*Store) error {

+ return s.WithAppMutation(ctx, []sqlGrant{g}, func(*Store) error { return nil })

+ })

+ if !errors.Is(err, ErrClaimSQLScope) {

+ t.Fatalf("nested mutation through the outer store: %v", err)

+ }

+ if calls != 0 {

+ t.Fatalf("ran %d remote checks before refusing", calls)

+ }

+ })

+ t.Run("plain outer query", func(t *testing.T) {

+ ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)

+ defer cancel()

+ var qerr error

+ err := s.withClaimSQL(ctx, []claimSQLScope{gateScope}, allowAllClaimSQL, func(*Store) error {

+ _, qerr = s.Operators()

+ return nil

+ })

+ if err != nil {

+ t.Fatal(err)

+ }

+ if !errors.Is(qerr, ErrClaimSQLScope) {

+ t.Fatalf("outer query inside the view: %v", qerr)

+ }

+ })

+ // Other goroutines still wait their turn rather than being refused.

+ t.Run("other goroutine waits", func(t *testing.T) {

+ release := make(chan struct{})

+ held := make(chan struct{})

+ done := make(chan error, 1)

+ go func() {

+ done <- s.withClaimSQL(t.Context(), []claimSQLScope{gateScope}, allowAllClaimSQL, func(*Store) error {

+ close(held)

+ <-release

+ return nil

+ })

+ }()

+ <-held

+ got := make(chan error, 1)

+ go func() { _, err := s.Operators(); got <- err }()

+ time.Sleep(50 * time.Millisecond)

+ close(release)

+ if err := <-done; err != nil {

+ t.Fatal(err)

+ }

+ if err := <-got; err != nil {

+ t.Fatalf("concurrent outer call refused: %v", err)

+ }

+ })

+}

diff --git a/internal/console/plan9_fix2_legacy_process_linux_test.go b/internal/console/plan9_fix2_legacy_process_linux_test.go

new file mode 100644

index 000000000..daaf66324

--- /dev/null

+++ b/internal/console/plan9_fix2_legacy_process_linux_test.go

@@ -0,0 +1,68 @@

+//go:build linux

+

+package console

+

+import (

+ "fmt"

+ "os"

+ "strings"

+ "testing"

+)

+

+func plan9Fix2Boot(t *testing.T) string {

+ t.Helper()

+ raw, err := os.ReadFile("/proc/sys/kernel/random/boot_id")

+ if err != nil {

+ t.Fatal(err)

+ }

+ return strings.TrimSpace(string(raw))

+}

+

+func plan9Fix2RootNamespace(t *testing.T, ns string, err error) {

+ t.Helper()

+ prior := hostReportRootPIDNamespace

+ t.Cleanup(func() { hostReportRootPIDNamespace = prior })

+ hostReportRootPIDNamespace = func() (string, error) { return ns, err }

+}

+

+func TestPlan9Fix2LegacyTwoPartIDProvedGoneInHostNamespace(t *testing.T) {

+ boot := plan9Fix2Boot(t)

+ ns, err := hostReportPIDNamespace()

+ if err != nil {

+ t.Fatal(err)

+ }

+ plan9Fix2RootNamespace(t, ns, nil)

+ if !hostReportProcessGone(boot+":999999999", hostReportProcessID(boot)) {

+ t.Fatal("main's absent boot:pid was not accepted during the roll")

+ }

+}

+

+func TestPlan9Fix2LegacyTwoPartIDNeverProvedOutsideHostNamespace(t *testing.T) {

+ boot := plan9Fix2Boot(t)

+ ns, _ := hostReportPIDNamespace()

+ next := hostReportProcessID(boot)

+ plan9Fix2RootNamespace(t, ns+"0", nil)

+ if hostReportProcessGone(boot+":999999999", next) {

+ t.Fatal("legacy ID proved gone from a namespace that is not init's")

+ }

+ plan9Fix2RootNamespace(t, "", os.ErrPermission)

+ if hostReportProcessGone(boot+":999999999", next) {

+ t.Fatal("legacy ID proved gone with an unreadable init namespace")

+ }

+}

+

+func TestPlan9Fix2LegacyTwoPartIDKeepsLivePIDAndBootChecks(t *testing.T) {

+ boot := plan9Fix2Boot(t)

+ ns, _ := hostReportPIDNamespace()

+ plan9Fix2RootNamespace(t, ns, nil)

+ next := hostReportProcessID(boot)

+ if hostReportProcessGone(fmt.Sprintf("%s:%d", boot, os.Getpid()), next) {

+ t.Fatal("a live legacy PID was proved gone")

+ }

+ if hostReportProcessGone("other-boot:999999999", hostReportProcessID("other-boot")) {

+ t.Fatal("a legacy ID from another boot was proved gone")

+ }

+ if hostReportProcessGone(boot+":999999999", hostReportProcessID("other-boot")) {

+ t.Fatal("boot mismatch between reports was ignored")

+ }

+}

diff --git a/internal/console/store_gate.go b/internal/console/store_gate.go

index 039ab3410..aac845c62 100644

--- a/internal/console/store_gate.go

+++ b/internal/console/store_gate.go

@@ -757,6 +757,9 @@ type claimGate struct {

armed atomic.Bool

mu sync.RWMutex

arm claimArm

+ // viewG is the goroutine holding the bound view's connection, 0 when

+ // none (claim_sql_outer.go).

+ viewG atomic.Uint64

}

func (g *claimGate) armSnapshot() claimArm {

@@ -1678,9 +1681,10 @@ func normalizeClaimScopes(in []claimSQLScope) ([]claimSQLScope, error) {

//

// Every scope must pass the remote check before the connection is acquired.

// Context, deadline and local capability checks guard each write and commit. The view holds the pool's connection for the duration,

-// so fn must use only the *Store it is given (an outer store call waits for

-// that connection until its context ends), and neither fn nor check may call

-// withClaimSQL again. Transactions, statements and rows from the view die when

+// so fn must use only the *Store it is given (an outer store call from fn's

+// goroutine is refused with errOuterStoreInView; one from a goroutine fn

+// starts waits for that connection until its context ends), and neither fn

+// nor check may call withClaimSQL again. Transactions, statements and rows from the view die when

// fn returns.

func (s *Store) withClaimSQL(ctx context.Context, scopes []claimSQLScope, check func(context.Context, claimSQLScope) error, fn func(*Store) error) error {

return s.withPreparedClaimSQL(ctx, scopes, check, func(ctx context.Context, _ claimSQLScope) error { return ctx.Err() }, nil, false, fn)

@@ -1693,6 +1697,9 @@ func (s *Store) withPreparedClaimSQL(ctx context.Context, scopes []claimSQLScope

if s.bound != nil {

return fmt.Errorf("%w: nested withClaimSQL", ErrClaimSQLScope)

}

+ if s.outerInsideView() {

+ return errOuterStoreInView

+ }

if fn == nil || check == nil || local == nil || len(scopes) == 0 {

return fmt.Errorf("%w: scopes, check and fn are required", ErrClaimSQLScope)

}

@@ -1736,7 +1743,9 @@ func (s *Store) withPreparedClaimSQL(ctx context.Context, scopes []claimSQLScope

vs := *s

vs.db = view

vs.bound = b

+ s.gate.viewG.Store(goroutineID())

defer func() {

+ s.gate.viewG.Store(0)

err = errors.Join(err, view.revoke())

if !batch {

// A batch reports this from endBatch, before its receipt.

@@ -1804,6 +1813,7 @@ func openGated(path, dsn string, tables map[string]claimSQLTable, migrate func(*

return nil, err

}

s := newStore(pool)

+ s.db = claimOuterDB{DB: pool, g: g}

s.rawBoot, s.gate = boot, g

// Trusted boot seam: unarmed, so migrations may run DDL and seed rows.

if err = migrate(s); err != nil {

diff --git a/internal/console/store_gate_test.go b/internal/console/store_gate_test.go

index 2d85598f2..614d05979 100644

--- a/internal/console/store_gate_test.go

+++ b/internal/console/store_gate_test.go

@@ -532,14 +532,14 @@ func TestClaimSQLViewCannotNestOrClose(t *testing.T) {

if err != nil {

t.Fatal(err)

}

- // An outer-store call inside the callback waits for the held connection

- // and gives up with its own context rather than hanging.

+ // An outer-store call inside the callback is refused at once rather than

+ // waiting for the connection the view holds (review m-5).

err = s.withClaimSQL(gateCtx(t), []claimSQLScope{gateScope}, allowAllClaimSQL, func(*Store) error {

short, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)

defer cancel()

_, err := s.db.ExecContext(short, `INSERT INTO users (address, created_at) VALUES ('outer@x', 1)`)

- if !errors.Is(err, context.DeadlineExceeded) {

- t.Errorf("outer call err = %v, want deadline exceeded", err)

+ if !errors.Is(err, ErrClaimSQLScope) {

+ t.Errorf("outer call err = %v, want ErrClaimSQLScope", err)

}

return nil

})

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

index 0ae011d26..da57a91b6 100644

--- a/internal/host/claimconstruction.go

+++ b/internal/host/claimconstruction.go

@@ -173,8 +173,9 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla

var inventory *hostbox.StopInventoryDB

var journals *hostbox.JournalService

if r.Mode != "agent" {

- // Offline migration has a finite row-scaled budget and keeps caller

- // cancellation, rather than inheriting ordinary construction's two minutes.

+ // The legacy journal check may wait on the migration lock while an

+ // operator runs the offline import (carlos system journal-migrate), so

+ // it keeps caller cancellation rather than construction's two minutes.

finishStartup()

ctx = callerCtx

if err = hostbox.PrepareRootJournalDirectory("/var/lib/carlos-claims"); err != nil {

@@ -190,7 +191,7 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla

return nil, err

}

inventory = journals.Inventory

- if err = journals.MigrateLegacy(ctx, local, actor.Box, actor.Enrollment, quiesceLegacyClaimJournals, legacyJournalProof(store, owners.RetirementAuthority.Directory)); err != nil {

+ if err = checkClaimJournalMigration(ctx, journals, local); err != nil {

launches.Close()

return nil, err

}

@@ -271,7 +272,7 @@ func OpenProductionClaimAssembly(ctx context.Context, store release.Store, r Cla

}, Guard: guard, Claims: store, DataDir: r.DataDir, Box: actor.Box}

wake.RetireArtifacts = claimArtifactRetirement{Local: retirement, Continue: hostclaim.SessionContinues}.Retire

wake.RecoverArtifacts = claimArtifactRetirement{Local: retirement, Continue: hostclaim.SessionContinues}.Recover

- wake.Continue = hostclaim.SessionContinues

+ wake.Continue = claimWakeContinue

wake.Operations = owners.Moves.Operations

wake.ProbeReady = owners.Moves.ProbeReady

if wake.ProbeReady == nil {

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

new file mode 100644

index 000000000..4ad9948b9

--- /dev/null

+++ b/internal/host/claimjournal_migration.go

@@ -0,0 +1,98 @@

+package host

+

+import (

+ "context"

+ "errors"

+ "flag"

+ "log"

+ "os"

+ "os/signal"

+ "path/filepath"

+ "syscall"

+

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

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

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

+)

+

+// checkClaimJournalMigration is root startup's only contact with the legacy

+// claim journals under the agent's data directory. It never imports and never

+// quiesces: legacy files the agent changes after the marker are reported and

+// ignored, so they cannot stop root's control plane.

+func checkClaimJournalMigration(ctx context.Context, journals *hostbox.JournalService, local string) error {

+ alert, err := journals.CheckLegacyMigration(ctx, local)

+ if alert != nil {

+ log.Printf("claim journal migration: %v", alert)

+ }

+ return err

+}

+

+// RunSystemJournalMigrate is the offline operator step that imports legacy

+// claim journals into root's journal: carlos system journal-migrate, run as

+// root with the edge and the agent stopped. SIGTERM and interrupt cancel the

+// import, so the edge's runtime mask is lifted on the way out.

+func RunSystemJournalMigrate(args []string) {

+ fs := flag.NewFlagSet("system journal-migrate", flag.ExitOnError)

+ _ = fs.Parse(args)

+ if os.Geteuid() != 0 || fs.NArg() != 0 {

+ log.Fatal(hostwrite.ErrUnavailable)

+ }

+ ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)

+ defer stop()

+ ctx, err := claimEntryContext(ctx, os.Getenv, openDeployStore)

+ if err != nil {

+ log.Fatal(err)

+ }

+ if !hostwrite.Enabled(claimEntryEnv(ctx)) {

+ log.Fatal(hostwrite.ErrUnavailable)

+ }

+ reg, err := openServiceRegistry(filepath.Join(dataDir(), "registry.db"))

+ if err != nil {

+ log.Fatal(err)

+ }

+ defer reg.Close()

+ provider := productionClaimOwners()

+ if provider == nil {

+ provider = ProductionClaimOwnerProvider{}

+ }

+ r := ClaimRuntimeResources{Mode: "instance-control", Registry: reg, DataDir: dataDir(), OptDir: release.OptDir(), UnitsDir: "/etc/systemd/system"}

+ if err = migrateClaimJournals(ctx, claimStore(ctx), r, provider); err != nil {

+ log.Fatal(err)

+ }

+}

+

+// migrateClaimJournals opens only what the import needs: the verified actor,

+// root's journal and the retirement directory for the legacy stop proofs.

+func migrateClaimJournals(ctx context.Context, store release.Store, r ClaimRuntimeResources, provider ClaimOwnerProvider) (resultErr error) {

+ if store == nil || provider == nil || r.Mode != "instance-control" || !filepath.IsAbs(r.DataDir) {

+ return hostwrite.ErrUnavailable

+ }

+ owners, err := provider.OpenClaimOwners(ctx, store, r)

+ if err != nil {

+ return err

+ }

+ if owners.Close == nil {

+ return hostwrite.ErrUnavailable

+ }

+ defer func() { resultErr = errors.Join(resultErr, owners.Close(context.WithoutCancel(ctx))) }()

+ if owners.RetirementAuthority == nil {

+ return hostwrite.ErrUnavailable

+ }

+ actor, err := owners.actorForConstruction(ctx, r.Mode)

+ if err != nil {

+ return err

+ }

+ if err = hostbox.PrepareRootJournalDirectory("/var/lib/carlos-claims"); err != nil {

+ return err

+ }

+ launches, err := hostbox.OpenLaunchJournal("/var/lib/carlos-claims/journals.db")

+ if err != nil {

+ return err

+ }

+ defer func() { resultErr = errors.Join(resultErr, launches.Close()) }()

+ journals, err := hostbox.NewJournalService(launches)

+ if err != nil {

+ return err

+ }

+ return journals.MigrateLegacy(ctx, filepath.Join(r.DataDir, ".hostclaims"), actor.Box, actor.Enrollment, quiesceLegacyClaimJournals, legacyJournalProof(store, owners.RetirementAuthority.Directory))

+}

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

index 73f845e08..44cdcfbd1 100644

--- a/internal/host/claimjournal_migration_linux.go

+++ b/internal/host/claimjournal_migration_linux.go

@@ -16,6 +16,13 @@ import (

var claimQuiesceOwner = claimStateOwner

+// claimEdgeMaskOwner records that the migration, not an operator, put the

+// edge's runtime mask in place. It holds the boot ID and is written before

+// the mask: a runtime mask ends at reboot, so a record from another boot is

+// stale. It lives in root's own journal directory, which the agent cannot

+// write.

+var claimEdgeMaskOwner = "/var/lib/carlos-claims/edge-mask"

+

var claimSystemctl = func(ctx context.Context, args ...string) ([]byte, error) {

return exec.CommandContext(ctx, "systemctl", args...).Output()

}

@@ -40,22 +47,49 @@ func quiesceLegacyClaimJournals(ctx context.Context) (func(), error) {

if err = claimEdgeStopped(ctx); err != nil {

return nil, err

}

- // A pre-existing runtime mask belongs to its operator, so leave it in place.

+ boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")

+ if err != nil {

+ return nil, err

+ }

+ record, err := os.ReadFile(claimEdgeMaskOwner)

+ if err != nil && !errors.Is(err, os.ErrNotExist) {

+ return nil, err

+ }

before, err := claimSystemctl(ctx, "is-enabled", "carlos-edge.service")

- masked := strings.TrimSpace(string(before)) == "masked" || strings.TrimSpace(string(before)) == "masked-runtime"

+ state := strings.TrimSpace(string(before))

+ masked := state == "masked" || state == "masked-runtime"

if err != nil && !masked {

return nil, err

}

+ // A runtime mask recorded in this boot was left by a migration that was

+ // killed while masked: it is ours to lift. Any other existing mask belongs

+ // to its operator and stays in place, and a record that names no live

+ // mask of ours is stale.

+ ours := state == "masked-runtime" && len(record) != 0 && string(record) == strings.TrimSpace(string(boot))

+ if !ours && record != nil {

+ if err = os.Remove(claimEdgeMaskOwner); err != nil {

+ return nil, err

+ }

if !masked {

+ if err = os.WriteFile(claimEdgeMaskOwner, []byte(strings.TrimSpace(string(boot))), 0600); err != nil {

+ return nil, err

+ }

if _, err = claimSystemctl(ctx, "mask", "--runtime", "carlos-edge.service"); err != nil {

return nil, err

}

+ ours = true

}

release := func() {

- if !masked {

- if _, err := claimSystemctl(context.Background(), "unmask", "--runtime", "carlos-edge.service"); err != nil {

- log.Printf("claim journal migration: edge runtime mask remains: %v", err)

- }

+ if !ours {

+ return

+ }

+ if _, err := claimSystemctl(context.Background(), "unmask", "--runtime", "carlos-edge.service"); err != nil {

+ log.Printf("claim journal migration: edge runtime mask remains: %v", err)

+ return

+ }

+ if err := os.Remove(claimEdgeMaskOwner); err != nil && !errors.Is(err, os.ErrNotExist) {

+ log.Printf("claim journal migration: edge mask record remains: %v", err)

}

// The edge may have started between the first check and the mask.

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

index d749539b6..07ef0f0ce 100644

--- a/internal/host/claimjournal_transport.go

+++ b/internal/host/claimjournal_transport.go

@@ -166,6 +166,21 @@ func (h ClaimRootHTTP) serveJournal(w http.ResponseWriter, r *http.Request, peer

}

+ if q.Op == "inventory-write" {

+ if h.Reconcile.Registry == nil {

+ w.WriteHeader(http.StatusForbidden)

+ return

+ }

+ row, err := h.Reconcile.Registry.Get(q.Inventory.Host)

+ if err != nil {

+ w.WriteHeader(http.StatusServiceUnavailable)

+ return

+ }

+ if hostbox.ValidateAgentInventory(q.Inventory, row, h.Roots["data"]) != nil {

+ w.WriteHeader(http.StatusForbidden)

+ return

+ }

out, err := h.Journals.Execute(r.Context(), owner, q)

if err != nil {

switch {

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

index 88b177954..7c771bec9 100644

--- a/internal/host/claimretire_transport.go

+++ b/internal/host/claimretire_transport.go

@@ -135,6 +135,7 @@ func (h ClaimRootHTTP) retire(ctx context.Context, target hostguard.FenceTarget)

return hostbox.ErrTypedInventory

}

a.RemoteRetire = nil

+ a.DataRoot = h.Roots["data"]

a.Box = target.Box

a.Reconciler.Guard = h.Stop.Coordinator.Guard

a.Stop = func(context.Context, registry.Route) (hostguard.FenceTarget, error) { return target, nil }

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

index d00780bc2..f7eeb770c 100644

--- a/internal/host/claimwakeassembly.go

+++ b/internal/host/claimwakeassembly.go

@@ -711,6 +711,11 @@ func (r *claimOwnedRep) Close(ctx context.Context) error {

return r.owner.release(ctx, r.host, r.attempt)

}

+// claimWakeContinue is the wake and selected-work checkpoint rule: an admitted

+// writer may observe a later phase of its own row after missing one (m-2). The

+// artifact retirement and the claim's own transitions keep SessionContinues.

+var claimWakeContinue = hostclaim.SessionObserves

+

// claimSameWriter reports whether b, observed at or after a, still names a's

// admitted writer session. continues is the session forward-step table; nil

// requires the same phase, as main did.

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

index d9caab4b2..a9a279e16 100644

--- a/internal/host/movesland.go

+++ b/internal/host/movesland.go

@@ -48,8 +48,8 @@ import (

//

// The claim never leads the record. Every phase starts with syncClaimPhase:

// equal is the common case; a claim behind the record (a crash between the

-// record CAS and the claim CAS) is walked forward along LandingOrder in one

-// claim CAS; a claim ahead is left alone and logged, and the phase's own

+// record CAS and the claim CAS) is walked forward along LandingOrder with

+// one claim CAS per adjacent step; a claim ahead is left alone and logged, and the phase's own

// RequireMoveOp then refuses, so nothing is written over it.

//

// Nothing here ever serves from moves/: the lane is read by exact version

@@ -343,7 +343,7 @@ func (l *landingRun) claim(ctx context.Context) (hostclaim.Observed, error) {

// syncClaimPhase is every phase's first step (§7.5): the claim's op.phase

// follows the record's landing_phase. Behind, it is walked forward one step

-// at a time in one CAS; ahead, it is left alone and logged (the phase's own

+// at a time, each in its own CAS; ahead, it is left alone and logged (the phase's own

// RequireMoveOp then refuses); equal is the common case. The observation

// returned is the one the phase works from, and names this box as the

// executor.

@@ -371,16 +371,15 @@ func (l *landingRun) syncClaimPhase(ctx context.Context) (hostclaim.Observed, er

log.Printf("moves: %s: the claim's landing phase %q is ahead of the record's %q; left alone", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)

return obs, nil

}

- obs, err = hostclaim.Update(ctx, l.s, obs, func(c *hostclaim.Claim) error {

- for i := ci; i < ri; i++ {

- if err := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, order[i], order[i+1])(c); err != nil {

- return err

- }

+ for i := ci; i < ri; i++ {

+ step := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, order[i], order[i+1])

+ if hostclaim.SessionPhase(hostclaim.KindLanding, order[i]) && hostclaim.SessionPhase(hostclaim.KindLanding, order[i+1]) {

+ step = hostclaim.KeepWriterSession(l.g.ID, step)

+ }

+ obs, err = hostclaim.Update(ctx, l.s, obs, step)

+ if err != nil {

+ return obs, fmt.Errorf("catch the claim up to %s: %w", l.rec.Move.LandingPhase, err)

}

- return nil

- })

- if err != nil {

- return obs, fmt.Errorf("catch the claim up to %s: %w", l.rec.Move.LandingPhase, err)

}

log.Printf("moves: %s: the claim's landing phase caught up from %q to %q", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)

return obs, nil

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

index 60373a7d3..23350d7e7 100644

--- a/internal/host/movesland_test.go

+++ b/internal/host/movesland_test.go

@@ -867,7 +867,7 @@ func TestLandingParkingWaitsForTheManifest(t *testing.T) {

// Phase catch-up (§7.5, round-2 finding 11): a crash after the record's CAS

// and before the claim's leaves the claim one step behind; the next pass

// walks the claim forward and continues. A claim several steps behind is

-// walked step by step in one write.

+// walked step by step, with each transition written separately.

func TestLandingPhaseCatchUp(t *testing.T) {

ctx := context.Background()

h := newLandingHarness(t)

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

new file mode 100644

index 000000000..d7f5cc923

--- /dev/null

+++ b/internal/host/plan9_fix2_observe_test.go

@@ -0,0 +1,30 @@

+package host

+

+import (

+ "testing"

+

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

+)

+

+// m-2: the wake checkpoint observes committed after a missed probed, for the

+// same writer only.

+func TestPlan9Fix2WakeCheckpointCatchesUpOnlyForTheSameWriter(t *testing.T) {

+ probing := plan9D1Session(hostclaim.PhaseLandingProbing, 11)

+ committed := plan9D1Session(hostclaim.PhaseLandingCommitted, 11)

+ a := &claimWakeAttempt{observed: probing}

+ if !a.advance(committed, claimWakeContinue) {

+ t.Fatal("an admitted writer lost its session across an unobserved probed")

+ }

+ if a.advance(plan9D1Session(hostclaim.PhaseLandingProbed, 11), claimWakeContinue) {

+ t.Fatal("the checkpoint went backward")

+ }

+ b := &claimWakeAttempt{observed: probing}

+ if b.advance(plan9D1Session(hostclaim.PhaseLandingCommitted, 12), claimWakeContinue) {

+ t.Fatal("a new session was caught up to")

+ }

+ other := plan9D1Session(hostclaim.PhaseLandingCommitted, 11)

+ other.Claim.Op.ID = "m2"

+ if b.advance(other, claimWakeContinue) {

+ t.Fatal("another op was caught up to")

+ }

+}

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

new file mode 100644

index 000000000..6b2f39408

--- /dev/null

+++ b/internal/host/plan9fix2_claimjournal_migration_linux_test.go

@@ -0,0 +1,128 @@

+//go:build linux

+

+package host

+

+import (

+ "context"

+ "os"

+ "path/filepath"

+ "slices"

+ "strings"

+ "testing"

+)

+

+// plan9Fix2Systemctl fakes systemctl for a stopped edge whose unit file state

+// is enabled. It records every call.

+func plan9Fix2Systemctl(t *testing.T, enabled string, calls *[]string) {

+ t.Helper()

+ restore := claimSystemctl

+ t.Cleanup(func() { claimSystemctl = restore })

+ restoreOwner := claimQuiesceOwner

+ t.Cleanup(func() { claimQuiesceOwner = restoreOwner })

+ claimQuiesceOwner = func() (int, int, error) { return 1 << 30, 1 << 30, nil }

+ claimSystemctl = func(_ context.Context, args ...string) ([]byte, error) {

+ *calls = append(*calls, strings.Join(args, " "))

+ switch args[0] {

+ case "is-enabled":

+ return []byte(enabled + "\n"), nil

+ case "show":

+ return []byte("inactive\n"), nil

+ }

+ return nil, nil

+ }

+}

+

+func plan9Fix2MaskOwner(t *testing.T) string {

+ t.Helper()

+ restore := claimEdgeMaskOwner

+ t.Cleanup(func() { claimEdgeMaskOwner = restore })

+ claimEdgeMaskOwner = filepath.Join(t.TempDir(), "edge-mask")

+ return claimEdgeMaskOwner

+}

+

+func plan9Fix2Boot(t *testing.T) string {

+ t.Helper()

+ boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id")

+ if err != nil {

+ t.Fatal(err)

+ }

+ return strings.TrimSpace(string(boot))

+}

+

+// m-7: a runtime mask left by a migration that was killed while masked is

+// recognised as the migration's own and lifted by the next attempt.

+func TestPlan9Fix2QuiesceLiftsMaskLeftByCrashedAttempt(t *testing.T) {

+ var calls []string

+ plan9Fix2Systemctl(t, "masked-runtime", &calls)

+ owner := plan9Fix2MaskOwner(t)

+ if err := os.WriteFile(owner, []byte(plan9Fix2Boot(t)), 0600); err != nil {

+ t.Fatal(err)

+ }

+ release, err := quiesceLegacyClaimJournals(t.Context())

+ if err != nil {

+ t.Fatal(err)

+ }

+ release()

+ if !slices.Contains(calls, "unmask --runtime carlos-edge.service") {

+ t.Fatalf("mask left by a crashed migration not lifted: %v", calls)

+ }

+ if _, err = os.Stat(owner); !os.IsNotExist(err) {

+ t.Fatalf("mask record kept after the mask was lifted: %v", err)

+ }

+}

+

+// m-7: the record is written before the mask, so a kill between the two still

+// leaves the mask recognisable.

+func TestPlan9Fix2QuiesceRecordsMaskBeforeMasking(t *testing.T) {

+ var calls []string

+ plan9Fix2Systemctl(t, "enabled", &calls)

+ owner := plan9Fix2MaskOwner(t)

+ restore := claimSystemctl

+ claimSystemctl = func(ctx context.Context, args ...string) ([]byte, error) {

+ if args[0] == "mask" {

+ if raw, err := os.ReadFile(owner); err != nil || string(raw) != plan9Fix2Boot(t) {

+ t.Errorf("masked before the record was written: %q %v", raw, err)

+ }

+ return restore(ctx, args...)

+ }

+ release, err := quiesceLegacyClaimJournals(t.Context())

+ if err != nil {

+ t.Fatal(err)

+ }

+ release()

+ if !slices.Contains(calls, "mask --runtime carlos-edge.service") || !slices.Contains(calls, "unmask --runtime carlos-edge.service") {

+ t.Fatalf("calls: %v", calls)

+ }

+ if _, err = os.Stat(owner); !os.IsNotExist(err) {

+ t.Fatalf("mask record kept: %v", err)

+ }

+}

+

+// m-7: an operator's runtime mask, or one with a record from another boot,

+// is never lifted.

+func TestPlan9Fix2QuiesceKeepsOperatorMask(t *testing.T) {

+ for _, record := range []string{"", "another-boot"} {

+ var calls []string

+ plan9Fix2Systemctl(t, "masked-runtime", &calls)

+ owner := plan9Fix2MaskOwner(t)

+ if record != "" {

+ if err := os.WriteFile(owner, []byte(record), 0600); err != nil {

+ t.Fatal(err)

+ }

+ release, err := quiesceLegacyClaimJournals(t.Context())

+ if err != nil {

+ t.Fatal(err)

+ }

+ release()

+ for _, call := range calls {

+ if strings.HasPrefix(call, "unmask") || strings.HasPrefix(call, "mask") {

+ t.Fatalf("record %q: operator mask touched: %v", record, calls)

+ }

+ if _, err = os.Stat(owner); !os.IsNotExist(err) {

+ t.Fatalf("stale record kept: %v", err)

+ }

+}

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

new file mode 100644

index 000000000..1a5e0a7ae

--- /dev/null

+++ b/internal/host/plan9fix2_inventory_linux_test.go

@@ -0,0 +1,171 @@

+//go:build linux

+

+package host

+

+import (

+ "bytes"

+ "context"

+ "encoding/json"

+ "errors"

+ "net/http"

+ "net/http/httptest"

+ "os"

+ "path/filepath"

+ "syscall"

+ "testing"

+

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

+)

+

+func TestR2InventoryNamesForeignArtifact(t *testing.T) {

+ for _, rootName := range []string{"opt", "units", "data"} {

+ t.Run(rootName, func(t *testing.T) {

+ root, target, reg, base := plan9Fix1RootRetire(t)

+ row, err := reg.Get(target.Host)

+ if err != nil {

+ t.Fatal(err)

+ }

+ row.DBPath = filepath.Join(base, "instances", target.Host, target.Host+".db")

+ if err = reg.Upsert(*row); err != nil {

+ t.Fatal(err)

+ }

+ root.Roots[rootName] = base

+ name := "carlos"

+ if rootName == "units" {

+ name = "carlos-edge.service.d/guard.conf"

+ }

+ if rootName == "data" {

+ name = "instances/another-host/secret"

+ }

+ victim := filepath.Join(base, name)

+ if err = os.MkdirAll(filepath.Dir(victim), 0700); err != nil {

+ t.Fatal(err)

+ }

+ if err = os.WriteFile(victim, []byte("root-only"), 0600); err != nil {

+ t.Fatal(err)

+ }

+ var st syscall.Stat_t

+ if err = syscall.Lstat(victim, &st); err != nil {

+ t.Fatal(err)

+ }

+ in, err := root.Reconcile.Inventory.ReadExactStopInventory(t.Context(), target)

+ if err != nil {

+ t.Fatal(err)

+ }

+ in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "path", ArtifactID: "foreign", ManagedRoot: rootName, RelativePath: name, Ownership: row.ClaimOwner, Materialization: "materialized", Device: uint64(st.Dev), Inode: st.Ino, Type: st.Mode & syscall.S_IFMT})

+ // Simulate a durable pre-fix or migrated agent inventory.

+ if err = root.Reconcile.Inventory.WriteStopInventory(t.Context(), in); err != nil {

+ t.Fatal(err)

+ }

+ err = root.retire(t.Context(), target)

+ if !errors.Is(err, hostbox.ErrTypedInventory) {

+ t.Errorf("foreign inventory not refused at retire: %v", err)

+ }

+ if raw, e := os.ReadFile(victim); e != nil || string(raw) != "root-only" {

+ t.Errorf("root moved foreign file: %v", e)

+ }

+

+ j, e := hostbox.OpenLaunchJournal(filepath.Join(base, "service.db"))

+ if e != nil {

+ t.Fatal(e)

+ }

+ defer j.Close()

+ root.Journals, e = hostbox.NewJournalService(j)

+ if e != nil {

+ t.Fatal(e)

+ }

+ root.Reconcile.Box, root.Reconcile.Enrollment = target.Box, target.Enrollment

+ f := newRecoveryFixture(t)

+ c := f.read().Claim

+ root.Stop.Coordinator.Evidence = f.claims

+ in.Host, in.Incarnation, in.InstallationStamp = c.Host, c.Holder.Incarnation, c.Holder.Stamp

+ row.Host, row.ClaimIncarnation, row.ClaimStamp = in.Host, in.Incarnation, in.InstallationStamp

+ row.DBPath = filepath.Join(base, "instances", in.Host, in.Host+".db")

+ in.Entries[0].RelativePath = in.Host

+ if e = reg.Upsert(*row); e != nil {

+ t.Fatal(e)

+ }

+ target.Host, target.Incarnation, target.Stamp = in.Host, in.Incarnation, in.InstallationStamp

+ var q hostbox.JournalRequest

+ _, remote := hostbox.RemoteJournals(func(_ context.Context, request hostbox.JournalRequest) (hostbox.JournalResponse, error) {

+ q = request

+ return hostbox.JournalResponse{V: 1, ID: request.ID}, nil

+ })

+ if e = remote.WriteStopInventory(t.Context(), in); e != nil {

+ t.Fatal(e)

+ }

+ body, _ := json.Marshal(q)

+ response := httptest.NewRecorder()

+ root.serveJournal(response, httptest.NewRequest(http.MethodPost, "/claim/journal", bytes.NewReader(body)), StopPeer{})

+ var out hostbox.JournalResponse

+ _ = json.Unmarshal(response.Body.Bytes(), &out)

+ if response.Code == http.StatusOK && out.Error == "" {

+ t.Error("foreign inventory accepted at write")

+ }

+ if _, e = root.Journals.Inventory.ReadExactStopInventory(t.Context(), target); e == nil {

+ t.Error("foreign inventory became durable")

+ }

+ in.Entries = in.Entries[:1]

+ in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "path", ManagedRoot: "data", RelativePath: filepath.Join("instances", in.Host), ArtifactID: "state", Ownership: row.ClaimOwner, Materialization: "intended"})

+ if e = remote.WriteStopInventory(t.Context(), in); e != nil {

+ t.Fatal(e)

+ }

+ body, _ = json.Marshal(q)

+ response = httptest.NewRecorder()

+ root.serveJournal(response, httptest.NewRequest(http.MethodPost, "/claim/journal", bytes.NewReader(body)), StopPeer{})

+ out = hostbox.JournalResponse{}

+ if e = json.Unmarshal(response.Body.Bytes(), &out); e != nil || response.Code != http.StatusOK || out.Error != "" {

+ t.Fatalf("own inventory refused: %d %+v %v", response.Code, out, e)

+ }

+ if _, e = root.Journals.Inventory.ReadExactStopInventory(t.Context(), target); e != nil {

+ t.Fatalf("own inventory not durable: %v", e)

+ }

+ })

+ }

+}

+

+func TestPlan9Fix2RetireOwnStateAndRecoverArchivedRoute(t *testing.T) {

+ root, target, reg, base := plan9Fix1RootRetire(t)

+ row, err := reg.Get(target.Host)

+ if err != nil {

+ t.Fatal(err)

+ }

+ state := filepath.Join(base, "instances", target.Host)

+ row.DBPath = filepath.Join(state, target.Host+".db")

+ if err = reg.Upsert(*row); err != nil {

+ t.Fatal(err)

+ }

+ if err = os.MkdirAll(state, 0700); err != nil {

+ t.Fatal(err)

+ }

+ if err = os.WriteFile(row.DBPath, []byte("own-state"), 0600); err != nil {

+ t.Fatal(err)

+ }

+ var st syscall.Stat_t

+ if err = syscall.Lstat(state, &st); err != nil {

+ t.Fatal(err)

+ }

+ in, err := root.Reconcile.Inventory.ReadExactStopInventory(t.Context(), target)

+ if err != nil {

+ t.Fatal(err)

+ }

+ in.Entries = append(in.Entries, hostbox.StopInventoryEntry{Kind: "path", ArtifactID: "state", ManagedRoot: "data", RelativePath: filepath.Join("instances", target.Host), Ownership: row.ClaimOwner, Materialization: "materialized", Device: uint64(st.Dev), Inode: st.Ino, Type: st.Mode & syscall.S_IFMT})

+ if err = root.Reconcile.Inventory.WriteStopInventory(t.Context(), in); err != nil {

+ t.Fatal(err)

+ }

+ // This fixture deliberately fails only the final remote acknowledgement.

+ first := root.retire(t.Context(), target)

+ if !errors.Is(first, ErrClaimStopIdentity) {

+ t.Fatalf("own-state retirement: %v", first)

+ }

+ if _, err = os.Stat(state); !errors.Is(err, os.ErrNotExist) {

+ t.Fatalf("own state not parked: %v", err)

+ }

+ if row, err = reg.Get(target.Host); err != nil || row != nil {

+ t.Fatalf("route not removed: %+v %v", row, err)

+ }

+ // Recovery gets its DB path from root's archive after the live row is gone.

+ if err = root.retire(t.Context(), target); !errors.Is(err, ErrClaimStopIdentity) {

+ t.Fatalf("archive recovery: %v", err)

+ }

+}

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

new file mode 100644

index 000000000..5725701d7

--- /dev/null

+++ b/internal/host/plan9fix2_landing_phase_test.go

@@ -0,0 +1,59 @@

+package host

+

+import (

+ "context"

+ "encoding/json"

+ "testing"

+

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

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

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

+)

+

+type plan9Fix2PhaseStore struct {

+ release.Store

+ phases []string

+ key string

+}

+

+func (s *plan9Fix2PhaseStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {

+ next, err := s.Store.Put(ctx, key, body, etag)

+ if err == nil && key == s.key {

+ var c hostclaim.Claim

+ if json.Unmarshal(body, &c) == nil {

+ s.phases = append(s.phases, c.Op.Phase)

+ }

+ return next, err

+}

+

+func TestPlan9Fix2LandingCatchUpPersistsEachAdjacentStep(t *testing.T) {

+ h := newLandingHarness(t)

+ h.flip(t)

+ h.lw = &movesWorker{cfg: h.landingCfg(t)}

+ key, err := hostclaim.Key(p1Host)

+ if err != nil {

+ t.Fatal(err)

+ }

+ store := &plan9Fix2PhaseStore{Store: h.s, key: key}

+ rec := h.read(t, p1Host)

+ rec.Move.LandingPhase = provision.LandProbed

+ run := landingRun{w: h.lw, s: store, rec: rec, g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}

+ before := h.claim(t)

+ if _, err = run.syncClaimPhase(t.Context()); err != nil {

+ t.Fatal(err)

+ }

+ order := provision.LandingOrder()

+ if len(store.phases) != len(order)-1 {

+ t.Fatalf("catch-up skipped durable phase transitions: %v", store.phases)

+ }

+ for i, phase := range store.phases {

+ if phase != order[i+1] {

+ t.Fatalf("transition %d: %s", i, phase)

+ }

+ after := h.claim(t)

+ if before.Holder.Session != after.Holder.Session || before.Holder.Stamp != after.Holder.Stamp || before.Holder.Generation != after.Holder.Generation {

+ t.Fatal("catch-up changed admitted writer")

+ }

+}

diff --git a/internal/hostbox/coldhandover.go b/internal/hostbox/coldhandover.go

index d5fe8e72f..475380a54 100644

--- a/internal/hostbox/coldhandover.go

+++ b/internal/hostbox/coldhandover.go

@@ -142,6 +142,9 @@ func (h ColdHandover) Run(ctx context.Context, prior hostguard.FenceTarget, next

if row == nil || !row.ClaimOwnership.Valid() || row.ClaimIncarnation != prior.Incarnation || row.ClaimStamp != prior.Stamp || row.Kind != "instance" || row.Unit != "" || row.Addr != "" || row.Addrs != "" || row.Backing != "" || row.Tunnel != "" || row.Platform || row.Canary || row.Replica {

return ErrUnsupported

}

+ if err := ValidateAgentInventory(in, row, h.Roots["data"]); err != nil {

+ return err

+ }

timers, err := h.Registry.TimersFor(prior.Host)

if err != nil {

return err

@@ -169,6 +172,9 @@ func (h ColdHandover) Run(ctx context.Context, prior hostguard.FenceTarget, next

if j.Phase == "complete" {

return h.Guard.CompleteColdHandover(ctx, j.Barrier, true, check, func() error { return nil })

}

+ if err := ValidateAgentInventory(j.Inventory, &j.Before.Route, h.Roots["data"]); err != nil {

+ return err

+ }

if h.afterIntent != nil && j.Phase == "prepared" {

if err := h.afterIntent(); err != nil {

return err

diff --git a/internal/hostbox/intended_recovery_test.go b/internal/hostbox/intended_recovery_test.go

index 9d6f636d1..ea7abcde7 100644

--- a/internal/hostbox/intended_recovery_test.go

+++ b/internal/hostbox/intended_recovery_test.go

@@ -502,3 +502,35 @@ func TestClaimReviewHandoverRefusesCompletedSuccessorProducer(t *testing.T) {

}

f.requireUnmoved(t)

}

+

+func TestPlan9Fix2ColdHandoverRefusesForeignInventory(t *testing.T) {

+ for _, root := range []string{"opt", "units", "data"} {

+ t.Run(root, func(t *testing.T) {

+ f := newIntendedFixture(t, nil)

+ victim := filepath.Join(f.data, "foreign")

+ if err := os.WriteFile(victim, []byte("root-only"), 0600); err != nil {

+ t.Fatal(err)

+ }

+ f.handover.Roots[root] = f.data

+ parker := TypedParker{Roots: f.handover.Roots, ParkedRoot: f.data, Incarnation: f.prior.Incarnation, Host: f.prior.Host, Retirement: "capture"}

+ a, err := parker.CaptureArtifact(root, "foreign", "old-owner", ArtifactPath)

+ if err != nil {

+ t.Fatal(err)

+ }

+ in, err := f.inventory.ReadExactStopInventory(t.Context(), f.prior)

+ if err != nil {

+ t.Fatal(err)

+ }

+ in.Entries = append(in.Entries, StopInventoryEntry{Kind: "path", ManagedRoot: root, RelativePath: a.Name, ArtifactID: "foreign", Ownership: a.Owner, Device: a.Dev, Inode: a.Ino, Type: a.Type, Materialization: "materialized"})

+ if err = f.inventory.WriteStopInventory(t.Context(), in); err != nil {

+ t.Fatal(err)

+ }

+ if err = f.handover.Run(t.Context(), f.prior, f.next); !errors.Is(err, ErrTypedInventory) {

+ t.Fatalf("foreign inventory: %v", err)

+ }

+ if raw, err := os.ReadFile(victim); err != nil || string(raw) != "root-only" {

+ t.Fatalf("foreign file moved: %v", err)

+ }

+ })

+ }

+}

diff --git a/internal/hostbox/inventoryscope.go b/internal/hostbox/inventoryscope.go

new file mode 100644

index 000000000..71746c434

--- /dev/null

+++ b/internal/hostbox/inventoryscope.go

@@ -0,0 +1,116 @@

+package hostbox

+

+import (

+ "encoding/json"

+ "errors"

+ "os"

+ "path/filepath"

+ "strings"

+

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

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

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

+)

+

+// ValidateAgentInventory confines agent-authored artifacts to this route's

+// state directory and registry rows. Opt and units need root-captured authority;

+// an inventory written by the agent never supplies that authority.

+func ValidateAgentInventory(in StopInventory, row *registry.Route, dataDir string) error {

+ if !component(in.Host) {

+ return ErrTypedInventory

+ }

+ if row != nil && (row.Host != in.Host || row.ClaimIncarnation != in.Incarnation || row.ClaimStamp != in.InstallationStamp || !row.ClaimOwnership.Valid()) {

+ return ErrTypedInventory

+ }

+ seen := make(map[string]bool)

+ for _, e := range in.Entries {

+ key := e.ManagedRoot + "\x00" + e.RelativePath

+ if seen[key] || !managedName(e.RelativePath) || (row != nil && e.Ownership != row.ClaimOwner) {

+ return ErrTypedInventory

+ }

+ seen[key] = true

+ switch e.Kind {

+ case "route":

+ if e.ManagedRoot != "registry" || e.RelativePath != in.Host || e.RowIdentity == "" || (row != nil && e.RowIdentity != row.ClaimRowID) || (row == nil && e.Materialization != "intended") {

+ return ErrTypedInventory

+ }

+ case "timer":

+ if row == nil || e.ManagedRoot != "registry" || !strings.HasPrefix(e.RelativePath, in.Host+"/") || !component(strings.TrimPrefix(e.RelativePath, in.Host+"/")) || e.RowIdentity == "" {

+ return ErrTypedInventory

+ }

+ case "path":

+ if row == nil || e.ManagedRoot != "data" || !filepath.IsAbs(dataDir) || !filepath.IsAbs(row.DBPath) {

+ return ErrTypedInventory

+ }

+ if row.DBPath != provision.NestedDBPath(dataDir, in.Host) && row.DBPath != provision.DefaultDBPath(dataDir, in.Host) {

+ return ErrTypedInventory

+ }

+ first := strings.SplitN(e.RelativePath, string(filepath.Separator), 2)[0]

+ if first == ".parked" || first == "hostguard" || first == ".hostclaims" || first == ".hostexec" || first == ".instances" || strings.HasPrefix(first, "registry.db") {

+ return ErrTypedInventory

+ }

+ parent, err := filepath.Rel(dataDir, filepath.Dir(row.DBPath))

+ if err != nil || (parent != "." && !managedName(parent)) {

+ return ErrTypedInventory

+ }

+ if parent == "." {

+ if first == "instances" {

+ return ErrTypedInventory

+ }

+ // Flat legacy layouts use the host-named state directory plus

+ // the database and its exact SQLite companion names.

+ db := filepath.Base(row.DBPath)

+ if db != in.Host+".db" || (e.RelativePath != in.Host && !strings.HasPrefix(e.RelativePath, in.Host+"/") && e.RelativePath != db && e.RelativePath != db+"-wal" && e.RelativePath != db+"-shm" && e.RelativePath != db+"-journal") {

+ return ErrTypedInventory

+ }

+ } else if e.RelativePath != parent && !strings.HasPrefix(e.RelativePath, parent+string(filepath.Separator)) {

+ return ErrTypedInventory

+ }

+ default:

+ return ErrTypedInventory

+ }

+ return nil

+}

+

+// retirementRoute recovers the row root archived before removing it. The

+// agent's inventory carries no database path and cannot choose this namespace.

+func (a ReconcileAdapter) retirementRoute(row registry.Route, target hostguard.FenceTarget) (*registry.Route, error) {

+ current, err := a.Registry.Get(row.Host)

+ if err != nil {

+ return nil, err

+ }

+ if current != nil && current.ClaimOwnership == row.ClaimOwnership {

+ return current, nil

+ }

+ raw, err := readTrustedJournal(filepath.Join(a.Retirement.JournalDir, target.FenceID+".json"))

+ if errors.Is(err, os.ErrNotExist) {

+ return nil, ErrTypedInventory

+ }

+ if err != nil {

+ return nil, err

+ }

+ var journal RetirementJournal

+ if decodeStrict(raw, &journal) != nil || journal.Target == nil || *journal.Target != target || journal.ID != target.FenceID {

+ return nil, ErrTypedJournal

+ }

+ var archives []registry.ClaimRowArchive

+ if json.Unmarshal(journal.Archives, &archives) != nil {

+ return nil, ErrTypedJournal

+ }

+ var found *registry.Route

+ for _, archive := range archives {

+ if archive.Row.Kind != "route" {

+ continue

+ }

+ var saved registry.Route

+ if found != nil || archive.Row.Host != row.Host || archive.Row.Ownership != row.ClaimOwnership || json.Unmarshal(archive.Contents, &saved) != nil || saved.Host != row.Host || saved.ClaimOwnership != row.ClaimOwnership {

+ return nil, ErrTypedJournal

+ }

+ found = &saved

+ }

+ if found == nil {

+ return nil, ErrTypedInventory

+ }

+ return found, nil

+}

diff --git a/internal/hostbox/journalmigration_linux.go b/internal/hostbox/journalmigration_linux.go

index 0aa05100b..96c40cace 100644

--- a/internal/hostbox/journalmigration_linux.go

+++ b/internal/hostbox/journalmigration_linux.go

@@ -22,17 +22,99 @@ import (

"golang.org/x/sys/unix"

)

-// MigrateLegacy runs only before serving requests. Quiesce must prove the old

+// ErrLegacyJournalMigration refuses root startup over legacy claim journals

+// that were never imported. Only the offline operator step imports them.

+var ErrLegacyJournalMigration = errors.New("legacy claim journals need the offline migration: run carlos system journal-migrate")

+

+// CheckLegacyMigration is root startup's only contact with the legacy

+// journals, which live in a directory the agent owns. It never opens them as

+// SQLite and never quiesces anything. With no legacy files and no marker (a

+// fresh box), it records the marker. Unimported legacy files refuse with

+// ErrLegacyJournalMigration. Once the marker exists, nothing the agent puts

+// there can stop root: legacy files that differ from the marked content are

+// ignored, and the returned alert says so.

+func (s *JournalService) CheckLegacyMigration(ctx context.Context, legacy string) (alert error, err error) {

+ if s == nil || s.Launches == nil || !filepath.IsAbs(legacy) {

+ return nil, ErrTypedJournal

+ }

+ unlock, err := lockJournalMigration(ctx, filepath.Join(filepath.Dir(s.Launches.path), "migration.lock"))

+ if err != nil {

+ return nil, err

+ }

+ defer unlock()

+ source, markedFiles, _, marked, err := s.journalMigrationMarker(ctx)

+ if err != nil {

+ return nil, err

+ }

+ files, present, err := legacyJournalFiles(legacy)

+ if marked {

+ if source != legacy {

+ return nil, ErrChanged

+ }

+ if err != nil {

+ return fmt.Errorf("legacy claim journal %s ignored after migration: %w", legacy, err), nil

+ }

+ if present == 0 || legacyContent(markedFiles) == files {

+ return nil, nil

+ }

+ return fmt.Errorf("legacy claim journal %s changed after migration and is ignored: %w", legacy, ErrChanged), nil

+ }

+ if err != nil {

+ return nil, err

+ }

+ if present != 0 {

+ return nil, ErrLegacyJournalMigration

+ }

+ tx, err := s.Launches.db.BeginTx(ctx, nil)

+ if err != nil {

+ return nil, err

+ }

+ defer tx.Rollback()

+ var count int

+ if err = tx.QueryRowContext(ctx, "SELECT (SELECT count(*) FROM launches)+(SELECT count(*) FROM stop_inventory)").Scan(&count); err != nil || count != 0 {

+ return nil, errors.Join(err, ErrTypedJournal)

+ }

+ if _, err = tx.ExecContext(ctx, "INSERT INTO journal_migration(version,source,files,rows) VALUES(1,?,?,?)", legacy, files, legacyRowsDigest(nil, nil)); err != nil {

+ return nil, err

+ }

+ return nil, tx.Commit()

+}

+

+// journalMigrationMarker creates or upgrades the marker table and reads it.

+func (s *JournalService) journalMigrationMarker(ctx context.Context) (source, files, rows string, marked bool, err error) {

+ if _, err = s.Launches.db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS journal_migration(version INTEGER PRIMARY KEY CHECK(version=1), source TEXT NOT NULL, files TEXT NOT NULL DEFAULT '', rows TEXT NOT NULL DEFAULT '')"); err != nil {

+ return

+ }

+ for _, column := range []string{"files", "rows"} {

+ var n int

+ if err = s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM pragma_table_info('journal_migration') WHERE name=?", column).Scan(&n); err != nil {

+ return

+ }

+ if n == 0 {

+ if _, err = s.Launches.db.ExecContext(ctx, "ALTER TABLE journal_migration ADD COLUMN "+column+" TEXT NOT NULL DEFAULT ''"); err != nil {

+ return

+ }

+ err = s.Launches.db.QueryRowContext(ctx, "SELECT source,files,rows FROM journal_migration WHERE version=1").Scan(&source, &files, &rows)

+ if errors.Is(err, sql.ErrNoRows) {

+ return "", "", "", false, nil

+ }

+ return source, files, rows, err == nil, err

+}

+

+// MigrateLegacy is the offline operator step (carlos system journal-migrate);

+// root startup only runs CheckLegacyMigration. Quiesce must prove the old

// writers are stopped and held stopped for this call. Root never opens SQLite

// on the old paths: bounded no-follow copies, including WAL, are made in the

-// protected directory and opened with an untrusted schema, read-only. Every

-// imported field is validated and re-marshalled; no raw body is stored. The

-// transaction imports both inventories and the marker together. A crash before

-// commit imports nothing; a crash afterward never imports a stale legacy

-// database again. The marker binds the legacy files' identity and the digest

-// of the rows it imported: if a rolled-back binary later wrote rows there,

-// the next startup refuses rather than silently ignoring them. Legacy

-// originals remain untouched.

+// protected directory, hashed as they are copied, and opened with an untrusted

+// schema, read-only. Every imported field is validated and re-marshalled; no

+// raw body is stored. The transaction imports both inventories and the marker

+// together. A crash before commit imports nothing; a crash afterward never

+// imports a stale legacy database again. The marker binds the digests of the

+// copied legacy bytes and of the rows it imported: if a rolled-back binary

+// later wrote rows there, this step refuses rather than silently ignoring

+// them. Legacy originals remain untouched.

func (s *JournalService) MigrateLegacy(ctx context.Context, legacy, box, enrollment string, quiesce func(context.Context) (func(), error), verify func(context.Context, LaunchInventory, StopInventory, StopKernelJournal) error) error {

if s == nil || s.Launches == nil || !filepath.IsAbs(legacy) || box == "" || enrollment == "" {

return ErrTypedJournal

@@ -55,20 +137,10 @@ func (s *JournalService) MigrateLegacy(ctx context.Context, legacy, box, enrollm

}

- if _, err := s.Launches.db.ExecContext(ctx, "CREATE TABLE IF NOT EXISTS journal_migration(version INTEGER PRIMARY KEY CHECK(version=1), source TEXT NOT NULL, files TEXT NOT NULL DEFAULT '', rows TEXT NOT NULL DEFAULT '')"); err != nil {

+ source, markedFiles, markedRows, marked, err := s.journalMigrationMarker(ctx)

+ if err != nil {

return err

}

- for _, column := range []string{"files", "rows"} {

- var n int

- if err = s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM pragma_table_info('journal_migration') WHERE name=?", column).Scan(&n); err != nil {

- return err

- }

- if n == 0 {

- if _, err = s.Launches.db.ExecContext(ctx, "ALTER TABLE journal_migration ADD COLUMN "+column+" TEXT NOT NULL DEFAULT ''"); err != nil {

- return err

- }

files, present, err := legacyJournalFiles(legacy)

if err != nil {

return err

@@ -76,20 +148,15 @@ func (s *JournalService) MigrateLegacy(ctx context.Context, legacy, box, enrollm

if present != 0 && present != 2 {

return ErrTypedJournal

}

- var source, markedFiles, markedRows string

- err = s.Launches.db.QueryRowContext(ctx, "SELECT source,files,rows FROM journal_migration WHERE version=1").Scan(&source, &markedFiles, &markedRows)

- marked := err == nil

if marked {

if source != legacy {

return ErrChanged

}

- // Unchanged legacy files, or legacy files removed after import: nothing

- // can have been written there since the marker.

- if files == markedFiles || present == 0 {

+ // Unchanged legacy content, or legacy files removed after import:

+ // nothing can have been written there since the marker.

+ if present == 0 || legacyContent(markedFiles) == files {

return nil

}

- } else if !errors.Is(err, sql.ErrNoRows) {

- return err

}

if present == 2 {

if quiesce == nil {

@@ -115,27 +182,30 @@ func (s *JournalService) MigrateLegacy(ctx context.Context, legacy, box, enrollm

// Prepare and verify bounded immutable records while no root SQLite

// connection or write transaction is held. The offline flock and edge

// exclusion remain held until the later atomic import finishes.

- launches, inventories, err := readLegacyJournals(ctx, legacy, filepath.Dir(s.Launches.path), box, enrollment)

- if err != nil {

- return err

- }

- // The snapshot must be exactly the files the identity names.

- after, _, err := legacyJournalFiles(legacy)

+ launches, inventories, copied, err := readLegacyJournals(ctx, legacy, filepath.Dir(s.Launches.path), box, enrollment)

if err != nil {

return err

}

- if after != files {

+ // The rows came from exactly the bytes the identity names: the digests

+ // are of the copies themselves, not of a later re-read.

+ if copied != files {

return fmt.Errorf("legacy journal changed while copied: %w", ErrChanged)

}

rows := legacyRowsDigest(launches, inventories)

if marked {

- // The files changed after import. Only identical rows (a touched file,

- // a checkpoint) keep the marker; new history from a rolled-back binary

- // needs the operator, never a silent skip or an unproven import.

- if markedRows == "" || rows != markedRows {

+ // The content changed after import. Only identical rows (a checkpoint)

+ // keep the marker; new history from a rolled-back binary needs the

+ // operator, never a silent skip or an unproven import. A marker from

+ // before rows were bound (4f342ec7) is bound now, but only when root

+ // already holds every legacy row.

+ if markedRows == "" {

+ if err = s.holdsLegacyRows(ctx, launches, inventories); err != nil {

+ return fmt.Errorf("legacy claim journal %s changed after migration: %w", legacy, err)

+ }

+ } else if rows != markedRows {

return fmt.Errorf("legacy claim journal %s changed after migration: %w", legacy, ErrChanged)

}

- _, err = s.Launches.db.ExecContext(ctx, "UPDATE journal_migration SET files=? WHERE version=1 AND rows=?", files, rows)

+ _, err = s.Launches.db.ExecContext(ctx, "UPDATE journal_migration SET files=?, rows=? WHERE version=1 AND rows=?", files, rows, markedRows)

return err

}

var count int

@@ -223,10 +293,41 @@ const (

legacyRowsMax = 10000

)

-// legacyJournalFiles names the exact legacy files: device, inode, size, change

-// times and a SHA-256 of the bytes of each database and WAL. File times are

-// coarse, so the digest is what proves the bytes are unchanged. Only no-follow,

-// single-link regular files within the copy bound are read.

+// holdsLegacyRows proves every legacy row's key is already in root's journal.

+func (s *JournalService) holdsLegacyRows(ctx context.Context, launches []LaunchInventory, inventories map[string]StopInventory) error {

+ for _, in := range launches {

+ var n int

+ if err := s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM launches WHERE identity=?", launchKey(in.Target)).Scan(&n); err != nil || n != 1 {

+ return errors.Join(err, ErrChanged)

+ }

+ for _, in := range inventories {

+ var n int

+ if err := s.Launches.db.QueryRowContext(ctx, "SELECT count(*) FROM stop_inventory WHERE host=? AND incarnation=? AND stamp=?", in.Host, in.Incarnation, in.InstallationStamp).Scan(&n); err != nil || n != 1 {

+ return errors.Join(err, ErrChanged)

+ }

+ return nil

+}

+

+// legacyContent reduces a marker's file identity to names and content

+// digests. Markers from 618f3a8a also carry device, inode, size and times;

+// the digest is their last field. A metadata-only change is not a change.

+func legacyContent(files string) string {

+ if files == "" || files == "absent" {

+ return files

+ }

+ parts := strings.Split(files, "/")

+ for i, part := range parts {

+ fields := strings.Split(part, ":")

+ parts[i] = fields[0] + ":" + fields[len(fields)-1]

+ }

+ return strings.Join(parts, "/")

+}

+

+// legacyJournalFiles names the exact legacy files by the SHA-256 of the bytes

+// of each database and WAL ("-" when absent). Only no-follow, single-link

+// regular files within the copy bound are read.

func legacyJournalFiles(legacy string) (string, int, error) {

dir, err := unix.Open(legacy, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)

if errors.Is(err, unix.ENOENT) {

@@ -261,7 +362,7 @@ func legacyJournalFiles(legacy string) (string, int, error) {

if n, err := io.Copy(h, io.LimitReader(f, legacyJournalFileMax+1)); err != nil || n != st.Size {

return "", errors.Join(err, ErrChanged)

}

- return fmt.Sprintf("%s:%d:%d:%d:%d.%d:%d.%d:%x", name, st.Dev, st.Ino, st.Size, st.Mtim.Sec, st.Mtim.Nsec, st.Ctim.Sec, st.Ctim.Nsec, h.Sum(nil)), nil

+ return fmt.Sprintf("%s:%x", name, h.Sum(nil)), nil

}()

if err != nil {

return "", 0, err

@@ -290,35 +391,38 @@ func legacyRowsDigest(launches []LaunchInventory, inventories map[string]StopInv

return hex.EncodeToString(sum[:])

}

-func readLegacyJournals(ctx context.Context, legacy, dest, box, enrollment string) ([]LaunchInventory, map[string]StopInventory, error) {

+// readLegacyJournals also returns the content identity of the bytes it copied,

+// in legacyJournalFiles' form.

+func readLegacyJournals(ctx context.Context, legacy, dest, box, enrollment string) ([]LaunchInventory, map[string]StopInventory, string, error) {

var launches []LaunchInventory

keys := map[string]bool{}

inventories := map[string]StopInventory{}

preparedBytes := 0

+ var identity []string

for _, name := range []string{"launches.db", "inventory.db"} {

- copied, cleanup, err := copyLegacyJournal(legacy, name, dest)

+ copied, digests, cleanup, err := copyLegacyJournal(legacy, name, dest)

if errors.Is(err, os.ErrNotExist) {

+ identity = append(identity, name+":-", name+"-wal:-")

continue

}

if err != nil {

- return nil, nil, err

+ return nil, nil, "", err

}

+ identity = append(identity, digests...)

err = func() error {

defer cleanup()

// The copy is agent-supplied bytes: no schema-defined SQL runs, no

- // statement writes, and the whole read is time bounded.

+ // statement writes, and the whole read is time bounded. The schema

+ // must be exactly the legacy code's before integrity_check reads a

+ // row, and integrity_check never evaluates a CHECK expression.

readCtx, cancel := context.WithTimeout(ctx, 2*time.Minute)

defer cancel()

- db, err := sql.Open("sqlite", "file:"+url.PathEscape(copied)+"?_pragma=trusted_schema(0)&_pragma=query_only(1)&_pragma=busy_timeout(5000)")

+ db, err := sql.Open("sqlite", "file:"+url.PathEscape(copied)+"?_pragma=trusted_schema(0)&_pragma=query_only(1)&_pragma=ignore_check_constraints(1)&_pragma=busy_timeout(5000)")

if err != nil {

return err

}

defer db.Close()

db.SetMaxOpenConns(1)

- var result string

- if err = db.QueryRowContext(readCtx, "PRAGMA integrity_check").Scan(&result); err != nil || result != "ok" {

- return errors.Join(err, ErrTypedJournal)

- }

tables := map[string]bool{"launches": true}

if name == "inventory.db" {

tables = map[string]bool{"stop_inventory": true, "journal_inventory_versions": true}

@@ -326,6 +430,10 @@ func readLegacyJournals(ctx context.Context, legacy, dest, box, enrollment strin

if err = checkLegacySchema(readCtx, db, tables); err != nil {

return err

}

+ var result string

+ if err = db.QueryRowContext(readCtx, "PRAGMA integrity_check").Scan(&result); err != nil || result != "ok" {

+ return errors.Join(err, ErrTypedJournal)

+ }

// Size every scanned cell before Scan allocates it. octet_length

// reads the stored length without loading the value.

bounds := "SELECT count(*), coalesce(sum(octet_length(body)),0), coalesce(max(octet_length(body)),0), coalesce(max(octet_length(identity)),0), 0 FROM launches"

@@ -393,32 +501,45 @@ func readLegacyJournals(ctx context.Context, legacy, dest, box, enrollment strin

return rows.Err()

}()

if err != nil {

- return nil, nil, err

+ return nil, nil, "", err

}

- return launches, inventories, nil

+ return launches, inventories, strings.Join(identity, "/"), nil

+}

+

+// legacyTableSQL is, byte for byte, the CREATE text each legacy table was

+// written with (SQLite stores it without IF NOT EXISTS).

+var legacyTableSQL = map[string]string{

+ "launches": "CREATE TABLE launches (identity TEXT PRIMARY KEY, body BLOB NOT NULL)",

+ "stop_inventory": "CREATE TABLE stop_inventory(host TEXT NOT NULL,incarnation TEXT NOT NULL,stamp INTEGER NOT NULL,body BLOB NOT NULL,PRIMARY KEY(host,incarnation,stamp))",

+ "journal_inventory_versions": "CREATE TABLE journal_inventory_versions(host TEXT NOT NULL, incarnation TEXT NOT NULL, stamp INTEGER NOT NULL, revision INTEGER NOT NULL CHECK(revision>0 AND typeof(revision)='integer'), request TEXT NOT NULL, PRIMARY KEY(host,incarnation,stamp))",

}

-// checkLegacySchema admits only the named tables and their implicit primary

-// key indexes. Triggers, views, virtual tables and any other schema object

-// refuse, as does any generated or hidden column: those run expressions on

-// SELECT.

+// checkLegacySchema admits only the named tables, each with exactly its legacy

+// CREATE text and a real root page, and their implicit primary key indexes,

+// which have no SQL text. Triggers, views, virtual tables (root page zero,

+// whatever the text says), expression indexes and any other schema object

+// refuse, so no schema-defined expression is left for a read to run.

func checkLegacySchema(ctx context.Context, db *sql.DB, tables map[string]bool) error {

- rows, err := db.QueryContext(ctx, "SELECT type,name,tbl_name,coalesce(sql,'') FROM sqlite_schema")

+ rows, err := db.QueryContext(ctx, "SELECT type,name,tbl_name,rootpage,sql FROM sqlite_schema")

if err != nil {

- return err

+ return errors.Join(err, ErrTypedJournal)

}

defer rows.Close()

var present []string

for rows.Next() {

- var kind, name, table, text string

- if err = rows.Scan(&kind, &name, &table, &text); err != nil {

- return err

+ var kind, name, table string

+ var root sql.NullInt64

+ var text sql.NullString

+ if err = rows.Scan(&kind, &name, &table, &root, &text); err != nil {

+ return errors.Join(err, ErrTypedJournal)

}

switch {

- case kind == "table" && tables[name] && name == table && !strings.HasPrefix(strings.ToUpper(strings.TrimSpace(text)), "CREATE VIRTUAL"):

+ case !root.Valid || root.Int64 <= 0:

+ return ErrTypedJournal

+ case kind == "table" && tables[name] && name == table && text.Valid && text.String == legacyTableSQL[name]:

present = append(present, name)

- case kind == "index" && tables[table] && strings.HasPrefix(name, "sqlite_autoindex_"+table+"_"):

+ case kind == "index" && tables[table] && strings.HasPrefix(name, "sqlite_autoindex_"+table+"_") && !text.Valid:

default:

return ErrTypedJournal

}

@@ -545,28 +666,40 @@ func validateImportedInventory(in StopInventory) error {

return nil

}

-func copyLegacyJournal(dir, name, dest string) (string, func(), error) {

+// legacyCopyHook lets tests act on a legacy file just before and just after

+// it is copied.

+var legacyCopyHook func(name string, copied bool)

+

+// copyLegacyJournal copies the database and its WAL, returning each file's

+// identity part ("name:sha256", or "name:-" for an absent WAL) computed from

+// the bytes written to the copy.

+func copyLegacyJournal(dir, name, dest string) (string, []string, func(), error) {

// A directory fd anchors the source components. The quiescence boundary,

// not a pair of pathname checks, makes the copied snapshot coherent.

fd, err := unix.Open(dir, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)

if err != nil {

- return "", nil, err

+ return "", nil, nil, err

}

defer unix.Close(fd)

temp, err := os.MkdirTemp(dest, "migration-")

if err != nil {

- return "", nil, err

+ return "", nil, nil, err

}

clean := func() { _ = os.RemoveAll(temp) }

path := filepath.Join(temp, name)

+ var digests []string

for _, suffix := range []string{"", "-wal"} {

+ if legacyCopyHook != nil {

+ legacyCopyHook(name+suffix, false)

+ }

f, e := unix.Openat(fd, name+suffix, unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_NONBLOCK|unix.O_CLOEXEC, 0)

if errors.Is(e, unix.ENOENT) && suffix != "" {

+ digests = append(digests, name+suffix+":-")

continue

}

if e != nil {

clean()

- return "", nil, e

+ return "", nil, nil, e

}

source := os.NewFile(uintptr(f), name+suffix)

var st unix.Stat_t

@@ -574,24 +707,29 @@ func copyLegacyJournal(dir, name, dest string) (string, func(), error) {

if e != nil || st.Mode&unix.S_IFMT != unix.S_IFREG || st.Nlink != 1 || st.Size > legacyJournalFileMax {

source.Close()

clean()

- return "", nil, ErrTypedJournal

+ return "", nil, nil, ErrTypedJournal

}

target, e := os.OpenFile(path+suffix, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0600)

if e != nil {

source.Close()

clean()

- return "", nil, e

+ return "", nil, nil, e

}

- n, e := io.Copy(target, io.LimitReader(source, legacyJournalFileMax+1))

+ h := sha256.New()

+ n, e := io.Copy(io.MultiWriter(target, h), io.LimitReader(source, legacyJournalFileMax+1))

syncErr := target.Sync()

closeErr := target.Close()

source.Close()

if e != nil || syncErr != nil || closeErr != nil || n != st.Size {

clean()

- return "", nil, fmt.Errorf("journal snapshot: %w", errors.Join(e, syncErr, closeErr, ErrChanged))

+ return "", nil, nil, fmt.Errorf("journal snapshot: %w", errors.Join(e, syncErr, closeErr, ErrChanged))

+ }

+ digests = append(digests, fmt.Sprintf("%s:%x", name+suffix, h.Sum(nil)))

+ if legacyCopyHook != nil {

+ legacyCopyHook(name+suffix, true)

}

- return path, clean, nil

+ return path, digests, clean, nil

}

func lockJournalMigration(ctx context.Context, path string) (func(), error) {

diff --git a/internal/hostbox/journalmigration_other.go b/internal/hostbox/journalmigration_other.go

index 02ad389c6..a1c533310 100644

--- a/internal/hostbox/journalmigration_other.go

+++ b/internal/hostbox/journalmigration_other.go

@@ -2,7 +2,16 @@

package hostbox

-import "context"

+import (

+ "context"

+ "errors"

+)

+

+var ErrLegacyJournalMigration = errors.New("legacy claim journals need the offline migration: run carlos system journal-migrate")

+

+func (s *JournalService) CheckLegacyMigration(context.Context, string) (error, error) {

+ return nil, ErrUnsupported

+}

func (s *JournalService) MigrateLegacy(context.Context, string, string, string, func(context.Context) (func(), error), func(context.Context, LaunchInventory, StopInventory, StopKernelJournal) error) error {

return ErrUnsupported

diff --git a/internal/hostbox/journalservice.go b/internal/hostbox/journalservice.go

index 258ba7bd0..8465aed1e 100644

--- a/internal/hostbox/journalservice.go

+++ b/internal/hostbox/journalservice.go

@@ -109,19 +109,21 @@ func journalCall(ctx context.Context, call JournalCall, req JournalRequest) (Jou

if err != nil {

return JournalResponse{}, err

}

- return journalSend(ctx, call, req)

+ return journalSend(ctx, call, req, false)

}

-// ReplayUncertain resends the exact request an uncertain call retained. Only

-// a committed receipt answers it once its execution window has passed.

+// ReplayUncertain resends the exact request an uncertain call retained. The

+// earlier attempt may have committed, so the replay stays uncertain on every

+// refusal, including after the receipt was pruned. Only a reply that names

+// this exact request with no error confirms it.

func (j *LaunchJournal) ReplayUncertain(ctx context.Context, uncertain *JournalUncertainError) (JournalResponse, error) {

if j == nil || j.remote == nil || uncertain == nil || uncertain.Request.ID == "" {

return JournalResponse{}, ErrUnsupported

}

- return journalSend(ctx, j.remote, uncertain.Request)

+ return journalSend(ctx, j.remote, uncertain.Request, true)

}

-func journalSend(ctx context.Context, call JournalCall, req JournalRequest) (JournalResponse, error) {

+func journalSend(ctx context.Context, call JournalCall, req JournalRequest, replay bool) (JournalResponse, error) {

out, err := call(ctx, req)

if err != nil {

return out, &JournalUncertainError{Request: req, Cause: err}

@@ -132,6 +134,11 @@ func journalSend(ctx context.Context, call JournalCall, req JournalRequest) (Jou

switch out.Error {

case "":

return out, nil

+ }

+ if replay {

+ return out, &JournalUncertainError{Request: req, Cause: ErrJournalUncertain}

+ }

+ switch out.Error {

case "missing":

return out, sql.ErrNoRows

case "changed":

@@ -295,7 +302,7 @@ func NewJournalService(j *LaunchJournal) (*JournalService, error) {

}

for _, q := range []string{inventoryVersionsSQL,

"CREATE TABLE IF NOT EXISTS stop_inventory(host TEXT NOT NULL,incarnation TEXT NOT NULL,stamp INTEGER NOT NULL,body BLOB NOT NULL,PRIMARY KEY(host,incarnation,stamp))",

- "CREATE TABLE IF NOT EXISTS journal_requests(id TEXT PRIMARY KEY, peer TEXT NOT NULL, digest TEXT NOT NULL, response BLOB NOT NULL, created INTEGER NOT NULL DEFAULT 0)",

+ "CREATE TABLE IF NOT EXISTS journal_requests(id TEXT PRIMARY KEY, peer TEXT NOT NULL, digest TEXT NOT NULL, response BLOB NOT NULL, created INTEGER NOT NULL DEFAULT 0, issued INTEGER NOT NULL DEFAULT 0)",

"CREATE TABLE IF NOT EXISTS journal_receipt_retention(id INTEGER PRIMARY KEY CHECK(id=1), reject_before INTEGER NOT NULL)",

"INSERT INTO journal_receipt_retention(id,reject_before) VALUES(1,0) ON CONFLICT(id) DO NOTHING",

} {

@@ -314,6 +321,15 @@ func NewJournalService(j *LaunchJournal) (*JournalService, error) {

return nil, err

}

+ var issued int

+ if err := j.db.QueryRow("SELECT count(*) FROM pragma_table_info('journal_requests') WHERE name='issued'").Scan(&issued); err != nil {

+ return nil, err

+ }

+ if issued == 0 {

+ if _, err := j.db.Exec("ALTER TABLE journal_requests ADD COLUMN issued INTEGER NOT NULL DEFAULT 0"); err != nil {

+ return nil, err

+ }

if _, err := j.db.Exec("CREATE INDEX IF NOT EXISTS journal_requests_created ON journal_requests(created)"); err != nil {

return nil, err

}

@@ -407,20 +423,18 @@ func (s *JournalService) Execute(ctx context.Context, peer string, q JournalRequ

if err = tx.QueryRowContext(ctx, "SELECT reject_before FROM journal_receipt_retention WHERE id=1").Scan(&rejectBefore); err != nil {

return out, err

}

- // Persist the rejection boundary with receipt deletion. A clock reversal

- // or root restart cannot make a pruned request executable again. The

- // allowance covers requests issued ahead of root's clock when committed.

- rejectBefore = max(rejectBefore, now.Add(-JournalReceiptRetention+JournalRequestWindow).UnixNano())

+ // The rejection boundary only ever rises to the issue time of a receipt

+ // this transaction deletes, so a pruned request can never execute again

+ // whatever the clock does, and a forward clock jump cannot raise it past

+ // real time: it covers only requests that actually committed.

if q.V == 2 {

age := now.Sub(time.Unix(0, q.IssuedAt))

if q.IssuedAt <= rejectBefore || age > JournalRequestWindow || age < -JournalRequestWindow {

return out, ErrJournalUncertain

}

- if _, err = tx.ExecContext(ctx, "UPDATE journal_receipt_retention SET reject_before=? WHERE id=1", rejectBefore); err != nil {

- return out, err

- }

- if _, err = tx.ExecContext(ctx, "DELETE FROM journal_requests WHERE rowid IN (SELECT rowid FROM journal_requests WHERE created < ? ORDER BY created LIMIT ?)", rejectBefore-int64(JournalRequestWindow), journalPruneBatch); err != nil {

+ rejectBefore, err = journalPrune(ctx, tx, rejectBefore, now)

+ if err != nil {

return out, err

}

ctx = context.WithValue(ctx, journalTransactionKey{}, tx)

@@ -464,7 +478,7 @@ func (s *JournalService) Execute(ctx context.Context, peer string, q JournalRequ

if err != nil {

return out, err

}

- if _, err = tx.ExecContext(ctx, "INSERT INTO journal_requests(id,peer,digest,response,created) VALUES(?,?,?,?,?)", q.ID, peer, digest, body, now.UnixNano()); err != nil {

+ if _, err = tx.ExecContext(ctx, "INSERT INTO journal_requests(id,peer,digest,response,created,issued) VALUES(?,?,?,?,?,?)", q.ID, peer, digest, body, now.UnixNano(), q.IssuedAt); err != nil {

return out, err

}

if err = tx.Commit(); err != nil {

@@ -501,3 +515,44 @@ func (s *JournalService) read(ctx context.Context, q JournalRequest) (JournalRes

}

return out, err

}

+

+// journalPrune deletes one bounded batch of receipts past retention and raises

+// the persisted boundary to cover their requests. A request executed within

+// JournalRequestWindow of its issue time, so an unrecorded issue time is at

+// most that far past the receipt's creation.

+func journalPrune(ctx context.Context, tx *sql.Tx, rejectBefore int64, now time.Time) (int64, error) {

+ cutoff := max(rejectBefore-int64(JournalRequestWindow), now.Add(-JournalReceiptRetention).UnixNano())

+ rows, err := tx.QueryContext(ctx, "SELECT rowid,created,issued FROM journal_requests WHERE created < ? ORDER BY created LIMIT ?", cutoff, journalPruneBatch)

+ if err != nil {

+ return rejectBefore, err

+ }

+ var ids []int64

+ covered := rejectBefore

+ for rows.Next() {

+ var id, created, issued int64

+ if err = rows.Scan(&id, &created, &issued); err != nil {

+ rows.Close()

+ return rejectBefore, err

+ }

+ if issued <= 0 {

+ issued = created + int64(JournalRequestWindow)

+ }

+ ids = append(ids, id)

+ covered = max(covered, issued)

+ }

+ if err = errors.Join(rows.Err(), rows.Close()); err != nil {

+ return rejectBefore, err

+ }

+ if len(ids) == 0 {

+ return rejectBefore, nil

+ }

+ if _, err = tx.ExecContext(ctx, "UPDATE journal_receipt_retention SET reject_before=? WHERE id=1", covered); err != nil {

+ return rejectBefore, err

+ }

+ for _, id := range ids {

+ if _, err = tx.ExecContext(ctx, "DELETE FROM journal_requests WHERE rowid=?", id); err != nil {

+ return rejectBefore, err

+ }

+ return covered, nil

+}

diff --git a/internal/hostbox/plan9_fix2_minors_test.go b/internal/hostbox/plan9_fix2_minors_test.go

new file mode 100644

index 000000000..5fad18c10

--- /dev/null

+++ b/internal/hostbox/plan9_fix2_minors_test.go

@@ -0,0 +1,164 @@

+package hostbox

+

+import (

+ "context"

+ "errors"

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

+ "testing"

+ "time"

+)

+

+// m-9: a replay keeps its uncertain state. Whatever typed refusal the transport

+// reports, the exact request may already have committed.

+func TestPlan9Fix2ReplayRefusalStaysUncertain(t *testing.T) {

+ for _, code := range []string{"refused", "changed", "missing", "unsupported", "boom"} {

+ t.Run(code, func(t *testing.T) {

+ client, _ := RemoteJournals(func(_ context.Context, q JournalRequest) (JournalResponse, error) {

+ return JournalResponse{V: 1, ID: q.ID, Error: code}, nil

+ })

+ q, err := journalPrepare(JournalRequest{Op: "producer"}, time.Now())

+ if err != nil {

+ t.Fatal(err)

+ }

+ _, err = client.ReplayUncertain(t.Context(), &JournalUncertainError{Request: q})

+ var uncertain *JournalUncertainError

+ if !errors.As(err, &uncertain) || uncertain.Request.ID != q.ID {

+ t.Fatalf("%s refusal on replay read as definite: %v", code, err)

+ }

+ if errors.Is(err, ErrChanged) || errors.Is(err, ErrTypedInventory) {

+ t.Fatalf("%s refusal on replay leaked a typed refusal: %v", code, err)

+ }

+ })

+ }

+}

+

+// m-9: success is only a positive reply for this exact request.

+func TestPlan9Fix2ReplaySuccessNeedsExactReceipt(t *testing.T) {

+ q, err := journalPrepare(JournalRequest{Op: "producer"}, time.Now())

+ if err != nil {

+ t.Fatal(err)

+ }

+ client, _ := RemoteJournals(func(context.Context, JournalRequest) (JournalResponse, error) {

+ return JournalResponse{V: 1, ID: "ffffffffffffffffffffffffffffffff"}, nil

+ })

+ var uncertain *JournalUncertainError

+ if _, err = client.ReplayUncertain(t.Context(), &JournalUncertainError{Request: q}); !errors.As(err, &uncertain) {

+ t.Fatalf("a reply for another request confirmed the replay: %v", err)

+ }

+ client, _ = RemoteJournals(func(_ context.Context, r JournalRequest) (JournalResponse, error) {

+ return JournalResponse{V: 1, ID: r.ID}, nil

+ })

+ if _, err = client.ReplayUncertain(t.Context(), &JournalUncertainError{Request: q}); err != nil {

+ t.Fatalf("exact confirmation refused: %v", err)

+ }

+}

+

+// m-11: a forward jump of the host clock that is later corrected must not

+// stall new mutations until real time catches up, and must not let a pruned

+// request execute again.

+func TestPlan9Fix2ForwardClockJumpThenCorrectionKeepsLiveness(t *testing.T) {

+ s, _, in := journalFixture(t)

+ origin := time.Now()

+ now := origin

+ s.Now = func() time.Time { return now }

+ issue := func(q JournalRequest) JournalRequest {

+ t.Helper()

+ q, err := journalPrepare(q, now)

+ if err != nil {

+ t.Fatal(err)

+ }

+ return q

+ }

+ if _, err := s.Execute(t.Context(), "peer", issue(JournalRequest{Op: "create", Launch: in})); err != nil {

+ t.Fatal(err)

+ }

+ old := issue(JournalRequest{Op: "producer", Target: in.Target, Producer: "old"})

+ if _, err := s.Execute(t.Context(), "peer", old); err != nil {

+ t.Fatal(err)

+ }

+ now = origin.Add(5 * 24 * time.Hour)

+ if _, err := s.Execute(t.Context(), "peer", issue(JournalRequest{Op: "producer", Target: in.Target, Producer: "jumped"})); err != nil {

+ t.Fatal(err)

+ }

+ var n int

+ if err := s.Launches.db.QueryRow("SELECT count(*) FROM journal_requests WHERE id=?", old.ID).Scan(&n); err != nil || n != 0 {

+ t.Fatalf("jump did not prune the old receipt: %d %v", n, err)

+ }

+ now = origin.Add(2 * time.Minute)

+ if _, err := s.Execute(t.Context(), "peer", issue(JournalRequest{Op: "producer", Target: in.Target, Producer: "corrected"})); err != nil {

+ t.Fatalf("mutation stalled after the clock was corrected: %v", err)

+ }

+ if _, err := s.Execute(t.Context(), "peer", old); !errors.Is(err, ErrJournalUncertain) {

+ t.Fatalf("a pruned request executed again: %v", err)

+ }

+}

+

+// m-2: an admitted writer that never checked probed may still observe the

+// committed phase when op, kind, session, stamp and generation are unchanged.

+// The transition itself still cannot skip a step.

+func TestPlan9Fix2SelectedWorkObservesCommittedAfterMissingProbed(t *testing.T) {

+ work, _, update := plan9D1Probe(t, hostclaim.SessionObserves)

+ if err := plan9D1Write(work); err != nil {

+ t.Fatal(err)

+ }

+ update(hostclaim.SetMovePhase("m1", hostclaim.KindLanding, hostclaim.PhaseLandingProbing, hostclaim.PhaseLandingProbed))

+ update(hostclaim.KeepWriterSession("m1", hostclaim.SetMovePhase("m1", hostclaim.KindLanding, hostclaim.PhaseLandingProbed, hostclaim.PhaseLandingCommitted)))

+ if err := plan9D1Write(work); err != nil {

+ t.Fatalf("the admitted writer lost its session across an unobserved probed: %v", err)

+ }

+}

+

+func TestPlan9Fix2SelectedWorkStillRefusesIdentityChangeAndBackwardSteps(t *testing.T) {

+ for name, change := range map[string]func(*hostclaim.Claim) error{

+ "session": func(c *hostclaim.Claim) error {

+ c.Op.Phase = hostclaim.PhaseLandingCommitted

+ c.Holder.Session++

+ return nil

+ },

+ "op": func(c *hostclaim.Claim) error {

+ c.Op.Phase = hostclaim.PhaseLandingCommitted

+ c.Op.ID = "m2"

+ return nil

+ },

+ "gen": func(c *hostclaim.Claim) error {

+ c.Op.Phase = hostclaim.PhaseLandingCommitted

+ c.Holder.Generation = "g2"

+ return nil

+ },

+ "rollback": func(c *hostclaim.Claim) error {

+ c.Op.Kind, c.Op.Phase = hostclaim.KindRestoring, "releasing"

+ return nil

+ },

+ } {

+ t.Run(name, func(t *testing.T) {

+ work, _, update := plan9D1Probe(t, hostclaim.SessionObserves)

+ update(change)

+ if err := plan9D1Write(work); err == nil {

+ t.Fatal("observation catch-up kept a changed session")

+ }

+ })

+ }

+ work, _, update := plan9D1Probe(t, hostclaim.SessionObserves)

+ update(hostclaim.SetMovePhase("m1", hostclaim.KindLanding, hostclaim.PhaseLandingProbing, hostclaim.PhaseLandingProbed))

+ if err := plan9D1Write(work); err != nil {

+ t.Fatal(err)

+ }

+ update(func(c *hostclaim.Claim) error { c.Op.Phase = hostclaim.PhaseLandingProbing; return nil })

+ if err := plan9D1Write(work); err == nil {

+ t.Fatal("observation went backward")

+ }

+}

+

+// m-2: the keep guard is transition authority and still refuses a skip.

+func TestPlan9Fix2KeepWriterSessionStillRefusesASkippedTransition(t *testing.T) {

+ _, store, update := plan9D1Probe(t, hostclaim.SessionObserves)

+ current, err := hostclaim.Read(t.Context(), store, "shop.example")

+ if err != nil {

+ t.Fatal(err)

+ }

+ _, err = hostclaim.Update(t.Context(), store, current, hostclaim.KeepWriterSession("m1", hostclaim.SetMovePhase("m1", hostclaim.KindLanding, hostclaim.PhaseLandingProbing, hostclaim.PhaseLandingCommitted)))

+ if err == nil {

+ t.Fatal("a session-keeping step skipped probed")

+ }

+ _ = update

+}

diff --git a/internal/hostbox/plan9fix2_inventory_test.go b/internal/hostbox/plan9fix2_inventory_test.go

new file mode 100644

index 000000000..91323d807

--- /dev/null

+++ b/internal/hostbox/plan9fix2_inventory_test.go

@@ -0,0 +1,90 @@

+package hostbox

+

+import (

+ "path/filepath"

+ "testing"

+

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

+)

+

+func TestPlan9Fix2InventoryNamespace(t *testing.T) {

+ data := t.TempDir()

+ row := registry.Route{Host: "app.example", DBPath: filepath.Join(data, "instances", "app.example", "app.example.db"), ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"}}

+ base := StopInventory{Host: row.Host, Incarnation: row.ClaimIncarnation, InstallationStamp: row.ClaimStamp, Complete: true}

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

+ name, kind, root, path, id string

+ want bool

+ }{

+ {"state", "path", "data", "instances/app.example", "", true},

+ {"child", "path", "data", "instances/app.example/app.example.db", "", true},

+ {"route", "route", "registry", "app.example", "row", true},

+ {"timer", "timer", "registry", "app.example/daily", "timer", true},

+ {"foreign state", "path", "data", "instances/other.example", "", false},

+ {"prefix sibling", "path", "data", "instances/app.example-other", "", false},

+ {"data root", "path", "data", ".", "", false},

+ {"escape", "path", "data", "instances/app.example/../other.example", "", false},

+ {"foreign route", "route", "registry", "other.example", "row", false},

+ {"foreign row id", "route", "registry", "app.example", "other-row", false},

+ {"foreign timer", "timer", "registry", "other.example/daily", "timer", false},

+ {"unit", "path", "units", "carlos-edge.service.d/guard.conf", "", false},

+ {"binary", "path", "opt", "carlos", "", false},

+ {"unknown kind", "unit-state", "data", "instances/app.example", "", false},

+ } {

+ t.Run(tc.name, func(t *testing.T) {

+ in := base

+ in.Entries = []StopInventoryEntry{{Kind: tc.kind, ManagedRoot: tc.root, RelativePath: filepath.FromSlash(tc.path), RowIdentity: tc.id, Ownership: "owner", Materialization: "materialized"}}

+ if err := ValidateAgentInventory(in, &row, data); (err == nil) != tc.want {

+ t.Fatalf("allowed=%v: %v", tc.want, err)

+ }

+ })

+ }

+ // A missing row can record the initial route intent, but grants no path authority.

+ base.Entries = []StopInventoryEntry{{Kind: "route", ManagedRoot: "registry", RelativePath: row.Host, RowIdentity: "new-row", Ownership: "owner", Materialization: "intended"}}

+ if err := ValidateAgentInventory(base, nil, data); err != nil {

+ t.Fatal(err)

+ }

+ base.Entries[0] = StopInventoryEntry{Kind: "path", ManagedRoot: "data", RelativePath: "instances/app.example", Ownership: "owner", Materialization: "intended"}

+ if err := ValidateAgentInventory(base, nil, data); err == nil {

+ t.Fatal("missing row granted path authority")

+ }

+ row.DBPath = filepath.Join(data, row.Host+".db")

+ for _, name := range []string{row.Host, row.Host + ".db", row.Host + ".db-wal", row.Host + ".db-shm"} {

+ base.Entries[0].RelativePath = name

+ if err := ValidateAgentInventory(base, &row, data); err != nil {

+ t.Fatalf("legacy %s: %v", name, err)

+ }

+ for _, name := range []string{".", ".hostclaims", row.Host + "-other", row.Host + ".db-other"} {

+ base.Entries[0].RelativePath = name

+ if err := ValidateAgentInventory(base, &row, data); err == nil {

+ t.Fatalf("legacy accepted foreign %s", name)

+ }

+}

+

+func TestPlan9Fix2AgentCannotChooseNamespaceThroughRegistryDBPath(t *testing.T) {

+ data := t.TempDir()

+ row := registry.Route{Host: "app.example", ClaimOwnership: registry.ClaimOwnership{ClaimIncarnation: "inc", ClaimStamp: 1, ClaimOwner: "owner", ClaimRowID: "row"}}

+ for _, parent := range []string{"instances/other.example", ".parked/other", ".hostexec", "hostguard", ".hostclaims", ".instances", "registry.db-backup"} {

+ t.Run(parent, func(t *testing.T) {

+ row.DBPath = filepath.Join(data, filepath.FromSlash(parent), "app.example.db")

+ in := StopInventory{Host: row.Host, Incarnation: "inc", InstallationStamp: 1, Entries: []StopInventoryEntry{{Kind: "path", ManagedRoot: "data", RelativePath: filepath.FromSlash(parent), Ownership: "owner", Materialization: "materialized"}}}

+ if err := ValidateAgentInventory(in, &row, data); err == nil {

+ t.Fatalf("agent chose root namespace %s through DBPath", parent)

+ }

+ })

+ }

+ row.Host = "hostguard"

+ row.DBPath = filepath.Join(data, "hostguard.db")

+ in := StopInventory{Host: row.Host, Incarnation: "inc", InstallationStamp: 1, Entries: []StopInventoryEntry{{Kind: "path", ManagedRoot: "data", RelativePath: "hostguard", Ownership: "owner", Materialization: "materialized"}}}

+ if err := ValidateAgentInventory(in, &row, data); err == nil {

+ t.Fatal("flat hostname selected root hostguard namespace")

+ }

+ row.Host = "instances"

+ row.DBPath = filepath.Join(data, "instances.db")

+ in.Host = row.Host

+ in.Entries[0].RelativePath = "instances/other.example"

+ if err := ValidateAgentInventory(in, &row, data); err == nil {

+ t.Fatal("flat hostname selected shared instances namespace")

+ }

+}

diff --git a/internal/hostbox/plan9fix2_journalmigration_linux_test.go b/internal/hostbox/plan9fix2_journalmigration_linux_test.go

new file mode 100644

index 000000000..d1ca42b6d

--- /dev/null

+++ b/internal/hostbox/plan9fix2_journalmigration_linux_test.go

@@ -0,0 +1,306 @@

+//go:build linux

+

+package hostbox

+

+import (

+ "crypto/sha256"

+ "database/sql"

+ "encoding/hex"

+ "errors"

+ "fmt"

+ "net/url"

+ "os"

+ "path/filepath"

+ "strings"

+ "syscall"

+ "testing"

+ "time"

+)

+

+// plan9Fix2Schema rewrites a legacy file's schema on one connection, as an

+// agent holding the file could.

+func plan9Fix2Schema(t *testing.T, path string, stmts ...string) {

+ t.Helper()

+ db, err := sql.Open("sqlite", "file:"+url.PathEscape(path)+"?_pragma=busy_timeout(5000)")

+ if err != nil {

+ t.Fatal(err)

+ }

+ defer db.Close()

+ db.SetMaxOpenConns(1)

+ for _, stmt := range append([]string{"PRAGMA writable_schema=ON"}, stmts...) {

+ if _, err = db.Exec(stmt); err != nil {

+ t.Fatal(stmt, err)

+ }

+}

+

+// I2-1: no expression in an agent-supplied schema is evaluated. A CHECK that

+// would allocate beyond SQLite's limit proves it: evaluating it fails with

+// "string or blob too big", so the refusal must come from the schema

+// allowlist, before integrity_check reads a row.

+func TestPlan9Fix2MigrationRefusesExpensiveCheckBeforeEvaluation(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ plan9Fix2Schema(t, filepath.Join(legacy, "launches.db"),

+ "UPDATE sqlite_schema SET sql='CREATE TABLE launches (identity TEXT PRIMARY KEY, body BLOB NOT NULL, CHECK(length(hex(zeroblob(2000000000)))>0))' WHERE name='launches'")

+ err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept)

+ if !errors.Is(err, ErrTypedJournal) {

+ t.Fatalf("planted CHECK accepted: %v", err)

+ }

+ if strings.Contains(err.Error(), "too big") {

+ t.Fatalf("planted CHECK was evaluated before the schema was refused: %v", err)

+ }

+}

+

+// I2-1 and m-6: a table entry with root page zero is a virtual table or view

+// whatever its text says; it is refused even when the text matches.

+func TestPlan9Fix2MigrationRefusesRootpageZero(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ plan9Fix2Schema(t, filepath.Join(legacy, "inventory.db"), "UPDATE sqlite_schema SET rootpage=0 WHERE name='journal_inventory_versions'")

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrTypedJournal) {

+ t.Fatalf("root page zero accepted: %v", err)

+ }

+ if got, err := s.Launches.HasHost(t.Context(), in.Target.Host); err != nil || got {

+ t.Fatalf("import committed: %v %v", got, err)

+ }

+}

+

+// I2-1: each allowed table must carry exactly the text the legacy code wrote.

+func TestPlan9Fix2MigrationRefusesAlteredTableText(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ plan9Fix2Schema(t, filepath.Join(legacy, "launches.db"),

+ "UPDATE sqlite_schema SET sql='CREATE TABLE launches (identity TEXT PRIMARY KEY, body BLOB NOT NULL )' WHERE name='launches'")

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrTypedJournal) {

+ t.Fatalf("altered table text accepted: %v", err)

+ }

+}

+

+// m-8: the marker binds content. A metadata-only change (ctime, mtime) to an

+// imported legacy file needs neither quiescence nor a refusal.

+func TestPlan9Fix2MarkerIgnoresMetadataOnlyChange(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatal(err)

+ }

+ later := time.Now().Add(time.Hour)

+ for _, name := range []string{"launches.db", "inventory.db"} {

+ path := filepath.Join(legacy, name)

+ if err := os.Chtimes(path, later, later); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.Chmod(path, 0640); err != nil {

+ t.Fatal(err)

+ }

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", nil, nil); err != nil {

+ t.Fatalf("metadata-only change refused: %v", err)

+ }

+}

+

+// m-8: a marker written before the marker had a binding (4f342ec7, files and

+// rows both empty after the column upgrade) is accepted when every legacy row

+// is one root already holds; the binding is then recorded, and later legacy

+// writes are refused as before.

+func TestPlan9Fix2MigrationAcceptsUnboundMarker(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatal(err)

+ }

+ if _, err := s.Launches.db.Exec("UPDATE journal_migration SET files='', rows=''"); err != nil {

+ t.Fatal(err)

+ }

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatalf("unbound marker refused: %v", err)

+ }

+ var files, rows string

+ if err := s.Launches.db.QueryRow("SELECT files,rows FROM journal_migration").Scan(&files, &rows); err != nil || files == "" || rows == "" {

+ t.Fatalf("binding not recorded: %q %q %v", files, rows, err)

+ }

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", nil, nil); err != nil {

+ t.Fatalf("bound marker refused: %v", err)

+ }

+ old, err := OpenLaunchJournal(filepath.Join(legacy, "launches.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ second := in

+ second.Target.Session = 2

+ if err = old.Create(t.Context(), second); err != nil {

+ t.Fatal(err)

+ }

+ old.Close()

+ if err = s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrChanged) {

+ t.Fatalf("post-marker legacy write accepted: %v", err)

+ }

+}

+

+// m-8: an unbound marker is not a licence to skip history root never held.

+func TestPlan9Fix2UnboundMarkerRefusesRowsRootNeverHeld(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatal(err)

+ }

+ if _, err := s.Launches.db.Exec("UPDATE journal_migration SET files='', rows=''"); err != nil {

+ t.Fatal(err)

+ }

+ old, err := OpenLaunchJournal(filepath.Join(legacy, "launches.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ second := in

+ second.Target.Session = 2

+ if err = old.Create(t.Context(), second); err != nil {

+ t.Fatal(err)

+ }

+ old.Close()

+ if err = s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrChanged) {

+ t.Fatalf("unbound marker skipped unheld history: %v", err)

+ }

+}

+

+// m-8: a marker from 618f3a8a binds device, inode and times beside the

+// digest. Only its digests are compared.

+func TestPlan9Fix2MigrationAcceptsMetadataBoundMarker(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatal(err)

+ }

+ var parts []string

+ for _, name := range []string{"launches.db", "launches.db-wal", "inventory.db", "inventory.db-wal"} {

+ raw, err := os.ReadFile(filepath.Join(legacy, name))

+ if errors.Is(err, os.ErrNotExist) {

+ parts = append(parts, name+":-")

+ continue

+ }

+ if err != nil {

+ t.Fatal(err)

+ }

+ st, _ := os.Stat(filepath.Join(legacy, name))

+ sys := st.Sys().(*syscall.Stat_t)

+ parts = append(parts, fmt.Sprintf("%s:%d:%d:%d:%d.%d:%d.%d:%s", name, sys.Dev, sys.Ino, st.Size(), sys.Mtim.Sec, sys.Mtim.Nsec, sys.Ctim.Sec+1, sys.Ctim.Nsec, plan9Fix2Digest(raw)))

+ }

+ if _, err := s.Launches.db.Exec("UPDATE journal_migration SET files=?", strings.Join(parts, "/")); err != nil {

+ t.Fatal(err)

+ }

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", nil, nil); err != nil {

+ t.Fatalf("618f3a8a marker with unchanged content refused: %v", err)

+ }

+}

+

+// m-10: the imported rows come from exactly the bytes the marker names. A

+// legacy file swapped for the copy and restored afterwards must refuse.

+func TestPlan9Fix2MigrationBindsCopiedBytes(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ other := plan9Fix1Legacy(t, in)

+ second := in

+ second.Target.Session = 2

+ old, err := OpenLaunchJournal(filepath.Join(other, "launches.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err = old.Create(t.Context(), second); err != nil {

+ t.Fatal(err)

+ }

+ old.Close()

+ original, err := os.ReadFile(filepath.Join(legacy, "launches.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ swapped, err := os.ReadFile(filepath.Join(other, "launches.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ restore := legacyCopyHook

+ t.Cleanup(func() { legacyCopyHook = restore })

+ legacyCopyHook = func(name string, copied bool) {

+ if name != "launches.db" {

+ return

+ }

+ body := swapped

+ if copied {

+ body = original

+ }

+ if err := os.WriteFile(filepath.Join(legacy, name), body, 0600); err != nil {

+ t.Error(err)

+ }

+ if err = s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrChanged) {

+ t.Fatalf("copy of different bytes imported: %v", err)

+ }

+ if _, err = s.Launches.Read(t.Context(), second.Target); !errors.Is(err, sql.ErrNoRows) {

+ t.Fatalf("swapped row imported: %v", err)

+ }

+}

+

+// I2-3: after the marker, legacy files under the agent's directory never stop

+// root's startup. Planted files are ignored and reported; nothing is quiesced.

+func TestPlan9Fix2StartupIgnoresLegacyFilesPlantedAfterMarker(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := t.TempDir()

+ if alert, err := s.CheckLegacyMigration(t.Context(), legacy); err != nil || alert != nil {

+ t.Fatalf("fresh startup: %v %v", alert, err)

+ }

+ // The agent plants a full legacy pair after the marker.

+ planted := plan9Fix1Legacy(t, in)

+ for _, name := range []string{"launches.db", "inventory.db"} {

+ raw, err := os.ReadFile(filepath.Join(planted, name))

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err = os.WriteFile(filepath.Join(legacy, name), raw, 0600); err != nil {

+ t.Fatal(err)

+ }

+ alert, err := s.CheckLegacyMigration(t.Context(), legacy)

+ if err != nil {

+ t.Fatalf("planted legacy files stopped root: %v", err)

+ }

+ if alert == nil {

+ t.Fatal("planted legacy files not reported")

+ }

+ if got, err := s.Launches.HasHost(t.Context(), in.Target.Host); err != nil || got {

+ t.Fatalf("planted rows imported: %v %v", got, err)

+ }

+ // Garbage and special files are ignored the same way.

+ if err = os.Remove(filepath.Join(legacy, "launches.db")); err != nil {

+ t.Fatal(err)

+ }

+ if err = syscall.Mkfifo(filepath.Join(legacy, "launches.db"), 0600); err != nil {

+ t.Fatal(err)

+ }

+ if alert, err = s.CheckLegacyMigration(t.Context(), legacy); err != nil || alert == nil {

+ t.Fatalf("planted fifo: %v %v", alert, err)

+ }

+}

+

+// I2-3: unmigrated legacy history is not silently dropped: startup refuses

+// until the operator runs the offline migration.

+func TestPlan9Fix2StartupRefusesUnmigratedLegacy(t *testing.T) {

+ s, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ if _, err := s.CheckLegacyMigration(t.Context(), legacy); !errors.Is(err, ErrLegacyJournalMigration) {

+ t.Fatalf("unmigrated legacy journals accepted at startup: %v", err)

+ }

+ var n int

+ if err := s.Launches.db.QueryRow("SELECT count(*) FROM journal_migration").Scan(&n); err != nil || n != 0 {

+ t.Fatalf("marker written over unmigrated history: %d %v", n, err)

+ }

+ if err := s.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); err != nil {

+ t.Fatal(err)

+ }

+ if alert, err := s.CheckLegacyMigration(t.Context(), legacy); err != nil || alert != nil {

+ t.Fatalf("migrated startup: %v %v", alert, err)

+ }

+}

+

+func plan9Fix2Digest(raw []byte) string {

+ sum := sha256.Sum256(raw)

+ return hex.EncodeToString(sum[:])

+}

diff --git a/internal/hostbox/plan9fix2_schema_size_linux_test.go b/internal/hostbox/plan9fix2_schema_size_linux_test.go

new file mode 100644

index 000000000..05308b8ac

--- /dev/null

+++ b/internal/hostbox/plan9fix2_schema_size_linux_test.go

@@ -0,0 +1,20 @@

+//go:build linux

+

+package hostbox

+

+import (

+ "errors"

+ "path/filepath"

+ "testing"

+)

+

+// Use the reviewer's exact expression. The larger CHECK regression separately

+// proves the allowlist runs before SQLite can evaluate any such expression.

+func TestPlan9Fix2MigrationRefusesReviewerCheck(t *testing.T) {

+ service, _, in := journalFixture(t)

+ legacy := plan9Fix1Legacy(t, in)

+ plan9Fix2Schema(t, filepath.Join(legacy, "launches.db"), "UPDATE sqlite_schema SET sql='CREATE TABLE launches (identity TEXT PRIMARY KEY, body BLOB NOT NULL, CHECK(length(hex(zeroblob(150000000)))>0))' WHERE name='launches'")

+ if err := service.MigrateLegacy(t.Context(), legacy, "box", "enrollment", plan9Fix1Freeze, plan9Fix1Accept); !errors.Is(err, ErrTypedJournal) {

+ t.Fatalf("reviewer CHECK was not refused: %v", err)

+ }

+}

diff --git a/internal/hostbox/reconcile_adapter.go b/internal/hostbox/reconcile_adapter.go

index 90112bc75..ceb403ebc 100644

--- a/internal/hostbox/reconcile_adapter.go

+++ b/internal/hostbox/reconcile_adapter.go

@@ -21,6 +21,7 @@ import (

)

type ReconcileAdapter struct {

+ DataRoot string

RemoteRetire func(context.Context, hostguard.FenceTarget) error

Census *EmptyInstallationCensus

Reconciler Reconciler

@@ -184,6 +185,13 @@ func (a ReconcileAdapter) Retire(ctx context.Context, row registry.Route) error

if !in.Complete || in.Box != target.Box || in.Enrollment != target.Enrollment || in.DirectoryRevision != target.DirectoryRevision {

return ErrTypedInventory

}

+ trustedRow, err := a.retirementRoute(row, target)

+ if err != nil {

+ return err

+ }

+ if err = ValidateAgentInventory(in, trustedRow, a.inventoryDataRoot()); err != nil {

+ return err

+ }

inventory := RetirementInventory{Host: row.Host, Captured: hostguard.Captured{Incarnation: row.ClaimIncarnation, Stamp: row.ClaimStamp}, Ownership: row.ClaimOwner}

for _, e := range in.Entries {

if e.Ownership != row.ClaimOwner || e.Materialization != "materialized" {

@@ -252,3 +260,10 @@ func (a ReconcileAdapter) Retire(ctx context.Context, row registry.Route) error

retirement.Rows = RegistryRows{Registry: a.Registry, Inventory: inventory}

return retirement.Run(ctx, target, intent)

}

+

+func (a ReconcileAdapter) inventoryDataRoot() string {

+ if a.DataRoot != "" {

+ return a.DataRoot

+ }

+ return a.Reconciler.Guard.DataDir

+}

diff --git a/internal/hostbox/work.go b/internal/hostbox/work.go

index 8b9f82abd..70e1d61f6 100644

--- a/internal/hostbox/work.go

+++ b/internal/hostbox/work.go

@@ -31,7 +31,7 @@ type WorkSource struct {

// Local checks quarantine, execution deadlines and failed cleanup under flock.

Local func(context.Context, *hostclaim.Claim, Selection) error

// Continue reports whether a selection last checked at one op phase may

- // keep working at another (hostclaim.SessionContinues). Nil keeps main's rule:

+ // keep working at another (hostclaim.SessionObserves). Nil keeps main's rule:

// any phase change ends the selection. Only the moves C1 integration sets

// it, and only with CARLOS_HOST_CLAIM and CARLOS_MOVES_PHASE1 both on.

Continue func(kind, from, to string) bool

diff --git a/internal/hostclaim/plan9_fix2_observe_test.go b/internal/hostclaim/plan9_fix2_observe_test.go

new file mode 100644

index 000000000..152f91eb0

--- /dev/null

+++ b/internal/hostclaim/plan9_fix2_observe_test.go

@@ -0,0 +1,30 @@

+package hostclaim

+

+import "testing"

+

+// m-2: transition authority and observation catch-up are different questions.

+func TestPlan9Fix2ObservationCatchesUpAcrossAdjacentTableEdgesOnly(t *testing.T) {

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

+ kind, from, to string

+ observes bool

+ }{

+ {KindLanding, PhaseLandingProbing, PhaseLandingProbing, true},

+ {KindLanding, PhaseLandingProbing, PhaseLandingProbed, true},

+ {KindLanding, PhaseLandingProbing, PhaseLandingCommitted, true},

+ {KindLanding, PhaseLandingProbed, PhaseLandingCommitted, true},

+ {KindLanding, PhaseLandingCommitted, PhaseLandingProbed, false},

+ {KindLanding, PhaseLandingProbed, PhaseLandingProbing, false},

+ {KindLanding, PhaseLandingProbing, "restored", false},

+ {KindLanding, "restored", PhaseLandingCommitted, false},

+ {KindRestoring, "releasing", "released", false},

+ {"", PhaseLandingProbing, PhaseLandingProbed, false},

+ } {

+ if got := SessionObserves(c.kind, c.from, c.to); got != c.observes {

+ t.Errorf("SessionObserves(%q,%q,%q) = %v, want %v", c.kind, c.from, c.to, got, c.observes)

+ }

+ // Write authority for an actual transition still never skips a step.

+ if SessionContinues(KindLanding, PhaseLandingProbing, PhaseLandingCommitted) {

+ t.Fatal("SessionContinues allows a skipped transition")

+ }

+}

diff --git a/internal/hostclaim/sessionstep.go b/internal/hostclaim/sessionstep.go

index a3e124f8b..0a8fb83f9 100644

--- a/internal/hostclaim/sessionstep.go

+++ b/internal/hostclaim/sessionstep.go

@@ -29,9 +29,8 @@ var sessionSteps = map[string][]string{

// SessionContinues reports whether a writer session last checked while the op

// of this kind was at phase from may keep working now that it is at phase to:

// the same phase or the adjacent forward step of its row, never a skip (M9).

-// A session that passes several steps keeps working only if a check observed

-// each one, so callers compare against the last phase they checked, not the

-// phase they selected at.

+// This is transition authority. A reader that missed a step uses

+// SessionObserves, comparing against the last phase it checked.

func SessionContinues(kind, from, to string) bool {

if from == to {

return true

@@ -45,6 +44,16 @@ func SessionContinues(kind, from, to string) bool {

return false

}

+// SessionObserves reports whether a writer session last checked at phase from

+// may observe the op at phase to: the same phase or any later phase on kind's

+// row, so a session that missed an intermediate phase can still catch up. It

+// grants observation only. Callers also bind op id, kind, session, stamp and

+// generation, and every transition that changes the claim is still guarded by

+// SessionContinues (KeepWriterSession), which never skips a step.

+func SessionObserves(kind, from, to string) bool {

+ return SessionReaches(kind, from, to)

+}

+

// SessionPhase reports whether phase is on kind's session row: a phase at

// which an admitted writer session of that kind may be working.

func SessionPhase(kind, phase string) bool {

diff --git a/system.go b/system.go

index e97a72d43..c746c5567 100644

--- a/system.go

+++ b/system.go

@@ -49,7 +49,7 @@ func systemVerb(args []string) (verb string, rest []string, component string) {

// untouched in rest for runSystemAntivirusVerb to read.

// layout is a family too: migrate|revert|status stay in rest for

// runSystemLayout.

- case "guard-arm", "guard-exec", "update", "status", "rollback", "cancel-activation", "apply", "antivirus", "layout", "geoip", "help", "-h", "--help":

+ case "guard-arm", "guard-exec", "journal-migrate", "update", "status", "rollback", "cancel-activation", "apply", "antivirus", "layout", "geoip", "help", "-h", "--help":

return args[0], args[1:], component

}

return "", args, component

@@ -62,6 +62,8 @@ func runSystem(args []string) {

host.RunSystemGuardArm(rest)

case "guard-exec":

host.RunSystemGuardExec(rest)

+ case "journal-migrate":

+ host.RunSystemJournalMigrate(rest)

case "apply":

host.RunSystemApply(rest)

carlos / platform Public

claim Plan 9 review round 2: agent inventory cannot name root paths, schema allowlist before SQLite evaluation, restored consoles adopt the live instance, explicit offline journal-migrate