fix: add mtime+size stability check and failure quarantine to watcher
Two related hardening fixes for the input folder poller (bug #22): 1. Partial-write protection. A freshly-dropped .mkv now has to present the same (mtime, size) on two consecutive ticks before processFn is called. A file mid-rsync that grows or has a moving mtime is skipped until it settles. The per-file (mtime, size) state lives in Watcher.seen and is rebuilt from the current glob each tick so the map cannot grow unboundedly. 2. Failure quarantine. When processFn returns an error, the file's (failedAt, mtime) is recorded in Watcher.failed and subsequent ticks skip it until either the backoff elapses or its mtime changes (user replaced or touched it). Previously a permanently-broken input -- e.g. a video-only mkv that trips the "no audio streams found" path added in the multi-audio fix -- would be retried and logged every 15 s forever. Backoff is 5 minutes: comfortably longer than the 10-30 s polling interval clamp so we are not effectively retrying every tick, but short enough that an operator fixing the underlying problem by replacing the file sees it picked up promptly on the next stable scan.
This commit is contained in:
@@ -2,15 +2,46 @@ package watcher
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"os"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"videnc-vibe/pkg/types"
|
"videnc-vibe/pkg/types"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// failureBackoff is how long we skip a file after processFn returned an error
|
||||||
|
// for it, provided the file has not been touched since (mtime unchanged). Chosen
|
||||||
|
// to be much larger than the polling interval so a permanently-broken input
|
||||||
|
// (e.g. video-only mkv producing "no audio streams found") does not spam the
|
||||||
|
// log every tick, but short enough that an operator who fixes the underlying
|
||||||
|
// problem by replacing the file sees it picked up promptly on the next stable
|
||||||
|
// scan.
|
||||||
|
const failureBackoff = 5 * time.Minute
|
||||||
|
|
||||||
|
// fileStat is the (mtime, size) pair used to decide whether a file has settled
|
||||||
|
// between two consecutive ticks.
|
||||||
|
type fileStat struct {
|
||||||
|
mtime time.Time
|
||||||
|
size int64
|
||||||
|
}
|
||||||
|
|
||||||
|
// failureRecord remembers that processFn failed for a given file, so we can
|
||||||
|
// back off rather than retrying on every tick.
|
||||||
|
type failureRecord struct {
|
||||||
|
failedAt time.Time
|
||||||
|
fileMtime time.Time
|
||||||
|
}
|
||||||
|
|
||||||
type Watcher struct {
|
type Watcher struct {
|
||||||
inputDir string
|
inputDir string
|
||||||
interval time.Duration
|
interval time.Duration
|
||||||
|
// seen tracks the last-observed (mtime, size) for every file currently in
|
||||||
|
// the input dir. A file is only handed to processFn once two consecutive
|
||||||
|
// ticks agree on both fields (partial-write protection).
|
||||||
|
seen map[string]fileStat
|
||||||
|
// failed tracks files that recently errored from processFn so we can skip
|
||||||
|
// them until either failureBackoff elapses or their mtime changes.
|
||||||
|
failed map[string]failureRecord
|
||||||
}
|
}
|
||||||
|
|
||||||
func New(inputDir string, intervalSeconds int) *Watcher {
|
func New(inputDir string, intervalSeconds int) *Watcher {
|
||||||
@@ -24,6 +55,8 @@ func New(inputDir string, intervalSeconds int) *Watcher {
|
|||||||
return &Watcher{
|
return &Watcher{
|
||||||
inputDir: inputDir,
|
inputDir: inputDir,
|
||||||
interval: interval,
|
interval: interval,
|
||||||
|
seen: make(map[string]fileStat),
|
||||||
|
failed: make(map[string]failureRecord),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -50,13 +83,54 @@ func (w *Watcher) scanAndProcess(processFn func(string) error) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
now := time.Now()
|
||||||
|
nextSeen := make(map[string]fileStat, len(files))
|
||||||
|
nextFailed := make(map[string]failureRecord, len(w.failed))
|
||||||
|
|
||||||
for _, file := range files {
|
for _, file := range files {
|
||||||
|
info, err := os.Stat(file)
|
||||||
|
if err != nil {
|
||||||
|
// File vanished between glob and stat, or unreadable. Drop any
|
||||||
|
// state for it by not carrying it forward.
|
||||||
|
fmt.Printf("Error stating %s: %v\n", file, err)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
current := fileStat{mtime: info.ModTime(), size: info.Size()}
|
||||||
|
nextSeen[file] = current
|
||||||
|
|
||||||
|
// Carry the failure record forward only if the file hasn't been
|
||||||
|
// touched since it failed; a changed mtime means the user replaced
|
||||||
|
// or modified the file and we should give it another chance.
|
||||||
|
if rec, ok := w.failed[file]; ok && rec.fileMtime.Equal(current.mtime) {
|
||||||
|
if now.Sub(rec.failedAt) < failureBackoff {
|
||||||
|
nextFailed[file] = rec
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Backoff expired; clear the record and let the file be
|
||||||
|
// re-processed if it is otherwise stable.
|
||||||
|
}
|
||||||
|
|
||||||
|
prev, ok := w.seen[file]
|
||||||
|
if !ok || prev.mtime != current.mtime || prev.size != current.size {
|
||||||
|
// First time we've seen this (mtime, size); wait one more tick
|
||||||
|
// to make sure the file isn't still being written.
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
if err := processFn(file); err != nil {
|
if err := processFn(file); err != nil {
|
||||||
fmt.Printf("Error processing %s: %v\n", file, err)
|
fmt.Printf("Error processing %s: %v\n", file, err)
|
||||||
|
nextFailed[file] = failureRecord{
|
||||||
|
failedAt: now,
|
||||||
|
fileMtime: current.mtime,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
w.seen = nextSeen
|
||||||
|
w.failed = nextFailed
|
||||||
|
}
|
||||||
|
|
||||||
func DetectMediaType(width, height int) types.MediaType {
|
func DetectMediaType(width, height int) types.MediaType {
|
||||||
pixels := width * height
|
pixels := width * height
|
||||||
if pixels < 600000 {
|
if pixels < 600000 {
|
||||||
|
|||||||
Reference in New Issue
Block a user