| 1 | //go:build unix |
| 2 | |
| 3 | package hostfiles |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "errors" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | type fakeWatcher struct { |
| 20 | ch chan event |
| 21 | mu sync.Mutex |
| 22 | degraded bool |
| 23 | repairs int |
| 24 | repairErr error // when nil, Repair succeeds and clears degraded |
| 25 | } |
| 26 | |
| 27 | func newFake() *fakeWatcher { return &fakeWatcher{ch: make(chan event, 64)} } |
| 28 | |
| 29 | func (f *fakeWatcher) Events() <-chan event { return f.ch } |
| 30 | func (f *fakeWatcher) Degraded() bool { f.mu.Lock(); defer f.mu.Unlock(); return f.degraded } |
| 31 | func (f *fakeWatcher) Close() error { return nil } |
| 32 | func (f *fakeWatcher) Repair() error { |
| 33 | f.mu.Lock() |
| 34 | defer f.mu.Unlock() |
| 35 | f.repairs++ |
| 36 | if f.repairErr == nil { |
| 37 | f.degraded = false |
| 38 | } |
| 39 | return f.repairErr |
| 40 | } |
| 41 | func (f *fakeWatcher) setDegraded(v bool) { f.mu.Lock(); f.degraded = v; f.mu.Unlock() } |
| 42 | |
| 43 | // eventsOnly publishes on events and nothing else: no repair scan and no |
| 44 | // polling can satisfy a test that uses it. polled is the opposite. |
| 45 | var ( |
| 46 | eventsOnly = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: time.Hour} |
| 47 | polled = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: 40 * time.Millisecond} |
| 48 | ) |
| 49 | |
| 50 | // startRun runs the loop and returns once its first full scan has |
| 51 | // finished, so nothing a test does afterwards can be found by that scan. |
| 52 | func startRun(t *testing.T, p *Publisher, dir string, w watcher, o Options) (stop func() error) { |
| 53 | t.Helper() |
| 54 | first := make(chan struct{}) |
| 55 | var once sync.Once |
| 56 | o.OnStats = func(SyncStats) { once.Do(func() { close(first) }) } |
| 57 | ctx, cancel := context.WithCancel(context.Background()) |
| 58 | done := make(chan error, 1) |
| 59 | go func() { done <- run(ctx, p, dir, o, w) }() |
| 60 | select { |
| 61 | case <-first: |
| 62 | case err := <-done: |
| 63 | t.Fatalf("run ended before its first scan: %v", err) |
| 64 | case <-time.After(2 * time.Second): |
| 65 | t.Fatal("run never finished its first scan") |
| 66 | } |
| 67 | return func() error { cancel(); return <-done } |
| 68 | } |
| 69 | |
| 70 | // startRunCounting is startRun with a callback on every full scan. |
| 71 | func startRunCounting(t *testing.T, p *Publisher, dir string, w watcher, o Options, onScan func()) (stop func() error) { |
| 72 | t.Helper() |
| 73 | first := make(chan struct{}) |
| 74 | var once sync.Once |
| 75 | o.OnStats = func(SyncStats) { onScan(); once.Do(func() { close(first) }) } |
| 76 | ctx, cancel := context.WithCancel(context.Background()) |
| 77 | done := make(chan error, 1) |
| 78 | go func() { done <- run(ctx, p, dir, o, w) }() |
| 79 | select { |
| 80 | case <-first: |
| 81 | case <-time.After(2 * time.Second): |
| 82 | t.Fatal("run never finished its first scan") |
| 83 | } |
| 84 | return func() error { cancel(); return <-done } |
| 85 | } |
| 86 | |
| 87 | // eventually polls until ok returns true or two seconds pass. |
| 88 | func eventually(t *testing.T, what string, ok func() bool) { |
| 89 | t.Helper() |
| 90 | deadline := time.Now().Add(2 * time.Second) |
| 91 | for time.Now().Before(deadline) { |
| 92 | if ok() { |
| 93 | return |
| 94 | } |
| 95 | time.Sleep(10 * time.Millisecond) |
| 96 | } |
| 97 | t.Fatalf("timed out waiting for: %s", what) |
| 98 | } |
| 99 | |
| 100 | func manifestHas(s release.Store, path, sha string) bool { |
| 101 | m, found, err := Load(context.Background(), s, pfx) |
| 102 | if err != nil || !found { |
| 103 | return false |
| 104 | } |
| 105 | e, ok := m.Entries[path] |
| 106 | return ok && !e.Deleted && e.SHA256 == sha |
| 107 | } |
| 108 | |
| 109 | func TestRunPublishesATouchedFileAfterTheDeadline(t *testing.T) { |
| 110 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 111 | p := open(t, s) |
| 112 | stop := startRun(t, p, dir, w, eventsOnly) |
| 113 | defer stop() |
| 114 | write(t, dir, "instance.key", "k1") |
| 115 | time.Sleep(3 * eventsOnly.Deadline) |
| 116 | if manifestHas(s, "instance.key", sum([]byte("k1"))) { |
| 117 | t.Fatal("published with no event: something other than the event path found it") |
| 118 | } |
| 119 | touched := time.Now() |
| 120 | w.ch <- event{rel: "instance.key"} |
| 121 | eventually(t, "instance.key published", func() bool { return manifestHas(s, "instance.key", sum([]byte("k1"))) }) |
| 122 | if took := time.Since(touched); took < eventsOnly.Deadline/2 { |
| 123 | t.Fatalf("published %s after the event; the deadline is %s", took, eventsOnly.Deadline) |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func TestRunPublishesADeleteAsATombstone(t *testing.T) { |
| 128 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 129 | p := open(t, s) |
| 130 | write(t, dir, "a", "a") |
| 131 | stop := startRun(t, p, dir, w, eventsOnly) // the first scan publishes "a" |
| 132 | defer stop() |
| 133 | eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) }) |
| 134 | if err := os.Remove(filepath.Join(dir, "a")); err != nil { |
| 135 | t.Fatal(err) |
| 136 | } |
| 137 | w.ch <- event{rel: "a"} |
| 138 | eventually(t, "a tombstoned", func() bool { |
| 139 | m, _, _ := Load(context.Background(), s, pfx) |
| 140 | return m.Entries["a"].Deleted |
| 141 | }) |
| 142 | } |
| 143 | |
| 144 | func TestRunRescansOnOverflow(t *testing.T) { |
| 145 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 146 | p := open(t, s) |
| 147 | stop := startRun(t, p, dir, w, eventsOnly) |
| 148 | defer stop() |
| 149 | write(t, dir, "missed.txt", "m") // no event for it: the queue overflowed |
| 150 | time.Sleep(3 * eventsOnly.Deadline) |
| 151 | if manifestHas(s, "missed.txt", sum([]byte("m"))) { |
| 152 | t.Fatal("found before the rescan was asked for") |
| 153 | } |
| 154 | w.ch <- event{rescan: true} |
| 155 | eventually(t, "missed.txt found by the rescan", func() bool { return manifestHas(s, "missed.txt", sum([]byte("m"))) }) |
| 156 | } |
| 157 | |
| 158 | // A2-5 residual: with a watch missing, changes are found by polling, not |
| 159 | // by waiting for the ten-minute scan. |
| 160 | func TestRunPollsWhileTheWatcherIsDegradedAndKeepsTryingToRepair(t *testing.T) { |
| 161 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 162 | w.setDegraded(true) |
| 163 | w.repairErr = errors.New("watch limit reached") |
| 164 | p := open(t, s) |
| 165 | stop := startRun(t, p, dir, w, polled) |
| 166 | defer stop() |
| 167 | write(t, dir, "quiet.txt", "q") // no event at all |
| 168 | eventually(t, "quiet.txt found by polling", func() bool { return manifestHas(s, "quiet.txt", sum([]byte("q"))) }) |
| 169 | eventually(t, "more than one repair attempt", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.repairs > 1 }) |
| 170 | } |
| 171 | |
| 172 | // The other half: once repair succeeds, polling stops. The loop runs |
| 173 | // with a short polling period, so if it kept polling the scan count would |
| 174 | // keep climbing; it must scan once more for what changed meanwhile and |
| 175 | // then go quiet. |
| 176 | func TestRunScansOnceAfterARepairThenStopsPolling(t *testing.T) { |
| 177 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 178 | w.setDegraded(true) |
| 179 | w.repairErr = errors.New("not yet") |
| 180 | p := open(t, s) |
| 181 | scans := 0 |
| 182 | var mu sync.Mutex |
| 183 | count := func() int { mu.Lock(); defer mu.Unlock(); return scans } |
| 184 | stop := startRunCounting(t, p, dir, w, polled, func() { mu.Lock(); scans++; mu.Unlock() }) |
| 185 | defer stop() |
| 186 | |
| 187 | // While repair fails, it polls: the count climbs. |
| 188 | n := count() |
| 189 | eventually(t, "polling while degraded", func() bool { return count() >= n+3 }) |
| 190 | |
| 191 | write(t, dir, "during.txt", "d") |
| 192 | w.mu.Lock() |
| 193 | w.repairErr = nil // the next Repair succeeds and clears degraded |
| 194 | w.mu.Unlock() |
| 195 | eventually(t, "the repair", func() bool { return !w.Degraded() }) |
| 196 | eventually(t, "during.txt found by the post-repair scan", func() bool { return manifestHas(s, "during.txt", sum([]byte("d"))) }) |
| 197 | |
| 198 | time.Sleep(2 * polled.Poll) // let any scan already under way finish |
| 199 | n = count() |
| 200 | time.Sleep(8 * polled.Poll) |
| 201 | if got := count(); got != n { |
| 202 | t.Fatalf("%d more scans in eight polling periods after the repair; polling should have stopped", got-n) |
| 203 | } |
| 204 | } |
| 205 | |
| 206 | // A watcher whose reader dies closes its channel. The loop must fall back |
| 207 | // to polling, not spin and not stop publishing. |
| 208 | func TestRunPollsAfterTheWatcherStops(t *testing.T) { |
| 209 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 210 | p := open(t, s) |
| 211 | stop := startRun(t, p, dir, w, polled) |
| 212 | defer stop() |
| 213 | close(w.ch) |
| 214 | write(t, dir, "after.txt", "a") |
| 215 | eventually(t, "after.txt found by polling", func() bool { return manifestHas(s, "after.txt", sum([]byte("a"))) }) |
| 216 | } |
| 217 | |
| 218 | func TestRunWithNoWatcherPolls(t *testing.T) { |
| 219 | s, dir := release.NewDirStore(t.TempDir()), t.TempDir() |
| 220 | p := open(t, s) |
| 221 | stop := startRun(t, p, dir, nil, polled) |
| 222 | defer stop() |
| 223 | write(t, dir, "x", "x") |
| 224 | eventually(t, "x found by polling", func() bool { return manifestHas(s, "x", sum([]byte("x"))) }) |
| 225 | } |
| 226 | |
| 227 | func TestRunStopsWhenSuperseded(t *testing.T) { |
| 228 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 229 | p, other := open(t, s), open(t, s) |
| 230 | ctx, cancel := context.WithCancel(context.Background()) |
| 231 | defer cancel() |
| 232 | done := make(chan error, 1) |
| 233 | go func() { done <- run(ctx, p, dir, eventsOnly, w) }() |
| 234 | stage(t, other, "theirs", "t") |
| 235 | commit(t, other) |
| 236 | write(t, dir, "mine", "m") |
| 237 | w.ch <- event{rel: "mine"} |
| 238 | select { |
| 239 | case err := <-done: |
| 240 | if !errors.Is(err, ErrSuperseded) { |
| 241 | t.Fatalf("err = %v, want ErrSuperseded", err) |
| 242 | } |
| 243 | case <-time.After(2 * time.Second): |
| 244 | t.Fatal("run did not stop after another writer published") |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | func TestRunIgnoresAnEventForAPathThatEscapes(t *testing.T) { |
| 249 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 250 | p := open(t, s) |
| 251 | stop := startRun(t, p, dir, w, eventsOnly) |
| 252 | w.ch <- event{rel: "../outside"} |
| 253 | write(t, dir, "ok", "ok") |
| 254 | w.ch <- event{rel: "ok"} |
| 255 | eventually(t, "ok published", func() bool { return manifestHas(s, "ok", sum([]byte("ok"))) }) |
| 256 | if err := stop(); !errors.Is(err, context.Canceled) { |
| 257 | t.Fatalf("run ended with %v; a bad path must be skipped, not fatal", err) |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | // logCapture is a log.SetOutput target that is safe to write from the loop. |
| 262 | type logCapture struct { |
| 263 | mu sync.Mutex |
| 264 | b strings.Builder |
| 265 | } |
| 266 | |
| 267 | func (l *logCapture) Write(p []byte) (int, error) { |
| 268 | l.mu.Lock() |
| 269 | defer l.mu.Unlock() |
| 270 | return l.b.Write(p) |
| 271 | } |
| 272 | func (l *logCapture) count(sub string) int { |
| 273 | l.mu.Lock() |
| 274 | defer l.mu.Unlock() |
| 275 | return strings.Count(l.b.String(), sub) |
| 276 | } |
| 277 | |
| 278 | func captureLog(t *testing.T) *logCapture { |
| 279 | t.Helper() |
| 280 | c := &logCapture{} |
| 281 | log.SetOutput(c) |
| 282 | t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| 283 | return c |
| 284 | } |
| 285 | |
| 286 | // A repair is a tree walk: a cause that lasts must cost one attempt and |
| 287 | // one log line per polling period at most, not one per tick. |
| 288 | func TestRunRepairsAtMostOncePerPollAndLogsTheTransitionOnce(t *testing.T) { |
| 289 | logs := captureLog(t) |
| 290 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 291 | w.setDegraded(true) |
| 292 | w.repairErr = errors.New("watch limit reached") |
| 293 | p := open(t, s) |
| 294 | stop := startRun(t, p, dir, w, polled) |
| 295 | time.Sleep(20 * polled.Poll) |
| 296 | stop() |
| 297 | w.mu.Lock() |
| 298 | n := w.repairs |
| 299 | w.mu.Unlock() |
| 300 | // Twenty periods; a tick is a quarter of the deadline, so per-tick |
| 301 | // would be ~80. The bounds are wide so a loaded CI runner cannot flake. |
| 302 | if n < 2 || n > 30 { |
| 303 | t.Fatalf("%d repair attempts in twenty polling periods; want about twenty", n) |
| 304 | } |
| 305 | if got := logs.count("watch coverage is missing"); got != 1 { |
| 306 | t.Fatalf("coverage-missing logged %d times, want once", got) |
| 307 | } |
| 308 | if got := logs.count("still missing"); got != 0 { |
| 309 | t.Fatalf("a per-attempt line was logged %d times", got) |
| 310 | } |
| 311 | } |
| 312 | |
| 313 | func TestRunLogsWhenCoverageIsRestored(t *testing.T) { |
| 314 | logs := captureLog(t) |
| 315 | s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake() |
| 316 | w.setDegraded(true) |
| 317 | p := open(t, s) // Repair succeeds on its first attempt |
| 318 | stop := startRun(t, p, dir, w, polled) |
| 319 | eventually(t, "restored line", func() bool { return logs.count("watch coverage restored") == 1 }) |
| 320 | stop() |
| 321 | if got := logs.count("watch coverage is missing"); got != 1 { |
| 322 | t.Fatalf("missing logged %d times, want once", got) |
| 323 | } |
| 324 | } |
| 325 | |
| 326 | // failingPuts refuses every object upload, so no scan of a directory with a |
| 327 | // file in it can finish. |
| 328 | type failingPuts struct { |
| 329 | release.Store |
| 330 | mu sync.Mutex |
| 331 | puts int |
| 332 | } |
| 333 | |
| 334 | func (f *failingPuts) Put(ctx context.Context, key string, body []byte, etag string) (string, error) { |
| 335 | if strings.HasSuffix(key, "manifest.json") { |
| 336 | return f.Store.Put(ctx, key, body, etag) |
| 337 | } |
| 338 | f.mu.Lock() |
| 339 | f.puts++ |
| 340 | f.mu.Unlock() |
| 341 | return "", errors.New("store unavailable") |
| 342 | } |
| 343 | |
| 344 | func TestRunRetriesAFailingScanOncePerPollAndLogsItOnce(t *testing.T) { |
| 345 | logs := captureLog(t) |
| 346 | fs := &failingPuts{Store: release.NewDirStore(t.TempDir())} |
| 347 | dir := t.TempDir() |
| 348 | write(t, dir, "f", "v") |
| 349 | p := open(t, fs) |
| 350 | ctx, cancel := context.WithCancel(context.Background()) |
| 351 | done := make(chan error, 1) |
| 352 | go func() { done <- run(ctx, p, dir, polled, newFake()) }() |
| 353 | time.Sleep(20 * polled.Poll) |
| 354 | cancel() |
| 355 | <-done |
| 356 | fs.mu.Lock() |
| 357 | n := fs.puts |
| 358 | fs.mu.Unlock() |
| 359 | if n < 2 || n > 30 { // per-tick would be ~80; see the repair test |
| 360 | t.Fatalf("%d scan attempts in twenty polling periods; want about twenty", n) |
| 361 | } |
| 362 | if got := logs.count("scan failing"); got != 1 { |
| 363 | t.Fatalf("the failure was logged %d times, want once", got) |
| 364 | } |
| 365 | } |
| 366 | |
| 367 | func TestRunWithZeroOptionsUsesTheDefaultsAndPublishesOnItsFirstScan(t *testing.T) { |
| 368 | s, dir := release.NewDirStore(t.TempDir()), t.TempDir() |
| 369 | write(t, dir, "a", "a") |
| 370 | p := open(t, s) |
| 371 | ctx, cancel := context.WithCancel(context.Background()) |
| 372 | done := make(chan error, 1) |
| 373 | go func() { done <- run(ctx, p, dir, Options{}, nil) }() |
| 374 | eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) }) |
| 375 | cancel() |
| 376 | if err := <-done; !errors.Is(err, context.Canceled) { |
| 377 | t.Fatal(err) |
| 378 | } |
| 379 | } |
| 380 | |
amadan