amadan

carlos / platform Public

host: a release whose receipt already discharged its obligation replays from the journaled receipt and the recorded positions, never asking root again

Complete removes the obligation and moves the claim, and the release is
journaled in a separate CAS. If that CAS lost a race or the agent
restarted, the next Release asked root for a second stop, which root
refuses (no obligation owed, another ETag), so the release could never be
returned and the operation went stuck after a clean stop. Release now
sees the journaled discharged receipt, re-verifies it (main's Complete in
its replay form, through the new RelocationReleaseDischarge.Verify), and
returns it with the positions this box recorded; with none recorded it
refuses. The test runs main's real Queue and Complete with a Collect that
models root's refusal, through a lost journal CAS and a restart.

Merge review fix round 1, finding 4.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 276837786f5724050ad6ba7f2a0d034947821659 parent 8b90c6b
4 files changed, +183 −4
  • internal/host/relocationeffects.go +44 −4
  • internal/host/relocationeffects_test.go +5 −0
  • internal/host/relocationreleases.go +15 −0
  • internal/host/relocationreleases_test.go +119 −0
diff --git a/internal/host/relocationeffects.go b/internal/host/relocationeffects.go
index ca17e96d7..32901efc5 100644
--- a/internal/host/relocationeffects.go
+++ b/internal/host/relocationeffects.go
@@ -653,16 +653,22 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
// from a restarted process's (localStopIdentity always has it).
return hostclaim.Release{}, ErrClaimStopUnavailable
}
+ if key, digest := obs.Claim.Op.Journal["release_stop_receipt"], obs.Claim.Op.Journal["release_stop_sha256"]; key != "" || digest != "" {
+ // Root's receipt already discharged the obligation, and the release
+ // was not journaled (the journal CAS lost a race, or the agent
+ // restarted after Complete). Root would refuse a second stop: the
+ // claim owes it nothing and has moved on. The release is the
+ // journaled receipt, verified again, with the recorded positions.
+ return e.replayDischarged(ctx, obs, id, key, digest)
+ }
// Root fences a relocation-release stop only while the claim owes it
// (hostguard.StopAuthorityMatches): the obligation is queued first, and
// the stop is asked at the claim that owes it.
if obs, err = e.Releases.Queue(ctx, obs); err != nil {
return hostclaim.Release{}, err
}
- 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,
- BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: fmt.Sprint(id.Actor.Revision)}
+ h := obs.Claim.Holder
+ req := releaseStopRequest(obs, id)
// 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.
@@ -728,6 +734,40 @@ func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed)
FinalTXID: txid, ManifestSeq: seq, Receipt: result.Key, ReceiptSHA256: result.Digest}, nil
}
+// releaseStopRequest is the release's stop request for obs's session, as
+// this box's identity id asks it.
+func releaseStopRequest(obs hostclaim.Observed, id RelocationIdentity) StopRequest {
+ c, h := obs.Claim, obs.Claim.Holder
+ return 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,
+ BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: fmt.Sprint(id.Actor.Revision)}
+}
+
+// replayDischarged returns the release whose receipt already discharged
+// the obligation: the claim journals key and digest and owes the operation
+// nothing, the receipt verifies again (main's Complete in its replay form),
+// and this box recorded the positions its flush reached for this stop.
+// Without a recorded flush it refuses rather than guess, as a replayed
+// receipt always does.
+func (e *RelocationEffects) replayDischarged(ctx context.Context, obs hostclaim.Observed, id RelocationIdentity, key, digest string) (hostclaim.Release, error) {
+ c, h := obs.Claim, obs.Claim.Holder
+ if err := releaseDischarged(c, key, digest); err != nil {
+ return hostclaim.Release{}, err
+ }
+ if err := e.Releases.Verify(ctx, obs, key, digest); err != nil {
+ return hostclaim.Release{}, err
+ }
+ recorded, flushed, err := e.Positions.Recorded(ctx, releaseStopRequest(obs, id))
+ if err != nil {
+ return hostclaim.Release{}, err
+ }
+ if !flushed {
+ return hostclaim.Release{}, fmt.Errorf("%w: a discharged receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt)
+ }
+ return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session,
+ FinalTXID: recorded.TXID, ManifestSeq: recorded.Seq, Receipt: key, ReceiptSHA256: digest}, nil
+}
+
// releaseDrain is the agent's drain for a release: step 1's action, its
// observation, steps 2 and 3, the durable positions, and only then the
// producers' own drain and acknowledgement.
diff --git a/internal/host/relocationeffects_test.go b/internal/host/relocationeffects_test.go
index ee2f134c3..f7a0b75cf 100644
--- a/internal/host/relocationeffects_test.go
+++ b/internal/host/relocationeffects_test.go
@@ -132,6 +132,11 @@ func (f *fakeReleaseObligation) Queue(_ context.Context, o hostclaim.Observed) (
return o, f.queueErr
}
+func (f *fakeReleaseObligation) Verify(context.Context, hostclaim.Observed, string, string) error {
+ *f.calls = append(*f.calls, "verify")
+ return nil
+}
+
func (f *fakeReleaseObligation) Complete(_ context.Context, o hostclaim.Observed, key, digest string) (hostclaim.Observed, error) {
*f.calls = append(*f.calls, "complete")
f.completed = append(f.completed, key+"@"+digest)
diff --git a/internal/host/relocationreleases.go b/internal/host/relocationreleases.go
index 20886ed40..f5a9ce3a9 100644
--- a/internal/host/relocationreleases.go
+++ b/internal/host/relocationreleases.go
@@ -83,6 +83,9 @@ func verifyDischargedReceipt(ctx context.Context, receipts RelocationReleaseRece
type RelocationReleaseDischarge interface {
Queue(context.Context, hostclaim.Observed) (hostclaim.Observed, error)
Complete(ctx context.Context, observed hostclaim.Observed, key, digest string) (hostclaim.Observed, error)
+ // Verify re-verifies a receipt observed has already discharged, and
+ // writes nothing.
+ Verify(ctx context.Context, observed hostclaim.Observed, key, digest string) error
}
// ReleaseObligation is production's RelocationReleaseDischarge: main's
@@ -134,3 +137,15 @@ func (o ReleaseObligation) Complete(ctx context.Context, obs hostclaim.Observed,
}
return o.Receipts.Complete(ctx, fresh, key, digest)
}
+
+// Verify re-verifies the discharged receipt key and digest name on
+// observed's claim, with main's Complete in its replay form.
+func (o ReleaseObligation) Verify(ctx context.Context, observed hostclaim.Observed, key, digest string) error {
+ if o.Receipts.Store == nil || observed.Claim == nil || observed.Claim.Holder == nil {
+ return ErrClaimStopUnavailable
+ }
+ if err := releaseDischarged(observed.Claim, key, digest); err != nil {
+ return err
+ }
+ return verifyDischargedReceipt(ctx, o.Receipts, observed.Claim, key, digest)
+}
diff --git a/internal/host/relocationreleases_test.go b/internal/host/relocationreleases_test.go
index d9fd0122c..23b07b291 100644
--- a/internal/host/relocationreleases_test.go
+++ b/internal/host/relocationreleases_test.go
@@ -7,6 +7,8 @@ import (
"encoding/hex"
"encoding/json"
"errors"
+ "fmt"
+ "path/filepath"
"strings"
"testing"
"time"
@@ -15,6 +17,7 @@ import (
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostevidence"
+ "github.com/carlosframework/platform/internal/hostguard"
"github.com/carlosframework/platform/internal/release"
)
@@ -298,3 +301,119 @@ func TestStopReceiptReleasesVerifiesRootsSignedReceiptAtThePin(t *testing.T) {
}
})
}
+
+// rootCollect models root's FenceAndStop for a relocation release: it
+// refuses unless the claim in the store owes the obligation at exactly the
+// request's ETag (hostguard.StopAuthorityMatches, the expected ETag), runs
+// the agent's drain, then publishes its signed receipt. calls counts the
+// requests that reached it.
+func (w releaseWorld) rootCollect(t *testing.T, calls *int) func(context.Context, StopRequest, StopDrainer) (StopResult, error) {
+ return func(ctx context.Context, r StopRequest, d StopDrainer) (StopResult, error) {
+ *calls++
+ o, err := hostclaim.Read(ctx, w.s, r.Host)
+ if err != nil {
+ return StopResult{}, err
+ }
+ owed := false
+ for _, ob := range o.Claim.Obligations {
+ owed = owed || (ob.Kind == RelocationReleaseObligation && ob.Op == r.OperationID && ob.Box == r.Box)
+ }
+ if !owed || o.ETag != r.ExpectedClaimETag {
+ return StopResult{}, ErrClaimStopIdentity
+ }
+ if _, err := d.DrainClaim(ctx, r, "fence"); err != nil {
+ return StopResult{}, err
+ }
+ key, digest := w.receipt(t)
+ return StopResult{Key: key, Digest: digest}, nil
+ }
+}
+
+// releaseEffects is one agent process's RelocationEffects over the world:
+// the real obligation (main's Queue and Complete), durable positions in
+// dir, root modelled by rootCollect, and this process's identity.
+func (w releaseWorld) releaseEffects(t *testing.T, dir string, pid int, calls *int) *RelocationEffects {
+ steps := &[]string{}
+ return &RelocationEffects{
+ Identity: func(context.Context) (RelocationIdentity, error) {
+ return RelocationIdentity{Actor: hostbox.Actor{Box: "box", Enrollment: w.en, Region: "ie", Revision: 1}, BootID: "boot", ControllerID: fmt.Sprint("ctl-", pid), PID: pid, ProcessStart: fmt.Sprint(pid * 10)}, nil
+ },
+ Producers: func(context.Context, hostclaim.Observed) (StopDrainer, error) { return recordingDrain{steps}, nil },
+ Collect: w.rootCollect(t, calls),
+ WritersStopped: func(_ context.Context, o hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) {
+ h := o.Claim.Holder
+ at := time.Date(2026, 10, 9, 1, 0, 0, 0, time.UTC)
+ target := hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: h.Session, Box: "box", Enrollment: w.en, BootID: "boot", ControllerID: fmt.Sprint("ctl-", pid)}
+ return ClaimWritersDrained{Target: target, Generation: h.Generation, WriterBox: h.WriterBox, FenceClosed: true, LaunchJoined: true, WorkDrained: true, SpawnsSettled: true,
+ CgroupEmpty: true, Cgroup: "carlos/shop", DBPaths: []string{"/data/shop.db"}, CheckedAt: at},
+ WritersStoppedEvidence{Host: o.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, Box: "box", Enrollment: w.en, CheckedAt: at}, nil
+ },
+ Releases: ReleaseObligation{Receipts: RelocationReleaseReceipts{Store: w.s, Directory: w.dir}},
+ FlushDB: func(context.Context, hostclaim.Observed) (uint64, error) { return 41, nil },
+ FlushFiles: func(context.Context, hostclaim.Observed) (int64, error) { return 8, nil },
+ Positions: ReleasePositionsFile{Dir: dir},
+ }
+}
+
+// Review finding 4: once root's receipt has discharged the obligation the
+// claim has moved on and owes root nothing, so a second stop is refused.
+// A release whose journal CAS was lost, or whose agent restarted after
+// Complete, must still be returned: from the journaled receipt, verified
+// again, and the recorded positions, never by asking root.
+func TestADischargedReleaseReplaysAfterALostJournalOrARestart(t *testing.T) {
+ ctx := context.Background()
+ w := newReleaseWorld(t)
+ dir := filepath.Join(t.TempDir(), "relocate")
+ calls := 0
+ first, err := w.releaseEffects(t, dir, 100, &calls).Release(ctx, w.obs)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if calls != 1 || first.FinalTXID != 41 || first.ManifestSeq != 8 {
+ t.Fatalf("first release: %+v, root asked %d times", first, calls)
+ }
+ fresh := func() hostclaim.Observed {
+ o, err := hostclaim.Read(ctx, w.s, w.obs.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return o
+ }
+ discharged := fresh()
+ if len(discharged.Claim.Obligations) != 0 || discharged.Claim.Op.Journal["release_stop_receipt"] != first.Receipt {
+ t.Fatalf("not discharged: %+v", discharged.Claim)
+ }
+ // The journal CAS lost: the same process retries at the claim as it is.
+ again, err := w.releaseEffects(t, dir, 100, &calls).Release(ctx, discharged)
+ if err != nil || again != first {
+ t.Fatalf("same process retry: %+v %v", again, err)
+ }
+ // The agent restarted after Complete: a new process, new controller.
+ restarted, err := w.releaseEffects(t, dir, 200, &calls).Release(ctx, fresh())
+ if err != nil || restarted != first {
+ t.Fatalf("after a restart: %+v %v", restarted, err)
+ }
+ if calls != 1 {
+ t.Fatalf("root was asked again for a discharged release: %d calls", calls)
+ }
+ // The release journals, and nothing was written by the replays.
+ if now := fresh(); now.ETag != discharged.ETag {
+ t.Fatal("a replay wrote the claim")
+ }
+ if _, err := hostclaim.Update(ctx, w.s, fresh(), hostclaim.JournalRelease("reloc", restarted)); err != nil {
+ t.Fatalf("the replayed release does not journal: %v", err)
+ }
+ // Without a recorded flush the replay refuses rather than guess.
+ w2 := newReleaseWorld(t)
+ calls2 := 0
+ if _, err := w2.releaseEffects(t, filepath.Join(t.TempDir(), "relocate"), 100, &calls2).Release(ctx, w2.obs); err != nil {
+ t.Fatal(err)
+ }
+ o2, err := hostclaim.Read(ctx, w2.s, w2.obs.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := w2.releaseEffects(t, filepath.Join(t.TempDir(), "elsewhere"), 300, &calls2).Release(ctx, o2); !errors.Is(err, ErrClaimStopReceipt) || calls2 != 1 {
+ t.Fatalf("a discharged release with no recorded positions: %v, root asked %d times", err, calls2)
+ }
+}