amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package park
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "testing"
8 "time"
9
10 "github.com/carlosframework/platform/internal/hostclaim"
11)
12
13func coverageFixture(t *testing.T, foreign bool) *fixture {
14 t.Helper()
15 f := newFixture(t)
16 allowFixtureMoveSweep(f)
17 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
18 if err != nil {
19 t.Fatal(err)
20 }
21 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
22 c.Obligations = nil
23 c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
24 c.Holder.Generation = "live"
25 c.Holder.Session = 3
26 c.Holder.Admissions = []hostclaim.AdmissionRecord{
27 {Region: "uk", PredecessorRegion: "ie", Predecessor: "dead"},
28 {Region: "ie", PredecessorRegion: "uk", Predecessor: "dead"},
29 {Region: "ie", PredecessorRegion: "ie", Predecessor: "unseen"},
30 {Region: "ie", PredecessorRegion: "ie", Predecessor: "stage-only"},
31 {Region: "ie", PredecessorRegion: "ie", Predecessor: "old-only"},
32 {Region: "uk", PredecessorRegion: "ie", Predecessor: ""},
33 }
34 return nil
35 })
36 if err != nil {
37 t.Fatal(err)
38 }
39 base := f.ob.Park.Location + "/"
40 for _, key := range []string{
41 base + "dead/data", base + "0000/data",
42 "moves/" + f.w.Host + "/stage-only/data",
43 "instances/" + f.w.Host + "/" + strings.Repeat("a", 64) + "/old-only/data",
44 } {
45 f.b.Put("data", key, []byte(key))
46 }
47 f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
48 authority := "ie"
49 if foreign {
50 authority = "uk"
51 }
52 return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: authority,
53 MarkerKnown: true, MarkerGeneration: "live", GenerationOf: func(r Ref) (string, bool) {
54 parts := strings.Split(r.Key, "/")
55 g := parts[len(parts)-2]
56 if g == "0000" {
57 g = ""
58 }
59 return g, true
60 }}, nil
61 }
62 return f
63}
64
65func TestSweepQueueCoversOnlyQueuedPredecessors(t *testing.T) {
66 for _, foreign := range []bool{false, true} {
67 name := "generation"
68 if foreign {
69 name = "replica"
70 }
71 t.Run(name, func(t *testing.T) {
72 f := coverageFixture(t, foreign)
73 before := f.b.Snapshot()
74 f.held.failed = true
75 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
76 t.Fatal(err)
77 }
78 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
79 if err != nil {
80 t.Fatal(err)
81 }
82 if len(obs.Claim.Obligations) == 0 || f.b.Snapshot() != before {
83 t.Fatal("queue boundary did not preserve pending work and storage")
84 }
85 for i, a := range obs.Claim.Holder.Admissions {
86 if want := i == 0 || i == 5; a.Covered != want {
87 t.Errorf("queue admission %d covered=%v, want %v", i, a.Covered, want)
88 }
89 }
90 })
91 }
92}
93
94func TestSweepAcknowledgementCoversOnlyRetiredPredecessors(t *testing.T) {
95 for _, scenario := range []string{"generation", "replica", "empty", "old-incarnation", "staging"} {
96 t.Run(scenario, func(t *testing.T) {
97 f := coverageFixture(t, scenario == "replica")
98 f.held.failed = true
99 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
100 t.Fatal(err)
101 }
102 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
103 if err != nil {
104 t.Fatal(err)
105 }
106 var chore hostclaim.Obligation
107 for _, o := range obs.Claim.Obligations {
108 match := o.Park.Location == f.ob.Park.Location && (o.Generation == "dead" || scenario == "replica")
109 if scenario == "old-incarnation" {
110 match = strings.HasSuffix(o.Park.Location, strings.Repeat("a", 64))
111 } else if scenario == "staging" {
112 match = strings.HasPrefix(o.Park.Location, "moves/")
113 }
114 if match {
115 chore = o
116 break
117 }
118 }
119 if chore.ID == "" {
120 t.Fatal("target chore was not queued")
121 }
122 if scenario == "empty" {
123 versions, err := f.e.list(t.Context(), f.w.Host, chore.Park.Location)
124 if err != nil {
125 t.Fatal(err)
126 }
127 for _, x := range versions {
128 if strings.Contains(x.Key, "/dead/") {
129 if err := f.b.DeleteVersion(t.Context(), "data", x.Key, x.VersionID); err != nil {
130 t.Fatal(err)
131 }
132 }
133 }
134 }
135 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
136 later := append([]hostclaim.AdmissionRecord(nil), c.Holder.Admissions...)
137 for i := range later {
138 later[i].Covered = false
139 }
140 c.Holder.Admissions = append(c.Holder.Admissions, later...)
141 return nil
142 })
143 if err != nil {
144 t.Fatal(err)
145 }
146 f.held.failed = false
147 f.e.After = func(step string) error {
148 if step == "before-completion" {
149 return ErrPending
150 }
151 return nil
152 }
153 w := Work{f.w.Host, chore.ID, chore.Op}
154 if err := f.e.Run(t.Context(), w); !errors.Is(err, ErrPending) {
155 t.Fatal(err)
156 }
157 pending, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
158 if err != nil {
159 t.Fatal(err)
160 }
161 for _, a := range pending.Claim.Holder.Admissions[6:] {
162 if a.Covered {
163 t.Fatal("later admission covered before acknowledgement")
164 }
165 }
166 f.e.After = nil
167 f.b.FailOn = func(op, key string) error {
168 if op == "copy" {
169 t.Error("completion retry copied again")
170 return ErrUnsafe
171 }
172 return nil
173 }
174 if err := f.e.Run(t.Context(), w); err != nil {
175 t.Fatal(err)
176 }
177 completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
178 if err != nil {
179 t.Fatal(err)
180 }
181 if _, present := find(completed.Claim, w); present || f.held.released != 1 || completed.Claim.QuietAfter["ie"].IsZero() {
182 t.Fatal("retirement was not acknowledged")
183 }
184 for i, a := range completed.Claim.Holder.Admissions[6:] {
185 want := (i == 0 && (scenario == "generation" || scenario == "replica")) || (i == 5 && scenario == "replica")
186 if a.Covered != want {
187 t.Errorf("ack admission %d covered=%v, want %v", i, a.Covered, want)
188 }
189 }
190 })
191 }
192}
193
194func TestSweepCoverageSurvivesRetirementRotation(t *testing.T) {
195 f := coverageFixture(t, true)
196 obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
197 if err != nil {
198 t.Fatal(err)
199 }
200 _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
201 c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Predecessor: "late"})
202 return nil
203 })
204 if err != nil {
205 t.Fatal(err)
206 }
207 f.held.failed = true
208 if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
209 t.Fatal(err)
210 }
211 obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
212 if err != nil {
213 t.Fatal(err)
214 }
215 var w Work
216 for _, o := range obs.Claim.Obligations {
217 if o.Park.Location == f.ob.Park.Location {
218 w = Work{f.w.Host, o.ID, o.Op}
219 }
220 }
221 if w.ID == "" || obs.Claim.Holder.Admissions[6].Covered {
222 t.Fatal("late generation was queued or covered without an object")
223 }
224 f.held.failed = false
225 for _, next := range []string{"late", "last"} {
226 f.e.After = func(step string) error {
227 if step == "manifest" {
228 f.b.Put("data", f.ob.Park.Location+"/"+next+"/data", []byte(next))
229 }
230 return nil
231 }
232 if err := f.e.Run(t.Context(), w); !errors.Is(err, ErrPending) {
233 t.Fatalf("retirement rotation: %v", err)
234 }
235 }
236 f.e.After = nil
237 if err := f.e.Run(t.Context(), w); err != nil {
238 t.Fatal(err)
239 }
240 completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
241 if err != nil {
242 t.Fatal(err)
243 }
244 if !completed.Claim.Holder.Admissions[6].Covered || completed.Claim.Holder.Admissions[2].Covered {
245 t.Fatalf("coverage lost a retired generation or covered unseen data: %+v", completed.Claim.Holder.Admissions)
246 }
247 if _, present := find(completed.Claim, w); present {
248 t.Fatal("rotated retirement did not complete")
249 }
250}
251