amadan

carlos / platform Public

host: a release recovers whenever another process flushed, by main's launch-owner rule over the boot, controller, PID and process start recorded with the positions

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 989072732c17c4bcbb6b7375e2c52c509589fe96 parent 6cff8a6
3 files changed, +135 −55
  • internal/host/relocationeffects.go +47 −19
  • internal/host/relocationeffects_test.go +68 −19
  • internal/host/relocationflush.go +20 −17
diff --git a/internal/host/relocationeffects.go b/internal/host/relocationeffects.go
index b7500aa92..c61791658 100644
--- a/internal/host/relocationeffects.go
+++ b/internal/host/relocationeffects.go
@@ -11,7 +11,6 @@ import (
"strings"
"sync"
- "github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
execctl "github.com/carlosframework/platform/internal/hostexec"
"github.com/carlosframework/platform/internal/hostrelocate"
@@ -19,11 +18,12 @@ import (
)
// RelocationIdentity is this box's verified actor and stop identity, as
-// Plan 6b's assembly reports it (claimStopLocal).
-type RelocationIdentity struct {
- Actor hostbox.Actor
- BootID, ControllerID string
-}
+// Plan 6b's assembly reports it: main's claimStopLocal, which
+// localStopIdentity fills (actor, boot, controller, and this process's PID
+// and start time). Release records the process with the flushed positions
+// and decides whether a recorded flush is this process's own with main's
+// claimLaunchOwner.local rule.
+type RelocationIdentity = claimStopLocal
// RelocationController is the execution controller as this package uses
// it: hostexec.Controller's Ready and Close, and Clear, which removes the
@@ -437,21 +437,40 @@ func (r *RelocationControllers) Close(ctx context.Context) error {
return first
}
+// ReleasePosition is one recorded release: the exact stop request, the
+// process that flushed it (PID and start time, beside the request's boot
+// and controller), and the positions its flush reached.
+type ReleasePosition struct {
+ Request StopRequest `json:"request"`
+ PID int `json:"pid"`
+ ProcessStart string `json:"process_start"`
+ TXID uint64 `json:"txid"`
+ Seq int64 `json:"seq"`
+}
+
+// owner is the recorded flush's launch tuple in main's form, so the
+// decision whether it is this process's own is claimLaunchOwner.local's.
+func (p ReleasePosition) owner() claimLaunchOwner {
+ return claimLaunchOwner{BootID: p.Request.BootID, ControllerID: p.Request.ControllerID, DirectoryRevision: p.Request.ExpectedDirectoryRevision,
+ PID: p.PID, ProcessStart: p.ProcessStart}
+}
+
// ReleasePositions keeps the flushed positions of one release durably,
// bound to the exact stop request (host, incarnation, stamp, session,
-// operation, phase, box, enrollment, boot, controller), from before root
-// completes its receipt. A replayed receipt (root's durable fence after a
+// operation, phase, box, enrollment, boot, controller) and the flushing
+// process, from before root completes its receipt. A replayed receipt (root's durable fence after a
// lost response or a crash before JournalRelease) is paired with them;
// without a record for that exact request the release is refused rather
// than guessed.
// Recorded matches the stop's installation and session (host, incarnation,
// stamp, session, operation, phase, obligation, box, enrollment), not its
// controller identity: a restarted agent asks with a new boot and
-// controller id, and gets back the request the flush was made under, so
-// it can ask root for that stop's durable receipt.
+// controller id (or the same ids in a new process), and gets back the
+// request the flush was made under, so it can ask root for that stop's
+// durable receipt.
type ReleasePositions interface {
- Record(ctx context.Context, r StopRequest, txid uint64, seq int64) error
- Recorded(ctx context.Context, r StopRequest) (recorded StopRequest, txid uint64, seq int64, ok bool, err error)
+ Record(ctx context.Context, p ReleasePosition) error
+ Recorded(ctx context.Context, r StopRequest) (recorded ReleasePosition, ok bool, err error)
}
// RelocationEffects is hostrelocate.Effects over providers Plan 6b's box
@@ -637,6 +656,11 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
if err != nil {
return hostclaim.Release{}, err
}
+ if id.PID <= 0 || id.ProcessStart == "" {
+ // Without this process's identity no flush could be told apart
+ // from a restarted process's (localStopIdentity always has it).
+ return hostclaim.Release{}, ErrClaimStopUnavailable
+ }
c, h := obs.Claim, obs.Claim.Holder
req := StopRequest{Region: id.Actor.Region, Host: c.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session,
OperationID: c.Op.ID, Phase: c.Op.Phase, ObligationID: RelocationReleaseObligation, Box: id.Actor.Box, Enrollment: id.Actor.Enrollment,
@@ -644,28 +668,31 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
// A flush this box already made for this stop, by any controller of
// this installation: after a restart the request is made under the
// identity that flushed, so root finds that stop's durable receipt.
- recorded, txid, seq, flushed, err := e.Positions.Recorded(ctx, req)
+ recorded, flushed, err := e.Positions.Recorded(ctx, req)
if err != nil {
return hostclaim.Release{}, err
}
+ txid, seq := recorded.TXID, recorded.Seq
if flushed {
// The stop is the recorded one: root's fence and its inventory
// check name the directory revision the flush was made under, so
// a retry keeps it even when the roster has moved on since
// (round 8 finding 2), whether or not this is a recovery.
- req.ExpectedDirectoryRevision = recorded.ExpectedDirectoryRevision
+ req.ExpectedDirectoryRevision = recorded.Request.ExpectedDirectoryRevision
}
var d StopDrainer = RecoveryReleaseDrain{}
- drain := &releaseDrain{e: e, obs: obs}
- if flushed && (recorded.BootID != req.BootID || recorded.ControllerID != req.ControllerID) {
- // The flushing process is gone with its producers: only root's
+ drain := &releaseDrain{e: e, obs: obs, id: id}
+ if flushed && !recorded.owner().local(id) {
+ // Another process flushed (main's rule: another boot or
+ // controller, or the same ids in a restarted process, whose
+ // producers ended with it): only root's
// durable receipt for the recorded identity can finish this
// release. The request names the recorded owner (boot, controller,
// directory revision) and the drain refuses; the Collect provider
// marks it a recovery request (6b's authenticated Recover path,
// Task 8a), under which root accepts another tuple for this box's
// session and replays the receipt.
- req.BootID, req.ControllerID = recorded.BootID, recorded.ControllerID
+ req.BootID, req.ControllerID = recorded.Request.BootID, recorded.Request.ControllerID
} else {
if drain.inner, err = e.Producers(ctx, obs); err != nil {
return hostclaim.Release{}, err
@@ -702,6 +729,7 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
type releaseDrain struct {
e *RelocationEffects
obs hostclaim.Observed
+ id RelocationIdentity // the process recorded with the positions
inner StopDrainer
txid uint64
seq int64
@@ -746,7 +774,7 @@ func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence stri
if d.seq, err = d.e.FlushFiles(ctx, d.obs); err != nil {
return DrainAcknowledgement{}, err
}
- if err := d.e.Positions.Record(ctx, r, d.txid, d.seq); err != nil {
+ if err := d.e.Positions.Record(ctx, ReleasePosition{Request: r, PID: d.id.PID, ProcessStart: d.id.ProcessStart, TXID: d.txid, Seq: d.seq}); err != nil {
return DrainAcknowledgement{}, err
}
ack, err := d.inner.DrainClaim(ctx, r, fence)
diff --git a/internal/host/relocationeffects_test.go b/internal/host/relocationeffects_test.go
index 5408488f3..226ae4eb1 100644
--- a/internal/host/relocationeffects_test.go
+++ b/internal/host/relocationeffects_test.go
@@ -94,22 +94,20 @@ func kept(r interface {
type memPositions struct {
calls *[]string
- req StopRequest
- txid uint64
- seq int64
+ p ReleasePosition
ok bool
}
-func (m *memPositions) Record(_ context.Context, r StopRequest, txid uint64, seq int64) error {
+func (m *memPositions) Record(_ context.Context, p ReleasePosition) error {
*m.calls = append(*m.calls, "record")
- m.req, m.txid, m.seq, m.ok = r, txid, seq, true
+ m.p, m.ok = p, true
return nil
}
-func (m *memPositions) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
- if !m.ok || !sameStop(m.req, r) {
- return StopRequest{}, 0, 0, false, nil
+func (m *memPositions) Recorded(_ context.Context, r StopRequest) (ReleasePosition, bool, error) {
+ if !m.ok || !sameStop(m.p.Request, r) {
+ return ReleasePosition{}, false, nil
}
- return m.req, m.txid, m.seq, true, nil
+ return m.p, true, nil
}
func effectsFixture(t *testing.T) (*RelocationEffects, hostclaim.Observed, *[]string, *StopRequest) {
@@ -122,7 +120,7 @@ func effectsFixture(t *testing.T) (*RelocationEffects, hostclaim.Observed, *[]st
Lifetime: context.Background(),
Controllers: testControllers(t),
Identity: func(context.Context) (RelocationIdentity, error) {
- return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot", ControllerID: "ctl", PID: 100, ProcessStart: "5000"}, nil
},
Controller: func(ctx context.Context, b execctl.Binding) (RelocationController, error) {
return &readyController{opened: ctx, st: execctl.State{Binding: execctl.Binding{Enrollment: "enr", BootID: "boot", ControllerID: "c9"}, LeaseETag: "lease-1", Expires: exp}}, nil
@@ -270,7 +268,7 @@ func TestRelocationReleaseReplaysOnlyRecordedPositions(t *testing.T) {
// (same boot, same controller); the retry still names the
// revision the stop was flushed under.
e.Identity = func(context.Context) (RelocationIdentity, error) {
- return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot", ControllerID: "ctl"}, nil
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot", ControllerID: "ctl", PID: 100, ProcessStart: "5000"}, nil
}
r, err := e.Release(context.Background(), obs)
if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
@@ -294,7 +292,7 @@ func TestRelocationReleaseReplaysOnlyRecordedPositions(t *testing.T) {
t.Fatal(err)
}
e.Identity = func(context.Context) (RelocationIdentity, error) {
- return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot-2", ControllerID: "ctl-2"}, nil
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 5}, BootID: "boot-2", ControllerID: "ctl-2", PID: 200, ProcessStart: "9000"}, nil
}
var asked StopRequest
recovery := false
@@ -314,6 +312,46 @@ func TestRelocationReleaseReplaysOnlyRecordedPositions(t *testing.T) {
t.Fatalf("root was not asked as the recorded owner's recovery: %+v recovery=%v", asked, recovery)
}
})
+ t.Run("a restarted process with the same boot and controller ids recovers", func(t *testing.T) {
+ // Fix round 1 (I-2), main's claimLaunchOwner.local: the same boot
+ // and controller ids in a restarted process are not this process.
+ // Its producers ended with the old one, so the request is a
+ // recovery under the recorded owner and no drain is offered.
+ e, obs, calls, _ := effectsFixture(t)
+ if _, err := e.Release(context.Background(), obs); err != nil {
+ t.Fatal(err)
+ }
+ *calls = nil
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot", ControllerID: "ctl", PID: 101, ProcessStart: "6000"}, nil
+ }
+ var asked StopRequest
+ recovery := false
+ e.Collect = func(_ context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ asked = r
+ _, recovery = d.(RecoveryReleaseDrain)
+ if _, err := d.DrainClaim(context.Background(), r, "fence"); !errors.Is(err, ErrClaimStopIdentity) {
+ return StopResult{}, errors.New("a restarted process offered its producers for the old process's stop")
+ }
+ return StopResult{Key: "k", Digest: strings.Repeat("a", 64)}, nil
+ }
+ r, err := e.Release(context.Background(), obs)
+ if err != nil || r.FinalTXID != 41 || r.ManifestSeq != 8 {
+ t.Fatalf("same ids, new process: %+v %v", r, err)
+ }
+ if !recovery || asked.BootID != "boot" || asked.ControllerID != "ctl" || len(*calls) != 0 {
+ t.Fatalf("not a recovery: recovery=%v asked=%+v ran %v", recovery, asked, *calls)
+ }
+ })
+ t.Run("an identity without its process", func(t *testing.T) {
+ e, obs, calls, _ := effectsFixture(t)
+ e.Identity = func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "src", Enrollment: "enr", Revision: 4}, BootID: "boot", ControllerID: "ctl"}, nil
+ }
+ if _, err := e.Release(context.Background(), obs); !errors.Is(err, ErrClaimStopUnavailable) || len(*calls) != 0 {
+ t.Fatalf("released without a process identity: %v ran %v", err, *calls)
+ }
+ })
t.Run("a record of another session", func(t *testing.T) {
e, obs, _, _ := effectsFixture(t)
if _, err := e.Release(context.Background(), obs); err != nil {
@@ -332,22 +370,25 @@ func TestReleasePositionsFileBindsTheExactStop(t *testing.T) {
f := ReleasePositionsFile{Dir: filepath.Join(t.TempDir(), "relocate")}
r := StopRequest{Host: "shop.example", Incarnation: "inc", InstallationStamp: 5, WriterSession: 6, OperationID: "rel-1", Phase: "releasing", ObligationID: RelocationReleaseObligation,
Box: "src", Enrollment: "enr", BootID: "boot", ControllerID: "ctl", ExpectedClaimETag: "e1", ExpectedDirectoryRevision: "4"}
- if _, _, _, ok, err := f.Recorded(ctx, r); ok || err != nil {
+ if _, ok, err := f.Recorded(ctx, r); ok || err != nil {
t.Fatalf("before any record: %v %v", ok, err)
}
- if err := f.Record(ctx, r, 41, 8); err != nil {
+ if err := f.Record(ctx, ReleasePosition{Request: r, TXID: 41, Seq: 8}); !errors.Is(err, ErrClaimStopUnavailable) {
+ t.Fatalf("a record without its process: %v", err)
+ }
+ if err := f.Record(ctx, ReleasePosition{Request: r, PID: 100, ProcessStart: "5000", TXID: 41, Seq: 8}); err != nil {
t.Fatal(err)
}
moved := r
moved.ExpectedClaimETag, moved.ExpectedDirectoryRevision = "e2", "5" // a target replacement moved the claim
- if rec, txid, seq, ok, err := f.Recorded(ctx, moved); !ok || err != nil || txid != 41 || seq != 8 || rec.ControllerID != "ctl" {
- t.Fatalf("recorded: %d %d %v %v", txid, seq, ok, err)
+ if rec, ok, err := f.Recorded(ctx, moved); !ok || err != nil || rec.TXID != 41 || rec.Seq != 8 || rec.Request.ControllerID != "ctl" || rec.PID != 100 || rec.ProcessStart != "5000" {
+ t.Fatalf("recorded: %+v %v %v", rec, ok, err)
}
// A restarted controller of the same installation finds the record,
// and gets the identity the flush was made under.
restarted := r
restarted.BootID, restarted.ControllerID = "boot-2", "ctl-2"
- if rec, _, _, ok, err := f.Recorded(ctx, restarted); !ok || err != nil || rec.BootID != "boot" || rec.ControllerID != "ctl" {
+ if rec, ok, err := f.Recorded(ctx, restarted); !ok || err != nil || rec.Request.BootID != "boot" || rec.Request.ControllerID != "ctl" {
t.Fatalf("after a restart: %+v %v %v", rec, ok, err)
}
for name, change := range map[string]func(*StopRequest){
@@ -359,16 +400,24 @@ func TestReleasePositionsFileBindsTheExactStop(t *testing.T) {
} {
other := r
change(&other)
- if _, _, _, ok, _ := f.Recorded(ctx, other); ok {
+ if _, ok, _ := f.Recorded(ctx, other); ok {
t.Fatalf("%s: another stop's positions were returned", name)
}
}
if err := os.WriteFile(filepath.Join(f.Dir, "shop.example.json"), []byte("{"), 0o600); err != nil {
t.Fatal(err)
}
- if _, _, _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
+ if _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
t.Fatalf("a damaged record: %v", err)
}
+ // A record that names no process would read as this process's own
+ // under claimLaunchOwner.local: it is damaged, never trusted.
+ if err := os.WriteFile(filepath.Join(f.Dir, "shop.example.json"), []byte(`{"v":1,"request":{"Host":"shop.example"},"txid":41,"seq":8}`), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := f.Recorded(ctx, r); !errors.Is(err, ErrClaimStopReceipt) {
+ t.Fatalf("a record without its process: %v", err)
+ }
}
func TestRelocationAcknowledgeRetainsTheControllerUnderTheAgentLifetime(t *testing.T) {
diff --git a/internal/host/relocationflush.go b/internal/host/relocationflush.go
index fe7770d44..6eade7bb3 100644
--- a/internal/host/relocationflush.go
+++ b/internal/host/relocationflush.go
@@ -86,13 +86,11 @@ func (f ReleaseFlushFiles) Flush(ctx context.Context, obs hostclaim.Observed) (i
return p.Commit(ctx)
}
-// releasePosition is one recorded release: the exact stop request and
-// the positions its flush reached.
+// releasePosition is one recorded release on disk: a versioned
+// ReleasePosition.
type releasePosition struct {
- V int `json:"v"`
- Request StopRequest `json:"request"`
- TXID uint64 `json:"txid"`
- Seq int64 `json:"seq"`
+ V int `json:"v"`
+ ReleasePosition
}
// ReleasePositionsFile is ReleasePositions on disk: one file per host under
@@ -111,15 +109,18 @@ func (f ReleasePositionsFile) path(host string) (string, error) {
return filepath.Join(f.Dir, host+".json"), nil
}
-func (f ReleasePositionsFile) Record(_ context.Context, r StopRequest, txid uint64, seq int64) error {
- path, err := f.path(r.Host)
+func (f ReleasePositionsFile) Record(_ context.Context, p ReleasePosition) error {
+ path, err := f.path(p.Request.Host)
if err != nil {
return err
}
+ if p.PID <= 0 || p.ProcessStart == "" {
+ return ErrClaimStopUnavailable // a record that cannot name its process could not be told from a restart's
+ }
if err := os.MkdirAll(f.Dir, 0o700); err != nil {
return err
}
- raw, err := json.Marshal(releasePosition{V: 1, Request: r, TXID: txid, Seq: seq})
+ raw, err := json.Marshal(releasePosition{V: 1, ReleasePosition: p})
if err != nil {
return err
}
@@ -156,24 +157,26 @@ func sameStop(a, b StopRequest) bool {
a.OperationID == b.OperationID && a.Phase == b.Phase && a.ObligationID == b.ObligationID && a.Box == b.Box && a.Enrollment == b.Enrollment
}
-func (f ReleasePositionsFile) Recorded(_ context.Context, r StopRequest) (StopRequest, uint64, int64, bool, error) {
+func (f ReleasePositionsFile) Recorded(_ context.Context, r StopRequest) (ReleasePosition, bool, error) {
path, err := f.path(r.Host)
if err != nil {
- return StopRequest{}, 0, 0, false, err
+ return ReleasePosition{}, false, err
}
raw, err := os.ReadFile(path)
if errors.Is(err, os.ErrNotExist) {
- return StopRequest{}, 0, 0, false, nil
+ return ReleasePosition{}, false, nil
}
if err != nil {
- return StopRequest{}, 0, 0, false, err
+ return ReleasePosition{}, false, err
}
var p releasePosition
- if err := json.Unmarshal(raw, &p); err != nil || p.V != 1 {
- return StopRequest{}, 0, 0, false, fmt.Errorf("%w: release positions for %s", ErrClaimStopReceipt, r.Host)
+ if err := json.Unmarshal(raw, &p); err != nil || p.V != 1 || p.PID <= 0 || p.ProcessStart == "" {
+ // A record without its process is damaged: it would read as this
+ // process's own under claimLaunchOwner.local.
+ return ReleasePosition{}, false, fmt.Errorf("%w: release positions for %s", ErrClaimStopReceipt, r.Host)
}
if !sameStop(p.Request, r) {
- return StopRequest{}, 0, 0, false, nil
+ return ReleasePosition{}, false, nil
}
- return p.Request, p.TXID, p.Seq, true, nil
+ return p.ReleasePosition, true, nil
}