Fix race condition on service stop

Jenkinsfile

@@ -1,2 +1,3 @@
-buildDebGolangWbgo defaultTargets: 'trixie-armhf trixie-arm64',
+buildDebGolangWbgo defaultTargets: 'current-armhf current-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.44.1) stable; urgency=medium
+
+  * Fix race condition on service stop
+
+ -- Nikolay Korotkiy <nikolay.korotkiy@wirenboard.com>  Wed, 01 Jul 2026 00:30:00 +0400
+
 wb-rules (2.44.0) stable; urgency=medium
 
   * Notify: sendEmail/sendSMS/sendWebhook/sendTelegramMessage accept an optional

wbrules/engine.go

@@ -625,7 +625,7 @@ type RuleEngine struct {
 	active          uint32 // atomic
 	cleanup         *ScopedCleanup
 	rev             uint32 // atomic
-	syncQueueActive bool
+	syncQueueActive uint32 // atomic
 	syncQueue       chan func()
 	syncQuitCh      chan chan struct{}
 	mqttClient      wbgong.MQTTClient // for service
@@ -687,7 +687,7 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn
 		cleanup:               MakeScopedCleanup(),
 		rev:                   0,
 		syncQueue:             make(chan func(), SYNC_QUEUE_LEN),
-		syncQueueActive:       true,
+		syncQueueActive:       ATOMIC_TRUE,
 		syncQuitCh:            make(chan chan struct{}, 1),
 		mqttClient:            mqtt,
 		driver:                driver,
@@ -982,7 +982,17 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) {
 	engine.eventBuffer.PushEvent(cce)
 }
 
-func (engine *RuleEngine) CallSync(thunk func()) {
+// callSync hands thunk to the sync loop and reports whether it was accepted.
+// The active-check and the send are done under statusMtx, which handleStop also
+// holds when it closes syncQueue — so a late caller (notably a GC finalizer via
+// MaybeCallSync) can never send on the closed channel. It returns false when the
+// engine is stopping/stopped and the queue is (being) closed.
+func (engine *RuleEngine) callSync(thunk func()) bool {
+	engine.statusMtx.Lock()
+	defer engine.statusMtx.Unlock()
+	if atomic.LoadUint32(&engine.syncQueueActive) != ATOMIC_TRUE {
+		return false
+	}
 	if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE {
 		delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT)
 		select {
@@ -996,12 +1006,19 @@ func (engine *RuleEngine) CallSync(thunk func()) {
 	} else {
 		engine.syncQueue <- thunk
 	}
+	return true
+}
+
+func (engine *RuleEngine) CallSync(thunk func()) {
+	if !engine.callSync(thunk) {
+		panic("[engine] CallSync called while sync loop is stopped")
+	}
 }
 
 func (engine *RuleEngine) MaybeCallSync(thunk func()) {
-	if engine.syncQueueActive {
-		engine.CallSync(thunk)
-	} else {
+	// If the sync loop isn't running (engine already stopped),
+	// callSync returns false and we run the thunk inline, as before.
+	if !engine.callSync(thunk) {
 		thunk()
 	}
 }
@@ -1296,7 +1313,7 @@ func (engine *RuleEngine) handleStop() {
 	engine.statusMtx.Lock()
 	engine.readyCh = nil
 	engine.driverReadyCh = nil
-	engine.syncQueueActive = false
+	atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE)
 	close(engine.syncQueue)
 	engine.statusMtx.Unlock()
 }
@@ -1345,9 +1362,22 @@ func (engine *RuleEngine) Start() {
 
 	engine.driver.OnDriverEvent(engine.driverEventHandler)
 	engine.driver.OnRetainReady(func(tx wbgong.DriverTx) {
-		engine.driverReadyCh <- struct{}{}
+		// Read the channel under statusMtx: handleStop() nils it on shutdown,
+		// and this callback runs on a driver goroutine. Skip if already stopped
+		// and send non-blocking (the buffer of 1 already carries the signal) so
+		// a late retain-ready can never block this goroutine forever.
+		engine.statusMtx.Lock()
+		ch := engine.driverReadyCh
+		engine.statusMtx.Unlock()
+		if ch == nil {
+			return
+		}
+		select {
+		case ch <- struct{}{}:
+		default:
+		}
 	})
-	engine.syncQueueActive = true
+	atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_TRUE)
 	atomic.StoreUint32(&engine.active, ENGINE_ACTIVE)
 
 	go engine.mainLoop()

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

@@ -21,6 +21,16 @@ const (
 	EXTRA_CTRL_CHANGE_WAIT_TIME_MS = 50
 )
 
+var initialWd string
+
+func init() {
+	var err error
+	initialWd, err = os.Getwd()
+	if err != nil {
+		panic(fmt.Sprintf("os.Getwd(): %v", err))
+	}
+}
+
 type fakeCron struct {
 	t       *testing.T
 	started bool
@@ -95,7 +105,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")
 	}
@@ -254,9 +264,7 @@ func (s *RuleSuiteBase) SetupTest(waitForRetained bool, ruleFiles ...string) {
 
 	engineOptions := NewESEngineOptions()
 	engineOptions.SetPersistentDBFile(s.PersistentDBFile)
-	currentDir, err := os.Getwd()
-	s.Ck("os.Getwd()", err)
-	defaultModulesPath := filepath.Join(currentDir, "..", "modules")
+	defaultModulesPath := filepath.Join(initialWd, "..", "modules")
 	moduleDirs := append(strings.Split(s.ModulesPath, ":"), defaultModulesPath)
 	engineOptions.SetModulesDirs(moduleDirs)
 	s.logClient = s.Broker.MakeClient("wbrules-log")

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")
 	}