From 5d9283be34ae2d6c0727cd811a72084081e3b5a6 Mon Sep 17 00:00:00 2001 From: Daulet Zhanguzin Date: Sat, 10 Apr 2021 19:49:59 +0100 Subject: [PATCH 1/2] Add a test to reproduce missed file update --- app/discovery/provider/file_test.go | 48 +++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/app/discovery/provider/file_test.go b/app/discovery/provider/file_test.go index 464f1bc..0d1465d 100644 --- a/app/discovery/provider/file_test.go +++ b/app/discovery/provider/file_test.go @@ -4,6 +4,7 @@ import ( "context" "io/ioutil" "os" + "sync" "testing" "time" @@ -48,9 +49,56 @@ func TestFile_Events(t *testing.T) { t.Log("event") events++ } + // expecting events from creation + 3 writes assert.Equal(t, 4, events) } +func TestFile_Events_BusyListener(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + + tmp, err := ioutil.TempFile(os.TempDir(), "reproxy-events-busy") + require.NoError(t, err) + _ = tmp.Close() + defer os.Remove(tmp.Name()) + + f := File{ + FileName: tmp.Name(), + CheckInterval: 10 * time.Millisecond, + Delay: 20 * time.Millisecond, + } + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + + for i := 0; i < 2; i++ { + time.Sleep(30 * time.Millisecond) + assert.NoError(t, ioutil.WriteFile(tmp.Name(), []byte("something"), 0600)) + } + }() + + ch := f.Events(ctx) + // exhaust creation and one write event + for i := 0; i < 2; i++ { + t.Log("event") + <- ch + } + + // wait until last write definitely has happened + wg.Wait() + // stay busy for CheckInterval before accepting from channel + time.Sleep(10 * time.Millisecond) + + events := 0 + for range ch { + t.Log("event") + events++ + } + assert.Equal(t, 1, events) +} + func TestFile_List(t *testing.T) { f := File{FileName: "testdata/config.yml"} From 936a0815a028d6101646424efe5e121958c5ae63 Mon Sep 17 00:00:00 2001 From: Daulet Zhanguzin Date: Sat, 10 Apr 2021 19:56:05 +0100 Subject: [PATCH 2/2] Fix missed file update if consumer of events is busy, e.g. waiting on a lock --- app/discovery/provider/file.go | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/app/discovery/provider/file.go b/app/discovery/provider/file.go index 971cf07..aeb119b 100644 --- a/app/discovery/provider/file.go +++ b/app/discovery/provider/file.go @@ -26,10 +26,12 @@ func (d *File) Events(ctx context.Context) <-chan struct{} { res := make(chan struct{}) // no need to queue multiple events - trySubmit := func(ch chan struct{}) { + trySubmit := func(ch chan struct{}) bool { select { case ch <- struct{}{}: + return true default: + return false } } @@ -50,8 +52,9 @@ func (d *File) Events(ctx context.Context) <-chan struct{} { } log.Printf("[DEBUG] file %s changed, %s -> %s", d.FileName, lastModif.Format(time.RFC3339Nano), fi.ModTime().Format(time.RFC3339Nano)) - lastModif = fi.ModTime() - trySubmit(res) + if trySubmit(res) { + lastModif = fi.ModTime() + } } case <-ctx.Done(): close(res)