amadan

carlos / platform Public

hostfiles: a changed file is published a second after its first change, open or not

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 243b0c9ca9583b5eb4cf3dac7a57ff646659ae2c parent a550cbd
9 files changed, +1124 −3
  • internal/hostfiles/deadline.go +36 −0
  • internal/hostfiles/deadline_test.go +38 −0
  • internal/hostfiles/dir_unix.go +14 −3
  • internal/hostfiles/publish_unix_test.go +135 −0
  • internal/hostfiles/run.go +184 −0
  • internal/hostfiles/run_test.go +257 −0
  • internal/hostfiles/watch_linux.go +218 −0
  • internal/hostfiles/watch_linux_test.go +232 −0
  • internal/hostfiles/watch_other.go +10 −0
diff --git a/internal/hostfiles/deadline.go b/internal/hostfiles/deadline.go
new file mode 100644
index 000000000..85e115ee1
--- /dev/null
+++ b/internal/hostfiles/deadline.go
@@ -0,0 +1,36 @@
+package hostfiles
+
+import (
+ "sort"
+ "time"
+)
+
+// queue holds, per path, the time by which its first unpublished
+// modification must be published. A later modification of a path already
+// waiting does not move its deadline, so a file written without pause is
+// still published every period.
+type queue struct {
+ d time.Duration
+ due map[string]time.Time
+}
+
+func newQueue(d time.Duration) *queue { return &queue{d: d, due: map[string]time.Time{}} }
+
+func (q *queue) Touch(rel string, now time.Time) {
+ if _, waiting := q.due[rel]; !waiting {
+ q.due[rel] = now.Add(q.d)
+ }
+}
+
+// Due removes and returns every path whose deadline has arrived.
+func (q *queue) Due(now time.Time) []string {
+ var out []string
+ for rel, at := range q.due {
+ if !at.After(now) {
+ out = append(out, rel)
+ delete(q.due, rel)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
diff --git a/internal/hostfiles/deadline_test.go b/internal/hostfiles/deadline_test.go
new file mode 100644
index 000000000..6c5f607df
--- /dev/null
+++ b/internal/hostfiles/deadline_test.go
@@ -0,0 +1,38 @@
+package hostfiles
+
+import (
+ "reflect"
+ "testing"
+ "time"
+)
+
+func TestQueueArmsOnTheFirstTouchAndLaterTouchesDoNotPushItBack(t *testing.T) {
+ q := newQueue(time.Second)
+ t0 := time.Unix(1000, 0)
+ q.Touch("a", t0)
+ q.Touch("a", t0.Add(900*time.Millisecond)) // still being written
+ if got := q.Due(t0.Add(999 * time.Millisecond)); len(got) != 0 {
+ t.Fatalf("due early: %v", got)
+ }
+ if got := q.Due(t0.Add(time.Second)); !reflect.DeepEqual(got, []string{"a"}) {
+ t.Fatalf("due = %v, want [a]", got)
+ }
+ if got := q.Due(t0.Add(2 * time.Second)); len(got) != 0 {
+ t.Fatalf("due twice: %v", got)
+ }
+ q.Touch("a", t0.Add(3*time.Second)) // a new modification arms a new deadline
+ if got := q.Due(t0.Add(4 * time.Second)); !reflect.DeepEqual(got, []string{"a"}) {
+ t.Fatalf("second deadline: %v", got)
+ }
+}
+
+func TestQueueReturnsDuePathsSorted(t *testing.T) {
+ q := newQueue(time.Second)
+ t0 := time.Unix(1000, 0)
+ for _, p := range []string{"c", "a", "b"} {
+ q.Touch(p, t0)
+ }
+ if got := q.Due(t0.Add(time.Second)); !reflect.DeepEqual(got, []string{"a", "b", "c"}) {
+ t.Fatalf("due = %v", got)
+ }
+}
diff --git a/internal/hostfiles/dir_unix.go b/internal/hostfiles/dir_unix.go
index 0ed6696f7..706756567 100644
--- a/internal/hostfiles/dir_unix.go
+++ b/internal/hostfiles/dir_unix.go
@@ -7,6 +7,7 @@ import (
"io"
"os"
"sort"
+ "strconv"
"strings"
"golang.org/x/sys/unix"
@@ -82,12 +83,22 @@ func notADirectory(err error) bool {
// descend opens the directory rel ("" or "a/b") beneath d, one component
// at a time. gone is true when some component is missing or is not a real
// directory: nothing of ours lives at a path under it.
+//
+// A component that is "", "." or ".." is refused with an error, never
+// reported as gone: publishOne relies on descend never leaving the tree.
+// descend("") returns d itself, so the caller must not close it.
func (d *dirHandle) descend(rel string) (sub *dirHandle, gone bool, err error) {
- cur, owned := d, false
- if rel == "" || rel == "." {
+ if rel == "" {
return d, false, nil
}
- for _, part := range strings.Split(rel, "/") {
+ parts := strings.Split(rel, "/")
+ for _, part := range parts {
+ if part == "" || part == "." || part == ".." {
+ return nil, false, errors.New("hostfiles: path component " + strconv.Quote(part) + " in " + strconv.Quote(rel) + " is not a plain name")
+ }
+ }
+ cur, owned := d, false
+ for _, part := range parts {
betweenLookAndOpen()
next, err := cur.openDir(part)
if owned {
diff --git a/internal/hostfiles/publish_unix_test.go b/internal/hostfiles/publish_unix_test.go
new file mode 100644
index 000000000..83d43f3fa
--- /dev/null
+++ b/internal/hostfiles/publish_unix_test.go
@@ -0,0 +1,135 @@
+//go:build unix
+
+package hostfiles
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// A parent directory swapped for a symlink, to outside or to an excluded
+// directory inside, must not let a queued path be read through it.
+func TestPublishOneNeverReadsThroughASwappedAncestor(t *testing.T) {
+ for name, target := range map[string]string{"a link to outside": "OUTSIDE", "a link to an excluded directory inside": "cache"} {
+ t.Run(name, func(t *testing.T) {
+ ctx, s, src, outside := context.Background(), release.NewDirStore(t.TempDir()), t.TempDir(), t.TempDir()
+ write(t, src, "sub/x", "ours")
+ write(t, src, "cache/x", "excluded")
+ if err := os.WriteFile(filepath.Join(outside, "x"), []byte("outside secret"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ skip := func(rel string, isDir bool) bool { return isDir && rel == "cache" }
+ p := open(t, s)
+ if _, err := SyncDir(ctx, p, src, skip); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ if target == "OUTSIDE" {
+ target = outside
+ }
+ swapAt(t, 1, func() { // before "sub" is opened
+ if err := os.RemoveAll(filepath.Join(src, "sub")); err != nil {
+ t.Error(err)
+ }
+ if err := os.Symlink(target, filepath.Join(src, "sub")); err != nil {
+ t.Error(err)
+ }
+ })
+ if _, err := publishOne(ctx, p, tree(t, src), "sub/x", skip); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ m, _, _ := Load(ctx, s, pfx)
+ if e := m.Entries["sub/x"]; !e.Deleted {
+ t.Fatalf("sub/x = %+v; the directory is gone, so the file is: want a tombstone, and no content read through the link", e)
+ }
+ })
+ }
+}
+
+// A queued path whose file exists but cannot be read is an error, and
+// what is published for it stays.
+func TestPublishOneKeepsAnUnreadableFilesPublishedEntry(t *testing.T) {
+ if os.Getuid() == 0 {
+ t.Skip("root can read anything")
+ }
+ ctx, s, src := context.Background(), release.NewDirStore(t.TempDir()), t.TempDir()
+ write(t, src, "locked", "v1")
+ p := open(t, s)
+ if _, err := SyncDir(ctx, p, src, nil); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ if err := os.Chmod(filepath.Join(src, "locked"), 0); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := publishOne(ctx, p, tree(t, src), "locked", nil); err == nil {
+ t.Fatal("publishOne reported success for a file it could not read")
+ }
+ if p.Dirty() {
+ t.Fatal("publishOne staged a change for a file it could not read")
+ }
+}
+
+// A regular file replaced by another regular file mid-look is read again
+// at the next deadline, not tombstoned.
+func TestPublishOneRetriesARegularFileReplacedMidLook(t *testing.T) {
+ ctx, s, src := context.Background(), release.NewDirStore(t.TempDir()), t.TempDir()
+ write(t, src, "f", "v1")
+ p := open(t, s)
+ if _, err := SyncDir(ctx, p, src, nil); err != nil {
+ t.Fatal(err)
+ }
+ commit(t, p)
+ swapAt(t, 1, func() {
+ tmp := filepath.Join(src, "f.new")
+ if err := os.WriteFile(tmp, []byte("v2"), 0o640); err != nil {
+ t.Error(err)
+ }
+ if err := os.Rename(tmp, filepath.Join(src, "f")); err != nil {
+ t.Error(err)
+ }
+ })
+ again, err := publishOne(ctx, p, tree(t, src), "f", nil)
+ if err != nil || !again || p.Dirty() {
+ t.Fatalf("again = %v, err = %v, dirty = %v; want a retry and nothing staged", again, err, p.Dirty())
+ }
+ again, err = publishOne(ctx, p, tree(t, src), "f", nil)
+ if err != nil || again {
+ t.Fatalf("second look: again = %v, err = %v", again, err)
+ }
+ commit(t, p)
+ if !manifestHas(s, "f", sum([]byte("v2"))) {
+ t.Fatal("the replacement was not published")
+ }
+}
+
+// descend must never leave the tree: a component that is empty, "." or
+// ".." is an error, not a missing directory.
+func TestDescendRefusesAnythingButPlainNames(t *testing.T) {
+ src := t.TempDir()
+ write(t, src, "a/x", "x")
+ write(t, src, "b/x", "x")
+ root := tree(t, src)
+ for _, rel := range []string{"..", "a/../b", "./a", "a//b", ".", "a/.", "/a", "a/"} {
+ sub, gone, err := root.descend(rel)
+ if err == nil || gone || sub != nil {
+ if sub != nil && sub != root {
+ sub.Close()
+ }
+ t.Errorf("descend(%q) = (%v, gone=%v, %v); want an error", rel, sub, gone, err)
+ }
+ }
+ if sub, gone, err := root.descend(""); err != nil || gone || sub != root {
+ t.Errorf("descend(\"\") = (%v, %v, %v); want the root itself", sub, gone, err)
+ }
+ sub, gone, err := root.descend("a")
+ if err != nil || gone || sub == nil {
+ t.Fatalf("descend(a) = (%v, %v, %v)", sub, gone, err)
+ }
+ sub.Close()
+}
diff --git a/internal/hostfiles/run.go b/internal/hostfiles/run.go
new file mode 100644
index 000000000..34bb5d20e
--- /dev/null
+++ b/internal/hostfiles/run.go
@@ -0,0 +1,184 @@
+package hostfiles
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "path"
+ "strings"
+ "time"
+)
+
+type Options struct {
+ Deadline time.Duration // first unpublished modification to publish
+ Scan time.Duration // repair scan while every directory is watched
+ Poll time.Duration // full-scan period while watch coverage is missing
+ Skip func(rel string, isDir bool) bool
+ OnStats func(SyncStats) // called after each full scan; may be nil
+}
+
+// DefaultOptions are region HA spec 1 §5.6's values.
+func DefaultOptions() Options {
+ return Options{Deadline: time.Second, Scan: 10 * time.Minute, Poll: time.Second}
+}
+
+type event struct {
+ rel string
+ rescan bool
+}
+
+// watcher reports changes under one directory. Its event channel closes
+// when it can report nothing more.
+type watcher interface {
+ Events() <-chan event
+ Degraded() bool // some directory may have no working watch
+ Repair() error // restore full coverage, or say why not
+ Close() error
+}
+
+// Run publishes dir's changes until ctx ends (it returns ctx.Err()) or
+// another writer supersedes the manifest (ErrSuperseded). Where there is
+// no watcher, or the watcher fails, it polls.
+func Run(ctx context.Context, p *Publisher, dir string, o Options) error {
+ w, err := newWatcher(dir, o.Skip)
+ if err != nil {
+ log.Printf("hostfiles: no watcher for %s (%v): polling every %s", dir, err, o.Poll)
+ return run(ctx, p, dir, o, nil)
+ }
+ defer w.Close()
+ return run(ctx, p, dir, o, w)
+}
+
+func run(ctx context.Context, p *Publisher, dir string, o Options, w watcher) error {
+ root, err := openTree(dir)
+ if err != nil {
+ return fmt.Errorf("hostfiles: %s: %w", dir, err)
+ }
+ defer root.Close()
+
+ var events <-chan event
+ if w != nil {
+ events = w.Events()
+ }
+ q := newQueue(o.Deadline)
+ tick := time.NewTicker(o.Deadline / 4)
+ defer tick.Stop()
+
+ needScan := true // the first pass is a full scan: changes made while nothing watched
+ var lastScan time.Time
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case ev, open := <-events:
+ if !open {
+ // The watcher has stopped for good. A nil channel never
+ // fires, and with no watcher every pass below polls.
+ log.Printf("hostfiles: watcher for %s stopped: polling every %s", dir, o.Poll)
+ events, w = nil, nil
+ continue
+ }
+ if ev.rescan {
+ needScan = true
+ } else {
+ q.Touch(ev.rel, time.Now())
+ }
+ continue
+ case <-tick.C:
+ }
+ 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
+ }
+ }
+ polling := w == nil || w.Degraded()
+ switch {
+ case polling && now.Sub(lastScan) >= o.Poll:
+ needScan = true
+ case !polling && now.Sub(lastScan) >= o.Scan:
+ needScan = true
+ }
+ if needScan {
+ st, err := syncRoot(ctx, p, root, o.Skip)
+ if err != nil {
+ if ctx.Err() != nil {
+ return ctx.Err()
+ }
+ log.Printf("hostfiles: %v", err)
+ } else {
+ needScan, lastScan = false, now
+ if o.OnStats != nil {
+ o.OnStats(st)
+ }
+ }
+ }
+ for _, rel := range q.Due(now) {
+ again, err := publishOne(ctx, p, root, rel, o.Skip)
+ if err != nil {
+ log.Printf("hostfiles: %s: %v", rel, err)
+ q.Touch(rel, now) // try again at the next deadline
+ continue
+ }
+ if again {
+ q.Touch(rel, now)
+ }
+ }
+ if p.Dirty() {
+ if _, err := p.Commit(ctx); errors.Is(err, ErrSuperseded) {
+ return err
+ } else if err != nil {
+ log.Printf("hostfiles: commit: %v", err) // staged changes stay; the next tick retries
+ }
+ }
+ }
+}
+
+// publishOne stages one path's current state. again is true when the
+// caller should read the path again at the next deadline: the file changed
+// while it was read (its image is published, flagged), or it was replaced
+// mid-look and nothing was read.
+//
+// A tombstone is staged only on proof: the name is gone, or something
+// that is not a regular file now stands there or above it. A file that
+// exists but could not be read is an error, and its manifest entry stays.
+func publishOne(ctx context.Context, p *Publisher, root *dirHandle, rel string, skip func(string, bool) bool) (again bool, err error) {
+ clean, err := CleanPath(rel)
+ if err != nil {
+ return false, nil // a path the watcher should never have produced: skip it
+ }
+ if skipped(skip, clean, false) {
+ return false, nil
+ }
+ parent, name := path.Split(clean)
+ dir, gone, err := root.descend(strings.TrimSuffix(parent, "/"))
+ if err != nil {
+ return false, err
+ }
+ if gone {
+ return false, p.StageDelete(clean)
+ }
+ if dir != root {
+ defer dir.Close()
+ }
+ body, mode, state, err := readStable(dir, name)
+ if err != nil {
+ return false, err
+ }
+ switch state {
+ case fileGone, fileNotRegular:
+ // StageDelete is a no-op for a path the manifest never had, which
+ // is what a directory's own event is.
+ return false, p.StageDelete(clean)
+ case fileUnreadable:
+ return false, fmt.Errorf("%s exists but could not be read; keeping what is published", clean)
+ case fileRetry:
+ return true, nil
+ case fileTooLarge:
+ return false, p.StageSkipped(clean, "too-large")
+ }
+ return state == fileTorn, p.Stage(ctx, clean, body, mode, state == fileTorn)
+}
diff --git a/internal/hostfiles/run_test.go b/internal/hostfiles/run_test.go
new file mode 100644
index 000000000..24bb9174a
--- /dev/null
+++ b/internal/hostfiles/run_test.go
@@ -0,0 +1,257 @@
+//go:build unix
+
+package hostfiles
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type fakeWatcher struct {
+ ch chan event
+ mu sync.Mutex
+ degraded bool
+ repairs int
+ repairErr error // when nil, Repair succeeds and clears degraded
+}
+
+func newFake() *fakeWatcher { return &fakeWatcher{ch: make(chan event, 64)} }
+
+func (f *fakeWatcher) Events() <-chan event { return f.ch }
+func (f *fakeWatcher) Degraded() bool { f.mu.Lock(); defer f.mu.Unlock(); return f.degraded }
+func (f *fakeWatcher) Close() error { return nil }
+func (f *fakeWatcher) Repair() error {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.repairs++
+ if f.repairErr == nil {
+ f.degraded = false
+ }
+ return f.repairErr
+}
+func (f *fakeWatcher) setDegraded(v bool) { f.mu.Lock(); f.degraded = v; f.mu.Unlock() }
+
+// eventsOnly publishes on events and nothing else: no repair scan and no
+// polling can satisfy a test that uses it. polled is the opposite.
+var (
+ eventsOnly = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: time.Hour}
+ polled = Options{Deadline: 40 * time.Millisecond, Scan: time.Hour, Poll: 40 * time.Millisecond}
+)
+
+// startRun runs the loop and returns once its first full scan has
+// finished, so nothing a test does afterwards can be found by that scan.
+func startRun(t *testing.T, p *Publisher, dir string, w watcher, o Options) (stop func() error) {
+ t.Helper()
+ first := make(chan struct{})
+ var once sync.Once
+ o.OnStats = func(SyncStats) { once.Do(func() { close(first) }) }
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- run(ctx, p, dir, o, w) }()
+ select {
+ case <-first:
+ case err := <-done:
+ t.Fatalf("run ended before its first scan: %v", err)
+ case <-time.After(2 * time.Second):
+ t.Fatal("run never finished its first scan")
+ }
+ return func() error { cancel(); return <-done }
+}
+
+// startRunCounting is startRun with a callback on every full scan.
+func startRunCounting(t *testing.T, p *Publisher, dir string, w watcher, o Options, onScan func()) (stop func() error) {
+ t.Helper()
+ first := make(chan struct{})
+ var once sync.Once
+ o.OnStats = func(SyncStats) { onScan(); once.Do(func() { close(first) }) }
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan error, 1)
+ go func() { done <- run(ctx, p, dir, o, w) }()
+ select {
+ case <-first:
+ case <-time.After(2 * time.Second):
+ t.Fatal("run never finished its first scan")
+ }
+ return func() error { cancel(); return <-done }
+}
+
+// eventually polls until ok returns true or two seconds pass.
+func eventually(t *testing.T, what string, ok func() bool) {
+ t.Helper()
+ deadline := time.Now().Add(2 * time.Second)
+ for time.Now().Before(deadline) {
+ if ok() {
+ return
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for: %s", what)
+}
+
+func manifestHas(s release.Store, path, sha string) bool {
+ m, found, err := Load(context.Background(), s, pfx)
+ if err != nil || !found {
+ return false
+ }
+ e, ok := m.Entries[path]
+ return ok && !e.Deleted && e.SHA256 == sha
+}
+
+func TestRunPublishesATouchedFileAfterTheDeadline(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, eventsOnly)
+ defer stop()
+ write(t, dir, "instance.key", "k1")
+ time.Sleep(3 * eventsOnly.Deadline)
+ if manifestHas(s, "instance.key", sum([]byte("k1"))) {
+ t.Fatal("published with no event: something other than the event path found it")
+ }
+ touched := time.Now()
+ w.ch <- event{rel: "instance.key"}
+ eventually(t, "instance.key published", func() bool { return manifestHas(s, "instance.key", sum([]byte("k1"))) })
+ if took := time.Since(touched); took < eventsOnly.Deadline/2 {
+ t.Fatalf("published %s after the event; the deadline is %s", took, eventsOnly.Deadline)
+ }
+}
+
+func TestRunPublishesADeleteAsATombstone(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p := open(t, s)
+ write(t, dir, "a", "a")
+ stop := startRun(t, p, dir, w, eventsOnly) // the first scan publishes "a"
+ defer stop()
+ eventually(t, "a published", func() bool { return manifestHas(s, "a", sum([]byte("a"))) })
+ if err := os.Remove(filepath.Join(dir, "a")); err != nil {
+ t.Fatal(err)
+ }
+ w.ch <- event{rel: "a"}
+ eventually(t, "a tombstoned", func() bool {
+ m, _, _ := Load(context.Background(), s, pfx)
+ return m.Entries["a"].Deleted
+ })
+}
+
+func TestRunRescansOnOverflow(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, eventsOnly)
+ defer stop()
+ write(t, dir, "missed.txt", "m") // no event for it: the queue overflowed
+ time.Sleep(3 * eventsOnly.Deadline)
+ if manifestHas(s, "missed.txt", sum([]byte("m"))) {
+ t.Fatal("found before the rescan was asked for")
+ }
+ w.ch <- event{rescan: true}
+ eventually(t, "missed.txt found by the rescan", func() bool { return manifestHas(s, "missed.txt", sum([]byte("m"))) })
+}
+
+// A2-5 residual: with a watch missing, changes are found by polling, not
+// by waiting for the ten-minute scan.
+func TestRunPollsWhileTheWatcherIsDegradedAndKeepsTryingToRepair(t *testing.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)
+ defer stop()
+ write(t, dir, "quiet.txt", "q") // no event at all
+ eventually(t, "quiet.txt found by polling", func() bool { return manifestHas(s, "quiet.txt", sum([]byte("q"))) })
+ eventually(t, "more than one repair attempt", func() bool { w.mu.Lock(); defer w.mu.Unlock(); return w.repairs > 1 })
+}
+
+// The other half: once repair succeeds, polling stops. The loop runs
+// with a short polling period, so if it kept polling the scan count would
+// keep climbing; it must scan once more for what changed meanwhile and
+// then go quiet.
+func TestRunScansOnceAfterARepairThenStopsPolling(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ w.setDegraded(true)
+ w.repairErr = errors.New("not yet")
+ p := open(t, s)
+ scans := 0
+ var mu sync.Mutex
+ count := func() int { mu.Lock(); defer mu.Unlock(); return scans }
+ stop := startRunCounting(t, p, dir, w, polled, func() { mu.Lock(); scans++; mu.Unlock() })
+ defer stop()
+
+ // While repair fails, it polls: the count climbs.
+ n := count()
+ eventually(t, "polling while degraded", func() bool { return count() >= n+3 })
+
+ write(t, dir, "during.txt", "d")
+ w.mu.Lock()
+ w.repairErr = nil // the next Repair succeeds and clears degraded
+ w.mu.Unlock()
+ eventually(t, "the repair", func() bool { return !w.Degraded() })
+ eventually(t, "during.txt found by the post-repair scan", func() bool { return manifestHas(s, "during.txt", sum([]byte("d"))) })
+
+ time.Sleep(2 * polled.Poll) // let any scan already under way finish
+ n = count()
+ time.Sleep(8 * polled.Poll)
+ if got := count(); got != n {
+ t.Fatalf("%d more scans in eight polling periods after the repair; polling should have stopped", got-n)
+ }
+}
+
+// A watcher whose reader dies closes its channel. The loop must fall back
+// to polling, not spin and not stop publishing.
+func TestRunPollsAfterTheWatcherStops(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, polled)
+ defer stop()
+ close(w.ch)
+ write(t, dir, "after.txt", "a")
+ eventually(t, "after.txt found by polling", func() bool { return manifestHas(s, "after.txt", sum([]byte("a"))) })
+}
+
+func TestRunWithNoWatcherPolls(t *testing.T) {
+ s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
+ p := open(t, s)
+ stop := startRun(t, p, dir, nil, polled)
+ defer stop()
+ write(t, dir, "x", "x")
+ eventually(t, "x found by polling", func() bool { return manifestHas(s, "x", sum([]byte("x"))) })
+}
+
+func TestRunStopsWhenSuperseded(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p, other := open(t, s), open(t, s)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- run(ctx, p, dir, eventsOnly, w) }()
+ stage(t, other, "theirs", "t")
+ commit(t, other)
+ write(t, dir, "mine", "m")
+ w.ch <- event{rel: "mine"}
+ select {
+ case err := <-done:
+ if !errors.Is(err, ErrSuperseded) {
+ t.Fatalf("err = %v, want ErrSuperseded", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("run did not stop after another writer published")
+ }
+}
+
+func TestRunIgnoresAnEventForAPathThatEscapes(t *testing.T) {
+ s, dir, w := release.NewDirStore(t.TempDir()), t.TempDir(), newFake()
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, eventsOnly)
+ w.ch <- event{rel: "../outside"}
+ write(t, dir, "ok", "ok")
+ w.ch <- event{rel: "ok"}
+ eventually(t, "ok published", func() bool { return manifestHas(s, "ok", sum([]byte("ok"))) })
+ if err := stop(); !errors.Is(err, context.Canceled) {
+ t.Fatalf("run ended with %v; a bad path must be skipped, not fatal", err)
+ }
+}
diff --git a/internal/hostfiles/watch_linux.go b/internal/hostfiles/watch_linux.go
new file mode 100644
index 000000000..2d950f8b4
--- /dev/null
+++ b/internal/hostfiles/watch_linux.go
@@ -0,0 +1,218 @@
+//go:build linux
+
+package hostfiles
+
+import (
+ "errors"
+ "fmt"
+ "io/fs"
+ "os"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "unsafe"
+
+ "golang.org/x/sys/unix"
+)
+
+const watchMask = unix.IN_MODIFY | unix.IN_CLOSE_WRITE | unix.IN_MOVED_FROM | unix.IN_MOVED_TO |
+ unix.IN_CREATE | unix.IN_DELETE | unix.IN_ATTRIB
+
+type inotifyWatcher struct {
+ root string
+ skip func(string, bool) bool
+ f *os.File // the inotify descriptor, non-blocking, so Close unblocks Read
+ ch chan event
+
+ mu sync.Mutex
+ wds map[int32]string // watch descriptor -> directory, relative to root ("" is root)
+
+ // Coverage is tracked by counting failures, not with a flag. failures
+ // goes up whenever a directory may have been left unwatched; repaired
+ // is the count Repair had seen when it started the walk that
+ // succeeded. A failure that arrives while Repair is walking is
+ // therefore never erased by that Repair: the two differ again.
+ failures atomic.Int64
+ repaired atomic.Int64
+ dead atomic.Bool // the reader has stopped; no event will ever arrive
+}
+
+// afterRepairWalk runs between Repair's walk and its bookkeeping. A test seam.
+var afterRepairWalk = func() {}
+
+func newWatcher(root string, skip func(string, bool) bool) (watcher, error) {
+ fd, err := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK)
+ if err != nil {
+ return nil, fmt.Errorf("inotify init: %w", err)
+ }
+ w := &inotifyWatcher{root: root, skip: skip, f: os.NewFile(uintptr(fd), "inotify"),
+ ch: make(chan event, 4096), wds: map[int32]string{}}
+ if err := w.addTree(""); err != nil {
+ w.failures.Add(1) // partial coverage: Run polls and calls Repair
+ }
+ go w.loop()
+ return w, nil
+}
+
+func (w *inotifyWatcher) Events() <-chan event { return w.ch }
+func (w *inotifyWatcher) Degraded() bool {
+ return w.dead.Load() || w.failures.Load() != w.repaired.Load()
+}
+func (w *inotifyWatcher) Close() error { return w.f.Close() }
+
+// Repair re-adds a watch on every directory. InotifyAddWatch on a path
+// already watched returns the same descriptor, so this is safe to repeat.
+// A watcher whose reader has stopped cannot be repaired.
+func (w *inotifyWatcher) Repair() error {
+ if w.dead.Load() {
+ return errors.New("the inotify reader has stopped")
+ }
+ seen := w.failures.Load()
+ if err := w.addTree(""); err != nil {
+ return err
+ }
+ afterRepairWalk()
+ // Only the failures this walk could have covered are cleared. One that
+ // arrived during the walk may concern a directory the walk had already
+ // passed, so it leaves the watcher degraded and Repair runs again.
+ w.repaired.Store(seen)
+ return nil
+}
+
+// addTree watches rel and every directory under it. It keeps going after a
+// failure, so one unreadable directory does not cost the rest their
+// watches, and returns the first error.
+func (w *inotifyWatcher) addTree(rel string) error {
+ var first error
+ start := filepath.Join(w.root, filepath.FromSlash(rel))
+ err := filepath.WalkDir(start, func(full string, d fs.DirEntry, err error) error {
+ if err != nil {
+ if first == nil {
+ first = err
+ }
+ return nil
+ }
+ if !d.IsDir() {
+ return nil // WalkDir does not follow symlinks, so a linked directory is never watched
+ }
+ r, rerr := filepath.Rel(w.root, full)
+ if rerr != nil {
+ return nil
+ }
+ r = filepath.ToSlash(r)
+ if r == "." {
+ r = ""
+ }
+ if r != "" && w.skip != nil && w.skip(r, true) {
+ return filepath.SkipDir
+ }
+ wd, aerr := unix.InotifyAddWatch(int(w.f.Fd()), full, watchMask|unix.IN_DONT_FOLLOW|unix.IN_ONLYDIR)
+ if aerr != nil {
+ if first == nil {
+ first = fmt.Errorf("watch %s: %w", full, aerr)
+ }
+ return nil
+ }
+ w.mu.Lock()
+ w.wds[int32(wd)] = r
+ w.mu.Unlock()
+ return nil
+ })
+ if err != nil && first == nil {
+ first = err
+ }
+ return first
+}
+
+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)
+ select {
+ case <-w.ch:
+ default:
+ }
+ select {
+ case w.ch <- event{rescan: true}:
+ default:
+ }
+ }
+}
+
+func (w *inotifyWatcher) loop() {
+ defer close(w.ch)
+ buf := make([]byte, 64<<10)
+ for {
+ n, err := w.f.Read(buf)
+ if err != nil {
+ // Closed by us, or broken. Either way nothing more will be
+ // read: the closed channel tells Run to poll from here on.
+ w.dead.Store(true)
+ return
+ }
+ for off := 0; off+unix.SizeofInotifyEvent <= n; {
+ raw := (*unix.InotifyEvent)(unsafe.Pointer(&buf[off]))
+ nameLen := int(raw.Len)
+ name := ""
+ if nameLen > 0 {
+ b := buf[off+unix.SizeofInotifyEvent : off+unix.SizeofInotifyEvent+nameLen]
+ for i, c := range b {
+ if c == 0 {
+ b = b[:i]
+ break
+ }
+ }
+ name = string(b)
+ }
+ off += unix.SizeofInotifyEvent + nameLen
+ w.handle(raw.Wd, raw.Mask, name)
+ }
+ }
+}
+
+func (w *inotifyWatcher) handle(wd int32, mask uint32, name string) {
+ if mask&unix.IN_Q_OVERFLOW != 0 {
+ // Events were dropped, and one of them may have been a new
+ // directory, which is then unwatched. Coverage is uncertain until
+ // Repair has walked the tree again.
+ w.failures.Add(1)
+ w.send(event{rescan: true})
+ return
+ }
+ w.mu.Lock()
+ dir, known := w.wds[wd]
+ if mask&unix.IN_IGNORED != 0 {
+ delete(w.wds, wd)
+ }
+ w.mu.Unlock()
+ if !known || name == "" {
+ return
+ }
+ rel := name
+ if dir != "" {
+ rel = dir + "/" + name
+ }
+ if mask&unix.IN_ISDIR != 0 {
+ if mask&(unix.IN_CREATE|unix.IN_MOVED_TO) != 0 {
+ if w.skip != nil && w.skip(rel, true) {
+ return
+ }
+ if err := w.addTree(rel); err != nil {
+ w.failures.Add(1)
+ }
+ }
+ // Files may have appeared in the new directory before its watch
+ // did, and a directory moved away takes its files with it: only a
+ // scan finds either.
+ w.send(event{rescan: true})
+ return
+ }
+ if w.skip != nil && w.skip(rel, false) {
+ return
+ }
+ w.send(event{rel: rel})
+}
diff --git a/internal/hostfiles/watch_linux_test.go b/internal/hostfiles/watch_linux_test.go
new file mode 100644
index 000000000..7ef7ec4a8
--- /dev/null
+++ b/internal/hostfiles/watch_linux_test.go
@@ -0,0 +1,232 @@
+//go:build linux
+
+package hostfiles
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "golang.org/x/sys/unix"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// realWatcher returns a working inotify watcher, or skips the test where
+// the kernel will not give one (a shared machine at its per-user limit).
+// CI must not skip: see the plan's Task 7, step 10.
+func realWatcher(t *testing.T, root string, skip func(string, bool) bool) watcher {
+ t.Helper()
+ w, err := newWatcher(root, skip)
+ if err != nil {
+ if os.Getenv("CARLOS_REQUIRE_INOTIFY") != "" {
+ t.Fatalf("inotify is required here: %v", err)
+ }
+ t.Skipf("no inotify instance available: %v", err)
+ }
+ t.Cleanup(func() { w.Close() })
+ return w
+}
+
+// startReal runs the loop on a real watcher with scans and polling off,
+// and returns once the first full scan is done: everything after that is
+// found by an inotify event or not at all.
+func startReal(t *testing.T, s release.Store, dir string) {
+ t.Helper()
+ w := realWatcher(t, dir, nil)
+ p := open(t, s)
+ stop := startRun(t, p, dir, w, Options{Deadline: 50 * time.Millisecond, Scan: time.Hour, Poll: time.Hour})
+ t.Cleanup(func() { stop() })
+}
+
+func TestWatcherPublishesAFileStillHeldOpen(t *testing.T) {
+ s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
+ startReal(t, s, dir)
+ f, err := os.Create(filepath.Join(dir, "open.log"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer f.Close() // deliberately not closed before the assertion
+ if _, err := f.WriteString("line one\n"); err != nil {
+ t.Fatal(err)
+ }
+ eventually(t, "open.log published while still open", func() bool {
+ return manifestHas(s, "open.log", sum([]byte("line one\n")))
+ })
+ // The first publication could have come from the file's creation. A
+ // second write through the same descriptor produces no create, no
+ // close and no rename: only IN_MODIFY can announce it.
+ if _, err := f.WriteString("line two\n"); err != nil {
+ t.Fatal(err)
+ }
+ eventually(t, "the second write published, still open", func() bool {
+ return manifestHas(s, "open.log", sum([]byte("line one\nline two\n")))
+ })
+}
+
+func TestWatcherSeesAFileInADirectoryCreatedAfterItStarted(t *testing.T) {
+ s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
+ startReal(t, s, dir)
+ write(t, dir, "new/deep/a.txt", "a")
+ eventually(t, "new/deep/a.txt published", func() bool {
+ return manifestHas(s, "new/deep/a.txt", sum([]byte("a")))
+ })
+ // And a later write inside the new directory is an event, not a scan.
+ time.Sleep(150 * time.Millisecond)
+ write(t, dir, "new/deep/b.txt", "b")
+ eventually(t, "new/deep/b.txt published", func() bool {
+ return manifestHas(s, "new/deep/b.txt", sum([]byte("b")))
+ })
+}
+
+func TestWatcherPublishesARenameAsDeleteAndCreate(t *testing.T) {
+ s, dir := release.NewDirStore(t.TempDir()), t.TempDir()
+ write(t, dir, "tmp", "v")
+ startReal(t, s, dir)
+ if err := os.Rename(filepath.Join(dir, "tmp"), filepath.Join(dir, "final")); err != nil {
+ t.Fatal(err)
+ }
+ eventually(t, "final present and tmp tombstoned", func() bool {
+ m, _, _ := Load(context.Background(), s, pfx)
+ return m.Entries["tmp"].Deleted && m.Entries["final"].SHA256 == sum([]byte("v"))
+ })
+}
+
+func TestWatcherDoesNotWatchASkippedDirectory(t *testing.T) {
+ root := t.TempDir()
+ if err := os.MkdirAll(filepath.Join(root, "cache"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ w := realWatcher(t, root, func(rel string, isDir bool) bool { return isDir && rel == "cache" })
+ write(t, root, "cache/x", "x")
+ select {
+ case ev := <-w.Events():
+ t.Fatalf("event from a skipped directory: %+v", ev)
+ case <-time.After(150 * time.Millisecond):
+ }
+}
+
+// unwatched builds a watcher with one directory that has no watch and no
+// event on its way: the state an overflow leaves when it swallows that
+// directory's creation. The directory exists before the watcher starts
+// and is excluded while the watcher is built, so the kernel never queues
+// an event that could install the watch later behind the test's back.
+func unwatched(t *testing.T) (w *inotifyWatcher, root string) {
+ t.Helper()
+ root = t.TempDir()
+ if err := os.MkdirAll(filepath.Join(root, "lost"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ var building atomic.Bool
+ building.Store(true)
+ w = realWatcher(t, root, func(rel string, isDir bool) bool {
+ return building.Load() && isDir && rel == "lost"
+ }).(*inotifyWatcher)
+ building.Store(false)
+ return w, root
+}
+
+// noEventFor fails if an event for rel arrives within 200ms.
+func noEventFor(t *testing.T, w *inotifyWatcher, rel string) {
+ t.Helper()
+ deadline := time.After(200 * time.Millisecond)
+ for {
+ select {
+ case ev := <-w.Events():
+ if ev.rel == rel {
+ t.Fatalf("an event for %s, which has no watch", rel)
+ }
+ case <-deadline:
+ return
+ }
+ }
+}
+
+func eventFor(t *testing.T, w *inotifyWatcher, rel string) {
+ t.Helper()
+ deadline := time.After(2 * time.Second)
+ for {
+ select {
+ case ev := <-w.Events():
+ if ev.rel == rel {
+ return
+ }
+ case <-deadline:
+ t.Fatalf("no event for %s", rel)
+ }
+ }
+}
+
+// Plan round 1 finding 8: a kernel overflow may have swallowed a
+// directory's creation, so coverage is uncertain until Repair has walked
+// the tree again. Only Repair can restore it here: remove the walk from
+// Repair and the last assertion fails.
+func TestWatcherOverflowMarksCoverageUncertainUntilRepaired(t *testing.T) {
+ w, root := unwatched(t)
+ if w.Degraded() {
+ t.Fatal("degraded before anything went wrong")
+ }
+ write(t, root, "lost/before.txt", "b")
+ noEventFor(t, w, "lost/before.txt") // proves the directory really is unwatched
+
+ w.handle(0, unix.IN_Q_OVERFLOW, "")
+ if !w.Degraded() {
+ t.Fatal("an overflow left coverage marked complete")
+ }
+ if err := w.Repair(); err != nil {
+ t.Fatal(err)
+ }
+ if w.Degraded() {
+ t.Fatal("still degraded after a successful repair")
+ }
+ write(t, root, "lost/after.txt", "a")
+ eventFor(t, w, "lost/after.txt")
+}
+
+// Plan round 2 finding 5: a failure that arrives while Repair is walking
+// may concern a directory the walk has already passed. That Repair must
+// not clear it.
+func TestWatcherRepairDoesNotEraseAFailureThatArrivedDuringItsWalk(t *testing.T) {
+ w, _ := unwatched(t)
+ w.handle(0, unix.IN_Q_OVERFLOW, "")
+ orig := afterRepairWalk
+ t.Cleanup(func() { afterRepairWalk = orig })
+ afterRepairWalk = func() {
+ afterRepairWalk = func() {}
+ w.handle(0, unix.IN_Q_OVERFLOW, "") // a second overflow, after the walk
+ }
+ if err := w.Repair(); err != nil {
+ t.Fatal(err)
+ }
+ if !w.Degraded() {
+ t.Fatal("the repair cleared a failure newer than its walk")
+ }
+ if err := w.Repair(); err != nil {
+ t.Fatal(err)
+ }
+ if w.Degraded() {
+ t.Fatal("a repair with nothing new during its walk left the watcher degraded")
+ }
+}
+
+func TestAStoppedWatcherCannotBeRepairedAndClosesItsChannel(t *testing.T) {
+ w := realWatcher(t, t.TempDir(), nil).(*inotifyWatcher)
+ w.Close()
+ select {
+ case _, open := <-w.Events():
+ if open {
+ t.Fatal("an event after Close")
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("the channel did not close")
+ }
+ if err := w.Repair(); err == nil {
+ t.Fatal("a stopped watcher reported a successful repair")
+ }
+ if !w.Degraded() {
+ t.Fatal("a stopped watcher is not degraded")
+ }
+}
diff --git a/internal/hostfiles/watch_other.go b/internal/hostfiles/watch_other.go
new file mode 100644
index 000000000..40d97caf6
--- /dev/null
+++ b/internal/hostfiles/watch_other.go
@@ -0,0 +1,10 @@
+//go:build !linux
+
+package hostfiles
+
+import "errors"
+
+// newWatcher has no implementation off Linux; Run polls instead.
+func newWatcher(root string, skip func(string, bool) bool) (watcher, error) {
+ return nil, errors.New("no file watcher on this OS")
+}