From d2221ec9219ebef3b2b8bf204844d5cdd3a98889 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 13 May 2026 21:14:45 +0400 Subject: [PATCH 01/19] Fix race condition on service stop --- Jenkinsfile | 1 + Makefile | 2 +- debian/changelog | 6 ++++++ wbrules/engine.go | 29 +++++++++++++++++++++++------ 4 files changed, 31 insertions(+), 7 deletions(-) diff --git a/Jenkinsfile b/Jenkinsfile index fcae3e1d..3535b506 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -1,2 +1,3 @@ buildDebGolangWbgo defaultTargets: 'bullseye-armhf bullseye-arm64', + defaultWbGoSoBranch: 'feature/amd64-build-race', defaultRunLintian: true diff --git a/Makefile b/Makefile index 99bdab29..b835a446 100644 --- a/Makefile +++ b/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 diff --git a/debian/changelog b/debian/changelog index 0f3a6903..c2e97c5d 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +wb-rules (2.40.1) stable; urgency=medium + + * Fix race condition on service stop + + -- Nikolay Korotkiy Wed, 13 May 2026 17:00:00 +0400 + wb-rules (2.40.0) stable; urgency=medium * Move notify/alarms to modules diff --git a/wbrules/engine.go b/wbrules/engine.go index ab57c26f..f11a1f92 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -628,6 +628,7 @@ type RuleEngine struct { syncQueueActive bool syncQueue chan func() syncQuitCh chan chan struct{} + mainLoopDone chan struct{} mqttClient wbgong.MQTTClient // for service driver wbgong.Driver driverReadyCh chan struct{} @@ -689,6 +690,7 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn syncQueue: make(chan func(), SYNC_QUEUE_LEN), syncQueueActive: true, syncQuitCh: make(chan chan struct{}, 1), + mainLoopDone: nil, mqttClient: mqtt, driver: driver, driverReadyCh: nil, @@ -999,7 +1001,11 @@ func (engine *RuleEngine) CallSync(thunk func()) { } func (engine *RuleEngine) MaybeCallSync(thunk func()) { - if engine.syncQueueActive { + engine.statusMtx.Lock() + syncQueueActive := engine.syncQueueActive + engine.statusMtx.Unlock() + + if syncQueueActive { engine.CallSync(thunk) } else { thunk() @@ -1297,7 +1303,6 @@ func (engine *RuleEngine) handleStop() { engine.readyCh = nil engine.driverReadyCh = nil engine.syncQueueActive = false - close(engine.syncQueue) engine.statusMtx.Unlock() } @@ -1339,18 +1344,25 @@ func (engine *RuleEngine) updateDebugEnabled() { } func (engine *RuleEngine) Start() { + engine.statusMtx.Lock() engine.readyCh = make(chan struct{}) engine.driverReadyCh = make(chan struct{}, 1) + engine.syncQueueActive = true + engine.mainLoopDone = make(chan struct{}) + engine.statusMtx.Unlock() + engine.eventBuffer = NewEventBuffer() engine.driver.OnDriverEvent(engine.driverEventHandler) engine.driver.OnRetainReady(func(tx wbgong.DriverTx) { engine.driverReadyCh <- struct{}{} }) - engine.syncQueueActive = true atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) - go engine.mainLoop() + go func() { + defer close(engine.mainLoopDone) + engine.mainLoop() + }() go engine.syncLoop() } @@ -1370,8 +1382,13 @@ func (engine *RuleEngine) Stop() { engine.syncQuitCh <- q <-q - // wait for main loop to release sync queue - <-engine.syncQueue + // wait for main loop shutdown sequence to finish + engine.statusMtx.Lock() + mainLoopDone := engine.mainLoopDone + engine.statusMtx.Unlock() + if mainLoopDone != nil { + <-mainLoopDone + } } func (engine *RuleEngine) IsActive() bool { From b31535d540f848a08a3c8208b22fc88b03837d10 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 13 May 2026 23:45:15 +0400 Subject: [PATCH 02/19] fix copilot suggestion --- wbrules/engine.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index f11a1f92..2af4ac1c 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -1377,11 +1377,6 @@ func (engine *RuleEngine) Stop() { engine.eventBuffer.Close() - // stop sync loop - q := make(chan struct{}) - engine.syncQuitCh <- q - <-q - // wait for main loop shutdown sequence to finish engine.statusMtx.Lock() mainLoopDone := engine.mainLoopDone @@ -1389,6 +1384,11 @@ func (engine *RuleEngine) Stop() { if mainLoopDone != nil { <-mainLoopDone } + + // stop sync loop + q := make(chan struct{}) + engine.syncQuitCh <- q + <-q } func (engine *RuleEngine) IsActive() bool { From d278f70d2259921ed62c4ea59e9d99e9f323d3ce Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Thu, 14 May 2026 00:19:17 +0400 Subject: [PATCH 03/19] fix copilot suggestion --- wbrules/engine.go | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/wbrules/engine.go b/wbrules/engine.go index 2af4ac1c..58f43716 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -985,6 +985,14 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { + engine.statusMtx.Lock() + syncQueueActive := engine.syncQueueActive + engine.statusMtx.Unlock() + + if !syncQueueActive { + return + } + if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { @@ -1287,6 +1295,11 @@ func (engine *RuleEngine) setupCron() { func (engine *RuleEngine) handleStop() { wbgong.Debug.Printf("engine stopped") + if engine.cron != nil { + engine.cron.Stop() + engine.cron = nil + } + engine.timersMutex.Lock() timerEntries := make([]*TimerEntry, 0, len(engine.timers)) for _, entry := range engine.timers { From 37c110f984f9d051f735c7c1c10a860854321241 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Thu, 14 May 2026 00:59:08 +0400 Subject: [PATCH 04/19] fix copilot suggestion --- wbrules/engine.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 58f43716..6337aee7 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -990,6 +990,7 @@ func (engine *RuleEngine) CallSync(thunk func()) { engine.statusMtx.Unlock() if !syncQueueActive { + thunk() return } @@ -1366,11 +1367,19 @@ func (engine *RuleEngine) Start() { engine.eventBuffer = NewEventBuffer() + atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) + engine.driver.OnDriverEvent(engine.driverEventHandler) + driverReadyCh := engine.driverReadyCh engine.driver.OnRetainReady(func(tx wbgong.DriverTx) { - engine.driverReadyCh <- struct{}{} + if driverReadyCh != nil { + select { + case driverReadyCh <- struct{}{}: + default: + // Channel already has signal or is full; don't block + } + } }) - atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) go func() { defer close(engine.mainLoopDone) From 3ed5e178e5adc34e5b013512caf0cd2188ce5f98 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Thu, 14 May 2026 00:00:27 +0300 Subject: [PATCH 05/19] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- wbrules/engine.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 6337aee7..bb5c8fa6 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -1314,8 +1314,14 @@ func (engine *RuleEngine) handleStop() { } engine.statusMtx.Lock() - engine.readyCh = nil - engine.driverReadyCh = nil + if engine.readyCh != nil { + close(engine.readyCh) + engine.readyCh = nil + } + if engine.driverReadyCh != nil { + close(engine.driverReadyCh) + engine.driverReadyCh = nil + } engine.syncQueueActive = false engine.statusMtx.Unlock() } From 5a9c7a679c78dd84b21dc8c0659d3e8c7ccd80dc Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Fri, 15 May 2026 16:08:03 +0400 Subject: [PATCH 06/19] tmp revert --- wbrules/engine.go | 67 ++++++++--------------------------------------- 1 file changed, 11 insertions(+), 56 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index bb5c8fa6..ab57c26f 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -628,7 +628,6 @@ type RuleEngine struct { syncQueueActive bool syncQueue chan func() syncQuitCh chan chan struct{} - mainLoopDone chan struct{} mqttClient wbgong.MQTTClient // for service driver wbgong.Driver driverReadyCh chan struct{} @@ -690,7 +689,6 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn syncQueue: make(chan func(), SYNC_QUEUE_LEN), syncQueueActive: true, syncQuitCh: make(chan chan struct{}, 1), - mainLoopDone: nil, mqttClient: mqtt, driver: driver, driverReadyCh: nil, @@ -985,15 +983,6 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { - engine.statusMtx.Lock() - syncQueueActive := engine.syncQueueActive - engine.statusMtx.Unlock() - - if !syncQueueActive { - thunk() - return - } - if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { @@ -1010,11 +999,7 @@ func (engine *RuleEngine) CallSync(thunk func()) { } func (engine *RuleEngine) MaybeCallSync(thunk func()) { - engine.statusMtx.Lock() - syncQueueActive := engine.syncQueueActive - engine.statusMtx.Unlock() - - if syncQueueActive { + if engine.syncQueueActive { engine.CallSync(thunk) } else { thunk() @@ -1296,11 +1281,6 @@ func (engine *RuleEngine) setupCron() { func (engine *RuleEngine) handleStop() { wbgong.Debug.Printf("engine stopped") - if engine.cron != nil { - engine.cron.Stop() - engine.cron = nil - } - engine.timersMutex.Lock() timerEntries := make([]*TimerEntry, 0, len(engine.timers)) for _, entry := range engine.timers { @@ -1314,15 +1294,10 @@ func (engine *RuleEngine) handleStop() { } engine.statusMtx.Lock() - if engine.readyCh != nil { - close(engine.readyCh) - engine.readyCh = nil - } - if engine.driverReadyCh != nil { - close(engine.driverReadyCh) - engine.driverReadyCh = nil - } + engine.readyCh = nil + engine.driverReadyCh = nil engine.syncQueueActive = false + close(engine.syncQueue) engine.statusMtx.Unlock() } @@ -1364,33 +1339,18 @@ func (engine *RuleEngine) updateDebugEnabled() { } func (engine *RuleEngine) Start() { - engine.statusMtx.Lock() engine.readyCh = make(chan struct{}) engine.driverReadyCh = make(chan struct{}, 1) - engine.syncQueueActive = true - engine.mainLoopDone = make(chan struct{}) - engine.statusMtx.Unlock() - engine.eventBuffer = NewEventBuffer() - atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) - engine.driver.OnDriverEvent(engine.driverEventHandler) - driverReadyCh := engine.driverReadyCh engine.driver.OnRetainReady(func(tx wbgong.DriverTx) { - if driverReadyCh != nil { - select { - case driverReadyCh <- struct{}{}: - default: - // Channel already has signal or is full; don't block - } - } + engine.driverReadyCh <- struct{}{} }) + engine.syncQueueActive = true + atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) - go func() { - defer close(engine.mainLoopDone) - engine.mainLoop() - }() + go engine.mainLoop() go engine.syncLoop() } @@ -1405,18 +1365,13 @@ func (engine *RuleEngine) Stop() { engine.eventBuffer.Close() - // wait for main loop shutdown sequence to finish - engine.statusMtx.Lock() - mainLoopDone := engine.mainLoopDone - engine.statusMtx.Unlock() - if mainLoopDone != nil { - <-mainLoopDone - } - // stop sync loop q := make(chan struct{}) engine.syncQuitCh <- q <-q + + // wait for main loop to release sync queue + <-engine.syncQueue } func (engine *RuleEngine) IsActive() bool { From 48792cea9ed7e451a3cb8ba788ea44d15f1c139b Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 11:27:14 +0400 Subject: [PATCH 07/19] atomic syncQueueActive --- wbrules/engine.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index ab57c26f..0fdfdb2c 100644 --- a/wbrules/engine.go +++ b/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, @@ -999,7 +999,7 @@ func (engine *RuleEngine) CallSync(thunk func()) { } func (engine *RuleEngine) MaybeCallSync(thunk func()) { - if engine.syncQueueActive { + if atomic.LoadUint32(&engine.syncQueueActive) == ATOMIC_TRUE { engine.CallSync(thunk) } else { thunk() @@ -1296,7 +1296,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() } @@ -1347,7 +1347,7 @@ func (engine *RuleEngine) Start() { 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() From 359828fabb2b6f4a1284cc66adca05cf13c25323 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 12:21:34 +0400 Subject: [PATCH 08/19] add syncQueueMtx --- wbrules/engine.go | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 0fdfdb2c..7e882578 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -626,6 +626,7 @@ type RuleEngine struct { cleanup *ScopedCleanup rev uint32 // atomic syncQueueActive uint32 // atomic + syncQueueMtx sync.RWMutex syncQueue chan func() syncQuitCh chan chan struct{} mqttClient wbgong.MQTTClient // for service @@ -983,6 +984,14 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { + engine.syncQueueMtx.RLock() + defer engine.syncQueueMtx.RUnlock() + + if atomic.LoadUint32(&engine.syncQueueActive) != ATOMIC_TRUE { + thunk() + return + } + if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { @@ -1296,9 +1305,12 @@ func (engine *RuleEngine) handleStop() { engine.statusMtx.Lock() engine.readyCh = nil engine.driverReadyCh = nil + engine.statusMtx.Unlock() + atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE) + engine.syncQueueMtx.Lock() close(engine.syncQueue) - engine.statusMtx.Unlock() + engine.syncQueueMtx.Unlock() } func (engine *RuleEngine) isDebugControl(ctrlSpec ControlSpec) bool { From 12aaf6443b40c11a1ee04519aa3741606266d21e Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 13:27:44 +0400 Subject: [PATCH 09/19] Revert "add syncQueueMtx" This reverts commit 359828fabb2b6f4a1284cc66adca05cf13c25323. --- wbrules/engine.go | 14 +------------- 1 file changed, 1 insertion(+), 13 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 7e882578..0fdfdb2c 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -626,7 +626,6 @@ type RuleEngine struct { cleanup *ScopedCleanup rev uint32 // atomic syncQueueActive uint32 // atomic - syncQueueMtx sync.RWMutex syncQueue chan func() syncQuitCh chan chan struct{} mqttClient wbgong.MQTTClient // for service @@ -984,14 +983,6 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { - engine.syncQueueMtx.RLock() - defer engine.syncQueueMtx.RUnlock() - - if atomic.LoadUint32(&engine.syncQueueActive) != ATOMIC_TRUE { - thunk() - return - } - if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { @@ -1305,12 +1296,9 @@ func (engine *RuleEngine) handleStop() { engine.statusMtx.Lock() engine.readyCh = nil engine.driverReadyCh = nil - engine.statusMtx.Unlock() - atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE) - engine.syncQueueMtx.Lock() close(engine.syncQueue) - engine.syncQueueMtx.Unlock() + engine.statusMtx.Unlock() } func (engine *RuleEngine) isDebugControl(ctrlSpec ControlSpec) bool { From 13bd6d4ca8a969bb42b0f4ec059431af0a097348 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 13:35:58 +0400 Subject: [PATCH 10/19] add syncStopCh --- wbrules/engine.go | 31 +++++++++++++++++++++++++++---- 1 file changed, 27 insertions(+), 4 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 0fdfdb2c..6a6c3d2e 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log" + "runtime" "sort" "strconv" "sync" @@ -627,6 +628,8 @@ type RuleEngine struct { rev uint32 // atomic 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,6 +690,7 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn cleanup: MakeScopedCleanup(), rev: 0, syncQueue: make(chan func(), SYNC_QUEUE_LEN), + syncStopCh: make(chan struct{}), syncQueueActive: ATOMIC_TRUE, syncQuitCh: make(chan chan struct{}, 1), mqttClient: mqtt, @@ -983,6 +987,11 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { + if atomic.LoadUint32(&engine.syncQueueActive) != ATOMIC_TRUE { + thunk() + return + } + if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { @@ -990,11 +999,20 @@ 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() + } } } @@ -1297,7 +1315,9 @@ func (engine *RuleEngine) handleStop() { engine.readyCh = nil engine.driverReadyCh = nil atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE) - close(engine.syncQueue) + engine.syncStopOnce.Do(func() { + close(engine.syncStopCh) + }) engine.statusMtx.Unlock() } @@ -1342,6 +1362,8 @@ 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) { @@ -1370,8 +1392,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 { From b848fdd2355868acce94f53626c90afe8c52e440 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 13:39:57 +0400 Subject: [PATCH 11/19] fixup --- wbrules/engine.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 6a6c3d2e..11d92baa 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -987,11 +987,6 @@ func (engine *RuleEngine) driverEventHandler(event wbgong.DriverEvent) { } func (engine *RuleEngine) CallSync(thunk func()) { - if atomic.LoadUint32(&engine.syncQueueActive) != ATOMIC_TRUE { - thunk() - return - } - if atomic.LoadUint32(&engine.debugEnabled) == ATOMIC_TRUE { delay := time.NewTimer(ENGINE_CALLSYNC_TIMEOUT) select { From f40f0de058ba17bc251ea227da927e45b4319ca7 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Mon, 18 May 2026 14:04:31 +0400 Subject: [PATCH 12/19] fix tests --- wbrules/persistent_storage_test.go | 2 +- wbrules/rule_reload_test.go | 4 ++-- wbrules/rule_test.go | 2 +- wbrules/vcells_storage_test.go | 2 +- 4 files changed, 5 insertions(+), 5 deletions(-) diff --git a/wbrules/persistent_storage_test.go b/wbrules/persistent_storage_test.go index 9360451c..2e3dcfa7 100644 --- a/wbrules/persistent_storage_test.go +++ b/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") } diff --git a/wbrules/rule_reload_test.go b/wbrules/rule_reload_test.go index b536e9e6..042223c6 100644 --- a/wbrules/rule_reload_test.go +++ b/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") } diff --git a/wbrules/rule_test.go b/wbrules/rule_test.go index 49dd27a2..5c2fd6e1 100644 --- a/wbrules/rule_test.go +++ b/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") } diff --git a/wbrules/vcells_storage_test.go b/wbrules/vcells_storage_test.go index 498bd1e0..4a80ff65 100644 --- a/wbrules/vcells_storage_test.go +++ b/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") } From 454c8f99a2dc6dbcec65692e3ce316edaad1aa58 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 03:10:43 +0400 Subject: [PATCH 13/19] revert --- wbrules/engine.go | 26 ++++---------------------- 1 file changed, 4 insertions(+), 22 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index ee987a04..cdd89a50 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "log" - "runtime" "sort" "strconv" "sync" @@ -628,8 +627,6 @@ type RuleEngine struct { rev uint32 // atomic syncQueueActive uint32 // atomic syncQueue chan func() - syncStopCh chan struct{} - syncStopOnce sync.Once syncQuitCh chan chan struct{} mqttClient wbgong.MQTTClient // for service driver wbgong.Driver @@ -690,7 +687,6 @@ func NewRuleEngine(driver wbgong.Driver, mqtt wbgong.MQTTClient, options *RuleEn cleanup: MakeScopedCleanup(), rev: 0, syncQueue: make(chan func(), SYNC_QUEUE_LEN), - syncStopCh: make(chan struct{}), syncQueueActive: ATOMIC_TRUE, syncQuitCh: make(chan chan struct{}, 1), mqttClient: mqtt, @@ -994,20 +990,11 @@ 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 { - select { - case engine.syncQueue <- thunk: - case <-engine.syncStopCh: - thunk() - } + engine.syncQueue <- thunk } } @@ -1310,9 +1297,7 @@ func (engine *RuleEngine) handleStop() { engine.readyCh = nil engine.driverReadyCh = nil atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_FALSE) - engine.syncStopOnce.Do(func() { - close(engine.syncStopCh) - }) + close(engine.syncQueue) engine.statusMtx.Unlock() } @@ -1357,8 +1342,6 @@ 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) { @@ -1387,9 +1370,8 @@ func (engine *RuleEngine) Stop() { engine.syncQuitCh <- q <-q - for atomic.LoadUint32(&engine.syncQueueActive) == ATOMIC_TRUE { - runtime.Gosched() - } + // wait for main loop to release sync queue + <-engine.syncQueue } func (engine *RuleEngine) IsActive() bool { From fb14d75b0210aace7086d8abd9afd2ec35b01416 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 04:16:25 +0400 Subject: [PATCH 14/19] fixup --- wbrules/rule_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/wbrules/rule_test.go b/wbrules/rule_test.go index 5c2fd6e1..dd8b810e 100644 --- a/wbrules/rule_test.go +++ b/wbrules/rule_test.go @@ -21,6 +21,8 @@ const ( EXTRA_CTRL_CHANGE_WAIT_TIME_MS = 50 ) +var initialWd, _ = os.Getwd() + type fakeCron struct { t *testing.T started bool @@ -254,9 +256,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") From 31fb0c409028306466b087988ff41401c5cd9b75 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 11:15:35 +0400 Subject: [PATCH 15/19] fixup --- wbrules/engine.go | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index cdd89a50..ce3c06ac 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -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,17 @@ func (engine *RuleEngine) CallSync(thunk func()) { } else { engine.syncQueue <- thunk } + return true +} + +func (engine *RuleEngine) CallSync(thunk func()) { + engine.callSync(thunk) } func (engine *RuleEngine) MaybeCallSync(thunk func()) { - if atomic.LoadUint32(&engine.syncQueueActive) == ATOMIC_TRUE { - engine.CallSync(thunk) - } else { + // If the sync loop isn't running (engine not started yet or already + // stopped), callSync returns false and we run the thunk inline, as before. + if !engine.callSync(thunk) { thunk() } } From e6265f4ed250657dc35a5574e129a26dd244e644 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 10:56:21 +0300 Subject: [PATCH 16/19] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- wbrules/rule_test.go | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/wbrules/rule_test.go b/wbrules/rule_test.go index dd8b810e..e65c0268 100644 --- a/wbrules/rule_test.go +++ b/wbrules/rule_test.go @@ -21,7 +21,15 @@ const ( EXTRA_CTRL_CHANGE_WAIT_TIME_MS = 50 ) -var initialWd, _ = os.Getwd() +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 From a3641b9b8ec5495f464cba8ef727873c3aa5baa8 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 17:34:05 +0400 Subject: [PATCH 17/19] fix copilot suggestion --- wbrules/engine.go | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index ce3c06ac..7ab1be8a 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -1360,7 +1360,20 @@ 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: + } }) atomic.StoreUint32(&engine.syncQueueActive, ATOMIC_TRUE) atomic.StoreUint32(&engine.active, ENGINE_ACTIVE) From 52f18a178bde90d6323cfdc1817709c845ca4cd8 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 16:42:25 +0300 Subject: [PATCH 18/19] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- wbrules/engine.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index 7ab1be8a..e3234915 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -1010,7 +1010,9 @@ func (engine *RuleEngine) callSync(thunk func()) bool { } func (engine *RuleEngine) CallSync(thunk func()) { - engine.callSync(thunk) + if !engine.callSync(thunk) { + panic("[engine] CallSync called while sync loop is stopped") + } } func (engine *RuleEngine) MaybeCallSync(thunk func()) { From da84abbd4b931fff9f5e1acde1897030d4666ff6 Mon Sep 17 00:00:00 2001 From: Nikolay Korotkiy Date: Wed, 1 Jul 2026 17:51:56 +0400 Subject: [PATCH 19/19] fix copilot suggestion --- wbrules/engine.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/wbrules/engine.go b/wbrules/engine.go index e3234915..1b5d7603 100644 --- a/wbrules/engine.go +++ b/wbrules/engine.go @@ -1016,8 +1016,8 @@ func (engine *RuleEngine) CallSync(thunk func()) { } func (engine *RuleEngine) MaybeCallSync(thunk func()) { - // If the sync loop isn't running (engine not started yet or already - // stopped), callSync returns false and we run the thunk inline, as before. + // 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() }