diff --git a/Jenkinsfile b/Jenkinsfile index 267189c..e22b89f 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -1,3 +1,4 @@ buildDebGolangWbgo defaultTargets: 'current-armhf current-arm64', + defaultWbGoSoBranch: 'feature/amd64-build-race', defaultGoVersion: '1.26', defaultRunLintian: true diff --git a/Makefile b/Makefile index 99bdab2..b835a44 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 488a7ea..b5482df 100644 --- a/debian/changelog +++ b/debian/changelog @@ -1,3 +1,9 @@ +wb-rules (2.46.2) stable; urgency=medium + + * Fix race condition on service stop + + -- Nikolay Korotkiy Fri, 07 Aug 2026 18:30:00 +0400 + wb-rules (2.46.1) stable; urgency=medium * Use go 1.26 compiler diff --git a/wbrules/engine.go b/wbrules/engine.go index b79d7bd..a5f1521 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 @@ -697,7 +697,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, @@ -993,7 +993,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 { @@ -1007,12 +1017,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() } } @@ -1313,7 +1330,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() } @@ -1362,9 +1379,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() diff --git a/wbrules/persistent_storage_test.go b/wbrules/persistent_storage_test.go index daa4b25..e35277e 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 b536e9e..042223c 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 6748f74..5df95e0 100644 --- a/wbrules/rule_test.go +++ b/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 fakeCronEntry struct { id cron.EntryID cmd func() @@ -109,7 +119,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") } @@ -268,9 +278,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") diff --git a/wbrules/vcells_storage_test.go b/wbrules/vcells_storage_test.go index 498bd1e..4a80ff6 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") }