| new file mode 100644 |
| index 000000000..61ab154b4 |
| --- /dev/null |
| +++ b/internal/host/movesloss_test.go |
| @@ -0,0 +1,758 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "crypto/ed25519" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "reflect" |
| + "strconv" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/genselect" |
| + "github.com/carlosframework/platform/internal/hostbox" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostops" |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/moves" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The loss and fence rules on the box (moves-after-Plan-12 Task 4: owed (a) |
| +// and (b), Plan 4 Task 10's box half; rulings Q5 as platform-1e answered |
| +// it, Q6, Q7's loss half, Q9's gate). |
| + |
| +const ( |
| + rcLossID = "loss_1" |
| + // rcSealN is the cut seal's maximum: the stopped database's position, |
| + // beyond the source replica's 30 when src_covered is false. |
| + rcSealN = 35 |
| +) |
| + |
| +// rcLossIntent is the moves recovery intent a declaration signs |
| +// (hostops.moveLossIntent's wire form). |
| +func (f *rcFixture) rcLossIntent(srcCovered bool) json.RawMessage { |
| + raw, err := json.Marshal(map[string]any{"kind": "move-target-lost", "move": poOp, "incarnation": f.inc, "rollback": rpRB, "target": "uk-box", "source": "ie-box", |
| + "target_region": "uk", "source_region": "ie", "record_key": "apps/bab/jam/instances/" + poHost + ".json", "record_etag": "etag-1", "record_seq": 1, |
| + "record_state": "rolling_back", "dirty": false, "return_lane": false, "src_covered": srcCovered, "cut_session": rcS, "src_gen": "src"}) |
| + if err != nil { |
| + f.t.Fatal(err) |
| + } |
| + return raw |
| +} |
| + |
| +// rcCutSeal is the cut's seal the claim journals. |
| +func rcCutSeal(srcCovered bool) moves.Seal { |
| + at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC) |
| + return moves.Seal{V: moves.SealVersion, N: rcSealN, FromDB: true, SrcCovered: srcCovered, SealedAt: at, LeaseUntil: at.Add(time.Minute), |
| + Replica: []moves.SealedObject{{Key: "moves/" + poHost + "/" + poOp + "/replica/0009/0000000000000001-0000000000000023.ltx", VersionID: "s1", Size: 100}}, |
| + Files: []moves.SealedObject{{Key: "moves/" + poHost + "/" + poOp + "/files/archive.sealed", VersionID: "s2", Size: 10}, {Key: "moves/" + poHost + "/" + poOp + "/files/archive.json", VersionID: "s3", Size: 10}}} |
| +} |
| + |
| +// rcTargetSource is the frozen source of one target session since the cut. |
| +func rcTargetSource(gen string, session uint64) hostclaim.WriteSource { |
| + return hostclaim.WriteSource{Reason: "target", Box: "uk-box", Generation: gen, Session: session, Region: "uk", Baseline: hostops.MoveLossReplicaBaseline + "uk/" + gen} |
| +} |
| + |
| +// rcCutSource is the frozen source of the source's cut session. |
| +func rcCutSource() hostclaim.WriteSource { |
| + return hostclaim.WriteSource{Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: hostops.MoveLossSealBaseline + poOp} |
| +} |
| + |
| +// declare turns f's clean return into the return after a declared-lost |
| +// target (Plan 4 Task 10): the claim as DeclareLost and the lost target's |
| +// release leave it, edit applied, the record's rollback citing the |
| +// declaration, and the root-signed declaration committed with sources. It |
| +// returns the root's public key. |
| +func (f *rcFixture) declare(srcCovered bool, sources []hostclaim.WriteSource, edit func(*hostclaim.Claim)) ed25519.PublicKey { |
| + t := f.t |
| + t.Helper() |
| + c := f.recoveryClaim() |
| + intent := f.rcLossIntent(srcCovered) |
| + c.Lost, c.LossRefs, c.Op.Participants = []string{"uk-box"}, []string{rcLossID}, []string{"ie-box"} |
| + c.Op.Journal["loss."+rcLossID] = string(intent) |
| + c.Op.Journal["loss_gate."+rcLossID] = `{"box":"uk-box","revoke_execution":true}` |
| + c.Op.Journal["lost_target_released"] = rcLossID |
| + c.Op.Journal[hostclaim.JournalSeal] = rcCutSeal(srcCovered).Encode() |
| + if edit != nil { |
| + edit(c) |
| + } |
| + f.putClaim(c) |
| + key, err := hostclaim.LossRecordKey(rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f.rec.Move.Rollback.LossRef, f.rec.Move.Rollback.Evidence = key, key |
| + pub, priv, err := ed25519.GenerateKey(nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + now := time.Now().UTC() |
| + r := hostclaim.LossRecord{V: 1, ID: rcLossID, State: "prepared", Actor: "operator", AcceptedAt: now, Host: poHost, HolderIncarnation: f.inc, HolderStamp: rcS, HolderSession: c.Holder.Session, |
| + ExpectedClaimGen: c.Gen, ExpectedClaimETag: "claim-etag", OpID: poOp, OpKind: hostclaim.KindRestoring, OpPhase: hostclaim.PhaseReleasing, Box: "uk-box", Enrollment: "enroll", QuarantineRevision: 1, |
| + CloudObservation: hostclaim.LossCloudObservation{ObservedAt: now, Account: "123", Region: "eu-west-2", InstanceID: "i-1", State: "terminated", RequestID: "request"}, |
| + OperatorAttestation: hostclaim.LossOperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true}, FencedAt: now, |
| + ConfirmedConsequences: hostclaim.LossConsequences{RevokeExecution: true, RemoveParticipation: true, RecoveryIntent: intent}, |
| + Timers: hostclaim.TimerLoss{Knowledge: hostclaim.TimersUnknown}, WriteSources: sources} |
| + created, err := hostclaim.CreateLossRecord(t.Context(), f.claims, release.RootSigner(priv), r) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + r.State, r.ResultClaimGen = "committed", c.Gen |
| + if _, err := hostclaim.ReplaceLossRecord(t.Context(), f.claims, release.RootSigner(priv), created, r); err != nil { |
| + t.Fatal(err) |
| + } |
| + return pub |
| +} |
| + |
| +// rcFenceEnrollment is the fenced target's enrollment. |
| +const rcFenceEnrollment = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" |
| + |
| +// fence turns f's clean return into the return after the target's |
| +// termination fence (spec rev 7.20 §7.10): the consumed fence, the |
| +// automatic loss of the target's session 20 in poOp, the record's rollback |
| +// citing the fence. It returns the derived record the writer passes. |
| +func (f *rcFixture) fence() hostclaim.LossRecord { |
| + t := f.t |
| + t.Helper() |
| + c := f.recoveryClaim() |
| + ref := hostclaim.FenceRef{Enrollment: rcFenceEnrollment, Key: "control/box-fences/" + rcFenceEnrollment + ".json", SHA256: strings.Repeat("b", 64)} |
| + raw, err := json.Marshal(ref) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + sum := sha256.Sum256([]byte(poOp + "\x00uk-box\x00" + ref.Enrollment + "\x00" + ref.SHA256)) |
| + id := "auto-" + hex.EncodeToString(sum[:16]) |
| + c.Op.Intent["target_enrollment"] = ref.Enrollment |
| + c.Op.Journal["fence.uk-box"], c.Op.Journal["fenced_target_released"] = string(raw), string(raw) |
| + c.Op.Journal[hostclaim.JournalSeal] = rcCutSeal(true).Encode() |
| + c.Op.Participants, c.Lost, c.LossRefs = []string{"ie-box"}, []string{"uk-box"}, []string{id} |
| + c.EvidenceRefs = []hostclaim.AutomaticLoss{{ID: id, Box: "uk-box", Enrollment: ref.Enrollment, Fence: ref, ObservedAt: time.Date(2026, 10, 1, 1, 0, 0, 0, time.UTC).Format(time.RFC3339Nano), |
| + Op: poOp, Kind: hostclaim.KindLanding, Phase: hostclaim.PhaseLandingProbing, Role: "target", Incarnation: f.inc, Stamp: 20, Session: 20, Generation: poOp, Timers: hostclaim.TimersUnknown}} |
| + f.putClaim(c) |
| + f.rec.Move.Rollback.LossRef, f.rec.Move.Rollback.Evidence = ref.Key, ref.Key |
| + record, ok := hostclaim.PendingMoveFenceLoss(c) |
| + if !ok { |
| + t.Fatal("no pending fence loss") |
| + } |
| + return record |
| +} |
| + |
| +// lossWriter is the claim's writer with the moves loss owner and baseline |
| +// validator, over the source's own bucket. |
| +func (f *rcFixture) lossWriter(pub ed25519.PublicKey) hostbox.Writer { |
| + return hostbox.Writer{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: "ie", Box: "ie-box", AdmissionEvidence: f.owners.AdmissionEvidence, |
| + Loss: &hostbox.LossAdmission{PublicKey: pub, Owner: hostops.MoveLossRecoveryOwner(), Sources: f.owners.LossSources}} |
| +} |
| + |
| +// Q6: a return after a declared loss is never covered: the probe's request |
| +// asks no coverage, the request matcher refuses clean-return for it, and the |
| +// recovery owners select and admit it uncovered. |
| +func TestDeclaredLossNeverAdmitsCleanReturn(t *testing.T) { |
| + t.Run("the probing pass asks no coverage", func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + h.released(t) |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil } |
| + h.srcPass(t) |
| + h.sourceAt(t, provision.RbSourceProbing) |
| + if len(h.probes) != 1 || h.probes[0].Coverage != recoveryCoverageClean || h.probes[0].Declared { |
| + t.Fatalf("before the loss: %+v", h.probes) |
| + } |
| + key := rcDeclareOnHarness(t, h) |
| + h.srcPass(t) |
| + if len(h.probes) != 2 { |
| + t.Fatalf("probes %+v", h.probes) |
| + } |
| + req := h.probes[1] |
| + if !req.Declared || !req.Clean || req.Coverage != "" || req.Generation != "" || req.SourceGeneration == "" { |
| + t.Fatalf("a declared return's request: %+v", req) |
| + } |
| + rec := h.read(t, p1Host) |
| + if rec.Move.Rollback.LossRef != key || recoveryProbeMatches(rec, req) != nil { |
| + t.Fatalf("the request does not match its record: %v", recoveryProbeMatches(rec, req)) |
| + } |
| + for name, mutate := range map[string]func(*recoveryProbeRequest){ |
| + "clean-return": func(r *recoveryProbeRequest) { r.Coverage = recoveryCoverageClean }, |
| + "sealed": func(r *recoveryProbeRequest) { r.Coverage = recoveryCoverageSealed }, |
| + "not declared": func(r *recoveryProbeRequest) { r.Declared = false }, |
| + "undeclared kind": func(r *recoveryProbeRequest) { r.Declared, r.Coverage = false, recoveryCoverageClean }, |
| + "dirty": func(r *recoveryProbeRequest) { r.Clean, r.Generation = false, r.RollbackID }, |
| + } { |
| + bad := req |
| + mutate(&bad) |
| + if err := recoveryProbeMatches(rec, bad); !errors.Is(err, errRecoveryProbeRequest) { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + } |
| + // A request that says declared for a record that does not is refused. |
| + undeclared := rec |
| + rb := *rec.Move.Rollback |
| + rb.LossRef = "" |
| + mv := *rec.Move |
| + mv.Rollback = &rb |
| + undeclared.Move = &mv |
| + if err := recoveryProbeMatches(undeclared, req); !errors.Is(err, errRecoveryProbeRequest) { |
| + t.Fatalf("declared over an undeclared record: %v", err) |
| + } |
| + }) |
| + t.Run("the owners admit it uncovered, through the loss owner", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + if in.Admission.Coverage != "" || in.Admission.LossID != "" { |
| + t.Fatalf("admission %+v", in.Admission) |
| + } |
| + w := f.lossWriter(pub) |
| + covered := in.Admission |
| + covered.Coverage = recoveryCoverageClean |
| + if _, err := w.Admit(t.Context(), o, covered, in.Baseline); err == nil { |
| + t.Fatal("a declared return admitted clean-return") |
| + } |
| + admitted, err := w.Admit(t.Context(), 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.Coverage != "" || a.LossID != rcLossID || a.Covered || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) { |
| + t.Fatalf("the loss admission: %+v", a) |
| + } |
| + }) |
| +} |
| + |
| +// rcDeclareOnHarness declares the recover harness's target lost: the claim |
| +// journals the declaration and the record's rollback cites its key. It |
| +// returns the key. |
| +func rcDeclareOnHarness(t *testing.T, h *recoverHarness) string { |
| + t.Helper() |
| + key, err := hostclaim.LossRecordKey(rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { |
| + c.LossRefs = append(c.LossRefs, rcLossID) |
| + c.Op.Journal["loss."+rcLossID] = `{"kind":"move-target-lost","target":"uk-box"}` |
| + }) |
| + rcSetLossRef(t, h, key) |
| + return key |
| +} |
| + |
| +// rcSetLossRef writes the record's rollback LossRef through provision's |
| +// record writer. |
| +func rcSetLossRef(t *testing.T, h *recoverHarness, ref string) { |
| + t.Helper() |
| + if _, err := provision.SetRecordMove(t.Context(), h.s, "bab", "app", p1Host, func(r *provision.Record) error { |
| + r.Move.Rollback.LossRef = ref |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// Q6: a return after the target's termination fence asks no coverage, and |
| +// its first admission takes the fenced target's frozen maximum. |
| +func TestFencedTargetReturnAsksUncovered(t *testing.T) { |
| + t.Run("the probing pass asks no coverage", func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + h.released(t) |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil } |
| + h.srcPass(t) |
| + ref := hostclaim.FenceRef{Enrollment: rcFenceEnrollment, Key: "control/box-fences/" + rcFenceEnrollment + ".json", SHA256: strings.Repeat("b", 64)} |
| + raw, _ := json.Marshal(ref) |
| + rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Journal["fence.uk-box"] = string(raw) }) |
| + rcSetLossRef(t, h, ref.Key) |
| + h.srcPass(t) |
| + if len(h.probes) != 2 || !h.probes[1].Declared || h.probes[1].Coverage != "" { |
| + t.Fatalf("probes %+v", h.probes) |
| + } |
| + }) |
| + t.Run("the owners admit the frozen target", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + f.fence() |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + if in.Admission.Coverage != "" { |
| + t.Fatalf("a fenced return asked %q", in.Admission.Coverage) |
| + } |
| + covered := in.Admission |
| + covered.Coverage = recoveryCoverageClean |
| + w := f.lossWriter(nil) |
| + if _, err := w.Admit(t.Context(), o, covered, in.Baseline); err == nil { |
| + t.Fatal("a fenced return admitted clean-return") |
| + } |
| + admitted, err := w.Admit(t.Context(), f.read(), in.Admission, in.Baseline) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1] |
| + // The target's replica is in its own region, which this box never |
| + // reads (platform-1e): the declaration names the admission's frozen |
| + // maximum, src_gen's 30 in this box's own bucket. |
| + if a.Coverage != "" || a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 30}) || len(a.AdditionalWritesLost) != 0 { |
| + t.Fatalf("the fenced admission: %+v", a) |
| + } |
| + }) |
| +} |
| + |
| +// Q6: every later session of the recovered source stays uncovered: the |
| +// loss is consumed by the first admission, but the declaration (or the |
| +// fence) is still the op's, so a restart asks, selects and admits no |
| +// coverage, and the claim refuses any. |
| +func TestLaterSourceSessionStaysUncovered(t *testing.T) { |
| + t.Run("the probing pass", func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + h.released(t) |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil } |
| + h.srcPass(t) |
| + rcDeclareOnHarness(t, h) |
| + rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { |
| + c.Op.Journal["loss_admitted."+rcLossID] = strconv.FormatUint(c.Holder.Session, 10) |
| + }) |
| + h.srcPass(t) |
| + if len(h.probes) != 2 || !h.probes[1].Declared || h.probes[1].Coverage != "" { |
| + t.Fatalf("probes %+v", h.probes) |
| + } |
| + }) |
| + for _, path := range []string{"declared", "fenced"} { |
| + t.Run("the owners: "+path, func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + var w hostbox.Writer |
| + if path == "declared" { |
| + w = f.lossWriter(f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil)) |
| + } else { |
| + f.fence() |
| + w = f.lossWriter(nil) |
| + } |
| + in := f.selectAdmission(f.read()) |
| + first, err := w.Admit(t.Context(), f.read(), in.Admission, in.Baseline) |
| + if 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 != "" || restart.Admission.ObservedGeneration != in.Admission.NewGeneration { |
| + t.Fatalf("the restart: %+v", restart.Admission) |
| + } |
| + covered := restart.Admission |
| + covered.Coverage = recoveryCoverageClean |
| + if err := f.owners.AdmissionEvidence(t.Context(), first, covered, restart.Baseline); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("a later session admitted clean-return: %v", err) |
| + } |
| + again, err := w.Admit(t.Context(), first, restart.Admission, restart.Baseline) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := again.Claim.Holder.Admissions[len(again.Claim.Holder.Admissions)-1] |
| + if a.Coverage != "" || a.LossID != "" || a.Generation != restart.Admission.NewGeneration { |
| + t.Fatalf("the restart's admission: %+v", a) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// Q6: the record and the claim must agree on the target's loss, by |
| +// identity; until they do the probe waits: no request is made, the owners |
| +// select nothing, and the recovery stays at probing. |
| +func TestRecordAndClaimDisagreeWaits(t *testing.T) { |
| + t.Run("the probing pass", func(t *testing.T) { |
| + for name, disagree := range map[string]func(t *testing.T, h *recoverHarness){ |
| + "the record alone": func(t *testing.T, h *recoverHarness) { rcSetLossRef(t, h, "control/host-loss/"+rcLossID+".json") }, |
| + "the claim alone": func(t *testing.T, h *recoverHarness) { rcDeclareOnHarness(t, h); rcSetLossRef(t, h, "") }, |
| + "another loss": func(t *testing.T, h *recoverHarness) { |
| + rcDeclareOnHarness(t, h) |
| + rcSetLossRef(t, h, "control/host-loss/loss_2.json") |
| + }, |
| + "a fence's for loss": func(t *testing.T, h *recoverHarness) { |
| + rcDeclareOnHarness(t, h) |
| + rcSetLossRef(t, h, "control/box-fences/x.json") |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + h := newRecoverHarness(t, false) |
| + h.released(t) |
| + h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil } |
| + h.srcPass(t) |
| + disagree(t, h) |
| + h.srcPass(t) |
| + if len(h.probes) != 1 { |
| + t.Fatalf("probed while the record and the claim disagree: %+v", h.probes) |
| + } |
| + h.sourceAt(t, provision.RbSourceProbing) |
| + }) |
| + } |
| + }) |
| + t.Run("the owners", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil) |
| + for name, ref := range map[string]string{"no loss": "", "another loss": "control/host-loss/loss_2.json"} { |
| + f.rec.Move.Rollback.LossRef = ref |
| + if _, err := f.owners.Select(t.Context(), f.route, f.read(), f.policy()); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + } |
| + }) |
| + t.Run("the adapter", func(t *testing.T) { |
| + f := newRecoveryFixture(t) |
| + inc := plan10J2AdapterClaim(t, f, hostclaim.PhaseProbing, "src-gen") |
| + rec := plan10J2AdapterRecord(inc, false) |
| + var events []string |
| + req := plan10J2AdapterRequest(&rec, "src-gen", &events) |
| + a := recoveryProbeAdapter{Box: "box-1", Claims: true, |
| + Wake: func(context.Context, registry.Route, activator.StartPolicy) error { |
| + events = append(events, "wake") |
| + return nil |
| + }, |
| + Route: func(host string) (*registry.Route, error) { return ®istry.Route{Host: host}, nil }, |
| + Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) { |
| + return plan10J2AdapterInputs(), nil |
| + }} |
| + // The claim journals a declaration the record does not cite. |
| + if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error { |
| + c.LossRefs = append(c.LossRefs, rcLossID) |
| + c.Op.Journal["loss."+rcLossID] = `{"kind":"move-target-lost","target":"` + c.Op.Intent["target"] + `"}` |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if out, err := a.probe(t.Context(), f.claims, rec, req); !errors.Is(err, hostbox.ErrChanged) || out != recoveryProbeWaiting || len(events) != 0 { |
| + t.Fatalf("%v %v %v", out, err, events) |
| + } |
| + }) |
| +} |
| + |
| +// 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 target replica's is the admission's frozen |
| +// maximum (the source baseline's, read in this box's own bucket by |
| +// genselect.GenerationMax), a seal's the journalled cut seal's. |
| +func TestLossSourcesValidatesEachFrozenBaseline(t *testing.T) { |
| + ctx := context.Background() |
| + t.Run("a declared loss's replica and seal", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + sources := []hostclaim.WriteSource{rcTargetSource(poOp, 20), rcCutSource()} |
| + pub := f.declare(false, sources, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []hostclaim.ValidatedWriteSource{ |
| + {Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp, RemoteMaxTXID: 30}, |
| + {Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: "seal:" + poOp, RemoteMaxTXID: rcSealN}, |
| + } |
| + if !reflect.DeepEqual(got, want) { |
| + t.Fatalf("validated %+v\nwant %+v", got, want) |
| + } |
| + for name, mutate := range map[string]func(*hostclaim.WriteSource){ |
| + "another target box": func(s *hostclaim.WriteSource) { s.Box = "de-box" }, |
| + "a replica of another gen": func(s *hostclaim.WriteSource) { s.Baseline = "replica:uk/other" }, |
| + "a replica of another region": func(s *hostclaim.WriteSource) { s.Baseline = "replica:de/" + poOp }, |
| + "a target outside its region": func(s *hostclaim.WriteSource) { s.Region, s.Baseline = "de", "replica:de/"+poOp }, |
| + "a session before the cut": func(s *hostclaim.WriteSource) { s.Session = rcS }, |
| + "a target named source-cut": func(s *hostclaim.WriteSource) { s.Reason = "source-cut" }, |
| + "a malformed baseline": func(s *hostclaim.WriteSource) { s.Baseline = "replica:uk" }, |
| + "an unknown baseline": func(s *hostclaim.WriteSource) { s.Baseline = "snapshot:x" }, |
| + } { |
| + bad := loss.Record |
| + bad.WriteSources = append([]hostclaim.WriteSource(nil), sources...) |
| + mutate(&bad.WriteSources[0]) |
| + if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("target, %s: %v", name, err) |
| + } |
| + } |
| + for name, mutate := range map[string]func(*hostclaim.WriteSource){ |
| + "another move's seal": func(s *hostclaim.WriteSource) { s.Baseline = "seal:mv_other" }, |
| + "another box": func(s *hostclaim.WriteSource) { s.Box = "uk-box" }, |
| + "another generation": func(s *hostclaim.WriteSource) { s.Generation = "other" }, |
| + "another session": func(s *hostclaim.WriteSource) { s.Session = rcS + 1 }, |
| + "another region": func(s *hostclaim.WriteSource) { s.Region = "uk" }, |
| + "a seal named target": func(s *hostclaim.WriteSource) { s.Reason = "target" }, |
| + "an unknown reason": func(s *hostclaim.WriteSource) { s.Reason = "other" }, |
| + "a replica for the cut seal": func(s *hostclaim.WriteSource) { s.Baseline = "replica:ie/src" }, |
| + } { |
| + bad := loss.Record |
| + bad.WriteSources = append([]hostclaim.WriteSource(nil), sources...) |
| + mutate(&bad.WriteSources[1]) |
| + if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("seal, %s: %v", name, err) |
| + } |
| + } |
| + // A seal that covered the source's replica holds no source-cut |
| + // writes: a source-cut source under it is not the cut's. |
| + covered := newRCFixture(t, rcClean) |
| + pub = covered.declare(true, sources, nil) |
| + o = covered.read() |
| + in = covered.selectAdmission(o) |
| + loss, err = hostclaim.ReadLossRecord(ctx, covered.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := covered.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("a source-cut under a covering seal: %v", err) |
| + } |
| + }) |
| + t.Run("the record is the pending declaration", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, mutate := range map[string]func(*hostclaim.LossRecord){ |
| + "another id": func(r *hostclaim.LossRecord) { r.ID = "loss_2" }, |
| + "another op": func(r *hostclaim.LossRecord) { r.OpID = "mv_other" }, |
| + "another box": func(r *hostclaim.LossRecord) { r.Box = "de-box" }, |
| + "another incarnation": func(r *hostclaim.LossRecord) { r.HolderIncarnation = "other" }, |
| + } { |
| + bad := loss.Record |
| + mutate(&bad) |
| + if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + } |
| + // The baseline must be the source baseline the recovery restores. |
| + other := *in.Baseline |
| + other.Generation = poOp |
| + if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, &other); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("another restore point: %v", err) |
| + } |
| + if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, nil); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("no restore point: %v", err) |
| + } |
| + }) |
| + t.Run("the automatic fence record", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + record := f.fence() |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + got, err := f.owners.LossSources(ctx, o, record, in.Admission, in.Baseline) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []hostclaim.ValidatedWriteSource{{Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp, RemoteMaxTXID: 30}} |
| + if !reflect.DeepEqual(got, want) { |
| + t.Fatalf("validated %+v", got) |
| + } |
| + bad := record |
| + bad.ID = "auto-other" |
| + if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("another automatic loss: %v", err) |
| + } |
| + }) |
| + t.Run("only a first recovery admission's loss", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := f.owners.LossSources(ctx, hostclaim.Observed{Host: poHost, Absent: true}, loss.Record, in.Admission, in.Baseline); err == nil { |
| + t.Fatal("no claim") |
| + } |
| + // No loss pending: the validator vouches for nothing. |
| + undeclared := newRCFixture(t, rcClean) |
| + uo := undeclared.read() |
| + if _, err := undeclared.owners.LossSources(ctx, uo, loss.Record, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("no pending loss: %v", err) |
| + } |
| + }) |
| +} |
| + |
| +// failingList is the source's own bucket, unreadable for listings. |
| +type failingList struct{ genselect.Reader } |
| + |
| +func (failingList) ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error) { |
| + return nil, errors.New("SlowDown") |
| +} |
| + |
| +// An unreadable source maximum is a wait, never a refusal (Q5): with no |
| +// cross-region read the one read is the admission's frozen maximum in this |
| +// box's own bucket, and while it cannot be read, or reads otherwise than |
| +// the selection, the validator answers hostbox.ErrChanged, which the probe |
| +// adapter turns into another pass. |
| +func TestLossSourcesWaitsOnAnUnreadableTarget(t *testing.T) { |
| + ctx := context.Background() |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + data := f.owners.Data |
| + f.owners.Data = failingList{data} |
| + _, err = f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline) |
| + if !errors.Is(err, hostbox.ErrChanged) || errors.Is(err, hostclaim.ErrLossRecord) { |
| + t.Fatalf("an unreadable bucket: %v", err) |
| + } |
| + f.owners.Data = data |
| + // The remote moved past the selection: wait for the next selection. |
| + f.bucket.Put("data", "instances/"+poHost+"/g-src/0000/0000000000000031-0000000000000031.ltx", f.ltxHeader(31, 31)) |
| + if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("a moved remote: %v", err) |
| + } |
| +} |
| + |
| +// Q7's loss half: the recovery's AdmissionEvidence accepts the writer's |
| +// loss substitution exactly: RemoteMaxTXID is the validated predecessor's |
| +// maximum and AdditionalWritesLost the other validated sources, from the |
| +// same LossSources call; anything else is refused. It journals the |
| +// lineage link and the source baseline before the CAS, as every recovery |
| +// admission does. |
| +func TestRecoveryAdmissionEvidenceAcceptsTheFrozenMaximum(t *testing.T) { |
| + ctx := context.Background() |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(false, []hostclaim.WriteSource{rcTargetSource(poOp, 20), rcCutSource()}, nil) |
| + o := f.read() |
| + in := f.selectAdmission(o) |
| + loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + // Without the validator's call for this observation, nothing vouches. |
| + if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("no validated sources: %v", err) |
| + } |
| + substituted := in.Admission |
| + substituted.RemoteMaxTXID = 30 |
| + substituted.AdditionalWritesLost = []hostclaim.WritesLost{{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}} |
| + for name, mutate := range map[string]func(*hostclaim.Admission){ |
| + "Select's maximum": func(a *hostclaim.Admission) { a.RemoteMaxTXID = in.Baseline.MaxTXID + 1 }, |
| + "no additional": func(a *hostclaim.Admission) { a.AdditionalWritesLost = nil }, |
| + "another additional maximum": func(a *hostclaim.Admission) { a.AdditionalWritesLost[0].RemoteMaxTXID = 30 }, |
| + "an extra additional": func(a *hostclaim.Admission) { |
| + a.AdditionalWritesLost = append(a.AdditionalWritesLost, hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 30}) |
| + }, |
| + "coverage": func(a *hostclaim.Admission) { a.Coverage = recoveryCoverageClean }, |
| + "a loss id": func(a *hostclaim.Admission) { a.LossID = rcLossID }, |
| + "kept": func(a *hostclaim.Admission) { a.KeepStamp = true }, |
| + } { |
| + if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); err != nil { |
| + t.Fatal(err) |
| + } |
| + bad := substituted |
| + bad.AdditionalWritesLost = append([]hostclaim.WritesLost(nil), substituted.AdditionalWritesLost...) |
| + mutate(&bad) |
| + if err := f.owners.AdmissionEvidence(ctx, o, bad, in.Baseline); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("%s: %v", name, err) |
| + } |
| + } |
| + // A validation for an earlier observation does not vouch for this one. |
| + if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); err != nil { |
| + t.Fatal(err) |
| + } |
| + stale := o |
| + stale.ETag = "another" |
| + if err := f.owners.AdmissionEvidence(ctx, stale, substituted, in.Baseline); !errors.Is(err, hostbox.ErrChanged) { |
| + t.Fatalf("another observation: %v", err) |
| + } |
| + // Through the claim's writer, with the moves loss owner: the |
| + // substitution passes, and the lineage and the source baseline are |
| + // journalled before the CAS. |
| + 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: poOp, Session: 20, RemoteMaxTXID: 30}) || |
| + !reflect.DeepEqual(a.AdditionalWritesLost, substituted.AdditionalWritesLost) || a.LossID != rcLossID { |
| + t.Fatalf("the loss admission: %+v", a) |
| + } |
| + j, err := moves.ReadJournal(f.dataDir, poHost, rpRB) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := j.Steps[rcStepLineage+a.Generation]; !ok { |
| + t.Fatalf("no lineage link for %s: %v", a.Generation, j.Steps) |
| + } |
| + if _, ok := j.Steps[rcStepSourceBaseline]; !ok { |
| + t.Fatalf("no source baseline: %v", j.Steps) |
| + } |
| + // The admitted session is the recovery's own. |
| + v, err := movesRecoveryOwners{f.owners}.recovery(ctx, admitted) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := (movesRecoveryOwners{f.owners}).lineage(poHost, v); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// Plan 4 Task 10 (witnesses wTwoSessionDeclaration, |
| +// 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. |
| +func TestLostTargetReturnDeclaresEverySessionOnce(t *testing.T) { |
| + ctx := context.Background() |
| + t.Run("two target sessions and the cut", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + sources := []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20), rcCutSource()} |
| + pub := f.declare(false, sources, func(c *hostclaim.Claim) { |
| + // The target's probe session closed and a restart died. |
| + 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"}) |
| + }) |
| + in := f.selectAdmission(f.read()) |
| + 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] |
| + want := []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 30}, {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: 30}) || !reflect.DeepEqual(a.AdditionalWritesLost, want) { |
| + t.Fatalf("declared %+v / %+v", a.WritesLost, a.AdditionalWritesLost) |
| + } |
| + }) |
| + t.Run("a target lost before its first admission", func(t *testing.T) { |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(false, []hostclaim.WriteSource{rcCutSource()}, func(c *hostclaim.Claim) { |
| + c.Holder.Session, c.Holder.WriterBox, c.Holder.Admissions = rcS, "ie-box", c.Holder.Admissions[:1] |
| + }) |
| + in := f.selectAdmission(f.read()) |
| + 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: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}) || len(a.AdditionalWritesLost) != 0 { |
| + t.Fatalf("the cut session, once: %+v / %+v", a.WritesLost, a.AdditionalWritesLost) |
| + } |
| + }) |
| + t.Run("a target lost before its first admission, the replica covered", func(t *testing.T) { |
| + // Nothing is frozen: the claim records its own predecessor at the |
| + // restore point, and moves vouches for an empty set. |
| + f := newRCFixture(t, rcClean) |
| + pub := f.declare(true, nil, func(c *hostclaim.Claim) { |
| + c.Holder.Session, c.Holder.WriterBox, c.Holder.Admissions = rcS, "ie-box", c.Holder.Admissions[:1] |
| + }) |
| + in := f.selectAdmission(f.read()) |
| + 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.LossID != rcLossID || len(a.AdditionalWritesLost) != 0 || (a.WritesLost != nil && a.WritesLost.RemoteMaxTXID != 30) { |
| + t.Fatalf("no frozen source: %+v", a) |
| + } |
| + }) |
| +} |