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 "context"
5 "crypto/ed25519"
6 "crypto/sha256"
7 "encoding/hex"
8 "encoding/json"
9 "errors"
10 "path/filepath"
11 "slices"
12 "strings"
13 "sync"
14 "sync/atomic"
15 "testing"
16 "time"
17
18 "github.com/aws/aws-sdk-go-v2/service/sns"
19 "github.com/carlosframework/platform/internal/alerting"
20 "github.com/carlosframework/platform/internal/hostclaim"
21 "github.com/carlosframework/platform/internal/release"
22)
23
24func lossPageRecord(id string, pending bool, knowledge hostclaim.TimerKnowledge) hostclaim.LossRecord {
25 at := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
26 r := hostclaim.LossRecord{V: 1, ID: id, State: "prepared", Actor: "operator", AcceptedAt: at,
27 Host: "host.example", HolderIncarnation: strings.Repeat("a", 64), HolderStamp: 1,
28 ExpectedClaimGen: 1, ExpectedClaimETag: "claim", OpID: "operation", OpKind: hostclaim.KindMoving,
29 OpPhase: "precommit", Box: "box-a", Enrollment: "enrollment", QuarantineRevision: 1, FencedAt: at,
30 CloudObservation: hostclaim.LossCloudObservation{ObservedAt: at, Account: "123", Region: "eu-west-1", InstanceID: "i-1", State: "stopped", RequestID: "request"},
31 OperatorAttestation: hostclaim.LossOperatorAttestation{MaintainComputeFence: true, ColdQuarantineStartOnly: true},
32 ConfirmedConsequences: hostclaim.LossConsequences{RecoveryIntent: json.RawMessage(`{"kind":"lost-target"}`)},
33 AlertsPending: pending, Timers: hostclaim.TimerLoss{Knowledge: knowledge}}
34 if knowledge != hostclaim.TimersUnknown {
35 r.Timers.Source = &hostclaim.TimerSource{Bucket: "bucket", Key: "timers", VersionID: "v1", SHA256: strings.Repeat("b", 64), CapturedAt: at}
36 }
37 return r
38}
39
40func lossPageWorld(t *testing.T) (*Server, *release.DirStore, release.Signer, *[]map[string]string) {
41 t.Helper()
42 raw := release.NewDirStore(t.TempDir())
43 _, key, err := ed25519.GenerateKey(nil)
44 if err != nil {
45 t.Fatal(err)
46 }
47 publishRestoreRoster(t, raw, key)
48 s := newEnabledRestoreConsole(t, raw, key, crConsole)
49 var pages []map[string]string
50 alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, body []byte) error {
51 var page map[string]string
52 if err := json.Unmarshal(body, &page); err != nil {
53 return err
54 }
55 pages = append(pages, page)
56 return nil
57 }})
58 t.Cleanup(func() { alerting.Arm(nil) })
59 return s, raw, release.RootSigner(key), &pages
60}
61
62func putLossPageRecord(t *testing.T, raw release.Store, signer release.Signer, r hostclaim.LossRecord) {
63 t.Helper()
64 if _, err := hostclaim.CreateLossRecord(t.Context(), raw, signer, r); err != nil {
65 t.Fatal(err)
66 }
67}
68
69func TestClaimLossPagesIntervalAndRestart(t *testing.T) {
70 s, raw, signer, pages := lossPageWorld(t)
71 for _, id := range []string{"loss-one", "loss-two"} {
72 putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
73 }
74 now := time.Date(2026, 10, 12, 13, 0, 0, 123, time.UTC)
75 pass := func() {
76 t.Helper()
77 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
78 t.Fatal(err)
79 }
80 }
81 pass()
82 pass()
83 if len(*pages) != 2 {
84 t.Fatalf("first passes: %v", *pages)
85 }
86 // Reopen the actual SQLite file; no in-memory limiter survives this.
87 path := s.store.gate.path
88 if err := s.store.Close(); err != nil {
89 t.Fatal(err)
90 }
91 st, err := OpenGated(path)
92 if err != nil {
93 t.Fatal(err)
94 }
95 t.Cleanup(func() { st.Close() })
96 s = &Server{cfg: s.cfg, store: st}
97 now = now.Add(time.Hour - time.Nanosecond)
98 pass()
99 if len(*pages) != 2 {
100 t.Fatalf("restart lost rate limit: %v", *pages)
101 }
102 now = now.Add(time.Nanosecond)
103 pass()
104 if len(*pages) != 4 {
105 t.Fatalf("interval did not repeat each record: %v", *pages)
106 }
107 key, _ := hostclaim.LossRecordKey("loss-one")
108 if err := raw.Delete(t.Context(), key); err != nil {
109 t.Fatal(err)
110 }
111 now = now.Add(time.Hour)
112 pass()
113 if len(*pages) != 5 || !strings.Contains((*pages)[4]["body"], "loss-two") {
114 t.Fatalf("absent pending record paged: %v", *pages)
115 }
116}
117
118func TestClaimLossPagesClearedAndTimers(t *testing.T) {
119 for _, knowledge := range []hostclaim.TimerKnowledge{hostclaim.TimersComplete, hostclaim.TimersPartial, hostclaim.TimersUnknown} {
120 t.Run(string(knowledge), func(t *testing.T) {
121 s, raw, signer, pages := lossPageWorld(t)
122 r := lossPageRecord("loss-timer", false, knowledge)
123 putLossPageRecord(t, raw, signer, r)
124 now := r.AcceptedAt
125 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); err != nil {
126 t.Fatal(err)
127 }
128 want := 1
129 if knowledge == hostclaim.TimersComplete {
130 want = 0
131 }
132 if len(*pages) != want {
133 t.Fatalf("pages: %v", *pages)
134 }
135 if want != 0 && (!strings.Contains((*pages)[0]["body"], r.Host) || !strings.Contains((*pages)[0]["body"], "re-arm")) {
136 t.Fatalf("timer instructions: %v", *pages)
137 }
138 key, _ := hostclaim.LossRecordKey(r.ID)
139 if err := raw.Delete(t.Context(), key); err != nil {
140 t.Fatal(err)
141 }
142 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(2 * time.Hour) }); err != nil {
143 t.Fatal(err)
144 }
145 if len(*pages) != want {
146 t.Fatal("cleared record paged")
147 }
148 })
149 }
150}
151
152func TestClaimLossPagesRecoveryWiringWithoutMovesOrClaims(t *testing.T) {
153 s, raw, signer, pages := lossPageWorld(t)
154 putLossPageRecord(t, raw, signer, lossPageRecord("orphan-loss", true, hostclaim.TimersComplete))
155 if s.cfg.MovesEnabled {
156 t.Fatal("fixture enabled moves")
157 }
158 if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
159 t.Fatal(err)
160 }
161 waitClaimLossPages(t, s)
162 if len(*pages) != 1 {
163 t.Fatalf("record without a claim was not paged: %v", *pages)
164 }
165}
166
167func TestClaimLossPagesFlagOff(t *testing.T) {
168 calls := 0
169 alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return nil }})
170 t.Cleanup(func() { alerting.Arm(nil) })
171 s := &Server{}
172 if err := s.pageClaimLossesAt(t.Context(), time.Now); err != nil {
173 t.Fatal(err)
174 }
175 if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
176 t.Fatal(err)
177 }
178 s.scheduleClaimLossPages(t.Context())
179 if s.claimLossPageDone != nil {
180 t.Fatal("disabled runtime started a worker")
181 }
182 if calls != 0 {
183 t.Fatal("disabled runtime paged")
184 }
185 st, err := Open(filepath.Join(t.TempDir(), "off.db"))
186 if err != nil {
187 t.Fatal(err)
188 }
189 defer st.Close()
190 var count int
191 if err := st.db.QueryRow(`SELECT count(*) FROM sqlite_master WHERE name = 'claim_loss_pages'`).Scan(&count); err != nil || count != 0 {
192 t.Fatalf("flag-off schema changed: %d %v", count, err)
193 }
194}
195
196func TestClaimLossPageFailedDeliveryIsRateLimited(t *testing.T) {
197 s, raw, signer, _ := lossPageWorld(t)
198 putLossPageRecord(t, raw, signer, lossPageRecord("failed-page", true, hostclaim.TimersComplete))
199 calls := 0
200 failure := errors.New("delivery failed")
201 alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error { calls++; return failure }})
202 now := time.Now()
203 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now }); !errors.Is(err, failure) {
204 t.Fatalf("delivery: %v", err)
205 }
206 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return now.Add(time.Minute) }); err != nil {
207 t.Fatal(err)
208 }
209 if calls != 1 {
210 t.Fatalf("attempts: %d", calls)
211 }
212}
213
214func TestClaimLossPagesAutomaticReportingDoesNotAcknowledge(t *testing.T) {
215 s, raw, _, pages := lossPageWorld(t)
216 ref := hostclaim.FenceRef{Enrollment: strings.Repeat("a", 32), SHA256: strings.Repeat("b", 64), Key: "control/box-fences/" + strings.Repeat("a", 32) + ".json"}
217 sum := sha256.Sum256([]byte("operation\x00box-a\x00" + ref.Enrollment + "\x00" + ref.SHA256))
218 id := "auto-" + hex.EncodeToString(sum[:16])
219 loss := hostclaim.AutomaticLoss{ID: id, Box: "box-a", Enrollment: ref.Enrollment, Fence: ref,
220 ObservedAt: time.Now().UTC().Format(time.RFC3339Nano), Op: "operation", Kind: hostclaim.KindLanding,
221 Phase: hostclaim.PhaseProbing, Role: "target", Incarnation: strings.Repeat("c", 64), Stamp: 1, Timers: hostclaim.TimersUnknown}
222 _, err := hostclaim.Update(t.Context(), raw, hostclaim.Observed{Host: "host.example", Absent: true}, func(c *hostclaim.Claim) error {
223 c.LossRefs = []string{id}
224 c.EvidenceRefs = []hostclaim.AutomaticLoss{loss}
225 return nil
226 })
227 if err != nil {
228 t.Fatal(err)
229 }
230 if err := s.RunClaimRecoveryPass(t.Context()); err != nil {
231 t.Fatal(err)
232 }
233 waitClaimLossPages(t, s)
234 if len(*pages) != 2 {
235 t.Fatalf("materialized record must page both conditions in the same pass: %v", *pages)
236 }
237 c, err := hostclaim.Read(t.Context(), raw, "host.example")
238 if err != nil || !c.Claim.EvidenceRefs[0].Reported {
239 t.Fatalf("reporting did not finish: %+v %v", c, err)
240 }
241 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return time.Now().Add(2 * time.Hour) }); err != nil {
242 t.Fatal(err)
243 }
244 if len(*pages) != 4 {
245 t.Fatal("claim reporting silenced pending operator alerts")
246 }
247}
248
249func TestClaimLossPageReservationSerializesPasses(t *testing.T) {
250 st := openGatedStore(t)
251 now := time.Now()
252 var wg sync.WaitGroup
253 var attempts atomic.Int64
254 for i := 0; i < 12; i++ {
255 wg.Add(1)
256 go func() {
257 defer wg.Done()
258 ok, err := st.ClaimLossPageAttempt(t.Context(), "pending loss", now)
259 if err != nil {
260 t.Error(err)
261 }
262 if ok {
263 attempts.Add(1)
264 }
265 }()
266 }
267 wg.Wait()
268 if attempts.Load() != 1 {
269 t.Fatalf("reserved %d attempts", attempts.Load())
270 }
271}
272
273func TestClaimLossPagesRejectBadEvidenceAndContinue(t *testing.T) {
274 s, raw, signer, pages := lossPageWorld(t)
275 if _, err := raw.Put(t.Context(), "control/host-loss/bad.json", []byte(`{}`), ""); err != nil {
276 t.Fatal(err)
277 }
278 putLossPageRecord(t, raw, signer, lossPageRecord("valid", true, hostclaim.TimersComplete))
279 if err := s.pageClaimLossesAt(t.Context(), time.Now); !errors.Is(err, hostclaim.ErrLossRecord) {
280 t.Fatalf("invalid signature accepted: %v", err)
281 }
282 if len(*pages) != 1 {
283 t.Fatalf("bad record blocked good record: %v", *pages)
284 }
285}
286
287func TestClaimLossPagesJournalFailureDoesNotSend(t *testing.T) {
288 s, raw, signer, pages := lossPageWorld(t)
289 putLossPageRecord(t, raw, signer, lossPageRecord("loss", true, hostclaim.TimersComplete))
290 if err := s.store.Close(); err != nil {
291 t.Fatal(err)
292 }
293 if err := s.pageClaimLossesAt(t.Context(), time.Now); err == nil {
294 t.Fatal("journal error hidden")
295 }
296 if len(*pages) != 0 {
297 t.Fatal("sent without durable reservation")
298 }
299}
300
301func waitClaimLossPages(t *testing.T, s *Server) {
302 t.Helper()
303 s.claimLossPageMu.Lock()
304 done := s.claimLossPageDone
305 s.claimLossPageMu.Unlock()
306 if done == nil {
307 t.Fatal("no loss paging worker scheduled")
308 }
309 select {
310 case <-done:
311 case <-time.After(5 * time.Second):
312 t.Fatal("loss paging worker did not finish")
313 }
314}
315
316func TestClaimLossPagesHourlyToDailyAcrossReopen(t *testing.T) {
317 s, raw, signer, pages := lossPageWorld(t)
318 r := lossPageRecord("old-loss", true, hostclaim.TimersUnknown)
319 putLossPageRecord(t, raw, signer, r)
320 first := r.AcceptedAt.Add(30 * 24 * time.Hour)
321 pass := func(at time.Time, want int) {
322 t.Helper()
323 if err := s.pageClaimLossesAt(t.Context(), func() time.Time { return at }); err != nil {
324 t.Fatal(err)
325 }
326 if len(*pages) != want {
327 t.Fatalf("at %s: got %d pages, want %d", at.Sub(first), len(*pages), want)
328 }
329 }
330 for hour := 0; hour < 24; hour++ {
331 pass(first.Add(time.Duration(hour)*time.Hour), 2*(hour+1))
332 }
333 path := s.store.gate.path
334 if err := s.store.Close(); err != nil {
335 t.Fatal(err)
336 }
337 st, err := OpenGated(path)
338 if err != nil {
339 t.Fatal(err)
340 }
341 t.Cleanup(func() { st.Close() })
342 s = &Server{cfg: s.cfg, store: st}
343 pass(first.Add(24*time.Hour-time.Nanosecond), 48)
344 // Once the first 24 hours end, the last attempt must be a day old.
345 pass(first.Add(24*time.Hour), 48)
346 pass(first.Add(47*time.Hour-time.Nanosecond), 48)
347 pass(first.Add(47*time.Hour), 50)
348 pass(first.Add(71*time.Hour-time.Nanosecond), 50)
349 pass(first.Add(71*time.Hour), 52)
350 for _, condition := range []string{"pending", "timers"} {
351 var saved int64
352 if err := st.db.QueryRow(`SELECT first_attempted_at FROM claim_loss_pages WHERE rule = ?`, condition+" "+r.ID).Scan(&saved); err != nil || saved != first.UnixNano() {
353 t.Fatalf("first %s attempt changed: %d %v", condition, saved, err)
354 }
355 }
356}
357
358func TestClaimLossPagesClockAtEachReservation(t *testing.T) {
359 s, raw, signer, _ := lossPageWorld(t)
360 for _, id := range []string{"a-early", "b-late"} {
361 putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
362 }
363 first := time.Now()
364 now := first
365 late := 0
366 calls := 0
367 alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(_ string, b []byte) error {
368 calls++
369 if strings.Contains(string(b), "b-late") {
370 late++
371 }
372 if calls == 1 {
373 now = now.Add(time.Hour)
374 }
375 return nil
376 }})
377 clock := func() time.Time { return now }
378 if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
379 t.Fatal(err)
380 }
381 now = now.Add(10 * time.Second)
382 if err := s.pageClaimLossesAt(t.Context(), clock); err != nil {
383 t.Fatal(err)
384 }
385 if late != 1 || calls != 3 {
386 t.Fatalf("late record retried ten seconds later: late=%d, total=%d", late, calls)
387 }
388 var reserved int64
389 if err := s.store.db.QueryRow(`SELECT attempted_at FROM claim_loss_pages WHERE rule = 'pending b-late'`).Scan(&reserved); err != nil || reserved != first.Add(time.Hour).UnixNano() {
390 t.Fatalf("late reservation: %d %v", reserved, err)
391 }
392}
393
394func TestClaimLossPagesBlockedSinkDoesNotBlockRecovery(t *testing.T) {
395 s, raw, signer, _ := lossPageWorld(t)
396 putLossPageRecord(t, raw, signer, lossPageRecord("blocking", true, hostclaim.TimersComplete))
397 entered, releaseSink := make(chan struct{}), make(chan struct{})
398 var unblock sync.Once
399 var calls atomic.Int64
400 alerting.Arm(&alerting.Sinks{WebhookURL: "http://unused", Post: func(string, []byte) error {
401 if calls.Add(1) == 1 {
402 close(entered)
403 <-releaseSink
404 }
405 return nil
406 }})
407 t.Cleanup(func() {
408 unblock.Do(func() { close(releaseSink) })
409 waitClaimLossPages(t, s)
410 })
411 var records, purges, admissions atomic.Int64
412 wrap := func(fn func(context.Context) error, count *atomic.Int64) func(context.Context) error {
413 return func(ctx context.Context) error { count.Add(1); return fn(ctx) }
414 }
415 r := s.cfg.HostClaimRecovery
416 r.SweepRecords = wrap(r.SweepRecords, &records)
417 r.SweepPurges = wrap(r.SweepPurges, &purges)
418 r.SweepAdmissions = wrap(r.SweepAdmissions, &admissions)
419 pass := func() {
420 t.Helper()
421 done := make(chan error, 1)
422 go func() { done <- s.RunClaimRecoveryPass(t.Context()) }()
423 select {
424 case err := <-done:
425 if err != nil {
426 t.Fatal(err)
427 }
428 case <-time.After(2 * time.Second):
429 unblock.Do(func() { close(releaseSink) })
430 <-done
431 t.Fatal("paging blocked recovery")
432 }
433 }
434 pass()
435 select {
436 case <-entered:
437 case <-time.After(2 * time.Second):
438 t.Fatal("sink was not called")
439 }
440 s.claimLossPageMu.Lock()
441 worker := s.claimLossPageDone
442 s.claimLossPageMu.Unlock()
443 for range 3 {
444 pass()
445 }
446 if records.Load() != 4 || purges.Load() != 4 || admissions.Load() != 4 {
447 t.Fatalf("sweeps delayed: records=%d purges=%d admissions=%d", records.Load(), purges.Load(), admissions.Load())
448 }
449 s.claimLossPageMu.Lock()
450 overlapped := s.claimLossPageDone != worker
451 s.claimLossPageMu.Unlock()
452 if overlapped || calls.Load() != 1 {
453 t.Fatal("paging workers overlapped")
454 }
455 unblock.Do(func() { close(releaseSink) })
456 waitClaimLossPages(t, s)
457 putLossPageRecord(t, raw, signer, lossPageRecord("next-worker", true, hostclaim.TimersComplete))
458 pass()
459 waitClaimLossPages(t, s)
460 if calls.Load() != 2 {
461 t.Fatal("next recovery did not schedule a new paging worker")
462 }
463}
464
465func TestClaimLossPageJournalUpgrade(t *testing.T) {
466 path := filepath.Join(t.TempDir(), "old.db")
467 st, err := Open(path)
468 if err != nil {
469 t.Fatal(err)
470 }
471 at := time.Now().Add(-23 * time.Hour)
472 _, err = st.db.Exec(`CREATE TABLE claim_loss_pages (rule TEXT PRIMARY KEY, attempted_at INTEGER NOT NULL);
473INSERT INTO claim_loss_pages VALUES ('pending old', ?)`, at.UnixNano())
474 if err != nil {
475 t.Fatal(err)
476 }
477 if err := st.Close(); err != nil {
478 t.Fatal(err)
479 }
480 for range 2 {
481 st, err := OpenGated(path)
482 if err != nil {
483 t.Fatal(err)
484 }
485 var first, last int64
486 err = st.db.QueryRow(`SELECT first_attempted_at, attempted_at FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&first, &last)
487 var cursor string
488 cursorErr := st.db.QueryRow(`SELECT last_key FROM claim_loss_pages WHERE rule = 'pending old'`).Scan(&cursor)
489 st.Close()
490 if err != nil || first != at.UnixNano() || last != at.UnixNano() {
491 t.Fatalf("upgrade lost durable attempt: %d %d %v", first, last, err)
492 }
493 if cursorErr != nil || cursor != "" {
494 t.Fatalf("upgrade scan progress: %q %v", cursor, cursorErr)
495 }
496 }
497}
498
499type lossPageDeadlineStore struct {
500 release.Store
501 deadline chan time.Time
502}
503
504func (s lossPageDeadlineStore) List(ctx context.Context, prefix string) ([]string, error) {
505 deadline, _ := ctx.Deadline()
506 s.deadline <- deadline
507 <-ctx.Done()
508 return nil, ctx.Err()
509}
510
511func TestClaimLossPagesWorkerBudget(t *testing.T) {
512 s, _, _, pages := lossPageWorld(t)
513 deadline := make(chan time.Time, 1)
514 s.cfg.Releases = lossPageDeadlineStore{Store: s.cfg.Releases, deadline: deadline}
515 ctx, cancel := context.WithCancel(t.Context())
516 defer cancel()
517 before := time.Now()
518 s.scheduleClaimLossPages(ctx)
519 select {
520 case at := <-deadline:
521 if at.Before(before.Add(claimLossPageBudget)) || at.After(time.Now().Add(claimLossPageBudget)) {
522 t.Errorf("scan has no per-pass budget: deadline=%v", at)
523 }
524 case <-time.After(time.Second):
525 t.Error("worker never started discovery")
526 }
527 cancel()
528 waitClaimLossPages(t, s)
529 if len(*pages) != 0 {
530 t.Fatal("canceled scan delivered a page")
531 }
532 s.claimLossPageMu.Lock()
533 done := s.claimLossPageDone
534 s.claimLossPageMu.Unlock()
535 s.scheduleClaimLossPages(ctx)
536 if s.claimLossPageDone != done {
537 t.Fatal("canceled recovery scheduled another worker")
538 }
539}
540
541type lossPageBlockingSNS struct {
542 entered chan time.Time
543}
544
545func (p lossPageBlockingSNS) Publish(ctx context.Context, _ *sns.PublishInput, _ ...func(*sns.Options)) (*sns.PublishOutput, error) {
546 deadline, _ := ctx.Deadline()
547 p.entered <- deadline
548 <-ctx.Done()
549 return nil, ctx.Err()
550}
551
552func TestClaimLossPagesSinkTimeoutAndCancellation(t *testing.T) {
553 s, raw, signer, _ := lossPageWorld(t)
554 putLossPageRecord(t, raw, signer, lossPageRecord("blocked-sns", true, hostclaim.TimersUnknown))
555 entered := make(chan time.Time, 1)
556 alerting.Arm(&alerting.Sinks{TopicARN: "test", Pub: lossPageBlockingSNS{entered: entered}})
557 ctx, cancel := context.WithCancel(t.Context())
558 defer cancel()
559 s.scheduleClaimLossPages(ctx)
560 select {
561 case deadline := <-entered:
562 if deadline.IsZero() || time.Until(deadline) > alerting.PublishTimeout {
563 t.Errorf("SNS sink has no timeout: %v", deadline)
564 }
565 case <-time.After(time.Second):
566 t.Error("worker never called SNS")
567 }
568 cancel()
569 waitClaimLossPages(t, s)
570 var attempts int
571 if err := s.store.db.QueryRow(`SELECT count(*) FROM claim_loss_pages`).Scan(&attempts); err != nil || attempts != 1 {
572 t.Fatalf("canceled pass reserved another condition: %d %v", attempts, err)
573 }
574}
575
576type lossPageSlowStore struct {
577 release.Store
578 reads []string
579 reverse bool
580}
581
582func (s *lossPageSlowStore) List(ctx context.Context, prefix string) ([]string, error) {
583 keys, err := s.Store.List(ctx, prefix)
584 if s.reverse {
585 slices.Reverse(keys)
586 }
587 return keys, err
588}
589
590func (s *lossPageSlowStore) Get(ctx context.Context, key string) ([]byte, string, error) {
591 s.reads = append(s.reads, key)
592 if len(s.reads) > 2 {
593 <-ctx.Done()
594 return nil, "", ctx.Err()
595 }
596 timer := time.NewTimer(10 * time.Millisecond)
597 defer timer.Stop()
598 select {
599 case <-timer.C:
600 return s.Store.Get(ctx, key)
601 case <-ctx.Done():
602 return nil, "", ctx.Err()
603 }
604}
605
606func TestClaimLossPagesScanProgressSurvivesRestartAndWraps(t *testing.T) {
607 s, raw, signer, pages := lossPageWorld(t)
608 for _, id := range []string{"b-retained", "c-retained", "d-retained", "e-retained", "f-retained", "z-pending"} {
609 putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "z-pending", hostclaim.TimersComplete))
610 }
611 slow := &lossPageSlowStore{Store: raw}
612 s.cfg.Releases = slow
613 pass := func(first string) {
614 t.Helper()
615 slow.reads = nil
616 ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
617 defer cancel()
618 if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
619 t.Fatalf("slow scan: %v", err)
620 }
621 key, _ := hostclaim.LossRecordKey(first)
622 if len(slow.reads) != 3 || slow.reads[0] != key {
623 t.Fatalf("scan reads: %v; want first %s and one unfinished read", slow.reads, key)
624 }
625 }
626 pass("b-retained")
627 if len(*pages) != 0 {
628 t.Fatalf("first scan reached beyond its budget: %v", *pages)
629 }
630 path := s.store.gate.path
631 if err := s.store.Close(); err != nil {
632 t.Fatal(err)
633 }
634 st, err := OpenGated(path)
635 if err != nil {
636 t.Fatal(err)
637 }
638 t.Cleanup(func() { st.Close() })
639 s = &Server{cfg: s.cfg, store: st}
640 pass("d-retained")
641 pass("f-retained")
642 if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "z-pending") {
643 t.Fatalf("pending record not delivered within three scans: %v", *pages)
644 }
645 putLossPageRecord(t, raw, signer, lossPageRecord("a-new", true, hostclaim.TimersComplete))
646 key, _ := hostclaim.LossRecordKey("z-pending")
647 if err := raw.Delete(t.Context(), key); err != nil {
648 t.Fatal(err)
649 }
650 slow.reverse = true
651 pass("a-new")
652 if len(*pages) != 2 || !strings.Contains((*pages)[1]["body"], "a-new") {
653 t.Fatalf("scan did not wrap past the removed cursor to the new early record: %v", *pages)
654 }
655}
656
657func TestClaimLossPagesScanRetriesUnfinishedPendingRead(t *testing.T) {
658 s, raw, signer, pages := lossPageWorld(t)
659 for _, id := range []string{"a-retained", "b-retained", "c-pending"} {
660 putLossPageRecord(t, raw, signer, lossPageRecord(id, id == "c-pending", hostclaim.TimersComplete))
661 }
662 s.cfg.Releases = &lossPageSlowStore{Store: raw}
663 ctx, cancel := context.WithTimeout(t.Context(), 250*time.Millisecond)
664 defer cancel()
665 if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
666 t.Fatalf("slow scan: %v", err)
667 }
668 if len(*pages) != 0 {
669 t.Fatalf("unfinished read paged: %v", *pages)
670 }
671 slow := &lossPageSlowStore{Store: raw}
672 s.cfg.Releases = slow
673 ctx, cancel = context.WithTimeout(t.Context(), 250*time.Millisecond)
674 defer cancel()
675 if err := s.pageClaimLossesAt(ctx, time.Now); !errors.Is(err, context.DeadlineExceeded) {
676 t.Fatalf("resumed scan: %v", err)
677 }
678 key, _ := hostclaim.LossRecordKey("c-pending")
679 if !slices.Equal(slow.reads, []string{key, "control/host-loss/a-retained.json", "control/host-loss/b-retained.json"}) || len(*pages) != 1 {
680 t.Fatalf("unfinished pending read skipped: reads=%v pages=%v", slow.reads, *pages)
681 }
682}
683
684// lossPageStalledStore never answers a read of one key.
685type lossPageStalledStore struct {
686 release.Store
687 stalled string
688 reads []string
689}
690
691func (s *lossPageStalledStore) Get(ctx context.Context, key string) ([]byte, string, error) {
692 s.reads = append(s.reads, key)
693 if key == s.stalled {
694 <-ctx.Done()
695 return nil, "", ctx.Err()
696 }
697 return s.Store.Get(ctx, key)
698}
699
700func TestClaimLossPagesPersistentlyStalledReadDoesNotStarve(t *testing.T) {
701 prior := claimLossReadBudget
702 claimLossReadBudget = 50 * time.Millisecond
703 t.Cleanup(func() { claimLossReadBudget = prior })
704 s, raw, signer, pages := lossPageWorld(t)
705 for _, id := range []string{"a-stalled", "b-pending"} {
706 putLossPageRecord(t, raw, signer, lossPageRecord(id, true, hostclaim.TimersComplete))
707 }
708 stalledKey, _ := hostclaim.LossRecordKey("a-stalled")
709 pendingKey, _ := hostclaim.LossRecordKey("b-pending")
710 store := &lossPageStalledStore{Store: raw, stalled: stalledKey}
711 s.cfg.Releases = store
712 for lap := 0; lap < 2; lap++ {
713 store.reads = nil
714 ctx, cancel := context.WithTimeout(t.Context(), 2*time.Second)
715 err := s.pageClaimLossesAt(ctx, time.Now)
716 cancel()
717 if err == nil || !errors.Is(err, context.DeadlineExceeded) {
718 t.Fatalf("lap %d: the stalled read should surface as a record error, got %v", lap, err)
719 }
720 if !slices.Contains(store.reads, pendingKey) {
721 t.Fatalf("lap %d: the record behind the stalled one was never read: %v", lap, store.reads)
722 }
723 if !slices.Contains(store.reads, stalledKey) {
724 t.Fatalf("lap %d: the stalled record was not retried: %v", lap, store.reads)
725 }
726 }
727 if len(*pages) != 1 || !strings.Contains((*pages)[0]["body"], "b-pending") {
728 t.Fatalf("pending record behind a stalled read was not paged exactly once: %v", *pages)
729 }
730}
731