fix(backend): close SSE reconcile race that could permanently drop a user's event triggers

backend/pkg/automation/automation.go

@@ -28,16 +28,25 @@ type GraphClient interface {
 	RevokeAppPassword(ctx context.Context, authHeader, token string) error
 }
 
+// ReconcileKicker lets Connect ask the SSE manager to reconcile its active consumers
+// immediately, instead of the newly-connected user's consumer only starting on the manager's
+// next periodic tick (up to sseReconcileInterval later). Satisfied by *sse.Manager.
+type ReconcileKicker interface {
+	Kick()
+}
+
 // Service implements the /me/automation status/connect/disconnect operations.
 type Service struct {
 	graph GraphClient
 	db    *localdb.DB
+	kick  ReconcileKicker
 	log   *slog.Logger
 }
 
-// New builds a Service.
-func New(graph GraphClient, db *localdb.DB, log *slog.Logger) *Service {
-	return &Service{graph: graph, db: db, log: log}
+// New builds a Service. kick may be nil, in which case a newly connected user's SSE consumer
+// is only picked up on the SSE manager's next periodic reconcile tick.
+func New(graph GraphClient, db *localdb.DB, kick ReconcileKicker, log *slog.Logger) *Service {
+	return &Service{graph: graph, db: db, kick: kick, log: log}
 }
 
 func toStatus(a *localdb.Automation) *model.AutomationStatus {
@@ -91,6 +100,13 @@ func (s *Service) Connect(ctx context.Context, authHeader string) (*model.Automa
 	}
 
 	s.log.Info("automation connected", "userID", userID)
+
+	// The user may already have an event trigger waiting on automation being connected —
+	// nudge the SSE manager to pick it up now rather than on its next periodic tick.
+	if s.kick != nil {
+		s.kick.Kick()
+	}
+
 	return toStatus(&a), nil
 }
 

backend/pkg/automation/automation_test.go

@@ -22,7 +22,7 @@ func TestStatusReportsConnectedForFutureExpiry(t *testing.T) {
 		t.Fatalf("UpsertAutomation: %v", err)
 	}
 
-	svc := New(&fakeGraphClient{}, db, discardLogger())
+	svc := New(&fakeGraphClient{}, db, nil, discardLogger())
 
 	status, err := svc.Status(ctx, "user-1")
 	if err != nil {
@@ -51,7 +51,7 @@ func TestStatusReportsDisconnectedForPastExpiry(t *testing.T) {
 		t.Fatalf("UpsertAutomation: %v", err)
 	}
 
-	svc := New(&fakeGraphClient{}, db, discardLogger())
+	svc := New(&fakeGraphClient{}, db, nil, discardLogger())
 
 	status, err := svc.Status(ctx, "user-1")
 	if err != nil {
@@ -69,7 +69,7 @@ func TestStatusReportsDisconnectedWhenNeverConnected(t *testing.T) {
 	db := testDB(t)
 	ctx := t.Context()
 
-	svc := New(&fakeGraphClient{}, db, discardLogger())
+	svc := New(&fakeGraphClient{}, db, nil, discardLogger())
 
 	status, err := svc.Status(ctx, "never-connected-user")
 	if err != nil {

backend/pkg/automation/renew_test.go

@@ -72,7 +72,7 @@ func TestRenewDueRenewsAutomationNearingExpiry(t *testing.T) {
 
 	newExpiry := time.Now().Add(defaultExpiry).Truncate(time.Second)
 	graph := &fakeGraphClient{mintToken: "new-password", mintExpiry: newExpiry}
-	svc := New(graph, db, discardLogger())
+	svc := New(graph, db, nil, discardLogger())
 
 	svc.renewDue(ctx)
 
@@ -114,7 +114,7 @@ func TestRenewDueSkipsAutomationNotNearingExpiry(t *testing.T) {
 	}
 
 	graph := &fakeGraphClient{}
-	svc := New(graph, db, discardLogger())
+	svc := New(graph, db, nil, discardLogger())
 
 	svc.renewDue(ctx)
 
@@ -165,7 +165,7 @@ func TestRenewDueContinuesPastAFailedRenewal(t *testing.T) {
 
 	newExpiry := time.Now().Add(defaultExpiry).Truncate(time.Second)
 	graph := &selectiveFailGraphClient{failForUsername: "admin", mintToken: "renewed", mintExpiry: newExpiry}
-	svc := New(graph, db, discardLogger())
+	svc := New(graph, db, nil, discardLogger())
 
 	svc.renewDue(ctx) // must not panic or stop early
 

backend/pkg/command/server.go

@@ -66,9 +66,15 @@ func RunServer(cfg config.Config) error {
 	}
 	defer db.Close()
 
-	automationService := automation.New(ocisClient, db, log)
+	// sseManager is constructed before the handlers below so its Kick method can be wired
+	// into them: both a workflow's event trigger being added and a user's automation being
+	// connected should nudge the SSE manager to reconcile immediately instead of waiting for
+	// its next periodic tick (up to sseReconcileInterval later).
+	sseManager := sse.New(db, store, ocisClient, graphExecutor, cfg.OCISURL, cfg.OCISInsecure, sseReconcileInterval, log)
+
+	automationService := automation.New(ocisClient, db, sseManager, log)
 
-	workflowsHandler := service.NewWorkflowsHandler(store, graphExecutor, ocisClient, db, log)
+	workflowsHandler := service.NewWorkflowsHandler(store, graphExecutor, ocisClient, db, sseManager, log)
 	automationHandler := service.NewAutomationHandler(automationService, ocisClient)
 
 	apiHandler := httpserver.New(httpserver.Options{
@@ -82,7 +88,6 @@ func RunServer(cfg config.Config) error {
 	apiServer := &http.Server{Addr: cfg.HTTPAddr, Handler: apiHandler}
 	debugSrv := &http.Server{Addr: cfg.DebugAddr, Handler: debugserver.New()}
 	sched := scheduler.New(db, store, graphExecutor, scheduleTickInterval, log)
-	sseManager := sse.New(db, store, ocisClient, graphExecutor, cfg.OCISURL, cfg.OCISInsecure, sseReconcileInterval, log)
 
 	g, gCtx := errgroup.WithContext(ctx)
 

backend/pkg/service/workflows.go

@@ -33,19 +33,29 @@ type TriggerIndexer interface {
 	DeleteTriggerIndexEntry(ctx context.Context, workflowID string) error
 }
 
+// ReconcileKicker lets a trigger-index write ask the SSE manager to reconcile its active
+// consumers immediately, instead of that consumer only starting on the manager's next
+// periodic tick (up to sseReconcileInterval later). Satisfied by *sse.Manager.
+type ReconcileKicker interface {
+	Kick()
+}
+
 // WorkflowsHandler implements the /me/workflows Graph-shaped REST API.
 type WorkflowsHandler struct {
 	store        *webdavstore.Store
 	executor     Executor
 	users        UserResolver
 	triggerIndex TriggerIndexer
+	kick         ReconcileKicker
 	log          *slog.Logger
 	now          func() time.Time
 }
 
-// NewWorkflowsHandler builds a WorkflowsHandler backed by the given store and executor.
-func NewWorkflowsHandler(store *webdavstore.Store, executor Executor, users UserResolver, triggerIndex TriggerIndexer, log *slog.Logger) *WorkflowsHandler {
-	return &WorkflowsHandler{store: store, executor: executor, users: users, triggerIndex: triggerIndex, log: log, now: time.Now}
+// NewWorkflowsHandler builds a WorkflowsHandler backed by the given store and executor. kick
+// may be nil, in which case trigger-index changes are only picked up on the SSE manager's
+// next periodic reconcile tick.
+func NewWorkflowsHandler(store *webdavstore.Store, executor Executor, users UserResolver, triggerIndex TriggerIndexer, kick ReconcileKicker, log *slog.Logger) *WorkflowsHandler {
+	return &WorkflowsHandler{store: store, executor: executor, users: users, triggerIndex: triggerIndex, kick: kick, log: log, now: time.Now}
 }
 
 // syncTriggerIndex keeps the local trigger index in sync with a workflow's current
@@ -79,6 +89,13 @@ func (h *WorkflowsHandler) syncTriggerIndex(ctx context.Context, authHeader stri
 	}
 	if err := h.triggerIndex.UpsertTriggerIndexEntry(ctx, entry); err != nil {
 		h.log.Error("update trigger index entry", "workflowID", wf.ID, "error", err)
+		return
+	}
+
+	// An event trigger just became active for this user — nudge the SSE manager to pick it
+	// up now rather than on its next periodic tick.
+	if entry.TriggerType == "event" && h.kick != nil {
+		h.kick.Kick()
 	}
 }
 

backend/pkg/sse/manager.go

@@ -77,7 +77,20 @@ type Manager struct {
 	httpClient *http.Client
 
 	mu     sync.Mutex
-	active map[string]context.CancelFunc // userID -> cancel
+	active map[string]activeConsumer // userID -> consumer
+	nextID uint64                    // monotonically increasing id, tags each activeConsumer
+
+	kick chan struct{}
+}
+
+// activeConsumer tracks one userID's running consumeForUser goroutine. id disambiguates
+// consumeForUser's self-removal (see deactivate) from a newer goroutine that reconcile may
+// have already started for the same userID by the time the old one gets around to cleaning
+// up after itself — without it, that self-removal could delete a live entry that isn't even
+// the one it owns.
+type activeConsumer struct {
+	cancel context.CancelFunc
+	id     uint64
 }
 
 // New builds a Manager.
@@ -96,11 +109,27 @@ func New(db TriggerStore, store WorkflowStore, paths PathResolver, executor Exec
 		interval:   interval,
 		log:        log,
 		httpClient: &http.Client{Transport: transport}, // no overall Timeout: this is a long-lived stream
-		active:     map[string]context.CancelFunc{},
+		active:     map[string]activeConsumer{},
+		kick:       make(chan struct{}, 1),
 	}
 }
 
-// Start blocks, reconciling active consumers every interval, until ctx is done.
+// Kick requests an immediate reconcile pass instead of waiting for the next periodic tick.
+// Callers that just made a change that could affect the wanted-consumer set — a workflow's
+// event trigger being added/enabled, or a user's automation getting connected — should call
+// this so the corresponding SSE consumer starts promptly instead of within the next interval
+// (up to sseReconcileInterval later). Non-blocking and safe to call from any goroutine; if a
+// kick is already pending, this is a no-op since one reconcile pass picks up all pending
+// changes anyway.
+func (m *Manager) Kick() {
+	select {
+	case m.kick <- struct{}{}:
+	default:
+	}
+}
+
+// Start blocks, reconciling active consumers every interval — or immediately whenever Kick
+// is called — until ctx is done.
 func (m *Manager) Start(ctx context.Context) {
 	m.reconcile(ctx)
 
@@ -113,6 +142,8 @@ func (m *Manager) Start(ctx context.Context) {
 			return
 		case <-ticker.C:
 			m.reconcile(ctx)
+		case <-m.kick:
+			m.reconcile(ctx)
 		}
 	}
 }
@@ -132,9 +163,9 @@ func (m *Manager) reconcile(ctx context.Context) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
-	for userID, cancel := range m.active {
+	for userID, c := range m.active {
 		if !wanted[userID] {
-			cancel()
+			c.cancel()
 			delete(m.active, userID)
 		}
 	}
@@ -143,26 +174,49 @@ func (m *Manager) reconcile(ctx context.Context) {
 			continue
 		}
 		cctx, cancel := context.WithCancel(ctx)
-		m.active[userID] = cancel
-		go m.consumeForUser(cctx, userID)
+		m.nextID++
+		id := m.nextID
+		m.active[userID] = activeConsumer{cancel: cancel, id: id}
+		go m.consumeForUser(cctx, userID, id)
 	}
 }
 
 func (m *Manager) stopAll() {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	for userID, cancel := range m.active {
-		cancel()
+	for userID, c := range m.active {
+		c.cancel()
+		delete(m.active, userID)
+	}
+}
+
+// deactivate removes userID's active-consumer entry, but only if it still belongs to the
+// caller (identified by id) — i.e. reconcile hasn't already replaced it with a newer
+// consumer for the same userID in the meantime. See consumeForUser and activeConsumer.
+func (m *Manager) deactivate(userID string, id uint64) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if c, ok := m.active[userID]; ok && c.id == id {
 		delete(m.active, userID)
 	}
 }
 
 // consumeForUser holds one SSE connection open for userID, reconnecting with backoff if it
-// drops, until ctx is cancelled (the user's last event trigger was removed, or shutdown).
-func (m *Manager) consumeForUser(ctx context.Context, userID string) {
+// drops, until ctx is cancelled (the user's last event trigger was removed, or shutdown). id
+// is the activeConsumer id reconcile assigned when it started this goroutine.
+//
+// Note on m.active bookkeeping: below this point, every return happens because ctx was
+// cancelled — by reconcile (userID dropped out of wanted) or by stopAll (shutdown) — and
+// both of those already remove the m.active entry themselves under the lock before
+// cancelling, so consumeForUser must not also delete it there; doing so could race with a
+// fresh entry a later reconcile pass has already installed for the same userID. The
+// GetAutomation failure below is the one return path ctx did NOT cause, so it's the one case
+// where this goroutine is responsible for cleaning up after itself.
+func (m *Manager) consumeForUser(ctx context.Context, userID string, id uint64) {
 	automation, err := m.db.GetAutomation(ctx, userID)
 	if err != nil {
 		m.log.Warn("sse manager: user has an event trigger but no automation connected", "userID", userID)
+		m.deactivate(userID, id)
 		return
 	}
 	authHeader := "Basic " + base64.StdEncoding.EncodeToString(fmt.Appendf(nil, "%s:%s", automation.Username, automation.AppPassword))

backend/pkg/sse/manager_test.go

@@ -179,6 +179,92 @@ func TestHandleEventIgnoresUnmappedSSEEventType(t *testing.T) {
 	}
 }
 
+// TestReconcileRetriesConsumerAfterTransientGetAutomationFailure regression-tests the bug
+// where consumeForUser inserted its m.active entry *before* calling GetAutomation, but never
+// removed it if GetAutomation failed (e.g. the user has an event trigger but hasn't connected
+// automation yet). Because the entry stayed in m.active forever, every subsequent reconcile
+// pass saw the userID as "already active" and never retried it, even after automation was
+// connected later.
+func TestReconcileRetriesConsumerAfterTransientGetAutomationFailure(t *testing.T) {
+	triggers := &fakeTriggerStore{
+		entries:     []localdb.TriggerIndexEntry{{WorkflowID: "wf-1", UserID: "user-1", TriggerType: "event", EventType: "upload"}},
+		automations: map[string]*localdb.Automation{}, // automation not connected yet
+	}
+	store := &fakeWorkflowStore{workflows: map[string]model.WorkflowDefinition{"wf-1": {ID: "wf-1", Enabled: true}}}
+	m := New(triggers, store, &fakePathResolver{}, &fakeExecutor{}, "http://unused-will-fail-fast", false, time.Hour, discardLogger())
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	m.reconcile(ctx)
+	// consumeForUser should observe the GetAutomation failure, log, and return — and in doing
+	// so remove its own now-dead entry from m.active so a later reconcile pass can retry it.
+	waitFor(t, time.Second, func() bool {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		_, ok := m.active["user-1"]
+		return !ok
+	})
+
+	// Automation gets connected sometime later.
+	triggers.automations["user-1"] = &localdb.Automation{UserID: "user-1", Username: "admin", AppPassword: "secret"}
+
+	// The next reconcile pass (in production, either the next tick or a Kick()) must retry
+	// the consumer now that the entry was cleared, instead of treating it as still active.
+	m.reconcile(ctx)
+	waitFor(t, time.Second, func() bool {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		_, ok := m.active["user-1"]
+		return ok
+	})
+}
+
+// TestKickTriggersImmediateReconcile regression-tests the up-to-30s blind window: without an
+// event-driven reconcile hook, Start only reconciles on its periodic ticker, so a trigger
+// added right after Start began running would have to wait for the next tick to pick up. Kick
+// lets callers (workflow trigger-index writes, automation Connect) request an immediate pass.
+func TestKickTriggersImmediateReconcile(t *testing.T) {
+	// A pre-existing "warm" trigger lets the test prove Start's initial synchronous
+	// reconcile — and hence its select loop — has actually run, without racing: checking
+	// for an *absence* of state right after `go m.Start(ctx)` would trivially succeed
+	// before Start's goroutine is even scheduled, proving nothing.
+	triggers := &fakeTriggerStore{
+		entries: []localdb.TriggerIndexEntry{{WorkflowID: "wf-warm", UserID: "user-warm", TriggerType: "event", EventType: "upload"}},
+		automations: map[string]*localdb.Automation{
+			"user-warm": {UserID: "user-warm", Username: "admin", AppPassword: "x"},
+			"user-1":    {UserID: "user-1", Username: "admin", AppPassword: "x"},
+		},
+	}
+	m := New(triggers, &fakeWorkflowStore{}, &fakePathResolver{}, &fakeExecutor{}, "http://unused-will-fail-fast", false, time.Hour, discardLogger())
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	go m.Start(ctx)
+	waitFor(t, time.Second, func() bool {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		_, ok := m.active["user-warm"]
+		return ok
+	})
+
+	// Only now, after reconcile() is known to have returned and the select loop entered,
+	// mutate the wanted set — simulating a workflow with an event trigger being created for
+	// user-1 — and ask for an immediate reconcile instead of waiting for the ticker.
+	triggers.entries = append(triggers.entries, localdb.TriggerIndexEntry{WorkflowID: "wf-1", UserID: "user-1", TriggerType: "event", EventType: "upload"})
+	m.Kick()
+
+	// This must succeed well within the 1-hour ticker interval, proving Kick — not the
+	// ticker — is what triggered the reconcile.
+	waitFor(t, time.Second, func() bool {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		_, ok := m.active["user-1"]
+		return ok
+	})
+}
+
 func TestReconcileStartsAndStopsConsumersAsTriggersChange(t *testing.T) {
 	triggers := &fakeTriggerStore{
 		entries:     []localdb.TriggerIndexEntry{{WorkflowID: "wf-1", UserID: "user-1", TriggerType: "event", EventType: "upload"}},