From 4b05e9755d97d3806168f0273dcf649fd104d9cc Mon Sep 17 00:00:00 2001 From: Nicolas Lenepveu Date: Sun, 31 May 2026 00:03:27 +0200 Subject: [PATCH] feat: throttle issue-watcher task fan-out (per-watcher cap) (#1113) --- apps/backend/cmd/kandev/orchestrator.go | 5 + .../internal/integrations/optional/int.go | 38 ++ .../integrations/optional/int_test.go | 55 +++ apps/backend/internal/jira/models.go | 65 ++-- apps/backend/internal/jira/service.go | 26 ++ .../internal/jira/service_issue_watch_test.go | 95 +++++ apps/backend/internal/jira/store.go | 51 ++- apps/backend/internal/linear/models.go | 45 ++- .../internal/linear/service_issue_watch.go | 26 ++ .../linear/service_issue_watch_test.go | 98 +++++ apps/backend/internal/linear/store.go | 28 ++ .../internal/linear/store_issue_watch.go | 55 ++- apps/backend/internal/orchestrator/service.go | 14 + .../internal/orchestrator/source_jira.go | 16 + .../internal/orchestrator/source_linear.go | 16 + .../internal/orchestrator/watcher_dispatch.go | 11 + .../orchestrator/watcher_dispatch_test.go | 6 + .../orchestrator/watcher_dispatch_wiring.go | 23 +- .../internal/orchestrator/watcher_throttle.go | 179 +++++++++ .../orchestrator/watcher_throttle_test.go | 358 ++++++++++++++++++ .../task/handlers/process_handlers_test.go | 3 + .../count_open_watcher_tasks_test.go | 89 +++++ .../internal/task/repository/interface.go | 5 + .../internal/task/repository/sqlite/task.go | 43 +++ .../task/service/handoff_workspace_test.go | 4 + .../jira/jira-issue-watch-dialog.tsx | 64 +++- .../linear/linear-issue-watch-dialog.tsx | 59 ++- .../linear/linear-issue-watch-form.test.ts | 91 +++++ .../linear/linear-issue-watch-form.ts | 84 ++++ apps/web/lib/types/jira.ts | 10 + apps/web/lib/types/linear.ts | 10 + 31 files changed, 1591 insertions(+), 81 deletions(-) create mode 100644 apps/backend/internal/integrations/optional/int.go create mode 100644 apps/backend/internal/integrations/optional/int_test.go create mode 100644 apps/backend/internal/orchestrator/watcher_throttle.go create mode 100644 apps/backend/internal/orchestrator/watcher_throttle_test.go create mode 100644 apps/backend/internal/task/repository/count_open_watcher_tasks_test.go create mode 100644 apps/web/components/linear/linear-issue-watch-form.test.ts diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index d39437741a..228f3e3f7d 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -116,6 +116,11 @@ func provideOrchestrator( // Wire issue task creator for auto-creating tasks from issue watch events orchestratorSvc.SetIssueTaskCreator(&issueTaskCreatorAdapter{svc: taskSvc}) + // Wire the per-watcher throttle gate. taskRepo exposes the JSON-scoped + // count of open watcher-created tasks; the orchestrator combines it with + // an in-process pending counter so a poll-tick burst can't overshoot. + orchestratorSvc.SetWatcherTaskCounter(taskRepo) + // Wire repository resolver for auto-cloning repos during review task creation if repoCloner != nil { orchestratorSvc.SetRepositoryResolver(&repositoryResolverAdapter{ diff --git a/apps/backend/internal/integrations/optional/int.go b/apps/backend/internal/integrations/optional/int.go new file mode 100644 index 0000000000..8a88c0889f --- /dev/null +++ b/apps/backend/internal/integrations/optional/int.go @@ -0,0 +1,38 @@ +// Package optional provides JSON field types that distinguish "absent" from +// "explicitly null" in PATCH-style request payloads — a distinction a plain +// `*int` cannot make, because both decode to a nil pointer. +package optional + +import "encoding/json" + +// Int is a tri-state integer field for partial-update request bodies: +// +// - key absent → Present == false (leave the target unchanged) +// - key present null → Present == true, Value == nil (clear to "unset") +// - key present N → Present == true, Value == &N (set to N) +// +// Declare it as a plain (non-pointer) struct field WITHOUT `omitempty`: +// +// MaxInflightTasks optional.Int `json:"maxInflightTasks"` +// +// encoding/json only invokes UnmarshalJSON when the key is present in the +// payload, so Present reliably reports whether the client sent the field. +type Int struct { + Present bool + Value *int +} + +// UnmarshalJSON records that the key was present and decodes null vs. number. +func (o *Int) UnmarshalJSON(data []byte) error { + o.Present = true + if string(data) == "null" { + o.Value = nil + return nil + } + var v int + if err := json.Unmarshal(data, &v); err != nil { + return err + } + o.Value = &v + return nil +} diff --git a/apps/backend/internal/integrations/optional/int_test.go b/apps/backend/internal/integrations/optional/int_test.go new file mode 100644 index 0000000000..8a83615f2a --- /dev/null +++ b/apps/backend/internal/integrations/optional/int_test.go @@ -0,0 +1,55 @@ +package optional + +import ( + "encoding/json" + "testing" +) + +func TestOptionalInt_UnmarshalJSON(t *testing.T) { + type payload struct { + Cap Int `json:"cap"` + } + + t.Run("absent key leaves Present false", func(t *testing.T) { + var p payload + if err := json.Unmarshal([]byte(`{"other":1}`), &p); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if p.Cap.Present { + t.Fatalf("expected Present=false for absent key, got Present=true") + } + if p.Cap.Value != nil { + t.Fatalf("expected nil Value for absent key, got %v", *p.Cap.Value) + } + }) + + t.Run("explicit null is present with nil value", func(t *testing.T) { + var p payload + if err := json.Unmarshal([]byte(`{"cap":null}`), &p); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !p.Cap.Present { + t.Fatalf("expected Present=true for explicit null") + } + if p.Cap.Value != nil { + t.Fatalf("expected nil Value for null, got %v", *p.Cap.Value) + } + }) + + t.Run("number is present with that value", func(t *testing.T) { + var p payload + if err := json.Unmarshal([]byte(`{"cap":7}`), &p); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if !p.Cap.Present || p.Cap.Value == nil || *p.Cap.Value != 7 { + t.Fatalf("expected Present=true Value=7, got Present=%v Value=%v", p.Cap.Present, p.Cap.Value) + } + }) + + t.Run("non-integer is an error", func(t *testing.T) { + var p payload + if err := json.Unmarshal([]byte(`{"cap":"nope"}`), &p); err == nil { + t.Fatalf("expected error for non-integer cap") + } + }) +} diff --git a/apps/backend/internal/jira/models.go b/apps/backend/internal/jira/models.go index 88b1a1a53d..b3465aa3dd 100644 --- a/apps/backend/internal/jira/models.go +++ b/apps/backend/internal/jira/models.go @@ -4,7 +4,11 @@ // expose these capabilities to the frontend. package jira -import "time" +import ( + "time" + + "github.com/kandev/kandev/internal/integrations/optional" +) // Auth method identifiers persisted in JiraConfig.AuthMethod. const ( @@ -156,19 +160,23 @@ const DefaultIssueWatchPollInterval = 300 // target workflow step's defaults determine where the resulting task runs, so // there's no `repos` column. type IssueWatch struct { - ID string `json:"id" db:"id"` - WorkspaceID string `json:"workspaceId" db:"workspace_id"` - WorkflowID string `json:"workflowId" db:"workflow_id"` - WorkflowStepID string `json:"workflowStepId" db:"workflow_step_id"` - JQL string `json:"jql" db:"jql"` - AgentProfileID string `json:"agentProfileId" db:"agent_profile_id"` - ExecutorProfileID string `json:"executorProfileId" db:"executor_profile_id"` - Prompt string `json:"prompt" db:"prompt"` - Enabled bool `json:"enabled" db:"enabled"` - PollIntervalSeconds int `json:"pollIntervalSeconds" db:"poll_interval_seconds"` - LastPolledAt *time.Time `json:"lastPolledAt,omitempty" db:"last_polled_at"` - CreatedAt time.Time `json:"createdAt" db:"created_at"` - UpdatedAt time.Time `json:"updatedAt" db:"updated_at"` + ID string `json:"id" db:"id"` + WorkspaceID string `json:"workspaceId" db:"workspace_id"` + WorkflowID string `json:"workflowId" db:"workflow_id"` + WorkflowStepID string `json:"workflowStepId" db:"workflow_step_id"` + JQL string `json:"jql" db:"jql"` + AgentProfileID string `json:"agentProfileId" db:"agent_profile_id"` + ExecutorProfileID string `json:"executorProfileId" db:"executor_profile_id"` + Prompt string `json:"prompt" db:"prompt"` + Enabled bool `json:"enabled" db:"enabled"` + PollIntervalSeconds int `json:"pollIntervalSeconds" db:"poll_interval_seconds"` + // MaxInflightTasks caps how many open watcher-created tasks this watch can + // hold at once. nil = uncapped. Values <= 0 are rejected at the API layer. + // See docs/specs/throttle-watcher-fanout/spec.md for the open-task definition. + MaxInflightTasks *int `json:"maxInflightTasks,omitempty" db:"max_inflight_tasks"` + LastPolledAt *time.Time `json:"lastPolledAt,omitempty" db:"last_polled_at"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` + UpdatedAt time.Time `json:"updatedAt" db:"updated_at"` } // IssueWatchTask deduplicates task creation per (watch, ticket) tuple. The @@ -187,14 +195,18 @@ type IssueWatchTask struct { // ticket matching a watch that has no existing dedup row. The orchestrator // consumes this to create (and optionally auto-start) a Kandev task. type NewJiraIssueEvent struct { - IssueWatchID string `json:"issueWatchId"` - WorkspaceID string `json:"workspaceId"` - WorkflowID string `json:"workflowId"` - WorkflowStepID string `json:"workflowStepId"` - AgentProfileID string `json:"agentProfileId"` - ExecutorProfileID string `json:"executorProfileId"` - Prompt string `json:"prompt"` - Issue *JiraTicket `json:"issue"` + IssueWatchID string `json:"issueWatchId"` + WorkspaceID string `json:"workspaceId"` + WorkflowID string `json:"workflowId"` + WorkflowStepID string `json:"workflowStepId"` + AgentProfileID string `json:"agentProfileId"` + ExecutorProfileID string `json:"executorProfileId"` + Prompt string `json:"prompt"` + // MaxInflightTasks mirrors the watch row's per-watcher throttle cap so the + // orchestrator's gate can read it without loading the row again. nil = + // uncapped. + MaxInflightTasks *int `json:"maxInflightTasks,omitempty"` + Issue *JiraTicket `json:"issue"` } // CreateIssueWatchRequest is the payload for POST /api/v1/jira/watches/issue. @@ -207,11 +219,14 @@ type CreateIssueWatchRequest struct { ExecutorProfileID string `json:"executorProfileId"` Prompt string `json:"prompt"` PollIntervalSeconds int `json:"pollIntervalSeconds"` + MaxInflightTasks *int `json:"maxInflightTasks,omitempty"` Enabled *bool `json:"enabled,omitempty"` } // UpdateIssueWatchRequest is the payload for PATCH /api/v1/jira/watches/issue/:id. -// All fields are pointers so the caller can omit ones it doesn't want to change. +// Most fields are pointers so callers can omit the ones they don't want to +// change. MaxInflightTasks uses optional.Int for tri-state PATCH semantics +// (absent = unchanged, null = uncapped, positive int = cap). type UpdateIssueWatchRequest struct { WorkflowID *string `json:"workflowId,omitempty"` WorkflowStepID *string `json:"workflowStepId,omitempty"` @@ -221,4 +236,8 @@ type UpdateIssueWatchRequest struct { Prompt *string `json:"prompt,omitempty"` Enabled *bool `json:"enabled,omitempty"` PollIntervalSeconds *int `json:"pollIntervalSeconds,omitempty"` + // MaxInflightTasks is tri-state so a partial PATCH that omits the field + // leaves the cap unchanged (a plain *int can't tell "omitted" from + // "null"). Absent = unchanged, null = uncapped, positive int = cap. + MaxInflightTasks optional.Int `json:"maxInflightTasks"` } diff --git a/apps/backend/internal/jira/service.go b/apps/backend/internal/jira/service.go index 3267db9f5c..71ad13be8e 100644 --- a/apps/backend/internal/jira/service.go +++ b/apps/backend/internal/jira/service.go @@ -511,6 +511,7 @@ func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchReq ExecutorProfileID: req.ExecutorProfileID, Prompt: req.Prompt, PollIntervalSeconds: req.PollIntervalSeconds, + MaxInflightTasks: req.MaxInflightTasks, Enabled: true, } if req.Enabled != nil { @@ -565,6 +566,9 @@ func (s *Service) UpdateIssueWatch(ctx context.Context, id string, req *UpdateIs if err := validatePollInterval(w.PollIntervalSeconds); err != nil { return nil, err } + if err := validateMaxInflightTasks(w.MaxInflightTasks); err != nil { + return nil, err + } if err := s.store.UpdateIssueWatch(ctx, w); err != nil { return nil, err } @@ -645,6 +649,7 @@ func (s *Service) publishNewJiraIssueEvent(ctx context.Context, w *IssueWatch, t AgentProfileID: w.AgentProfileID, ExecutorProfileID: w.ExecutorProfileID, Prompt: w.Prompt, + MaxInflightTasks: w.MaxInflightTasks, Issue: ticket, }) if err := eb.Publish(ctx, events.JiraNewIssue, evt); err != nil { @@ -684,6 +689,21 @@ func validateIssueWatchCreate(req *CreateIssueWatchRequest) error { return err } } + if err := validateMaxInflightTasks(req.MaxInflightTasks); err != nil { + return err + } + return nil +} + +// validateMaxInflightTasks rejects non-positive caps. A nil pointer is +// "uncapped" and explicitly allowed. +func validateMaxInflightTasks(v *int) error { + if v == nil { + return nil + } + if *v <= 0 { + return fmt.Errorf("%w: maxInflightTasks must be a positive integer", ErrInvalidConfig) + } return nil } @@ -720,4 +740,10 @@ func applyIssueWatchPatch(w *IssueWatch, req *UpdateIssueWatchRequest) { if req.PollIntervalSeconds != nil { w.PollIntervalSeconds = *req.PollIntervalSeconds } + // MaxInflightTasks is tri-state (optional.Int): only apply it when the + // PATCH actually carried the key, so a partial update that omits it + // leaves the cap intact. Present+null clears the cap; present+int sets it. + if req.MaxInflightTasks.Present { + w.MaxInflightTasks = req.MaxInflightTasks.Value + } } diff --git a/apps/backend/internal/jira/service_issue_watch_test.go b/apps/backend/internal/jira/service_issue_watch_test.go index 4ba69c0ae6..5eeb359aa5 100644 --- a/apps/backend/internal/jira/service_issue_watch_test.go +++ b/apps/backend/internal/jira/service_issue_watch_test.go @@ -4,6 +4,8 @@ import ( "context" "errors" "testing" + + "github.com/kandev/kandev/internal/integrations/optional" ) // withSearchResults returns a fakeClient that always returns the given tickets @@ -46,6 +48,99 @@ func TestService_CreateIssueWatch_DefaultsAndValidation(t *testing.T) { } } +func TestService_IssueWatch_MaxInflightTasks(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + // Default create (no cap) persists NULL → reads back nil (uncapped); the + // column default of 5 must not leak through the explicit-NULL INSERT. + uncapped, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + JQL: "project = PROJ", + }) + if err != nil { + t.Fatalf("create uncapped: %v", err) + } + if uncapped.MaxInflightTasks != nil { + t.Fatalf("expected nil (uncapped), got %v", *uncapped.MaxInflightTasks) + } + got, err := f.svc.GetIssueWatch(ctx, uncapped.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.MaxInflightTasks != nil { + t.Fatalf("uncapped did not round-trip as nil: %v", *got.MaxInflightTasks) + } + + // Positive cap round-trips. + cap5 := 5 + capped, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + JQL: "project = PROJ", MaxInflightTasks: &cap5, + }) + if err != nil { + t.Fatalf("create capped: %v", err) + } + if capped.MaxInflightTasks == nil || *capped.MaxInflightTasks != 5 { + t.Fatalf("cap not persisted: %v", capped.MaxInflightTasks) + } + + // Non-positive caps rejected on create. + for _, bad := range []int{0, -1} { + b := bad + if _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + JQL: "project = PROJ", MaxInflightTasks: &b, + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for cap=%d, got %v", bad, err) + } + } + + // PATCH is tri-state: present+int sets, present+null clears, absent leaves + // the cap unchanged. The absent case must NOT silently drop the cap. + cap3 := 3 + updated, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: &cap3}, + }) + if err != nil { + t.Fatalf("update set cap: %v", err) + } + if updated.MaxInflightTasks == nil || *updated.MaxInflightTasks != 3 { + t.Fatalf("cap not updated: %v", updated.MaxInflightTasks) + } + + // Partial PATCH that omits MaxInflightTasks preserves the existing cap. + newPrompt := "changed" + preserved, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + Prompt: &newPrompt, + }) + if err != nil { + t.Fatalf("partial update: %v", err) + } + if preserved.MaxInflightTasks == nil || *preserved.MaxInflightTasks != 3 { + t.Fatalf("partial PATCH wrongly cleared the cap: %v", preserved.MaxInflightTasks) + } + + // Explicit null clears back to uncapped. + cleared, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: nil}, + }) + if err != nil { + t.Fatalf("update clear cap: %v", err) + } + if cleared.MaxInflightTasks != nil { + t.Fatalf("expected cap cleared to nil, got %v", *cleared.MaxInflightTasks) + } + + // Non-positive cap rejected on update. + zero := 0 + if _, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: &zero}, + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for update cap=0, got %v", err) + } +} + func TestService_UpdateIssueWatch_PartialPatch(t *testing.T) { f := newSvcFixture(t) ctx := context.Background() diff --git a/apps/backend/internal/jira/store.go b/apps/backend/internal/jira/store.go index ceb2921aeb..af70ca9091 100644 --- a/apps/backend/internal/jira/store.go +++ b/apps/backend/internal/jira/store.go @@ -67,6 +67,10 @@ const createTablesSQL = ` prompt TEXT NOT NULL DEFAULT '', enabled BOOLEAN NOT NULL DEFAULT 1, poll_interval_seconds INTEGER NOT NULL DEFAULT 300, + -- Cap on concurrent open watcher-created tasks for this watch. + -- NULL = uncapped. Positive integer = cap. Values <= 0 are rejected at + -- the API layer. See docs/specs/throttle-watcher-fanout/. + max_inflight_tasks INTEGER DEFAULT 5, last_polled_at DATETIME, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL @@ -101,6 +105,30 @@ func (s *Store) initSchema() error { if err := s.addInstanceTypeColumn(); err != nil { return err } + if err := s.addMaxInflightTasksColumn(); err != nil { + return err + } + return nil +} + +// addMaxInflightTasksColumn brings older databases up to the current schema by +// adding the max_inflight_tasks column to jira_issue_watches when missing. +// Existing rows backfill to the default (5). A fresh install hits the +// column-already-present branch since createTablesSQL declares the column. +func (s *Store) addMaxInflightTasksColumn() error { + cols, err := s.tableColumns("jira_issue_watches") + if err != nil { + return err + } + if len(cols) == 0 { + return nil + } + if _, ok := cols["max_inflight_tasks"]; ok { + return nil + } + if _, err := s.db.Exec(`ALTER TABLE jira_issue_watches ADD COLUMN max_inflight_tasks INTEGER DEFAULT 5`); err != nil { + return fmt.Errorf("add max_inflight_tasks column: %w", err) + } return nil } @@ -341,7 +369,17 @@ func (s *Store) UpdateAuthHealth(ctx context.Context, ok bool, errMsg string, ch const issueWatchColumns = `id, workspace_id, workflow_id, workflow_step_id, jql, agent_profile_id, executor_profile_id, prompt, enabled, - poll_interval_seconds, last_polled_at, created_at, updated_at` + poll_interval_seconds, max_inflight_tasks, last_polled_at, created_at, updated_at` + +// nullableInt converts a *int into a value suitable for a nullable SQL column. +// A nil pointer becomes a SQL NULL; a non-nil pointer becomes the underlying +// int. Used for max_inflight_tasks where NULL means "uncapped". +func nullableInt(v *int) interface{} { + if v == nil { + return nil + } + return *v +} // CreateIssueWatch persists a new issue watch row. ID and timestamps are // assigned here so callers can pass a partially-populated struct. @@ -357,10 +395,11 @@ func (s *Store) CreateIssueWatch(ctx context.Context, w *IssueWatch) error { } _, err := s.db.ExecContext(ctx, ` INSERT INTO jira_issue_watches (`+issueWatchColumns+`) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, w.ID, w.WorkspaceID, w.WorkflowID, w.WorkflowStepID, w.JQL, w.AgentProfileID, w.ExecutorProfileID, w.Prompt, w.Enabled, - w.PollIntervalSeconds, w.LastPolledAt, w.CreatedAt, w.UpdatedAt) + w.PollIntervalSeconds, nullableInt(w.MaxInflightTasks), + w.LastPolledAt, w.CreatedAt, w.UpdatedAt) return err } @@ -429,11 +468,13 @@ func (s *Store) UpdateIssueWatch(ctx context.Context, w *IssueWatch) error { _, err := s.db.ExecContext(ctx, ` UPDATE jira_issue_watches SET workflow_id = ?, workflow_step_id = ?, jql = ?, agent_profile_id = ?, executor_profile_id = ?, prompt = ?, - enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ? + enabled = ?, poll_interval_seconds = ?, max_inflight_tasks = ?, + last_polled_at = ?, updated_at = ? WHERE id = ?`, w.WorkflowID, w.WorkflowStepID, w.JQL, w.AgentProfileID, w.ExecutorProfileID, w.Prompt, - w.Enabled, w.PollIntervalSeconds, w.LastPolledAt, w.UpdatedAt, w.ID) + w.Enabled, w.PollIntervalSeconds, nullableInt(w.MaxInflightTasks), + w.LastPolledAt, w.UpdatedAt, w.ID) return err } diff --git a/apps/backend/internal/linear/models.go b/apps/backend/internal/linear/models.go index 0e1bc42c31..bf44cf72ee 100644 --- a/apps/backend/internal/linear/models.go +++ b/apps/backend/internal/linear/models.go @@ -3,7 +3,11 @@ // and WebSocket handlers that expose these capabilities to the frontend. package linear -import "time" +import ( + "time" + + "github.com/kandev/kandev/internal/integrations/optional" +) // AuthMethodAPIKey is the only auth method Linear supports today: a Personal // API Key sent as the `Authorization` header (no Bearer prefix). The constant @@ -194,9 +198,13 @@ type IssueWatch struct { Prompt string `json:"prompt" db:"prompt"` Enabled bool `json:"enabled" db:"enabled"` PollIntervalSeconds int `json:"pollIntervalSeconds" db:"poll_interval_seconds"` - LastPolledAt *time.Time `json:"lastPolledAt,omitempty" db:"last_polled_at"` - CreatedAt time.Time `json:"createdAt" db:"created_at"` - UpdatedAt time.Time `json:"updatedAt" db:"updated_at"` + // MaxInflightTasks caps how many open watcher-created tasks this watch can + // hold at once. nil = uncapped. Values <= 0 are rejected at the API layer. + // See docs/specs/throttle-watcher-fanout/spec.md for the open-task definition. + MaxInflightTasks *int `json:"maxInflightTasks,omitempty" db:"max_inflight_tasks"` + LastPolledAt *time.Time `json:"lastPolledAt,omitempty" db:"last_polled_at"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` + UpdatedAt time.Time `json:"updatedAt" db:"updated_at"` } // IssueWatchTask deduplicates task creation per (watch, issue) tuple. The @@ -217,14 +225,18 @@ type IssueWatchTask struct { // issue matching a watch that has no existing dedup row. The orchestrator // consumes this to create (and optionally auto-start) a Kandev task. type NewLinearIssueEvent struct { - IssueWatchID string `json:"issueWatchId"` - WorkspaceID string `json:"workspaceId"` - WorkflowID string `json:"workflowId"` - WorkflowStepID string `json:"workflowStepId"` - AgentProfileID string `json:"agentProfileId"` - ExecutorProfileID string `json:"executorProfileId"` - Prompt string `json:"prompt"` - Issue *LinearIssue `json:"issue"` + IssueWatchID string `json:"issueWatchId"` + WorkspaceID string `json:"workspaceId"` + WorkflowID string `json:"workflowId"` + WorkflowStepID string `json:"workflowStepId"` + AgentProfileID string `json:"agentProfileId"` + ExecutorProfileID string `json:"executorProfileId"` + Prompt string `json:"prompt"` + // MaxInflightTasks mirrors the watch row's per-watcher throttle cap so the + // orchestrator's gate can read it without loading the row again. nil = + // uncapped. + MaxInflightTasks *int `json:"maxInflightTasks,omitempty"` + Issue *LinearIssue `json:"issue"` } // CreateIssueWatchRequest is the payload for POST /api/v1/linear/watches/issue. @@ -237,11 +249,14 @@ type CreateIssueWatchRequest struct { ExecutorProfileID string `json:"executorProfileId"` Prompt string `json:"prompt"` PollIntervalSeconds int `json:"pollIntervalSeconds"` + MaxInflightTasks *int `json:"maxInflightTasks,omitempty"` Enabled *bool `json:"enabled,omitempty"` } // UpdateIssueWatchRequest is the payload for PATCH /api/v1/linear/watches/issue/:id. -// All fields are pointers so the caller can omit ones it doesn't want to change. +// Most fields are pointers so callers can omit the ones they don't want to +// change. MaxInflightTasks uses optional.Int for tri-state PATCH semantics +// (absent = unchanged, null = uncapped, positive int = cap). type UpdateIssueWatchRequest struct { WorkflowID *string `json:"workflowId,omitempty"` WorkflowStepID *string `json:"workflowStepId,omitempty"` @@ -251,4 +266,8 @@ type UpdateIssueWatchRequest struct { Prompt *string `json:"prompt,omitempty"` Enabled *bool `json:"enabled,omitempty"` PollIntervalSeconds *int `json:"pollIntervalSeconds,omitempty"` + // MaxInflightTasks is tri-state so a partial PATCH that omits the field + // leaves the cap unchanged (a plain *int can't tell "omitted" from + // "null"). Absent = unchanged, null = uncapped, positive int = cap. + MaxInflightTasks optional.Int `json:"maxInflightTasks"` } diff --git a/apps/backend/internal/linear/service_issue_watch.go b/apps/backend/internal/linear/service_issue_watch.go index 6875534a1a..7d3526045f 100644 --- a/apps/backend/internal/linear/service_issue_watch.go +++ b/apps/backend/internal/linear/service_issue_watch.go @@ -40,6 +40,7 @@ func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchReq ExecutorProfileID: req.ExecutorProfileID, Prompt: req.Prompt, PollIntervalSeconds: req.PollIntervalSeconds, + MaxInflightTasks: req.MaxInflightTasks, Enabled: true, } if req.Enabled != nil { @@ -93,6 +94,9 @@ func (s *Service) UpdateIssueWatch(ctx context.Context, id string, req *UpdateIs if err := validatePollInterval(w.PollIntervalSeconds); err != nil { return nil, err } + if err := validateMaxInflightTasks(w.MaxInflightTasks); err != nil { + return nil, err + } if err := s.store.UpdateIssueWatch(ctx, w); err != nil { return nil, err } @@ -174,6 +178,7 @@ func (s *Service) publishNewLinearIssueEvent(ctx context.Context, w *IssueWatch, AgentProfileID: w.AgentProfileID, ExecutorProfileID: w.ExecutorProfileID, Prompt: w.Prompt, + MaxInflightTasks: w.MaxInflightTasks, Issue: issue, }) if err := eb.Publish(ctx, events.LinearNewIssue, evt); err != nil { @@ -212,6 +217,21 @@ func validateIssueWatchCreate(req *CreateIssueWatchRequest) error { return err } } + if err := validateMaxInflightTasks(req.MaxInflightTasks); err != nil { + return err + } + return nil +} + +// validateMaxInflightTasks rejects non-positive caps. A nil pointer is +// "uncapped" and explicitly allowed. +func validateMaxInflightTasks(v *int) error { + if v == nil { + return nil + } + if *v <= 0 { + return fmt.Errorf("%w: maxInflightTasks must be a positive integer", ErrInvalidConfig) + } return nil } @@ -329,4 +349,10 @@ func applyIssueWatchPatch(w *IssueWatch, req *UpdateIssueWatchRequest) { if req.PollIntervalSeconds != nil { w.PollIntervalSeconds = *req.PollIntervalSeconds } + // MaxInflightTasks is tri-state (optional.Int): only apply it when the + // PATCH actually carried the key, so a partial update that omits it + // leaves the cap intact. Present+null clears the cap; present+int sets it. + if req.MaxInflightTasks.Present { + w.MaxInflightTasks = req.MaxInflightTasks.Value + } } diff --git a/apps/backend/internal/linear/service_issue_watch_test.go b/apps/backend/internal/linear/service_issue_watch_test.go index 1ddb0a09fa..b6f0f1032e 100644 --- a/apps/backend/internal/linear/service_issue_watch_test.go +++ b/apps/backend/internal/linear/service_issue_watch_test.go @@ -5,6 +5,8 @@ import ( "errors" "math" "testing" + + "github.com/kandev/kandev/internal/integrations/optional" ) func TestService_CreateIssueWatch_AcceptsRichFilters(t *testing.T) { @@ -124,6 +126,102 @@ func TestService_UpdateIssueWatch_PartialPatch(t *testing.T) { } } +func TestService_IssueWatch_MaxInflightTasks(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + // Default create (no cap supplied) persists NULL → reads back as nil + // (uncapped). The store column default of 5 must NOT leak through the + // INSERT path, which names the column explicitly with a NULL bind. + uncapped, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }) + if err != nil { + t.Fatalf("create uncapped: %v", err) + } + if uncapped.MaxInflightTasks != nil { + t.Fatalf("expected nil (uncapped), got %v", *uncapped.MaxInflightTasks) + } + got, err := f.svc.GetIssueWatch(ctx, uncapped.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got.MaxInflightTasks != nil { + t.Fatalf("uncapped did not round-trip as nil: %v", *got.MaxInflightTasks) + } + + // Create with a positive cap round-trips through the nullable column. + cap5 := 5 + capped, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, MaxInflightTasks: &cap5, + }) + if err != nil { + t.Fatalf("create capped: %v", err) + } + if capped.MaxInflightTasks == nil || *capped.MaxInflightTasks != 5 { + t.Fatalf("cap not persisted: %v", capped.MaxInflightTasks) + } + + // Non-positive caps are rejected on create. + for _, bad := range []int{0, -1} { + b := bad + if _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, MaxInflightTasks: &b, + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for cap=%d, got %v", bad, err) + } + } + + // PATCH is tri-state: present+int sets the cap, present+null clears it, + // absent leaves it unchanged. The absent case is the footgun greptile/the + // local reviewer flagged — a partial update must NOT silently drop the cap. + cap3 := 3 + updated, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: &cap3}, + }) + if err != nil { + t.Fatalf("update set cap: %v", err) + } + if updated.MaxInflightTasks == nil || *updated.MaxInflightTasks != 3 { + t.Fatalf("cap not updated: %v", updated.MaxInflightTasks) + } + + // Partial PATCH that omits MaxInflightTasks (Present=false) must preserve + // the existing cap of 3 — not reset it to uncapped. + newPrompt := "changed" + preserved, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + Prompt: &newPrompt, + }) + if err != nil { + t.Fatalf("partial update: %v", err) + } + if preserved.MaxInflightTasks == nil || *preserved.MaxInflightTasks != 3 { + t.Fatalf("partial PATCH wrongly cleared the cap: %v", preserved.MaxInflightTasks) + } + + // Explicit null clears the cap back to uncapped. + cleared, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: nil}, + }) + if err != nil { + t.Fatalf("update clear cap: %v", err) + } + if cleared.MaxInflightTasks != nil { + t.Fatalf("expected cap cleared to nil, got %v", *cleared.MaxInflightTasks) + } + + // Non-positive cap is rejected on update too. + zero := 0 + if _, err := f.svc.UpdateIssueWatch(ctx, capped.ID, &UpdateIssueWatchRequest{ + MaxInflightTasks: optional.Int{Present: true, Value: &zero}, + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for update cap=0, got %v", err) + } +} + func TestValidateFilterBounds_RejectsNonFiniteAndNegativeEstimates(t *testing.T) { nan := math.NaN() posInf := math.Inf(1) diff --git a/apps/backend/internal/linear/store.go b/apps/backend/internal/linear/store.go index 0e913e4b5a..8e9ace8354 100644 --- a/apps/backend/internal/linear/store.go +++ b/apps/backend/internal/linear/store.go @@ -63,6 +63,10 @@ const createTablesSQL = ` prompt TEXT NOT NULL DEFAULT '', enabled BOOLEAN NOT NULL DEFAULT 1, poll_interval_seconds INTEGER NOT NULL DEFAULT 300, + -- Cap on concurrent open watcher-created tasks for this watch. + -- NULL = uncapped. Positive integer = cap. Values <= 0 are rejected at + -- the API layer. See docs/specs/throttle-watcher-fanout/. + max_inflight_tasks INTEGER DEFAULT 5, last_polled_at DATETIME, created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL @@ -92,6 +96,30 @@ func (s *Store) initSchema() error { if _, err := s.db.Exec(createTablesSQL); err != nil { return err } + if err := s.addMaxInflightTasksColumn(); err != nil { + return err + } + return nil +} + +// addMaxInflightTasksColumn brings older databases up to the current schema by +// adding the max_inflight_tasks column to linear_issue_watches when missing. +// Existing rows backfill to the default (5). A fresh install hits the +// column-already-present branch since createTablesSQL declares the column. +func (s *Store) addMaxInflightTasksColumn() error { + cols, err := s.tableColumns("linear_issue_watches") + if err != nil { + return err + } + if len(cols) == 0 { + return nil + } + if _, ok := cols["max_inflight_tasks"]; ok { + return nil + } + if _, err := s.db.Exec(`ALTER TABLE linear_issue_watches ADD COLUMN max_inflight_tasks INTEGER DEFAULT 5`); err != nil { + return fmt.Errorf("add max_inflight_tasks column: %w", err) + } return nil } diff --git a/apps/backend/internal/linear/store_issue_watch.go b/apps/backend/internal/linear/store_issue_watch.go index 3219704433..298ae65496 100644 --- a/apps/backend/internal/linear/store_issue_watch.go +++ b/apps/backend/internal/linear/store_issue_watch.go @@ -15,19 +15,20 @@ import ( // the way SQLite holds it. The store marshals/unmarshals at this boundary so // service callers see a typed SearchFilter and never have to think about JSON. type issueWatchRow struct { - ID string `db:"id"` - WorkspaceID string `db:"workspace_id"` - WorkflowID string `db:"workflow_id"` - WorkflowStepID string `db:"workflow_step_id"` - FilterJSON string `db:"filter_json"` - AgentProfileID string `db:"agent_profile_id"` - ExecutorProfileID string `db:"executor_profile_id"` - Prompt string `db:"prompt"` - Enabled bool `db:"enabled"` - PollIntervalSeconds int `db:"poll_interval_seconds"` - LastPolledAt *time.Time `db:"last_polled_at"` - CreatedAt time.Time `db:"created_at"` - UpdatedAt time.Time `db:"updated_at"` + ID string `db:"id"` + WorkspaceID string `db:"workspace_id"` + WorkflowID string `db:"workflow_id"` + WorkflowStepID string `db:"workflow_step_id"` + FilterJSON string `db:"filter_json"` + AgentProfileID string `db:"agent_profile_id"` + ExecutorProfileID string `db:"executor_profile_id"` + Prompt string `db:"prompt"` + Enabled bool `db:"enabled"` + PollIntervalSeconds int `db:"poll_interval_seconds"` + MaxInflightTasks sql.NullInt64 `db:"max_inflight_tasks"` + LastPolledAt *time.Time `db:"last_polled_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` } func (r *issueWatchRow) toIssueWatch() (*IssueWatch, error) { @@ -37,6 +38,11 @@ func (r *issueWatchRow) toIssueWatch() (*IssueWatch, error) { return nil, fmt.Errorf("decode filter: %w", err) } } + var maxInflight *int + if r.MaxInflightTasks.Valid { + v := int(r.MaxInflightTasks.Int64) + maxInflight = &v + } return &IssueWatch{ ID: r.ID, WorkspaceID: r.WorkspaceID, @@ -48,6 +54,7 @@ func (r *issueWatchRow) toIssueWatch() (*IssueWatch, error) { Prompt: r.Prompt, Enabled: r.Enabled, PollIntervalSeconds: r.PollIntervalSeconds, + MaxInflightTasks: maxInflight, LastPolledAt: r.LastPolledAt, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt, @@ -64,7 +71,7 @@ func encodeFilter(f SearchFilter) (string, error) { const issueWatchColumns = `id, workspace_id, workflow_id, workflow_step_id, filter_json, agent_profile_id, executor_profile_id, prompt, enabled, - poll_interval_seconds, last_polled_at, created_at, updated_at` + poll_interval_seconds, max_inflight_tasks, last_polled_at, created_at, updated_at` // CreateIssueWatch persists a new issue watch row. ID and timestamps are // assigned here so callers can pass a partially-populated struct. @@ -84,13 +91,23 @@ func (s *Store) CreateIssueWatch(ctx context.Context, w *IssueWatch) error { } _, err = s.db.ExecContext(ctx, ` INSERT INTO linear_issue_watches (`+issueWatchColumns+`) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, w.ID, w.WorkspaceID, w.WorkflowID, w.WorkflowStepID, filterJSON, w.AgentProfileID, w.ExecutorProfileID, w.Prompt, w.Enabled, - w.PollIntervalSeconds, w.LastPolledAt, w.CreatedAt, w.UpdatedAt) + w.PollIntervalSeconds, nullableInt(w.MaxInflightTasks), w.LastPolledAt, w.CreatedAt, w.UpdatedAt) return err } +// nullableInt converts a *int into a value suitable for a nullable SQL column. +// A nil pointer becomes a SQL NULL; a non-nil pointer becomes the underlying +// int. Used for max_inflight_tasks where NULL means "uncapped". +func nullableInt(v *int) interface{} { + if v == nil { + return nil + } + return *v +} + // GetIssueWatch returns a single watch by ID, or nil when no row matches. func (s *Store) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) { var row issueWatchRow @@ -168,11 +185,13 @@ func (s *Store) UpdateIssueWatch(ctx context.Context, w *IssueWatch) error { _, err = s.db.ExecContext(ctx, ` UPDATE linear_issue_watches SET workflow_id = ?, workflow_step_id = ?, filter_json = ?, agent_profile_id = ?, executor_profile_id = ?, prompt = ?, - enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ? + enabled = ?, poll_interval_seconds = ?, max_inflight_tasks = ?, + last_polled_at = ?, updated_at = ? WHERE id = ?`, w.WorkflowID, w.WorkflowStepID, filterJSON, w.AgentProfileID, w.ExecutorProfileID, w.Prompt, - w.Enabled, w.PollIntervalSeconds, w.LastPolledAt, w.UpdatedAt, w.ID) + w.Enabled, w.PollIntervalSeconds, nullableInt(w.MaxInflightTasks), + w.LastPolledAt, w.UpdatedAt, w.ID) return err } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 0af2bd96d1..39790ee72b 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -267,6 +267,20 @@ type Service struct { // Constructed lazily once issueTaskCreator is wired (see SetIssueTaskCreator). watcherCoordinator *WatcherDispatchCoordinator + // Watcher throttle state. watcherTaskCount counts committed open watcher- + // created tasks (DB-backed source of truth across restarts). pendingByWatch + // tracks in-process events whose dedup row has not yet been written — + // without it, a burst of events read the same stale COUNT(*) before any + // goroutine commits, and the cap is silently overshot. Both reads and the + // pending increment happen under watcherMu to prevent the burst race. + watcherTaskCount WatcherTaskCounter + watcherMu sync.Mutex + pendingByWatch map[string]int + // watcherSaturated tracks per-watch whether the last gate result was + // "deferred", so we log the state-transition Warn ("cap reached" / + // "cap cleared") only once per transition instead of every event. + watcherSaturated map[string]bool + // Jira service for issue watch dedup operations jiraService JiraService // jiraSource adapts jiraService onto WatcherSource. Built once in diff --git a/apps/backend/internal/orchestrator/source_jira.go b/apps/backend/internal/orchestrator/source_jira.go index 032cea3442..390ec6b79b 100644 --- a/apps/backend/internal/orchestrator/source_jira.go +++ b/apps/backend/internal/orchestrator/source_jira.go @@ -82,6 +82,22 @@ func (s *JiraWatcherSource) AttachTaskID(ctx context.Context, evt any, taskID st return s.service.AssignIssueWatchTaskID(ctx, e.IssueWatchID, e.Issue.Key, taskID) } +func (s *JiraWatcherSource) WatchID(evt any) string { + e, ok := evt.(*jira.NewJiraIssueEvent) + if !ok || e == nil { + return "" + } + return e.IssueWatchID +} + +func (s *JiraWatcherSource) MaxInflightTasks(evt any) *int { + e, ok := evt.(*jira.NewJiraIssueEvent) + if !ok || e == nil { + return nil + } + return e.MaxInflightTasks +} + func (s *JiraWatcherSource) AutoStartParams(evt any) AutoStartParams { e, ok := evt.(*jira.NewJiraIssueEvent) if !ok || e == nil { diff --git a/apps/backend/internal/orchestrator/source_linear.go b/apps/backend/internal/orchestrator/source_linear.go index 1f7df9c11b..413b1c2f8b 100644 --- a/apps/backend/internal/orchestrator/source_linear.go +++ b/apps/backend/internal/orchestrator/source_linear.go @@ -83,6 +83,22 @@ func (s *LinearWatcherSource) AttachTaskID(ctx context.Context, evt any, taskID return s.service.AssignIssueWatchTaskID(ctx, e.IssueWatchID, e.Issue.Identifier, taskID) } +func (s *LinearWatcherSource) WatchID(evt any) string { + e, ok := evt.(*linear.NewLinearIssueEvent) + if !ok || e == nil { + return "" + } + return e.IssueWatchID +} + +func (s *LinearWatcherSource) MaxInflightTasks(evt any) *int { + e, ok := evt.(*linear.NewLinearIssueEvent) + if !ok || e == nil { + return nil + } + return e.MaxInflightTasks +} + func (s *LinearWatcherSource) AutoStartParams(evt any) AutoStartParams { e, ok := evt.(*linear.NewLinearIssueEvent) if !ok || e == nil { diff --git a/apps/backend/internal/orchestrator/watcher_dispatch.go b/apps/backend/internal/orchestrator/watcher_dispatch.go index 23fd48372e..8bc3f24951 100644 --- a/apps/backend/internal/orchestrator/watcher_dispatch.go +++ b/apps/backend/internal/orchestrator/watcher_dispatch.go @@ -96,6 +96,17 @@ type WatcherSource interface { // AutoStartParams returns the parameters needed to kick the task off // when its workflow step is configured for auto-start. AutoStartParams(evt any) AutoStartParams + + // WatchID extracts the per-integration watch identifier from the event + // payload. Used by the throttle gate to key the per-watch pending + // counter and look up the per-watch cap. Returns "" if the event has + // no watch (the gate then treats it as unthrottled). + WatchID(evt any) string + + // MaxInflightTasks returns the per-watch cap on open watcher-created + // tasks, or nil when the watch is uncapped. The orchestrator gate uses + // this to decide whether to defer the event. + MaxInflightTasks(evt any) *int } // Dispatch runs one event through the full pipeline. Safe to call from a diff --git a/apps/backend/internal/orchestrator/watcher_dispatch_test.go b/apps/backend/internal/orchestrator/watcher_dispatch_test.go index aec1009af6..3bc16f412e 100644 --- a/apps/backend/internal/orchestrator/watcher_dispatch_test.go +++ b/apps/backend/internal/orchestrator/watcher_dispatch_test.go @@ -33,6 +33,8 @@ type fakeWatcherSource struct { buildErr error attachErr error autoStart AutoStartParams + watchID string + maxInflightTasks *int recordedReserve int recordedRelease int recordedBuild int @@ -67,6 +69,10 @@ func (f *fakeWatcherSource) AutoStartParams(_ any) AutoStartParams { return p } +func (f *fakeWatcherSource) WatchID(_ any) string { return f.watchID } + +func (f *fakeWatcherSource) MaxInflightTasks(_ any) *int { return f.maxInflightTasks } + // fakeTaskCreator captures the request and returns a canned task or error. type fakeTaskCreator struct { createErr error diff --git a/apps/backend/internal/orchestrator/watcher_dispatch_wiring.go b/apps/backend/internal/orchestrator/watcher_dispatch_wiring.go index 59201cb845..77b7853493 100644 --- a/apps/backend/internal/orchestrator/watcher_dispatch_wiring.go +++ b/apps/backend/internal/orchestrator/watcher_dispatch_wiring.go @@ -65,7 +65,28 @@ func (s *Service) dispatchWatcherEvent(ctx context.Context, integration string, s.logger.Warn(fmt.Sprintf("watcher coordinator not configured, skipping %s task creation", integration)) return } + + // Per-watcher throttle. Synchronous: we MUST acquire the slot before + // spawning the goroutine, otherwise a tight burst of bus events all + // read the same pre-creation DB count and overshoot the cap. See + // docs/specs/throttle-watcher-fanout/spec.md. + // + // The slot covers the WHOLE dispatch pipeline, including the case where + // Dispatch returns early because Reserve loses the dedup race (no task + // created). In that window the pending counter briefly holds a slot that + // maps to no task, so a poll/`/trigger` collision can momentarily + // over-throttle a watch by one. The slot is released when the goroutine + // exits and the DB count is unaffected, so this self-heals on the next + // tick — acceptable for v1. + release, ok := s.acquireWatcherSlot(ctx, integration, src.WatchID(evt), src.MaxInflightTasks(evt)) + if !ok { + return + } + // Detach from cancellation but keep request-scoped values (tracing, etc.): // the bus delivery context may be cancelled before task creation finishes. - go coordinator.Dispatch(context.WithoutCancel(ctx), src, evt) + go func() { + defer release() + coordinator.Dispatch(context.WithoutCancel(ctx), src, evt) + }() } diff --git a/apps/backend/internal/orchestrator/watcher_throttle.go b/apps/backend/internal/orchestrator/watcher_throttle.go new file mode 100644 index 0000000000..db65dc0ad2 --- /dev/null +++ b/apps/backend/internal/orchestrator/watcher_throttle.go @@ -0,0 +1,179 @@ +package orchestrator + +import ( + "context" + "expvar" + "strings" + + "go.uber.org/zap" +) + +// WatcherTaskCounter exposes the count of open watcher-created tasks for a +// single (integration, watchID) pair. Implemented by the task repository. +// Kept as a narrow interface so the orchestrator's throttle gate can be +// tested without spinning up SQLite. +type WatcherTaskCounter interface { + CountOpenWatcherCreatedTasks(ctx context.Context, integration, watchID string) (int, error) +} + +// SetWatcherTaskCounter wires the open-task counter used by the watcher +// throttle gate. Safe to call once during boot. When unset, the gate falls +// open (no throttling) — the orchestrator must remain usable in tests and +// during early init before the task repository is plumbed through. +func (s *Service) SetWatcherTaskCounter(c WatcherTaskCounter) { + s.watcherTaskCount = c +} + +// Watcher throttle metrics, exposed via stdlib's /debug/vars handler. Counter +// labels match the office routing metrics format ("k=v;k=v") so a downstream +// Prometheus translator can read both with the same parser. Saturation is +// recorded as a counter of state-transition events (cap_reached / +// cap_cleared); a live "currently saturated" gauge would drift across +// restarts so we stick with deltas. +var ( + watcherDispatchAttemptedTotal = expvar.NewMap("watcher_dispatch_attempted_total") + watcherDispatchDeferredTotal = expvar.NewMap("watcher_dispatch_deferred_total") + watcherCapTransitionsTotal = expvar.NewMap("watcher_cap_transitions_total") +) + +// watcherMetricLabel builds a "k1=v1;k2=v2" label string for an expvar map +// key. Kept package-local so it doesn't clash with the office package's +// metricLabel helper. +func watcherMetricLabel(pairs ...string) string { + if len(pairs)%2 != 0 { + return "" + } + parts := make([]string, 0, len(pairs)/2) + for i := 0; i < len(pairs); i += 2 { + parts = append(parts, pairs[i]+"="+pairs[i+1]) + } + return strings.Join(parts, ";") +} + +// watcherSlotKey is the (integration, watchID) compound key used to scope +// the pending counter and the saturation state map. Linear and Jira watches +// can share id strings; the integration prefix keeps them separate. +func watcherSlotKey(integration, watchID string) string { + return integration + "|" + watchID +} + +// acquireWatcherSlot is the synchronous slot acquisition that gates the +// `go coordinator.Dispatch(...)` spawn. It is the front-pressure valve for +// the Linear and Jira watchers: with a per-watch cap, an unexpectedly broad +// filter cannot fan out into dozens of concurrent tasks on a single poll +// tick. +// +// Returns (release, ok). When ok is true the caller MUST invoke release() +// exactly once (typically via `defer release()` in the dispatch goroutine) +// to drop the in-process pending count. When ok is false the caller MUST +// drop the event — no goroutine, no dedup row — so the next poll tick can +// retry. +// +// Bypass paths (all return release=no-op, ok=true): +// +// - cap is nil (uncapped watch) +// - cap is <= 0 (defensive: API rejects these, but a stale row should not +// freeze the gate) +// - watchID is empty (event has no watch — treat as unthrottled) +// - watcherTaskCount is unset (early boot / tests) +// - the count query errors (fail-open: a transient DB blip must not +// silently stall the watcher; one-event overshoot is acceptable) +// +// Throttling path: when count(db) + pending(in-process) >= cap, the gate +// defers. Pending is incremented atomically with the read under watcherMu +// so a burst of events arriving in the same poll tick cannot collectively +// read the same stale DB count and all pass. +func (s *Service) acquireWatcherSlot(ctx context.Context, integration, watchID string, maxInflight *int) (func(), bool) { + watcherDispatchAttemptedTotal.Add(watcherMetricLabel("integration", integration), 1) + noop := func() {} + + if maxInflight == nil || *maxInflight <= 0 || watchID == "" { + return noop, true + } + if s.watcherTaskCount == nil { + return noop, true + } + + count, err := s.watcherTaskCount.CountOpenWatcherCreatedTasks(ctx, integration, watchID) + if err != nil { + s.logger.Warn("watcher throttle: count failed, failing open", + zap.String("integration", integration), + zap.String("watch_id", watchID), + zap.Error(err)) + return noop, true + } + + key := watcherSlotKey(integration, watchID) + s.watcherMu.Lock() + defer s.watcherMu.Unlock() + if s.pendingByWatch == nil { + s.pendingByWatch = make(map[string]int) + } + if s.watcherSaturated == nil { + s.watcherSaturated = make(map[string]bool) + } + pending := s.pendingByWatch[key] + if count+pending >= *maxInflight { + s.recordSaturatedLocked(integration, watchID, key, count, pending, *maxInflight) + return noop, false + } + s.recordClearedLocked(integration, watchID, key) + s.pendingByWatch[key] = pending + 1 + return s.releaseFunc(key), true +} + +func (s *Service) releaseFunc(key string) func() { + return func() { + s.watcherMu.Lock() + defer s.watcherMu.Unlock() + if s.pendingByWatch == nil { + return + } + if v := s.pendingByWatch[key]; v <= 1 { + delete(s.pendingByWatch, key) + } else { + s.pendingByWatch[key] = v - 1 + } + } +} + +// recordSaturatedLocked logs a one-shot Warn the first time the gate sees +// (count + pending) >= cap for this watch, and bumps the deferred counter. +// Subsequent deferrals during the same saturation window are Debug-only. +// Must be called with s.watcherMu held. +func (s *Service) recordSaturatedLocked(integration, watchID, key string, count, pending, maxInflight int) { + watcherDispatchDeferredTotal.Add(watcherMetricLabel("integration", integration), 1) + if s.watcherSaturated[key] { + s.logger.Debug("watcher throttle: deferring event (cap held)", + zap.String("integration", integration), + zap.String("watch_id", watchID), + zap.Int("count", count), + zap.Int("pending", pending), + zap.Int("cap", maxInflight)) + return + } + s.watcherSaturated[key] = true + watcherCapTransitionsTotal.Add( + watcherMetricLabel("integration", integration, "transition", "reached"), 1) + s.logger.Warn("watcher throttle: cap reached, deferring further events until tasks drain", + zap.String("integration", integration), + zap.String("watch_id", watchID), + zap.Int("count", count), + zap.Int("pending", pending), + zap.Int("cap", maxInflight)) +} + +// recordClearedLocked emits a single Warn when the gate transitions back +// from saturated → not-saturated for this watch. Cheap when not saturated: +// just a map lookup. Must be called with s.watcherMu held. +func (s *Service) recordClearedLocked(integration, watchID, key string) { + if !s.watcherSaturated[key] { + return + } + delete(s.watcherSaturated, key) + watcherCapTransitionsTotal.Add( + watcherMetricLabel("integration", integration, "transition", "cleared"), 1) + s.logger.Warn("watcher throttle: cap cleared, resuming dispatch", + zap.String("integration", integration), + zap.String("watch_id", watchID)) +} diff --git a/apps/backend/internal/orchestrator/watcher_throttle_test.go b/apps/backend/internal/orchestrator/watcher_throttle_test.go new file mode 100644 index 0000000000..0cb1f55975 --- /dev/null +++ b/apps/backend/internal/orchestrator/watcher_throttle_test.go @@ -0,0 +1,358 @@ +package orchestrator + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/task/models" +) + +// fakeWatcherCounter implements WatcherTaskCounter for tests. Each test sets +// the value returned for a (integration, watchID) lookup. +type fakeWatcherCounter struct { + mu sync.Mutex + counts map[string]int + err error + queryLog []string +} + +func newFakeWatcherCounter() *fakeWatcherCounter { + return &fakeWatcherCounter{counts: map[string]int{}} +} + +func (f *fakeWatcherCounter) set(integration, watchID string, n int) { + f.mu.Lock() + defer f.mu.Unlock() + f.counts[integration+"|"+watchID] = n +} + +func (f *fakeWatcherCounter) CountOpenWatcherCreatedTasks(_ context.Context, integration, watchID string) (int, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.queryLog = append(f.queryLog, integration+"|"+watchID) + if f.err != nil { + return 0, f.err + } + return f.counts[integration+"|"+watchID], nil +} + +// nopServiceWithCounter builds a Service with just enough wiring for the +// throttle gate. The watcher coordinator is intentionally nil — these tests +// exercise only acquireWatcherSlot. +func nopServiceWithCounter(t *testing.T, counter WatcherTaskCounter) *Service { + t.Helper() + return &Service{ + logger: nopLogger(t), + watcherTaskCount: counter, + } +} + +func ptrInt(v int) *int { return &v } + +func TestAcquireWatcherSlot_NilCapBypasses(t *testing.T) { + counter := newFakeWatcherCounter() + s := nopServiceWithCounter(t, counter) + + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "watch-1", nil) + if !ok { + t.Fatal("expected acquire to pass when cap is nil (uncapped)") + } + if release == nil { + t.Fatal("expected non-nil release func") + } + if len(counter.queryLog) != 0 { + t.Fatalf("expected no DB query for uncapped watch, got %v", counter.queryLog) + } + release() +} + +func TestAcquireWatcherSlot_EmptyWatchIDBypasses(t *testing.T) { + s := nopServiceWithCounter(t, newFakeWatcherCounter()) + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "", ptrInt(1)) + if !ok || release == nil { + t.Fatal("expected bypass when watch id is empty") + } + release() +} + +func TestAcquireWatcherSlot_UnderCapAcquires(t *testing.T) { + counter := newFakeWatcherCounter() + counter.set("linear", "w-1", 2) + s := nopServiceWithCounter(t, counter) + + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(5)) + if !ok { + t.Fatal("expected acquire to pass when count+pending < cap") + } + release() +} + +func TestAcquireWatcherSlot_AtCapDefers(t *testing.T) { + counter := newFakeWatcherCounter() + counter.set("linear", "w-1", 5) + s := nopServiceWithCounter(t, counter) + + _, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(5)) + if ok { + t.Fatal("expected acquire to fail when count >= cap") + } +} + +func TestAcquireWatcherSlot_PendingCountedAgainstCap(t *testing.T) { + // Carlos's burst race: two events arrive back-to-back. DB count is 0 for + // both (no dedup row written yet). Cap is 1. Only the first should + // acquire — the synchronous mutex + pending counter must reject the + // second. + counter := newFakeWatcherCounter() + s := nopServiceWithCounter(t, counter) + + release1, ok1 := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !ok1 { + t.Fatal("expected first acquire to pass") + } + defer release1() + + _, ok2 := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if ok2 { + t.Fatal("expected second acquire to defer (pending=1, cap=1)") + } +} + +func TestAcquireWatcherSlot_ReleaseRestoresSlot(t *testing.T) { + counter := newFakeWatcherCounter() + s := nopServiceWithCounter(t, counter) + + release1, ok1 := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !ok1 { + t.Fatal("expected first acquire to pass") + } + release1() + + // After release, the next event should pass (pending is back to 0). + release2, ok2 := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !ok2 { + t.Fatal("expected second acquire to pass after first released") + } + release2() +} + +func TestAcquireWatcherSlot_DifferentWatchesIsolated(t *testing.T) { + counter := newFakeWatcherCounter() + counter.set("linear", "w-1", 1) + counter.set("linear", "w-2", 0) + s := nopServiceWithCounter(t, counter) + + // w-1 is at cap, w-2 is empty — they must not share pending state. + _, ok1 := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if ok1 { + t.Fatal("expected w-1 to be at cap") + } + release2, ok2 := s.acquireWatcherSlot(context.Background(), "linear", "w-2", ptrInt(1)) + if !ok2 { + t.Fatal("expected w-2 to be acquirable (independent watch)") + } + release2() +} + +func TestAcquireWatcherSlot_DifferentIntegrationsIsolated(t *testing.T) { + counter := newFakeWatcherCounter() + s := nopServiceWithCounter(t, counter) + + // Same watch id across two integrations must NOT collide. + releaseLin, okLin := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !okLin { + t.Fatal("expected linear w-1 to acquire") + } + defer releaseLin() + + releaseJira, okJira := s.acquireWatcherSlot(context.Background(), "jira", "w-1", ptrInt(1)) + if !okJira { + t.Fatal("expected jira w-1 to acquire independently") + } + releaseJira() +} + +func TestAcquireWatcherSlot_NonPositiveCapTreatedAsUncapped(t *testing.T) { + // The API rejects <= 0 but a stale row could theoretically reach the gate + // with 0 or negative. Treat as uncapped (fail-open) rather than block + // everything. + counter := newFakeWatcherCounter() + counter.set("linear", "w-1", 100) + s := nopServiceWithCounter(t, counter) + + for _, c := range []int{0, -1, -100} { + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(c)) + if !ok { + t.Fatalf("expected non-positive cap (%d) to bypass gate", c) + } + release() + } +} + +func TestAcquireWatcherSlot_CountErrorFailsOpen(t *testing.T) { + counter := newFakeWatcherCounter() + counter.err = errors.New("db down") + s := nopServiceWithCounter(t, counter) + + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !ok { + t.Fatal("expected fail-open on count error") + } + release() +} + +func TestAcquireWatcherSlot_NilCounterBypasses(t *testing.T) { + // Service constructed before WatcherTaskCounter is wired must not panic. + s := nopServiceWithCounter(t, nil) + release, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-1", ptrInt(1)) + if !ok { + t.Fatal("expected nil counter to bypass gate (fail-open)") + } + release() +} + +// TestAcquireWatcherSlot_ConcurrentBurstRespectsCap fires N goroutines at the +// gate simultaneously with cap=1 and DB count=0. Exactly one must acquire; the +// rest must defer. Unlike the sequential dispatch test below, this exercises +// watcherMu under real contention — run with `-race` to catch a regression +// that drops the lock. Acquirers hold their slot (never release) so the cap +// stays saturated for the duration of the burst. +func TestAcquireWatcherSlot_ConcurrentBurstRespectsCap(t *testing.T) { + const goroutines = 64 + for _, capValue := range []int{1, 5} { + counter := newFakeWatcherCounter() // count=0 for the watch + s := nopServiceWithCounter(t, counter) + + var acquired int64 + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < goroutines; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + if _, ok := s.acquireWatcherSlot(context.Background(), "linear", "w-burst", ptrInt(capValue)); ok { + atomic.AddInt64(&acquired, 1) + // Hold the slot — do NOT release, so the cap stays full. + } + }() + } + close(start) + wg.Wait() + + if int(acquired) != capValue { + t.Fatalf("cap=%d: expected exactly %d acquisitions under %d concurrent callers, got %d", + capValue, capValue, goroutines, acquired) + } + } +} + +// TestDispatchWatcherEvent_GateBlocksBurstRace is the regression test for +// Carlos's burst race: 10 events arrive back-to-back with cap=1, the +// CreateIssueTask call blocks (simulating a slow DB write that hasn't +// committed the dedup row yet), and only ONE event must reach the +// coordinator. Without the pending counter, all 10 read count=0 and pass. +func TestDispatchWatcherEvent_GateBlocksBurstRace(t *testing.T) { + counter := newFakeWatcherCounter() // count=0 for w-1 + src := &fakeWatcherSource{ + name: "linear", + reserveOK: true, + buildReq: &IssueTaskRequest{ + WorkspaceID: "ws-1", + WorkflowStepID: "step-1", + }, + watchID: "w-1", + maxInflightTasks: ptrInt(1), + } + + // Block CreateIssueTask so the admitted goroutine holds its slot for the + // duration of the burst. `entered` / `done` are buffered to `burst` so a + // (wrongly) over-admitted goroutine can never block on signaling — the + // test must observe the over-admission, not deadlock on it. + gate := make(chan struct{}) + entered := make(chan struct{}, burstSize) + done := make(chan struct{}, burstSize) + creator := &blockingTaskCreator{block: gate, entered: entered, done: done} + + starter := &fakeTaskStarter{} + s := &Service{ + logger: nopLogger(t), + watcherTaskCount: counter, + issueTaskCreator: creator, + watcherCoordinator: newTestCoordinator(t, nil, false, starter), + } + s.watcherCoordinator.SetTaskCreator(creator) + + // Fire the burst back-to-back. Only one should pass the gate; the rest + // must defer without spawning a goroutine. + for i := 0; i < burstSize; i++ { + s.dispatchWatcherEvent(context.Background(), "linear", src, "evt") + } + + // Gate acquisition is synchronous (it happens before the goroutine spawns), + // so once the burst loop returns the pending counter is final. Asserting on + // it here catches over-admission deterministically, independent of how the + // admitted goroutine(s) are scheduled — this is the real regression guard. + s.watcherMu.Lock() + pending := s.pendingByWatch["linear|w-1"] + s.watcherMu.Unlock() + if pending != 1 { + t.Fatalf("expected exactly 1 slot acquired across the burst, got %d", pending) + } + + // Confirm the admitted goroutine actually reached CreateIssueTask. Bounded + // so a broken gate that admits nothing fails fast instead of hanging. + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("no goroutine reached CreateIssueTask within 5s") + } + close(gate) + + // Wait for the admitted goroutine to finish before asserting the call + // count, so a late over-admitted call can't slip past the assertion. + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("admitted goroutine did not finish within 5s") + } + if got := creator.callCount(); got != 1 { + t.Fatalf("expected exactly 1 CreateIssueTask call (gate enforced), got %d", got) + } +} + +// burstSize is the number of events fired at the gate in the burst-race test. +const burstSize = 10 + +// blockingTaskCreator counts calls and blocks each one on `block` until it is +// closed. It signals `entered` when a call reaches CreateIssueTask and `done` +// when that call unblocks, so a test can wait deterministically (with a +// timeout) instead of polling. Both channels are buffered by the caller so an +// over-admitted goroutine never blocks on signaling. +type blockingTaskCreator struct { + mu sync.Mutex + calls int + block <-chan struct{} + entered chan<- struct{} + done chan<- struct{} +} + +func (b *blockingTaskCreator) CreateIssueTask(_ context.Context, _ *IssueTaskRequest) (*models.Task, error) { + b.mu.Lock() + b.calls++ + b.mu.Unlock() + b.entered <- struct{}{} + <-b.block + b.done <- struct{}{} + return &models.Task{ID: "t-blocked"}, nil +} + +func (b *blockingTaskCreator) callCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return b.calls +} diff --git a/apps/backend/internal/task/handlers/process_handlers_test.go b/apps/backend/internal/task/handlers/process_handlers_test.go index f85cfd794f..beb8582e92 100644 --- a/apps/backend/internal/task/handlers/process_handlers_test.go +++ b/apps/backend/internal/task/handlers/process_handlers_test.go @@ -73,6 +73,9 @@ func (m *mockRepository) ArchiveTask(ctx context.Context, id string) error { func (m *mockRepository) ListTasksForAutoArchive(ctx context.Context) ([]*models.Task, error) { return nil, nil } +func (m *mockRepository) CountOpenWatcherCreatedTasks(_ context.Context, _, _ string) (int, error) { + return 0, nil +} func (m *mockRepository) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error { return nil } diff --git a/apps/backend/internal/task/repository/count_open_watcher_tasks_test.go b/apps/backend/internal/task/repository/count_open_watcher_tasks_test.go new file mode 100644 index 0000000000..2151b25a57 --- /dev/null +++ b/apps/backend/internal/task/repository/count_open_watcher_tasks_test.go @@ -0,0 +1,89 @@ +package repository + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +func TestSQLiteRepository_CountOpenWatcherCreatedTasks(t *testing.T) { + repo, cleanup := createTestSQLiteRepo(t) + defer cleanup() + ctx := context.Background() + + _ = repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Workspace"}) + _ = repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "WF"}) + + // Linear watch A has 2 open tasks + 1 completed + 1 archived → count should be 2. + mkTask := func(id, watchKey, watchID string, state v1.TaskState, archived bool) { + t.Helper() + task := &models.Task{ + ID: id, + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + Title: id, + State: state, + Metadata: map[string]interface{}{ + watchKey: watchID, + }, + } + if err := repo.CreateTask(ctx, task); err != nil { + t.Fatalf("create task %s: %v", id, err) + } + if archived { + if err := repo.ArchiveTask(ctx, id); err != nil { + t.Fatalf("archive task %s: %v", id, err) + } + } + } + + mkTask("t-open-1", "linear_issue_watch_id", "watch-a", v1.TaskStateTODO, false) + mkTask("t-open-2", "linear_issue_watch_id", "watch-a", v1.TaskStateInProgress, false) + mkTask("t-completed", "linear_issue_watch_id", "watch-a", v1.TaskStateCompleted, false) + mkTask("t-archived", "linear_issue_watch_id", "watch-a", v1.TaskStateTODO, true) + + // Different watch — must not count toward watch-a. + mkTask("t-other-linear", "linear_issue_watch_id", "watch-b", v1.TaskStateTODO, false) + + // Different integration — Jira watch with the same id must not bleed in. + mkTask("t-jira", "jira_issue_watch_id", "watch-a", v1.TaskStateTODO, false) + + // User-created task with no watcher metadata. + mkTask("t-user", "unrelated_key", "watch-a", v1.TaskStateTODO, false) + + got, err := repo.CountOpenWatcherCreatedTasks(ctx, "linear", "watch-a") + if err != nil { + t.Fatalf("count: %v", err) + } + if got != 2 { + t.Fatalf("expected 2 open linear tasks for watch-a, got %d", got) + } + + gotB, err := repo.CountOpenWatcherCreatedTasks(ctx, "linear", "watch-b") + if err != nil { + t.Fatalf("count watch-b: %v", err) + } + if gotB != 1 { + t.Fatalf("expected 1 open linear task for watch-b, got %d", gotB) + } + + gotJira, err := repo.CountOpenWatcherCreatedTasks(ctx, "jira", "watch-a") + if err != nil { + t.Fatalf("count jira: %v", err) + } + if gotJira != 1 { + t.Fatalf("expected 1 open jira task for watch-a, got %d", gotJira) + } + + // Unknown integration falls through without erroring. + gotUnknown, err := repo.CountOpenWatcherCreatedTasks(ctx, "github", "watch-a") + if err != nil { + t.Fatalf("count github: %v", err) + } + if gotUnknown != 0 { + t.Fatalf("expected 0 open github tasks, got %d", gotUnknown) + } +} diff --git a/apps/backend/internal/task/repository/interface.go b/apps/backend/internal/task/repository/interface.go index 1c9b7e8b57..b9ad14800f 100644 --- a/apps/backend/internal/task/repository/interface.go +++ b/apps/backend/internal/task/repository/interface.go @@ -35,6 +35,11 @@ type TaskRepository interface { // archived by the named cascade. Returns whether the row was updated. UnarchiveTaskByCascade(ctx context.Context, id, cascadeID string) (bool, error) ListTasksForAutoArchive(ctx context.Context) ([]*models.Task, error) + // CountOpenWatcherCreatedTasks returns the number of open watcher-created + // tasks for a single (integration, watchID) pair. Open = non-archived AND + // state NOT IN (COMPLETED, FAILED, CANCELLED). Used by the + // orchestrator's watcher throttle gate to enforce a per-watch cap. + CountOpenWatcherCreatedTasks(ctx context.Context, integration, watchID string) (int, error) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error CountTasksByWorkflow(ctx context.Context, workflowID string) (int, error) CountTasksByWorkflowStep(ctx context.Context, stepID string) (int, error) diff --git a/apps/backend/internal/task/repository/sqlite/task.go b/apps/backend/internal/task/repository/sqlite/task.go index 6843eeab9d..5e3f67ca59 100644 --- a/apps/backend/internal/task/repository/sqlite/task.go +++ b/apps/backend/internal/task/repository/sqlite/task.go @@ -730,6 +730,49 @@ func (r *Repository) ListTasksForAutoArchive(ctx context.Context) ([]*models.Tas return r.scanTasks(rows) } +// watcherMetadataKeyByIntegration maps an integration name to the metadata key +// that watcher-created tasks carry. Used by CountOpenWatcherCreatedTasks to +// scope the COUNT to a single watcher. Returns "" for unknown integrations so +// the count query returns 0 without erroring. +func watcherMetadataKeyByIntegration(integration string) string { + switch integration { + case "linear": + return "linear_issue_watch_id" + case "jira": + return "jira_issue_watch_id" + default: + return "" + } +} + +// CountOpenWatcherCreatedTasks returns the number of open watcher-created tasks +// for a single (integration, watchID) pair. "Open" means non-archived AND not +// in a terminal state (COMPLETED, FAILED, CANCELLED). Watcher-created tasks +// are identified via the integration's JSON metadata key, scoped per +// integration so a Linear and Jira watch with the same id don't collide. +// +// Unknown integrations return (0, nil) — the throttle gate falls open for +// integrations not yet wired into the JSON predicate map. +func (r *Repository) CountOpenWatcherCreatedTasks(ctx context.Context, integration, watchID string) (int, error) { + key := watcherMetadataKeyByIntegration(integration) + if key == "" || watchID == "" { + return 0, nil + } + query := r.ro.Rebind(` + SELECT COUNT(*) FROM tasks + WHERE archived_at IS NULL + AND state NOT IN (?, ?, ?) + AND json_extract(metadata, '$.` + key + `') = ? + `) + var n int + if err := r.ro.QueryRowxContext(ctx, query, + v1.TaskStateCompleted, v1.TaskStateFailed, v1.TaskStateCancelled, watchID, + ).Scan(&n); err != nil { + return 0, err + } + return n, nil +} + // UpdateTaskState updates the state of a task func (r *Repository) UpdateTaskState(ctx context.Context, id string, state v1.TaskState) error { result, err := r.db.ExecContext(ctx, r.db.Rebind(`UPDATE tasks SET state = ?, updated_at = ? WHERE id = ?`), state, time.Now().UTC(), id) diff --git a/apps/backend/internal/task/service/handoff_workspace_test.go b/apps/backend/internal/task/service/handoff_workspace_test.go index 06de8396c9..c2feb1de6c 100644 --- a/apps/backend/internal/task/service/handoff_workspace_test.go +++ b/apps/backend/internal/task/service/handoff_workspace_test.go @@ -348,6 +348,10 @@ func (r *phase4TaskRepo) ListTasksForAutoArchive(context.Context) ([]*models.Tas r.panicNotUsed("ListTasksForAutoArchive") return nil, nil } +func (r *phase4TaskRepo) CountOpenWatcherCreatedTasks(context.Context, string, string) (int, error) { + r.panicNotUsed("CountOpenWatcherCreatedTasks") + return 0, nil +} func (r *phase4TaskRepo) UpdateTaskState(context.Context, string, v1.TaskState) error { r.panicNotUsed("UpdateTaskState") return nil diff --git a/apps/web/components/jira/jira-issue-watch-dialog.tsx b/apps/web/components/jira/jira-issue-watch-dialog.tsx index 53b247c784..22f12097be 100644 --- a/apps/web/components/jira/jira-issue-watch-dialog.tsx +++ b/apps/web/components/jira/jira-issue-watch-dialog.tsx @@ -60,8 +60,27 @@ type FormState = { prompt: string; enabled: boolean; pollInterval: number; + /** + * Per-watcher throttle cap as a free-text input: empty string means + * "uncapped" (sent as null), non-empty must parse to a positive integer. + */ + maxInflightTasks: string; }; +function maxInflightTasksString(v: number | null | undefined): string { + if (v === undefined || v === null) return ""; + if (!Number.isFinite(v) || v <= 0) return ""; + return String(v); +} + +function parseMaxInflightTasks(raw: string): number | null | "invalid" { + const t = raw.trim(); + if (t === "") return null; + const n = Number(t); + if (!Number.isInteger(n) || n <= 0) return "invalid"; + return n; +} + const DEFAULT_JQL = `project = PROJ AND status = "Open" ORDER BY created DESC`; function makeEmptyForm(workspaceId: string): FormState { @@ -75,6 +94,7 @@ function makeEmptyForm(workspaceId: string): FormState { prompt: DEFAULT_JIRA_ISSUE_WATCH_PROMPT, enabled: true, pollInterval: 300, + maxInflightTasks: "5", }; } @@ -89,6 +109,7 @@ function formStateFromWatch(w: JiraIssueWatch): FormState { prompt: w.prompt.trim() ? w.prompt : DEFAULT_JIRA_ISSUE_WATCH_PROMPT, enabled: w.enabled, pollInterval: w.pollIntervalSeconds, + maxInflightTasks: maxInflightTasksString(w.maxInflightTasks), }; } @@ -327,6 +348,38 @@ function AutomationFields({ ); } +function MaxInflightTasksField({ + form, + setForm, +}: { + form: FormState; + setForm: React.Dispatch>; +}) { + const parsed = parseMaxInflightTasks(form.maxInflightTasks); + const invalid = parsed === "invalid"; + return ( +
+ +

+ Cap on open tasks created by this watcher. Leave blank for no cap. New matches are deferred + to the next poll when the cap is reached. +

+ setForm((p) => ({ ...p, maxInflightTasks: e.target.value }))} + min={1} + step={1} + placeholder="(no cap)" + aria-invalid={invalid} + /> + {invalid && ( +

Enter a positive integer or leave blank.

+ )} +
+ ); +} + function SettingsFields({ form, setForm, @@ -349,6 +402,7 @@ function SettingsFields({ max={3600} /> +
@@ -397,16 +451,23 @@ export function JiraIssueWatchDialog({ // refs) or when the dialog was opened from a single-workspace surface. const workspaceLocked = !!watch || !!workspaceId; + const parsedMaxInflight = parseMaxInflightTasks(form.maxInflightTasks); const canSave = !!form.workspaceId && !!form.jql.trim() && !!form.workflowId && !!form.workflowStepId && - !!form.prompt.trim(); + !!form.prompt.trim() && + parsedMaxInflight !== "invalid"; const handleSave = useCallback(async () => { setSaving(true); try { + const maxInflight = parseMaxInflightTasks(form.maxInflightTasks); + if (maxInflight === "invalid") { + // canSave gates the button but guard the handler too — see Linear dialog. + return; + } const payload = { jql: form.jql, workflowId: form.workflowId, @@ -416,6 +477,7 @@ export function JiraIssueWatchDialog({ prompt: form.prompt, enabled: form.enabled, pollIntervalSeconds: form.pollInterval, + maxInflightTasks: maxInflight, }; if (watch) { await onUpdate(watch.id, payload); diff --git a/apps/web/components/linear/linear-issue-watch-dialog.tsx b/apps/web/components/linear/linear-issue-watch-dialog.tsx index 611abdf0b9..c52ab75a0f 100644 --- a/apps/web/components/linear/linear-issue-watch-dialog.tsx +++ b/apps/web/components/linear/linear-issue-watch-dialog.tsx @@ -38,11 +38,12 @@ import { CREATOR_ANY, type FormState, type LinearPriority, - buildFilterPayload, + buildWatchPayload, creatorPlaceholder, - filterIsEmpty, formStateFromWatch, + isWatchFormReady, makeEmptyForm, + parseMaxInflightTasks, userOptionLabel, } from "./linear-issue-watch-form"; import type { @@ -452,6 +453,38 @@ function AutomationFields({ ); } +function MaxInflightTasksField({ + form, + setForm, +}: { + form: FormState; + setForm: React.Dispatch>; +}) { + const parsed = parseMaxInflightTasks(form.maxInflightTasks); + const invalid = parsed === "invalid"; + return ( +
+ +

+ Cap on open tasks created by this watcher. Leave blank for no cap. New matches are deferred + to the next poll when the cap is reached. +

+ setForm((p) => ({ ...p, maxInflightTasks: e.target.value }))} + min={1} + step={1} + placeholder="(no cap)" + aria-invalid={invalid} + /> + {invalid && ( +

Enter a positive integer or leave blank.

+ )} +
+ ); +} + function SettingsFields({ form, setForm, @@ -474,6 +507,7 @@ function SettingsFields({ max={3600} />
+
@@ -515,28 +549,13 @@ export function LinearIssueWatchDialog({ }, [watch, open, workspaceId, activeWorkspaceId]); const workspaceLocked = !!watch || !!workspaceId; - - const canSave = - !!form.workspaceId && - !filterIsEmpty(form) && - !!form.workflowId && - !!form.workflowStepId && - !!form.prompt.trim(); + const canSave = isWatchFormReady(form); const handleSave = useCallback(async () => { + const payload = buildWatchPayload(form); + if (!payload) return; // re-checks the cap input — see canSave gate setSaving(true); try { - const filter = buildFilterPayload(form); - const payload = { - filter, - workflowId: form.workflowId, - workflowStepId: form.workflowStepId, - agentProfileId: form.agentProfileId, - executorProfileId: form.executorProfileId, - prompt: form.prompt, - enabled: form.enabled, - pollIntervalSeconds: form.pollInterval, - }; if (watch) { await onUpdate(watch.id, payload); } else { diff --git a/apps/web/components/linear/linear-issue-watch-form.test.ts b/apps/web/components/linear/linear-issue-watch-form.test.ts new file mode 100644 index 0000000000..f81793178c --- /dev/null +++ b/apps/web/components/linear/linear-issue-watch-form.test.ts @@ -0,0 +1,91 @@ +import { describe, it, expect } from "vitest"; + +import { + buildWatchPayload, + isWatchFormReady, + makeEmptyForm, + maxInflightTasksString, + parseMaxInflightTasks, + type FormState, +} from "./linear-issue-watch-form"; + +describe("parseMaxInflightTasks", () => { + it("treats blank / whitespace as uncapped (null)", () => { + expect(parseMaxInflightTasks("")).toBeNull(); + expect(parseMaxInflightTasks(" ")).toBeNull(); + }); + + it("accepts positive integers", () => { + expect(parseMaxInflightTasks("5")).toBe(5); + expect(parseMaxInflightTasks(" 12 ")).toBe(12); + }); + + it("rejects zero, negatives, and non-integers", () => { + expect(parseMaxInflightTasks("0")).toBe("invalid"); + expect(parseMaxInflightTasks("-1")).toBe("invalid"); + expect(parseMaxInflightTasks("1.5")).toBe("invalid"); + expect(parseMaxInflightTasks("abc")).toBe("invalid"); + }); +}); + +describe("maxInflightTasksString", () => { + it("renders null/undefined and non-positive values as blank (uncapped)", () => { + expect(maxInflightTasksString(null)).toBe(""); + expect(maxInflightTasksString(undefined)).toBe(""); + expect(maxInflightTasksString(0)).toBe(""); + expect(maxInflightTasksString(-3)).toBe(""); + }); + + it("renders positive values as the integer string", () => { + expect(maxInflightTasksString(5)).toBe("5"); + }); +}); + +// makeEmptyForm seeds a valid filter-less skeleton; tests add a filter field +// and a destination so isWatchFormReady's other gates pass and we isolate the +// cap behaviour. +function readyForm(overrides: Partial = {}): FormState { + return { + ...makeEmptyForm("ws-1"), + teamKey: "ENG", + workflowId: "wf-1", + workflowStepId: "step-1", + prompt: "do the thing", + ...overrides, + }; +} + +describe("isWatchFormReady", () => { + it("is true for a complete form with a valid cap", () => { + expect(isWatchFormReady(readyForm({ maxInflightTasks: "5" }))).toBe(true); + }); + + it("is true when the cap is blank (uncapped)", () => { + expect(isWatchFormReady(readyForm({ maxInflightTasks: "" }))).toBe(true); + }); + + it("is false when the cap is invalid", () => { + expect(isWatchFormReady(readyForm({ maxInflightTasks: "0" }))).toBe(false); + }); + + it("is false when the filter is empty", () => { + expect(isWatchFormReady(readyForm({ teamKey: "", maxInflightTasks: "5" }))).toBe(false); + }); +}); + +describe("buildWatchPayload", () => { + it("maps a blank cap to null", () => { + const payload = buildWatchPayload(readyForm({ maxInflightTasks: "" })); + expect(payload).not.toBeNull(); + expect(payload?.maxInflightTasks).toBeNull(); + }); + + it("maps a positive cap to the integer", () => { + const payload = buildWatchPayload(readyForm({ maxInflightTasks: "7" })); + expect(payload?.maxInflightTasks).toBe(7); + }); + + it("returns null when the cap is invalid", () => { + expect(buildWatchPayload(readyForm({ maxInflightTasks: "-1" }))).toBeNull(); + }); +}); diff --git a/apps/web/components/linear/linear-issue-watch-form.ts b/apps/web/components/linear/linear-issue-watch-form.ts index 962336eaf0..658958d146 100644 --- a/apps/web/components/linear/linear-issue-watch-form.ts +++ b/apps/web/components/linear/linear-issue-watch-form.ts @@ -34,6 +34,13 @@ export interface FormState { prompt: string; enabled: boolean; pollInterval: number; + /** + * Per-watcher throttle cap as a free-text input: empty string means + * "uncapped" (sent as null), non-empty must parse to a positive integer. + * Kept as a string so the user can clear the field without it snapping + * back to a number. + */ + maxInflightTasks: string; } export function makeEmptyForm(workspaceId: string): FormState { @@ -55,6 +62,7 @@ export function makeEmptyForm(workspaceId: string): FormState { prompt: DEFAULT_LINEAR_ISSUE_WATCH_PROMPT, enabled: true, pollInterval: 300, + maxInflightTasks: "5", }; } @@ -82,6 +90,82 @@ export function formStateFromWatch(w: LinearIssueWatch): FormState { prompt: w.prompt || DEFAULT_LINEAR_ISSUE_WATCH_PROMPT, enabled: w.enabled, pollInterval: w.pollIntervalSeconds, + maxInflightTasks: maxInflightTasksString(w.maxInflightTasks), + }; +} + +/** + * Renders the watch's stored throttle cap for the dialog input. `null` / + * `undefined` map to an empty string ("uncapped"); positive integers are + * shown as-is. Non-positive values from a stale row are clamped to empty + * — the backend rejects them on save anyway, and showing a misleading "0" + * would let users think the cap was enforced. + */ +export function maxInflightTasksString(v: number | null | undefined): string { + if (v === undefined || v === null) return ""; + if (!Number.isFinite(v) || v <= 0) return ""; + return String(v); +} + +/** + * Parses the throttle-cap input back into a payload value. Returns: + * - `null` when the input is blank (user wants uncapped), + * - the integer when it's a positive whole number, + * - `"invalid"` when the input is non-empty but unparseable / non-positive, + * so the dialog can surface an inline validation error before submit. + */ +export function parseMaxInflightTasks(raw: string): number | null | "invalid" { + const t = raw.trim(); + if (t === "") return null; + const n = Number(t); + if (!Number.isInteger(n) || n <= 0) return "invalid"; + return n; +} + +/** + * Aggregates the dialog's "can the Save button enable?" rule. Pulled out of + * the dialog so the file stays under the 600-line ceiling and the rule has + * one named home — handy for unit tests as the form grows. + */ +export function isWatchFormReady(form: FormState): boolean { + return ( + !!form.workspaceId && + !filterIsEmpty(form) && + !!form.workflowId && + !!form.workflowStepId && + !!form.prompt.trim() && + parseMaxInflightTasks(form.maxInflightTasks) !== "invalid" + ); +} + +/** + * Builds the API payload from the dialog state. Returns `null` when the + * throttle cap input fails validation, so the caller can short-circuit + * the submit without throwing. + */ +export function buildWatchPayload(form: FormState): { + filter: LinearSearchFilter; + workflowId: string; + workflowStepId: string; + agentProfileId: string; + executorProfileId: string; + prompt: string; + enabled: boolean; + pollIntervalSeconds: number; + maxInflightTasks: number | null; +} | null { + const maxInflight = parseMaxInflightTasks(form.maxInflightTasks); + if (maxInflight === "invalid") return null; + return { + filter: buildFilterPayload(form), + workflowId: form.workflowId, + workflowStepId: form.workflowStepId, + agentProfileId: form.agentProfileId, + executorProfileId: form.executorProfileId, + prompt: form.prompt, + enabled: form.enabled, + pollIntervalSeconds: form.pollInterval, + maxInflightTasks: maxInflight, }; } diff --git a/apps/web/lib/types/jira.ts b/apps/web/lib/types/jira.ts index 9fe0e3f552..a03676f95c 100644 --- a/apps/web/lib/types/jira.ts +++ b/apps/web/lib/types/jira.ts @@ -109,6 +109,12 @@ export interface JiraIssueWatch { prompt: string; enabled: boolean; pollIntervalSeconds: number; + /** + * Cap on concurrent open watcher-created tasks for this watch. + * `null`/omitted means uncapped. Positive integers are accepted; the backend + * rejects values ≤ 0. + */ + maxInflightTasks?: number | null; /** Last poll timestamp, or null when the watch has never run. */ lastPolledAt?: string | null; createdAt: string; @@ -124,6 +130,8 @@ export interface CreateJiraIssueWatchInput { executorProfileId?: string; prompt?: string; pollIntervalSeconds?: number; + /** Per-watch throttle cap; null = uncapped, positive int = cap. */ + maxInflightTasks?: number | null; enabled?: boolean; } @@ -137,4 +145,6 @@ export interface UpdateJiraIssueWatchInput { prompt?: string; enabled?: boolean; pollIntervalSeconds?: number; + /** Per-watch throttle cap; null = uncapped, positive int = cap. */ + maxInflightTasks?: number | null; } diff --git a/apps/web/lib/types/linear.ts b/apps/web/lib/types/linear.ts index ec107a7f7f..936ea82838 100644 --- a/apps/web/lib/types/linear.ts +++ b/apps/web/lib/types/linear.ts @@ -130,6 +130,12 @@ export interface LinearIssueWatch { prompt: string; enabled: boolean; pollIntervalSeconds: number; + /** + * Cap on concurrent open watcher-created tasks for this watch. + * `null`/omitted means uncapped. Positive integers are accepted; the backend + * rejects values ≤ 0. + */ + maxInflightTasks?: number | null; /** Last poll timestamp, or null when the watch has never run. */ lastPolledAt?: string | null; createdAt: string; @@ -145,6 +151,8 @@ export interface CreateLinearIssueWatchInput { executorProfileId?: string; prompt?: string; pollIntervalSeconds?: number; + /** Per-watch throttle cap; null = uncapped, positive int = cap. */ + maxInflightTasks?: number | null; enabled?: boolean; } @@ -158,4 +166,6 @@ export interface UpdateLinearIssueWatchInput { prompt?: string; enabled?: boolean; pollIntervalSeconds?: number; + /** Per-watch throttle cap; null = uncapped, positive int = cap. */ + maxInflightTasks?: number | null; }