diff --git a/Makefile b/Makefile index 8a4d2b7..8cd8c69 100644 --- a/Makefile +++ b/Makefile @@ -1,3 +1,4 @@ +SHELL := /bin/bash .PHONY: lint lint: @@ -7,5 +8,5 @@ lint: .PHONY: test test: which go-test-coverage || go install github.com/vladopajic/go-test-coverage/v2@latest - go test -v ./... -coverprofile=./cover.out -covermode=atomic -coverpkg=./... -json | python3 testutil/colourise-go-test-output.py >/dev/null + set -o pipefail; go test -race -v ./... -coverprofile=./cover.out -covermode=atomic -coverpkg=./... -json | python3 testutil/colourise-go-test-output.py >/dev/null go-test-coverage --config=./.testcoverage.yaml diff --git a/connection_metrics_test.go b/connection_metrics_test.go index 0c7e58e..2510eb9 100644 --- a/connection_metrics_test.go +++ b/connection_metrics_test.go @@ -2,6 +2,7 @@ package hal_test import ( "context" + "sync/atomic" "testing" "time" @@ -27,7 +28,7 @@ func TestMetricsInstrumentation(t *testing.T) { WithAction(func(ctx context.Context, trigger hal.EntityInterface) { light.TurnOn() }) - + conn.RegisterAutomations(automation) // Trigger a state change that will run automation @@ -54,14 +55,14 @@ func TestMetricsAutomationTriggered(t *testing.T) { sensor := hal.NewBinarySensor("test.sensor") conn.RegisterEntities(sensor) - var automationExecuted bool + var automationExecuted atomic.Bool automation := hal.NewAutomation(). WithName("test_automation"). WithEntities(sensor). WithAction(func(ctx context.Context, trigger hal.EntityInterface) { - automationExecuted = true + automationExecuted.Store(true) }) - + conn.RegisterAutomations(automation) // Trigger automation @@ -72,10 +73,11 @@ func TestMetricsAutomationTriggered(t *testing.T) { NewState: &homeassistant.State{State: "on"}, }, }) - time.Sleep(10 * time.Millisecond) - // Verify automation was triggered (which means metrics were collected) - assert.Assert(t, automationExecuted, "Expected automation to be triggered") + // Wait for the automation to run (which means metrics were collected). + testutil.WaitFor(t, "automation triggered", func() bool { + return automationExecuted.Load() + }, func() {}) } func TestMetricsIntegrationBasicScenarios(t *testing.T) { @@ -89,24 +91,24 @@ func TestMetricsIntegrationBasicScenarios(t *testing.T) { conn.RegisterEntities(sensor, light1, light2) // Create multiple automations for the same trigger - var automation1Executed, automation2Executed bool - + var automation1Executed, automation2Executed atomic.Bool + automation1 := hal.NewAutomation(). WithName("automation_1"). WithEntities(sensor). WithAction(func(ctx context.Context, trigger hal.EntityInterface) { - automation1Executed = true + automation1Executed.Store(true) light1.TurnOn() }) - + automation2 := hal.NewAutomation(). WithName("automation_2"). WithEntities(sensor). WithAction(func(ctx context.Context, trigger hal.EntityInterface) { - automation2Executed = true + automation2Executed.Store(true) light2.TurnOn() }) - + conn.RegisterAutomations(automation1, automation2) // Test normal state change (should trigger automations and record metrics) @@ -117,15 +119,15 @@ func TestMetricsIntegrationBasicScenarios(t *testing.T) { NewState: &homeassistant.State{State: "on"}, }, }) - time.Sleep(50 * time.Millisecond) - // Verify both automations were triggered - assert.Assert(t, automation1Executed, "Expected automation 1 to be triggered") - assert.Assert(t, automation2Executed, "Expected automation 2 to be triggered") - + // Wait for both automations to run. + testutil.WaitFor(t, "both automations triggered", func() bool { + return automation1Executed.Load() && automation2Executed.Load() + }, func() {}) + // Reset for next test - automation1Executed = false - automation2Executed = false + automation1Executed.Store(false) + automation2Executed.Store(false) // Test state change from HAL's own user (should NOT trigger automations) server.SendEvent(homeassistant.Event{ @@ -141,8 +143,8 @@ func TestMetricsIntegrationBasicScenarios(t *testing.T) { time.Sleep(50 * time.Millisecond) // Verify automations were NOT triggered (loop protection) - assert.Assert(t, !automation1Executed, "Expected automation 1 NOT to be triggered for own actions") - assert.Assert(t, !automation2Executed, "Expected automation 2 NOT to be triggered for own actions") - + assert.Assert(t, !automation1Executed.Load(), "Expected automation 1 NOT to be triggered for own actions") + assert.Assert(t, !automation2Executed.Load(), "Expected automation 2 NOT to be triggered for own actions") + // If we get here without panics/errors, metrics collection is working properly -} \ No newline at end of file +} diff --git a/entities.go b/entities.go index 388d171..8e759f9 100644 --- a/entities.go +++ b/entities.go @@ -2,6 +2,7 @@ package hal import ( "reflect" + "sync" "github.com/dansimau/hal/homeassistant" ) @@ -21,7 +22,11 @@ type Entities []EntityInterface // Entity is a base type for all entities that can be embedded into other types. type Entity struct { connection *Connection - state homeassistant.State + + // mutex guards state, which is read by automations and user code while the + // WebSocket reader goroutine writes incoming state changes. + mutex sync.RWMutex + state homeassistant.State } func NewEntity(id string) *Entity { @@ -35,14 +40,23 @@ func (e *Entity) BindConnection(connection *Connection) { } func (e *Entity) GetID() string { + e.mutex.RLock() + defer e.mutex.RUnlock() + return e.state.EntityID } func (e *Entity) SetState(state homeassistant.State) { + e.mutex.Lock() + defer e.mutex.Unlock() + e.state = state } func (e *Entity) GetState() homeassistant.State { + e.mutex.RLock() + defer e.mutex.RUnlock() + return e.state } diff --git a/logger/service.go b/logger/service.go index c8b8a94..4595a8d 100644 --- a/logger/service.go +++ b/logger/service.go @@ -109,17 +109,24 @@ func (s *Service) Start() { default: // Channel is still open } + // Capture the current stop channel so the pruning goroutine watches the + // channel that was live when it started, rather than reading the shared + // field (which Start may replace) without holding the lock. + stopChan := s.stopChan hasDB := s.db != nil s.mu.Unlock() if hasDB { - go s.pruneLogs() + go s.pruneLogs(stopChan) } slog.Info("Logging service started") } // Stop stops the logging service func (s *Service) Stop() { + s.mu.Lock() + defer s.mu.Unlock() + select { case <-s.stopChan: // Already stopped @@ -489,14 +496,15 @@ func (s *Service) writeLogText(level slog.Level, entityID string, logText string } } -// pruneLogs runs in a goroutine to periodically remove old logs -func (s *Service) pruneLogs() { +// pruneLogs runs in a goroutine to periodically remove old logs. It watches the +// stop channel captured by Start rather than the shared field. +func (s *Service) pruneLogs(stopChan <-chan struct{}) { ticker := time.NewTicker(s.pruneInterval) defer ticker.Stop() for { select { - case <-s.stopChan: + case <-stopChan: return case <-ticker.C: s.mu.RLock() diff --git a/store/sqlite.go b/store/sqlite.go index b27fbc4..e11c953 100644 --- a/store/sqlite.go +++ b/store/sqlite.go @@ -1,6 +1,8 @@ package store import ( + "strings" + "github.com/glebarez/sqlite" // Pure go SQLite driver, checkout https://github.com/glebarez/sqlite for details "gorm.io/gorm" ) @@ -21,6 +23,20 @@ func Open(path string) (*Store, error) { return nil, err } + // In-memory SQLite databases are scoped to a single connection: every new + // connection in the pool gets its own empty database, so a migration run on + // one connection is invisible to a query that lands on another ("no such + // table"). Pin the pool to a single connection so the whole Store shares one + // in-memory database. File-backed databases keep the default pool. + if strings.Contains(path, ":memory:") { + sqlDB, err := db.DB() + if err != nil { + return nil, err + } + + sqlDB.SetMaxOpenConns(1) + } + // Configure SQLite settings if err := db.Exec("PRAGMA journal_mode = WAL").Error; err != nil { return nil, err diff --git a/timer.go b/timer.go index ec555e8..c148bf1 100644 --- a/timer.go +++ b/timer.go @@ -2,6 +2,7 @@ package hal import ( "context" + "sync" "time" "github.com/benbjohnson/clock" @@ -9,10 +10,14 @@ import ( // Timer wraps time.Timer to add functionality for checking if the timer is running. type Timer struct { + // mutex guards the fields below, which the timer callback goroutine mutates + // concurrently with callers of StartContext/Cancel/IsRunning. + mutex sync.Mutex clock clock.Clock timer *clock.Timer running bool ctx context.Context + fn func(context.Context) } func NewTimer(clock clock.Clock) *Timer { @@ -22,29 +27,33 @@ func NewTimer(clock clock.Clock) *Timer { } func (t *Timer) Cancel() { + t.mutex.Lock() + defer t.mutex.Unlock() + if t.timer == nil { return } t.timer.Stop() + t.running = false } // StartContext starts a timer with context that will be passed to the callback func (t *Timer) StartContext(ctx context.Context, fn func(context.Context), duration time.Duration) { + t.mutex.Lock() + defer t.mutex.Unlock() + if t.clock == nil { t.clock = clock.New() } + // Record the latest callback and context so that resetting an existing + // timer picks them up instead of firing the ones captured on first start. t.ctx = ctx + t.fn = fn if t.timer == nil { - t.timer = t.clock.AfterFunc(duration, func() { - t.running = false - - if fn != nil { - fn(ctx) - } - }) + t.timer = t.clock.AfterFunc(duration, t.fire) } else { t.timer.Reset(duration) } @@ -52,6 +61,19 @@ func (t *Timer) StartContext(ctx context.Context, fn func(context.Context), dura t.running = true } +// fire runs when the timer expires. It reads the most recently configured +// callback and context under the lock so a Reset takes effect. +func (t *Timer) fire() { + t.mutex.Lock() + t.running = false + fn, ctx := t.fn, t.ctx + t.mutex.Unlock() + + if fn != nil { + fn(ctx) + } +} + // Start starts the timer or resets it to a new duration. // Deprecated: Use StartContext to propagate context for tracing. func (t *Timer) Start(fn func(), duration time.Duration) { @@ -64,5 +86,8 @@ func (t *Timer) Start(fn func(), duration time.Duration) { // IsRunning returns whether the timer is currently running. func (t *Timer) IsRunning() bool { + t.mutex.Lock() + defer t.mutex.Unlock() + return t.running } diff --git a/timer_test.go b/timer_test.go index 1263cd9..f0f906d 100644 --- a/timer_test.go +++ b/timer_test.go @@ -1,6 +1,7 @@ package hal_test import ( + "sync/atomic" "testing" "time" @@ -10,17 +11,17 @@ import ( func TestTimer(t *testing.T) { t.Parallel() - timerRan := false + var timerRan atomic.Bool var timer hal.Timer timer.Start(func() { - timerRan = true + timerRan.Store(true) }, 100*time.Millisecond) time.Sleep(200 * time.Millisecond) - if !timerRan { + if !timerRan.Load() { t.Error("Timer did not run") } }