From 332ee0cd21995f23f4c6e8a0d70b67ae3ded8ee7 Mon Sep 17 00:00:00 2001 From: aniket866 Date: Fri, 16 Oct 2026 00:37:09 +0530 Subject: [PATCH 1/2] fix(watcher): debounce and serialize artifact imports Signed-off-by: aniket866 --- pkg/watcher/watchManager.go | 215 ++++++++++++++++++++++++------- pkg/watcher/watchManager_test.go | 167 ++++++++++++++++++++++++ 2 files changed, 338 insertions(+), 44 deletions(-) create mode 100644 pkg/watcher/watchManager_test.go diff --git a/pkg/watcher/watchManager.go b/pkg/watcher/watchManager.go index fa205f7..ccc6015 100644 --- a/pkg/watcher/watchManager.go +++ b/pkg/watcher/watchManager.go @@ -20,38 +20,56 @@ import ( "context" "fmt" "log" + "path/filepath" "sync" + "time" "github.com/fsnotify/fsnotify" "github.com/microcks/microcks-cli/pkg/config" ) +const importDebounce = 300 * time.Millisecond + +type pendingImport struct { + entry config.WatchEntry + deadline time.Time +} + type WatchManager struct { - fileWatcher *fsnotify.Watcher - configPath string - watchEntries map[string]config.WatchEntry - lock sync.Mutex + fileWatcher *fsnotify.Watcher + configPath string + watchEntries map[string]config.WatchEntry + watchDirs map[string]struct{} + triggerImport func(context.Context, config.WatchEntry) + lock sync.Mutex } func NewWatchManger(configPath string) (*WatchManager, error) { - fw, err := fsnotify.NewWatcher() + absConfigPath, err := filepath.Abs(configPath) if err != nil { return nil, err } - err = fw.Add(configPath) + fw, err := fsnotify.NewWatcher() if err != nil { return nil, err } + configDir := filepath.Dir(absConfigPath) + if err := fw.Add(configDir); err != nil { + fw.Close() + return nil, err + } + wm := &WatchManager{ fileWatcher: fw, - configPath: configPath, + configPath: absConfigPath, watchEntries: make(map[string]config.WatchEntry), + watchDirs: map[string]struct{}{configDir: {}}, } - err = wm.Reload() - if err != nil { + if err := wm.Reload(); err != nil { + fw.Close() return nil, err } @@ -65,33 +83,48 @@ func (wm *WatchManager) Reload() error { } newFiles := map[string]config.WatchEntry{} + newDirs := map[string]struct{}{filepath.Dir(wm.configPath): {}} for _, entry := range cfg.Entries { - newFiles[entry.FilePath] = entry + absPath, err := filepath.Abs(entry.FilePath) + if err != nil { + log.Printf("[WARN] Cannot resolve watched file %s: %v", entry.FilePath, err) + continue + } + newFiles[filepath.Clean(absPath)] = entry + newDirs[filepath.Dir(filepath.Clean(absPath))] = struct{}{} } - // Remove stale watchers - for file := range wm.watchEntries { + // Watch containing directories so replacing a file atomically doesn't + // detach the watcher from its inode. + for dir := range newDirs { + if _, exists := wm.watchDirs[dir]; exists { + continue + } + if err := wm.fileWatcher.Add(dir); err != nil { + log.Printf("[WARN] Cannot watch directory %s: %v", dir, err) + delete(newDirs, dir) + continue + } + log.Printf("[INFO] Watcher added on directory %s", dir) + } - if _, exists := newFiles[file]; !exists { - if err := wm.fileWatcher.Remove(file); err != nil { - log.Printf("[WARN] Cannot remove watcher for file %s: %v", file, err) - } + for dir := range wm.watchDirs { + if _, exists := newDirs[dir]; exists { + continue + } + if err := wm.fileWatcher.Remove(dir); err != nil { + log.Printf("[WARN] Cannot remove watcher for directory %s: %v", dir, err) } } - // Add new watchers for file := range newFiles { if _, exists := wm.watchEntries[file]; !exists { - err := wm.fileWatcher.Add(file) - if err != nil { - log.Printf("[WARN] Cannot watch file %s: %v", file, err) - continue - } - log.Printf("[INFO] Watcher added on %s", file) + log.Printf("[INFO] Watching file %s", file) } } wm.watchEntries = newFiles + wm.watchDirs = newDirs return nil } @@ -100,33 +133,127 @@ func (wm *WatchManager) Run() { } func (wm *WatchManager) RunContext(ctx context.Context) { + runCtx, cancel := context.WithCancel(ctx) + defer cancel() defer wm.fileWatcher.Close() + + imports := make(chan config.WatchEntry) + go func() { + for { + select { + case <-runCtx.Done(): + return + case entry := <-imports: + if runCtx.Err() != nil { + return + } + if wm.triggerImport != nil { + wm.triggerImport(runCtx, entry) + } else { + TriggerImportWithContext(runCtx, entry) + } + } + } + }() + + pending := make(map[string]pendingImport) + queued := make(map[string]config.WatchEntry) + ready := make([]string, 0) + debounceTimer := time.NewTimer(time.Hour) + if !debounceTimer.Stop() { + <-debounceTimer.C + } + defer debounceTimer.Stop() + + eventErrors := wm.fileWatcher.Errors for { + var debounceC <-chan time.Time + if len(pending) > 0 { + var deadline time.Time + for _, item := range pending { + if deadline.IsZero() || item.deadline.Before(deadline) { + deadline = item.deadline + } + } + if !debounceTimer.Stop() { + select { + case <-debounceTimer.C: + default: + } + } + delay := time.Until(deadline) + if delay < 0 { + delay = 0 + } + debounceTimer.Reset(delay) + debounceC = debounceTimer.C + } + + var importC chan config.WatchEntry + var nextImport config.WatchEntry + if len(ready) > 0 { + importC = imports + nextImport = queued[ready[0]] + } + select { - case <-ctx.Done(): + case <-runCtx.Done(): return - case event := <-wm.fileWatcher.Events: - if event.Op&fsnotify.Write == fsnotify.Write { - if event.Name == wm.configPath { - fmt.Println("[INFO] Reloading config...") - wm.lock.Lock() - err := wm.Reload() - wm.lock.Unlock() - if err != nil { - // A bad config edit shouldn't kill the watcher; log and - // keep the previous config until the next valid save. - log.Printf("[ERROR] Config reload failed, keeping previous config: %v", err) - } - } else { - wm.lock.Lock() - entry, exists := wm.watchEntries[event.Name] - wm.lock.Unlock() - if exists { - go TriggerImportWithContext(ctx, entry) - } + case event, ok := <-wm.fileWatcher.Events: + if !ok { + return + } + if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename) == 0 { + continue + } + + eventPath, err := filepath.Abs(event.Name) + if err != nil { + continue + } + eventPath = filepath.Clean(eventPath) + if eventPath == wm.configPath { + fmt.Println("[INFO] Reloading config...") + wm.lock.Lock() + err := wm.Reload() + wm.lock.Unlock() + if err != nil { + log.Printf("[ERROR] Config reload failed, keeping previous config: %v", err) } + continue + } + + wm.lock.Lock() + entry, exists := wm.watchEntries[eventPath] + wm.lock.Unlock() + if exists { + pending[eventPath] = pendingImport{ + entry: entry, + deadline: time.Now().Add(importDebounce), + } + } + case <-debounceC: + now := time.Now() + for file, item := range pending { + if item.deadline.After(now) { + continue + } + delete(pending, file) + if _, exists := queued[file]; exists { + queued[file] = item.entry + continue + } + queued[file] = item.entry + ready = append(ready, file) + } + case importC <- nextImport: + delete(queued, ready[0]) + ready = ready[1:] + case err, ok := <-eventErrors: + if !ok { + eventErrors = nil + continue } - case err := <-wm.fileWatcher.Errors: log.Printf("[ERROR] Watcher error: %v", err) } } diff --git a/pkg/watcher/watchManager_test.go b/pkg/watcher/watchManager_test.go new file mode 100644 index 0000000..f4ab1b0 --- /dev/null +++ b/pkg/watcher/watchManager_test.go @@ -0,0 +1,167 @@ +/* + * Copyright The Microcks Authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package watcher + +import ( + "context" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/microcks/microcks-cli/pkg/config" +) + +func startTestWatcher(t *testing.T, filenames []string, trigger func(context.Context, config.WatchEntry)) ([]string, context.CancelFunc) { + t.Helper() + + dir := t.TempDir() + files := make([]string, 0, len(filenames)) + entries := make([]config.WatchEntry, 0, len(filenames)) + for _, filename := range filenames { + filePath := filepath.Join(dir, filename) + if err := os.WriteFile(filePath, []byte("initial"), 0600); err != nil { + t.Fatal(err) + } + files = append(files, filePath) + entries = append(entries, config.WatchEntry{ + FilePath: filePath, + Context: []string{"test"}, + }) + } + + configPath := filepath.Join(dir, "watch.yaml") + if err := config.WriteLocalWatchConfig(config.WatchConfig{Entries: entries}, configPath); err != nil { + t.Fatal(err) + } + + manager, err := NewWatchManger(configPath) + if err != nil { + t.Fatal(err) + } + manager.triggerImport = trigger + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + manager.RunContext(ctx) + close(done) + }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Error("watch manager did not stop after cancellation") + } + }) + + return files, cancel +} + +func TestRunContextDebouncesWriteBurst(t *testing.T) { + imports := make(chan string, 10) + files, _ := startTestWatcher(t, []string{"api.yaml"}, func(_ context.Context, entry config.WatchEntry) { + imports <- entry.FilePath + }) + + for i := 0; i < 20; i++ { + if err := os.WriteFile(files[0], []byte("updated"), 0600); err != nil { + t.Fatal(err) + } + time.Sleep(10 * time.Millisecond) + } + + select { + case got := <-imports: + if got != files[0] { + t.Fatalf("imported %q, want %q", got, files[0]) + } + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for debounced import") + } + + select { + case got := <-imports: + t.Fatalf("write burst triggered a duplicate import for %q", got) + case <-time.After(450 * time.Millisecond): + } +} + +func TestRunContextSerializesImportsAndWatchesAtomicReplacement(t *testing.T) { + started := make(chan string, 10) + release := make(chan struct{}, 10) + var active atomic.Int32 + var maxActive atomic.Int32 + files, _ := startTestWatcher(t, []string{"api.yaml"}, func(ctx context.Context, entry config.WatchEntry) { + current := active.Add(1) + for observed := maxActive.Load(); current > observed && !maxActive.CompareAndSwap(observed, current); observed = maxActive.Load() { + } + started <- entry.FilePath + select { + case <-release: + case <-ctx.Done(): + } + active.Add(-1) + }) + + target := files[0] + temp := target + ".tmp" + if err := os.WriteFile(temp, []byte("replacement"), 0600); err != nil { + t.Fatal(err) + } + if err := os.Remove(target); err != nil { + t.Fatal(err) + } + if err := os.Rename(temp, target); err != nil { + t.Fatal(err) + } + + select { + case got := <-started: + if got != target { + t.Fatalf("imported %q, want %q", got, target) + } + case <-time.After(3 * time.Second): + t.Fatal("atomic replacement did not trigger an import") + } + + if err := os.WriteFile(target, []byte("changed during import"), 0600); err != nil { + t.Fatal(err) + } + time.Sleep(importDebounce + 100*time.Millisecond) + select { + case got := <-started: + t.Fatalf("started concurrent import for %q", got) + default: + } + + release <- struct{}{} + select { + case got := <-started: + if got != target { + t.Fatalf("imported %q, want %q", got, target) + } + case <-time.After(3 * time.Second): + t.Fatal("change during import was not imported") + } + if got := maxActive.Load(); got != 1 { + t.Fatalf("maximum concurrent imports = %d, want 1", got) + } + release <- struct{}{} +} From 0bc309202a374df1d31e8111ed8ed7408b3c55ae Mon Sep 17 00:00:00 2001 From: aniket866 Date: Fri, 16 Oct 2026 00:55:03 +0530 Subject: [PATCH 2/2] fix(watcher): use test watcher cancellation handles Signed-off-by: aniket866 --- pkg/watcher/watchManager_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pkg/watcher/watchManager_test.go b/pkg/watcher/watchManager_test.go index f4ab1b0..9086886 100644 --- a/pkg/watcher/watchManager_test.go +++ b/pkg/watcher/watchManager_test.go @@ -76,9 +76,10 @@ func startTestWatcher(t *testing.T, filenames []string, trigger func(context.Con func TestRunContextDebouncesWriteBurst(t *testing.T) { imports := make(chan string, 10) - files, _ := startTestWatcher(t, []string{"api.yaml"}, func(_ context.Context, entry config.WatchEntry) { + files, cancel := startTestWatcher(t, []string{"api.yaml"}, func(_ context.Context, entry config.WatchEntry) { imports <- entry.FilePath }) + defer cancel() for i := 0; i < 20; i++ { if err := os.WriteFile(files[0], []byte("updated"), 0600); err != nil { @@ -108,7 +109,7 @@ func TestRunContextSerializesImportsAndWatchesAtomicReplacement(t *testing.T) { release := make(chan struct{}, 10) var active atomic.Int32 var maxActive atomic.Int32 - files, _ := startTestWatcher(t, []string{"api.yaml"}, func(ctx context.Context, entry config.WatchEntry) { + files, cancel := startTestWatcher(t, []string{"api.yaml"}, func(ctx context.Context, entry config.WatchEntry) { current := active.Add(1) for observed := maxActive.Load(); current > observed && !maxActive.CompareAndSwap(observed, current); observed = maxActive.Load() { } @@ -119,6 +120,7 @@ func TestRunContextSerializesImportsAndWatchesAtomicReplacement(t *testing.T) { } active.Add(-1) }) + defer cancel() target := files[0] temp := target + ".tmp"