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"}},