| 1 | package hostfiles |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "log" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | "testing" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | // relayLog captures the package's log lines for one test. |
| 15 | type relayLog struct { |
| 16 | mu sync.Mutex |
| 17 | buf bytes.Buffer |
| 18 | } |
| 19 | |
| 20 | func (l *relayLog) Write(p []byte) (int, error) { |
| 21 | l.mu.Lock() |
| 22 | defer l.mu.Unlock() |
| 23 | return l.buf.Write(p) |
| 24 | } |
| 25 | |
| 26 | func (l *relayLog) count(substr string) int { |
| 27 | l.mu.Lock() |
| 28 | defer l.mu.Unlock() |
| 29 | return strings.Count(l.buf.String(), substr) |
| 30 | } |
| 31 | |
| 32 | func captureRelayLog(t *testing.T) *relayLog { |
| 33 | t.Helper() |
| 34 | l := &relayLog{} |
| 35 | prev := log.Writer() |
| 36 | log.SetOutput(l) |
| 37 | t.Cleanup(func() { log.SetOutput(prev) }) |
| 38 | return l |
| 39 | } |
| 40 | |
| 41 | type flakyStager struct { |
| 42 | mu sync.Mutex |
| 43 | fail int |
| 44 | staged [][]byte |
| 45 | block chan struct{} |
| 46 | } |
| 47 | |
| 48 | func (f *flakyStager) StagePlatform(_ context.Context, name string, body []byte) error { |
| 49 | if f.block != nil { |
| 50 | <-f.block |
| 51 | } |
| 52 | f.mu.Lock() |
| 53 | defer f.mu.Unlock() |
| 54 | if f.fail > 0 { |
| 55 | f.fail-- |
| 56 | return errors.New("store unreachable") |
| 57 | } |
| 58 | f.staged = append(f.staged, append([]byte(nil), body...)) |
| 59 | return nil |
| 60 | } |
| 61 | |
| 62 | func (f *flakyStager) last() []byte { |
| 63 | f.mu.Lock() |
| 64 | defer f.mu.Unlock() |
| 65 | if len(f.staged) == 0 { |
| 66 | return nil |
| 67 | } |
| 68 | return f.staged[len(f.staged)-1] |
| 69 | } |
| 70 | |
| 71 | // Ruling 22: Publish never waits on the store; Run stages the latest body, |
| 72 | // retrying a failed stage, and stages nothing again once it has. Two |
| 73 | // failed stages log one "failing" line and the stage that follows one |
| 74 | // "recovered" line: transitions, as run.go logs a scan, not every retry. |
| 75 | func TestPlatformRelayStagesTheLatestBodyWithoutBlockingItsCaller(t *testing.T) { |
| 76 | logs := captureRelayLog(t) |
| 77 | st := &flakyStager{fail: 2, block: make(chan struct{})} |
| 78 | r := &PlatformRelay{Name: "instance-key-adoption.json", Stage: st, Retry: 10 * time.Millisecond} |
| 79 | ctx, cancel := context.WithCancel(context.Background()) |
| 80 | defer cancel() |
| 81 | done := make(chan error, 1) |
| 82 | go func() { done <- r.Run(ctx) }() |
| 83 | published := make(chan struct{}) |
| 84 | go func() { r.Publish([]byte("one")); r.Publish([]byte("two")); close(published) }() |
| 85 | select { |
| 86 | case <-published: |
| 87 | case <-time.After(time.Second): |
| 88 | t.Fatal("Publish waited on a store that does not answer") |
| 89 | } |
| 90 | close(st.block) |
| 91 | deadline := time.Now().Add(2 * time.Second) |
| 92 | for string(st.last()) != "two" { |
| 93 | if time.Now().After(deadline) { |
| 94 | t.Fatalf("staged %q, want the latest body after two failed stages", st.last()) |
| 95 | } |
| 96 | time.Sleep(5 * time.Millisecond) |
| 97 | } |
| 98 | time.Sleep(50 * time.Millisecond) |
| 99 | st.mu.Lock() |
| 100 | n := len(st.staged) |
| 101 | st.mu.Unlock() |
| 102 | if n != 1 { |
| 103 | t.Fatalf("staged %d times, want once", n) |
| 104 | } |
| 105 | if failing, recovered := logs.count(r.Name+": stage failing"), logs.count(r.Name+": stage recovered"); failing != 1 || recovered != 1 { |
| 106 | t.Fatalf("logged %d failing and %d recovered lines across two failed stages, want one of each", failing, recovered) |
| 107 | } |
| 108 | cancel() |
| 109 | if err := <-done; !errors.Is(err, context.Canceled) { |
| 110 | t.Fatalf("Run = %v", err) |
| 111 | } |
| 112 | } |
| 113 | |
amadan