amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package hostrelocate
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "slices"
8 "testing"
9
10 "github.com/carlosframework/platform/internal/hostclaim"
11 "github.com/carlosframework/platform/internal/release"
12)
13
14// driveToProbing runs a drain until dst's participant is in probing with
15// its local install recorded and no attempt counted.
16func driveToProbing(t *testing.T, f *fixture, src, dst *fakeBox) (Participant, Participant) {
17 t.Helper()
18 ctx := context.Background()
19 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
20 t.Fatal(err)
21 }
22 ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
23 pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
24 for i := 0; i < 40; i++ {
25 if o := f.claim(); o.Claim.Op.Phase == hostclaim.PhaseRelocatingProbing {
26 return ps, pd
27 }
28 f.w.Step(ctx, host)
29 ps.Step(ctx, host)
30 if f.claim().Claim.Op.Phase == hostclaim.PhaseInstalled && f.claim().Claim.Op.Journal["local_install"] != "" {
31 pd.Step(ctx, host) // ToProbing
32 continue
33 }
34 pd.Step(ctx, host)
35 }
36 t.Fatalf("never probing: %+v", f.claim().Claim.Op)
37 return ps, pd
38}
39
40// Ruling 48: while root refuses the relocating kind, the participant
41// counts no attempt; once it answers, the next pass counts one.
42func TestParticipantCountsNoAttemptWhileRootRefusesTheKind(t *testing.T) {
43 f := newFixture(t)
44 ctx := context.Background()
45 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", preflightErr: errors.New("root: relocating unsupported")}
46 _, pd := driveToProbing(t, f, src, dst)
47 for i := 0; i < 3; i++ {
48 if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
49 t.Fatalf("pass %d: %v", i, err)
50 }
51 if n := f.claim().Claim.Op.Journal["probe_attempts.dst"]; n != "" {
52 t.Fatalf("an attempt was counted while root refused: %s", n)
53 }
54 }
55 dst.preflightErr = nil
56 if _, err := pd.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
57 t.Fatal(err)
58 }
59 if n := f.claim().Claim.Op.Journal["probe_attempts.dst"]; n != "1" {
60 t.Fatalf("attempts after root answered: %q", n)
61 }
62}
63
64// Ruling 58: once P's opening failed, an attempt root cannot launch is not
65// made; the operation goes probe-failed at once, with one attempt counted.
66func TestParticipantFailsAProbeRootCannotRelaunch(t *testing.T) {
67 f := newFixture(t)
68 ctx := context.Background()
69 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", failProbes: true}
70 _, pd := driveToProbing(t, f, src, dst)
71 if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
72 t.Fatalf("the first attempt: %v", err)
73 }
74 if !hostclaim.PreparedOpened(f.claim().Claim) {
75 t.Fatal("P was not opened by the first attempt")
76 }
77 dst.preflightErr = ErrLaterAttemptUnlaunchable
78 o, err := pd.Step(ctx, host)
79 if err != nil {
80 t.Fatal(err)
81 }
82 if o.Claim.Op.Phase != hostclaim.PhaseProbeFailed || o.Claim.Op.Journal["probe_attempts.dst"] != "1" || len(dst.certified) != 1 {
83 t.Fatalf("phase %s attempts %q admissions %d", o.Claim.Op.Phase, o.Claim.Op.Journal["probe_attempts.dst"], len(dst.certified))
84 }
85}
86
87// flakyStore is the claim store with one failure armed: the next Get (the
88// read after a passed probe) or Put (the CAS that records it) fails once,
89// as a transient store error would, and the store is itself again.
90type flakyStore struct {
91 release.Store
92 armed bool
93 puts bool // fail a Put rather than a Get
94}
95
96var errFlaky = errors.New("the claim store failed once")
97
98func (s *flakyStore) Get(ctx context.Context, key string) ([]byte, string, error) {
99 if s.armed && !s.puts {
100 s.armed = false
101 return nil, "", errFlaky
102 }
103 return s.Store.Get(ctx, key)
104}
105
106func (s *flakyStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
107 if s.armed && s.puts {
108 s.armed = false
109 return "", errFlaky
110 }
111 return s.Store.Put(ctx, key, body, ifMatch)
112}
113
114// Final review A2 (Astra 2): after Probe returned passed, the claim read
115// or the CAS that records the result fails once. The next pass recovers
116// the result from the box's durable attempt record, keyed by the admitted
117// session, and journals it: no Preflight, no second attempt, no
118// probe-failed, and the live session is the one journaled. A record that
119// cannot be read is retried, never taken as "no probe".
120func TestParticipantRecoversAPassedProbeTheClaimDidNotRecord(t *testing.T) {
121 for name, puts := range map[string]bool{"the claim read fails once": false, "the CAS fails once": true} {
122 t.Run(name, func(t *testing.T) {
123 f := newFixture(t)
124 ctx := context.Background()
125 flaky := &flakyStore{Store: f.s, puts: puts}
126 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", flaky: flaky}
127 _, pd := driveToProbing(t, f, src, dst)
128 pd.Claims = flaky
129 if _, err := pd.Step(ctx, host); !errors.Is(err, errFlaky) {
130 t.Fatalf("the pass after the probe passed: %v", err)
131 }
132 c := f.claim().Claim
133 if !hostclaim.PreparedOpened(c) || c.Op.Journal["probe"] != "" || flaky.armed {
134 t.Fatalf("after the failure: opened %v, probe %q, armed %v", hostclaim.PreparedOpened(c), c.Op.Journal["probe"], flaky.armed)
135 }
136 dst.preflightErr = ErrLaterAttemptUnlaunchable // what root would answer a new attempt
137 dst.effects = nil
138 dst.passedErr = errors.New("the attempt journal cannot be read")
139 if _, err := pd.Step(ctx, host); !errors.Is(err, dst.passedErr) {
140 t.Fatalf("an unreadable record: %v", err)
141 }
142 dst.passedErr = nil
143 o, err := pd.Step(ctx, host)
144 if err != nil {
145 t.Fatalf("the recovering pass: %v", err)
146 }
147 var p hostclaim.Probe
148 if o.Claim.Op.Phase != hostclaim.PhaseRelocatingProbing || json.Unmarshal([]byte(o.Claim.Op.Journal["probe"]), &p) != nil || p.Session != o.Claim.Holder.Session || p.Box != "dst" || p.At.IsZero() {
149 t.Fatalf("phase %s probe %q", o.Claim.Op.Phase, o.Claim.Op.Journal["probe"])
150 }
151 if len(dst.certified) != 1 || o.Claim.Op.Journal["probe_attempts.dst"] != "1" || slices.Contains(dst.effects, "preflight") {
152 t.Fatalf("admissions %d attempts %q effects %v", len(dst.certified), o.Claim.Op.Journal["probe_attempts.dst"], dst.effects)
153 }
154 })
155 }
156}
157
158// Ruling 62: an install that writes the claim (the typed inventory) has its
159// receipt recorded on a fresh read, not a conflict loop on the stale one.
160func TestParticipantRecordsItsReceiptAfterAnEffectWroteTheClaim(t *testing.T) {
161 f := newFixture(t)
162 ctx := context.Background()
163 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst", writesOnInstall: true}
164 if _, err := f.w.Start(ctx, host, Request{Reason: hostclaim.ReasonDrain, Source: "src", Target: "dst"}); err != nil {
165 t.Fatal(err)
166 }
167 ps := Participant{Claims: f.s, Box: "src", Enrollment: enrollment("src"), Effects: src, Now: f.clock}
168 pd := Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: dst, Now: f.clock}
169 for i := 0; i < 40 && f.claim().Claim.Op.Phase != hostclaim.PhaseInstalled; i++ {
170 f.w.Step(ctx, host)
171 ps.Step(ctx, host)
172 pd.Step(ctx, host)
173 }
174 if f.claim().Claim.Op.Phase != hostclaim.PhaseInstalled {
175 t.Fatalf("never installed: %+v", f.claim().Claim.Op)
176 }
177 if _, err := pd.Step(ctx, host); err != nil {
178 t.Fatalf("the receipt after an effect wrote the claim: %v", err)
179 }
180 c := f.claim().Claim
181 if c.Op.Journal["local_install"] == "" || c.Op.Journal["test.inventory"] != "dst" {
182 t.Fatalf("receipt %q inventory %q", c.Op.Journal["local_install"], c.Op.Journal["test.inventory"])
183 }
184}
185
186// passProbe runs dst's first attempt to a journaled passing probe.
187func passProbe(t *testing.T, f *fixture, pd Participant) {
188 t.Helper()
189 if _, err := pd.Step(context.Background(), host); err != nil {
190 t.Fatalf("the probe: %v", err)
191 }
192 if f.claim().Claim.Op.Journal["probe"] == "" {
193 t.Fatal("no probe journaled")
194 }
195}
196
197// Ruling 67's first scenario: the execution lease is lost right after
198// RecordProbe. The participant reopens a controller and ends nothing, so
199// no probe_stop is ever journaled by it (the worker's bound decides).
200func TestParticipantEndsNoProbeWithoutItsLease(t *testing.T) {
201 f := newFixture(t)
202 ctx := context.Background()
203 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
204 _, pd := driveToProbing(t, f, src, dst)
205 passProbe(t, f, pd)
206 dst.restarted = true // the lease is gone; Ready reports no controller
207 dst.effects = nil
208 if _, err := pd.Step(ctx, host); !errors.Is(err, ErrNotYet) {
209 t.Fatalf("a step without the lease: %v", err)
210 }
211 if slices.Contains(dst.effects, "end-probe") || f.claim().Claim.Op.Journal["probe_stop"] != "" {
212 t.Fatalf("ended the probe without a ready controller: %v", dst.effects)
213 }
214}
215
216// Ruling 67's second scenario: root keeps refusing the stop. The error is
217// returned and nothing is journaled; a later pass that gets the receipt
218// records it.
219func TestParticipantRecordsNoProbeStopWhileRootRefuses(t *testing.T) {
220 f := newFixture(t)
221 ctx := context.Background()
222 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
223 _, pd := driveToProbing(t, f, src, dst)
224 passProbe(t, f, pd)
225 refused := errors.New("root: stop refused")
226 dst.endProbeErr = refused
227 for i := 0; i < 2; i++ {
228 if _, err := pd.Step(ctx, host); !errors.Is(err, refused) || f.claim().Claim.Op.Journal["probe_stop"] != "" {
229 t.Fatalf("pass %d: %v", i, err)
230 }
231 }
232 dst.endProbeErr = nil
233 if _, err := pd.Step(ctx, host); err != nil || f.claim().Claim.Op.Journal["probe_stop"] == "" {
234 t.Fatalf("the stop once root answered: %v", err)
235 }
236}
237
238// Fix round 1, M4 (ruling 62): sameInstallation accepts a fresh read that
239// differs only in an effect's own bookkeeping, and refuses one whose
240// installation moved: another stamp, install stamp, prepared P, session
241// or phase, or a stuck mark. (The claim store itself refuses a changed
242// incarnation, or an installed op's executor moved off its target.)
243func TestSameInstallationRefusesAMovedInstallation(t *testing.T) {
244 ctx := context.Background()
245 installed := func(t *testing.T) (*fixture, hostclaim.Observed, Participant) {
246 f := newFixture(t)
247 releasedAndInstalling(t, f)
248 f.worker(ErrNotYet) // P is not built yet
249 p := f.claim().Claim.Op.Journal["prepared"]
250 f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.built"] = p; return nil })
251 if o := f.worker(nil); o.Claim.Op.Phase != hostclaim.PhaseInstalled {
252 t.Fatalf("not installed: %s", o.Claim.Op.Phase)
253 }
254 return f, f.claim(), Participant{Claims: f.s, Box: "dst", Enrollment: enrollment("dst"), Effects: &fakeBox{f: f, box: "dst"}, Now: f.clock}
255 }
256 t.Run("the effect's own bookkeeping", func(t *testing.T) {
257 f, obs, p := installed(t)
258 f.participant(func(c *hostclaim.Claim) error { c.Op.Journal["test.inventory"] = "dst"; return nil })
259 fresh, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled)
260 if err != nil || fresh.ETag == obs.ETag || fresh.Claim.Op.Journal["test.inventory"] != "dst" {
261 t.Fatalf("a fresh read of the same installation: %v", err)
262 }
263 })
264 for name, change := range map[string]func(*hostclaim.Claim){
265 "another stamp": func(c *hostclaim.Claim) { c.Holder.Stamp++ },
266 "another install stamp": func(c *hostclaim.Claim) { c.Op.Journal["install_stamp"] += "0" },
267 "another prepared P": func(c *hostclaim.Claim) { c.Op.Journal["prepared"] = "P9" },
268 "another session": func(c *hostclaim.Claim) { c.Holder.Session++ },
269 "stuck": func(c *hostclaim.Claim) { c.Op.Journal["stuck"] = "operator" },
270 } {
271 t.Run(name, func(t *testing.T) {
272 f, obs, p := installed(t)
273 f.participant(func(c *hostclaim.Claim) error { change(c); return nil })
274 if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseInstalled); !errors.Is(err, ErrRefused) || !errors.Is(err, ErrNotYet) {
275 t.Fatalf("a moved installation was accepted: %v", err)
276 }
277 })
278 }
279 t.Run("another phase", func(t *testing.T) {
280 _, obs, p := installed(t)
281 if _, err := p.sameInstallation(ctx, obs, hostclaim.PhaseRelocatingProbing); !errors.Is(err, ErrRefused) {
282 t.Fatalf("another phase was accepted: %v", err)
283 }
284 })
285}
286
287// Fix round 1, I3 and M1: a journaled release that is unreadable, or that
288// names another writer than the predecessor, is a claim defect. The
289// participant reads it before any attempt is counted: the operation is
290// stuck for an operator, no attempt is spent, and no probe is made.
291func TestParticipantRefusesADamagedReleaseBeforeCountingAnAttempt(t *testing.T) {
292 ctx := context.Background()
293 for name, damage := range map[string]func(*hostclaim.Claim) error{
294 "unparseable": func(c *hostclaim.Claim) error { c.Op.Journal["release"] = `{"released_by":`; return nil },
295 "another session": func(c *hostclaim.Claim) error {
296 var rel hostclaim.Release
297 if err := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel); err != nil {
298 return err
299 }
300 rel.Session++
301 raw, err := json.Marshal(rel)
302 c.Op.Journal["release"] = string(raw)
303 return err
304 },
305 } {
306 t.Run(name, func(t *testing.T) {
307 f := newFixture(t)
308 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
309 _, pd := driveToProbing(t, f, src, dst)
310 if _, err := hostclaim.Update(ctx, f.s, f.claim(), damage); err != nil {
311 t.Fatal(err)
312 }
313 if _, err := pd.Step(ctx, host); err != nil {
314 t.Fatal(err)
315 }
316 c := f.claim().Claim
317 if c.Op.Journal["stuck"] == "" || c.Op.Journal["probe_attempts.dst"] != "" || len(dst.certified) != 0 {
318 t.Fatalf("stuck %q attempts %q probes %d", c.Op.Journal["stuck"], c.Op.Journal["probe_attempts.dst"], len(dst.certified))
319 }
320 })
321 }
322 // A sound release certifies P's opening, and only it.
323 f := newFixture(t)
324 src, dst := &fakeBox{f: f, box: "src"}, &fakeBox{f: f, box: "dst"}
325 _, pd := driveToProbing(t, f, src, dst)
326 if certified, err := CertifiedFor(f.claim().Claim); err != nil || !certified {
327 t.Fatalf("a drain's release: %v %v", certified, err)
328 }
329 if _, err := pd.Step(ctx, host); err != nil && !errors.Is(err, ErrNotYet) {
330 t.Fatal(err)
331 }
332 if certified, err := CertifiedFor(f.claim().Claim); err != nil || certified {
333 t.Fatalf("after P's opening: %v %v", certified, err)
334 }
335}
336