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

[LukasHirt]

Summary

The per-user SSE consumer manager (backend/pkg/sse/manager.go) had two related timing/race bugs that could cause event-triggered workflows to silently never fire uploads:

1. Dead consumer never retried. 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). Since the entry never left m.active, every later reconcile() pass saw the user as "already active" and never retried the consumer, even after automation was later connected — the consumer was dead forever.

   Fix: consumeForUser now removes its own m.active entry when it exits due to a GetAutomation failure. All the other return paths happen because ctx was cancelled by reconcile's removal logic or by shutdown — those already remove the entry themselves, so consumeForUser must not also delete it there (that would race with a fresher entry a later reconcile pass may have already installed for the same userID). This is guarded with a small per-entry generation id (activeConsumer.id) so self-cleanup only ever deletes the entry it actually owns.

2. Up-to-30s blind window. reconcile() only ran on a fixed 30s ticker, so a workflow's event trigger being created/enabled, or a user's automation being connected, could miss an upload that happened before the next tick opened that user's SSE stream.

   Fix: added Manager.Kick() — a non-blocking, drop-if-pending signal — that Start's select loop also listens on alongside the ticker. Wired it into the two write paths that can affect the wanted-consumer set: WorkflowsHandler.syncTriggerIndex (workflow create/update, in pkg/service/workflows.go) and automation.Service.Connect (in pkg/automation/automation.go). Both take an optional Kick()-satisfying interface so they stay decoupled from the concrete *sse.Manager type; backend/pkg/command/server.go was reordered to construct the sse.Manager first and wire it into both.

Test plan

TDD: wrote the following tests against the old code first and watched them fail for the right reason, then implemented the fix and watched them pass.

• TestReconcileRetriesConsumerAfterTransientGetAutomationFailure (new, pkg/sse/manager_test.go) — reproduces bug chore(deps): Bump actions/setup-node from 6.3.0 to 7.0.0 #1: automation missing on first reconcile, connected later, asserts the consumer is retried.
• TestKickTriggersImmediateReconcile (new, pkg/sse/manager_test.go) — reproduces bug chore(deps): Bump actions/checkout from 6.0.2 to 7.0.0 #2: uses a "warm-up" trigger to prove Start's select loop is actually running (avoiding a scheduling race), then asserts Kick() picks up a newly-added trigger well within the (1-hour, in the test) ticker interval.
• All existing pkg/sse tests continue to pass, including TestReconcileStartsAndStopsConsumersAsTriggersChange (removal-path coverage).

Ran locally, matching CI's backend job exactly:

