Fix race condition on service stop

Jenkinsfile

@@ -1,2 +1,3 @@
 buildDebGolangWbgo defaultTargets: 'bullseye-armhf bullseye-arm64',
+                   defaultWbGoSoBranch: 'feature/amd64-build-race',
                    defaultRunLintian: true

Makefile

@@ -19,7 +19,7 @@ GOTEST ?= $(GO) test
 GCFLAGS :=
 LDFLAGS := -X main.version=`git describe --tags --always --dirty`
 GO_FLAGS := -buildvcs=false
-GO_TEST_FLAGS := -v -cover
+GO_TEST_FLAGS := -v -cover -race
 
 ifeq ($(DEBUG),)
 	LDFLAGS += -s -w

debian/changelog

@@ -1,3 +1,9 @@
+wb-rules (2.40.1) stable; urgency=medium
+
+  * Fix race condition on service stop
+
+ -- Nikolay Korotkiy <nikolay.korotkiy@wirenboard.com>  Wed, 13 May 2026 17:00:00 +0400
+
 wb-rules (2.40.0) stable; urgency=medium
 
   * Move notify/alarms to modules

wbrules/engine.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"runtime"
 	"sort"
 	"strconv"
 	"sync"
@@ -625,8 +626,10 @@ type RuleEngine struct {
 	active          uint32 // atomic
 	cleanup         *ScopedCleanup
 	rev             uint32 // atomic
-	syncQueueActive bool
+	syncQueueActive uint32 // atomic
 	syncQueue       chan func()
+	syncStopCh      chan struct{}
+	syncStopOnce    sync.Once
 	syncQuitCh      chan chan struct{}
 	mqttClient      wbgong.MQTTClient // for service
 	driver          wbgong.Driver
@@ -687,7 +690,8 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn
 		cleanup:               MakeScopedCleanup(),
 		rev:                   0,
 		syncQueue:             make(chan func(), SYNC_QUEUE_LEN),
-		syncQueueActive:       true,
+		syncStopCh:            make(chan struct{}),
+		syncQueueActive:       ATOMIC_TRUE,
 		syncQuitCh:            make(chan chan struct{}, 1),
 		mqttClient:            mqtt,
 		driver:                driver,
@@ -990,16 +994,25 @@ func (engine *RuleEngine) CallSync(thunk func()) {
 			if !delay.Stop() {
 				<-delay.C
 			}
+		case <-engine.syncStopCh:
+			if !delay.Stop() {
+				<-delay.C
+			}
+			thunk()
 		case <-delay.C:
 			panic("[engine] CallSync stuck!")
 		}
 	} else {
-		engine.syncQueue <- thunk
+		select {
+		case engine.syncQueue <- thunk:
+		case <-engine.syncStopCh:
+			thunk()
+		}
 	}
 }
 
 func (engine *RuleEngine) MaybeCallSync(thunk func()) {
-	if engine.syncQueueActive {
+	if atomic.LoadUint32(&engine.syncQueueActive) == ATOMIC_TRUE {
 		engine.CallSync(thunk)
 	} else {
 		thunk()
@@ -1296,8 +1309,10 @@ func (engine *RuleEngine) handleStop() {
 	engine.statusMtx.Lock()
 	engine.readyCh = nil
 	engine.driverReadyCh = nil
-	engine.syncQueueActive = false
-	close(engine.syncQueue)
+	atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE)
+	engine.syncStopOnce.Do(func() {
+		close(engine.syncStopCh)
+	})
 	engine.statusMtx.Unlock()
 }
 
@@ -1342,12 +1357,14 @@ func (engine *RuleEngine) Start() {
 	engine.readyCh = make(chan struct{})
 	engine.driverReadyCh = make(chan struct{}, 1)
 	engine.eventBuffer = NewEventBuffer()
+	engine.syncStopCh = make(chan struct{})
+	engine.syncStopOnce = sync.Once{}
 
 	engine.driver.OnDriverEvent(engine.driverEventHandler)
 	engine.driver.OnRetainReady(func(tx wbgong.DriverTx) {
 		engine.driverReadyCh <- struct{}{}
 	})
-	engine.syncQueueActive = true
+	atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_TRUE)
 	atomic.StoreUint32(&engine.active, ENGINE_ACTIVE)
 
 	go engine.mainLoop()
@@ -1370,8 +1387,9 @@ func (engine *RuleEngine) Stop() {
 	engine.syncQuitCh <- q
 	<-q
 
-	// wait for main loop to release sync queue
-	<-engine.syncQueue
+	for atomic.LoadUint32(&engine.syncQueueActive) == ATOMIC_TRUE {
+		runtime.Gosched()
+	}
 }
 
 func (engine *RuleEngine) IsActive() bool {

wbrules/persistent_storage_test.go

@@ -18,7 +18,7 @@ func (s *PersistentStorageSuite) SetupFixture() {
 
 	// we need to create separated temp directory because persistent DB file
 	// should be keeped between tests
-	s.tmpDir, err = os.MkdirTemp("", "wbrulestest")
+	s.tmpDir, err = os.MkdirTemp("/dev/shm", "wbrulestest")
 	if err != nil {
 		s.FailNow("can't create temp directory")
 	}

wbrules/rule_reload_test.go

@@ -17,7 +17,7 @@ type RuleReloadSuite struct {
 
 func (s *RuleReloadSuite) SetupTest() {
 	var err error
-	s.reloadTmpDir, err = os.MkdirTemp("", "wbrulestest")
+	s.reloadTmpDir, err = os.MkdirTemp("/dev/shm", "wbrulestest")
 	if err != nil {
 		s.FailNow("can't create temp directory")
 	}
@@ -336,7 +336,7 @@ type RuleReloadForceDefaultSuite struct {
 
 func (s *RuleReloadForceDefaultSuite) SetupTest() {
 	var err error
-	s.reloadTmpDir, err = os.MkdirTemp("", "wbrulestest")
+	s.reloadTmpDir, err = os.MkdirTemp("/dev/shm", "wbrulestest")
 	if err != nil {
 		s.FailNow("can't create temp directory")
 	}

wbrules/rule_test.go

@@ -95,7 +95,7 @@ var updatesVerifyRx = regexp.MustCompile(`^\[(changed|removed)\] (.*)`)
 
 // creates necessary file paths if some are not defined already
 func (s *RuleSuiteBase) createTempFiles() {
-	tmpDir, err := os.MkdirTemp("", "wbrulestest")
+	tmpDir, err := os.MkdirTemp("/dev/shm", "wbrulestest")
 	if err != nil {
 		s.FailNow("can't create temp directory")
 	}

wbrules/vcells_storage_test.go

@@ -15,7 +15,7 @@ type VirtualCellsStorageSuite struct {
 
 func (s *VirtualCellsStorageSuite) SetupFixture() {
 	var err error
-	s.tmpDir, err = os.MkdirTemp("", "wbrulestest")
+	s.tmpDir, err = os.MkdirTemp("/dev/shm", "wbrulestest")
 	if err != nil {
 		s.FailNow("can't create temp directory")
 	}