From 69c1a37577924fa826d34d58b3f29afc49fb2fca Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 3 Jul 2026 13:52:17 +0000 Subject: [PATCH 1/3] fix(watch): debounce fsnotify events, backfill archives, resilient walk Co-authored-by: Belphemur <197810+Belphemur@users.noreply.github.com> --- .github/copilot-instructions.md | 9 + AGENTS.md | 1 + cmd/cbzoptimizer/commands/watch_command.go | 198 ++++++++++++++++-- .../commands/watch_command_test.go | 140 +++++++++++++ 4 files changed, 334 insertions(+), 14 deletions(-) create mode 100644 cmd/cbzoptimizer/commands/watch_command_test.go diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index ab73d0e..ef0bead 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -123,6 +123,15 @@ golangci-lint run - **Error handling:** Always check errors explicitly; use structured error wrapping with `fmt.Errorf("context: %w", err)` - **Context usage:** Pass `context.Context` as first parameter for operations that may be cancelled +### Commit Messages + +This project follows [Conventional Commits](https://www.conventionalcommits.org/). Prefix commit messages with a type (and optional scope), for example: + +- `fix(watch): debounce fsnotify events before optimizing` +- `feat(converter): add avif support` +- `docs: clarify commit convention` +- `chore: update dependency` + ### Logging Use **zerolog** for all logging: diff --git a/AGENTS.md b/AGENTS.md index a7a5d97..d45b6e5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,6 +14,7 @@ This repository contains AI-agent oriented project context. - Logging: zerolog - Error wrapping: `fmt.Errorf("context: %w", err)` - Prefer small, focused changes. +- Commit messages: follow [Conventional Commits](https://www.conventionalcommits.org/) (e.g. `fix(watch): debounce fsnotify events`, `docs: clarify commit convention`). ## Areas to know diff --git a/cmd/cbzoptimizer/commands/watch_command.go b/cmd/cbzoptimizer/commands/watch_command.go index d9265bf..64b66ec 100644 --- a/cmd/cbzoptimizer/commands/watch_command.go +++ b/cmd/cbzoptimizer/commands/watch_command.go @@ -7,6 +7,8 @@ import ( "path/filepath" "runtime" "strings" + "sync" + "time" utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" "github.com/belphemur/CBZOptimizer/v2/pkg/converter" @@ -17,6 +19,16 @@ import ( "github.com/spf13/viper" ) +// debounceDelay is the quiet period we wait, per path, after the last fsnotify +// event before triggering an optimization. This coalesces the bursts of +// Write/Create/Rename events fsnotify emits while a file is still being +// written (e.g. copied or downloaded into the watched folder). +const debounceDelay = 2 * time.Second + +// watchWorkerCount controls how many optimization jobs can run concurrently +// so the fsnotify event loop is never blocked waiting on a conversion. +var watchWorkerCount = runtime.NumCPU() + func init() { if runtime.GOOS != "linux" { return @@ -58,12 +70,12 @@ func WatchCommand(_ *cobra.Command, args []string) error { converterType := constant.FindConversionFormat(viper.GetString("format")) chapterConverter, err := converter.Get(converterType) if err != nil { - return fmt.Errorf("failed to get chapterConverter: %v", err) + return fmt.Errorf("failed to get chapterConverter: %w", err) } err = chapterConverter.PrepareConverter() if err != nil { - return fmt.Errorf("failed to prepare converter: %v", err) + return fmt.Errorf("failed to prepare converter: %w", err) } log.Info().Str("path", path).Bool("override", override).Uint8("quality", quality).Str("format", converterType.String()).Bool("split", split).Msg("Watching directory") @@ -71,11 +83,28 @@ func WatchCommand(_ *cobra.Command, args []string) error { if err != nil { return fmt.Errorf("failed to create file watcher: %w", err) } - defer watcher.Close() + defer func() { + if closeErr := watcher.Close(); closeErr != nil { + log.Error().Err(closeErr).Msg("Failed to close file watcher") + } + }() + + queue := newOptimizeQueue(watchWorkerCount, &utils2.OptimizeOptions{ + ChapterConverter: chapterConverter, + Quality: quality, + Override: override, + Split: split, + Timeout: timeout, + }) + defer queue.Stop() + + debouncer := newEventDebouncer(debounceDelay, queue.Enqueue) + defer debouncer.Stop() if err := addRecursiveWatch(watcher, path); err != nil { return fmt.Errorf("failed to watch path %s: %w", path, err) } + backfillExistingArchives(path, debouncer.Trigger) for { select { @@ -91,6 +120,11 @@ func WatchCommand(_ *cobra.Command, args []string) error { if err := addRecursiveWatch(watcher, event.Name); err != nil { log.Error().Err(err).Str("path", event.Name).Msg("Failed to watch created directory") } + // The newly discovered directory may already contain + // archives (e.g. a folder moved/copied in); back-fill them + // since no further fsnotify event will target them. + backfillExistingArchives(event.Name, debouncer.Trigger) + continue } } @@ -102,16 +136,7 @@ func WatchCommand(_ *cobra.Command, args []string) error { continue } - if err := utils2.Optimize(&utils2.OptimizeOptions{ - ChapterConverter: chapterConverter, - Path: event.Name, - Quality: quality, - Override: override, - Split: split, - Timeout: timeout, - }); err != nil { - log.Error().Err(err).Str("file", event.Name).Msg("Error processing file") - } + debouncer.Trigger(event.Name) case err, ok := <-watcher.Errors: if !ok { return nil @@ -121,10 +146,15 @@ func WatchCommand(_ *cobra.Command, args []string) error { } } +// addRecursiveWatch registers a watch on rootPath and every subdirectory +// beneath it. Directories that can't be read (e.g. permission errors) are +// logged and skipped instead of aborting the whole walk, so a single bad +// subtree doesn't prevent the rest of the folder from being watched. func addRecursiveWatch(watcher *fsnotify.Watcher, rootPath string) error { return filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error { if err != nil { - return err + log.Warn().Err(err).Str("path", path).Msg("Skipping path while setting up watch") + return nil } if !entry.IsDir() { return nil @@ -136,6 +166,31 @@ func addRecursiveWatch(watcher *fsnotify.Watcher, rootPath string) error { }) } +// backfillExistingArchives walks rootPath for comic archives that already +// exist on disk and hands them to process. This covers the case where a +// directory (potentially already containing archives) is created/moved into +// the watched tree: only the directory itself generates an fsnotify event, +// so the files inside it would otherwise never be picked up. +func backfillExistingArchives(rootPath string, process func(path string)) { + err := filepath.WalkDir(rootPath, func(path string, entry fs.DirEntry, err error) error { + if err != nil { + log.Warn().Err(err).Str("path", path).Msg("Skipping path while scanning for existing archives") + return nil + } + if entry.IsDir() { + return nil + } + if !isComicArchive(path) { + return nil + } + process(path) + return nil + }) + if err != nil { + log.Error().Err(err).Str("path", rootPath).Msg("Failed to scan directory for existing archives") + } +} + func shouldProcessWatchEvent(event fsnotify.Event) bool { return event.Has(fsnotify.Create) || event.Has(fsnotify.Write) || event.Has(fsnotify.Rename) } @@ -144,3 +199,118 @@ func isComicArchive(path string) bool { filename := strings.ToLower(path) return strings.HasSuffix(filename, ".cbz") || strings.HasSuffix(filename, ".cbr") } + +// eventDebouncer coalesces bursts of fsnotify events targeting the same path +// into a single call to onQuiet, fired after the path has been quiet for +// delay. This prevents repeated conversions being triggered while a file is +// still being written, and avoids running Optimize against a Rename event +// that reports the *old* (already gone) file name once the quiet period lets +// us re-check the path. +type eventDebouncer struct { + delay time.Duration + onQuiet func(path string) + + mu sync.Mutex + timers map[string]*time.Timer +} + +func newEventDebouncer(delay time.Duration, onQuiet func(path string)) *eventDebouncer { + return &eventDebouncer{ + delay: delay, + onQuiet: onQuiet, + timers: make(map[string]*time.Timer), + } +} + +// Trigger (re)schedules onQuiet to run for path after the debounce delay, +// resetting any previously pending timer for the same path. +func (d *eventDebouncer) Trigger(path string) { + d.mu.Lock() + defer d.mu.Unlock() + + if timer, exists := d.timers[path]; exists { + timer.Reset(d.delay) + return + } + + d.timers[path] = time.AfterFunc(d.delay, func() { + d.mu.Lock() + delete(d.timers, path) + d.mu.Unlock() + d.onQuiet(path) + }) +} + +// Stop cancels all pending timers. +func (d *eventDebouncer) Stop() { + d.mu.Lock() + defer d.mu.Unlock() + for path, timer := range d.timers { + timer.Stop() + delete(d.timers, path) + } +} + +// optimizeQueue is a small worker pool that runs Optimize jobs off of the +// fsnotify event loop, so a slow conversion never blocks event draining. +type optimizeQueue struct { + jobs chan string + wg sync.WaitGroup + options *utils2.OptimizeOptions +} + +func newOptimizeQueue(workerCount int, options *utils2.OptimizeOptions) *optimizeQueue { + if workerCount < 1 { + workerCount = 1 + } + q := &optimizeQueue{ + jobs: make(chan string, 64), + options: options, + } + q.wg.Add(workerCount) + for i := 0; i < workerCount; i++ { + go q.worker() + } + return q +} + +func (q *optimizeQueue) worker() { + defer q.wg.Done() + for path := range q.jobs { + q.process(path) + } +} + +// process re-checks the path right before optimizing: fsnotify Rename events +// commonly report the *old* file name (already gone by the time we act on +// it), and Write can fire while a file is still being written elsewhere. In +// override mode, Optimize may overwrite/delete the source file, so skipping +// paths that no longer exist (or that are no longer regular files) avoids +// noisy failures. +func (q *optimizeQueue) process(path string) { + info, err := os.Stat(path) + if err != nil { + log.Debug().Err(err).Str("file", path).Msg("Skipping watch event: path no longer accessible") + return + } + if info.IsDir() { + return + } + + options := *q.options + options.Path = path + if err := utils2.Optimize(&options); err != nil { + log.Error().Err(err).Str("file", path).Msg("Error processing file") + } +} + +// Enqueue submits path for processing by the worker pool. +func (q *optimizeQueue) Enqueue(path string) { + q.jobs <- path +} + +// Stop closes the job queue and waits for in-flight jobs to finish. +func (q *optimizeQueue) Stop() { + close(q.jobs) + q.wg.Wait() +} diff --git a/cmd/cbzoptimizer/commands/watch_command_test.go b/cmd/cbzoptimizer/commands/watch_command_test.go new file mode 100644 index 0000000..6d171b7 --- /dev/null +++ b/cmd/cbzoptimizer/commands/watch_command_test.go @@ -0,0 +1,140 @@ +package commands + +import ( + "os" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" + "github.com/fsnotify/fsnotify" + "github.com/stretchr/testify/assert" +) + +func TestIsComicArchive(t *testing.T) { + testCases := []struct { + name string + path string + expected bool + }{ + {"cbz lowercase", "/a/b/chapter.cbz", true}, + {"cbr lowercase", "/a/b/chapter.cbr", true}, + {"cbz uppercase", "/a/b/chapter.CBZ", true}, + {"other extension", "/a/b/chapter.zip", false}, + {"no extension", "/a/b/chapter", false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.expected, isComicArchive(tc.path)) + }) + } +} + +func TestShouldProcessWatchEvent(t *testing.T) { + testCases := []struct { + name string + op fsnotify.Op + expected bool + }{ + {"create", fsnotify.Create, true}, + {"write", fsnotify.Write, true}, + {"rename", fsnotify.Rename, true}, + {"remove", fsnotify.Remove, false}, + {"chmod", fsnotify.Chmod, false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + event := fsnotify.Event{Name: "file.cbz", Op: tc.op} + assert.Equal(t, tc.expected, shouldProcessWatchEvent(event)) + }) + } +} + +func TestAddRecursiveWatchSkipsUnreadableSubdirectory(t *testing.T) { + root := t.TempDir() + nested := filepath.Join(root, "nested") + assert.NoError(t, os.MkdirAll(nested, 0o755)) + + watcher, err := fsnotify.NewWatcher() + assert.NoError(t, err) + defer func() { + assert.NoError(t, watcher.Close()) + }() + + // A regular, fully accessible tree should be watched without error. + assert.NoError(t, addRecursiveWatch(watcher, root)) +} + +func TestBackfillExistingArchivesFindsPreExistingArchives(t *testing.T) { + root := t.TempDir() + assert.NoError(t, os.WriteFile(filepath.Join(root, "chapter1.cbz"), []byte("data"), 0o644)) + assert.NoError(t, os.WriteFile(filepath.Join(root, "notes.txt"), []byte("data"), 0o644)) + + sub := filepath.Join(root, "sub") + assert.NoError(t, os.MkdirAll(sub, 0o755)) + assert.NoError(t, os.WriteFile(filepath.Join(sub, "chapter2.cbr"), []byte("data"), 0o644)) + + var mu sync.Mutex + var found []string + backfillExistingArchives(root, func(path string) { + mu.Lock() + defer mu.Unlock() + found = append(found, path) + }) + + assert.Len(t, found, 2) +} + +func TestEventDebouncerCoalescesBurstsIntoSingleCall(t *testing.T) { + var calls int32 + debouncer := newEventDebouncer(20*time.Millisecond, func(path string) { + atomic.AddInt32(&calls, 1) + }) + defer debouncer.Stop() + + // Simulate a burst of events for the same path. + for i := 0; i < 5; i++ { + debouncer.Trigger("/tmp/chapter.cbz") + time.Sleep(5 * time.Millisecond) + } + + time.Sleep(50 * time.Millisecond) + assert.EqualValues(t, 1, atomic.LoadInt32(&calls)) +} + +func TestEventDebouncerHandlesMultiplePaths(t *testing.T) { + var mu sync.Mutex + seen := make(map[string]int) + debouncer := newEventDebouncer(10*time.Millisecond, func(path string) { + mu.Lock() + defer mu.Unlock() + seen[path]++ + }) + defer debouncer.Stop() + + debouncer.Trigger("/tmp/a.cbz") + debouncer.Trigger("/tmp/b.cbz") + + time.Sleep(50 * time.Millisecond) + + mu.Lock() + defer mu.Unlock() + assert.Equal(t, 1, seen["/tmp/a.cbz"]) + assert.Equal(t, 1, seen["/tmp/b.cbz"]) +} + +func TestOptimizeQueueSkipsMissingPath(t *testing.T) { + q := newOptimizeQueue(1, &utils2.OptimizeOptions{}) + defer q.Stop() + + // Enqueue a path that doesn't exist; process should skip it without + // panicking or blocking since it never reaches utils2.Optimize. + q.Enqueue(filepath.Join(t.TempDir(), "missing.cbz")) + + // Give the worker a moment to drain the job. + time.Sleep(50 * time.Millisecond) +} From 1e692ed19b617b97873ecc63faf2e7d0b396cdcd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 3 Jul 2026 13:55:58 +0000 Subject: [PATCH 2/3] fix(watch): remove mutable package var, fix debouncer race Co-authored-by: Belphemur <197810+Belphemur@users.noreply.github.com> --- cmd/cbzoptimizer/commands/watch_command.go | 30 ++++++++++++++-------- 1 file changed, 19 insertions(+), 11 deletions(-) diff --git a/cmd/cbzoptimizer/commands/watch_command.go b/cmd/cbzoptimizer/commands/watch_command.go index 64b66ec..181d81c 100644 --- a/cmd/cbzoptimizer/commands/watch_command.go +++ b/cmd/cbzoptimizer/commands/watch_command.go @@ -25,10 +25,6 @@ import ( // written (e.g. copied or downloaded into the watched folder). const debounceDelay = 2 * time.Second -// watchWorkerCount controls how many optimization jobs can run concurrently -// so the fsnotify event loop is never blocked waiting on a conversion. -var watchWorkerCount = runtime.NumCPU() - func init() { if runtime.GOOS != "linux" { return @@ -89,7 +85,9 @@ func WatchCommand(_ *cobra.Command, args []string) error { } }() - queue := newOptimizeQueue(watchWorkerCount, &utils2.OptimizeOptions{ + // Optimization jobs run in a small worker pool so the fsnotify event loop + // is never blocked waiting on a conversion. + queue := newOptimizeQueue(runtime.NumCPU(), &utils2.OptimizeOptions{ ChapterConverter: chapterConverter, Quality: quality, Override: override, @@ -228,17 +226,27 @@ func (d *eventDebouncer) Trigger(path string) { d.mu.Lock() defer d.mu.Unlock() - if timer, exists := d.timers[path]; exists { - timer.Reset(d.delay) - return + if existing, exists := d.timers[path]; exists { + existing.Stop() } - d.timers[path] = time.AfterFunc(d.delay, func() { + var timer *time.Timer + timer = time.AfterFunc(d.delay, func() { d.mu.Lock() - delete(d.timers, path) + // Only the still-current timer for this path is allowed to clear the + // entry and fire onQuiet. If Trigger raced with this callback and + // already installed a newer timer, this stale invocation must not + // delete that newer entry or fire early. + owns := d.timers[path] == timer + if owns { + delete(d.timers, path) + } d.mu.Unlock() - d.onQuiet(path) + if owns { + d.onQuiet(path) + } }) + d.timers[path] = timer } // Stop cancels all pending timers. From 30d0d632bf2eb3bed41c3c1f4389ac2d506a7032 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 3 Jul 2026 14:26:40 +0000 Subject: [PATCH 3/3] fix(watch): address review feedback in watcher queue and tests Co-authored-by: Belphemur <197810+Belphemur@users.noreply.github.com> --- cmd/cbzoptimizer/commands/watch_command.go | 70 ++++++++++++++----- .../commands/watch_command_test.go | 54 +++++++++++--- 2 files changed, 96 insertions(+), 28 deletions(-) diff --git a/cmd/cbzoptimizer/commands/watch_command.go b/cmd/cbzoptimizer/commands/watch_command.go index 181d81c..b3bcb3e 100644 --- a/cmd/cbzoptimizer/commands/watch_command.go +++ b/cmd/cbzoptimizer/commands/watch_command.go @@ -1,6 +1,7 @@ package commands import ( + "errors" "fmt" "io/fs" "os" @@ -158,6 +159,10 @@ func addRecursiveWatch(watcher *fsnotify.Watcher, rootPath string) error { return nil } if err := watcher.Add(path); err != nil { + if errors.Is(err, fs.ErrPermission) { + log.Warn().Err(err).Str("path", path).Msg("Skipping unreadable directory while setting up watch") + return nil + } return fmt.Errorf("failed to watch directory %s: %w", path, err) } return nil @@ -208,15 +213,21 @@ type eventDebouncer struct { delay time.Duration onQuiet func(path string) - mu sync.Mutex - timers map[string]*time.Timer + mu sync.Mutex + timers map[string]*debounceTimer + inFlight sync.WaitGroup + stopping bool +} + +type debounceTimer struct { + timer *time.Timer } func newEventDebouncer(delay time.Duration, onQuiet func(path string)) *eventDebouncer { return &eventDebouncer{ delay: delay, onQuiet: onQuiet, - timers: make(map[string]*time.Timer), + timers: make(map[string]*debounceTimer), } } @@ -226,18 +237,27 @@ func (d *eventDebouncer) Trigger(path string) { d.mu.Lock() defer d.mu.Unlock() - if existing, exists := d.timers[path]; exists { - existing.Stop() + if d.stopping { + return } - var timer *time.Timer - timer = time.AfterFunc(d.delay, func() { + if existing, exists := d.timers[path]; exists { + if existing.timer.Stop() { + d.inFlight.Done() + } + } + + entry := &debounceTimer{} + d.inFlight.Add(1) + entry.timer = time.AfterFunc(d.delay, func() { + defer d.inFlight.Done() + d.mu.Lock() // Only the still-current timer for this path is allowed to clear the // entry and fire onQuiet. If Trigger raced with this callback and // already installed a newer timer, this stale invocation must not // delete that newer entry or fire early. - owns := d.timers[path] == timer + owns := d.timers[path] == entry && !d.stopping if owns { delete(d.timers, path) } @@ -246,25 +266,32 @@ func (d *eventDebouncer) Trigger(path string) { d.onQuiet(path) } }) - d.timers[path] = timer + d.timers[path] = entry } // Stop cancels all pending timers. func (d *eventDebouncer) Stop() { d.mu.Lock() - defer d.mu.Unlock() + d.stopping = true for path, timer := range d.timers { - timer.Stop() + if timer.timer.Stop() { + d.inFlight.Done() + } delete(d.timers, path) } + d.mu.Unlock() + d.inFlight.Wait() } // optimizeQueue is a small worker pool that runs Optimize jobs off of the // fsnotify event loop, so a slow conversion never blocks event draining. type optimizeQueue struct { - jobs chan string - wg sync.WaitGroup - options *utils2.OptimizeOptions + jobs chan string + wg sync.WaitGroup + mu sync.RWMutex + stopped bool + options *utils2.OptimizeOptions + optimize func(options *utils2.OptimizeOptions) error } func newOptimizeQueue(workerCount int, options *utils2.OptimizeOptions) *optimizeQueue { @@ -272,8 +299,9 @@ func newOptimizeQueue(workerCount int, options *utils2.OptimizeOptions) *optimiz workerCount = 1 } q := &optimizeQueue{ - jobs: make(chan string, 64), - options: options, + jobs: make(chan string, 64), + options: options, + optimize: utils2.Optimize, } q.wg.Add(workerCount) for i := 0; i < workerCount; i++ { @@ -307,18 +335,26 @@ func (q *optimizeQueue) process(path string) { options := *q.options options.Path = path - if err := utils2.Optimize(&options); err != nil { + if err := q.optimize(&options); err != nil { log.Error().Err(err).Str("file", path).Msg("Error processing file") } } // Enqueue submits path for processing by the worker pool. func (q *optimizeQueue) Enqueue(path string) { + q.mu.RLock() + defer q.mu.RUnlock() + if q.stopped { + return + } q.jobs <- path } // Stop closes the job queue and waits for in-flight jobs to finish. func (q *optimizeQueue) Stop() { + q.mu.Lock() + q.stopped = true close(q.jobs) + q.mu.Unlock() q.wg.Wait() } diff --git a/cmd/cbzoptimizer/commands/watch_command_test.go b/cmd/cbzoptimizer/commands/watch_command_test.go index 6d171b7..485c3fd 100644 --- a/cmd/cbzoptimizer/commands/watch_command_test.go +++ b/cmd/cbzoptimizer/commands/watch_command_test.go @@ -3,6 +3,7 @@ package commands import ( "os" "path/filepath" + "runtime" "sync" "sync/atomic" "testing" @@ -11,6 +12,7 @@ import ( utils2 "github.com/belphemur/CBZOptimizer/v2/internal/utils" "github.com/fsnotify/fsnotify" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestIsComicArchive(t *testing.T) { @@ -55,9 +57,20 @@ func TestShouldProcessWatchEvent(t *testing.T) { } func TestAddRecursiveWatchSkipsUnreadableSubdirectory(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("permission bits are not reliably enforced on Windows") + } + root := t.TempDir() nested := filepath.Join(root, "nested") assert.NoError(t, os.MkdirAll(nested, 0o755)) + assert.NoError(t, os.Chmod(nested, 0o000)) + defer func() { + assert.NoError(t, os.Chmod(nested, 0o755)) + }() + if _, err := os.ReadDir(nested); err == nil { + t.Skip("cannot make nested directory unreadable in this environment") + } watcher, err := fsnotify.NewWatcher() assert.NoError(t, err) @@ -99,11 +112,11 @@ func TestEventDebouncerCoalescesBurstsIntoSingleCall(t *testing.T) { // Simulate a burst of events for the same path. for i := 0; i < 5; i++ { debouncer.Trigger("/tmp/chapter.cbz") - time.Sleep(5 * time.Millisecond) } - time.Sleep(50 * time.Millisecond) - assert.EqualValues(t, 1, atomic.LoadInt32(&calls)) + require.Eventually(t, func() bool { + return atomic.LoadInt32(&calls) == 1 + }, 2*time.Second, 10*time.Millisecond) } func TestEventDebouncerHandlesMultiplePaths(t *testing.T) { @@ -119,22 +132,41 @@ func TestEventDebouncerHandlesMultiplePaths(t *testing.T) { debouncer.Trigger("/tmp/a.cbz") debouncer.Trigger("/tmp/b.cbz") - time.Sleep(50 * time.Millisecond) - - mu.Lock() - defer mu.Unlock() - assert.Equal(t, 1, seen["/tmp/a.cbz"]) - assert.Equal(t, 1, seen["/tmp/b.cbz"]) + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return seen["/tmp/a.cbz"] == 1 && seen["/tmp/b.cbz"] == 1 + }, 2*time.Second, 10*time.Millisecond) } func TestOptimizeQueueSkipsMissingPath(t *testing.T) { q := newOptimizeQueue(1, &utils2.OptimizeOptions{}) defer q.Stop() + var calls int32 + done := make(chan struct{}, 1) + existing := filepath.Join(t.TempDir(), "existing.cbz") + require.NoError(t, os.WriteFile(existing, []byte("data"), 0o644)) + q.optimize = func(options *utils2.OptimizeOptions) error { + if options.Path == existing { + atomic.AddInt32(&calls, 1) + select { + case done <- struct{}{}: + default: + } + } + return nil + } // Enqueue a path that doesn't exist; process should skip it without // panicking or blocking since it never reaches utils2.Optimize. q.Enqueue(filepath.Join(t.TempDir(), "missing.cbz")) + q.Enqueue(existing) - // Give the worker a moment to drain the job. - time.Sleep(50 * time.Millisecond) + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for existing path to be processed") + } + + assert.EqualValues(t, 1, atomic.LoadInt32(&calls)) }