carlos / platform Public

hostfiles: repair and failing scans retry once per poll and log transitions; zero options take defaults

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude agent, operated by paul@keymail.dev 2501d73ef53011a45224129d1027840387dff1ed parent 243b0c9
3 files changed, +159 −11
  • internal/hostfiles/run.go +36 −7
  • internal/hostfiles/run_test.go +120 −0
  • internal/hostfiles/watch_linux.go +3 −4
diff --git a/internal/hostfiles/run.go b/internal/hostfiles/run.go
index 34bb5d20e..4f9c18245 100644
--- a/internal/hostfiles/run.go
+++ b/internal/hostfiles/run.go
@@ -51,6 +51,16 @@ func Run(ctx context.Context, p *Publisher, dir string, o Options) error {
}
func run(ctx context.Context, p *Publisher, dir string, o Options, w watcher) error {
+ def := DefaultOptions()
+ if o.Deadline <= 0 {
+ o.Deadline = def.Deadline
+ }
+ if o.Scan <= 0 {
+ o.Scan = def.Scan
+ }
+ if o.Poll <= 0 {
+ o.Poll = def.Poll
+ }
root, err := openTree(dir)
if err != nil {
return fmt.Errorf("hostfiles: %s: %w", dir, err)
@@ -66,7 +76,8 @@ func run(ctx context.Context, p *Publisher, dir string, o Options, w watcher) er
defer tick.Stop()
needScan := true // the first pass is a full scan: changes made while nothing watched
- var lastScan time.Time
+ var lastScan, lastScanTry, lastRepair time.Time
+ coverageLost, scanFailing := false, false // for logging on transitions only
for {
select {
case <-ctx.Done():
@@ -89,27 +100,45 @@ func run(ctx context.Context, p *Publisher, dir string, o Options, w watcher) er
}
now := time.Now()
if w != nil && w.Degraded() {
- if err := w.Repair(); err != nil {
- log.Printf("hostfiles: %s: watch coverage still missing: %v", dir, err)
- } else {
- needScan = true // whatever changed while coverage was missing
+ if !coverageLost {
+ coverageLost = true
+ log.Printf("hostfiles: %s: watch coverage is missing: polling every %s until it is restored", dir, o.Poll)
+ }
+ // A repair is a full tree walk: try it once per polling period.
+ if now.Sub(lastRepair) >= o.Poll {
+ lastRepair = now
+ if err := w.Repair(); err == nil {
+ needScan = true // whatever changed while coverage was missing
+ }
}
}
polling := w == nil || w.Degraded()
+ if coverageLost && !polling {
+ coverageLost = false
+ log.Printf("hostfiles: %s: watch coverage restored", dir)
+ }
switch {
case polling && now.Sub(lastScan) >= o.Poll:
needScan = true
case !polling && now.Sub(lastScan) >= o.Scan:
needScan = true
}
- if needScan {
+ if needScan && (!scanFailing || now.Sub(lastScanTry) >= o.Poll) {
+ lastScanTry = now
st, err := syncRoot(ctx, p, root, o.Skip)
if err != nil {
if ctx.Err() != nil {
return ctx.Err()
}
- log.Printf("hostfiles: %v", err)
+ if !scanFailing {
+ scanFailing = true
+ log.Printf("hostfiles: %s: scan failing, retrying every %s: %v", dir, o.Poll, err)
+ }
} else {
+ if scanFailing {
+ scanFailing = false
+ log.Printf("hostfiles: %s: scan recovered", dir)
+ }
needScan, lastScan = false, now
if o.OnStats != nil {
o.OnStats(st)
diff --git a/internal/hostfiles/run_test.go b/internal/hostfiles/run_test.go
index 24bb9174a..5ed8ca61c 100644
--- a/internal/hostfiles/run_test.go
+++ b/internal/hostfiles/run_test.go
@@ -5,8 +5,10 @@ package hostfiles
import (
"context"
"errors"
+ "log"
"os"
"path/filepath"
+ "strings"
"sync"
"testing"
"time"
@@ -255,3 +257,121 @@ func TestRunIgnoresAnEventForAPathThatEscapes(t *testing.T) {
t.Fatalf("run ended with %v; a bad path must be skipped, not fatal", err)
}
}
+
+// logCapture is a log.SetOutput target that is safe to write from the loop.
+type logCapture struct {
+ mu sync.Mutex
+ b strings.Builder
+}
+
+func (l *logCapture) Write(p []byte) (int, error) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.b.Write(p)
+}
+func (l *logCapture) count(sub string) int {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return strings.Count(l.b.String(), sub)
+}
+
+func captureLog(t *testing.T) *logCapture {
+ t.Helper()
+ c := &logCapture{}
+ log.SetOutput(c)
+ t.Cleanup(func() { log.SetOutput(os.Stderr) })
+ return c
+}
+
+// A repair is a tree walk: a cause that lasts must cost one attempt and
+// one log line per polling period at most, not one per tick.
+func TestRunRepairsAtMostOncePerPollAndLogsTheTransitionOnce(t *testing.T) {
+ logs := captureLog(t)
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ w.setDegraded(true)
+ w.repairErr = errors.New("watch limit reached")
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, polled)
+ time.Sleep(10 * polled.Poll)
+ stop()
+ w.mu.Lock()
+ n := w.repairs
+ w.mu.Unlock()
+ if n < 3 || n > 13 { // ten periods; a tick is a quarter of the deadline, so per-tick would be ~40
+ t.Fatalf("%d repair attempts in ten polling periods; want about ten", n)
+ }
+ if got := logs.count("watch coverage is missing"); got != 1 {
+ t.Fatalf("coverage-missing logged %d times, want once", got)
+ }
+ if got := logs.count("still missing"); got != 0 {
+ t.Fatalf("a per-attempt line was logged %d times", got)
+ }
+}
+
+func TestRunLogsWhenCoverageIsRestored(t *testing.T) {
+ logs := captureLog(t)
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ w.setDegraded(true)
+ p := open(t, s) // Repair succeeds on its first attempt
+ stop := startRun(t, p, dir, w, polled)
+ eventually(t, "restored line", func() bool { return logs.count("watch coverage restored") == 1 })
+ stop()
+ if got := logs.count("watch coverage is missing"); got != 1 {
+ t.Fatalf("missing logged %d times, want once", got)
+ }
+}
+
+// failingPuts refuses every object upload, so no scan of a directory with a
+// file in it can finish.
+type failingPuts struct {
+ release.Store
+ mu sync.Mutex
+ puts int
+}
+
+func (f *failingPuts) Put(ctx context.Context, key string, body []byte, etag string) (string, error) {
+ if strings.HasSuffix(key, "manifest.json") {
+ return f.Store.Put(ctx, key, body, etag)
+ }
+ f.mu.Lock()
+ f.puts++
+ f.mu.Unlock()
+ return "", errors.New("store unavailable")
+}
+
+func TestRunRetriesAFailingScanOncePerPollAndLogsItOnce(t *testing.T) {
+ logs := captureLog(t)
+ fs := &failingPuts{Store: release.NewDirStore(t.TempDir())}
+ dir := t.TempDir()
+ write(t, dir, "f", "v")
+ p := open(t, fs)
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- run(ctx, p, dir, polled, newFake()) }()
+ time.Sleep(10 * polled.Poll)
+ cancel()
+ <-done
+ fs.mu.Lock()
+ n := fs.puts
+ fs.mu.Unlock()
+ if n < 3 || n > 13 {
+ t.Fatalf("%d scan attempts in ten polling periods; want about ten", n)
+ }
+ if got := logs.count("scan failing"); got != 1 {
+ t.Fatalf("the failure was logged %d times, want once", got)
+ }
+}
+
+func TestRunWithZeroOptionsUsesTheDefaultsAndPublishesOnItsFirstScan(t *testing.T) {
+ s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
+ write(t, dir, "a", "a")
+ p := open(t, s)
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- run(ctx, p, dir, Options{}, nil) }()
+ eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) })
+ cancel()
+ if err := <-done; !errors.Is(err, context.Canceled) {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/hostfiles/watch_linux.go b/internal/hostfiles/watch_linux.go
index 2d950f8b4..da41c52fb 100644
--- a/internal/hostfiles/watch_linux.go
+++ b/internal/hostfiles/watch_linux.go
@@ -128,10 +128,9 @@ func (w *inotifyWatcher) send(ev event) {
select {
case w.ch <- ev:
default:
- // Our own buffer is full: the same remedy as a kernel overflow.
- // A path was dropped, so only a scan finds it now; and it may
- // have been a new directory, so coverage is uncertain too.
- w.failures.Add(1)
+ // Our own buffer is full. What was dropped is a file path (a new
+ // directory's watch is added before its event is sent), so coverage
+ // is intact; only a scan can find the dropped path now.
select {
case <-w.ch:
default: