package config import ( "context" "errors" "fmt" "os" "reflect" "slices" "sync" "sync/atomic" "time" ) type WatchOptions struct { PollInterval time.Duration // Minimum 100ms; each poll reads and fingerprints the file. Debounce time.Duration // Quiet interval before publishing a value or error. MaxWatchers int ReloadTimeout time.Duration // Expired reads/parses cannot publish. VerifyPermissions bool } func DefaultWatchOptions() WatchOptions { return WatchOptions{DefaultPollInterval, DefaultDebounce, DefaultMaxWatchers, DefaultReloadTimeout, true} } type watcher struct { mu sync.Mutex ctx context.Context cancel context.CancelFunc opts WatchOptions filePath string lastMode os.FileMode watching atomic.Bool watchers map[int64]chan string watcherID int64 observed string handled string timeoutKey string observedAt time.Time read func(context.Context, string, fileSettings) (*fileSnapshot, error) } func normalizeWatchOptions(opts WatchOptions) WatchOptions { if opts.PollInterval < MinPollInterval { opts.PollInterval = MinPollInterval } if opts.Debounce < 0 { opts.Debounce = 0 } if opts.MaxWatchers <= 0 { opts.MaxWatchers = DefaultMaxWatchers } if opts.ReloadTimeout <= 0 { opts.ReloadTimeout = DefaultReloadTimeout } return opts } func (c *Config) AutoUpdate() { c.AutoUpdateWithOptions(DefaultWatchOptions()) } func (c *Config) AutoUpdateWithOptions(opts WatchOptions) { c.mutex.Lock() defer c.mutex.Unlock() c.startWatcherLocked(opts) } func (c *Config) startWatcherLocked(opts WatchOptions) { if c.configFilePath == "" { return } if c.watcher != nil { if c.watcher.filePath == c.configFilePath && c.watcher.watching.Load() { return } c.watcher.stop() } ctx, cancel := context.WithCancel(context.Background()) w := &watcher{ctx: ctx, cancel: cancel, opts: normalizeWatchOptions(opts), filePath: c.configFilePath, watchers: make(map[int64]chan string), read: readConfigFile} if info, err := os.Stat(w.filePath); err == nil { w.lastMode = info.Mode() } w.watching.Store(true) c.watcher = w go w.watchLoop(c) } func (c *Config) StopAutoUpdate() { c.mutex.Lock() defer c.mutex.Unlock() if c.watcher != nil { c.watcher.stop() c.watcher = nil } } func (c *Config) Watch() <-chan string { return c.WatchWithOptions(DefaultWatchOptions()) } func (c *Config) WatchWithOptions(opts WatchOptions) <-chan string { c.mutex.Lock() defer c.mutex.Unlock() c.startWatcherLocked(opts) if c.watcher == nil { return closedWatchChannel() } return c.watcher.subscribe() } func (c *Config) WatchFile(path string, formatHint ...string) error { if len(formatHint) > 1 { return ErrFileFormat } if len(formatHint) == 1 { if err := c.SetFileFormat(formatHint[0]); err != nil { return err } } c.mutex.RLock() opts := DefaultWatchOptions() if c.watcher != nil { opts = c.watcher.opts } c.mutex.RUnlock() // A failed load leaves the existing watcher and valid state intact. if err := c.LoadFile(path); err != nil { return err } c.AutoUpdateWithOptions(opts) return nil } func (c *Config) IsWatching() bool { c.mutex.RLock() defer c.mutex.RUnlock() return c.watcher != nil && c.watcher.watching.Load() } func (c *Config) WatcherCount() int { c.mutex.RLock() defer c.mutex.RUnlock() if c.watcher == nil { return 0 } c.watcher.mu.Lock() defer c.watcher.mu.Unlock() return len(c.watcher.watchers) } func (w *watcher) watchLoop(c *Config) { ticker := time.NewTicker(w.opts.PollInterval) defer ticker.Stop() for { select { case <-w.ctx.Done(): return case <-ticker.C: w.checkAndReload(c) } } } // One loop owns observations and debounce state. I/O runs outside the Config lock; // publication checks context, watcher identity and file generation under that lock. func (w *watcher) checkAndReload(c *Config) { if w.ctx.Err() != nil { return } ctx, cancel := context.WithTimeout(w.ctx, w.opts.ReloadTimeout) defer cancel() c.mutex.RLock() opts := c.fileSettingsLocked() generation := c.fileGeneration c.mutex.RUnlock() snapshot, readErr := w.read(ctx, w.filePath, opts) key := "" event := "" if readErr != nil { key = "error:" + readErr.Error() switch { case errors.Is(readErr, context.DeadlineExceeded): event = EventReloadTimeout case errors.Is(readErr, ErrConfigNotFound): event = EventFileDeleted default: event = EventReloadError + ":" + readErr.Error() } } else { key = fmt.Sprintf("%x:%o", snapshot.digest, snapshot.info.Mode()) if w.opts.VerifyPermissions && w.lastMode != 0 && (snapshot.info.Mode()&0077) != (w.lastMode&0077) { event = EventPermissionsChanged } } if key != w.observed { w.observed, w.observedAt = key, time.Now() } if key == w.handled || time.Since(w.observedAt) < w.opts.Debounce { return } if w.ctx.Err() != nil { return } if event != "" { if event == EventReloadTimeout { w.notifyTimeout(key) } else { w.notifyWatchers(event) w.handled = key } return } root, comments, err := parseFile(snapshot) if err != nil { w.notifyWatchers(EventReloadError + ":" + err.Error()) w.handled = key return } c.mutex.Lock() defer c.mutex.Unlock() if c.watcher != w || c.configFilePath != w.filePath || c.fileGeneration != generation || w.ctx.Err() != nil { return } if ctx.Err() != nil { w.notifyTimeout(key) return } old := make(map[string]any, len(c.items)) for path, item := range c.items { old[path] = item.currentValue } if err := c.applyFileLocked(root, comments, w.filePath, ctx.Err); err != nil { if errors.Is(err, context.DeadlineExceeded) { w.notifyTimeout(key) } else { w.notifyWatchers(EventReloadError + ":" + err.Error()) w.handled = key } return } w.lastMode, w.handled = snapshot.info.Mode(), key w.timeoutKey = "" var changed []string for path, item := range c.items { if !reflect.DeepEqual(old[path], item.currentValue) { changed = append(changed, path) } } slices.Sort(changed) for _, path := range changed { w.notifyWatchers(path) } } // A timeout is retryable even when the file contents have not changed. Suppress // repeated notifications for the same observation without suppressing its retry. func (w *watcher) notifyTimeout(key string) { if w.timeoutKey != key { w.notifyWatchers(EventReloadTimeout) w.timeoutKey = key } } func closedWatchChannel() <-chan string { ch := make(chan string); close(ch); return ch } func (w *watcher) subscribe() <-chan string { w.mu.Lock() defer w.mu.Unlock() if w.ctx.Err() != nil || len(w.watchers) >= w.opts.MaxWatchers { return closedWatchChannel() } ch := make(chan string, WatchChannelBuffer) w.watcherID++ w.watchers[w.watcherID] = ch return ch } func (w *watcher) notifyWatchers(path string) { w.mu.Lock() defer w.mu.Unlock() if w.ctx.Err() != nil { return } for _, ch := range w.watchers { select { case ch <- path: default: } } } func (w *watcher) stop() { w.mu.Lock() defer w.mu.Unlock() w.cancel() w.watching.Store(false) for id, ch := range w.watchers { close(ch) delete(w.watchers, id) } } func (c *Config) snapshot() map[string]any { c.mutex.RLock() defer c.mutex.RUnlock() out := make(map[string]any, len(c.items)) for path, item := range c.items { out[path] = cloneOwned(item.currentValue) } return out }