fix(systemd): keep user unit state in sync at startup and on restart

fsnotify only reports invocation links, so odio-api missed user units that
settle without a new one: a unit snapshotted while activating (odio-screen
waiting on odio-api), and a restart whose start event landed while the stop
was still being watched. Follow the former, and re-read the latter.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013gFB9WgpMajs1zcfrxyanA

Author: b0bbywan

backend/systemd/fsnotify.go

@@ -105,16 +105,53 @@ func (l *Listener) dispatchFSNotify(event fsnotify.Event) {
 		return
 	}
 
-	if _, loaded := l.watcherMap.LoadOrStore(serviceName, true); !loaded {
-		go l.waitForStableState(serviceName)
+	l.track(serviceName)
+}
+
+// track starts a watcher for service, or marks the running one: an event it
+// has not seen may postdate the state it is about to settle on (a restart).
+func (l *Listener) track(service string) {
+	l.watchMu.Lock()
+	defer l.watchMu.Unlock()
+	if l.watching == nil {
+		l.watching = make(map[string]bool)
+	}
+	if _, running := l.watching[service]; running {
+		l.watching[service] = true
+		return
 	}
+	l.watching[service] = false
+	go l.waitForStableState(service)
+}
+
+// settle ends the watch on service, unless an event came in meanwhile: then it
+// clears the mark and reports false, and the watcher reads the state again.
+func (l *Listener) settle(service string) bool {
+	l.watchMu.Lock()
+	defer l.watchMu.Unlock()
+	if l.watching[service] {
+		l.watching[service] = false
+		return false
+	}
+	delete(l.watching, service)
+	return true
+}
+
+func (l *Listener) untrack(service string) {
+	l.watchMu.Lock()
+	defer l.watchMu.Unlock()
+	delete(l.watching, service)
 }
 
 func (l *Listener) waitForStableState(service string) {
 	timeout := l.backend.config.Timeout
 	ctx, cancel := context.WithTimeout(l.backend.ctx, timeout)
+	settled := false
 	defer func() {
-		l.watcherMap.Delete(service)
+		// Once settled the entry may already belong to a newer watcher.
+		if !settled {
+			l.untrack(service)
+		}
 		if ctx.Err() == context.DeadlineExceeded {
 			logger.Warn("[systemd] %s failed to start in less than %s, cache might be out of sync", service, timeout)
 			refreshCtx, refreshCancel := context.WithTimeout(l.backend.ctx, 5*time.Second)
@@ -146,8 +183,13 @@ func (l *Listener) waitForStableState(service string) {
 				timer.Reset(waitTime)
 				continue
 			}
-			switch unit.ActiveState {
-			case "active", "inactive", "failed":
+			if isStableState(unit.ActiveState) {
+				if !l.settle(service) {
+					logger.Debug("[systemd] %s/%s changed while settling on %s, reading again", ScopeUser, service, unit.ActiveState)
+					timer.Reset(0)
+					continue
+				}
+				settled = true
 				logger.Debug("[systemd] %s/%s reached stable state: %s", ScopeUser, service, unit.ActiveState)
 				l.backend.notifyService(*unit)
 				return
@@ -162,3 +204,33 @@ func (l *Listener) waitForStableState(service string) {
 		}
 	}
 }
+
+func isStableState(state string) bool {
+	switch state {
+	case "active", "inactive", "failed":
+		return true
+	}
+	return false
+}
+
+// trackTransitional follows the watched user units the startup snapshot caught
+// mid-transition: their invocation link predates the watcher, so no event comes.
+func (l *Listener) trackTransitional(services []Service) {
+	if l.supportsUTMP {
+		return // D-Bus signals report the end of the transition
+	}
+	for _, name := range transitionalUnits(services, l.userWatched) {
+		l.track(name)
+	}
+}
+
+// transitionalUnits lists the watched user units not yet in a stable state.
+func transitionalUnits(services []Service, watched map[string]bool) []string {
+	var names []string
+	for _, svc := range services {
+		if svc.Scope == ScopeUser && watched[svc.Name] && !isStableState(svc.ActiveState) {
+			names = append(names, svc.Name)
+		}
+	}
+	return names
+}

backend/systemd/systemd.go

@@ -72,7 +72,8 @@ func (s *SystemdBackend) Start() error {
 	logger.Debug("[systemd] starting backend (utmp=%v)", s.config.SupportsUTMP)
 
 	// Load the cache at startup
-	if _, err := s.ListServices(); err != nil {
+	services, err := s.ListServices()
+	if err != nil {
 		return err
 	}
 
@@ -81,6 +82,7 @@ func (s *SystemdBackend) Start() error {
 	if err := s.listener.Start(); err != nil {
 		return err
 	}
+	s.listener.trackTransitional(services)
 
 	logger.Info("[systemd] backend started successfully")
 	return nil

backend/systemd/transitional_test.go

@@ -0,0 +1,79 @@
+package systemd
+
+import (
+	"slices"
+	"testing"
+)
+
+func TestTransitionalUnits(t *testing.T) {
+	services := []Service{
+		{Name: "odio-screen.service", Scope: ScopeUser, ActiveState: "activating"},
+		{Name: "mpd.service", Scope: ScopeUser, ActiveState: "active"},
+		{Name: "spotifyd.service", Scope: ScopeUser, ActiveState: "failed"},
+		{Name: "qbzd.service", Scope: ScopeUser, ActiveState: "deactivating"},
+		{Name: "bluetooth.service", Scope: ScopeSystem, ActiveState: "activating"},
+		{Name: "unwatched.service", Scope: ScopeUser, ActiveState: "reloading"},
+	}
+	watched := map[string]bool{
+		"odio-screen.service": true,
+		"mpd.service":         true,
+		"spotifyd.service":    true,
+		"qbzd.service":        true,
+		"bluetooth.service":   true,
+	}
+
+	got := transitionalUnits(services, watched)
+	want := []string{"odio-screen.service", "qbzd.service"}
+	if !slices.Equal(got, want) {
+		t.Errorf("transitionalUnits = %v, want %v", got, want)
+	}
+}
+
+func TestTrackTransitionalLeavesDBusModeToSignals(t *testing.T) {
+	l := &Listener{
+		supportsUTMP: true,
+		userWatched:  map[string]bool{"odio-screen.service": true},
+	}
+
+	l.trackTransitional([]Service{{Name: "odio-screen.service", Scope: ScopeUser, ActiveState: "activating"}})
+
+	if _, tracked := l.watching["odio-screen.service"]; tracked {
+		t.Error("odio-screen.service tracked in D-Bus mode, want signals to cover it")
+	}
+}
+
+func TestTrackMarksARunningWatch(t *testing.T) {
+	// A restart: the stop event started the watch, the start event must not be lost.
+	l := &Listener{watching: map[string]bool{"odio-screen.service": false}}
+
+	l.track("odio-screen.service")
+
+	if marked := l.watching["odio-screen.service"]; !marked {
+		t.Error("running watch not marked, the start event would be dropped")
+	}
+}
+
+func TestSettleEndsAQuietWatch(t *testing.T) {
+	l := &Listener{watching: map[string]bool{"odio-screen.service": false}}
+
+	if !l.settle("odio-screen.service") {
+		t.Fatal("settle = false, want true with no event since")
+	}
+	if _, tracked := l.watching["odio-screen.service"]; tracked {
+		t.Error("settled watch still tracked")
+	}
+}
+
+func TestSettleReadsAgainAfterAnEvent(t *testing.T) {
+	l := &Listener{watching: map[string]bool{"odio-screen.service": true}}
+
+	if l.settle("odio-screen.service") {
+		t.Fatal("settle = true, want false: an event came in meanwhile")
+	}
+	if marked, tracked := l.watching["odio-screen.service"]; !tracked || marked {
+		t.Errorf("watching = %v, want still tracked with the mark cleared", l.watching)
+	}
+	if !l.settle("odio-screen.service") {
+		t.Error("second settle = false, want true once the event was read")
+	}
+}

backend/systemd/types.go

@@ -23,7 +23,10 @@ type Listener struct {
 	// Deduplication: last known state per service/scope
 	lastState   map[string]string
 	lastStateMu sync.RWMutex
-	watcherMap  sync.Map
+
+	// User units watched to a stable state; true = an event came in since.
+	watching map[string]bool
+	watchMu  sync.Mutex
 }
 
 type UnitScope string