carlos / platform Public

moves: each lost tuple's boundary is its own generation's maximum in the restore, 0 when the restore omits it (Task 4 loss-boundary rule, platform-1e item 3)

Replaces the interim refusal (ErrLossTargetUnreadable, 96144320). A
target tuple's RemoteMaxTXID is genselect.GenerationMax over the restore
plan for that tuple's generation: 0 for a generation the restore does not
contain (every target generation behind a J4 return, every fresh one),
that generation's maximum when it is the restore's own. The source-cut
tuple keeps the journalled seal's N in src_gen. The fenced later-session
test admits through the writer again (no hand-written fence_admitted).

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

claude 1 week ago 92079fae3265274dca5524c44baa82374faccfaf parent 90a5873

3 files changed, +114 −85

internal/host/movesloss.go +19 −35
internal/host/movesloss_test.go +95 −47
internal/host/movesprobeowners.go +0 −3

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

index f88a16489..a6994e21d 100644

--- a/internal/host/movesloss.go

+++ b/internal/host/movesloss.go

@@ -3,12 +3,9 @@ package host

import (

"context"

"encoding/json"

- "errors"

"fmt"

- "log"

"reflect"

"strconv"

- "time"

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

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

@@ -38,12 +35,16 @@ import (

// own data bucket; platform-1e): each frozen source is vouched for from the

// maxima already recorded.

-// - replica:<region>/<gen>, a target session since the cut: the target's

-// replica is in the target's region, which this box never reads, and a

-// fresh target generation restarts its TXIDs at [1, 1] (spec §5.1), so

-// no maximum this box holds bounds its writes. Until platform-1e rules

-// how such a generation is declared, the admission is refused, typed

-// (ErrLossTargetUnreadable), with a standing log line (fix round 1).

+// - replica:<region>/<gen>, a target session since the cut: the loss

+// boundary is the highest TXID of the tuple's own generation that the

+// restore contains (platform-1e's loss-boundary rule), never a number

+// carried across generations: a fresh target generation restarts its

+// TXIDs at [1, 1] (spec §5.1). It is genselect.GenerationMax over the

+// restore plan: a generation the plan does not contain (every target

+// generation behind a J4 return, which restores src_gen) is 0, every

+// write in it lost. Every tuple is kept, however many share 0. The

+// claim compares the declared maximum with this validated one

+// (hostclaim/lost.go AdmitWriterWithLoss).

// - seal:<move>, the source's cut session when the seal did not cover the

// source's replica: the cut seal the claim journals, its maximum N.

@@ -62,14 +63,6 @@ import (

// check of the recovery's evidence, the lineage link and the source

// baseline's record before the CAS included, is unchanged.

-// ErrLossTargetUnreadable refuses a loss admission that would have to

-// record a remote maximum for a lost target's generation, which lives in

-// the target region's bucket this box cannot read (fix round 1, interim

-// until platform-1e rules how such a generation is declared). A fresh

-// target generation restarts its TXIDs at [1, 1] (spec §5.1), so no

-// maximum of another generation bounds its writes.

-var ErrLossTargetUnreadable = errors.New("moves: a lost target generation's maximum is in a region this box cannot read")

// claimLossRefs are the keys of the target's loss the claim journals under

// the op: the declaration's record key, and the fence record's key.

func claimLossRefs(c *hostclaim.Claim) (map[string]bool, error) {

@@ -167,7 +160,7 @@ func (r movesRecoveryOwners) movesLossSources(ctx context.Context, o hostclaim.O

}

out := make([]hostclaim.ValidatedWriteSource, 0, len(loss.WriteSources))

for _, s := range loss.WriteSources {

- max, err := r.sourceMaximum(c, v, cut, s)

+ max, err := r.sourceMaximum(c, v, cut, baseline, s)

if err != nil {

return nil, err

}

@@ -232,19 +225,9 @@ func (r movesRecoveryOwners) frozenMaximum(v recoveryProbe, baseline *genselect.

return nil

}

-// refusalNotice logs a standing refusal at most once a minute per key.

-func (r movesRecoveryOwners) refusalNotice(key, format string, args ...any) {

- now := time.Now()

- if last, ok := r.m.lossRefusals.Load(key); ok && now.Sub(last.(time.Time)) < time.Minute {

- return

- }

- r.m.lossRefusals.Store(key, now)

- log.Printf(format, args...)

// sourceMaximum vouches for one frozen source (see the file comment) and

// returns its maximum.

-func (r movesRecoveryOwners) sourceMaximum(c *hostclaim.Claim, v recoveryProbe, cut uint64, s hostclaim.WriteSource) (uint64, error) {

+func (r movesRecoveryOwners) sourceMaximum(c *hostclaim.Claim, v recoveryProbe, cut uint64, baseline *genselect.Plan, s hostclaim.WriteSource) (uint64, error) {

m := r.m

refused := func(why string) (uint64, error) {

return 0, fmt.Errorf("%w: frozen source %s/%s/%d: %s", hostclaim.ErrLossRecord, s.Box, s.Generation, s.Session, why)

@@ -261,12 +244,13 @@ func (r movesRecoveryOwners) sourceMaximum(c *hostclaim.Claim, v recoveryProbe,

if !park.SameRegion(v.homeRegion, region, s.Region) || !park.SameRegion(v.homeRegion, s.Region, c.Op.Intent["to"]) {

return refused("not the target region's replica")

}

- // The target's generation is in the target region's bucket, which

- // this box never reads, and its TXIDs restart at [1, 1]: no

- // maximum this box holds bounds its writes (fix round 1, interim

- // until platform-1e rules the declaration).

- r.refusalNotice(v.rec.Host+"#"+s.Generation, "moves: %s: the recovery of %s refuses its loss admission: the lost target's generation %s (session %d) is in %s, whose bucket this box cannot read", v.rec.Host, v.op, s.Generation, s.Session, s.Region)

- return 0, fmt.Errorf("%w: generation %s, session %d, in %s", ErrLossTargetUnreadable, s.Generation, s.Session, s.Region)

+ // The highest TXID of the tuple's own generation the restore

+ // contains: 0 for a generation outside it.

+ max, err := genselect.GenerationMax(baseline, v.holder.Location, s.Generation)

+ if err != nil {

+ return 0, fmt.Errorf("%w: the restore plan: %v", hostbox.ErrChanged, err)

+ }

+ return max, nil

case "seal":

if s.Reason != "source-cut" || identity != v.op || s.Box != m.Box || selectedGeneration(s.Generation) != selectedGeneration(v.srcGen) || s.Session != cut ||

!park.SameRegion(v.homeRegion, s.Region, m.Region) {

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

index 81e479764..f6867a9fd 100644

--- a/internal/host/movesloss_test.go

+++ b/internal/host/movesloss_test.go

@@ -267,8 +267,9 @@ func rcSetLossRef(t *testing.T, h *recoverHarness, ref string) {

}

// Q6: a return after the target's termination fence asks no coverage. Its

-// first admission would declare the fenced target's generation, which this

-// box cannot read: it is refused, typed (fix round 1, interim).

+// first admission declares the fenced target's generation, which the

+// restore (src_gen) does not contain, at 0: every write in it is lost

+// (platform-1e's loss-boundary rule).

func TestFencedTargetReturnAsksUncovered(t *testing.T) {

t.Run("the probing pass asks no coverage", func(t *testing.T) {

h := newRecoverHarness(t, false)

@@ -284,7 +285,7 @@ func TestFencedTargetReturnAsksUncovered(t *testing.T) {

t.Fatalf("probes %+v", h.probes)

}

})

- t.Run("the owners refuse the frozen target's admission", func(t *testing.T) {

+ t.Run("the owners admit the frozen target at 0", func(t *testing.T) {

f := newRCFixture(t, rcClean)

f.fence()

o := f.read()

@@ -298,12 +299,13 @@ func TestFencedTargetReturnAsksUncovered(t *testing.T) {

if _, err := w.Admit(t.Context(), o, covered, in.Baseline); err == nil {

t.Fatal("a fenced return admitted clean-return")

}

- before := f.read()

- if _, err := w.Admit(t.Context(), before, in.Admission, in.Baseline); !errors.Is(err, ErrLossTargetUnreadable) {

- t.Fatalf("the fenced target's generation: %v", err)

+ admitted, err := w.Admit(t.Context(), f.read(), in.Admission, in.Baseline)

+ if err != nil {

+ t.Fatal(err)

}

- if after := f.read(); after.ETag != before.ETag {

- t.Fatal("the claim moved")

+ a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]

+ if a.Coverage != "" || a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) || len(a.AdditionalWritesLost) != 0 {

+ t.Fatalf("the fenced admission: %+v", a)

}

})

}

@@ -343,20 +345,19 @@ func TestLaterSourceSessionStaysUncovered(t *testing.T) {

t.Fatal(err)

}

} else {

- // A fenced target's first admission is refused for now (fix

- // round 1); the fence's consumption is journalled by hand.

- record := f.fence()

+ f.fence()

w = f.lossWriter(nil)

- c := f.read().Claim

- c.Op.Journal["fence_admitted."+record.ID] = "true"

- f.putClaim(c)

- first = f.read()

+ in = f.selectAdmission(f.read())

+ var err error

+ if first, err = w.Admit(t.Context(), f.read(), in.Admission, in.Baseline); err != nil {

+ t.Fatal(err)

+ }

}

if pending, err := recoveryLossPending(first.Claim); err != nil || pending {

t.Fatalf("the loss is still pending: %v %v", pending, err)

}

restart := f.selectAdmission(first)

- if restart.Admission.Coverage != "" || (path == "declared" && restart.Admission.ObservedGeneration != in.Admission.NewGeneration) {

+ if restart.Admission.Coverage != "" || restart.Admission.ObservedGeneration != in.Admission.NewGeneration {

t.Fatalf("the restart: %+v", restart.Admission)

}

covered := restart.Admission

@@ -448,10 +449,11 @@ func TestRecordAndClaimDisagreeWaits(t *testing.T) {

// Q5 as platform-1e answered it: the loss baseline validator vouches for

// each frozen source from the maxima already recorded, never reading

-// another region's bucket: a seal's is the journalled cut seal's. A target

-// replica's generation is in a region this box cannot read and restarts its

-// TXIDs, so a well-formed one is refused, typed (fix round 1, interim); a

-// malformed one is refused as not the recovery's.

+// another region's bucket: a seal's is the journalled cut seal's N; a

+// target replica's is the highest TXID of its own generation the restore

+// contains, 0 for a generation the restore (src_gen) does not contain

+// (platform-1e's loss-boundary rule). A malformed source is refused as not

+// the recovery's.

func TestLossSourcesValidatesEachFrozenBaseline(t *testing.T) {

ctx := context.Background()

t.Run("a declared loss's replica and seal", func(t *testing.T) {

@@ -464,8 +466,12 @@ func TestLossSourcesValidatesEachFrozenBaseline(t *testing.T) {

if err != nil {

t.Fatal(err)

}

- if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); !errors.Is(err, ErrLossTargetUnreadable) || errors.Is(err, hostbox.ErrChanged) {

- t.Fatalf("a target replica: %v", err)

+ both, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline)

+ if err != nil {

+ t.Fatal(err)

+ }

+ if len(both) != 2 || both[0].RemoteMaxTXID != 0 || both[0].Generation != poOp || both[1].RemoteMaxTXID != rcSealN {

+ t.Fatalf("the target at 0 and the cut at N: %+v", both)

}

cutOnly := loss.Record

cutOnly.WriteSources = []hostclaim.WriteSource{rcCutSource()}

@@ -561,9 +567,12 @@ func TestLossSourcesValidatesEachFrozenBaseline(t *testing.T) {

record := f.fence()

o := f.read()

in := f.selectAdmission(o)

- // Its one source is the fenced target's generation: refused, typed.

- if _, err := f.owners.LossSources(ctx, o, record, in.Admission, in.Baseline); !errors.Is(err, ErrLossTargetUnreadable) {

- t.Fatalf("the fenced target: %v", err)

+ // Its one source is the fenced target's generation, outside the

+ // restore: 0.

+ got, err := f.owners.LossSources(ctx, o, record, in.Admission, in.Baseline)

+ want := []hostclaim.ValidatedWriteSource{{Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp, RemoteMaxTXID: 0}}

+ if err != nil || !reflect.DeepEqual(got, want) {

+ t.Fatalf("the fenced target: %+v %v", got, err)

}

bad := record

bad.ID = "auto-other"

@@ -719,9 +728,10 @@ func TestRecoveryAdmissionEvidenceAcceptsTheFrozenMaximum(t *testing.T) {

// wNoDuplicateDeclarationEarly/Late): behind a lost target the source's

// recovery declares every target session since the cut and, when the seal

// did not cover the source's replica, the cut session, each once: the

-// predecessor's as writes_lost, the rest as additional_writes_lost. Until

-// platform-1e rules how a target generation this box cannot read is

-// declared, a declaration that freezes one is refused whole (fix round 1).

+// predecessor's as writes_lost, the rest as additional_writes_lost. Each

+// tuple's maximum is the highest TXID of its own generation the restore

+// contains (platform-1e's loss-boundary rule): 0 for a target generation

+// outside it, the journalled seal's N for the cut session.

func TestLostTargetReturnDeclaresEverySessionOnce(t *testing.T) {

ctx := context.Background()

t.Run("two target sessions and the cut", func(t *testing.T) {

@@ -732,13 +742,40 @@ func TestLostTargetReturnDeclaresEverySessionOnce(t *testing.T) {

c.Holder.Session = 25

c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})

})

- before := f.read()

- in := f.selectAdmission(before)

- if _, err := f.lossWriter(pub).Admit(ctx, before, in.Admission, in.Baseline); !errors.Is(err, ErrLossTargetUnreadable) {

- t.Fatalf("two target generations this box cannot read: %v", err)

+ in := f.selectAdmission(f.read())

+ admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)

+ if err != nil {

+ t.Fatal(err)

}

- if after := f.read(); after.ETag != before.ETag {

- t.Fatal("the claim moved")

+ a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]

+ want := []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}, {Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}

+ if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "t2", Session: 25, RemoteMaxTXID: 0}) || !reflect.DeepEqual(a.AdditionalWritesLost, want) {

+ t.Fatalf("declared %+v / %+v", a.WritesLost, a.AdditionalWritesLost)

+ }

+ })

+ t.Run("a lost tuple in the restore's own generation", func(t *testing.T) {

+ // The target's session wrote into src itself: the restore contains

+ // that generation, so its maximum is GenerationMax(src), 30.

+ f := newRCFixture(t, rcClean)

+ pub := f.declare(true, []hostclaim.WriteSource{{Reason: "target", Box: "uk-box", Generation: "src", Session: 20, Region: "uk", Baseline: hostops.MoveLossReplicaBaseline + "uk/src"}}, func(c *hostclaim.Claim) {

+ c.Holder.Admissions[1].Generation, c.Holder.Admissions[1].Predecessor = "src", "src"

+ })

+ in := f.selectAdmission(f.read())

+ loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)

+ if err != nil {

+ t.Fatal(err)

+ }

+ got, err := f.owners.LossSources(ctx, f.read(), loss.Record, in.Admission, in.Baseline)

+ if err != nil || len(got) != 1 || got[0].RemoteMaxTXID != 30 || in.Baseline.Generation != "src" {

+ t.Fatalf("validated %+v %v", got, err)

+ }

+ admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)

+ if err != nil {

+ t.Fatal(err)

+ }

+ a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]

+ if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "src", Session: 20, RemoteMaxTXID: 30}) {

+ t.Fatalf("declared %+v", a.WritesLost)

}

})

t.Run("a target lost before its first admission", func(t *testing.T) {

@@ -773,12 +810,14 @@ func TestLostTargetReturnDeclaresEverySessionOnce(t *testing.T) {

})

}

-// Fix round 1 (Critical): a fresh target generation restarts its TXIDs at

-// [1, 1] (spec §5.1), so the source's maximum, 30, is no loss boundary for

-// it: the target's generation t2 acknowledged a write at TXID 2 that the

-// restore of src omits. With no read of the target's region, the loss

-// admission is refused, typed, never admitted with a borrowed maximum.

-func TestLostTargetGenerationResetIsRefused(t *testing.T) {

+// The loss-boundary rule (platform-1e): a fresh target generation restarts

+// its TXIDs at [1, 1] (spec §5.1), so the source's maximum, 30, is no loss

+// boundary for it: t2 acknowledged a write at TXID 2 that the restore of

+// src omits. A lost tuple's maximum is the highest TXID of its own

+// generation the restore contains: 0 here, so every write of t2 (and of

+// mv's generation) is declared lost. The validated source carries the same

+// 0, which the claim compares (lost.go:194).

+func TestLostTargetGenerationOutsideTheRestoreDeclaresAllItsWrites(t *testing.T) {

ctx := context.Background()

f := newRCFixture(t, rcClean)

// The target's own bucket, which the source never reads: t2 restarted

@@ -790,13 +829,22 @@ func TestLostTargetGenerationResetIsRefused(t *testing.T) {

c.Holder.Session = 25

c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})

})

- before := f.read()

- in := f.selectAdmission(before)

- if _, err := f.lossWriter(pub).Admit(ctx, before, in.Admission, in.Baseline); !errors.Is(err, ErrLossTargetUnreadable) {

- t.Fatalf("a target generation this box cannot read was admitted or refused untyped: %v", err)

+ in := f.selectAdmission(f.read())

+ loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)

+ if err != nil {

+ t.Fatal(err)

+ }

+ validated, err := f.owners.LossSources(ctx, f.read(), loss.Record, in.Admission, in.Baseline)

+ if err != nil || len(validated) != 2 || validated[0].Generation != "t2" || validated[0].RemoteMaxTXID != 0 || validated[1].RemoteMaxTXID != 0 {

+ t.Fatalf("validated %+v %v", validated, err)

}

- after := f.read()

- if after.ETag != before.ETag || len(after.Claim.Holder.Admissions) != len(before.Claim.Holder.Admissions) || hostclaim.OperationLossConsumed(after.Claim, rcLossID) {

- t.Fatalf("the claim moved: %+v", after.Claim.Holder.Admissions)

+ admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)

+ if err != nil {

+ t.Fatal(err)

+ }

+ a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]

+ if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "t2", Session: 25, RemoteMaxTXID: 0}) ||

+ !reflect.DeepEqual(a.AdditionalWritesLost, []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}}) || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) {

+ t.Fatalf("declared %+v / %+v", a.WritesLost, a.AdditionalWritesLost)

}

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

index b8770c8e9..7fda90e53 100644

--- a/internal/host/movesprobeowners.go

+++ b/internal/host/movesprobeowners.go

@@ -156,9 +156,6 @@ type movesProbeOwners struct {

// recovery's AdmissionEvidence of the same admission consumes

// (movesloss.go).

lossValidated map[string]lossValidation

- // lossRefusals are the standing loss-admission refusals logged, by key

- // and time (movesloss.go refusalNotice).

- lossRefusals sync.Map

}

var (