• go build ./...
• go vet ./...
• go test ./...
• go test ./... -race (extra, not in CI but run to validate the new mutex/generation-id logic)
• go build -tags=e2e ./... / go vet -tags=e2e ./... (constructor signature changes didn't break e2e compilation)

No unrelated code touched — the share-file action node commits on this branch's ancestry were left untouched; this PR only modifies pkg/sse/manager.go (+tests), pkg/service/workflows.go, pkg/automation/automation.go, and pkg/command/server.go wiring.

[dj4oC]

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

Stats: +204/-26 across 7 files — a concurrency fix, so I traced the actual interleavings rather than reading the diff at face value.

Overview

Fixes two real bugs in the per-user SSE consumer manager: (1) a consumer that failed its initial GetAutomation call left a permanent, never-retried "active" entry, since nothing removed it; (2) reconcile() only ran on a fixed 30s ticker, so a newly-added trigger or newly-connected automation could miss events for up to that long. The fix adds a per-entry generation id to activeConsumer so a goroutine's deferred self-cleanup can never clobber a newer entry for the same user, plus a non-blocking Kick() channel wired into the two write paths that can affect the wanted-consumer set.

What I specifically traced through

This is exactly the kind of change where the fix for one race can quietly introduce another, so I worked through the interleavings by hand rather than trusting the doc comments at face value:

• The generation-id guard is doing real work, not just defensive decoration. I constructed the scenario it's meant to prevent: a slow GetAutomation call for user U (id=5) is still in flight when U's trigger is removed (reconcile's stop-loop cancels+deletes the id=5 entry), then immediately re-added before the old goroutine notices (a new reconcile starts a fresh id=6 goroutine). When the old (id=5) goroutine's GetAutomation finally returns — possibly because its ctx was cancelled — it calls deactivate(U, 5), and the c.id == id check correctly refuses to delete the newer id=6 entry. Without the id, this would be a real "dead goroutine deletes a live one's bookkeeping" bug.
• The claim that every other return path in consumeForUser is ctx-caused — which is what the fix's whole "only this one path needs self-cleanup" premise rests on — I checked against the full, unchanged function body (fetched from main), not just the diff. Every return in the reconnect loop is gated by ctx.Done()/ctx.Err() != nil. The claim holds; the fix isn't missing a second spot that needed the same treatment.
• reconcile()'s stop-loop calls cancel() then delete() on the same entry while holding m.mu for the whole function — so by the time any other code (including deactivate) can acquire the lock, both halves of that removal have already happened atomically. No window where a cancelled goroutine could see a half-updated map.
• TestKickTriggersImmediateReconcile's "warm-up trigger" indirection is solving a real problem, not just being cautious: without first confirming (via waitFor on a pre-existing trigger) that Start's initial synchronous reconcile() has already run and the goroutine has moved into its select loop, the test's own triggers.entries = append(...) would race against that initial reconcile's unsynchronized read of the same slice — and the test could pass "by accident" if Start's first pass happened to catch the new entry on its own, which wouldn't actually prove Kick() did anything. The warm-up establishes a real happens-before relationship before the mutation, which is the right fix for a test that's specifically trying to prove an event-driven (not ticker-driven) code path fired.

I didn't find a bug in any of this — it's a correct, well-reasoned fix for a genuinely subtle pair of races.

Code quality & style

• Scoping Kick() to only fire on entry.TriggerType == "event" (not "schedule") in syncTriggerIndex is correctly targeted — a schedule-trigger save has nothing to do with the SSE manager, so there's no reason to wake it. Worth noting there's an intentional asymmetry: adding/enabling an event trigger kicks immediately, but removing/disabling one doesn't — which is the right call, since a late consumer shutdown (up to 30s of an unnecessary open SSE stream) is harmless, while a late consumer start (up to 30s of silently missed uploads) is the actual bug being fixed.
• Two separate one-method ReconcileKicker interfaces (in automation and service) rather than a shared type is consistent with how this codebase already defines interfaces locally at the point of use elsewhere (Executor, FileClient, etc.), not a new pattern to question.
• kick may be nil being explicitly documented and handled (if s.kick != nil) rather than assumed makes the existing tests' nil argument obviously safe rather than something to double-check.

Cross-PR note

NewWorkflowsHandler's constructor signature gains a kick parameter here, and syncTriggerIndex's body is touched by this PR and by two other currently-open PRs (#30 adds a webhook case to the same function's switch statement; #31 adds an entry.SpaceID = ... line to the same event-trigger branch). Once this PR's signature change lands, PR #30's own test file (which calls NewWorkflowsHandler(...) with the current 5-argument shape) will need a nil kick argument added, the same mechanical update this PR already made to automation_test.go/renew_test.go. Worth sequencing accordingly rather than a surprise at merge time.

Test coverage

Both new tests are regression tests in the proper TDD sense (written to fail against the old code first, per the PR body) and each is targeted at exactly one of the two bugs. Combined with the existing TestReconcileStartsAndStopsConsumersAsTriggersChange (confirmed still passing, covering the removal path this fix deliberately didn't touch), coverage is appropriately complete for the scope of this change. The PR body also notes go test ./... -race was run locally specifically to validate the new mutex/generation-id logic — exactly the right extra step for this kind of change, even though it's not part of CI.

Summary

A correct, carefully-reasoned fix for a real production bug (permanently dead consumers, up to 30s of missed events on new triggers). I traced the concurrency claims by hand rather than taking them at face value and didn't find a gap. Nothing here blocks merge.

---

🤖 Generated with Claude Code