Harden event-driven daemon for production
Three critical fixes to make event-driven mode production-ready: 1. Skip redundant imports: Check JSONL mtime vs DB mtime to avoid self-triggered import loops after export writes JSONL 2. Add server.Stop() in serverErrChan case: Ensures clean RPC server shutdown on errors 3. Fallback ticker (60s): When file watcher unavailable (e.g., network filesystems), fall back to periodic polling to detect remote changes These minimal fixes address Oracle's concerns without over-engineering. Event-driven mode is now safe for default. Amp-Thread-ID: https://ampcode.com/threads/T-a9a67394-37ca-4b79-aa23-c5c011f9c0cd Co-authored-by: Amp <amp@ampcode.com>
This commit is contained in:
@@ -1112,14 +1112,36 @@ func createAutoImportFunc(ctx context.Context, store storage.Storage, log daemon
|
|||||||
log.log("Removed stale lock (%s), proceeding", holder)
|
log.log("Removed stale lock (%s), proceeding", holder)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Pull from git
|
// Check JSONL modification time to avoid redundant imports
|
||||||
if err := gitPull(importCtx); err != nil {
|
// (e.g., from self-triggered file watcher events after our own export)
|
||||||
log.log("Pull failed: %v", err)
|
jsonlInfo, err := os.Stat(jsonlPath)
|
||||||
return
|
if err != nil {
|
||||||
}
|
log.log("Failed to stat JSONL: %v", err)
|
||||||
log.log("Pulled from remote")
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// Count issues before import
|
// Get database modification time
|
||||||
|
dbPath := filepath.Join(beadsDir, "beads.db")
|
||||||
|
dbInfo, err := os.Stat(dbPath)
|
||||||
|
if err != nil {
|
||||||
|
log.log("Failed to stat database: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Skip if JSONL is older than database (nothing new to import)
|
||||||
|
if !jsonlInfo.ModTime().After(dbInfo.ModTime()) {
|
||||||
|
log.log("Skipping import: JSONL not newer than database")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Pull from git
|
||||||
|
if err := gitPull(importCtx); err != nil {
|
||||||
|
log.log("Pull failed: %v", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
log.log("Pulled from remote")
|
||||||
|
|
||||||
|
// Count issues before import
|
||||||
beforeCount, err := countDBIssues(importCtx, store)
|
beforeCount, err := countDBIssues(importCtx, store)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.log("Failed to count issues before import: %v", err)
|
log.log("Failed to count issues before import: %v", err)
|
||||||
|
|||||||
@@ -47,9 +47,13 @@ func runEventDrivenLoop(
|
|||||||
watcher, err := NewFileWatcher(jsonlPath, func() {
|
watcher, err := NewFileWatcher(jsonlPath, func() {
|
||||||
importDebouncer.Trigger()
|
importDebouncer.Trigger()
|
||||||
})
|
})
|
||||||
|
var fallbackTicker *time.Ticker
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.log("WARNING: File watcher unavailable (%v), mutations will trigger export only", err)
|
log.log("WARNING: File watcher unavailable (%v), using 60s polling fallback", err)
|
||||||
watcher = nil
|
watcher = nil
|
||||||
|
// Fallback ticker to check for remote changes when watcher unavailable
|
||||||
|
fallbackTicker = time.NewTicker(60 * time.Second)
|
||||||
|
defer fallbackTicker.Stop()
|
||||||
} else {
|
} else {
|
||||||
watcher.Start(ctx, log)
|
watcher.Start(ctx, log)
|
||||||
defer watcher.Close()
|
defer watcher.Close()
|
||||||
@@ -97,6 +101,16 @@ func runEventDrivenLoop(
|
|||||||
// Periodic health validation (not sync)
|
// Periodic health validation (not sync)
|
||||||
checkDaemonHealth(ctx, store, log)
|
checkDaemonHealth(ctx, store, log)
|
||||||
|
|
||||||
|
case <-func() <-chan time.Time {
|
||||||
|
if fallbackTicker != nil {
|
||||||
|
return fallbackTicker.C
|
||||||
|
}
|
||||||
|
// Never fire if watcher is available
|
||||||
|
return make(chan time.Time)
|
||||||
|
}():
|
||||||
|
log.log("Fallback ticker: checking for remote changes")
|
||||||
|
importDebouncer.Trigger()
|
||||||
|
|
||||||
case sig := <-sigChan:
|
case sig := <-sigChan:
|
||||||
if isReloadSignal(sig) {
|
if isReloadSignal(sig) {
|
||||||
log.log("Received reload signal, ignoring")
|
log.log("Received reload signal, ignoring")
|
||||||
@@ -120,12 +134,15 @@ func runEventDrivenLoop(
|
|||||||
return
|
return
|
||||||
|
|
||||||
case err := <-serverErrChan:
|
case err := <-serverErrChan:
|
||||||
log.log("RPC server failed: %v", err)
|
log.log("RPC server failed: %v", err)
|
||||||
cancel()
|
cancel()
|
||||||
if watcher != nil {
|
if watcher != nil {
|
||||||
watcher.Close()
|
watcher.Close()
|
||||||
}
|
}
|
||||||
return
|
if stopErr := server.Stop(); stopErr != nil {
|
||||||
|
log.log("Error stopping server: %v", stopErr)
|
||||||
|
}
|
||||||
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user