Watchdog: prevent concurrent access (#25232)

plugin/watchdog.go

@@ -69,6 +69,7 @@ func (o *watchdogPlugin) wdt(ctx context.Context, set func() error) {
 func setter[T comparable](o *watchdogPlugin, set func(T) error, reset []T) func(T) error {
 	return func(val T) error {
 		o.mu.Lock()
+		defer o.mu.Unlock()
 
 		// stop wdt on new write
 		if o.cancel != nil {
@@ -82,12 +83,13 @@ func setter[T comparable](o *watchdogPlugin, set func(T) error, reset []T) func(
 			ctx, o.cancel = context.WithCancel(context.Background())
 
 			go o.wdt(ctx, func() error {
+				o.mu.Lock()
+				defer o.mu.Unlock()
+
 				return set(val)
 			})
 		}
 
-		o.mu.Unlock()
-
 		return set(val)
 	}
 }

plugin/watchdog_test.go

@@ -0,0 +1,46 @@
+package plugin
+
+import (
+	"errors"
+	"math/rand/v2"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/evcc-io/evcc/util"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+func TestWatchdogSetterConcurrency(t *testing.T) {
+	p := &watchdogPlugin{
+		log:     util.NewLogger("foo"),
+		timeout: 10 * time.Nanosecond,
+	}
+
+	var u atomic.Uint32
+
+	set := setter(p, func(i int) error {
+		if !u.CompareAndSwap(0, 1) {
+			return errors.New("race")
+		}
+
+		time.Sleep(time.Duration(rand.Int32N(int32(p.timeout))))
+
+		if !u.CompareAndSwap(1, 0) {
+			return errors.New("race")
+		}
+
+		return nil
+	}, nil)
+
+	var eg errgroup.Group
+
+	for range 100 {
+		eg.Go(func() error {
+			return set(1)
+		})
+	}
+
+	require.NoError(t, eg.Wait())
+}