From bbff2f4ba1b50ae463f3eeee133ed051983fce5d Mon Sep 17 00:00:00 2001 From: jamesread Date: Thu, 10 Sep 2026 16:20:28 +0100 Subject: [PATCH] fix: synchronize watcher and execution log state Co-authored-by: Cursor --- service/internal/executor/executor.go | 32 +++++++++++----- service/internal/executor/executor_test.go | 22 +++++++++++ .../internal/filehelper/file_change_notify.go | 8 +++- .../filehelper/file_change_notify_test.go | 37 +++++++++++++++++++ 4 files changed, 88 insertions(+), 11 deletions(-) create mode 100644 service/internal/filehelper/file_change_notify_test.go diff --git a/service/internal/executor/executor.go b/service/internal/executor/executor.go index 8bdec7ca..8cd4c8f7 100644 --- a/service/internal/executor/executor.go +++ b/service/internal/executor/executor.go @@ -153,6 +153,18 @@ type InternalLogEntry struct { TimedOut bool } +func cloneInternalLogEntry(entry *InternalLogEntry) *InternalLogEntry { + if entry == nil { + return nil + } + + cloned := *entry + cloned.Arguments = maps.Clone(entry.Arguments) + cloned.Tags = slices.Clone(entry.Tags) + + return &cloned +} + // .Binding can be nil, so we need to handle that. func (e *InternalLogEntry) GetBindingId() string { if e.Binding == nil { @@ -273,7 +285,7 @@ func (e *Executor) GetLogTrackingIds(startOffset int64, pageCount int64) ([]*Int if totalLogCount > 0 { for i := startIndex; i >= endIndex; i-- { - trackingIds = append(trackingIds, e.logs[e.logsTrackingIdsByDate[i]]) + trackingIds = append(trackingIds, cloneInternalLogEntry(e.logs[e.logsTrackingIdsByDate[i]])) } } @@ -303,7 +315,7 @@ func (e *Executor) filterLogsByACL(cfg *config.Config, user *authpublic.Authenti entry := e.logs[trackingId] if shouldIncludeLogEntry(cfg, user, entry, filterDate, hasDateFilter) { - filtered = append(filtered, entry) + filtered = append(filtered, cloneInternalLogEntry(entry)) } } @@ -399,26 +411,28 @@ func (e *Executor) GetLogTrackingIdsACL(cfg *config.Config, user *authpublic.Aut func (e *Executor) GetLog(trackingID string) (*InternalLogEntry, bool) { e.logmutex.RLock() + defer e.logmutex.RUnlock() entry, found := e.logs[trackingID] - - e.logmutex.RUnlock() - - return entry, found + return cloneInternalLogEntry(entry), found } func (e *Executor) GetLogsByBindingId(bindingId string) []*InternalLogEntry { e.logmutex.RLock() + defer e.logmutex.RUnlock() logs, found := e.LogsByBindingId[bindingId] - e.logmutex.RUnlock() - if !found { return make([]*InternalLogEntry, 0) } - return logs + cloned := make([]*InternalLogEntry, 0, len(logs)) + for _, entry := range logs { + cloned = append(cloned, cloneInternalLogEntry(entry)) + } + + return cloned } // shouldCountExecution checks if a log entry should be counted for rate limiting. diff --git a/service/internal/executor/executor_test.go b/service/internal/executor/executor_test.go index 2261a172..f7c23002 100644 --- a/service/internal/executor/executor_test.go +++ b/service/internal/executor/executor_test.go @@ -38,6 +38,28 @@ func testingExecutor() (*Executor, *config.Config) { return e, cfg } +func TestGetLogReturnsDefensiveCopy(t *testing.T) { + e := DefaultExecutor(config.DefaultConfig()) + e.logs["tracking-id"] = &InternalLogEntry{ + Arguments: map[string]string{"message": "original"}, + Output: "original", + Tags: []string{"original"}, + } + + entry, found := e.GetLog("tracking-id") + require.True(t, found) + + entry.Arguments["message"] = "changed" + entry.Output = "changed" + entry.Tags[0] = "changed" + + stored, found := e.GetLog("tracking-id") + require.True(t, found) + assert.Equal(t, "original", stored.Arguments["message"]) + assert.Equal(t, "original", stored.Output) + assert.Equal(t, []string{"original"}, stored.Tags) +} + func TestCreateExecutorAndExec(t *testing.T) { e, cfg := testingExecutor() diff --git a/service/internal/filehelper/file_change_notify.go b/service/internal/filehelper/file_change_notify.go index a6aec552..5f38536a 100644 --- a/service/internal/filehelper/file_change_notify.go +++ b/service/internal/filehelper/file_change_notify.go @@ -273,13 +273,17 @@ func processDebounce(ctx *watchContext) { if logEntry.callbackComplete || logEntry.callbackWrapper == nil { log.Debugf("fsnotify event callback queued within debounce delay: %v", ctx.filename) + callback := ctx.callback + eventName := ctx.event.Name logEntry.callbackComplete = false logEntry.callbackWrapper = time.AfterFunc(debounceDelay, func() { - log.Debugf("fsnotify event callback being fired: %v", ctx.filename) + log.Debugf("fsnotify event callback being fired: %v", eventName) - ctx.callback(ctx.event.Name) + callback(eventName) + debounceWriteLogMutex.Lock() logEntry.callbackComplete = true + debounceWriteLogMutex.Unlock() }) } else { log.Debugf("fsnotify event suppressed because it's within the debounce delay: %v", ctx.filename) diff --git a/service/internal/filehelper/file_change_notify_test.go b/service/internal/filehelper/file_change_notify_test.go new file mode 100644 index 00000000..48c1d88d --- /dev/null +++ b/service/internal/filehelper/file_change_notify_test.go @@ -0,0 +1,37 @@ +package filehelper + +import ( + "testing" + "time" + + "github.com/fsnotify/fsnotify" + "github.com/stretchr/testify/require" +) + +func TestProcessDebounceCapturesEventName(t *testing.T) { + debounceWriteLogMutex.Lock() + debounceWriteLog = make(map[string]*FsNotifyLogEntry) + debounceWriteLogMutex.Unlock() + + callbackNames := make(chan string, 1) + firstEvent := fsnotify.Event{Name: "first"} + ctx := &watchContext{ + callback: func(filename string) { + callbackNames <- filename + }, + event: &firstEvent, + filename: t.Name(), + } + + processDebounce(ctx) + + secondEvent := fsnotify.Event{Name: "second"} + ctx.event = &secondEvent + + select { + case callbackName := <-callbackNames: + require.Equal(t, firstEvent.Name, callbackName) + case <-time.After(time.Second): + t.Fatal("debounced callback did not run") + } +}