From 79f4285338ea6a8dace4a3dd0933eddb305e14b0 Mon Sep 17 00:00:00 2001 From: Lukas Hirt Date: Fri, 24 Jul 2026 19:18:22 +0200 Subject: [PATCH] fix(backend): close SSE reconcile race that could permanently drop a user's event triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two related bugs in the per-user SSE consumer manager (pkg/sse/manager.go) combined to cause an up-to-30s or, in the worst case, permanent gap between a workflow's event trigger becoming active and its uploads actually firing: 1. consumeForUser inserted its m.active[userID] 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, or their app-password was revoked). Because the entry never left m.active, every later reconcile() pass treated that user as "already active" and never retried the consumer — even once automation was connected. Fixed by having consumeForUser remove its own entry on that one failure path. The other return paths (ctx cancelled by reconcile's removal logic or by shutdown) already have their entries removed by the canceller, so self-removal there is skipped to avoid racing with a fresher entry a later reconcile pass may have already installed for the same userID — guarded with a per-entry generation id (activeConsumer.id) so the cleanup only ever deletes the entry it actually owns. 2. reconcile() only ran on a fixed 30s ticker, so a workflow's event trigger being added/enabled, or a user's automation being connected, could miss an upload that happened in the blind window before the next tick opened that user's SSE stream. Added Manager.Kick(), a non-blocking, drop-if-pending signal that Start's select loop also listens on, and wired it into the two write paths that can affect the wanted-consumer set: WorkflowsHandler.syncTriggerIndex (workflow create/update) and automation.Service.Connect. Tests added to pkg/sse/manager_test.go under TDD, watched failing against the old code before the fix: - TestReconcileRetriesConsumerAfterTransientGetAutomationFailure - TestKickTriggersImmediateReconcile Full backend suite (go build, go vet, go test ./... -race) passes. Signed-off-by: Lukas Hirt --- backend/pkg/automation/automation.go | 22 +++++- backend/pkg/automation/automation_test.go | 6 +- backend/pkg/automation/renew_test.go | 6 +- backend/pkg/command/server.go | 11 ++- backend/pkg/service/workflows.go | 23 +++++- backend/pkg/sse/manager.go | 76 +++++++++++++++++--- backend/pkg/sse/manager_test.go | 86 +++++++++++++++++++++++ 7 files changed, 204 insertions(+), 26 deletions(-) diff --git a/backend/pkg/automation/automation.go b/backend/pkg/automation/automation.go index c78c2ed..82e7cc4 100644 --- a/backend/pkg/automation/automation.go +++ b/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 } diff --git a/backend/pkg/automation/automation_test.go b/backend/pkg/automation/automation_test.go index f9aa4fc..13e06c3 100644 --- a/backend/pkg/automation/automation_test.go +++ b/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 { diff --git a/backend/pkg/automation/renew_test.go b/backend/pkg/automation/renew_test.go index 57744e5..bc8ffb1 100644 --- a/backend/pkg/automation/renew_test.go +++ b/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 diff --git a/backend/pkg/command/server.go b/backend/pkg/command/server.go index 55d4a9d..0277c97 100644 --- a/backend/pkg/command/server.go +++ b/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) diff --git a/backend/pkg/service/workflows.go b/backend/pkg/service/workflows.go index e97f2b8..41df8ee 100644 --- a/backend/pkg/service/workflows.go +++ b/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() } } diff --git a/backend/pkg/sse/manager.go b/backend/pkg/sse/manager.go index 9d4f028..717b2d8 100644 --- a/backend/pkg/sse/manager.go +++ b/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)) diff --git a/backend/pkg/sse/manager_test.go b/backend/pkg/sse/manager_test.go index b73f8cf..49e7c0c 100644 --- a/backend/pkg/sse/manager_test.go +++ b/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"}},