Fix/2402/retry events synchronously at startup (#2414)

* notify subscribers synchronously on startup
* fix race condition in test, jitter now equals initialDelay. Reduced max retry count to 20 (eliminates overflow)
* remove initial delay from retry
* no retry when attempts <= 0 or >= max

network/dag/notifier.go

@@ -37,7 +37,7 @@ import (
 const (
 	defaultRetryDelay      = time.Second
 	retriesFailedThreshold = 10
-	maxRetries             = 100
+	maxRetries             = 20
 	// TransactionEventType is used as Type in an Event when a transaction is added to the DAG.
 	TransactionEventType = "transaction"
 	// PayloadEventType is used as Type in an Event when a payload is written to the DB.
@@ -239,16 +239,33 @@ func (p *notifier) Run() error {
 	if !p.isPersistent() {
 		return nil
 	}
-	return p.db.ReadShelf(p.ctx, p.shelfName(), func(reader stoabs.Reader) error {
+	// we're going to retry all events synchronously at startup. For the ones that fail we'll start the retry loop
+	failedAtStartup := make([]Event, 0)
+	err := p.db.ReadShelf(p.ctx, p.shelfName(), func(reader stoabs.Reader) error {
 		return reader.Iterate(func(k stoabs.Key, v []byte) error {
 			event := Event{}
 			_ = json.Unmarshal(v, &event)
 
-			p.retry(event)
+			if err := p.notifyNow(event); err != nil {
+				if event.Retries < maxRetries {
+					failedAtStartup = append(failedAtStartup, event)
+				}
+			}
 
 			return nil
 		}, stoabs.BytesKey{})
 	})
+	if err != nil {
+		return err
+	}
+
+	// for all events from failedAtStartup, call retry
+	// this may still produce errors in the logs or even duplicate errors since notifyNow also failed
+	// but rather duplicate errors then errors produced from overloading the DB with transactions
+	for _, event := range failedAtStartup {
+		p.retry(event)
+	}
+	return nil
 }
 
 func (p *notifier) GetFailedEvents() (events []Event, err error) {
@@ -330,21 +347,24 @@ func (p *notifier) Notify(event Event) {
 func (p *notifier) retry(event Event) {
 	delay := p.retryDelay
 	initialCount := event.Retries + 1
+	attempts := maxRetries - uint(initialCount)
+	if attempts <= 0 || attempts >= maxRetries {
+		return
+	}
 
 	for i := 0; i < initialCount; i++ {
 		delay *= 2
 	}
 
 	go func(ctx context.Context) {
-		// also an initial delay
-		time.Sleep(delay)
 		err := retry.Do(func() error {
 			return p.notifyNow(event)
 		},
-			retry.Attempts(maxRetries-uint(initialCount)),
+			retry.Attempts(attempts),
 			retry.MaxDelay(24*time.Hour),
+			retry.MaxJitter(p.retryDelay),
 			retry.Delay(delay),
-			retry.DelayType(retry.BackOffDelay),
+			retry.DelayType(retry.CombineDelay(retry.BackOffDelay, retry.RandomDelay)),
 			retry.Context(ctx),
 			retry.LastErrorOnly(true),
 			retry.OnRetry(func(n uint, err error) {

network/dag/notifier_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/nuts-foundation/go-stoabs"
 	"github.com/nuts-foundation/go-stoabs/bbolt"
 	"github.com/nuts-foundation/nuts-node/crypto/hash"
+	"github.com/nuts-foundation/nuts-node/storage"
 	"github.com/nuts-foundation/nuts-node/test"
 	"github.com/nuts-foundation/nuts-node/test/io"
 	"github.com/prometheus/client_golang/prometheus"
@@ -34,6 +35,8 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 	"path"
+	"runtime"
+	"strings"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -313,13 +316,14 @@ func TestNotifier_Notify(t *testing.T) {
 		timeFunc = func() time.Time {
 			return now
 		}
-		s := NewNotifier(t.Name(), counter.callback, WithPersistency(kvStore)).(*notifier)
+		s := NewNotifier(t.Name(), counter.callback, WithRetryDelay(2*time.Second), WithPersistency(kvStore)).(*notifier)
 		defer s.Close()
 		kvStore.Write(ctx, func(tx stoabs.WriteTx) error {
 			return s.Save(tx, event)
 		})
 
-		s.Notify(event)
+		// we use retry here since Notify will run notifyNow twice asynchronously
+		s.retry(event)
 
 		test.WaitFor(t, func() (bool, error) {
 			return counter.N.Load() == 1, nil
@@ -360,11 +364,7 @@ func TestNotifier_Notify(t *testing.T) {
 }
 
 func TestNotifier_Run(t *testing.T) {
-	ctx := context.Background()
-	filePath := io.TestDirectory(t)
 	transaction, _, _ := CreateTestTransaction(0)
-	kvStore, _ := bbolt.CreateBBoltStore(path.Join(filePath, "test.db"))
-	counter := callbackCounter{}
 	payload := "payload"
 	event := Event{
 		Type:        TransactionEventType,
@@ -373,18 +373,52 @@ func TestNotifier_Run(t *testing.T) {
 		Transaction: transaction,
 		Payload:     []byte(payload),
 	}
-	s := NewNotifier(t.Name(), counter.callbackFinished, WithPersistency(kvStore), WithRetryDelay(time.Millisecond)).(*notifier)
 
-	_ = kvStore.WriteShelf(ctx, s.shelfName(), func(writer stoabs.Writer) error {
-		bytes, _ := json.Marshal(event)
-		return writer.Put(stoabs.BytesKey(event.Hash.Slice()), bytes)
+	t.Run("OK - callback called", func(t *testing.T) {
+		ctx := context.Background()
+		filePath := io.TestDirectory(t)
+		kvStore := storage.CreateTestBBoltStore(t, path.Join(filePath, "test.db"))
+		counter := callbackCounter{}
+		s := NewNotifier(t.Name(), counter.callbackFinished, WithPersistency(kvStore), WithRetryDelay(time.Millisecond)).(*notifier)
+		defer s.Close()
+
+		_ = kvStore.WriteShelf(ctx, s.shelfName(), func(writer stoabs.Writer) error {
+			bytes, _ := json.Marshal(event)
+			return writer.Put(stoabs.BytesKey(event.Hash.Slice()), bytes)
+		})
+
+		err := s.Run()
+		require.NoError(t, err)
+
+		test.WaitFor(t, func() (bool, error) {
+			return counter.N.Load() == 1, nil
+		}, time.Second, "timeout while waiting for receiver")
 	})
 
-	s.Run()
+	t.Run("OK - callback errors", func(t *testing.T) {
+		ctx := context.Background()
+		filePath := io.TestDirectory(t)
+		kvStore := storage.CreateTestBBoltStore(t, path.Join(filePath, "test.db"))
+		counter := callbackCounter{}
+		counter.setCallbackError(errors.New("error"))
+		s := NewNotifier(t.Name(), counter.callbackFailure, WithPersistency(kvStore), WithRetryDelay(time.Millisecond)).(*notifier)
+		defer s.Close()
+
+		_ = kvStore.WriteShelf(ctx, s.shelfName(), func(writer stoabs.Writer) error {
+			bytes, _ := json.Marshal(event)
+			return writer.Put(stoabs.BytesKey(event.Hash.Slice()), bytes)
+		})
+
+		err := s.Run()
+		require.NoError(t, err)
 
-	test.WaitFor(t, func() (bool, error) {
-		return counter.N.Load() == 1, nil
-	}, time.Second, "timeout while waiting for receiver")
+		stack := make([]byte, 4*1024)
+		test.WaitFor(t, func() (bool, error) {
+			runtime.Stack(stack, true)
+			index := strings.Index(string(stack), "dag.(*notifier).retry.func1")
+			return index != -1, nil
+		}, time.Second, "timeout while waiting for go routine to start")
+	})
 }
 
 func TestNotifier_VariousFlows(t *testing.T) {
@@ -447,7 +481,7 @@ func TestNotifier_VariousFlows(t *testing.T) {
 		kvStore, _ := bbolt.CreateBBoltStore(path.Join(filePath, "test.db"))
 		counter := callbackCounter{}
 		notifiedCounter := &prometheusCounter{}
-		event := Event{Hash: hash.EmptyHash(), Transaction: transaction, Retries: 95}
+		event := Event{Hash: hash.EmptyHash(), Transaction: transaction, Retries: maxRetries - 5}
 		s := NewNotifier(t.Name(), counter.callbackFailure, WithPersistency(kvStore), WithRetryDelay(time.Nanosecond), withCounters(notifiedCounter, nil)).(*notifier)
 		defer s.Close()
 
@@ -464,8 +498,8 @@ func TestNotifier_VariousFlows(t *testing.T) {
 				return nil
 			})
 
-			return e.Retries == 100, nil
-		}, time.Second, "timeout while waiting for receiver")
+			return e.Retries == maxRetries, nil
+		}, 2*time.Second, "timeout while waiting for receiver")
 
 		events, err := s.GetFailedEvents()
 
@@ -532,7 +566,7 @@ func TestNotifier_VariousFlows(t *testing.T) {
 				return nil
 			})
 			return e.Retries >= maxRetries, nil
-		}, time.Second, "timeout while waiting for receiver")
+		}, 5*time.Second, "timeout while waiting for receiver")
 
 		events, err := s.GetFailedEvents()