amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package boxvolumes
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "testing"
8 "time"
9
10 "github.com/aws/aws-sdk-go-v2/aws"
11 awsmiddleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
12 "github.com/aws/aws-sdk-go-v2/service/ec2"
13 "github.com/aws/aws-sdk-go-v2/service/ec2/types"
14 smithymiddleware "github.com/aws/smithy-go/middleware"
15
16 "github.com/carlosframework/platform/internal/boxid"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20var (
21 cloud = boxid.Cloud{Provider: "aws", Account: "111122223333", Region: "eu-west-1", InstanceID: "i-0abc"}
22 entry = Entry{Box: "annie-welding", Enrollment: "enr-annie-1", Cloud: cloud}
23 writer = WrittenBy{Box: "bert-sawing", Enrollment: "enr-bert-1"}
24 t0 = time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
25)
26
27type vol struct {
28 dev, id string
29 dot bool
30}
31
32func reply(owner string, state types.InstanceStateName, vols ...vol) *ec2.DescribeInstancesOutput {
33 var maps []types.InstanceBlockDeviceMapping
34 for _, v := range vols {
35 maps = append(maps, types.InstanceBlockDeviceMapping{DeviceName: aws.String(v.dev),
36 Ebs: &types.EbsInstanceBlockDevice{VolumeId: aws.String(v.id), DeleteOnTermination: aws.Bool(v.dot), Status: types.AttachmentStatusAttached}})
37 }
38 var md smithymiddleware.Metadata
39 awsmiddleware.SetRequestIDMetadata(&md, "req-1")
40 return &ec2.DescribeInstancesOutput{ResultMetadata: md, Reservations: []types.Reservation{{OwnerId: aws.String(owner),
41 Instances: []types.Instance{{InstanceId: aws.String("i-0abc"), State: &types.InstanceState{Name: state}, BlockDeviceMappings: maps}}}}}
42}
43
44var root, data, scratch = vol{"/dev/xvda", "vol-root", true}, vol{"/dev/xvdb", "vol-data", true}, vol{"/dev/xvdc", "vol-keep", false}
45
46func observe(t *testing.T, out *ec2.DescribeInstancesOutput) Observation {
47 t.Helper()
48 o, err := Observe(out, cloud)
49 if err != nil {
50 t.Fatal(err)
51 }
52 return o
53}
54
55func TestObserveReadsEBSMappingsInVolumeOrder(t *testing.T) {
56 out := reply("111122223333", types.InstanceStateNameRunning, root, data)
57 out.Reservations[0].Instances[0].BlockDeviceMappings = append(out.Reservations[0].Instances[0].BlockDeviceMappings,
58 types.InstanceBlockDeviceMapping{DeviceName: aws.String("/dev/sdz")}) // instance store: no Ebs
59 o := observe(t, out)
60 want := []Device{{"/dev/xvdb", "vol-data", true, "attached"}, {"/dev/xvda", "vol-root", true, "attached"}}
61 if o.State != "running" || o.RequestID != "req-1" || len(o.Devices) != 2 || o.Devices[0] != want[0] || o.Devices[1] != want[1] {
62 t.Fatalf("observation = %+v", o)
63 }
64}
65
66// Review Focus 3: a terminated instance's reply has no mappings. Recorded,
67// it would claim every volume was removed; it must not be recordable.
68func TestObserveRefusesWhatIsNotTheInstancesMapping(t *testing.T) {
69 for name, out := range map[string]*ec2.DescribeInstancesOutput{
70 "terminated": reply("111122223333", types.InstanceStateNameTerminated),
71 "another owner": reply("999999999999", types.InstanceStateNameRunning, root),
72 "another instance": {Reservations: []types.Reservation{{OwnerId: aws.String("111122223333"), Instances: []types.Instance{{InstanceId: aws.String("i-0zzz")}}}}},
73 "empty reply": {},
74 "nil reply": nil,
75 } {
76 t.Run(name, func(t *testing.T) {
77 if o, err := Observe(out, cloud); !errors.Is(err, ErrNotRecordable) {
78 t.Fatalf("observation = %+v, err = %v; want ErrNotRecordable", o, err)
79 }
80 })
81 }
82 // A shutting-down read still lists the mappings and may be recorded.
83 if o := observe(t, reply("111122223333", types.InstanceStateNameShuttingDown, root)); len(o.Devices) != 1 {
84 t.Fatalf("shutting-down: %+v", o)
85 }
86}
87
88func TestNextWritesOnChangeAndAtLeastHourly(t *testing.T) {
89 first, ok := Next(nil, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), t0)
90 if !ok || first.ObservedAt != "2026-10-06T09:00:00Z" || len(first.RemovedVolumes) != 0 || first.RemovedVolumes == nil {
91 t.Fatalf("first = %+v ok=%v", first, ok)
92 }
93 same := observe(t, reply("111122223333", types.InstanceStateNameRunning, root))
94 if _, ok := Next(&first, entry, writer, same, t0.Add(59*time.Minute)); ok {
95 t.Fatal("wrote an unchanged mapping inside the hour")
96 }
97 if _, ok := Next(&first, entry, writer, same, t0.Add(time.Hour)); !ok {
98 t.Fatal("no record after an hour")
99 }
100 grown, ok := Next(&first, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root, scratch)), t0.Add(time.Minute))
101 if !ok || len(grown.BlockDevices) != 2 {
102 t.Fatalf("an attached volume was not recorded: %+v", grown)
103 }
104 shrunk, ok := Next(&grown, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), t0.Add(2*time.Minute))
105 if !ok || len(shrunk.RemovedVolumes) != 1 || shrunk.RemovedVolumes[0] != "vol-keep" {
106 t.Fatalf("a detached volume is not in removed_volumes: %+v", shrunk)
107 }
108 if _, ok := Next(&first, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameShuttingDown, root)), t0.Add(time.Minute)); !ok {
109 t.Fatal("a state change was not recorded")
110 }
111}
112
113func TestWriteIsCreateOnlyAndListReturnsObservationOrder(t *testing.T) {
114 ctx, st := context.Background(), release.NewDirStore(t.TempDir())
115 a, _ := Next(nil, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), t0)
116 b, _ := Next(&a, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root, data)), t0.Add(10*time.Minute))
117 for _, r := range []Record{b, a} {
118 if _, err := Write(ctx, st, r); err != nil {
119 t.Fatal(err)
120 }
121 }
122 if _, err := Write(ctx, st, a); err != nil {
123 t.Fatalf("the same record twice: %v", err)
124 }
125 other := a
126 other.RequestID = "req-other"
127 if _, err := Write(ctx, st, other); !errors.Is(err, release.ErrCASConflict) {
128 t.Fatalf("a different record under the same key: err = %v", err)
129 }
130 got, err := List(ctx, st, entry.Enrollment)
131 if err != nil {
132 t.Fatal(err)
133 }
134 if len(got) != 2 || got[0].ObservedAt != a.ObservedAt || got[1].ObservedAt != b.ObservedAt {
135 t.Fatalf("list = %+v", got)
136 }
137 if _, err := Write(ctx, st, Record{Enrollment: "../edges", ObservedAt: a.ObservedAt}); err == nil {
138 t.Fatal("wrote outside the prefix")
139 }
140}
141
142func TestListFailsOnARecordItCannotRead(t *testing.T) {
143 ctx, st := context.Background(), release.NewDirStore(t.TempDir())
144 if _, err := st.Put(ctx, Key(entry.Enrollment, t0), []byte(`{"v":1,"surprise":true}`), ""); err != nil {
145 t.Fatal(err)
146 }
147 if _, err := List(ctx, st, entry.Enrollment); err == nil {
148 t.Fatal("an unreadable record read as a gap")
149 }
150}
151
152func records(t *testing.T, steps ...[]vol) []Record {
153 t.Helper()
154 var out []Record
155 var prev *Record
156 for i, vs := range steps {
157 rec, _ := Next(prev, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, vs...)), t0.Add(time.Duration(i)*10*time.Minute))
158 out = append(out, rec)
159 prev = &out[len(out)-1]
160 }
161 return out
162}
163
164func TestDischarge(t *testing.T) {
165 enrolled := t0.Add(-5 * time.Minute)
166 terminated := t0.Add(30 * time.Minute)
167 if v := Discharge(records(t, []vol{root}, []vol{root, data}, []vol{root, data}), entry, enrolled, terminated); !v.Discharge {
168 t.Fatalf("all true, nothing removed: %+v", v)
169 }
170 cases := map[string]struct {
171 recs []Record
172 at time.Time
173 }{
174 "no record at all": {nil, enrolled},
175 "a false flag": {records(t, []vol{root}, []vol{root, scratch}), enrolled},
176 "a removed volume": {records(t, []vol{root, data}, []vol{root}), enrolled},
177 "a replaced root volume": {records(t, []vol{root}, []vol{{"/dev/xvda", "vol-root2", true}}), enrolled},
178 "first record long after": {records(t, []vol{root}), t0.Add(-2 * time.Hour)},
179 "only another enrollment's": {func() []Record { r := records(t, []vol{root}); r[0].Enrollment = "enr-other"; return r }(), enrolled},
180 "only another instance's": {func() []Record { r := records(t, []vol{root}); r[0].Cloud.InstanceID = "i-0zzz"; return r }(), enrolled},
181 }
182 for name, tc := range cases {
183 t.Run(name, func(t *testing.T) {
184 if v := Discharge(tc.recs, entry, tc.at, terminated); v.Discharge || v.Reason == "" {
185 t.Fatalf("verdict = %+v, want the chores left open with a reason", v)
186 }
187 })
188 }
189}
190
191// A removal is caught from the sequence even when the record that should
192// have listed it was never written (two workers, one missed write).
193func TestDischargeSeesARemovalNoRecordListed(t *testing.T) {
194 recs := records(t, []vol{root, data}, []vol{root})
195 recs[1].RemovedVolumes = []string{}
196 if v := Discharge(recs, entry, t0, t0.Add(time.Hour)); v.Discharge {
197 t.Fatal("discharged across a silent removal")
198 }
199}
200
201// Controller ruling, 2026-10-06: the newest record must be no more than
202// LastRecordWindow older than the termination observation. A recorder that
203// stopped long before the box died cannot speak for its last hours.
204func TestDischargeNeedsARecentLastRecord(t *testing.T) {
205 recs := records(t, []vol{root}, []vol{root}) // newest at t0+10m
206 newest := t0.Add(10 * time.Minute)
207 if v := Discharge(recs, entry, t0, newest.Add(LastRecordWindow)); !v.Discharge {
208 t.Fatalf("exactly at the window: %+v", v)
209 }
210 if v := Discharge(recs, entry, t0, newest.Add(LastRecordWindow+time.Second)); v.Discharge || v.Reason == "" {
211 t.Fatalf("a newest record older than the window discharged: %+v", v)
212 }
213}
214
215// Astra plan R1-6: what List returns is whatever is stored. A stored record
216// that is not evidence of a running instance's mappings keeps the chores
217// open, however it got there. Each bad record sits beside a good one so
218// that no removal is seen between them: only the record's own check can
219// refuse it.
220func TestDischargeRefusesStoredRecordsThatAreNotEvidence(t *testing.T) {
221 ctx := context.Background()
222 good := records(t, []vol{root})
223 const before = `"enrollment":"enr-annie-1","cloud":{"provider":"aws","account":"111122223333","region":"eu-west-1","instance_id":"i-0abc"},"observed_at":"2026-10-06T08:55:00Z"`
224 const after = `"enrollment":"enr-annie-1","cloud":{"provider":"aws","account":"111122223333","region":"eu-west-1","instance_id":"i-0abc"},"observed_at":"2026-10-06T09:05:00Z"`
225 type bad struct {
226 body string
227 at time.Duration // from t0; the good record is at t0
228 }
229 cases := map[string]bad{
230 // Before the good record: an empty mapping followed by one volume is
231 // an attach, not a removal.
232 "only the binding fields": {`{"v":1,` + before + `}`, -5 * time.Minute},
233 "a terminated read": {`{"v":1,` + before + `,"instance_state":"terminated","block_devices":[],"removed_volumes":[]}`, -5 * time.Minute},
234 // After it, keeping vol-root, so nothing is removed.
235 "a volume with no id": {`{"v":1,` + after + `,"instance_state":"running","block_devices":[{"device":"/dev/xvda","volume_id":"vol-root","delete_on_termination":true,"status":"attached"},{"device":"/dev/xvdb","volume_id":"","delete_on_termination":true,"status":"attached"}],"removed_volumes":[]}`, 5 * time.Minute},
236 "a version 2 record": {`{"v":2,` + after + `,"instance_state":"running","block_devices":[{"device":"/dev/xvda","volume_id":"vol-root","delete_on_termination":true,"status":"attached"}],"removed_volumes":[]}`, 5 * time.Minute},
237 // Astra plan R2-2: one case per check, each failing only that check.
238 "a running read with no volume": {`{"v":1,` + before + `,"instance_state":"running","block_devices":[],"removed_volumes":[]}`, -5 * time.Minute},
239 "a terminated read with its root mapping": {`{"v":1,` + after + `,"instance_state":"terminated","block_devices":[{"device":"/dev/xvda","volume_id":"vol-root","delete_on_termination":true,"status":"attached"}],"removed_volumes":[]}`, 5 * time.Minute},
240 "a volume with no device": {`{"v":1,` + after + `,"instance_state":"running","block_devices":[{"device":"/dev/xvda","volume_id":"vol-root","delete_on_termination":true,"status":"attached"},{"device":"","volume_id":"vol-data","delete_on_termination":true,"status":"attached"}],"removed_volumes":[]}`, 5 * time.Minute},
241 }
242 const goodAfter = `{"v":1,` + after + `,"instance_state":"running","block_devices":[{"device":"/dev/xvda","volume_id":"vol-root","delete_on_termination":true,"status":"attached"}],"removed_volumes":[]}`
243 swap := func(old, new string) string {
244 if !strings.Contains(goodAfter, old) {
245 t.Fatalf("no %q in the template", old)
246 }
247 return strings.Replace(goodAfter, old, new, 1)
248 }
249 cases["a detaching volume"] = bad{swap(`"status":"attached"`, `"status":"detaching"`), 5 * time.Minute}
250 cases["a detached volume"] = bad{swap(`"status":"attached"`, `"status":"detached"`), 5 * time.Minute}
251 cases["no status"] = bad{swap(`"status":"attached"`, `"status":""`), 5 * time.Minute}
252 cases["no enrollment"] = bad{swap(`"enrollment":"enr-annie-1"`, `"enrollment":""`), 5 * time.Minute}
253 cases["another enrollment"] = bad{swap(`"enrollment":"enr-annie-1"`, `"enrollment":"enr-other"`), 5 * time.Minute}
254 cases["another instance"] = bad{swap(`"instance_id":"i-0abc"`, `"instance_id":"i-0zzz"`), 5 * time.Minute}
255 cases["a region in another case"] = bad{swap(`"region":"eu-west-1"`, `"region":"EU-WEST-1"`), 5 * time.Minute}
256 cases["an unlabelled version 2 record"] = bad{`{"v":2}`, 5 * time.Minute}
257 for name, tc := range cases {
258 t.Run(name, func(t *testing.T) {
259 st := release.NewDirStore(t.TempDir())
260 if _, err := Write(ctx, st, good[0]); err != nil {
261 t.Fatal(err)
262 }
263 if _, err := st.Put(ctx, Key(entry.Enrollment, t0.Add(tc.at)), []byte(tc.body), ""); err != nil {
264 t.Fatal(err)
265 }
266 recs, err := List(ctx, st, entry.Enrollment)
267 if err != nil {
268 t.Fatal(err)
269 }
270 if len(recs) != 2 {
271 t.Fatalf("listed %d records, want 2", len(recs))
272 }
273 if v := Discharge(recs, entry, t0, t0.Add(time.Hour)); v.Discharge || v.Reason == "" {
274 t.Fatalf("verdict = %+v, want the chores left open", v)
275 }
276 })
277 }
278 // The good record alone, stored and listed, still discharges.
279 st := release.NewDirStore(t.TempDir())
280 if _, err := Write(ctx, st, good[0]); err != nil {
281 t.Fatal(err)
282 }
283 recs, _ := List(ctx, st, entry.Enrollment)
284 if v := Discharge(recs, entry, t0, t0.Add(time.Hour)); !v.Discharge {
285 t.Fatalf("a stored good record: %+v", v)
286 }
287}
288
289// A gap longer than LastRecordWindow between two records is a stretch the
290// recorder did not cover.
291func TestDischargeRefusesAnInteriorGap(t *testing.T) {
292 a, _ := Next(nil, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), t0)
293 late := t0.Add(30 * 24 * time.Hour)
294 b, _ := Next(&a, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), late)
295 if v := Discharge([]Record{a, b}, entry, t0, late.Add(time.Hour)); v.Discharge || v.Reason == "" {
296 t.Fatalf("a 30 day gap discharged: %+v", v)
297 }
298 c, _ := Next(&a, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root)), t0.Add(LastRecordWindow))
299 if v := Discharge([]Record{a, c}, entry, t0, t0.Add(LastRecordWindow)); !v.Discharge {
300 t.Fatalf("a gap of exactly the window: %+v", v)
301 }
302}
303
304// Records must lie between enrollment and termination, and both times must
305// be real and in order.
306func TestDischargeRefusesTimesThatDoNotFit(t *testing.T) {
307 recs := records(t, []vol{root}, []vol{root}) // t0, t0+10m
308 var zero time.Time
309 cases := map[string]struct{ enrolled, terminated time.Time }{
310 "zero termination": {t0, zero},
311 "zero enrollment": {zero, t0.Add(time.Hour)},
312 "terminated before enrolled": {t0.Add(time.Hour), t0},
313 "a record before enrollment": {t0.Add(time.Minute), t0.Add(time.Hour)},
314 "a record after termination": {t0, t0.Add(5 * time.Minute)},
315 }
316 for name, tc := range cases {
317 t.Run(name, func(t *testing.T) {
318 if v := Discharge(recs, entry, tc.enrolled, tc.terminated); v.Discharge || v.Reason == "" {
319 t.Fatalf("verdict = %+v", v)
320 }
321 })
322 }
323}
324
325// Final review 1: as EC2 shuts an instance down, its DeleteOnTermination
326// volumes show "detaching" on the way to deletion. A shutting-down record
327// listing them is the normal last record of a box and must discharge; a
328// detaching volume with a false flag, or one in any other state, must not.
329func TestDischargeAcceptsDetachingDeleteOnTerminationVolumesWhileShuttingDown(t *testing.T) {
330 running, _ := Next(nil, entry, writer, observe(t, reply("111122223333", types.InstanceStateNameRunning, root, data)), t0)
331 shutting := func(dot bool) Record {
332 out := reply("111122223333", types.InstanceStateNameShuttingDown, root, vol{data.dev, data.id, dot})
333 for _, m := range out.Reservations[0].Instances[0].BlockDeviceMappings {
334 m.Ebs.Status = types.AttachmentStatusDetaching
335 }
336 rec, ok := Next(&running, entry, writer, observe(t, out), t0.Add(10*time.Minute))
337 if !ok {
338 t.Fatal("the shutting-down read was not a new record")
339 }
340 return rec
341 }
342 if v := Discharge([]Record{running, shutting(true)}, entry, t0, t0.Add(15*time.Minute)); !v.Discharge {
343 t.Fatalf("shutting down, every volume detaching and DeleteOnTermination: %+v", v)
344 }
345 if v := Discharge([]Record{running, shutting(false)}, entry, t0, t0.Add(15*time.Minute)); v.Discharge || v.Reason == "" {
346 t.Fatalf("shutting down with a detaching volume that is not DeleteOnTermination: %+v", v)
347 }
348 if reason := incomplete(shutting(false)); reason == "" {
349 t.Fatal("incomplete accepted a detaching volume with a false flag")
350 }
351 stopping := shutting(true)
352 stopping.InstanceState = "stopping"
353 if v := Discharge([]Record{running, stopping}, entry, t0, t0.Add(15*time.Minute)); v.Discharge || v.Reason == "" {
354 t.Fatalf("a stopping record with detaching volumes: %+v", v)
355 }
356 detached := shutting(true)
357 detached.BlockDevices[0].Status = "detached"
358 if v := Discharge([]Record{running, detached}, entry, t0, t0.Add(15*time.Minute)); v.Discharge || v.Reason == "" {
359 t.Fatalf("a shutting-down record with a detached volume: %+v", v)
360 }
361}
362