amadan

carlos / platform Public

hostclaim: Plan 12 fix round 1 (write-ahead, identity-bound recovery handover; replay re-checks timers and route; real non-empty restore test; root stop still required for the cut stamp; grant-only fenced release token; narrowed loss gate with destroyed-local-only cleanup kept; census evidence and vouching; loss report alerts without blocking; exact target obligation snapshot)

Answers Astra's two P1s and two P2s and Opus's P2-1 and P3-1..6 on
42ae89c7. The recovery intent is journalled before the admission CAS and
resumed by identity (incarnation, stamp, session, generation, writer box,
op), so unrelated claim updates no longer strand a handover.
Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude b0ac408ea7e6c9e564af52de6eb0977295b202c3 parent 42ae89c
36 files changed, +1237 −106
  • internal/activator/claimwake.go +2 −2
  • internal/activator/plan12_resume_test.go +28 −3
  • internal/host/claimhandover_crossbox.go +1 −0
  • internal/host/claimhandover_recovery.go +10 −13
  • internal/host/claimrecoveryjournal.go +105 −14
  • internal/host/claimretirementcensus.go +15 −23
  • internal/host/claimretryrecovery_test.go +21 −5
  • internal/host/claimwakeassembly.go +9 −2
  • internal/host/claimwakestop.go +4 −2
  • internal/host/plan12_census_test.go +1 −1
  • internal/host/plan12_fix1_test.go +218 −0
  • internal/host/plan12_handover_test.go +20 −6
  • internal/host/plan12_restore_linux_test.go +180 −16
  • internal/hostbox/plan12_fence_loss_test.go +19 −6
  • internal/hostbox/plan12_loss_test.go +1 −1
  • internal/hostclaim/lost.go +11 −0
  • internal/hostclaim/lost_test.go +4 −0
  • internal/hostclaim/move_target_fence.go +17 −1
  • internal/hostclaim/operation_loss.go +34 −1
  • internal/hostclaim/plan12_fence_test.go +26 −4
  • internal/hostclaim/plan12_fix1_loss_test.go +132 −0
  • internal/hostclaim/writer_identity.go +11 −0
  • internal/hostguard/writerstop.go +13 −0
  • internal/hostops/fencerelease/token.go +7 −0
  • internal/hostops/internal/fencerelease/token.go +27 −0
  • internal/hostops/internal/fencerelease/token_test.go +37 −0
  • internal/hostops/move_fence_release.go +3 −0
  • internal/hostops/move_local_loss_test.go +31 −0
  • internal/hostops/plan12_fence_test.go +42 −0
  • internal/hostops/plan12_fix1_recovery_test.go +70 −0
  • internal/hostops/recovery.go +8 −1
  • internal/park/sweep.go +5 −1
  • internal/provision/plan9_b8_census_test.go +2 −3
  • internal/registry/plan12_fix1_journal_test.go +49 −0
  • internal/registry/recoveryjournal.go +60 −0
  • internal/registry/remote_dispatch.go +14 −1
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go
index d4e0710df..5468cfe9d 100644
--- a/internal/activator/claimwake.go
+++ b/internal/activator/claimwake.go
@@ -308,7 +308,7 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance,
return nil, hostbox.ErrChanged
}
if admitted.Claim.Holder.Session <= observed.Claim.Holder.Session || admitted.Claim.Holder.Generation == observed.Claim.Holder.Generation {
- if !recoveryProbeKind(observed.Claim.Op.Kind) || observed.ETag == "" || admitted.ETag != observed.ETag || admitted.Host != observed.Host || admitted.Claim.Holder.Session != observed.Claim.Holder.Session || admitted.Claim.Holder.Generation != observed.Claim.Holder.Generation || r.ResumedAdmission == nil {
+ if !recoveryProbeKind(observed.Claim.Op.Kind) || !hostclaim.SameWriterIdentity(observed, admitted) || r.ResumedAdmission == nil {
return nil, hostbox.ErrChanged
}
if err := r.ResumedAdmission(ctx, observed, admitted); err != nil {
@@ -360,7 +360,7 @@ func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostc
if err != nil {
return err
}
- if fresh.ETag != admitted.ETag {
+ if fresh.ETag != admitted.ETag && !(recoveryProbeKind(admitted.Claim.Op.Kind) && hostclaim.SameWriterIdentity(fresh, admitted)) {
return hostbox.ErrChanged
}
s, err := r.Prepare(ctx, e.route, admitted, policy)
diff --git a/internal/activator/plan12_resume_test.go b/internal/activator/plan12_resume_test.go
index 81a628e4d..a12694e2c 100644
--- a/internal/activator/plan12_resume_test.go
+++ b/internal/activator/plan12_resume_test.go
@@ -13,21 +13,46 @@ import (
)
func TestPlan12RecoveryResumeReachesPrepare(t *testing.T) {
- for _, mode := range []string{"verified", "missing", "refused", "changed"} {
+ for _, mode := range []string{"verified", "missing", "refused", "changed", "changed-after-resume", "session", "op", "writer", "stamp", "kind", "phase", "exec"} {
t.Run(mode, func(t *testing.T) {
o := hostclaim.Observed{Host: "x.example", ETag: "etag", Claim: &hostclaim.Claim{Holder: &hostclaim.Holder{Incarnation: "inc", Session: 3, Stamp: 3, Generation: "gen", WriterBox: "box"}, Op: hostclaim.Op{ID: "move", Kind: hostclaim.KindRestoring, Phase: hostclaim.PhaseProbing, Exec: "box"}}}
prepared := errors.New("prepare reached")
- r := &ClaimRuntime{Box: "box", Read: func(context.Context, string) (hostclaim.Observed, error) { return o, nil }, StopForeign: func(context.Context, string) error { return nil }, Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil }, Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
+ current := o
+ r := &ClaimRuntime{Box: "box", Read: func(context.Context, string) (hostclaim.Observed, error) { return current, nil }, StopForeign: func(context.Context, string) error { return nil }, Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil }, Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) {
next := o
+ claim := *o.Claim
+ holder := *claim.Holder
+ next.Claim = &claim
+ claim.Holder = &holder
+ switch mode {
+ case "session":
+ holder.Session--
+ case "op":
+ claim.Op.ID = "other"
+ case "writer":
+ holder.WriterBox = "other"
+ case "stamp":
+ holder.Stamp++
+ case "kind":
+ claim.Op.Kind = hostclaim.KindReverting
+ case "phase":
+ claim.Op.Phase = "other"
+ case "exec":
+ claim.Op.Exec = "other"
+ }
if mode == "changed" {
next.ETag = "changed"
}
+ current = next
return next, nil
}, Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) {
return ClaimStart{}, prepared
}, FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil }}
if mode != "missing" {
r.ResumedAdmission = func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
+ if mode == "changed-after-resume" {
+ current.ETag = "changed"
+ }
if mode == "refused" {
return hostbox.ErrChanged
}
@@ -36,7 +61,7 @@ func TestPlan12RecoveryResumeReachesPrepare(t *testing.T) {
}
s := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: r}}
_, err := s.doClaimWake(t.Context(), &actEntry{host: o.Host, route: registry.Route{Host: o.Host}, recoveryPolicy: &StartPolicy{Probe: true, MaxStarts: 5, Window: 30 * time.Second}})
- if mode == "verified" {
+ if mode == "verified" || mode == "changed" || mode == "changed-after-resume" {
if !errors.Is(err, prepared) {
t.Fatalf("resume did not reach prepare: %v", err)
}
diff --git a/internal/host/claimhandover_crossbox.go b/internal/host/claimhandover_crossbox.go
index a2c9603dd..01ddd6a8a 100644
--- a/internal/host/claimhandover_crossbox.go
+++ b/internal/host/claimhandover_crossbox.go
@@ -180,6 +180,7 @@ func wireClaimAdmission(w *ClaimWakeAssembly, box string, local func(context.Con
return relocation.AfterAdmission(ctx, prior, admitted)
}
}
+ w.recoveryIntent = landing.prepareRecovery
w.recoveryPrior = landing.pendingRecovery
w.recoveryPrelaunched = landing.completeRecovery
w.BeforeAdmission = claimPredecessorStop(w, box, before)
diff --git a/internal/host/claimhandover_recovery.go b/internal/host/claimhandover_recovery.go
index ef22a2841..37655558b 100644
--- a/internal/host/claimhandover_recovery.go
+++ b/internal/host/claimhandover_recovery.go
@@ -101,7 +101,7 @@ func (h claimCrossBoxHandover) beforeRecovery(ctx context.Context, o hostclaim.O
}
old := o.Claim.Holder
return h.Guard.With(ctx, o.Host, hostguard.Captured{Incarnation: old.Incarnation, Stamp: old.Stamp}, func(c *hostclaim.Claim) error {
- if c.Gen != o.Claim.Gen || c.Op.ID != o.Claim.Op.ID || c.Op.Kind != o.Claim.Op.Kind || c.Op.Phase != o.Claim.Op.Phase || c.Op.Exec != h.Box || c.Holder.Session != old.Session {
+ if !hostclaim.SameWriterIdentity(o, hostclaim.Observed{Host: o.Host, Claim: c}) {
return hostbox.ErrChanged
}
return nil
@@ -138,26 +138,19 @@ func (h claimCrossBoxHandover) afterRecovery(ctx context.Context, prior, admitte
if err != nil {
return err
}
- if current.ETag != admitted.ETag {
+ if !hostclaim.SameWriterIdentity(current, admitted) {
return hostbox.ErrChanged
}
if h.Registry == nil {
return hostbox.ErrTypedInventory
}
- journal, err := json.Marshal(claimRecoveryJournal{Prior: prior, Admitted: admitted})
- if err != nil {
- return err
- }
- if err = h.Registry.BeginRecoveryHandover(ctx, claimRecoveryJournalID(admitted), string(journal)); err != nil {
- return err
- }
// Activate retires the prior local stamp through the descriptor-based guard.
// The restore hook subsequently retires the source database and marker.
if err = h.Guard.Activate(ctx, admitted.Host, h.Box, nil); err != nil {
return err
}
return h.Guard.With(ctx, admitted.Host, hostguard.Captured{Incarnation: next.Incarnation, Stamp: next.Stamp}, func(c *hostclaim.Claim) error {
- if c.Gen != ac.Gen || c.Holder.Session != next.Session || c.Holder.Generation != next.Generation || c.Holder.WriterBox != h.Box || c.Op.ID != op.ID || c.Op.Kind != op.Kind || c.Op.Phase != op.Phase || c.Op.Exec != h.Box {
+ if !hostclaim.SameWriterIdentity(admitted, hostclaim.Observed{Host: admitted.Host, Claim: c}) {
return hostbox.ErrChanged
}
return nil
@@ -220,6 +213,8 @@ func (h claimCrossBoxHandover) installRecoveryInventory(ctx context.Context, pri
if !inv.Complete || inv.Host != prior.Host || inv.Incarnation != old.Incarnation || inv.InstallationStamp != stamp || inv.Box != actor.Box || inv.Enrollment != actor.Enrollment {
return hostbox.ErrTypedInventory
}
+ sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", prior.Host, next.Incarnation, next.Stamp, actor.Box)))
+ owner := hex.EncodeToString(sum[:])
current, err := h.Inventory.ReadInstallation(ctx, prior.Host, next.Incarnation, next.Stamp)
if err == nil {
if !current.Complete || current.Box != actor.Box || current.Enrollment != actor.Enrollment || current.DirectoryRevision != fmt.Sprint(actor.Revision) {
@@ -232,14 +227,16 @@ func (h claimCrossBoxHandover) installRecoveryInventory(ctx context.Context, pri
if e != nil {
return e
}
- return hostbox.ValidateAgentInventory(current, row, h.Guard.DataDir)
+ if err := hostbox.ValidateAgentInventory(current, row, h.Guard.DataDir); err != nil {
+ return err
+ }
+ inv.InstallationStamp = next.Stamp
+ return h.rebindRecoveryRows(ctx, &inv, stamp, owner)
}
if !errors.Is(err, sql.ErrNoRows) {
return err
}
inv.InstallationStamp, inv.DirectoryRevision = next.Stamp, fmt.Sprint(actor.Revision)
- sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%s", prior.Host, next.Incarnation, next.Stamp, actor.Box)))
- owner := hex.EncodeToString(sum[:])
if err := h.rebindRecoveryRows(ctx, &inv, stamp, owner); err != nil {
return err
}
diff --git a/internal/host/claimrecoveryjournal.go b/internal/host/claimrecoveryjournal.go
index 763b78a43..d3628a2b9 100644
--- a/internal/host/claimrecoveryjournal.go
+++ b/internal/host/claimrecoveryjournal.go
@@ -10,28 +10,117 @@ import (
"github.com/carlosframework/platform/internal/hostclaim"
)
-type claimRecoveryJournal struct{ Prior, Admitted hostclaim.Observed }
+type claimRecoveryJournal struct {
+ Prior hostclaim.Observed
+ Generation string
+ Sessions []uint64
+}
func claimRecoveryJournalID(o hostclaim.Observed) string {
h := o.Claim.Holder
- sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d", o.Host, h.Incarnation, h.Stamp)))
+ sum := sha256.Sum256([]byte(fmt.Sprintf("%s\x00%s\x00%d\x00%d\x00%s", o.Host, h.Incarnation, h.Stamp, h.Session, o.Claim.Op.ID)))
return hex.EncodeToString(sum[:])
}
+
+func (j claimRecoveryJournal) successor(o hostclaim.Observed, box string) bool {
+ if j.Prior.Claim == nil || j.Prior.Claim.Holder == nil || o.Claim == nil || o.Claim.Holder == nil {
+ return false
+ }
+ p, c := j.Prior.Claim, o.Claim
+ a, b := p.Holder, c.Holder
+ if j.Prior.Host != o.Host || a.Incarnation != b.Incarnation || p.Op.ID != c.Op.ID || p.Op.Kind != c.Op.Kind || p.Op.Phase != c.Op.Phase || p.Op.Exec != c.Op.Exec || c.Op.Exec != box || b.WriterBox != box || b.Session <= a.Session || b.Stamp != b.Session || b.Generation != j.Generation || len(b.Admissions) == 0 {
+ return false
+ }
+ allowed := false
+ for _, session := range j.Sessions {
+ if session == b.Session {
+ allowed = true
+ break
+ }
+ }
+ if !allowed {
+ return false
+ }
+ last := b.Admissions[len(b.Admissions)-1]
+ return last.Session == b.Session && last.Box == box && last.Generation == j.Generation && last.Mode == "cold" && last.Predecessor == recoveryPredecessor(j.Prior).Generation
+}
+
+func (h claimCrossBoxHandover) recoveryJournal(ctx context.Context, o hostclaim.Observed) (string, string, *claimRecoveryJournal, error) {
+ if h.Registry == nil {
+ return "", "", nil, hostbox.ErrTypedInventory
+ }
+ bodies, err := h.Registry.PendingRecoveryHandovers(ctx)
+ if err != nil {
+ return "", "", nil, err
+ }
+ var found *claimRecoveryJournal
+ var id, body string
+ for key, raw := range bodies {
+ var j claimRecoveryJournal
+ if json.Unmarshal([]byte(raw), &j) != nil || j.Prior.Claim == nil || j.Prior.Claim.Holder == nil || j.Generation == "" {
+ return "", "", nil, hostbox.ErrChanged
+ }
+ if j.Prior.Host != o.Host {
+ continue
+ }
+ if found != nil || key != claimRecoveryJournalID(j.Prior) || (!hostclaim.SameWriterIdentity(j.Prior, o) && !j.successor(o, h.Box)) {
+ return "", "", nil, hostbox.ErrChanged
+ }
+ found, id, body = &j, key, raw
+ }
+ return id, body, found, nil
+}
+
+func (h claimCrossBoxHandover) prepareRecovery(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission) (hostclaim.Admission, error) {
+ if !claimInitialSourceRecovery(o, h.Box) && !(claimRecoveryObservation(o) && claimRemotePredecessor(o, h.Box)) {
+ return in, nil
+ }
+ if err := h.recoveryEvidence(ctx, o); err != nil {
+ return in, err
+ }
+ id, body, j, err := h.recoveryJournal(ctx, o)
+ if err != nil {
+ return in, err
+ }
+ if j != nil {
+ if !hostclaim.SameWriterIdentity(j.Prior, o) {
+ return in, hostbox.ErrChanged
+ }
+ in.NewGeneration = j.Generation
+ session := o.Claim.Gen + 1
+ for _, existing := range j.Sessions {
+ if existing == session {
+ return in, nil
+ }
+ }
+ j.Sessions = append(j.Sessions, session)
+ raw, err := json.Marshal(j)
+ if err != nil {
+ return in, err
+ }
+ return in, h.Registry.ReviseRecoveryHandover(ctx, id, body, string(raw))
+ }
+ if in.Mode != "cold" || in.KeepStamp || in.NewGeneration == "" {
+ return in, hostbox.ErrChanged
+ }
+ raw, err := json.Marshal(claimRecoveryJournal{Prior: o, Generation: in.NewGeneration, Sessions: []uint64{o.Claim.Gen + 1}})
+ if err != nil {
+ return in, err
+ }
+ return in, h.Registry.BeginRecoveryHandover(ctx, claimRecoveryJournalID(o), string(raw))
+}
+
func (h claimCrossBoxHandover) pendingRecovery(ctx context.Context, o hostclaim.Observed) (hostclaim.Observed, bool, error) {
if !claimRecoveryObservation(o) || o.Claim.Holder == nil || h.Registry == nil {
return o, false, nil
}
- body, err := h.Registry.PendingRecoveryHandover(ctx, claimRecoveryJournalID(o))
- if err != nil {
+ id, body, j, err := h.recoveryJournal(ctx, o)
+ if err != nil || j == nil {
return o, false, err
}
- if body == "" {
+ if hostclaim.SameWriterIdentity(j.Prior, o) {
return o, false, nil
}
- var j claimRecoveryJournal
- if json.Unmarshal([]byte(body), &j) != nil || j.Admitted.ETag != o.ETag || j.Admitted.Host != o.Host || j.Prior.Claim == nil || j.Prior.Claim.Holder == nil {
- return o, false, hostbox.ErrChanged
- }
if h.Launches == nil {
return o, false, hostbox.ErrTypedInventory
}
@@ -45,8 +134,7 @@ func (h claimCrossBoxHandover) pendingRecovery(ctx context.Context, o hostclaim.
return o, false, err
}
if len(launches) > 0 {
- // Once prelaunch exists, ordinary root stopping must settle that session.
- return o, false, h.Registry.CompleteRecoveryHandover(ctx, claimRecoveryJournalID(o), body)
+ return o, false, h.Registry.CompleteRecoveryHandover(ctx, id, body)
}
return j.Prior, true, nil
}
@@ -54,9 +142,12 @@ func (h claimCrossBoxHandover) completeRecovery(ctx context.Context, o hostclaim
if !claimRecoveryObservation(o) || o.Claim.Holder == nil || h.Registry == nil {
return nil
}
- body, err := h.Registry.PendingRecoveryHandover(ctx, claimRecoveryJournalID(o))
- if err != nil || body == "" {
+ id, body, j, err := h.recoveryJournal(ctx, o)
+ if err != nil || j == nil {
return err
}
- return h.Registry.CompleteRecoveryHandover(ctx, claimRecoveryJournalID(o), body)
+ if !j.successor(o, h.Box) {
+ return hostbox.ErrChanged
+ }
+ return h.Registry.CompleteRecoveryHandover(ctx, id, body)
}
diff --git a/internal/host/claimretirementcensus.go b/internal/host/claimretirementcensus.go
index 08d9d993f..7f21e9f4e 100644
--- a/internal/host/claimretirementcensus.go
+++ b/internal/host/claimretirementcensus.go
@@ -101,7 +101,7 @@ func claimFreeMoveProof(o hostclaim.Observed) (park.MoveSweepProof, error) {
}
func (p productionClaimRegional) retirement(ctx context.Context, o hostclaim.Observed, v []lease.ObjectVersion) (string, time.Time, error) {
- labels, err := ClaimRetirementCensus(o, v)
+ _, err := ClaimRetirementCensus(o, v)
var dependency *ClaimDependencyError
if err != nil && !errors.As(err, &dependency) {
return "", time.Time{}, err
@@ -113,30 +113,22 @@ func (p productionClaimRegional) retirement(ctx context.Context, o hostclaim.Obs
if now.IsZero() {
return "", time.Time{}, hostbox.ErrChanged
}
- var filtered []lease.ObjectVersion
- for _, x := range v {
- if strings.HasPrefix(x.Key, "moves/"+o.Host+"/") {
- filtered = append(filtered, x)
+ if proof, err := claimFreeMoveProof(o); err == nil {
+ if proof.RetiredAt.After(now) {
+ return "", time.Time{}, hostbox.ErrChanged
}
+ return proof.Incarnation, proof.RetiredAt, nil
}
- if len(filtered) == 0 {
- return labels["instances/"+o.Host], now, nil
+ if p.Parks == nil || p.Parks.Evidence == nil {
+ return "", time.Time{}, claimRetirementDependency()
}
- if proof, err := claimFreeMoveProof(o); err == nil {
- return proof.Incarnation, now, nil
- }
- label := labels["instances/"+o.Host]
- if o.Claim.Holder != nil {
- label = o.Claim.Holder.Incarnation
- } else if len(o.Claim.RetiredIncarnations) > 0 {
- label = o.Claim.RetiredIncarnations[0]
- }
- if p.Parks != nil && p.Parks.Evidence != nil {
- inc, at, err := p.Parks.Evidence.VerifyRetirement(ctx, o, filtered)
- if err == nil && !at.IsZero() && !at.After(now) {
- return inc, now, nil
- }
+ // The moves provider must vouch for instance generations as well as lanes.
+ inc, at, err := p.Parks.Evidence.VerifyRetirement(ctx, o, v)
+ if err != nil {
+ return "", time.Time{}, err
+ }
+ if inc == "" || at.IsZero() || at.After(now) {
+ return "", time.Time{}, hostbox.ErrChanged
}
- // A refused lane is dropped by VerifyMovesSweep; it cannot block proven instance residue.
- return label, now, nil
+ return inc, at, nil
}
diff --git a/internal/host/claimretryrecovery_test.go b/internal/host/claimretryrecovery_test.go
index aa875546b..a7d1f96ff 100644
--- a/internal/host/claimretryrecovery_test.go
+++ b/internal/host/claimretryrecovery_test.go
@@ -65,10 +65,12 @@ func (recoveryInputs) Publish(context.Context, hostclaim.Observed, ClaimLaunchIn
}
type recoveryKernel struct {
- journal *hostbox.LaunchJournal
- alive bool
- verified []hostguard.FenceTarget
- recovered []hostguard.FenceTarget
+ noProcess bool
+ noProcessCalls int
+ journal *hostbox.LaunchJournal
+ alive bool
+ verified []hostguard.FenceTarget
+ recovered []hostguard.FenceTarget
}
func (k *recoveryKernel) evidence(inv StopInventory, j hostbox.StopKernelJournal) StopKernelEvidence {
@@ -109,6 +111,8 @@ func (k *recoveryKernel) RecoverStopped(ctx context.Context, t hostguard.FenceTa
}
type recoveryFixture struct {
+ data genselect.Reader
+ bucket string
t *testing.T
claims release.Store
evidence release.Store
@@ -256,7 +260,7 @@ func (f *recoveryFixture) controller(boot, controller string, pid int, start str
}
a.bindClaimWakeStop()
source := hostbox.WorkSource{Guard: f.guard, Box: "box-1", Region: "ie", Now: func() time.Time { return f.now }, Policy: func(context.Context, *hostclaim.Claim, hostbox.Selection, hostwrite.Identity) error { return nil }, Local: func(context.Context, *hostclaim.Claim, hostbox.Selection) error { return nil }}
- writer := hostbox.Writer{Claims: f.claims, Box: "box-1", Region: "ie", AdmissionEvidence: func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error { return nil }}
+ writer := hostbox.Writer{Data: f.data, Bucket: f.bucket, Claims: f.claims, Box: "box-1", Region: "ie", AdmissionEvidence: func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error { return nil }}
runtime, err := w.Build(t.Context(), source, writer)
if err != nil {
t.Fatal(err)
@@ -622,3 +626,15 @@ func TestClaimRetryRecoveryRetainsRecordedDirectoryRevision(t *testing.T) {
t.Fatalf("retry = %+v", retry)
}
}
+
+func (k *recoveryKernel) VerifyNoProcess(ctx context.Context, target hostguard.FenceTarget, inv StopInventory) (StopKernelEvidence, error) {
+ k.noProcessCalls++
+ if !k.noProcess {
+ return StopKernelEvidence{}, ErrClaimStopInventory
+ }
+ has, err := k.journal.HasHost(ctx, target.Host)
+ if err != nil || has {
+ return StopKernelEvidence{}, ErrClaimStopInventory
+ }
+ return StopKernelEvidence{InventorySHA256: digestStop(inv), LocalJournalSHA256: strings.Repeat("c", 64), CgroupIdentities: []string{"absent:carlos-exec@shop.example"}, UnitInvocationIDs: []string{}}, nil
+}
diff --git a/internal/host/claimwakeassembly.go b/internal/host/claimwakeassembly.go
index 56aec0dac..6a21e881b 100644
--- a/internal/host/claimwakeassembly.go
+++ b/internal/host/claimwakeassembly.go
@@ -90,6 +90,7 @@ type claimSyncer interface {
// ClaimWakeAssembly owns cold restore, lease transfer and failure cleanup.
// Root supplies Stop and Launch; moves supplies only its input/publication API.
type ClaimWakeAssembly struct {
+ recoveryIntent func(context.Context, hostclaim.Observed, hostclaim.Admission) (hostclaim.Admission, error)
recoveryPrior func(context.Context, hostclaim.Observed) (hostclaim.Observed, bool, error)
recoveryPrelaunched func(context.Context, hostclaim.Observed) error
retained []*retainedClaimWork
@@ -207,7 +208,7 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
runtime := &activator.ClaimRuntime{Box: writer.Box, Read: read, RecoveryAttemptsReady: w.recoveryAttemptsReady}
runtime.ResumedAdmission = func(call context.Context, prior, admitted hostclaim.Observed) error {
a := w.get(admitted.Host)
- if a == nil || prior.ETag == "" || prior.ETag != admitted.ETag || prior.Host != admitted.Host {
+ if a == nil || !hostclaim.SameWriterIdentity(prior, admitted) {
return hostbox.ErrChanged
}
current, err := read(call, admitted.Host)
@@ -216,7 +217,7 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
}
a.mu.Lock()
defer a.mu.Unlock()
- if !a.resumedAdmission || a.observed.ETag != admitted.ETag || current.ETag != admitted.ETag {
+ if !a.resumedAdmission || !hostclaim.SameWriterIdentity(a.observed, admitted) || !hostclaim.SameWriterIdentity(current, admitted) {
return hostbox.ErrChanged
}
a.resumedAdmission = false
@@ -338,6 +339,12 @@ func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource
}
admitted := o
if !resuming {
+ if w.recoveryIntent != nil {
+ in.Admission, err = w.recoveryIntent(call, o, in.Admission)
+ if err != nil {
+ return admitted, err
+ }
+ }
admitted, err = writer.Admit(call, o, in.Admission, in.Baseline)
} else {
in.Admission.NewGeneration = o.Claim.Holder.Generation
diff --git a/internal/host/claimwakestop.go b/internal/host/claimwakestop.go
index cf3b8d883..6f4aaf5b3 100644
--- a/internal/host/claimwakestop.go
+++ b/internal/host/claimwakestop.go
@@ -176,7 +176,9 @@ func (a ClaimAssembly) stopCapturedSession(ctx context.Context, captured hostcla
return err
}
if a.stopReaches() != nil && claimInitialSourceRecovery(captured, local.Actor.Box) {
- return (claimCrossBoxHandover{Guard: a.Guard, Evidence: a.Claims, Box: local.Actor.Box}).beforeRecovery(ctx, captured)
+ if err := (claimCrossBoxHandover{Guard: a.Guard, Evidence: a.Claims, Box: local.Actor.Box}).sourceRecoveryEvidence(ctx, captured); err != nil {
+ return err
+ }
}
if h.Session == 0 {
// No writer was ever admitted, so nothing was launched or journaled.
@@ -279,7 +281,7 @@ func (a ClaimAssembly) collectCapturedStop(ctx context.Context, captured hostcla
actor := local.Actor
claim, holder := observed.Claim, observed.Claim.Holder
obligation := ""
- if id, ok := a.operationWriterStop(claim, actor); writerSession && ok {
+ if id, ok := a.operationWriterStop(claim, actor); (writerSession || zero) && ok {
obligation = id
} else if claim.Op.Kind != hostclaim.KindServing {
for _, ob := range claim.Obligations {
diff --git a/internal/host/plan12_census_test.go b/internal/host/plan12_census_test.go
index 338f784a6..ec06b09d1 100644
--- a/internal/host/plan12_census_test.go
+++ b/internal/host/plan12_census_test.go
@@ -96,7 +96,7 @@ func TestPlan12LegacyAndMovesHaveIndependentLabels(t *testing.T) {
}
p := productionClaimRegional{Now: func() time.Time { return time.Unix(100, 0) }}
label, at, err := p.retirement(t.Context(), o, versions[:1])
- if err != nil || label != legacy || at.IsZero() {
+ if err == nil || label != "" || !at.IsZero() {
t.Fatal(label, at, err)
}
}
diff --git a/internal/host/plan12_fix1_test.go b/internal/host/plan12_fix1_test.go
new file mode 100644
index 000000000..9f5069cc9
--- /dev/null
+++ b/internal/host/plan12_fix1_test.go
@@ -0,0 +1,218 @@
+package host
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/registry"
+ "path/filepath"
+ "reflect"
+ "testing"
+ "time"
+)
+
+func TestPlan12InventoryReplayChecksAllTimers(t *testing.T) {
+ for _, mode := range []string{"foreign-owner", "stale-incarnation", "route-changed", "ok"} {
+ t.Run(mode, func(t *testing.T) {
+ f, prior, h := plan12Handover(t, hostclaim.KindRestoring, "stop")
+ path := filepath.Join(t.TempDir(), "registry.db")
+ original, err := h.Registry.Get(prior.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ h.Registry, err = registry.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer h.Registry.Close()
+ if err = h.Registry.Upsert(*original); err != nil {
+ t.Fatal(err)
+ }
+ in := hostclaim.Admission{ObservedOpID: prior.Claim.Op.ID, ObservedPhase: prior.Claim.Op.Phase, ObservedSession: prior.Claim.Holder.Session, ObservedGeneration: prior.Claim.Holder.Generation, Box: h.Box, Mode: "cold", NewGeneration: "successor", Region: "ie", PredecessorRegion: "uk"}
+ admitted, err := hostclaim.Update(t.Context(), f.claims, prior, hostclaim.AdmitWriter(in))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := h.After(t.Context(), prior, admitted); err != nil {
+ t.Fatal(err)
+ }
+ row, err := h.Registry.Get(prior.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if mode == "route-changed" {
+ row.Version = "changed"
+ if err := h.Registry.Upsert(*row); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if mode == "foreign-owner" || mode == "stale-incarnation" {
+ owner := row.ClaimOwnership
+ owner.ClaimRowID = "injected"
+ if mode == "foreign-owner" {
+ owner.ClaimOwner = "another-box"
+ } else {
+ owner.ClaimIncarnation = "stale"
+ }
+ if err := h.Registry.UpsertTimer(registry.Timer{Host: prior.Host, Name: "injected", Kind: "cron", Spec: "* * * * *"}); err != nil {
+ t.Fatal(err)
+ }
+ raw, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer raw.Close()
+ if _, err := raw.Exec("UPDATE timers SET claim_incarnation=?,claim_stamp=?,claim_owner=?,claim_row_id=? WHERE host=?", owner.ClaimIncarnation, owner.ClaimStamp, owner.ClaimOwner, owner.ClaimRowID, prior.Host); err != nil {
+ t.Fatal(err)
+ }
+ }
+ err = h.After(t.Context(), prior, admitted)
+ if (err == nil) != (mode == "ok") {
+ t.Fatalf("replay %s: %v", mode, err)
+ }
+ })
+ }
+}
+
+func TestPlan12InitialSourceRecoveryRequiresRootStop(t *testing.T) {
+ for _, session := range []uint64{0, 2} {
+ t.Run(fmt.Sprint(session), func(t *testing.T) {
+ f, o, _ := plan12Handover(t, hostclaim.KindRestoring, "stop")
+ key := moves.StoppedKey(moves.ReportPrefix("bab", "jam", o.Host, o.Claim.Op.ID), "source")
+ report, _ := json.Marshal(plan12Stopped(o, "source", "box-1"))
+ if _, err := f.claims.Put(t.Context(), key, report, ""); err != nil {
+ t.Fatal(err)
+ }
+ o, err := hostclaim.Update(t.Context(), f.claims, o, func(c *hostclaim.Claim) error {
+ c.Holder.Session = session
+ c.Holder.WriterBox, c.Holder.Generation = "box-1", "src"
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{{Session: session, Box: "box-1", Generation: "src", Mode: "cold", Region: "ie"}}
+ if session == 0 {
+ c.Holder.WriterBox, c.Holder.Generation = "", ""
+ c.Holder.Admissions = nil
+ }
+ c.Op.Journal[hostclaim.JournalCutSession] = fmt.Sprint(session)
+ c.Op.Journal[hostclaim.JournalSrcGen] = c.Holder.Generation
+ c.Op.Journal[hostclaim.JournalEvidenceSource] = key
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := f.controller("boot-1", "controller-1", 4101, "start-1")
+ c.wake.Continue = hostclaim.SessionContinues
+ c.assembly.Guard = f.guard
+ c.assembly.bindClaimWakeStop()
+ if err := c.wake.Stop(t.Context(), o); err == nil {
+ t.Fatal("agent report substituted for root stop")
+ }
+ if len(f.requests) == 0 {
+ t.Fatal("root was not asked to stop the cut installation")
+ }
+ })
+ }
+}
+
+type plan12RetirementEvidence struct {
+ parkRegistrationEvidence
+ seen []lease.ObjectVersion
+ at time.Time
+ err error
+}
+
+func (e *plan12RetirementEvidence) VerifyRetirement(_ context.Context, o hostclaim.Observed, v []lease.ObjectVersion) (string, time.Time, error) {
+ e.seen = append([]lease.ObjectVersion(nil), v...)
+ return o.Claim.Holder.Incarnation, e.at, e.err
+}
+func TestPlan12RetirementUsesOriginalEvidenceAndFullCensus(t *testing.T) {
+ for _, refuses := range []bool{false, true} {
+ t.Run(fmt.Sprint(refuses), func(t *testing.T) {
+ f := newRecoveryFixture(t)
+ o := f.read()
+ versions := []lease.ObjectVersion{{Key: o.Claim.Holder.Location + "/g-old/db", VersionID: "one"}, {Key: "moves/" + o.Host + "/old/db", VersionID: "two"}}
+ proof := &plan12RetirementEvidence{at: time.Unix(100, 0)}
+ if refuses {
+ proof.err = errors.New("move did not retire this generation")
+ }
+ p := productionClaimRegional{Now: func() time.Time { return time.Unix(200, 0) }, Parks: &claimMoveParkSources{Evidence: proof}}
+ _, at, err := p.retirement(t.Context(), o, versions)
+ if !reflect.DeepEqual(proof.seen, versions) {
+ t.Fatalf("incomplete retirement census: %+v", proof.seen)
+ }
+ if refuses {
+ if !errors.Is(err, proof.err) {
+ t.Fatalf("ignored refusal: %v", err)
+ }
+ } else if err != nil || !at.Equal(proof.at) {
+ t.Fatalf("retirement time=%v error=%v", at, err)
+ }
+ })
+ }
+}
+func TestPlan12RetirementRefusalAlertOncePerLocation(t *testing.T) {
+ o := plan12Retired(t)
+ loc := "instances/" + o.Host
+ view := park.View{ClaimETag: o.ETag, Region: "ie", RefusedLocations: map[string]error{loc: claimRetirementDependency()}}
+ versions := []lease.ObjectVersion{{Key: loc + "/0000/db", VersionID: "one"}, {Key: loc + "/0000/db", VersionID: "two"}}
+ d, err := park.Decide(o, view, versions, "")
+ if err != nil || len(d.Alerts) != 1 || len(d.Candidates) != 0 {
+ t.Fatalf("decision=%+v err=%v", d, err)
+ }
+}
+
+func TestPlan12PendingRecoveryRejectsChangedIdentity(t *testing.T) {
+ for _, field := range []string{"same", "session", "op", "writer", "incarnation", "stamp", "generation", "kind", "phase", "exec"} {
+ t.Run(field, func(t *testing.T) {
+ f, o, h := plan12Handover(t, hostclaim.KindRestoring, "stop")
+ h.Launches = f.journal
+ in := hostclaim.Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Box: h.Box, Mode: "cold", NewGeneration: "successor", Region: "ie", PredecessorRegion: "uk"}
+ in, err := h.prepareRecovery(t.Context(), o, in)
+ if err != nil {
+ t.Fatal(err)
+ }
+ next, err := hostclaim.Update(t.Context(), f.claims, o, hostclaim.AdmitWriter(in))
+ if err != nil {
+ t.Fatal(err)
+ }
+ c := next.Claim
+ holder := c.Holder
+ switch field {
+ case "session":
+ holder.Session++
+ holder.Stamp = holder.Session
+ holder.Admissions[len(holder.Admissions)-1].Session = holder.Session
+ case "op":
+ c.Op.ID = "other"
+ case "writer":
+ holder.WriterBox = "other"
+ case "incarnation":
+ holder.Incarnation = "other"
+ case "stamp":
+ holder.Stamp++
+ case "generation":
+ holder.Generation = "other"
+ case "kind":
+ c.Op.Kind = hostclaim.KindReverting
+ case "phase":
+ c.Op.Phase = "other"
+ case "exec":
+ c.Op.Exec = "other"
+ }
+ _, pending, err := h.pendingRecovery(t.Context(), next)
+ if field == "same" {
+ if err != nil || !pending {
+ t.Fatal(pending, err)
+ }
+ } else if !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatalf("changed %s accepted: %v", field, err)
+ }
+ })
+ }
+}
diff --git a/internal/host/plan12_handover_test.go b/internal/host/plan12_handover_test.go
index 2572762ee..e4a6d1012 100644
--- a/internal/host/plan12_handover_test.go
+++ b/internal/host/plan12_handover_test.go
@@ -56,7 +56,7 @@ func plan12Handover(t *testing.T, kind, mode string) (*recoveryFixture, hostclai
journal = map[string]string{"fence.box-2": string(refRaw)}
}
if mode == "loss" || mode == "wrong-loss-op" {
- journal = map[string]string{"loss.loss_1": "{}"}
+ journal = map[string]string{"loss.loss_1": `{"kind":"move-target-lost","target":"box-2"}`}
}
var err error
o, err = hostclaim.Update(t.Context(), f.claims, o, func(c *hostclaim.Claim) error {
@@ -221,11 +221,25 @@ func TestPlan12ConsumedFenceSurvivesRosterChange(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- n := 0
- scope := func(loc string) hostclaim.ParkScope {
- return hostclaim.ParkScope{Location: loc, Incarnation: o.Claim.Holder.Incarnation, EvidenceAt: fence.ObservedAt(), RequestLimit: 2 * time.Minute}
- }
- o, err = hostclaim.Update(t.Context(), f.claims, o, hostclaim.ReleaseFencedTarget(hostclaim.FencedTargetRelease{OpID: o.Claim.Op.ID, RollbackID: rb, Target: "box-2", TargetRegion: "uk", SourceRegion: "ie", Fence: fence, Location: scope(o.Claim.Holder.Location), Lane: scope("moves/" + o.Host + "/" + o.Claim.Op.ID), NewID: func() string { n++; return fmt.Sprint("park-", n) }}))
+
+ // Admission consumes a previously grant-released claim. The grant itself
+ // is exercised in hostops.TestPlan12RecoveryFenceGrant.
+ o, err = hostclaim.Update(t.Context(), f.claims, o, func(c *hostclaim.Claim) error {
+ raw, _ := json.Marshal(ref)
+ sum := sha256.Sum256([]byte(c.Op.ID + "\x00box-2\x00" + ref.Enrollment + "\x00" + ref.SHA256))
+ id := "auto-" + hex.EncodeToString(sum[:16])
+ loss := hostclaim.AutomaticLoss{ID: id, Box: "box-2", Enrollment: ref.Enrollment, Fence: ref, ObservedAt: fence.ObservedAt().Format(time.RFC3339Nano), Op: c.Op.ID, Kind: c.Op.Kind, Phase: c.Op.Phase, Role: "target", Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Session: c.Holder.Session, Generation: c.Holder.Generation, Timers: hostclaim.TimersUnknown}
+ if len(c.Holder.Admissions) > 0 {
+ a := c.Holder.Admissions[len(c.Holder.Admissions)-1]
+ loss.LastAdmission = &a
+ }
+ c.Op.Kind, c.Op.Phase, c.Op.Exec = hostclaim.KindRestoring, hostclaim.PhaseReleased, "none"
+ c.Op.Intent["rollback"] = rb
+ c.Op.Participants = []string{c.Op.Intent["source"]}
+ c.Op.Journal["fence.box-2"], c.Op.Journal["fenced_target_released"] = string(raw), string(raw)
+ c.Lost, c.LossRefs, c.EvidenceRefs = []string{"box-2"}, []string{id}, []hostclaim.AutomaticLoss{loss}
+ return nil
+ })
if err != nil {
t.Fatal(err)
}
diff --git a/internal/host/plan12_restore_linux_test.go b/internal/host/plan12_restore_linux_test.go
index 52c9e7e67..21683a401 100644
--- a/internal/host/plan12_restore_linux_test.go
+++ b/internal/host/plan12_restore_linux_test.go
@@ -3,19 +3,26 @@
package host
import (
+ "bytes"
"context"
"database/sql"
+ "encoding/binary"
"encoding/json"
"errors"
+ "fmt"
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/cgroup"
+ "github.com/carlosframework/platform/internal/genselect"
"github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/park"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
"golang.org/x/sys/unix"
"os"
"os/exec"
@@ -25,11 +32,13 @@ import (
)
func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
- for _, name := range []string{hostclaim.KindRestoring, hostclaim.KindReverting, "clean-return", "never-restoring", "never-reverting", "never-clean-return", "missing-inventory", "wrong-inventory", "missing-registry", "interrupted"} {
+ for _, name := range []string{hostclaim.KindRestoring, hostclaim.KindReverting, "clean-return", "never-restoring", "never-reverting", "never-clean-return", "missing-inventory", "wrong-inventory", "missing-registry", "interrupted", "admission-crash", "prejournal-failure", "intent-crash", "evidence-read-failure", "concurrent-update", "prelaunch-update"} {
t.Run(name, func(t *testing.T) {
kind, clean := name, name == "clean-return" || name == "never-clean-return"
bad := name == "missing-inventory" || name == "wrong-inventory" || name == "missing-registry"
- if bad || name == "interrupted" {
+ beforeAdmission := name == "prejournal-failure" || name == "intent-crash" || name == "evidence-read-failure"
+ retryCase := name == "interrupted" || name == "admission-crash" || beforeAdmission || name == "prelaunch-update"
+ if bad || retryCase || name == "concurrent-update" {
kind = hostclaim.KindRestoring
}
if name == "never-restoring" {
@@ -44,6 +53,7 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
}
f, o, h := plan12Handover(t, hostclaim.KindRestoring, "stop")
stamp := o.Claim.Holder.Stamp
+ baseline := plan12RestoreBaseline(t, f)
db := provision.NestedDBPath(f.guard.DataDir, o.Host)
dbRel, _ := filepath.Rel(f.guard.DataDir, db)
dirRel, _ := filepath.Rel(f.guard.DataDir, filepath.Dir(db))
@@ -111,7 +121,17 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
}
o = f.read()
c := f.controller("boot-1", "controller-1", 4101, "start-1")
- c.wake.Inputs = plan12ReturnInputs{plan9D2Inputs: plan9D2Inputs{recoveryInputs{f}}, clean: clean}
+ if claimInitialSourceRecovery(o, "box-1") {
+ if never {
+ f.kernel.noProcess = true
+ } else {
+ target := recoveryTuple(o, c.local)
+ if err := f.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{ControllerPID: 4101, ProcessStart: "start-1", Target: target, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/" + claimContainmentName(target), Device: 1, Inode: target.Session}}, NoState: true}); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ c.wake.Inputs = plan12ReturnInputs{plan9D2Inputs: plan9D2Inputs{recoveryInputs{f}}, clean: clean, baseline: baseline}
c.wake.Continue = hostclaim.SessionContinues
c.wake.Operations = &plan9D2Attempts{}
c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
@@ -154,6 +174,7 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
h.Registry = nil
}
h.Launches = f.journal
+ savedRegistry := h.Registry
interrupted := false
if name == "interrupted" {
h.afterRecoveryRows = func() error { interrupted = true; return hostbox.ErrChanged }
@@ -162,9 +183,44 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
}
}
wireClaimAdmission(c.wake, "box-1", c.wake.AfterAdmission, h, nil)
+ if name == "concurrent-update" {
+ h.afterRecoveryRows = func() error {
+ _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error { c.Op.Journal["concurrent"] = "acknowledged"; return nil })
+ return err
+ }
+ wireClaimAdmission(c.wake, "box-1", c.wake.AfterAdmission, h, nil)
+ }
+ if name == "intent-crash" {
+ prepare := c.wake.recoveryIntent
+ c.wake.recoveryIntent = func(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission) (hostclaim.Admission, error) {
+ in, err := prepare(ctx, o, in)
+ if err != nil {
+ return in, err
+ }
+ interrupted = true
+ return in, hostbox.ErrChanged
+ }
+ }
+ if name == "prelaunch-update" {
+ c.wake.recoveryPrelaunched = func(context.Context, hostclaim.Observed) error { interrupted = true; return hostbox.ErrChanged }
+ }
+ if name == "admission-crash" {
+ c.wake.AfterAdmission = func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
+ interrupted = true
+ return hostbox.ErrChanged
+ }
+ }
+ if name == "prejournal-failure" {
+ h.Registry = nil
+ wireClaimAdmission(c.wake, "box-1", c.wake.AfterAdmission, h, nil)
+ }
if err := c.wake.stopPredecessor(t.Context(), o); err != nil {
t.Fatal(err)
}
+ if name == "evidence-read-failure" {
+ h.Evidence = &plan12FailEvidence{Store: f.claims, fail: true}
+ wireClaimAdmission(c.wake, "box-1", c.wake.AfterAdmission, h, nil)
+ }
live, cancel := context.WithCancel(t.Context())
defer cancel()
done := make(chan struct{})
@@ -173,20 +229,37 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
c.wake.pending[o.Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
c.wake.mu.Unlock()
admitted, err := c.runtime.Admit(t.Context(), registry.Route{Host: o.Host}, o, activator.StartPolicy{Probe: true})
- if name == "interrupted" {
+ if retryCase {
+ if name == "prejournal-failure" {
+ interrupted = err != nil
+ h.Registry = savedRegistry
+ }
+ if name == "evidence-read-failure" {
+ interrupted = err != nil
+ h.Evidence = f.claims
+ }
if err == nil || !interrupted {
t.Fatal("did not interrupt after registry commit", err)
}
first := f.read()
+ first, e = hostclaim.Update(t.Context(), f.claims, first, func(c *hostclaim.Claim) error {
+ c.Op.Journal["unrelated"] = "park acknowledged"
+ return nil
+ })
+ if e != nil {
+ t.Fatal(e)
+ }
row, e := h.Registry.Get(o.Host)
- if e != nil || row.ClaimStamp != first.Claim.Holder.Stamp {
+ if e != nil || name == "interrupted" && row.ClaimStamp != first.Claim.Holder.Stamp {
t.Fatal("registry did not commit", e)
}
- if _, e = h.Inventory.ReadInstallation(t.Context(), o.Host, first.Claim.Holder.Incarnation, first.Claim.Holder.Stamp); !errors.Is(e, sql.ErrNoRows) {
- t.Fatal("successor inventory already written", e)
+ if name == "interrupted" {
+ if _, e = h.Inventory.ReadInstallation(t.Context(), o.Host, first.Claim.Holder.Incarnation, first.Claim.Holder.Stamp); !errors.Is(e, sql.ErrNoRows) {
+ t.Fatal("successor inventory already written", e)
+ }
}
c = f.controller("boot-2", "controller-2", 4102, "start-2")
- c.wake.Inputs = plan12ReturnInputs{plan9D2Inputs: plan9D2Inputs{recoveryInputs{f}}}
+ c.wake.Inputs = plan12ReturnInputs{plan9D2Inputs: plan9D2Inputs{recoveryInputs{f}}, baseline: baseline}
c.wake.Continue = hostclaim.SessionContinues
c.wake.Operations = &plan9D2Attempts{}
c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
@@ -199,20 +272,32 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
if e := c.wake.stopPredecessor(t.Context(), first); e != nil {
t.Fatal("retry chose predecessor stop before handover", e)
}
+ if name == "prelaunch-update" {
+ bodies, err := h.Registry.PendingRecoveryHandovers(t.Context())
+ if err != nil || len(bodies) != 0 {
+ t.Fatalf("prelaunch journal stayed pending: %+v %v", bodies, err)
+ }
+ if len(f.requests) == 0 {
+ t.Fatal("prelaunch retry did not reach root stop")
+ }
+ return
+ }
retryCtx, retryCancel := context.WithCancel(t.Context())
defer retryCancel()
retryDone := make(chan struct{})
go func() { <-retryCtx.Done(); close(retryDone) }()
c.wake.pending[o.Host] = &claimWakeAttempt{observed: first, held: recoveryHeld{}, ctx: retryCtx, cancel: retryCancel, done: retryDone}
admitted, err = c.runtime.Admit(t.Context(), registry.Route{Host: o.Host}, first, activator.StartPolicy{Probe: true})
- if err != nil || admitted.Claim.Holder.Session != first.Claim.Holder.Session {
+ if err != nil || !beforeAdmission && admitted.Claim.Holder.Session != first.Claim.Holder.Session {
t.Fatal("retry did not resume admitted session", err)
}
- if err = c.runtime.ResumedAdmission(t.Context(), first, admitted); err != nil {
- t.Fatal("activator could not verify resumed admission", err)
- }
- if err := c.runtime.ResumedAdmission(t.Context(), first, admitted); !errors.Is(err, hostbox.ErrChanged) {
- t.Fatal("resume verification was not single use", err)
+ if !beforeAdmission {
+ if err = c.runtime.ResumedAdmission(t.Context(), first, admitted); err != nil {
+ t.Fatal("activator could not verify resumed admission", err)
+ }
+ if err := c.runtime.ResumedAdmission(t.Context(), first, admitted); !errors.Is(err, hostbox.ErrChanged) {
+ t.Fatal("resume verification was not single use", err)
+ }
}
}
if bad {
@@ -229,14 +314,33 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
t.Fatal(err)
}
if never {
+ if f.kernel.noProcessCalls == 0 {
+ t.Fatal("R4 witness was not checked")
+ }
return
}
if err := launch.Restore(t.Context()); err != nil {
t.Fatal(err)
}
- if got, err := os.ReadFile(db + ".gen"); err != nil || string(got) == "src" {
+ if got, err := os.ReadFile(db + ".gen"); err != nil || string(got) != admitted.Claim.Holder.Generation+"\n" {
t.Fatal(string(got), err)
}
+ restored, err := sql.Open("sqlite", "file:"+db+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer restored.Close()
+ var value string
+ if err = restored.QueryRow("SELECT value FROM fixture").Scan(&value); err != nil || value != "selected recovery baseline" {
+ t.Fatalf("restored database: %q %v", value, err)
+ }
+ retired := filepath.Join(f.guard.DataDir, ".parked", admitted.Claim.Holder.Incarnation, o.Host, fmt.Sprintf("superseded-%d-%d", admitted.Claim.Holder.Session, stamp), "artifacts", dbRel)
+ if got, err := os.ReadFile(retired); err != nil || string(got) != "source database" {
+ t.Fatalf("retired predecessor: %q %v", got, err)
+ }
+ if got, err := os.ReadFile(retired + ".gen"); err != nil || string(got) != "src\n" {
+ t.Fatalf("retired marker: %q %v", got, err)
+ }
if tag := f.read().Claim.Inventory["box-1"].Paths[dbRel]; tag != admitted.Claim.Holder.Stamp {
t.Fatal("old ownership", tag)
}
@@ -246,15 +350,75 @@ func TestPlan12RecoveryRestoreSourceDatabase(t *testing.T) {
type plan12ReturnInputs struct {
plan9D2Inputs
- clean bool
+ clean bool
+ baseline *genselect.Plan
}
func (in plan12ReturnInputs) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) {
selected, err := in.plan9D2Inputs.Select(ctx, rt, o, p)
selected.Admission.KeepStamp = false
+ if in.baseline != nil {
+ selected.Empty = false
+ selected.Baseline = in.baseline
+ selected.Admission.RemoteMaxTXID = in.baseline.MaxTXID
+ }
selected.DBPath = provision.NestedDBPath(in.f.guard.DataDir, o.Host)
if in.clean {
selected.Admission.Coverage = "clean-return"
}
return selected, err
}
+
+func plan12RestoreBaseline(t *testing.T, f *recoveryFixture) *genselect.Plan {
+ t.Helper()
+ path := filepath.Join(t.TempDir(), "selected.db")
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec("CREATE TABLE fixture(value TEXT); INSERT INTO fixture VALUES ('selected recovery baseline')"); err != nil {
+ t.Fatal(err)
+ }
+ if err := db.Close(); err != nil {
+ t.Fatal(err)
+ }
+ raw, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ size := int(binary.BigEndian.Uint16(raw[16:18]))
+ if size == 1 {
+ size = 65536
+ }
+ spec := ltx.FileSpec{Header: ltx.Header{Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: uint32(size), Commit: uint32(len(raw) / size), MinTXID: 1, MaxTXID: 1}}
+ for off := 0; off < len(raw); off += size {
+ pg := uint32(off/size + 1)
+ if pg != ltx.LockPgno(uint32(size)) {
+ spec.Pages = append(spec.Pages, ltx.PageSpec{Header: ltx.PageHeader{Pgno: pg}, Data: raw[off : off+size]})
+ }
+ }
+ var out bytes.Buffer
+ if _, err := spec.WriteTo(&out); err != nil {
+ t.Fatal(err)
+ }
+ key := "instances/" + recoveryHost + "/g-src/0009/" + ltx.FormatFilename(1, 1)
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ version := bucket.Put("data", key, out.Bytes())
+ bucket.Put("data", key, []byte("newer unselected version"))
+ f.data, f.bucket = bucket, "data"
+ return &genselect.Plan{Generation: "src", MaxTXID: 1, Pin: &genselect.Pin{SnapshotCRC64: uint64(spec.Trailer.FileChecksum)}, Files: []genselect.File{{Key: key, VersionID: version, Level: 9, MinTXID: 1, MaxTXID: 1, Size: int64(out.Len())}}}
+}
+
+type plan12FailEvidence struct {
+ release.Store
+ fail bool
+}
+
+func (s *plan12FailEvidence) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if s.fail {
+ s.fail = false
+ return nil, "", hostbox.ErrChanged
+ }
+ return s.Store.Get(ctx, key)
+}
diff --git a/internal/hostbox/plan12_fence_loss_test.go b/internal/hostbox/plan12_fence_loss_test.go
index 93d03a293..6e1c7c0e1 100644
--- a/internal/hostbox/plan12_fence_loss_test.go
+++ b/internal/hostbox/plan12_fence_loss_test.go
@@ -7,7 +7,6 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
- "fmt"
"github.com/carlosframework/platform/internal/genselect"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/termfence"
@@ -46,11 +45,25 @@ func TestPlan12FencedTargetLossMaximum(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- scope := func(loc string) hostclaim.ParkScope {
- return hostclaim.ParkScope{Location: loc, Incarnation: o.Claim.Holder.Incarnation, EvidenceAt: now, RequestLimit: 2 * time.Minute}
- }
- n := 0
- o, err = hostclaim.Update(t.Context(), s, o, hostclaim.ReleaseFencedTarget(hostclaim.FencedTargetRelease{OpID: "move", RollbackID: "rb_abcdefghijklmnopqrstuvwxyz", Target: "target", TargetRegion: "uk", SourceRegion: "ie", Location: scope(o.Claim.Holder.Location), Lane: scope("moves/" + o.Host + "/move"), NewID: func() string { n++; return fmt.Sprint("park-", n) }, Fence: fence}))
+
+ // Admission consumes a previously grant-released claim. The grant itself
+ // is exercised in hostops.TestPlan12RecoveryFenceGrant.
+ o, err = hostclaim.Update(t.Context(), s, o, func(c *hostclaim.Claim) error {
+ raw, _ := json.Marshal(ref)
+ sum := sha256.Sum256([]byte(c.Op.ID + "\x00target\x00" + ref.Enrollment + "\x00" + ref.SHA256))
+ id := "auto-" + hex.EncodeToString(sum[:16])
+ loss := hostclaim.AutomaticLoss{ID: id, Box: "target", Enrollment: ref.Enrollment, Fence: ref, ObservedAt: fence.ObservedAt().Format(time.RFC3339Nano), Op: c.Op.ID, Kind: c.Op.Kind, Phase: c.Op.Phase, Role: "target", Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Session: c.Holder.Session, Generation: c.Holder.Generation, Timers: hostclaim.TimersUnknown}
+ if len(c.Holder.Admissions) > 0 {
+ a := c.Holder.Admissions[len(c.Holder.Admissions)-1]
+ loss.LastAdmission = &a
+ }
+ c.Op.Kind, c.Op.Phase, c.Op.Exec = hostclaim.KindRestoring, hostclaim.PhaseReleased, "none"
+ c.Op.Intent["rollback"] = "rb_abcdefghijklmnopqrstuvwxyz"
+ c.Op.Participants = []string{c.Op.Intent["source"]}
+ c.Op.Journal["fence.target"], c.Op.Journal["fenced_target_released"] = string(raw), string(raw)
+ c.Lost, c.LossRefs, c.EvidenceRefs = []string{"target"}, []string{id}, []hostclaim.AutomaticLoss{loss}
+ return nil
+ })
if err != nil {
t.Fatal(err)
}
diff --git a/internal/hostbox/plan12_loss_test.go b/internal/hostbox/plan12_loss_test.go
index 3e3e9dc2a..5fcaf8a0e 100644
--- a/internal/hostbox/plan12_loss_test.go
+++ b/internal/hostbox/plan12_loss_test.go
@@ -35,7 +35,7 @@ func plan12LossAdmissionCases(t *testing.T, modes []string) {
_, store, o := workFixture(t)
var err error
o, err = hostclaim.Update(t.Context(), store, o, func(c *hostclaim.Claim) error {
- c.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: "op", Phase: hostclaim.PhaseProbing, Exec: "box", Participants: []string{"box"}, Journal: map[string]string{"loss.loss_1": "{}"}}
+ c.Op = hostclaim.Op{Kind: hostclaim.KindRestoring, ID: "op", Phase: hostclaim.PhaseProbing, Exec: "box", Participants: []string{"box"}, Intent: map[string]string{"target": "target"}, Journal: map[string]string{"loss.loss_1": `{"kind":"move-target-lost","target":"target"}`}}
c.Holder.Session, c.Holder.Generation, c.Holder.WriterBox = 2, "target-gen", "target"
if mode == "source-cut" || mode == "covered-source" || mode == "no-source" {
c.Holder.WriterBox = "box"
diff --git a/internal/hostclaim/lost.go b/internal/hostclaim/lost.go
index ebe3eb993..f5cc2c82f 100644
--- a/internal/hostclaim/lost.go
+++ b/internal/hostclaim/lost.go
@@ -9,6 +9,7 @@ import (
"fmt"
"io"
"reflect"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/release"
@@ -241,6 +242,11 @@ func declareLostTransition(in LostDeclaration, owner LossOperationOwner) func(*C
if lossRefPresent(c, in.LossID) {
return ErrLossAlreadyApplied
}
+ for key := range c.Op.Journal {
+ if strings.HasPrefix(key, "loss.") && !(destroyedLocalOnly(in.Consequences) && stringPresent(c.Lost, in.Box)) {
+ return ErrWrongState
+ }
+ }
copy, err := cloneClaim(c)
if err != nil {
return err
@@ -298,6 +304,11 @@ func declareLostTransition(in LostDeclaration, owner LossOperationOwner) func(*C
// Recovery intent is recorded before any authority or chore effects in
// this transition. Update makes the entire sequence one claim CAS.
c.Op.Journal[journalKey] = intent
+ gate, err := json.Marshal(operationLossGate{Box: in.Box, RevokeExecution: in.Consequences.RevokeExecution})
+ if err != nil {
+ return err
+ }
+ c.Op.Journal["loss_gate."+in.LossID] = string(gate)
if isExec {
c.Op.Exec = "none"
diff --git a/internal/hostclaim/lost_test.go b/internal/hostclaim/lost_test.go
index b972411d9..8558def7e 100644
--- a/internal/hostclaim/lost_test.go
+++ b/internal/hostclaim/lost_test.go
@@ -238,6 +238,10 @@ func TestDeclareLostNewerRevisionAppliesSecondConsequences(t *testing.T) {
if len(stopped.Claim.Obligations) != 1 {
t.Fatal("compute stop discharged local storage")
}
+ stopped, err = Update(ctx, s, stopped, func(c *Claim) error { c.Op.ID = "delete-second"; c.Op.Journal = nil; return nil })
+ if err != nil {
+ t.Fatal(err)
+ }
second := lossForObservation(t, stopped, "destroyed", 10, "box-a", LossConsequences{
DischargeLocal: []ObligationRef{{ID: "local", Op: "delete", Retirement: "ret"}},
DestroyedStorage: []string{"delete/local/ret"}, RecoveryIntent: []byte(`{"kind":"destroyed-storage"}`),
diff --git a/internal/hostclaim/move_target_fence.go b/internal/hostclaim/move_target_fence.go
index 33e14e51d..514943b72 100644
--- a/internal/hostclaim/move_target_fence.go
+++ b/internal/hostclaim/move_target_fence.go
@@ -3,6 +3,7 @@ package hostclaim
import (
"context"
"encoding/json"
+ "github.com/carlosframework/platform/internal/hostops/fencerelease"
"github.com/carlosframework/platform/internal/release"
"time"
)
@@ -38,6 +39,7 @@ func VerifyMoveTargetFence(ctx context.Context, s release.Store, o Observed, ref
// FencedTargetRelease has the same scopes as loss release, with verified fence authority.
type FencedTargetRelease struct {
+ Token fencerelease.Token
OpID, RollbackID, Target, TargetRegion, SourceRegion string
Location, Lane ParkScope
ReturnLane *ParkScope
@@ -47,6 +49,14 @@ type FencedTargetRelease struct {
// ReleaseFencedTarget declares and consumes the fence in one CAS, including all parks.
func ReleaseFencedTarget(r FencedTargetRelease) func(*Claim) error {
+ return func(c *Claim) error {
+ if c == nil || c.Holder == nil || !r.Token.Consume(c.Host, c.Holder.Incarnation, c.Gen, r.OpID, r.RollbackID, r.Target) {
+ return ErrFence
+ }
+ return releaseFencedTarget(r)(c)
+ }
+}
+func releaseFencedTarget(r FencedTargetRelease) func(*Claim) error {
return func(c *Claim) error {
if err := RequireOp(r.OpID)(c); err != nil {
return err
@@ -86,6 +96,12 @@ func ReleaseFencedTarget(r FencedTargetRelease) func(*Claim) error {
break
}
}
+ owned := map[ObligationRef]bool{}
+ for _, ob := range c.Obligations {
+ if ob.Op == r.OpID && (ob.Kind == "park-generation" || ob.Kind == "park-replica") && ob.Owner == "box:"+r.Target {
+ owned[ObligationRef{ID: ob.ID, Op: ob.Op, Retirement: ob.Retirement}] = true
+ }
+ }
return inOneWrite(c, func(c *Claim) error {
setJournal(c, "fence."+r.Target, f.ref)
c.Op.Exec = "none"
@@ -94,7 +110,7 @@ func ReleaseFencedTarget(r FencedTargetRelease) func(*Claim) error {
}, func(c *Claim) error {
for i := range c.Obligations {
o := &c.Obligations[i]
- if o.Op == r.OpID && (o.Kind == "park-generation" || o.Kind == "park-replica") && (o.Region == r.TargetRegion || o.Owner == "box:"+r.Target) {
+ if owned[ObligationRef{ID: o.ID, Op: o.Op, Retirement: o.Retirement}] {
o.Owner = "region:" + o.Region
loss.Reassigned = append(loss.Reassigned, ObligationRef{ID: o.ID, Op: o.Op, Retirement: o.Retirement})
}
diff --git a/internal/hostclaim/operation_loss.go b/internal/hostclaim/operation_loss.go
index 6962ec13d..07cbbde80 100644
--- a/internal/hostclaim/operation_loss.go
+++ b/internal/hostclaim/operation_loss.go
@@ -1,10 +1,16 @@
package hostclaim
import (
+ "encoding/json"
"strconv"
"strings"
)
+type operationLossGate struct {
+ Box string `json:"box"`
+ RevokeExecution bool `json:"revoke_execution"`
+}
+
// OperationLossID selects only a declaration journalled under this operation.
// Old claim-level LossRefs alone do not turn a later move into a loss recovery.
func OperationLossID(c *Claim) (string, error) {
@@ -17,7 +23,34 @@ func OperationLossID(c *Claim) (string, error) {
continue
}
next := strings.TrimPrefix(key, "loss.")
- if c.Op.ID == "" || !safeLossID(next) || !lossRefPresent(c, next) || id != "" {
+ if c.Op.ID == "" || !safeLossID(next) || !lossRefPresent(c, next) {
+ return "", ErrLossRecord
+ }
+ var gate operationLossGate
+ if raw := c.Op.Journal["loss_gate."+next]; raw != "" {
+ if json.Unmarshal([]byte(raw), &gate) != nil || gate.Box == "" {
+ return "", ErrLossRecord
+ }
+ } else {
+ var legacy struct {
+ Kind, Target string
+ RevokeExecution bool `json:"revoke_execution"`
+ }
+ if json.Unmarshal([]byte(c.Op.Journal[key]), &legacy) != nil {
+ return "", ErrLossRecord
+ }
+ gate.Box, gate.RevokeExecution = legacy.Target, legacy.RevokeExecution
+ if gate.Box == "" && legacy.Kind == "move-target-lost" {
+ gate.Box = c.Op.Intent["target"]
+ }
+ if gate.Box == "" && !gate.RevokeExecution {
+ return "", ErrLossRecord
+ }
+ }
+ if gate.Box != c.Op.Intent["target"] && !gate.RevokeExecution {
+ continue
+ }
+ if id != "" {
return "", ErrLossRecord
}
id = next
diff --git a/internal/hostclaim/plan12_fence_test.go b/internal/hostclaim/plan12_fence_test.go
index 51b3955f0..7ef859d49 100644
--- a/internal/hostclaim/plan12_fence_test.go
+++ b/internal/hostclaim/plan12_fence_test.go
@@ -10,12 +10,18 @@ import (
func TestPlan12TargetFenceRelease(t *testing.T) {
for _, kind := range []string{KindLanding, KindRestoring} {
- for _, mode := range []string{"ok", "wrong-op", "wrong-target", "wrong-enrollment", "wrong-region", "unverified", "zero-proof", "stale", "committed", "partial-return"} {
+ for _, mode := range []string{"ok", "wrong-op", "wrong-target", "wrong-enrollment", "wrong-region", "unverified", "zero-proof", "stale", "committed", "partial-return", "parks", "no-grant"} {
t.Run(kind+"/"+mode, func(t *testing.T) {
s, o, ref, a := c7Fixture(t)
var err error
o, err = Update(t.Context(), s, o, func(c *Claim) error {
c.Obligations = nil
+ if mode == "parks" {
+ for _, owner := range []string{"box:adi-knitting", "box:other", "region:ie"} {
+ id := "existing-" + strconv.Itoa(len(c.Obligations))
+ c.Obligations = append(c.Obligations, Obligation{ID: id, Op: c.Op.ID, Kind: "park-generation", Generation: id, Region: "ie", Owner: owner, Retirement: id})
+ }
+ }
c.Op.Kind, c.Op.Phase, c.Op.Exec = kind, PhaseProbing, "adi-knitting"
c.Op.Participants = []string{"bri-knitting", "adi-knitting"}
c.Op.Intent["target"], c.Op.Intent["source"], c.Op.Intent["to"], c.Op.Intent["target_enrollment"] = "adi-knitting", "bri-knitting", "ie", ref.Enrollment
@@ -70,9 +76,13 @@ func TestPlan12TargetFenceRelease(t *testing.T) {
s.puts = 0
var got Observed
if err == nil {
- got, err = Update(t.Context(), s, o, ReleaseFencedTarget(r))
+ transition := releaseFencedTarget(r)
+ if mode == "no-grant" {
+ transition = ReleaseFencedTarget(r)
+ }
+ got, err = Update(t.Context(), s, o, transition)
}
- if mode != "ok" && mode != "partial-return" {
+ if mode != "ok" && mode != "partial-return" && mode != "parks" {
if err == nil {
t.Fatal("accepted", mode)
}
@@ -87,8 +97,20 @@ func TestPlan12TargetFenceRelease(t *testing.T) {
if s.puts != 1 || got.Claim.Op.Phase != PhaseReleased || got.Claim.Op.Exec != "none" || stringPresent(got.Claim.Op.Participants, r.Target) || len(got.Claim.Obligations) < 2 {
t.Fatal(got.Claim)
}
+ if mode == "parks" {
+ gotLoss := got.Claim.EvidenceRefs[len(got.Claim.EvidenceRefs)-1]
+ want := []ObligationRef{{ID: "existing-0", Op: o.Claim.Op.ID, Retirement: "existing-0"}}
+ if !reflect.DeepEqual(gotLoss.Reassigned, want) {
+ t.Fatalf("reassigned %+v", gotLoss.Reassigned)
+ }
+ for _, ob := range got.Claim.Obligations {
+ if ob.ID == "existing-1" && ob.Owner != "box:other" {
+ t.Fatal("stole another box's park")
+ }
+ }
+ }
clone, _ := cloneClaim(got.Claim)
- if err := ReleaseFencedTarget(r)(clone); err != nil || !reflect.DeepEqual(clone, got.Claim) {
+ if err := releaseFencedTarget(r)(clone); err != nil || !reflect.DeepEqual(clone, got.Claim) {
t.Fatal("replay", err)
}
if err := ReturnToSource(r.OpID, "bri-knitting", false, ReturnFence{})(clone); err != nil {
diff --git a/internal/hostclaim/plan12_fix1_loss_test.go b/internal/hostclaim/plan12_fix1_loss_test.go
new file mode 100644
index 000000000..9fecb59e9
--- /dev/null
+++ b/internal/hostclaim/plan12_fix1_loss_test.go
@@ -0,0 +1,132 @@
+package hostclaim
+
+import (
+ "crypto/ed25519"
+ "errors"
+ "reflect"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestPlan12ParticipantLossKeepsExecutorAdmission(t *testing.T) {
+ for _, target := range []bool{false, true} {
+ t.Run(map[bool]string{false: "participant", true: "target"}[target], func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ pub, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o := servingClaim(t, t.Context(), s)
+ o, err = Update(t.Context(), s, o, func(c *Claim) error {
+ c.Op = Op{ID: "move", Kind: KindLanding, Phase: PhaseProbing, Exec: "box-a", Participants: []string{"box-a", "box-b"}, Intent: map[string]string{"target": "box-a"}}
+ if target {
+ c.Op.Intent["target"] = "box-b"
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := lossForObservation(t, o, "participant", 1, "box-b", LossConsequences{RemoveParticipation: true, RecoveryIntent: []byte(`{"kind":"participant-lost"}`)})
+ if _, err = CreateLossRecord(t.Context(), s, release.RootSigner(priv), r); err != nil {
+ t.Fatal(err)
+ }
+ o, err = DeclareLost(t.Context(), s, pub, o, r.ID, testLossOwner{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ h := o.Claim.Holder
+ _, err = Update(t.Context(), s, o, AdmitWriter(Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: h.Session, ObservedGeneration: h.Generation, Box: "box-a", Mode: "cold", NewGeneration: "healthy-next", Region: "ie"}))
+ if target && !errors.Is(err, ErrLossRecord) || !target && err != nil {
+ t.Fatalf("admission: %v", err)
+ }
+ })
+ }
+}
+
+func TestPlan12DeclareLostRefusesSecondSameOperation(t *testing.T) {
+ for _, which := range []string{"different-box", "new-loss-id", "revoke-execution", "local-only-different-box"} {
+ t.Run(which, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ pub, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ o := servingClaim(t, t.Context(), s)
+ o, err = Update(t.Context(), s, o, func(c *Claim) error {
+ c.Op = Op{ID: "move", Kind: KindLanding, Phase: PhaseProbing, Exec: "box-a", Participants: []string{"box-a", "box-b", "box-c"}}
+ c.Obligations = []Obligation{{ID: "local", Op: "move", Kind: "park-local", Box: "box-c", Retirement: "ret", Owner: "box:box-c"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ first := lossForObservation(t, o, "first", 1, "box-b", LossConsequences{RemoveParticipation: true, RecoveryIntent: []byte(`{}`)})
+ if _, err = CreateLossRecord(t.Context(), s, release.RootSigner(priv), first); err != nil {
+ t.Fatal(err)
+ }
+ o, err = DeclareLost(t.Context(), s, pub, o, first.ID, testLossOwner{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ box := "box-c"
+ effects := LossConsequences{RemoveParticipation: true, RecoveryIntent: []byte(`{}`)}
+ switch which {
+ case "new-loss-id":
+ box, effects.RemoveParticipation = "box-b", false
+ case "revoke-execution":
+ box, effects.RevokeExecution = "box-a", true
+ case "local-only-different-box":
+ effects = LossConsequences{
+ DischargeLocal: []ObligationRef{{ID: "local", Op: "move", Retirement: "ret"}},
+ DestroyedStorage: []string{"move/local/ret"}, RecoveryIntent: []byte(`{"kind":"destroyed-local-only"}`),
+ }
+ }
+ second := lossForObservation(t, o, "second", 2, box, effects)
+ if _, err = CreateLossRecord(t.Context(), s, release.RootSigner(priv), second); err != nil {
+ t.Fatal(err)
+ }
+ before := make(map[string]ObservedLossRecord)
+ for _, id := range []string{first.ID, second.ID} {
+ r, err := ReadLossRecord(t.Context(), s, pub, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ before[id] = r
+ }
+ if _, err = DeclareLost(t.Context(), s, pub, o, second.ID, testLossOwner{}); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("second declaration: %v", err)
+ }
+ fresh, err := Read(t.Context(), s, o.Host)
+ if err != nil || fresh.ETag != o.ETag || !reflect.DeepEqual(fresh.Claim, o.Claim) {
+ t.Fatal("refusal mutated claim or obligations", err)
+ }
+ for id, prior := range before {
+ after, err := ReadLossRecord(t.Context(), s, pub, id)
+ if err != nil || !reflect.DeepEqual(prior, after) {
+ t.Fatal("refusal mutated loss record", id, err)
+ }
+ }
+ })
+ }
+}
+
+func TestPlan12OperationLossGateIgnoresOtherParticipants(t *testing.T) {
+ c := &Claim{Op: Op{ID: "move", Intent: map[string]string{"target": "target"}, Journal: map[string]string{"loss.target": "{}", "loss.participant": "{}", "loss_gate.target": `{"box":"target"}`, "loss_gate.participant": `{"box":"participant"}`}}, LossRefs: []string{"target", "participant"}}
+ for i := 0; i < 100; i++ {
+ id, err := OperationLossID(c)
+ if err != nil || id != "target" {
+ t.Fatalf("loss selection %q %v", id, err)
+ }
+ }
+}
+
+func TestPlan12OperationLossGateIncludesRevokedExecutor(t *testing.T) {
+ c := &Claim{Op: Op{ID: "move", Intent: map[string]string{"target": "target"}, Journal: map[string]string{
+ "loss.executor": "{}", "loss_gate.executor": `{"box":"source","revoke_execution":true}`,
+ }}, LossRefs: []string{"executor"}}
+ if id, err := OperationLossID(c); err != nil || id != "executor" {
+ t.Fatalf("revoked executor loss selection %q %v", id, err)
+ }
+}
diff --git a/internal/hostclaim/writer_identity.go b/internal/hostclaim/writer_identity.go
new file mode 100644
index 000000000..deb48683c
--- /dev/null
+++ b/internal/hostclaim/writer_identity.go
@@ -0,0 +1,11 @@
+package hostclaim
+
+// SameWriterIdentity ignores claim revisions belonging to unrelated chores.
+func SameWriterIdentity(a, b Observed) bool {
+ if a.Absent || b.Absent || a.Host != b.Host || a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil {
+ return false
+ }
+ x, y := a.Claim.Holder, b.Claim.Holder
+ p, q := a.Claim.Op, b.Claim.Op
+ return x.Incarnation == y.Incarnation && x.Stamp == y.Stamp && x.Session == y.Session && x.Generation == y.Generation && x.WriterBox == y.WriterBox && p.ID == q.ID && p.Kind == q.Kind && p.Phase == q.Phase && p.Exec == q.Exec
+}
diff --git a/internal/hostguard/writerstop.go b/internal/hostguard/writerstop.go
index d64fb558f..86d0c32ec 100644
--- a/internal/hostguard/writerstop.go
+++ b/internal/hostguard/writerstop.go
@@ -118,6 +118,19 @@ func RecoveryWriter(c *hostclaim.Claim, box string) (hostclaim.AdmissionRecord,
return operationWriter(c, box, c.Op.Kind)
}
func RecoveryWriterStop(c *hostclaim.Claim, box, region string) (string, bool) {
+ if c != nil && c.Holder != nil && region != "" && box != "" && (c.Op.Kind == hostclaim.KindRestoring || c.Op.Kind == hostclaim.KindReverting) && c.Op.ID != "" && c.Op.Phase == hostclaim.PhaseProbing && c.Op.Exec == box && c.Op.Intent["source"] == box {
+ h := c.Holder
+ stamp, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
+ if err == nil && stamp > 0 && h.Stamp == stamp && h.Session == 0 && h.Generation == "" && h.WriterBox == "" && len(h.Admissions) == 0 && c.Op.Journal[hostclaim.JournalCutSession] == "0" && c.Op.Journal[hostclaim.JournalSrcGen] == "" {
+ id := WriterStopObligation(0)
+ for _, ob := range c.Obligations {
+ if ob.ID == id {
+ return "", false
+ }
+ }
+ return id, true
+ }
+ }
if _, ok := RecoveryWriter(c, box); !ok {
return "", false
}
diff --git a/internal/hostops/fencerelease/token.go b/internal/hostops/fencerelease/token.go
new file mode 100644
index 000000000..2f988bf71
--- /dev/null
+++ b/internal/hostops/fencerelease/token.go
@@ -0,0 +1,7 @@
+package fencerelease
+
+import private "github.com/carlosframework/platform/internal/hostops/internal/fencerelease"
+
+// Token exposes the value type without exposing its constructor.
+// Go's internal import rule reserves construction to hostops.
+type Token = private.Token
diff --git a/internal/hostops/internal/fencerelease/token.go b/internal/hostops/internal/fencerelease/token.go
new file mode 100644
index 000000000..0d62ede33
--- /dev/null
+++ b/internal/hostops/internal/fencerelease/token.go
@@ -0,0 +1,27 @@
+package fencerelease
+
+import "sync/atomic"
+
+type token struct {
+ host, incarnation, op, rollback, target string
+ gen uint64
+ live *atomic.Bool
+}
+
+type Token = token
+
+// New is inside hostops' import boundary. The recovery grant calls it only
+// after checking the record, sequence, freeze and exact claim observation.
+func New(host, incarnation string, gen uint64, op, rollback, target string) Token {
+ live := new(atomic.Bool)
+ live.Store(true)
+ return token{host: host, incarnation: incarnation, gen: gen, op: op, rollback: rollback, target: target, live: live}
+}
+func (t token) Consume(host, incarnation string, gen uint64, op, rollback, target string) bool {
+ return t.live != nil && t.host == host && t.incarnation == incarnation && t.gen == gen && t.op == op && t.rollback == rollback && t.target == target && t.live.CompareAndSwap(true, false)
+}
+func (t token) Revoke() {
+ if t.live != nil {
+ t.live.Store(false)
+ }
+}
diff --git a/internal/hostops/internal/fencerelease/token_test.go b/internal/hostops/internal/fencerelease/token_test.go
new file mode 100644
index 000000000..9964aea62
--- /dev/null
+++ b/internal/hostops/internal/fencerelease/token_test.go
@@ -0,0 +1,37 @@
+package fencerelease
+
+import "testing"
+
+func TestTokenScopeAndSingleUse(t *testing.T) {
+ for _, mode := range []string{"ok", "host", "incarnation", "generation", "operation", "rollback", "target", "zero", "revoked"} {
+ t.Run(mode, func(t *testing.T) {
+ token := New("host", "inc", 7, "op", "rb", "target")
+ host, inc, op, rb, target := "host", "inc", "op", "rb", "target"
+ gen := uint64(7)
+ switch mode {
+ case "host":
+ host = "other"
+ case "incarnation":
+ inc = "other"
+ case "generation":
+ gen++
+ case "operation":
+ op = "other"
+ case "rollback":
+ rb = "other"
+ case "target":
+ target = "other"
+ case "zero":
+ token = Token{}
+ case "revoked":
+ token.Revoke()
+ }
+ if token.Consume(host, inc, gen, op, rb, target) != (mode == "ok") {
+ t.Fatal("wrong token scope")
+ }
+ if mode == "ok" && token.Consume(host, inc, gen, op, rb, target) {
+ t.Fatal("reused token")
+ }
+ })
+ }
+}
diff --git a/internal/hostops/move_fence_release.go b/internal/hostops/move_fence_release.go
index 5935b032d..939bf2c21 100644
--- a/internal/hostops/move_fence_release.go
+++ b/internal/hostops/move_fence_release.go
@@ -5,6 +5,7 @@ import (
"crypto/rand"
"encoding/json"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostops/internal/fencerelease"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/release"
@@ -118,6 +119,8 @@ func (g *MoveRecoveryGrant) ReleaseFencedTarget(ctx context.Context, ref hostcla
if err := moveAdmission(ctx, g.w.raw, o.Host, c); err != nil {
return err
}
+ r.Token = fencerelease.New(c.Host, c.Holder.Incarnation, c.Gen, r.OpID, r.RollbackID, r.Target)
+ defer r.Token.Revoke()
return hostclaim.ReleaseFencedTarget(r)(c)
})
return err
diff --git a/internal/hostops/move_local_loss_test.go b/internal/hostops/move_local_loss_test.go
index 66d4d8824..4dd837b46 100644
--- a/internal/hostops/move_local_loss_test.go
+++ b/internal/hostops/move_local_loss_test.go
@@ -108,6 +108,37 @@ func TestJ4DestroyedLocalOnlyAlreadyDeclaredTarget(t *testing.T) {
}
}
+func TestJ4DestroyedLocalOnlyKeepsOperationLossGate(t *testing.T) {
+ w := newLossWorld(t, "landing", func(c *hostclaim.Claim, _ *provision.Record, _ *AppAdmission) {
+ c.Inventory = map[string]hostclaim.InventoryEntry{"uk-box": {Paths: map[string]uint64{"root": 21}}}
+ })
+ first := w.inspect(t, "uk-label")
+ before, err := w.declare(w.submission(first))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if id, err := hostclaim.OperationLossID(before.Claim); err != nil || id != first.Details.ID {
+ t.Fatalf("target loss gate = %q, %v", id, err)
+ }
+ local := w.inspect(t, "uk-label")
+ sub := w.submission(local)
+ for _, step := range []string{"discharge", "replay"} {
+ got, err := w.declare(sub)
+ if err != nil {
+ t.Fatal(step, err)
+ }
+ if id, err := hostclaim.OperationLossID(got.Claim); err != nil || id != first.Details.ID {
+ t.Fatalf("%s changed target loss gate: %q, %v", step, id, err)
+ }
+ if !reflect.DeepEqual(before.Claim.Lost, got.Claim.Lost) || !reflect.DeepEqual(before.Claim.Op, got.Claim.Op) {
+ t.Fatal(step, "added loss or changed operation")
+ }
+ if _, exists := got.Claim.Op.Journal["loss."+local.Details.ID]; exists {
+ t.Fatal("local-only declaration added recovery loss")
+ }
+ }
+}
+
func TestJ4DestroyedLocalOnlyRequiresExactAttestation(t *testing.T) {
for _, which := range []string{"missing", "wrong-retirement", "terminated-only", "stale-claim", "stale-record"} {
t.Run(which, func(t *testing.T) {
diff --git a/internal/hostops/plan12_fence_test.go b/internal/hostops/plan12_fence_test.go
index 60cd19933..145040314 100644
--- a/internal/hostops/plan12_fence_test.go
+++ b/internal/hostops/plan12_fence_test.go
@@ -5,6 +5,8 @@ import (
"crypto/ed25519"
"encoding/json"
"errors"
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/moves"
"github.com/carlosframework/platform/internal/provision"
@@ -200,6 +202,9 @@ func TestPlan12RecoveryFenceGrant(t *testing.T) {
if err != nil || replayTag != stampTag || w.claim(t).ETag != claimTag {
t.Fatal("parked replay wrote state", err)
}
+ if mode == "admitted" {
+ plan12GrantLossMaximum(t, w)
+ }
// Source opening is owned by the source; seed its completed evidence.
current = w.claim(t)
if _, err := hostclaim.Update(t.Context(), w.raw, current, func(c *hostclaim.Claim) error {
@@ -260,3 +265,40 @@ func TestPlan12RecoveryFenceGrant(t *testing.T) {
})
}
}
+
+func plan12GrantLossMaximum(t *testing.T, w *c7World) {
+ t.Helper()
+ o := w.claim(t)
+ source := o.Claim.Op.Intent["source"]
+ o, err := hostclaim.Update(t.Context(), w.raw, o, hostclaim.ReturnToSource(o.Claim.Op.ID, source, false, hostclaim.ReturnFence{}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = hostclaim.Update(t.Context(), w.raw, o, hostclaim.SetRestoringPhase(o.Claim.Op.ID, hostclaim.PhaseClaimed, hostclaim.PhaseProbing))
+ if err != nil {
+ t.Fatal(err)
+ }
+ in := hostclaim.Admission{ObservedOpID: o.Claim.Op.ID, ObservedPhase: o.Claim.Op.Phase, ObservedSession: o.Claim.Holder.Session, ObservedGeneration: o.Claim.Holder.Generation, Box: source, Mode: "cold", NewGeneration: "recovered", Region: "uk", PredecessorRegion: "ie", RemoteMaxTXID: 3}
+ checked := false
+ writer := hostbox.Writer{Claims: w.raw, Box: source, Region: "uk", AdmissionEvidence: func(_ context.Context, _ hostclaim.Observed, a hostclaim.Admission, p *genselect.Plan) error {
+ checked = true
+ if a.RemoteMaxTXID != 7 || p.MaxTXID != 3 {
+ t.Fatalf("loss=%d restore=%d", a.RemoteMaxTXID, p.MaxTXID)
+ }
+ return nil
+ }, Loss: &hostbox.LossAdmission{Sources: func(_ context.Context, _ hostclaim.Observed, l hostclaim.LossRecord, _ hostclaim.Admission, _ *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) {
+ if len(l.WriteSources) != 1 {
+ t.Fatalf("sources %+v", l.WriteSources)
+ }
+ s := l.WriteSources[0]
+ return []hostclaim.ValidatedWriteSource{{Reason: s.Reason, Box: s.Box, Session: s.Session, Generation: s.Generation, Region: s.Region, Baseline: s.Baseline, RemoteMaxTXID: 7}}, nil
+ }}}
+ admitted, err := writer.Admit(t.Context(), o, in, &genselect.Plan{Generation: "src", MaxTXID: 3})
+ if err != nil {
+ t.Fatal(err)
+ }
+ records := admitted.Claim.Holder.Admissions
+ if !checked || records[len(records)-1].WritesLost.RemoteMaxTXID != 7 {
+ t.Fatal("fenced maximum lost")
+ }
+}
diff --git a/internal/hostops/plan12_fix1_recovery_test.go b/internal/hostops/plan12_fix1_recovery_test.go
new file mode 100644
index 000000000..5efa84f7f
--- /dev/null
+++ b/internal/hostops/plan12_fix1_recovery_test.go
@@ -0,0 +1,70 @@
+package hostops
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "github.com/carlosframework/platform/internal/alerting"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "strings"
+ "testing"
+ "time"
+)
+
+func TestPlan12ReportingFailureStillRunsRecovery(t *testing.T) {
+ f := newResolveFixture(t, true)
+ o, err := hostclaim.Update(t.Context(), f.raw, f.moving, func(c *hostclaim.Claim) error {
+ ref := hostclaim.FenceRef{Enrollment: strings.Repeat("a", 32), SHA256: strings.Repeat("b", 64), Key: "control/box-fences/" + strings.Repeat("a", 32) + ".json"}
+ sum := sha256.Sum256([]byte(c.Op.ID + "\x00box-b\x00" + ref.Enrollment + "\x00" + ref.SHA256))
+ id := "auto-" + hex.EncodeToString(sum[:16])
+ c.LossRefs = []string{id}
+ c.EvidenceRefs = []hostclaim.AutomaticLoss{{ID: id, Box: "box-b", Enrollment: ref.Enrollment, Fence: ref, ObservedAt: f.now.Format(time.RFC3339Nano), Op: c.Op.ID, Kind: hostclaim.KindLanding, Phase: hostclaim.PhaseProbing, Role: "target", Incarnation: c.Holder.Incarnation, Stamp: c.Holder.Stamp, Timers: hostclaim.TimersUnknown}}
+ return nil
+ })
+ // The raw fixture below supplies a retained evidence row; validation of the
+ // row belongs to ReportAutomaticLosses, whose failure is the point here.
+ if err != nil {
+ _ = o
+ t.Fatal(err)
+ }
+ noWork := func(context.Context) error { return nil }
+ wrong := func(context.Context, hostclaim.Observed) error { t.Fatal("generic recovery"); return nil }
+ failure := errors.New("loss signer unavailable")
+ calls := 0
+ pages := 0
+ alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, _ []byte) error { pages++; return nil }})
+ t.Cleanup(func() { alerting.Arm(nil) })
+ r := Recovery{Controller: Controller{Store: f.raw, Control: f.w, Authority: &f.authority, Now: func() time.Time { return f.now }, NewID: func() string { return "recovery" }}, Purge: wrong, Placement: wrong, SweepRecords: noWork, SweepPurges: noWork, SweepAdmissions: noWork}
+ r.ReportAutomaticLosses = func(ctx context.Context, host string) error {
+ before, err := hostclaim.Read(ctx, f.raw, host)
+ if err != nil {
+ return err
+ }
+ _, err = hostclaim.Update(ctx, f.raw, before, func(c *hostclaim.Claim) error {
+ if c.Op.Journal == nil {
+ c.Op.Journal = map[string]string{}
+ }
+ c.Op.Journal["report-attempt"] = "failed"
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+ return failure
+ }
+ r.Moves = func(ctx context.Context, g *MoveRecoveryGrant) error {
+ calls++
+ fresh, err := hostclaim.Read(ctx, f.raw, g.Observation().Host)
+ if err != nil {
+ return err
+ }
+ if fresh.ETag != g.Observation().ClaimETag || fresh.Claim.Op.Journal["report-attempt"] != "failed" {
+ t.Fatal("recovery used observation before report")
+ }
+ return nil
+ }
+ if err := f.w.RunRecovery(t.Context(), r.Pass); !errors.Is(err, failure) || calls != 1 || pages != 1 {
+ t.Fatalf("pass=%v recovery calls=%d", err, calls)
+ }
+}
diff --git a/internal/hostops/recovery.go b/internal/hostops/recovery.go
index 11a5cc2e1..32b6735ac 100644
--- a/internal/hostops/recovery.go
+++ b/internal/hostops/recovery.go
@@ -3,6 +3,8 @@ package hostops
import (
"context"
"errors"
+ "fmt"
+ "github.com/carlosframework/platform/internal/alerting"
"os"
"strings"
"time"
@@ -80,7 +82,9 @@ func (r Recovery) Pass(ctx context.Context) error {
if r.ReportAutomaticLosses != nil && len(obs.Claim.EvidenceRefs) > 0 {
if err := r.ReportAutomaticLosses(ctx, host); err != nil {
all = append(all, err)
- continue
+ if pageErr := alerting.Page(ctx, alerting.LevelWarning, hostclaim.ErrLossRecord.Error(), fmt.Sprintf("%s: %v", host, err)); pageErr != nil {
+ all = append(all, pageErr)
+ }
}
obs, err = hostclaim.Read(ctx, c.Store, host)
if err != nil {
@@ -88,6 +92,9 @@ func (r Recovery) Pass(ctx context.Context) error {
continue
}
}
+ if obs.Absent || obs.Claim == nil {
+ continue
+ }
op := obs.Claim.Op
// Zero time is unknown age, never old enough. Freeing empty parking is safe
// independently of age; this includes Plan 4's holderless sweep retries.
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index 8eabf28df..6ab70bf84 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -150,6 +150,7 @@ func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, lega
d.Alerts = append(d.Alerts, Alert{c.Holder.Location, "marker does not match the claim or an unfinished admission run"})
}
groups := map[Candidate]bool{}
+ refused := map[string]bool{}
seen := map[Ref]bool{}
for _, x := range versions {
loc, err := locationOf(c.Host, x.Key)
@@ -161,7 +162,10 @@ func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, lega
continue
}
if refusal := v.RefusedLocations[loc]; refusal != nil && !(c.Holder != nil && loc == c.Holder.Location && v.Region == v.AuthoritativeRegion) {
- d.Alerts = append(d.Alerts, Alert{loc, refusal.Error()})
+ if !refused[loc] {
+ d.Alerts = append(d.Alerts, Alert{loc, refusal.Error()})
+ refused[loc] = true
+ }
continue
}
candidate := Candidate{Location: loc}
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go
index 601b60f26..37fb725c9 100644
--- a/internal/provision/plan9_b8_census_test.go
+++ b/internal/provision/plan9_b8_census_test.go
@@ -474,12 +474,11 @@ func Reviewed(ctx context.Context, s Store, k string) error {
var plan9B8CensusInventory = []censusEntry{
{"internal/hostbox/parkleases.go", "ReplicaParkLeaseSet", "key-site", "lit:instances", "0ca8d9b743402fda9d15cc215464423b35e21c1c2b47f91294d76188b7d43fb6"},
{"internal/host/claimretirementcensus.go", "ClaimRetirementCensus", "key-site", "lit:instances", "abc0bd8c5f5d328abfac8821f42f8f23954723ed6506747f33b0197824e2611f"},
- {"internal/host/claimretirementcensus.go", "productionClaimRegional.retirement", "key-site", "lit:instances", "19306389daacc704a94cfb434b7d557ced96abe09958a01be38a0707b3ea0a6f"},
{"internal/host/claimproduction.go", "productionClaimRegional.View", "key-site", "lit:instances", "61ff9ff47054b6a96cb24467c7aaf41c112b7c5129987af5ce27e0b7f862348e"},
{"internal/hostops/move_loss_admission.go", "moveLossOwner.ValidateLossRecovery", "key-site", "key:RecordKey", "ea2aa35a069ffdc846ecb208c33940245fc865795188818e86ea7a93135dd6a9"},
{"internal/host/claimretirementcensus.go", "claimArchivedLocations", "key-site", "lit:instances", "9ba8ea4ecb5ad63621c1322467b9a80f8f5703896ba1fc3f7cb1addd56418950"},
{"internal/host/claimretirementcensus.go", "claimLegacyIncarnation", "key-site", "lit:instances", "31a14ea105c67e00a24f06b9500a221ee5bfc47402b9c33cecfde807e51db530"},
- {"internal/hostops/move_fence_release.go", "MoveRecoveryGrant.ReleaseFencedTarget", "record-writer", "key:WriteRecord,key:writeRecord", "9af3294f6fcd6c00266eca06a5002b5dd904342938aeae7da155d025f23677d7"},
+ {"internal/hostops/move_fence_release.go", "MoveRecoveryGrant.ReleaseFencedTarget", "record-writer", "key:WriteRecord,key:writeRecord", "031a3a8169a92775bff8a1a1b08cfe82a5a613156a1bce45dc9cfcb34c56befe"},
{"internal/hostops/move_fence_release.go", "MoveRecoveryGrant.StampLostMoveTargetParked", "record-writer", "key:WriteRecord", "b082db5634a2f4d9a5b45cc4a9cb725964e9bb24b7aeb53a6346167d7737227c"},
{"internal/hostops/move_recovery_grant.go", "MoveRecoveryGrant.Convert", "record-writer", "key:WriteRecord", "794963a24a0a50d980505f83ac912973a10267b21f74bd41c56584ab9b0b25a1"},
// The recovery pass's move log writer: one create-only entry of a move's
@@ -849,7 +848,7 @@ var plan9B8CensusInventory = []censusEntry{
{"internal/ops/litestream.go", "FleetDBsWithSkipped", "key-site", "lit:instances", "f62291df0685db15529c4946dca358a29aba66fc31e81aab299de579e37f60f1"},
{"internal/park/remote.go", "locationOf", "key-site", "lit:instances", "83b3efca31b3faadcd5997dc181d81b7bb8fc866cc4a448683bf8423b1bea866"},
{"internal/park/remote.go", "Engine.run", "versioned-transport", "sel:CopyVersion,sel:PutIfAbsent", "3de5b215e4b7aba67768243891b3f6d8ff8c77a465aefe5168c789cb012b1ea9"},
- {"internal/park/sweep.go", "Decide", "key-site", "lit:instances", "334144f96a8b19448790578370923c290b1ef50142e8ffc7a7f5933281f7e057"},
+ {"internal/park/sweep.go", "Decide", "key-site", "lit:instances", "d03c3af1430816d3f9517a3944e553291805f97a317e54b96a02b9fbfbb5efb5"},
{"internal/park/sweep.go", "Engine.Discover", "key-site", "lit:instances", "69b83e6e859c1c9809ba76d46cb0192888df1ecdcd161b4f2b6435733eecdfa3"},
{"internal/provision/claim_app_writes.go", "UpdateRecord", "record-writer", "key:UpdateRecord,key:claimRawUpdateRecord", "caf714cb78e1280d489c7ad5b3cefb1420e812d541108b1fe13d733abed97229"},
{"internal/provision/provision.go", "ManifestKey", "key-site", "lit:instances", "b8ee0e77bdfade19af79a4012c411c1eb0b6ddca16037e5536fa8597f7cde270"},
diff --git a/internal/registry/plan12_fix1_journal_test.go b/internal/registry/plan12_fix1_journal_test.go
new file mode 100644
index 000000000..48e3a77c9
--- /dev/null
+++ b/internal/registry/plan12_fix1_journal_test.go
@@ -0,0 +1,49 @@
+package registry
+
+import (
+ "errors"
+ "path/filepath"
+ "testing"
+)
+
+func TestPlan12RecoveryIntentJournalDurable(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err = r.BeginRecoveryHandover(t.Context(), "intent", "before"); err != nil {
+ t.Fatal(err)
+ }
+ if err = r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ r, err = Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer r.Close()
+ pending, err := r.PendingRecoveryHandovers(t.Context())
+ if err != nil || pending["intent"] != "before" {
+ t.Fatal(pending, err)
+ }
+ if err = r.ReviseRecoveryHandover(t.Context(), "intent", "before", "after"); err != nil {
+ t.Fatal(err)
+ }
+ if err = r.ReviseRecoveryHandover(t.Context(), "intent", "before", "stale"); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatal("stale revision", err)
+ }
+ if err = r.CompleteRecoveryHandover(t.Context(), "intent", "before"); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatal("stale completion", err)
+ }
+ if err = r.CompleteRecoveryHandover(t.Context(), "intent", "after"); err != nil {
+ t.Fatal(err)
+ }
+ pending, err = r.PendingRecoveryHandovers(t.Context())
+ if err != nil || len(pending) != 0 {
+ t.Fatal(pending, err)
+ }
+ if err = r.ReviseRecoveryHandover(t.Context(), "intent", "after", "reopened"); !errors.Is(err, ErrClaimOwnership) {
+ t.Fatal("reopened completed intent", err)
+ }
+}
diff --git a/internal/registry/recoveryjournal.go b/internal/registry/recoveryjournal.go
index 70501d3f7..a9a9631c7 100644
--- a/internal/registry/recoveryjournal.go
+++ b/internal/registry/recoveryjournal.go
@@ -3,6 +3,7 @@ package registry
import (
"context"
"database/sql"
+ "encoding/json"
"errors"
)
@@ -80,3 +81,62 @@ func (r *Registry) recoveryJournal(ctx context.Context) error {
_, err := r.sql.ExecContext(ctx, `CREATE TABLE IF NOT EXISTS recovery_handovers(id TEXT PRIMARY KEY,body TEXT NOT NULL,complete INTEGER NOT NULL)`)
return err
}
+
+func (r *Registry) PendingRecoveryHandovers(ctx context.Context) (map[string]string, error) {
+ if r.remote != nil {
+ out, err := r.remote.call(ctx, remoteRequest{Method: "PendingRecoveryHandovers"}, nil)
+ if err != nil {
+ return nil, err
+ }
+ var bodies map[string]string
+ err = json.Unmarshal([]byte(out.Text), &bodies)
+ return bodies, err
+ }
+ if err := r.recoveryJournal(ctx); err != nil {
+ return nil, err
+ }
+ rows, err := r.sql.QueryContext(ctx, `SELECT id,body FROM recovery_handovers WHERE complete=0`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ bodies := map[string]string{}
+ for rows.Next() {
+ var id, body string
+ if err := rows.Scan(&id, &body); err != nil {
+ return nil, err
+ }
+ bodies[id] = body
+ }
+ return bodies, rows.Err()
+}
+
+func (r *Registry) ReviseRecoveryHandover(ctx context.Context, id, prior, body string) error {
+ if r.remote != nil {
+ _, err := r.remote.call(ctx, remoteRequest{Method: "ReviseRecoveryHandover", Id: id, Expected: prior, Journal: body}, nil)
+ return err
+ }
+ if id == "" || prior == "" || body == "" {
+ return ErrClaimOwnership
+ }
+ if err := r.recoveryJournal(ctx); err != nil {
+ return err
+ }
+ tx, err := r.beginClaimTransaction(ctx, true)
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ result, err := tx.ExecContext(ctx, `UPDATE recovery_handovers SET body=? WHERE id=? AND body=? AND complete=0`, body, id, prior)
+ if err != nil {
+ return err
+ }
+ n, err := result.RowsAffected()
+ if err != nil {
+ return err
+ }
+ if n != 1 {
+ return ErrClaimOwnership
+ }
+ return tx.Commit()
+}
diff --git a/internal/registry/remote_dispatch.go b/internal/registry/remote_dispatch.go
index 00b7f40f1..cf85fc2eb 100644
--- a/internal/registry/remote_dispatch.go
+++ b/internal/registry/remote_dispatch.go
@@ -1,6 +1,9 @@
package registry
-import "context"
+import (
+ "context"
+ "encoding/json"
+)
type remoteRequest struct {
Method string
@@ -134,6 +137,16 @@ func (r *Registry) dispatchRemote(ctx context.Context, q remoteRequest, callback
err = r.SetTimerStatusRegistration(q.Host, q.Name, q.Registration, q.Status)
case "FireTimerNowRegistration":
err = r.FireTimerNowRegistration(q.Host, q.Name, q.Registration, q.Now)
+ case "ReviseRecoveryHandover":
+ err = r.ReviseRecoveryHandover(ctx, q.Id, q.Expected, q.Journal)
+ case "PendingRecoveryHandovers":
+ var bodies map[string]string
+ bodies, err = r.PendingRecoveryHandovers(ctx)
+ if err == nil {
+ var raw []byte
+ raw, err = json.Marshal(bodies)
+ out.Text = string(raw)
+ }
case "BeginRecoveryHandover":
err = r.BeginRecoveryHandover(ctx, q.Id, q.Journal)
case "PendingRecoveryHandover":