amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/ed25519"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "reflect"
11 "strconv"
12 "strings"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/activator"
17 "github.com/carlosframework/platform/internal/genselect"
18 "github.com/carlosframework/platform/internal/hostbox"
19 "github.com/carlosframework/platform/internal/hostclaim"
20 "github.com/carlosframework/platform/internal/hostops"
21 "github.com/carlosframework/platform/internal/lease"
22 "github.com/carlosframework/platform/internal/moves"
23 "github.com/carlosframework/platform/internal/provision"
24 "github.com/carlosframework/platform/internal/registry"
25 "github.com/carlosframework/platform/internal/release"
26 "github.com/superfly/ltx"
27)
28
29// The loss and fence rules on the box (moves-after-Plan-12 Task 4: owed (a)
30// and (b), Plan 4 Task 10's box half; rulings Q5 as platform-1e answered
31// it, Q6, Q7's loss half, Q9's gate).
32
33const (
34 rcLossID = "loss_1"
35 // rcSealN is the cut seal's maximum: the stopped database's position,
36 // beyond the source replica's 30 when src_covered is false.
37 rcSealN = 35
38)
39
40// rcLossIntent is the moves recovery intent a declaration signs
41// (hostops.moveLossIntent's wire form).
42func (f *rcFixture) rcLossIntent(srcCovered bool) json.RawMessage {
43 raw, err := json.Marshal(map[string]any{"kind": "move-target-lost", "move": poOp, "incarnation": f.inc, "rollback": rpRB, "target": "uk-box", "source": "ie-box",
44 "target_region": "uk", "source_region": "ie", "record_key": "apps/bab/jam/instances/" + poHost + ".json", "record_etag": "etag-1", "record_seq": 1,
45 "record_state": "rolling_back", "dirty": false, "return_lane": false, "src_covered": srcCovered, "cut_session": rcS, "src_gen": "src"})
46 if err != nil {
47 f.t.Fatal(err)
48 }
49 return raw
50}
51
52// rcCutSeal is the cut's seal the claim journals.
53func rcCutSeal(srcCovered bool) moves.Seal {
54 at := time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC)
55 return moves.Seal{V: moves.SealVersion, N: rcSealN, FromDB: true, SrcCovered: srcCovered, SealedAt: at, LeaseUntil: at.Add(time.Minute),
56 Replica: []moves.SealedObject{{Key: "moves/" + poHost + "/" + poOp + "/replica/0009/0000000000000001-0000000000000023.ltx", VersionID: "s1", Size: 100}},
57 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}}}
58}
59
60// rcTargetSource is the frozen source of one target session since the cut.
61func rcTargetSource(gen string, session uint64) hostclaim.WriteSource {
62 return hostclaim.WriteSource{Reason: "target", Box: "uk-box", Generation: gen, Session: session, Region: "uk", Baseline: hostops.MoveLossReplicaBaseline + "uk/" + gen}
63}
64
65// rcCutSource is the frozen source of the source's cut session.
66func rcCutSource() hostclaim.WriteSource {
67 return hostclaim.WriteSource{Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: hostops.MoveLossSealBaseline + poOp}
68}
69
70// rcBeforeTarget is a claim whose target was lost before its first
71// admission: the holder is still the source's cut session.
72func rcBeforeTarget(c *hostclaim.Claim) {
73 c.Holder.Session, c.Holder.WriterBox, c.Holder.Admissions = rcS, "ie-box", c.Holder.Admissions[:1]
74}
75
76// declare turns f's clean return into the return after a declared-lost
77// target (Plan 4 Task 10): the claim as DeclareLost and the lost target's
78// release leave it, edit applied, the record's rollback citing the
79// declaration, and the root-signed declaration committed with sources. It
80// returns the root's public key.
81func (f *rcFixture) declare(srcCovered bool, sources []hostclaim.WriteSource, edit func(*hostclaim.Claim)) ed25519.PublicKey {
82 t := f.t
83 t.Helper()
84 c := f.recoveryClaim()
85 intent := f.rcLossIntent(srcCovered)
86 c.Lost, c.LossRefs, c.Op.Participants = []string{"uk-box"}, []string{rcLossID}, []string{"ie-box"}
87 c.Op.Journal["loss."+rcLossID] = string(intent)
88 c.Op.Journal["loss_gate."+rcLossID] = `{"box":"uk-box","revoke_execution":true}`
89 c.Op.Journal["lost_target_released"] = rcLossID
90 c.Op.Journal[hostclaim.JournalSeal] = rcCutSeal(srcCovered).Encode()
91 if edit != nil {
92 edit(c)
93 }
94 f.putClaim(c)
95 key, err := hostclaim.LossRecordKey(rcLossID)
96 if err != nil {
97 t.Fatal(err)
98 }
99 f.rec.Move.Rollback.LossRef, f.rec.Move.Rollback.Evidence = key, key
100 pub, priv, err := ed25519.GenerateKey(nil)
101 if err != nil {
102 t.Fatal(err)
103 }
104 now := time.Now().UTC()
105 r := hostclaim.LossRecord{V: 1, ID: rcLossID, State: "prepared", Actor: "operator", AcceptedAt: now, Host: poHost, HolderIncarnation: f.inc, HolderStamp: rcS, HolderSession: c.Holder.Session,
106 ExpectedClaimGen: c.Gen, ExpectedClaimETag: "claim-etag", OpID: poOp, OpKind: hostclaim.KindRestoring, OpPhase: hostclaim.PhaseReleasing, Box: "uk-box", Enrollment: "enroll", QuarantineRevision: 1,
107 CloudObservation: hostclaim.LossCloudObservation{ObservedAt: now, Account: "123", Region: "eu-west-2", InstanceID: "i-1", State: "terminated", RequestID: "request"},
108 OperatorAttestation: hostclaim.LossOperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true}, FencedAt: now,
109 ConfirmedConsequences: hostclaim.LossConsequences{RevokeExecution: true, RemoveParticipation: true, RecoveryIntent: intent},
110 Timers: hostclaim.TimerLoss{Knowledge: hostclaim.TimersUnknown}, WriteSources: sources}
111 created, err := hostclaim.CreateLossRecord(t.Context(), f.claims, release.RootSigner(priv), r)
112 if err != nil {
113 t.Fatal(err)
114 }
115 r.State, r.ResultClaimGen = "committed", c.Gen
116 if _, err := hostclaim.ReplaceLossRecord(t.Context(), f.claims, release.RootSigner(priv), created, r); err != nil {
117 t.Fatal(err)
118 }
119 return pub
120}
121
122// rcFenceEnrollment is the fenced target's enrollment.
123const rcFenceEnrollment = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
124
125// fence turns f's clean return into the return after the target's
126// termination fence (spec rev 7.20 §7.10): the consumed fence, the
127// automatic loss of the target's session 20 in poOp, the record's rollback
128// citing the fence. It returns the derived record the writer passes.
129func (f *rcFixture) fence() hostclaim.LossRecord {
130 t := f.t
131 t.Helper()
132 c := f.recoveryClaim()
133 ref := hostclaim.FenceRef{Enrollment: rcFenceEnrollment, Key: "control/box-fences/" + rcFenceEnrollment + ".json", SHA256: strings.Repeat("b", 64)}
134 raw, err := json.Marshal(ref)
135 if err != nil {
136 t.Fatal(err)
137 }
138 sum := sha256.Sum256([]byte(poOp + "\x00uk-box\x00" + ref.Enrollment + "\x00" + ref.SHA256))
139 id := "auto-" + hex.EncodeToString(sum[:16])
140 c.Op.Intent["target_enrollment"] = ref.Enrollment
141 c.Op.Journal["fence.uk-box"], c.Op.Journal["fenced_target_released"] = string(raw), string(raw)
142 c.Op.Journal[hostclaim.JournalSeal] = rcCutSeal(true).Encode()
143 c.Op.Participants, c.Lost, c.LossRefs = []string{"ie-box"}, []string{"uk-box"}, []string{id}
144 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),
145 Op: poOp, Kind: hostclaim.KindLanding, Phase: hostclaim.PhaseLandingProbing, Role: "target", Incarnation: f.inc, Stamp: 20, Session: 20, Generation: poOp, Timers: hostclaim.TimersUnknown}}
146 f.putClaim(c)
147 f.rec.Move.Rollback.LossRef, f.rec.Move.Rollback.Evidence = ref.Key, ref.Key
148 record, ok := hostclaim.PendingMoveFenceLoss(c)
149 if !ok {
150 t.Fatal("no pending fence loss")
151 }
152 return record
153}
154
155// lossWriter is the claim's writer with the moves loss owner and baseline
156// validator, over the source's own bucket.
157func (f *rcFixture) lossWriter(pub ed25519.PublicKey) hostbox.Writer {
158 return hostbox.Writer{Claims: f.claims, Data: f.bucket, Bucket: "data", Region: "ie", Box: "ie-box", AdmissionEvidence: f.owners.AdmissionEvidence,
159 Loss: &hostbox.LossAdmission{PublicKey: pub, Owner: hostops.MoveLossRecoveryOwner(), Sources: f.owners.LossSources}}
160}
161
162// Q6: a return after a declared loss is never covered: the probe's request
163// asks no coverage, the request matcher refuses clean-return for it, and the
164// recovery owners select and admit it uncovered.
165func TestDeclaredLossNeverAdmitsCleanReturn(t *testing.T) {
166 t.Run("the probing pass asks no coverage", func(t *testing.T) {
167 h := newRecoverHarness(t, false)
168 h.released(t)
169 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
170 h.srcPass(t)
171 h.sourceAt(t, provision.RbSourceProbing)
172 if len(h.probes) != 1 || h.probes[0].Coverage != recoveryCoverageClean || h.probes[0].Declared {
173 t.Fatalf("before the loss: %+v", h.probes)
174 }
175 key := rcDeclareOnHarness(t, h)
176 h.srcPass(t)
177 if len(h.probes) != 2 {
178 t.Fatalf("probes %+v", h.probes)
179 }
180 req := h.probes[1]
181 if !req.Declared || !req.Clean || req.Coverage != "" || req.Generation != "" || req.SourceGeneration == "" {
182 t.Fatalf("a declared return's request: %+v", req)
183 }
184 rec := h.read(t, p1Host)
185 if rec.Move.Rollback.LossRef != key || recoveryProbeMatches(rec, req) != nil {
186 t.Fatalf("the request does not match its record: %v", recoveryProbeMatches(rec, req))
187 }
188 for name, mutate := range map[string]func(*recoveryProbeRequest){
189 "clean-return": func(r *recoveryProbeRequest) { r.Coverage = recoveryCoverageClean },
190 "sealed": func(r *recoveryProbeRequest) { r.Coverage = recoveryCoverageSealed },
191 "not declared": func(r *recoveryProbeRequest) { r.Declared = false },
192 "undeclared kind": func(r *recoveryProbeRequest) { r.Declared, r.Coverage = false, recoveryCoverageClean },
193 "dirty": func(r *recoveryProbeRequest) { r.Clean, r.Generation = false, r.RollbackID },
194 } {
195 bad := req
196 mutate(&bad)
197 if err := recoveryProbeMatches(rec, bad); !errors.Is(err, errRecoveryProbeRequest) {
198 t.Fatalf("%s: %v", name, err)
199 }
200 }
201 // A request that says declared for a record that does not is refused.
202 undeclared := rec
203 rb := *rec.Move.Rollback
204 rb.LossRef = ""
205 mv := *rec.Move
206 mv.Rollback = &rb
207 undeclared.Move = &mv
208 if err := recoveryProbeMatches(undeclared, req); !errors.Is(err, errRecoveryProbeRequest) {
209 t.Fatalf("declared over an undeclared record: %v", err)
210 }
211 })
212 t.Run("the owners admit it uncovered, through the loss owner", func(t *testing.T) {
213 // The target was lost before its first admission: the cut session
214 // is the one frozen source (a target generation is refused, fix
215 // round 1).
216 f := newRCFixture(t, rcClean)
217 pub := f.declare(false, []hostclaim.WriteSource{rcCutSource()}, rcBeforeTarget)
218 o := f.read()
219 in := f.selectAdmission(o)
220 if in.Admission.Coverage != "" || in.Admission.LossID != "" {
221 t.Fatalf("admission %+v", in.Admission)
222 }
223 w := f.lossWriter(pub)
224 covered := in.Admission
225 covered.Coverage = recoveryCoverageClean
226 if _, err := w.Admit(t.Context(), o, covered, in.Baseline); err == nil {
227 t.Fatal("a declared return admitted clean-return")
228 }
229 admitted, err := w.Admit(t.Context(), f.read(), in.Admission, in.Baseline)
230 if err != nil {
231 t.Fatal(err)
232 }
233 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
234 if a.Coverage != "" || a.LossID != rcLossID || a.Covered || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) {
235 t.Fatalf("the loss admission: %+v", a)
236 }
237 })
238}
239
240// rcDeclareOnHarness declares the recover harness's target lost: the claim
241// journals the declaration and the record's rollback cites its key. It
242// returns the key.
243func rcDeclareOnHarness(t *testing.T, h *recoverHarness) string {
244 t.Helper()
245 key, err := hostclaim.LossRecordKey(rcLossID)
246 if err != nil {
247 t.Fatal(err)
248 }
249 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
250 c.LossRefs = append(c.LossRefs, rcLossID)
251 c.Op.Journal["loss."+rcLossID] = `{"kind":"move-target-lost","target":"uk-box"}`
252 })
253 rcSetLossRef(t, h, key)
254 return key
255}
256
257// rcSetLossRef writes the record's rollback LossRef through provision's
258// record writer.
259func rcSetLossRef(t *testing.T, h *recoverHarness, ref string) {
260 t.Helper()
261 if _, err := provision.SetRecordMove(t.Context(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {
262 r.Move.Rollback.LossRef = ref
263 return nil
264 }); err != nil {
265 t.Fatal(err)
266 }
267}
268
269// Q6: a return after the target's termination fence asks no coverage. Its
270// first admission declares the fenced target's generation, which the
271// restore (src_gen) does not contain, at 0: every write in it is lost
272// (platform-1e's loss-boundary rule).
273func TestFencedTargetReturnAsksUncovered(t *testing.T) {
274 t.Run("the probing pass asks no coverage", func(t *testing.T) {
275 h := newRecoverHarness(t, false)
276 h.released(t)
277 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
278 h.srcPass(t)
279 ref := hostclaim.FenceRef{Enrollment: rcFenceEnrollment, Key: "control/box-fences/" + rcFenceEnrollment + ".json", SHA256: strings.Repeat("b", 64)}
280 raw, _ := json.Marshal(ref)
281 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Journal["fence.uk-box"] = string(raw) })
282 rcSetLossRef(t, h, ref.Key)
283 h.srcPass(t)
284 if len(h.probes) != 2 || !h.probes[1].Declared || h.probes[1].Coverage != "" {
285 t.Fatalf("probes %+v", h.probes)
286 }
287 })
288 t.Run("the owners admit the frozen target at 0", func(t *testing.T) {
289 f := newRCFixture(t, rcClean)
290 f.fence()
291 o := f.read()
292 in := f.selectAdmission(o)
293 if in.Admission.Coverage != "" {
294 t.Fatalf("a fenced return asked %q", in.Admission.Coverage)
295 }
296 covered := in.Admission
297 covered.Coverage = recoveryCoverageClean
298 w := f.lossWriter(nil)
299 if _, err := w.Admit(t.Context(), o, covered, in.Baseline); err == nil {
300 t.Fatal("a fenced return admitted clean-return")
301 }
302 admitted, err := w.Admit(t.Context(), f.read(), in.Admission, in.Baseline)
303 if err != nil {
304 t.Fatal(err)
305 }
306 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
307 if a.Coverage != "" || a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}) || len(a.AdditionalWritesLost) != 0 {
308 t.Fatalf("the fenced admission: %+v", a)
309 }
310 })
311}
312
313// Q6: every later session of the recovered source stays uncovered: the
314// loss is consumed by the first admission, but the declaration (or the
315// fence) is still the op's, so a restart asks, selects and admits no
316// coverage, and the claim refuses any.
317func TestLaterSourceSessionStaysUncovered(t *testing.T) {
318 t.Run("the probing pass", func(t *testing.T) {
319 h := newRecoverHarness(t, false)
320 h.released(t)
321 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
322 h.srcPass(t)
323 rcDeclareOnHarness(t, h)
324 rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
325 c.Op.Journal["loss_admitted."+rcLossID] = strconv.FormatUint(c.Holder.Session, 10)
326 })
327 h.srcPass(t)
328 if len(h.probes) != 2 || !h.probes[1].Declared || h.probes[1].Coverage != "" {
329 t.Fatalf("probes %+v", h.probes)
330 }
331 })
332 for _, path := range []string{"declared", "fenced"} {
333 t.Run("the owners: "+path, func(t *testing.T) {
334 f := newRCFixture(t, rcClean)
335 var (
336 w hostbox.Writer
337 first hostclaim.Observed
338 in ClaimLaunchInputs
339 )
340 if path == "declared" {
341 w = f.lossWriter(f.declare(false, []hostclaim.WriteSource{rcCutSource()}, rcBeforeTarget))
342 in = f.selectAdmission(f.read())
343 var err error
344 if first, err = w.Admit(t.Context(), f.read(), in.Admission, in.Baseline); err != nil {
345 t.Fatal(err)
346 }
347 } else {
348 f.fence()
349 w = f.lossWriter(nil)
350 in = f.selectAdmission(f.read())
351 var err error
352 if first, err = w.Admit(t.Context(), f.read(), in.Admission, in.Baseline); err != nil {
353 t.Fatal(err)
354 }
355 }
356 if pending, err := recoveryLossPending(first.Claim); err != nil || pending {
357 t.Fatalf("the loss is still pending: %v %v", pending, err)
358 }
359 restart := f.selectAdmission(first)
360 if restart.Admission.Coverage != "" || restart.Admission.ObservedGeneration != in.Admission.NewGeneration {
361 t.Fatalf("the restart: %+v", restart.Admission)
362 }
363 covered := restart.Admission
364 covered.Coverage = recoveryCoverageClean
365 if err := f.owners.AdmissionEvidence(t.Context(), first, covered, restart.Baseline); !errors.Is(err, hostbox.ErrChanged) {
366 t.Fatalf("a later session admitted clean-return: %v", err)
367 }
368 again, err := w.Admit(t.Context(), first, restart.Admission, restart.Baseline)
369 if err != nil {
370 t.Fatal(err)
371 }
372 a := again.Claim.Holder.Admissions[len(again.Claim.Holder.Admissions)-1]
373 if a.Coverage != "" || a.LossID != "" || a.Generation != restart.Admission.NewGeneration {
374 t.Fatalf("the restart's admission: %+v", a)
375 }
376 })
377 }
378}
379
380// Q6: the record and the claim must agree on the target's loss, by
381// identity; until they do the probe waits: no request is made, the owners
382// select nothing, and the recovery stays at probing.
383func TestRecordAndClaimDisagreeWaits(t *testing.T) {
384 t.Run("the probing pass", func(t *testing.T) {
385 for name, disagree := range map[string]func(t *testing.T, h *recoverHarness){
386 "the record alone": func(t *testing.T, h *recoverHarness) { rcSetLossRef(t, h, "control/host-loss/"+rcLossID+".json") },
387 "the claim alone": func(t *testing.T, h *recoverHarness) { rcDeclareOnHarness(t, h); rcSetLossRef(t, h, "") },
388 "another loss": func(t *testing.T, h *recoverHarness) {
389 rcDeclareOnHarness(t, h)
390 rcSetLossRef(t, h, "control/host-loss/loss_2.json")
391 },
392 "a fence's for loss": func(t *testing.T, h *recoverHarness) {
393 rcDeclareOnHarness(t, h)
394 rcSetLossRef(t, h, "control/box-fences/x.json")
395 },
396 } {
397 t.Run(name, func(t *testing.T) {
398 h := newRecoverHarness(t, false)
399 h.released(t)
400 h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
401 h.srcPass(t)
402 disagree(t, h)
403 h.srcPass(t)
404 if len(h.probes) != 1 {
405 t.Fatalf("probed while the record and the claim disagree: %+v", h.probes)
406 }
407 h.sourceAt(t, provision.RbSourceProbing)
408 })
409 }
410 })
411 t.Run("the owners", func(t *testing.T) {
412 f := newRCFixture(t, rcClean)
413 f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil)
414 for name, ref := range map[string]string{"no loss": "", "another loss": "control/host-loss/loss_2.json"} {
415 f.rec.Move.Rollback.LossRef = ref
416 if _, err := f.owners.Select(t.Context(), f.route, f.read(), f.policy()); !errors.Is(err, hostbox.ErrChanged) {
417 t.Fatalf("%s: %v", name, err)
418 }
419 }
420 })
421 t.Run("the adapter", func(t *testing.T) {
422 f := newRecoveryFixture(t)
423 inc := plan10J2AdapterClaim(t, f, hostclaim.PhaseProbing, "src-gen")
424 rec := plan10J2AdapterRecord(inc, false)
425 var events []string
426 req := plan10J2AdapterRequest(&rec, "src-gen", &events)
427 a := recoveryProbeAdapter{Box: "box-1", Claims: true,
428 Wake: func(context.Context, registry.Route, activator.StartPolicy) error {
429 events = append(events, "wake")
430 return nil
431 },
432 Route: func(host string) (*registry.Route, error) { return &registry.Route{Host: host}, nil },
433 Current: func(context.Context, release.Store, provision.Record) (provision.MoveInputs, error) {
434 return plan10J2AdapterInputs(), nil
435 }}
436 // The claim journals a declaration the record does not cite.
437 if _, err := hostclaim.Update(t.Context(), f.claims, f.read(), func(c *hostclaim.Claim) error {
438 c.LossRefs = append(c.LossRefs, rcLossID)
439 c.Op.Journal["loss."+rcLossID] = `{"kind":"move-target-lost","target":"` + c.Op.Intent["target"] + `"}`
440 return nil
441 }); err != nil {
442 t.Fatal(err)
443 }
444 if out, err := a.probe(t.Context(), f.claims, rec, req); !errors.Is(err, hostbox.ErrChanged) || out != recoveryProbeWaiting || len(events) != 0 {
445 t.Fatalf("%v %v %v", out, err, events)
446 }
447 })
448}
449
450// Q5 as platform-1e answered it: the loss baseline validator vouches for
451// each frozen source from the maxima already recorded, never reading
452// another region's bucket: a seal's is the journalled cut seal's N; a
453// target replica's is the highest TXID of its own generation the restore
454// contains, 0 for a generation the restore (src_gen) does not contain
455// (platform-1e's loss-boundary rule). A malformed source is refused as not
456// the recovery's.
457func TestLossSourcesValidatesEachFrozenBaseline(t *testing.T) {
458 ctx := context.Background()
459 t.Run("a declared loss's replica and seal", func(t *testing.T) {
460 f := newRCFixture(t, rcClean)
461 sources := []hostclaim.WriteSource{rcTargetSource(poOp, 20), rcCutSource()}
462 pub := f.declare(false, sources, nil)
463 o := f.read()
464 in := f.selectAdmission(o)
465 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
466 if err != nil {
467 t.Fatal(err)
468 }
469 both, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline)
470 if err != nil {
471 t.Fatal(err)
472 }
473 if len(both) != 2 || both[0].RemoteMaxTXID != 0 || both[0].Generation != poOp || both[1].RemoteMaxTXID != rcSealN {
474 t.Fatalf("the target at 0 and the cut at N: %+v", both)
475 }
476 cutOnly := loss.Record
477 cutOnly.WriteSources = []hostclaim.WriteSource{rcCutSource()}
478 got, err := f.owners.LossSources(ctx, o, cutOnly, in.Admission, in.Baseline)
479 if err != nil {
480 t.Fatal(err)
481 }
482 want := []hostclaim.ValidatedWriteSource{{Reason: "source-cut", Box: "ie-box", Generation: "src", Session: rcS, Region: "ie", Baseline: "seal:" + poOp, RemoteMaxTXID: rcSealN}}
483 if !reflect.DeepEqual(got, want) {
484 t.Fatalf("validated %+v\nwant %+v", got, want)
485 }
486 for name, mutate := range map[string]func(*hostclaim.WriteSource){
487 "another target box": func(s *hostclaim.WriteSource) { s.Box = "de-box" },
488 "a replica of another gen": func(s *hostclaim.WriteSource) { s.Baseline = "replica:uk/other" },
489 "a replica of another region": func(s *hostclaim.WriteSource) { s.Baseline = "replica:de/" + poOp },
490 "a target outside its region": func(s *hostclaim.WriteSource) { s.Region, s.Baseline = "de", "replica:de/"+poOp },
491 "a session before the cut": func(s *hostclaim.WriteSource) { s.Session = rcS },
492 "a target named source-cut": func(s *hostclaim.WriteSource) { s.Reason = "source-cut" },
493 "a malformed baseline": func(s *hostclaim.WriteSource) { s.Baseline = "replica:uk" },
494 "an unknown baseline": func(s *hostclaim.WriteSource) { s.Baseline = "snapshot:x" },
495 } {
496 bad := loss.Record
497 bad.WriteSources = append([]hostclaim.WriteSource(nil), sources...)
498 mutate(&bad.WriteSources[0])
499 if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
500 t.Fatalf("target, %s: %v", name, err)
501 }
502 }
503 for name, mutate := range map[string]func(*hostclaim.WriteSource){
504 "another move's seal": func(s *hostclaim.WriteSource) { s.Baseline = "seal:mv_other" },
505 "another box": func(s *hostclaim.WriteSource) { s.Box = "uk-box" },
506 "another generation": func(s *hostclaim.WriteSource) { s.Generation = "other" },
507 "another session": func(s *hostclaim.WriteSource) { s.Session = rcS + 1 },
508 "another region": func(s *hostclaim.WriteSource) { s.Region = "uk" },
509 "a seal named target": func(s *hostclaim.WriteSource) { s.Reason = "target" },
510 "an unknown reason": func(s *hostclaim.WriteSource) { s.Reason = "other" },
511 "a replica for the cut seal": func(s *hostclaim.WriteSource) { s.Baseline = "replica:ie/src" },
512 } {
513 bad := loss.Record
514 bad.WriteSources = []hostclaim.WriteSource{rcCutSource()}
515 mutate(&bad.WriteSources[0])
516 if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
517 t.Fatalf("seal, %s: %v", name, err)
518 }
519 }
520 // A seal that covered the source's replica holds no source-cut
521 // writes: a source-cut source under it is not the cut's.
522 covered := newRCFixture(t, rcClean)
523 pub = covered.declare(true, []hostclaim.WriteSource{rcCutSource()}, nil)
524 o = covered.read()
525 in = covered.selectAdmission(o)
526 loss, err = hostclaim.ReadLossRecord(ctx, covered.claims, pub, rcLossID)
527 if err != nil {
528 t.Fatal(err)
529 }
530 if _, err := covered.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
531 t.Fatalf("a source-cut under a covering seal: %v", err)
532 }
533 })
534 t.Run("the record is the pending declaration", func(t *testing.T) {
535 f := newRCFixture(t, rcClean)
536 pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil)
537 o := f.read()
538 in := f.selectAdmission(o)
539 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
540 if err != nil {
541 t.Fatal(err)
542 }
543 for name, mutate := range map[string]func(*hostclaim.LossRecord){
544 "another id": func(r *hostclaim.LossRecord) { r.ID = "loss_2" },
545 "another op": func(r *hostclaim.LossRecord) { r.OpID = "mv_other" },
546 "another box": func(r *hostclaim.LossRecord) { r.Box = "de-box" },
547 "another incarnation": func(r *hostclaim.LossRecord) { r.HolderIncarnation = "other" },
548 } {
549 bad := loss.Record
550 mutate(&bad)
551 if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
552 t.Fatalf("%s: %v", name, err)
553 }
554 }
555 // The baseline must be the source baseline the recovery restores.
556 other := *in.Baseline
557 other.Generation = poOp
558 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, &other); !errors.Is(err, hostbox.ErrChanged) {
559 t.Fatalf("another restore point: %v", err)
560 }
561 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, nil); !errors.Is(err, hostbox.ErrChanged) {
562 t.Fatalf("no restore point: %v", err)
563 }
564 })
565 t.Run("the automatic fence record", func(t *testing.T) {
566 f := newRCFixture(t, rcClean)
567 record := f.fence()
568 o := f.read()
569 in := f.selectAdmission(o)
570 // Its one source is the fenced target's generation, outside the
571 // restore: 0.
572 got, err := f.owners.LossSources(ctx, o, record, in.Admission, in.Baseline)
573 want := []hostclaim.ValidatedWriteSource{{Reason: "target", Box: "uk-box", Generation: poOp, Session: 20, Region: "uk", Baseline: "replica:uk/" + poOp, RemoteMaxTXID: 0}}
574 if err != nil || !reflect.DeepEqual(got, want) {
575 t.Fatalf("the fenced target: %+v %v", got, err)
576 }
577 bad := record
578 bad.ID = "auto-other"
579 if _, err := f.owners.LossSources(ctx, o, bad, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
580 t.Fatalf("another automatic loss: %v", err)
581 }
582 })
583 t.Run("only a first recovery admission's loss", func(t *testing.T) {
584 f := newRCFixture(t, rcClean)
585 pub := f.declare(true, []hostclaim.WriteSource{rcTargetSource(poOp, 20)}, nil)
586 o := f.read()
587 in := f.selectAdmission(o)
588 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
589 if err != nil {
590 t.Fatal(err)
591 }
592 if _, err := f.owners.LossSources(ctx, hostclaim.Observed{Host: poHost, Absent: true}, loss.Record, in.Admission, in.Baseline); err == nil {
593 t.Fatal("no claim")
594 }
595 // No loss pending: the validator vouches for nothing.
596 undeclared := newRCFixture(t, rcClean)
597 uo := undeclared.read()
598 if _, err := undeclared.owners.LossSources(ctx, uo, loss.Record, in.Admission, in.Baseline); !errors.Is(err, hostclaim.ErrLossRecord) {
599 t.Fatalf("no pending loss: %v", err)
600 }
601 })
602}
603
604// failingList is the source's own bucket, unreadable for listings.
605type failingList struct{ genselect.Reader }
606
607func (failingList) ListVersions(context.Context, string, string) ([]lease.ObjectVersion, error) {
608 return nil, errors.New("SlowDown")
609}
610
611// The validator answers from the restore plan alone (platform-1e's ruling
612// on #236): it lists no bucket, so an unreadable own bucket changes
613// nothing; a restore plan whose selected files do not reach its maximum,
614// or that is not the source baseline, waits (hostbox.ErrChanged), which the
615// probe adapter turns into another pass.
616func TestLossSourcesReadOnlyTheRestorePlan(t *testing.T) {
617 ctx := context.Background()
618 f := newRCFixture(t, rcClean)
619 pub := f.declare(false, []hostclaim.WriteSource{rcCutSource()}, rcBeforeTarget)
620 o := f.read()
621 in := f.selectAdmission(o)
622 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
623 if err != nil {
624 t.Fatal(err)
625 }
626 f.owners.Data = failingList{f.owners.Data}
627 got, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline)
628 if err != nil || len(got) != 1 || got[0].RemoteMaxTXID != rcSealN {
629 t.Fatalf("with an unreadable bucket: %+v %v", got, err)
630 }
631 short := *in.Baseline
632 short.MaxTXID++
633 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, &short); !errors.Is(err, hostbox.ErrChanged) {
634 t.Fatalf("a plan whose files fall short of its maximum: %v", err)
635 }
636 bare := *in.Baseline
637 bare.Files, bare.Root = nil, nil
638 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, &bare); !errors.Is(err, hostbox.ErrChanged) {
639 t.Fatalf("a plan with only a maximum: %v", err)
640 }
641}
642
643// Q7's loss half: the recovery's AdmissionEvidence accepts the writer's
644// loss substitution exactly: RemoteMaxTXID is the validated predecessor's
645// maximum (here the cut session's frozen seal N, the target lost before its
646// first admission) and AdditionalWritesLost the other validated sources,
647// from the same LossSources call; anything else is refused. It journals the
648// lineage link and the source baseline before the CAS, as every recovery
649// admission does.
650func TestRecoveryAdmissionEvidenceAcceptsTheFrozenMaximum(t *testing.T) {
651 ctx := context.Background()
652 f := newRCFixture(t, rcClean)
653 pub := f.declare(false, []hostclaim.WriteSource{rcCutSource()}, rcBeforeTarget)
654 o := f.read()
655 in := f.selectAdmission(o)
656 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
657 if err != nil {
658 t.Fatal(err)
659 }
660 // Without the validator's call for this observation, nothing vouches.
661 if err := f.owners.AdmissionEvidence(ctx, o, in.Admission, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
662 t.Fatalf("no validated sources: %v", err)
663 }
664 substituted := in.Admission
665 substituted.RemoteMaxTXID = rcSealN
666 for name, mutate := range map[string]func(*hostclaim.Admission){
667 "Select's maximum": func(a *hostclaim.Admission) { a.RemoteMaxTXID = in.Baseline.MaxTXID },
668 "another maximum": func(a *hostclaim.Admission) { a.RemoteMaxTXID = rcSealN + 1 },
669 "an additional": func(a *hostclaim.Admission) {
670 a.AdditionalWritesLost = []hostclaim.WritesLost{{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}
671 },
672 "coverage": func(a *hostclaim.Admission) { a.Coverage = recoveryCoverageClean },
673 "a loss id": func(a *hostclaim.Admission) { a.LossID = rcLossID },
674 "kept": func(a *hostclaim.Admission) { a.KeepStamp = true },
675 } {
676 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); err != nil {
677 t.Fatal(err)
678 }
679 bad := substituted
680 bad.AdditionalWritesLost = append([]hostclaim.WritesLost(nil), substituted.AdditionalWritesLost...)
681 mutate(&bad)
682 if err := f.owners.AdmissionEvidence(ctx, o, bad, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
683 t.Fatalf("%s: %v", name, err)
684 }
685 }
686 // A validation for an earlier observation does not vouch for this one.
687 if _, err := f.owners.LossSources(ctx, o, loss.Record, in.Admission, in.Baseline); err != nil {
688 t.Fatal(err)
689 }
690 stale := o
691 stale.ETag = "another"
692 if err := f.owners.AdmissionEvidence(ctx, stale, substituted, in.Baseline); !errors.Is(err, hostbox.ErrChanged) {
693 t.Fatalf("another observation: %v", err)
694 }
695 // Through the claim's writer, with the moves loss owner: the
696 // substitution passes, and the lineage and the source baseline are
697 // journalled before the CAS.
698 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
699 if err != nil {
700 t.Fatal(err)
701 }
702 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
703 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}) ||
704 len(a.AdditionalWritesLost) != 0 || a.LossID != rcLossID {
705 t.Fatalf("the loss admission: %+v", a)
706 }
707 j, err := moves.ReadJournal(f.dataDir, poHost, rpRB)
708 if err != nil {
709 t.Fatal(err)
710 }
711 if _, ok := j.Steps[rcStepLineage+a.Generation]; !ok {
712 t.Fatalf("no lineage link for %s: %v", a.Generation, j.Steps)
713 }
714 if _, ok := j.Steps[rcStepSourceBaseline]; !ok {
715 t.Fatalf("no source baseline: %v", j.Steps)
716 }
717 // The admitted session is the recovery's own.
718 v, err := movesRecoveryOwners{f.owners}.recovery(ctx, admitted)
719 if err != nil {
720 t.Fatal(err)
721 }
722 if err := (movesRecoveryOwners{f.owners}).lineage(poHost, v); err != nil {
723 t.Fatal(err)
724 }
725}
726
727// Plan 4 Task 10 (witnesses wTwoSessionDeclaration,
728// wNoDuplicateDeclarationEarly/Late): behind a lost target the source's
729// recovery declares every target session since the cut and, when the seal
730// did not cover the source's replica, the cut session, each once: the
731// predecessor's as writes_lost, the rest as additional_writes_lost. Each
732// tuple's maximum is the highest TXID of its own generation the restore
733// contains (platform-1e's loss-boundary rule): 0 for a target generation
734// outside it, the journalled seal's N for the cut session.
735func TestLostTargetReturnDeclaresEverySessionOnce(t *testing.T) {
736 ctx := context.Background()
737 t.Run("two target sessions and the cut", func(t *testing.T) {
738 f := newRCFixture(t, rcClean)
739 sources := []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20), rcCutSource()}
740 pub := f.declare(false, sources, func(c *hostclaim.Claim) {
741 // The target's probe session closed and a restart died.
742 c.Holder.Session = 25
743 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
744 })
745 in := f.selectAdmission(f.read())
746 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
747 if err != nil {
748 t.Fatal(err)
749 }
750 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
751 want := []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}, {Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}
752 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "t2", Session: 25, RemoteMaxTXID: 0}) || !reflect.DeepEqual(a.AdditionalWritesLost, want) {
753 t.Fatalf("declared %+v / %+v", a.WritesLost, a.AdditionalWritesLost)
754 }
755 })
756 t.Run("a lost tuple in the restore's own generation", func(t *testing.T) {
757 // The target's session wrote into src itself: the restore contains
758 // that generation, so its maximum is GenerationMax(src), 30.
759 f := newRCFixture(t, rcClean)
760 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) {
761 c.Holder.Admissions[1].Generation, c.Holder.Admissions[1].Predecessor = "src", "src"
762 })
763 in := f.selectAdmission(f.read())
764 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
765 if err != nil {
766 t.Fatal(err)
767 }
768 got, err := f.owners.LossSources(ctx, f.read(), loss.Record, in.Admission, in.Baseline)
769 if err != nil || len(got) != 1 || got[0].RemoteMaxTXID != 30 || in.Baseline.Generation != "src" {
770 t.Fatalf("validated %+v %v", got, err)
771 }
772 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
773 if err != nil {
774 t.Fatal(err)
775 }
776 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
777 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "src", Session: 20, RemoteMaxTXID: 30}) {
778 t.Fatalf("declared %+v", a.WritesLost)
779 }
780 })
781 t.Run("a target lost before its first admission", func(t *testing.T) {
782 f := newRCFixture(t, rcClean)
783 pub := f.declare(false, []hostclaim.WriteSource{rcCutSource()}, rcBeforeTarget)
784 in := f.selectAdmission(f.read())
785 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
786 if err != nil {
787 t.Fatal(err)
788 }
789 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
790 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}) || len(a.AdditionalWritesLost) != 0 {
791 t.Fatalf("the cut session, once: %+v / %+v", a.WritesLost, a.AdditionalWritesLost)
792 }
793 })
794 t.Run("a target lost before its first admission, the replica covered", func(t *testing.T) {
795 // Nothing is frozen: the claim records its own predecessor at the
796 // restore point, and moves vouches for an empty set.
797 f := newRCFixture(t, rcClean)
798 pub := f.declare(true, nil, func(c *hostclaim.Claim) {
799 c.Holder.Session, c.Holder.WriterBox, c.Holder.Admissions = rcS, "ie-box", c.Holder.Admissions[:1]
800 })
801 in := f.selectAdmission(f.read())
802 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
803 if err != nil {
804 t.Fatal(err)
805 }
806 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
807 if a.LossID != rcLossID || len(a.AdditionalWritesLost) != 0 || (a.WritesLost != nil && a.WritesLost.RemoteMaxTXID != 30) {
808 t.Fatalf("no frozen source: %+v", a)
809 }
810 })
811}
812
813// The loss-boundary rule (platform-1e): a fresh target generation restarts
814// its TXIDs at [1, 1] (spec §5.1), so the source's maximum, 30, is no loss
815// boundary for it: t2 acknowledged a write at TXID 2 that the restore of
816// src omits. A lost tuple's maximum is the highest TXID of its own
817// generation the restore contains: 0 here, so every write of t2 (and of
818// mv's generation) is declared lost. The validated source carries the same
819// 0, which the claim compares (lost.go:194).
820func TestLostTargetGenerationOutsideTheRestoreDeclaresAllItsWrites(t *testing.T) {
821 ctx := context.Background()
822 f := newRCFixture(t, rcClean)
823 // The target's own bucket, which the source never reads: t2 restarted
824 // at [1, 1] and holds TXID 2.
825 f.bucket.AddBucket("data-uk")
826 f.bucket.Put("data-uk", "instances/"+poHost+"/g-t2/0000/"+ltx.FormatFilename(1, 2), f.ltxHeader(1, 2))
827 sources := []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20)}
828 pub := f.declare(true, sources, func(c *hostclaim.Claim) {
829 c.Holder.Session = 25
830 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
831 })
832 in := f.selectAdmission(f.read())
833 loss, err := hostclaim.ReadLossRecord(ctx, f.claims, pub, rcLossID)
834 if err != nil {
835 t.Fatal(err)
836 }
837 validated, err := f.owners.LossSources(ctx, f.read(), loss.Record, in.Admission, in.Baseline)
838 if err != nil || len(validated) != 2 || validated[0].Generation != "t2" || validated[0].RemoteMaxTXID != 0 || validated[1].RemoteMaxTXID != 0 {
839 t.Fatalf("validated %+v %v", validated, err)
840 }
841 admitted, err := f.lossWriter(pub).Admit(ctx, f.read(), in.Admission, in.Baseline)
842 if err != nil {
843 t.Fatal(err)
844 }
845 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
846 if a.WritesLost == nil || *a.WritesLost != (hostclaim.WritesLost{Box: "uk-box", Generation: "t2", Session: 25, RemoteMaxTXID: 0}) ||
847 !reflect.DeepEqual(a.AdditionalWritesLost, []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}}) || !hostclaim.OperationLossConsumed(admitted.Claim, rcLossID) {
848 t.Fatalf("declared %+v / %+v", a.WritesLost, a.AdditionalWritesLost)
849 }
850}
851
852// Item 4 of platform-1e's review: the loss admission end to end, through
853// the real writer, the moves validator, the moves loss owner and the
854// claim's AdmitWriterWithLoss or AdmitWriterWithFencedTargetLoss. Every
855// target tuple the restore omits is 0 in the validated evidence and in the
856// persisted entry, every tuple is kept, the cut session is the seal's N; a
857// replay declares nothing again, and an omitted or duplicated tuple is
858// refused with the claim unmoved.
859func TestLossAdmissionEndToEndThroughTheWriter(t *testing.T) {
860 ctx := context.Background()
861 twoSessions := func(c *hostclaim.Claim) {
862 c.Holder.Session = 25
863 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Session: 25, Box: "uk-box", Generation: "t2", Predecessor: poOp, Region: "uk", PredecessorRegion: "uk", Mode: "cold"})
864 }
865 type setup struct {
866 w hostbox.Writer
867 // want are the validated maxima by generation; primary and more the
868 // persisted entries.
869 want map[string]uint64
870 primary hostclaim.WritesLost
871 more []hostclaim.WritesLost
872 }
873 paths := map[string]func(f *rcFixture) setup{
874 "declared": func(f *rcFixture) setup {
875 pub := f.declare(false, []hostclaim.WriteSource{rcTargetSource("t2", 25), rcTargetSource(poOp, 20), rcCutSource()}, twoSessions)
876 return setup{w: f.lossWriter(pub), want: map[string]uint64{"t2": 0, poOp: 0, "src": rcSealN},
877 primary: hostclaim.WritesLost{Box: "uk-box", Generation: "t2", Session: 25, RemoteMaxTXID: 0},
878 more: []hostclaim.WritesLost{{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}, {Box: "ie-box", Generation: "src", Session: rcS, RemoteMaxTXID: rcSealN}}}
879 },
880 "fenced": func(f *rcFixture) setup {
881 f.fence()
882 return setup{w: f.lossWriter(nil), want: map[string]uint64{poOp: 0}, primary: hostclaim.WritesLost{Box: "uk-box", Generation: poOp, Session: 20, RemoteMaxTXID: 0}}
883 },
884 }
885 // recording wraps the writer's loss sources: it keeps what the moves
886 // validator answered and lets a case alter what the writer receives.
887 recording := func(w *hostbox.Writer, alter func([]hostclaim.ValidatedWriteSource) []hostclaim.ValidatedWriteSource) *[]hostclaim.ValidatedWriteSource {
888 var got []hostclaim.ValidatedWriteSource
889 inner := w.Loss.Sources
890 w.Loss.Sources = func(ctx context.Context, o hostclaim.Observed, l hostclaim.LossRecord, in hostclaim.Admission, p *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) {
891 v, err := inner(ctx, o, l, in, p)
892 got = append([]hostclaim.ValidatedWriteSource(nil), v...)
893 if err == nil && alter != nil {
894 v = alter(v)
895 }
896 return v, err
897 }
898 return &got
899 }
900 for name, build := range paths {
901 t.Run(name+": admitted, zero in the evidence and the entry", func(t *testing.T) {
902 f := newRCFixture(t, rcClean)
903 st := build(f)
904 validated := recording(&st.w, nil)
905 o := f.read()
906 in := f.selectAdmission(o)
907 admitted, err := st.w.Admit(ctx, o, in.Admission, in.Baseline)
908 if err != nil {
909 t.Fatal(err)
910 }
911 if len(*validated) != len(st.want) {
912 t.Fatalf("validated %+v", *validated)
913 }
914 for _, v := range *validated {
915 if want, ok := st.want[v.Generation]; !ok || v.RemoteMaxTXID != want {
916 t.Fatalf("validated %s at %d, want %d", v.Generation, v.RemoteMaxTXID, want)
917 }
918 }
919 a := admitted.Claim.Holder.Admissions[len(admitted.Claim.Holder.Admissions)-1]
920 if a.WritesLost == nil || *a.WritesLost != st.primary || len(a.AdditionalWritesLost) != len(st.more) || (len(st.more) != 0 && !reflect.DeepEqual(a.AdditionalWritesLost, st.more)) {
921 t.Fatalf("persisted %+v / %+v", a.WritesLost, a.AdditionalWritesLost)
922 }
923 // Replay: the same admission again, over the observation it was
924 // made on and over the claim it made, declares nothing again.
925 after := f.read()
926 if _, err := st.w.Admit(ctx, o, in.Admission, in.Baseline); err == nil {
927 t.Fatal("a replay over the stale observation was admitted")
928 }
929 if _, err := st.w.Admit(ctx, after, in.Admission, in.Baseline); err == nil {
930 t.Fatal("a replay over the admitted claim was admitted")
931 }
932 again := f.read()
933 declared := 0
934 for _, e := range again.Claim.Holder.Admissions {
935 if e.WritesLost != nil {
936 declared++
937 }
938 }
939 if again.ETag != after.ETag || declared != 1 {
940 t.Fatalf("the replay moved the claim: %d declaring admissions", declared)
941 }
942 })
943 for refusal, alter := range map[string]func([]hostclaim.ValidatedWriteSource) []hostclaim.ValidatedWriteSource{
944 "omission": func(v []hostclaim.ValidatedWriteSource) []hostclaim.ValidatedWriteSource { return v[:len(v)-1] },
945 "duplication": func(v []hostclaim.ValidatedWriteSource) []hostclaim.ValidatedWriteSource { return append(v, v[0]) },
946 } {
947 t.Run(name+": "+refusal+" refused", func(t *testing.T) {
948 f := newRCFixture(t, rcClean)
949 st := build(f)
950 recording(&st.w, alter)
951 o := f.read()
952 in := f.selectAdmission(o)
953 if _, err := st.w.Admit(ctx, o, in.Admission, in.Baseline); err == nil {
954 t.Fatalf("an admission with an %s was accepted", refusal)
955 }
956 if after := f.read(); after.ETag != o.ETag {
957 t.Fatal("the claim moved")
958 }
959 })
960 }
961 }
962}
963