Timeout on flush in testing (#16864)

* Timeout on flush in testing

At the end of each test the queues are flushed. At present there is no limit on the
length of time a flush can take which can lead to long flushes.

However, if the CI task is cancelled we lose the log information as to where the long
flush was taking place.

This PR simply adds a default time limit of 2 minutes - at which point an error will
be produced. This should allow us to more easily find the culprit.

Signed-off-by: Andrew Thornton <art27@cantab.net>
* return better error

Signed-off-by: Andrew Thornton <art27@cantab.net>
Co-authored-by: 6543 <6543@obermui.de>

diff --git a/integrations/testlogger.go b/integrations/testlogger.go
index 9498ad655b33166479b108966a5edeee9b10d15a..ff408b314c868233be8e3dae8c4251af0f5b2580 100644 (file)
--- a/integrations/testlogger.go
+++ b/integrations/testlogger.go
@@ -121,7 +121,7 @@ func PrintCurrentTest(t testing.TB, skip ...int) func() {
                                fmt.Fprintf(os.Stdout, "+++ %s ... still flushing after %v ...\n", t.Name(), slowFlush)
                        }
                })
-               if err := queue.GetManager().FlushAll(context.Background(), -1); err != nil {
+               if err := queue.GetManager().FlushAll(context.Background(), 2*time.Minute); err != nil {
                        t.Errorf("Flushing queues failed with error %v", err)
                }
                timer.Stop()

diff --git a/modules/queue/manager.go b/modules/queue/manager.go
index a88933191aabfc64c6b036b6947a4ca4a490dd64..23e96155a91dc792e23b7bb39cb5708a4acac3fd 100644 (file)
--- a/modules/queue/manager.go
+++ b/modules/queue/manager.go
@@ -9,6 +9,7 @@ import (
        "fmt"
        "reflect"
        "sort"
+       "strings"
        "sync"
        "time"
 
@@ -169,7 +170,17 @@ func (m *Manager) FlushAll(baseCtx context.Context, timeout time.Duration) error
        for {
                select {
                case <-ctx.Done():
-                       return ctx.Err()
+                       mqs := m.ManagedQueues()
+                       nonEmptyQueues := []string{}
+                       for _, mq := range mqs {
+                               if !mq.IsEmpty() {
+                                       nonEmptyQueues = append(nonEmptyQueues, mq.Name)
+                               }
+                       }
+                       if len(nonEmptyQueues) > 0 {
+                               return fmt.Errorf("flush timeout with non-empty queues: %s", strings.Join(nonEmptyQueues, ", "))
+                       }
+                       return nil
                default:
                }
                mqs := m.ManagedQueues()