lixen/config
SHA256
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.2.0 toml-only, harden concurrency and env export, dep updates
Test / test (push) Canceled after 0s
Details

Browse Source
This commit is contained in:
lixen committed 2026-09-28 16:55:58 -04:00
1 parent 3c35601179
commit e08467bf38
42 files changed
+2533 -4274
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

watch.go
+176 -311
View File
@@ -1,240 +1,146 @@
// FILE: lixenwraith/config/watch.go
package config
import (
"context"
"errors"
"fmt"
"os"
"reflect"
"slices"
"sync"
"sync/atomic"
"time"
)
// WatchOptions configures file watching behavior
type WatchOptions struct {
// PollInterval for file stat checks (minimum 100ms)
PollInterval time.Duration
// Debounce duration to avoid rapid reloads
Debounce time.Duration
// MaxWatchers limits concurrent watch channels
MaxWatchers int
// ReloadTimeout for file reload operations
ReloadTimeout time.Duration
// VerifyPermissions checks file hasn't been replaced with different permissions
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
}
// DefaultWatchOptions returns sensible defaults for file watching
func DefaultWatchOptions() WatchOptions {
return WatchOptions{
PollInterval: DefaultPollInterval,
Debounce: DefaultDebounce,
MaxWatchers: DefaultMaxWatchers,
ReloadTimeout: DefaultReloadTimeout,
VerifyPermissions: true,
}
return WatchOptions{DefaultPollInterval, DefaultDebounce, DefaultMaxWatchers, DefaultReloadTimeout, true}
}
// watcher manages file watching state
type watcher struct {
mu sync.RWMutex
ctx context.Context
cancel context.CancelFunc
opts WatchOptions
filePath string
lastModTime time.Time
lastSize int64
lastMode os.FileMode
watching atomic.Bool
reloadInProgress atomic.Bool
watchers map[int64]chan string // subscriber channels
watcherID atomic.Int64
debounceTimer *time.Timer
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)
}
// configWatcher extends Config with watching capabilities
type configWatcher struct {
*Config
watcher *watcher
}
// AutoUpdate enables automatic configuration reloading when the file changes
func (c *Config) AutoUpdate() {
c.AutoUpdateWithOptions(DefaultWatchOptions())
}
// AutoUpdateWithOptions enables automatic configuration reloading with custom options
func (c *Config) AutoUpdateWithOptions(opts WatchOptions) {
// Validate options
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 = 100
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()
// Get path of current file to watch
filePath := c.getConfigFilePath()
if filePath == "" {
// No file configured, nothing to watch
c.startWatcherLocked(opts)
}
func (c *Config) startWatcherLocked(opts WatchOptions) {
if c.configFilePath == "" {
return
}
// Stop existing watcher if path changed
if c.watcher != nil && c.watcher.filePath != filePath {
if c.watcher != nil {
if c.watcher.filePath == c.configFilePath && c.watcher.watching.Load() {
return
}
c.watcher.stop()
c.watcher = nil
}
// Initialize watcher if needed
if c.watcher == nil {
ctx, cancel := context.WithCancel(context.Background())
c.watcher = &watcher{
ctx: ctx,
cancel: cancel,
opts: opts,
filePath: filePath,
watchers: make(map[int64]chan string),
}
// Get initial file state
if info, err := os.Stat(filePath); err == nil {
c.watcher.lastModTime = info.ModTime()
c.watcher.lastSize = info.Size()
c.watcher.lastMode = info.Mode()
}
// Start watching
go c.watcher.watchLoop(c)
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)
}
// StopAutoUpdate stops automatic configuration reloading
func (c *Config) StopAutoUpdate() {
c.mutex.Lock()
defer c.mutex.Unlock()
if c.watcher != nil {
c.watcher.stop()
c.watcher = nil
}
}
// Watch returns a channel that receives paths of changed configuration values
func (c *Config) Watch() <-chan string {
return c.WatchWithOptions(DefaultWatchOptions())
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()
}
// WatchFile stops any existing file watcher, loads a new configuration file,
// and starts a new watcher on that file path. Optionally accepts format hint
func (c *Config) WatchFile(filePath string, formatHint ...string) error {
// Stop any currently running watcher
c.StopAutoUpdate()
// Set format hint if provided
if len(formatHint) > 0 {
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 wrapError(ErrFileFormat, fmt.Errorf("invalid format hint: %w", err))
return err
}
}
// Load the new file
if err := c.LoadFile(filePath); err != nil {
return fmt.Errorf("failed to load new file for watching: %w", err) // Already wrapped with error category in LoadFile
}
// Get previous watcher options if available
c.mutex.RLock()
opts := DefaultWatchOptions()
if c.watcher != nil {
opts = c.watcher.opts
}
c.mutex.RUnlock()
// Create and start a new watcher with the new file path
// 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
}
// WatchWithOptions returns a channel with custom watch options
// should not restart the watcher if it's already running with the same file
func (c *Config) WatchWithOptions(opts WatchOptions) <-chan string {
c.mutex.RLock()
watcher := c.watcher
filePath := c.configFilePath
c.mutex.RUnlock()
// If no file configured, return closed channel
if filePath == "" {
ch := make(chan string)
close(ch)
return ch
}
// If watcher exists and is watching the current file, just subscribe
if watcher != nil && watcher.filePath == filePath && watcher.watching.Load() {
return watcher.subscribe()
}
// First ensure auto-update is running
c.AutoUpdateWithOptions(opts)
c.mutex.RLock()
watcher = c.watcher
c.mutex.RUnlock()
if watcher == nil {
// No file to watch, return closed channel
ch := make(chan string)
close(ch)
return ch
}
return watcher.subscribe()
}
// IsWatching returns true if auto-update is enabled
func (c *Config) IsWatching() bool {
c.mutex.RLock()
defer c.mutex.RUnlock()
return c.watcher != nil && c.watcher.watching.Load()
}
// WatcherCount returns the number of active watch channels
func (c *Config) WatcherCount() int {
c.mutex.RLock()
defer c.mutex.RUnlock()
if c.watcher == nil {
return 0
}
c.watcher.mu.RLock()
defer c.watcher.mu.RUnlock()
c.watcher.mu.Lock()
defer c.watcher.mu.Unlock()
return len(c.watcher.watchers)
}
// watchLoop is the main file watching loop
func (w *watcher) watchLoop(c *Config) {
if !w.watching.CompareAndSwap(false, true) {
return // Already watching
}
defer w.watching.Store(false)
ticker := time.NewTicker(w.opts.PollInterval)
defer ticker.Stop()
for {
select {
case <-w.ctx.Done():
@@ -245,188 +151,147 @@ func (w *watcher) watchLoop(c *Config) {
}
}
// checkAndReload checks if file changed and triggers reload
// 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) {
info, err := os.Stat(w.filePath)
if err != nil {
if os.IsNotExist(err) {
// File was deleted, notify watchers
w.notifyWatchers(EventFileDeleted)
}
if w.ctx.Err() != nil {
return
}
// Check for changes
changed := false
// Compare modification time and size
if !info.ModTime().Equal(w.lastModTime) || info.Size() != w.lastSize {
changed = true
}
// SECURITY: Verify permissions haven't changed suspiciously
if w.opts.VerifyPermissions && w.lastMode != 0 {
if info.Mode() != w.lastMode {
// Permission change detected
if (info.Mode() & 0077) != (w.lastMode & 0077) {
// World/group permissions changed - potential security issue
w.notifyWatchers(EventPermissionsChanged)
// Don't reload on permission change for security
return
}
}
}
if changed {
// Update tracked state
w.lastModTime = info.ModTime()
w.lastSize = info.Size()
w.lastMode = info.Mode()
// Debounce rapid changes
w.mu.Lock()
if w.debounceTimer != nil {
w.debounceTimer.Stop()
}
w.debounceTimer = time.AfterFunc(w.opts.Debounce, func() {
w.performReload(c)
})
w.mu.Unlock()
}
}
// performReload reloads the configuration file
func (w *watcher) performReload(c *Config) {
// Prevent concurrent reloads
if !w.reloadInProgress.CompareAndSwap(false, true) {
return
}
defer w.reloadInProgress.Store(false)
// Create a timeout context for reload
ctx, cancel := context.WithTimeout(w.ctx, w.opts.ReloadTimeout)
defer cancel()
// Track what changed
oldValues := c.snapshot()
// Reload file in a goroutine with timeout
done := make(chan error, 1)
go func() {
done <- c.loadFile(w.filePath)
}()
select {
case err := <-done:
if err != nil {
// Reload failed, notify error
w.notifyWatchers(fmt.Sprintf("%s:%v", EventReloadError, err))
return
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()
}
// Compare and notify changes
newValues := c.snapshot()
for path, newVal := range newValues {
if oldVal, existed := oldValues[path]; !existed || !reflect.DeepEqual(oldVal, newVal) {
w.notifyWatchers(path)
}
} 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
}
// Check for deletions
for path := range oldValues {
if _, exists := newValues[path]; !exists {
w.notifyWatchers(path)
}
}
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
}
case <-ctx.Done():
// Reload timeout
w.notifyWatchers(EventReloadTimeout)
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)
}
}
// subscribe creates a new watcher channel
// 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()
// Check watcher limit
if len(w.watchers) >= w.opts.MaxWatchers {
// Return closed channel to prevent resource exhaustion
ch := make(chan string)
close(ch)
return ch
if w.ctx.Err() != nil || len(w.watchers) >= w.opts.MaxWatchers {
return closedWatchChannel()
}
// Create buffered channel to prevent blocking
ch := make(chan string, WatchChannelBuffer)
id := w.watcherID.Add(1)
w.watchers[id] = ch
// Cleanup goroutine
go func() {
<-w.ctx.Done()
w.mu.Lock()
delete(w.watchers, id)
close(ch)
w.mu.Unlock()
}()
w.watcherID++
w.watchers[w.watcherID] = ch
return ch
}
// notifyWatchers sends change notification to all subscribers
func (w *watcher) notifyWatchers(path string) {
w.mu.RLock()
defer w.mu.RUnlock()
for id, ch := range w.watchers {
w.mu.Lock()
defer w.mu.Unlock()
if w.ctx.Err() != nil {
return
}
for _, ch := range w.watchers {
select {
case ch <- path:
// Sent successfully
default:
// Channel full or closed, skip
// Could implement removal of dead watchers here
_ = id
}
}
}
// stop terminates the watcher
func (w *watcher) stop() {
if w.cancel != nil {
w.cancel()
}
// Stop debounce timer
w.mu.Lock()
if w.debounceTimer != nil {
w.debounceTimer.Stop()
w.debounceTimer = nil
}
w.mu.Unlock()
// Wait for watch loop to exit with timeout
deadline := time.Now().Add(ShutdownTimeout)
for w.watching.Load() && time.Now().Before(deadline) {
time.Sleep(SpinWaitInterval)
defer w.mu.Unlock()
w.cancel()
w.watching.Store(false)
for id, ch := range w.watchers {
close(ch)
delete(w.watchers, id)
}
}
// getConfigFilePath returns the current config file path
func (c *Config) getConfigFilePath() string {
// Access the tracked config file path
return c.configFilePath
}
// snapshot creates a snapshot of current values
func (c *Config) snapshot() map[string]any {
c.mutex.RLock()
defer c.mutex.RUnlock()
snapshot := make(map[string]any, len(c.items))
out := make(map[string]any, len(c.items))
for path, item := range c.items {
snapshot[path] = item.currentValue
out[path] = cloneOwned(item.currentValue)
}
return snapshot
}
return out
}