amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package console
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "reflect"
9 "sort"
10 "strings"
11 "sync"
12 "testing"
13 "time"
14
15 "github.com/carlosframework/platform/internal/provision"
16 "github.com/carlosframework/platform/internal/release"
17 "github.com/carlosframework/platform/internal/runs"
18)
19
20func (fx *runFx) record(t *testing.T) *provision.Record {
21 t.Helper()
22 rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost)
23 if err != nil {
24 t.Fatal(err)
25 }
26 return rec
27}
28
29// plainStore hides every optional extension of the store under it (ListMod
30// among them), as the test fakes and the fleet store door do.
31type plainStore struct{ release.Store }
32
33func (fx *runFx) putRunJSON(t *testing.T, key string, v any) {
34 t.Helper()
35 raw, _ := json.Marshal(v)
36 if _, err := fx.rel.Put(context.Background(), key, raw, ""); err != nil {
37 t.Fatal(err)
38 }
39}
40
41func TestRunSweepExpiresAQueuedRunWhoseCallerLeft(t *testing.T) {
42 fx := runFixture(t)
43 id := fx.startRun(t)
44 fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Second))
45 if r := fx.record(t).ActiveRun; r.State != provision.RunQueued {
46 t.Fatalf("expired before its deadline: %+v", r)
47 }
48 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
49 if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
50 t.Fatalf("not expired 60 s past the request (codex R3-3): %+v", r)
51 }
52 events, _ := fx.st.Events(OpsAccountID, "shop", 20)
53 found := false
54 for _, e := range events {
55 found = found || (e.Action == "run-end" && strings.Contains(e.Detail, id) && strings.Contains(e.Detail, `"expired"`))
56 }
57 if !found {
58 t.Error("no run-end expired row")
59 }
60 // The expire route's own result: a follower can read why it ended.
61 res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
62 if err != nil || res == nil || res.Refused != runs.RefusedExpired {
63 t.Fatalf("expired result %+v, %v", res, err)
64 }
65 if fx.startRun(t) == "" {
66 t.Fatal("an expired run blocked the next")
67 }
68}
69
70// The deadline that counts is the record's, stamped at admission after the
71// upload. request.json's requested_at, which an upload can leave minutes
72// old, must not expire a run whose window has only just opened.
73func TestRunSweepExpiresByTheRecordsDeadlineNotTheRequests(t *testing.T) {
74 fx := runFixture(t)
75 id := fx.startRun(t)
76 req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", id)
77 if err != nil {
78 t.Fatal(err)
79 }
80 req.RequestedAt = time.Now().Add(-10 * time.Minute)
81 raw, _ := json.Marshal(req)
82 _, etag, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "shop", id))
83 if err != nil {
84 t.Fatal(err)
85 }
86 if _, err := fx.rel.Put(context.Background(), runs.RequestKey(release.OpsAccount, "shop", id), raw, etag); err != nil {
87 t.Fatal(err)
88 }
89 fx.srv.runSweep(context.Background(), time.Now().Add(10*time.Second))
90 if r := fx.record(t).ActiveRun; r.State != provision.RunQueued {
91 t.Fatalf("expired by requested_at: %+v", r)
92 }
93}
94
95func TestRunSweepExpiryLosesToALatePickup(t *testing.T) {
96 fx := runFixture(t)
97 id := fx.startRun(t)
98 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
99 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
100 if r := fx.record(t).ActiveRun; r.State != provision.RunPicked {
101 t.Fatalf("the sweep overrode a pickup: %+v", r)
102 }
103 if res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id); res != nil {
104 t.Fatalf("the losing sweep wrote a result: %+v", res)
105 }
106 if ends := fx.runEnds(t, id); len(ends) != 0 {
107 t.Fatalf("the losing sweep wrote run-end rows: %q", ends)
108 }
109}
110
111// The other order: the sweep's CAS lands first, and the box's pickup is the
112// one that changes nothing.
113func TestRunSweepExpiryBeatsALatePickupTheOtherWay(t *testing.T) {
114 fx := runFixture(t)
115 id := fx.startRun(t)
116 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Minute))
117 _, err := provision.SetRecordRun(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost,
118 provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
119 if !errors.Is(err, provision.ErrRunConflict) {
120 t.Fatalf("a pickup after the sweep expired the run: %v", err)
121 }
122 if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
123 t.Fatalf("%+v", r)
124 }
125}
126
127func TestRunSweepMarksALostRunDone(t *testing.T) {
128 fx := runFixture(t)
129 id := fx.startRun(t)
130 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
131 fx.chunk(t, id, 1, runs.StreamOut, "partial")
132 fx.chunk(t, id, 2, runs.StreamErr, "more")
133 fx.srv.runSweep(context.Background(), time.Now().Add(30*time.Minute))
134 if r := fx.record(t).ActiveRun; r.State != provision.RunPicked {
135 t.Fatalf("lost before an hour past its deadline: %+v", r)
136 }
137 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
138 if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
139 t.Fatalf("lost run not done: %+v", r)
140 }
141 res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
142 if err != nil || res == nil || res.Interrupted != runs.InterruptedLost {
143 t.Fatalf("lost result %+v, %v", res, err)
144 }
145 if res.LastSeq != 2 || res.Box != "box-1" {
146 t.Fatalf("lost result does not name the last chunk and box: %+v", res)
147 }
148 if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
149 t.Fatalf("run-end rows %q", ends)
150 }
151 fx.srv.runSweep(context.Background(), time.Now().Add(3*time.Hour))
152 if ends := fx.runEnds(t, id); len(ends) != 1 {
153 t.Fatalf("a second sweep wrote another row: %q", ends)
154 }
155}
156
157func TestRunSweepMarksACancellingRunLost(t *testing.T) {
158 fx := runFixture(t)
159 id := fx.startRun(t)
160 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
161 fx.transition(t, provision.RunCancelTransition(id))
162 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
163 if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
164 t.Fatalf("a cancelling run whose box is gone is not done: %+v", r)
165 }
166}
167
168// A box that wrote its own result and died before moving the run to done:
169// its result stands, and the row says result, not lost.
170func TestRunSweepLostRunKeepsABoxResultThatWonTheWrite(t *testing.T) {
171 fx := runFixture(t)
172 id := fx.startRun(t)
173 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
174 own := runs.Result{Exit: 3, Started: time.Now().Add(-time.Hour), Ended: time.Now().Add(-59 * time.Minute)}
175 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, own); err != nil {
176 t.Fatal(err)
177 }
178 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
179 if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
180 t.Fatalf("%+v", r)
181 }
182 res, _ := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
183 if res == nil || res.Exit != 3 || res.Interrupted != "" {
184 t.Fatalf("the box's result was overwritten: %+v", res)
185 }
186 if ends := fx.runEnds(t, id); len(ends) != 1 || strings.Contains(ends[0], `"lost"`) {
187 t.Fatalf("run-end rows %q", ends)
188 }
189}
190
191func TestRunSweepClearsTerminalRunsAfterADay(t *testing.T) {
192 fx := runFixture(t)
193 id := fx.startRun(t)
194 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
195 ended := time.Now()
196 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Ended: ended}); err != nil {
197 t.Fatal(err)
198 }
199 fx.transition(t, provision.RunFinish(id))
200 fx.srv.runSweep(context.Background(), ended.Add(23*time.Hour))
201 if fx.record(t).ActiveRun == nil {
202 t.Fatal("cleared before 24 hours")
203 }
204 fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour))
205 if r := fx.record(t).ActiveRun; r != nil {
206 t.Fatalf("not cleared after 24 hours: %+v", r)
207 }
208 if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
209 t.Fatalf("objects left %q", left)
210 }
211}
212
213// A cancelled or expired run is ended by the console, which wrote its result
214// at that moment: the day counts from there.
215func TestRunSweepClearsACancelledRunADayAfterItsResult(t *testing.T) {
216 fx := runFixture(t)
217 id := fx.startRun(t)
218 if status, body := fx.post(t, id, "cancel", ""); status != 200 {
219 t.Fatalf("cancel: %d %s", status, body)
220 }
221 fx.srv.runSweep(context.Background(), time.Now().Add(23*time.Hour))
222 if fx.record(t).ActiveRun == nil {
223 t.Fatal("cleared before 24 hours")
224 }
225 fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour))
226 if r := fx.record(t).ActiveRun; r != nil {
227 t.Fatalf("not cleared after 24 hours: %+v", r)
228 }
229 if left := fx.left(id); len(left) != 0 {
230 t.Fatalf("objects left %q", left)
231 }
232}
233
234// A console that died between the record's CAS and the result and run-end
235// row leaves a terminal run with neither. The sweep writes them, once.
236func TestRunSweepFinishesAnExpiredRunThatHasNoResult(t *testing.T) {
237 fx := runFixture(t)
238 id := fx.startRun(t)
239 fx.transition(t, provision.RunExpire(id))
240 fx.srv.runSweep(context.Background(), time.Now())
241 res, err := runs.ReadResult(context.Background(), fx.rel, release.OpsAccount, "shop", id)
242 if err != nil || res == nil || res.Refused != runs.RefusedExpired {
243 t.Fatalf("result %+v, %v", res, err)
244 }
245 fx.srv.runSweep(context.Background(), time.Now())
246 if ends := fx.runEnds(t, id); len(ends) != 1 {
247 t.Fatalf("run-end rows %q", ends)
248 }
249}
250
251func TestRunSweepDeletesOrphanRunDirectories(t *testing.T) {
252 fx := runFixture(t)
253 ctx := context.Background()
254 old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
255 fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
256 young := old
257 young.ID, young.RequestedAt = "run_dddddddddddddddddddddddddd", time.Now()
258 fx.putRunJSON(t, runs.RequestKey(young.Account, young.App, young.ID), young)
259 fx.srv.runSweep(ctx, time.Now())
260 if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 {
261 t.Errorf("a day-old orphan survived: %q", left)
262 }
263 if left, _ := fx.rel.List(ctx, runs.Prefix(young.Account, young.App, young.ID)); len(left) == 0 {
264 t.Error("a fresh orphan (a create in flight) was deleted")
265 }
266}
267
268// The run the record names is never an orphan, however old its request.
269func TestRunSweepKeepsTheRunTheRecordNames(t *testing.T) {
270 fx := runFixture(t)
271 id := fx.startRun(t)
272 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(100*time.Hour)))
273 fx.srv.runSweep(context.Background(), time.Now().Add(48*time.Hour))
274 if left := fx.left(id); len(left) == 0 {
275 t.Fatal("the picked run's objects were deleted")
276 }
277}
278
279// A directory with no request.json (a stray result.json a stale retry can
280// leave, or an audited marker left by a cut-short delete) is reaped once
281// its newest object is a day old, when the store lists modification times.
282func TestRunSweepReapsAStrayRunDirectoryByItsNewestObject(t *testing.T) {
283 fx := runFixture(t)
284 ctx := context.Background()
285 id := "run_eeeeeeeeeeeeeeeeeeeeeeeeee"
286 if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "shop", id), []byte(`{}`), ""); err != nil {
287 t.Fatal(err)
288 }
289 fx.srv.runSweep(ctx, time.Now())
290 fx.srv.runSweep(ctx, time.Now())
291 if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
292 t.Fatal("a stray written a moment ago was deleted")
293 }
294 fx.srv.runSweep(ctx, time.Now().Add(25*time.Hour))
295 if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) != 0 {
296 t.Fatalf("a day-old stray survived: %q", left)
297 }
298}
299
300// A store that cannot list modification times leaves strays alone.
301func TestRunSweepLeavesAStrayAloneWithoutModTimes(t *testing.T) {
302 fx := runFixture(t)
303 ctx := context.Background()
304 fx.srv.cfg.Releases = plainStore{fx.rel}
305 if _, ok := fx.srv.cfg.Releases.(release.ModLister); ok {
306 t.Fatal("the wrapper still lists modification times")
307 }
308 id := "run_ffffffffffffffffffffffffff"
309 if _, err := fx.rel.Put(ctx, runs.ResultKey(release.OpsAccount, "shop", id), []byte(`{}`), ""); err != nil {
310 t.Fatal(err)
311 }
312 for i := 0; i < 3; i++ {
313 fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
314 }
315 if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
316 t.Fatal("a stray was deleted by a store that cannot say how old it is")
317 }
318}
319
320func TestRunSweepLeavesAnUnreadableRequestAlone(t *testing.T) {
321 fx := runFixture(t)
322 ctx := context.Background()
323 id := "run_hhhhhhhhhhhhhhhhhhhhhhhhhh"
324 if _, err := fx.rel.Put(ctx, runs.RequestKey(release.OpsAccount, "shop", id), bytes.Repeat([]byte("x"), 8), ""); err != nil {
325 t.Fatal(err)
326 }
327 fx.srv.runSweep(ctx, time.Now().Add(48*time.Hour))
328 if left, _ := fx.rel.List(ctx, runs.Prefix(release.OpsAccount, "shop", id)); len(left) == 0 {
329 t.Fatal("a directory whose request could not be read was deleted")
330 }
331}
332
333// A follower's poll that lands between the sweep's lost result and its done
334// CAS wins the audit marker. The row must still say lost.
335func TestRunSweepLostRunRowSaysLostWhenAPollGotThereFirst(t *testing.T) {
336 fx := runFixture(t)
337 id := fx.startRun(t)
338 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
339 lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now().Add(2 * time.Hour)}
340 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, lost); err != nil {
341 t.Fatal(err)
342 }
343 if p := fx.page(t, id, 0); p.Result == nil {
344 t.Fatal("the poll did not see the result")
345 }
346 fx.srv.runSweep(context.Background(), time.Now().Add(2*time.Hour))
347 if r := fx.record(t).ActiveRun; r.State != provision.RunDone {
348 t.Fatalf("%+v", r)
349 }
350 if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
351 t.Fatalf("run-end rows %q", ends)
352 }
353}
354
355// A run a box finished that nobody followed or acked gets its one row from
356// the sweep, before the sweep clears it.
357func TestRunSweepWritesTheRowOfAFinishedRunNobodyFollowed(t *testing.T) {
358 fx := runFixture(t)
359 id := fx.startRun(t)
360 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
361 ended := time.Now()
362 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 2, Ended: ended}); err != nil {
363 t.Fatal(err)
364 }
365 fx.transition(t, provision.RunFinish(id))
366 fx.srv.runSweep(context.Background(), ended.Add(time.Hour))
367 if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) {
368 t.Fatalf("run-end rows %q", ends)
369 }
370 fx.srv.runSweep(context.Background(), ended.Add(25*time.Hour))
371 if r := fx.record(t).ActiveRun; r != nil {
372 t.Fatalf("not cleared: %+v", r)
373 }
374 if ends := fx.runEnds(t, id); len(ends) != 1 {
375 t.Fatalf("run-end rows after the clear %q", ends)
376 }
377}
378
379// A run the sweep marked done and then died: no row. A later pass writes it,
380// as lost.
381func TestRunSweepWritesTheRowOfALostRunItsSweepDiedOn(t *testing.T) {
382 fx := runFixture(t)
383 id := fx.startRun(t)
384 fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Minute)))
385 lost := runs.Result{Interrupted: runs.InterruptedLost, Box: "box-1", Ended: time.Now()}
386 if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, lost); err != nil {
387 t.Fatal(err)
388 }
389 fx.transition(t, provision.RunFinish(id))
390 fx.srv.runSweep(context.Background(), time.Now().Add(time.Hour))
391 if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) {
392 t.Fatalf("run-end rows %q", ends)
393 }
394}
395
396// Expiry and lost-run closing run on every pass; retention and orphans only
397// on housekeeping passes.
398func TestRunSweepHousekeepingPhasesSkipOffPasses(t *testing.T) {
399 fx := runFixture(t)
400 ctx := context.Background()
401 old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
402 fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
403 fx.startRun(t)
404 fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false)
405 if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
406 t.Fatalf("expiry did not run on an off pass: %+v", r)
407 }
408 if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) == 0 {
409 t.Fatal("the orphan phase ran on an off pass")
410 }
411 fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), false)
412 if fx.record(t).ActiveRun == nil {
413 t.Fatal("the retention phase ran on an off pass")
414 }
415 fx.srv.runSweepPass(ctx, time.Now().Add(26*time.Hour), true)
416 if left, _ := fx.rel.List(ctx, runs.Prefix(old.Account, old.App, old.ID)); len(left) != 0 {
417 t.Fatalf("the orphan phase did not run on a housekeeping pass: %q", left)
418 }
419 if r := fx.record(t).ActiveRun; r != nil {
420 t.Fatalf("the retention phase did not run on a housekeeping pass: %+v", r)
421 }
422}
423
424// hangingStore blocks every call until its context is done.
425type hangingStore struct{ release.Store }
426
427func (hangingStore) Get(ctx context.Context, _ string) ([]byte, string, error) {
428 <-ctx.Done()
429 return nil, "", ctx.Err()
430}
431func (hangingStore) List(ctx context.Context, _ string) ([]string, error) {
432 <-ctx.Done()
433 return nil, ctx.Err()
434}
435func (hangingStore) ListDirs(ctx context.Context, _ string) ([]string, error) {
436 <-ctx.Done()
437 return nil, ctx.Err()
438}
439
440func TestRunSweepPassIsBoundedByATimeout(t *testing.T) {
441 fx := runFixture(t)
442 fx.srv.cfg.Releases = hangingStore{fx.rel}
443 old := runSweepTimeout
444 runSweepTimeout = 50 * time.Millisecond
445 defer func() { runSweepTimeout = old }()
446 done := make(chan struct{})
447 go func() { fx.srv.runSweep(context.Background(), time.Now()); close(done) }()
448 select {
449 case <-done:
450 case <-time.After(10 * time.Second):
451 t.Fatal("a hung store stalled the pass")
452 }
453}
454
455// A run a box finished with no follower, replaced by a new admission before
456// a housekeeping pass, is an orphan when the sweep comes to it. Its row is
457// written before its objects go.
458func TestRunSweepWritesTheRowOfAReplacedRunBeforeDeletingIt(t *testing.T) {
459 fx := runFixture(t)
460 id := fx.finishedRun(t)
461 fx.startRun(t) // the admission replaces the finished run on the record
462 if ends := fx.runEnds(t, id); len(ends) != 0 {
463 t.Fatalf("run-end rows before the sweep %q", ends)
464 }
465 fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour))
466 if left := fx.left(id); len(left) != 0 {
467 t.Fatalf("the orphan survived: %q", left)
468 }
469 if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) {
470 t.Fatalf("run-end rows %q", ends)
471 }
472}
473
474// An orphan with no result (its record let it go before any box wrote one)
475// gets a lost result and its row says lost. A run that already has its row
476// does not get a second.
477func TestRunSweepWritesALostRowForAnOrphanWithNoResult(t *testing.T) {
478 fx := runFixture(t)
479 ctx := context.Background()
480 old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, By: "alice@keymail.dev", RequestedAt: time.Now().Add(-25 * time.Hour)}
481 fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
482 fx.chunk(t, old.ID, 1, runs.StreamOut, "a")
483 rowed := old
484 rowed.ID = "run_dddddddddddddddddddddddddd"
485 fx.putRunJSON(t, runs.RequestKey(rowed.Account, rowed.App, rowed.ID), rowed)
486 if _, err := fx.rel.Put(ctx, runs.AuditedKey(rowed.Account, rowed.App, rowed.ID), []byte("result"), ""); err != nil {
487 t.Fatal(err)
488 }
489 fx.srv.runSweep(ctx, time.Now())
490 for _, id := range []string{old.ID, rowed.ID} {
491 if left := fx.left(id); len(left) != 0 {
492 t.Fatalf("%s survived: %q", id, left)
493 }
494 }
495 ends := fx.runEnds(t, old.ID)
496 if len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) || !strings.Contains(ends[0], `"last_seq":1`) {
497 t.Fatalf("run-end rows %q", ends)
498 }
499 if ends := fx.runEnds(t, rowed.ID); len(ends) != 0 {
500 t.Fatalf("a run with its row got another: %q", ends)
501 }
502}
503
504// A marker the bucket refuses leaves the orphan for a later pass, so its
505// row is never lost to the delete.
506func TestRunSweepKeepsAnOrphanWhoseRowCouldNotBeMarked(t *testing.T) {
507 fx := runFixture(t)
508 ctx := context.Background()
509 old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)}
510 fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
511 refuse := true
512 fx.srv.cfg.Releases = &faultStore{Store: fx.rel, put: func(key string) (bool, error) {
513 if refuse && strings.HasSuffix(key, "/audited") {
514 return false, errors.New("access denied")
515 }
516 return false, nil
517 }}
518 fx.srv.runSweep(ctx, time.Now())
519 if left := fx.left(old.ID); len(left) == 0 {
520 t.Fatal("the orphan was deleted with no row")
521 }
522 refuse = false
523 fx.srv.runSweep(ctx, time.Now())
524 if left := fx.left(old.ID); len(left) != 0 {
525 t.Fatalf("the orphan survived: %q", left)
526 }
527 if ends := fx.runEnds(t, old.ID); len(ends) != 1 {
528 t.Fatalf("run-end rows %q", ends)
529 }
530}
531
532// One record that will not read does not stop expiry for the rest of the
533// fleet, but it does stop the orphan phase: a run that record names could
534// look like an orphan.
535func TestRunSweepReadsPastAnUnreadableRecordButDeletesNoOrphan(t *testing.T) {
536 fx := runFixture(t)
537 ctx := context.Background()
538 putShopRecord(t, fx.consoleFixture, provision.Record{Host: "other.example.com", Channel: "stable", CreatedAt: time.Now().UTC()})
539 fx.startRun(t)
540 old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: "other.example.com", RequestedAt: time.Now().Add(-25 * time.Hour)}
541 fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old)
542 unreadable := provision.RecordKey(release.OpsAccount, "shop", "other.example.com")
543 broken := true
544 fx.srv.cfg.Releases = &faultStore{Store: fx.rel, get: func(key string) error {
545 if broken && key == unreadable {
546 return errors.New("slow down")
547 }
548 return nil
549 }}
550 fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
551 if r := fx.record(t).ActiveRun; r.State != provision.RunExpired {
552 t.Fatalf("one unreadable record stopped the expiry: %+v", r)
553 }
554 if left := fx.left(old.ID); len(left) == 0 {
555 t.Fatal("an orphan was deleted on a pass that could not read every record")
556 }
557 broken = false
558 fx.srv.runSweep(ctx, time.Now().Add(2*time.Minute))
559 if left := fx.left(old.ID); len(left) != 0 {
560 t.Fatalf("the orphan survived a full read: %q", left)
561 }
562}
563
564// listCountingStore is a DirStore with the change index on that records every
565// listing made through it, so a test can say which prefixes a pass walked.
566// Its embedded DirStore keeps every optional extension (conditional reads,
567// ETag listings, ModLister) a record view and the sweeps use.
568type listCountingStore struct {
569 *release.DirStore
570 mu sync.Mutex
571 lists []string // "op prefix", in call order
572}
573
574func (c *listCountingStore) note(op, prefix string) {
575 c.mu.Lock()
576 c.lists = append(c.lists, op+" "+prefix)
577 c.mu.Unlock()
578}
579
580// taken returns the listings made since the last call and forgets them.
581func (c *listCountingStore) taken() []string {
582 c.mu.Lock()
583 defer c.mu.Unlock()
584 out := c.lists
585 c.lists = nil
586 return out
587}
588
589func (c *listCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
590 c.note("List", prefix)
591 return c.DirStore.List(ctx, prefix)
592}
593
594func (c *listCountingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
595 c.note("ListPage", prefix)
596 return c.DirStore.ListPage(ctx, prefix, startAfter, limit)
597}
598
599func (c *listCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
600 c.note("ListDirs", prefix)
601 return c.DirStore.ListDirs(ctx, prefix)
602}
603
604func (c *listCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
605 c.note("ListETags", prefix)
606 return c.DirStore.ListETags(ctx, prefix)
607}
608
609// indexedDeployment swaps fx's deployment store for a listCountingStore over the
610// same bucket, with the change index on, as the console's opener builds it.
611func indexedDeployment(t *testing.T, fx *consoleFixture) *listCountingStore {
612 t.Helper()
613 d, ok := fx.rel.(*deletableStore)
614 if !ok {
615 t.Fatalf("fixture store is %T", fx.rel)
616 }
617 ds := release.NewDirStore(d.root)
618 ds.EnableChangeIndex()
619 cs := &listCountingStore{DirStore: ds}
620 fx.rel, fx.srv.cfg.Releases = cs, cs
621 return cs
622}
623
624// armRecordView arms srv's record view over live, as cmd/console does with
625// CARLOS_CHANGE_INDEX=read, and runs the backstop's first rebuild.
626func armRecordView(t *testing.T, srv *Server, live release.Store) *provision.RecordView {
627 t.Helper()
628 v := provision.NewRecordView(live, release.NewChangeWatch(live))
629 if err := v.Rebuild(context.Background()); err != nil {
630 t.Fatal(err)
631 }
632 srv.recordView = v
633 return v
634}
635
636// runsListings keeps the listings under a runs/ prefix.
637func runsListings(lists []string) []string {
638 var out []string
639 for _, l := range lists {
640 if strings.Contains(l, "/runs/") {
641 out = append(out, l)
642 }
643 }
644 sort.Strings(out)
645 return out
646}
647
648// Armed, the run sweep reads its records from the record view: once the
649// view has caught up, a pass with nothing written lists nothing at all.
650// Disarmed, the same pass walks the bucket as before the index.
651func TestRunSweepReadsViewWhenArmed(t *testing.T) {
652 t.Parallel()
653 fx := runFixture(t)
654 cs := indexedDeployment(t, fx.consoleFixture)
655 ctx := context.Background()
656 id := fx.startRun(t)
657
658 cs.taken()
659 fx.srv.runSweepPass(ctx, time.Now(), false)
660 if lists := cs.taken(); len(lists) == 0 {
661 t.Fatal("a disarmed pass listed nothing: the counter does not see the walk")
662 }
663
664 armRecordView(t, fx.srv, cs)
665 // The expiry acts on the run the view holds, and its write bumps the
666 // app's scope, which the next pass's refresh re-reads.
667 fx.srv.runSweepPass(ctx, time.Now().Add(2*time.Minute), false)
668 if r := fx.record(t).ActiveRun; r == nil || r.ID != id || r.State != provision.RunExpired {
669 t.Fatalf("armed pass did not expire the run it read from the view: %+v", r)
670 }
671 fx.srv.runSweepPass(ctx, time.Now().Add(3*time.Minute), false)
672 cs.taken()
673 for i := 0; i < 2; i++ {
674 fx.srv.runSweepPass(ctx, time.Now().Add(4*time.Minute), false)
675 }
676 if lists := cs.taken(); len(lists) != 0 {
677 t.Fatalf("armed passes with nothing written listed the bucket: %q", lists)
678 }
679}
680
681// putAppRecord writes an instance record of app in the operator account.
682func putAppRecord(t *testing.T, s release.Store, app, host string) {
683 t.Helper()
684 rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount,
685 Channel: "stable", CreatedAt: time.Now().UTC()}
686 if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
687 t.Fatal(err)
688 }
689}
690
691// Between daily full sweeps, orphan housekeeping lists runs/ only for apps
692// with a run-start row in the last 48 hours (codex R2-1): shop ran an hour
693// ago, cafe never did, so only shop's runs/ is listed.
694func TestOrphanHousekeepingListsOnlyRecentRunApps(t *testing.T) {
695 t.Parallel()
696 fx := runFixture(t)
697 cs := indexedDeployment(t, fx.consoleFixture)
698 putAppRecord(t, cs, "cafe", "cafe.example.com")
699 now := time.Now()
700 if err := fx.st.RecordActivity(OpsAccountID, "shop", "alice@keymail.dev", Activity{Action: "run-start"}, now.Add(-time.Hour)); err != nil {
701 t.Fatal(err)
702 }
703 armRecordView(t, fx.srv, cs)
704 fx.srv.lastOrphanFull = now.Add(-time.Hour)
705 cs.taken()
706 fx.srv.runSweepPass(context.Background(), now, true)
707 want := []string{"ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop")}
708 if got := runsListings(cs.taken()); !reflect.DeepEqual(got, want) {
709 t.Fatalf("runs/ listings %q, want %q", got, want)
710 }
711 if !fx.srv.lastOrphanFull.Equal(now.Add(-time.Hour)) {
712 t.Fatalf("a narrowed sweep moved the full sweep's clock to %s", fx.srv.lastOrphanFull)
713 }
714}
715
716// The first housekeeping after start, and then one a day, sweeps every app
717// with records, whatever the activity rows say.
718func TestOrphanHousekeepingFullSweepDaily(t *testing.T) {
719 t.Parallel()
720 fx := runFixture(t)
721 cs := indexedDeployment(t, fx.consoleFixture)
722 putAppRecord(t, cs, "cafe", "cafe.example.com")
723 armRecordView(t, fx.srv, cs)
724 ctx := context.Background()
725 all := []string{
726 "ListDirs " + runs.RunsPrefix(release.OpsAccount, "cafe"),
727 "ListDirs " + runs.RunsPrefix(release.OpsAccount, "shop"),
728 }
729 now := time.Now()
730
731 cs.taken()
732 fx.srv.runSweepPass(ctx, now, true)
733 if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) {
734 t.Fatalf("first housekeeping listed %q, want every app with records %q", got, all)
735 }
736 if !fx.srv.lastOrphanFull.Equal(now) {
737 t.Fatalf("lastOrphanFull = %s after a full sweep at %s", fx.srv.lastOrphanFull, now)
738 }
739
740 fx.srv.runSweepPass(ctx, now.Add(10*time.Minute), true)
741 if got := runsListings(cs.taken()); len(got) != 0 {
742 t.Fatalf("housekeeping with no recent run listed %q", got)
743 }
744
745 later := now.Add(orphanFullSweepEvery + time.Minute)
746 fx.srv.runSweepPass(ctx, later, true)
747 if got := runsListings(cs.taken()); !reflect.DeepEqual(got, all) {
748 t.Fatalf("a day later housekeeping listed %q, want %q", got, all)
749 }
750 if !fx.srv.lastOrphanFull.Equal(later) {
751 t.Fatalf("lastOrphanFull = %s, want %s", fx.srv.lastOrphanFull, later)
752 }
753}
754