From 353a8acc009f6d26ec05d5f4ffbfc0293244a0f8 Mon Sep 17 00:00:00 2001 From: Greg Allen Date: Mon, 10 Aug 2026 18:19:26 -0400 Subject: [PATCH] feat(#5959): consolidate GitLab poll into single auto-promoting schedule Replace the dual fast/full poll schedules with a single schedule that runs every 5 minutes and auto-promotes to a full poll when 15+ minutes have elapsed since the last full poll. Changes: - setupGitLabPipelineSchedules now creates one schedule ("fullsend poll" at */5) for all tiers, removing the enterprise/free distinction and the FULLSEND_POLL_MODE variable. - Poller.Run auto-selects poll mode at the start of each cycle by reading FULLSEND_LAST_POLL_AT_FULL: if the watermark is absent or 15+ minutes old, a full poll runs; otherwise a fast (slash-commands- only) poll runs. - Options.SlashCommandsOnly removed; replaced by runtime auto-promotion logic in shouldFullPoll. Options.FullPollInterval added for configurability (defaults to 15 min). - CLI --poll-mode flag and FULLSEND_POLL_MODE env var removed. Closes #5959 Signed-off-by: Claude Signed-off-by: Greg Allen --- ...0067-gitlab-cron-polling-event-dispatch.md | 17 ++ internal/cli/poll.go | 35 ++-- internal/cli/poll_test.go | 2 +- internal/cli/repos.go | 2 +- internal/cli/repos_gitlab.go | 51 ++---- internal/cli/repos_gitlab_test.go | 111 +------------ internal/poll/dispatch_test.go | 4 +- internal/poll/poll.go | 69 ++++++-- internal/poll/poll_test.go | 152 +++++++++++++++--- internal/poll/state.go | 6 +- internal/poll/state_test.go | 9 +- internal/poll/types.go | 12 +- 12 files changed, 260 insertions(+), 210 deletions(-) diff --git a/docs/ADRs/0067-gitlab-cron-polling-event-dispatch.md b/docs/ADRs/0067-gitlab-cron-polling-event-dispatch.md index ac6305ca15..8e439d7a22 100644 --- a/docs/ADRs/0067-gitlab-cron-polling-event-dispatch.md +++ b/docs/ADRs/0067-gitlab-cron-polling-event-dispatch.md @@ -319,6 +319,23 @@ This is an accepted tradeoff — the alternative (sharing a processed-note-IDs set or cross-reading watermarks between modes) adds state coupling that complicates the independent-schedule design. +> **Update (2026-08, #5959):** The dual-schedule architecture above was replaced +> by a single `*/5 * * * *` schedule with automatic full-poll promotion. The +> poller now decides at runtime whether to run a fast poll or full poll based on +> elapsed time since the last full poll (`FULLSEND_LAST_POLL_AT_FULL`). This +> eliminates the tier distinction (Premium vs Free), the separate fast/full +> schedules, and the `FULLSEND_POLL_MODE` variable. The fast-poll watermark +> (`FULLSEND_LAST_POLL_AT_FAST`) is still used for slash-command-only cycles. +> Free tier in-CI polling is no longer supported by this schedule (Free tier's +> minimum interval is 60 minutes); Free tier users should use off-system polling +> (`fullsend poll` on a VM or Kubernetes CronJob) as documented in "GitLab tier +> considerations" below. Superseded sections: "Multi-frequency polling" above, +> the "5 minutes on Premium/Ultimate, 60 minutes on Free tier" reference in the +> cron-poller introduction, "Multi-frequency polling" and fast-poll MR note +> limitation under "Slash command latency", the Free tier 60-minute interval +> references in "GitLab tier considerations", and the "5 minutes on Premium, 60 +> minutes on Free" latency in "Consequences". + ### Event routing The design goal is **functional event-type parity with GitHub** — users see the diff --git a/internal/cli/poll.go b/internal/cli/poll.go index 10da0d68ee..22a186d870 100644 --- a/internal/cli/poll.go +++ b/internal/cli/poll.go @@ -18,17 +18,16 @@ import ( func newPollCmd() *cobra.Command { var ( - forgeFlag string - inputDriver string - projectPath string - gitlabURL string - outputPath string - pollModeFlag string - fullsendDir string - jiraURL string - jiraProject string - jqlOverride string - targetRepo string + forgeFlag string + inputDriver string + projectPath string + gitlabURL string + outputPath string + fullsendDir string + jiraURL string + jiraProject string + jqlOverride string + targetRepo string ) cmd := &cobra.Command{ @@ -55,8 +54,6 @@ func newPollCmd() *cobra.Command { return fmt.Errorf("--project or CI_PROJECT_PATH is required") } - slashCommandsOnly := pollModeFlag == "fast" || os.Getenv("FULLSEND_POLL_MODE") == "fast" - glClient, err := gitlab.New(forgeToken, gitlab.WithBaseURL(gitlabURL)) if err != nil { return fmt.Errorf("create GitLab client: %w", err) @@ -83,12 +80,11 @@ func newPollCmd() *cobra.Command { } opts := poll.Options{ - SlashCommandsOnly: slashCommandsOnly, - BotUserID: botUserID, - GitLabURL: gitlabURL, - PipelineRef: pipelineRef, - PollJobURL: os.Getenv("CI_JOB_URL"), - DispatchSecret: os.Getenv("FULLSEND_DISPATCH_SECRET"), + BotUserID: botUserID, + GitLabURL: gitlabURL, + PipelineRef: pipelineRef, + PollJobURL: os.Getenv("CI_JOB_URL"), + DispatchSecret: os.Getenv("FULLSEND_DISPATCH_SECRET"), } poller := poll.New(pollClient, router, projectPath, opts) @@ -101,7 +97,6 @@ func newPollCmd() *cobra.Command { cmd.Flags().StringVar(&projectPath, "project", "", "GitLab project path (default: $CI_PROJECT_PATH)") cmd.Flags().StringVar(&gitlabURL, "gitlab-url", "https://gitlab.com", "GitLab instance URL") cmd.Flags().StringVar(&outputPath, "output", "", "Path to write dispatches JSON (jira-poll only; ignored by --forge gitlab)") - cmd.Flags().StringVar(&pollModeFlag, "poll-mode", "", "Poll mode: fast (slash commands only) or full") cmd.Flags().StringVar(&fullsendDir, "fullsend-dir", "", "path to the .fullsend configuration directory") _ = cmd.MarkFlagRequired("fullsend-dir") cmd.Flags().StringVar(&jiraURL, "jira-url", "", "Jira instance base URL (default: $JIRA_BASE_URL)") diff --git a/internal/cli/poll_test.go b/internal/cli/poll_test.go index f461816d3b..1892499c0d 100644 --- a/internal/cli/poll_test.go +++ b/internal/cli/poll_test.go @@ -15,7 +15,7 @@ import ( func clearPollEnv(t *testing.T) { t.Helper() for _, v := range []string{ - "FULLSEND_FORGE_TOKEN", "CI_PROJECT_PATH", "FULLSEND_POLL_MODE", + "FULLSEND_FORGE_TOKEN", "CI_PROJECT_PATH", "CI_COMMIT_REF_NAME", "CI_DEFAULT_BRANCH", "CI_JOB_URL", "JIRA_BASE_URL", "GITHUB_REPOSITORY", "JIRA_TOKEN", "JIRA_USER_EMAIL", diff --git a/internal/cli/repos.go b/internal/cli/repos.go index e63ee96aac..426b40caca 100644 --- a/internal/cli/repos.go +++ b/internal/cli/repos.go @@ -725,7 +725,7 @@ func runReposInstall(ctx context.Context, opts *reposInstallConfig) error { continue } - _, schedErr := setupGitLabPipelineSchedules(ctx, fc.Client, glClient, printer, r.Owner, r.Repo, targetRepo.DefaultBranch) + schedErr := setupGitLabPipelineSchedules(ctx, fc.Client, printer, r.Owner, r.Repo, targetRepo.DefaultBranch) if schedErr != nil { printer.StepWarn(fmt.Sprintf("[%s] Pipeline schedule setup failed: %v", repoFullName, schedErr)) } diff --git a/internal/cli/repos_gitlab.go b/internal/cli/repos_gitlab.go index 32b3576af8..681c580b25 100644 --- a/internal/cli/repos_gitlab.go +++ b/internal/cli/repos_gitlab.go @@ -217,18 +217,10 @@ func storeSecretManagerToken(ctx context.Context, gcpClient gcf.GCFClient, print return nil } -// setupGitLabPipelineSchedules creates pipeline schedules for polling. -// Enterprise instances get dual fast/full poll schedules; free/CE instances -// get a single hourly schedule. -func setupGitLabPipelineSchedules(ctx context.Context, client forge.Client, glClient *gitlab.LiveClient, printer *ui.Printer, owner, repo, defaultBranch string) (bool, error) { - printer.StepStart("Detecting GitLab tier") - isEnterprise := glClient != nil && glClient.IsEnterprise(ctx) - if isEnterprise { - printer.StepDone("Detected tier: enterprise") - } else { - printer.StepDone("Detected tier: free") - } - +// setupGitLabPipelineSchedules creates a single pipeline schedule for +// polling. The schedule runs every 5 minutes; the poller auto-promotes +// to a full poll when 15+ minutes have elapsed since the last full poll. +func setupGitLabPipelineSchedules(ctx context.Context, client forge.Client, printer *ui.Printer, owner, repo, defaultBranch string) error { // Delete existing fullsend schedules to avoid duplicates on re-install. existing, listErr := client.ListPipelineSchedules(ctx, owner, repo) if listErr != nil { @@ -245,34 +237,15 @@ func setupGitLabPipelineSchedules(ctx context.Context, client forge.Client, glCl } } - printer.StepStart("Creating pipeline schedules") - if isEnterprise { - fastID, err := client.CreatePipelineSchedule(ctx, owner, repo, defaultBranch, - "fullsend fast poll", "*/5 * * * *", - map[string]string{"FULLSEND_POLL_MODE": "fast"}) - if err != nil { - printer.StepFail("Failed to create fast poll schedule") - return isEnterprise, fmt.Errorf("creating fast poll schedule: %w", err) - } - fullID, err := client.CreatePipelineSchedule(ctx, owner, repo, defaultBranch, - "fullsend full poll", "*/15 * * * *", - map[string]string{"FULLSEND_POLL_MODE": "full"}) - if err != nil { - printer.StepFail("Failed to create full poll schedule") - return isEnterprise, fmt.Errorf("creating full poll schedule: %w", err) - } - printer.StepDone(fmt.Sprintf("Created dual poll schedules (fast: ID %d, full: ID %d)", fastID, fullID)) - } else { - scheduleID, err := client.CreatePipelineSchedule(ctx, owner, repo, defaultBranch, - "fullsend poll", "0 * * * *", nil) - if err != nil { - printer.StepFail("Failed to create poll schedule") - return isEnterprise, fmt.Errorf("creating poll schedule: %w", err) - } - printer.StepDone(fmt.Sprintf("Created hourly poll schedule (ID %d)", scheduleID)) + printer.StepStart("Creating pipeline schedule") + scheduleID, err := client.CreatePipelineSchedule(ctx, owner, repo, defaultBranch, + "fullsend poll", "*/5 * * * *", nil) + if err != nil { + printer.StepFail("Failed to create poll schedule") + return fmt.Errorf("creating poll schedule: %w", err) } - - return isEnterprise, nil + printer.StepDone(fmt.Sprintf("Created poll schedule (ID %d)", scheduleID)) + return nil } // cleanupGitLabPipelineSchedules removes all fullsend-prefixed pipeline diff --git a/internal/cli/repos_gitlab_test.go b/internal/cli/repos_gitlab_test.go index 0b680bb968..6307179ea6 100644 --- a/internal/cli/repos_gitlab_test.go +++ b/internal/cli/repos_gitlab_test.go @@ -251,125 +251,42 @@ func TestSetupGitLabBotToken(t *testing.T) { func TestSetupGitLabPipelineSchedules(t *testing.T) { ctx := context.Background() - t.Run("enterprise gets dual schedules", func(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": true}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - + t.Run("creates single poll schedule", func(t *testing.T) { fake := &forge.FakeClient{} var buf bytes.Buffer printer := ui.New(&buf) - isEnterprise, err := setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") + err := setupGitLabPipelineSchedules(ctx, fake, printer, "group", "project", "main") require.NoError(t, err) - assert.True(t, isEnterprise) - require.Len(t, fake.CreatedSchedules, 2) - assert.Equal(t, "*/5 * * * *", fake.CreatedSchedules[0].Cron) - assert.Equal(t, "*/15 * * * *", fake.CreatedSchedules[1].Cron) - }) - - t.Run("free tier gets hourly schedule", func(t *testing.T) { - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": false}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - - fake := &forge.FakeClient{} - var buf bytes.Buffer - printer := ui.New(&buf) - - isEnterprise, err := setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") - require.NoError(t, err) - assert.False(t, isEnterprise) require.Len(t, fake.CreatedSchedules, 1) - assert.Equal(t, "0 * * * *", fake.CreatedSchedules[0].Cron) + assert.Equal(t, "*/5 * * * *", fake.CreatedSchedules[0].Cron) + assert.Equal(t, "fullsend poll", fake.CreatedSchedules[0].Description) }) } -func TestSetupGitLabPipelineSchedules_FreeScheduleError(t *testing.T) { +func TestSetupGitLabPipelineSchedules_ScheduleError(t *testing.T) { ctx := context.Background() - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": false}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - fake := forge.NewFakeClient() fake.Errors["CreatePipelineSchedule"] = fmt.Errorf("quota exceeded") var buf bytes.Buffer printer := ui.New(&buf) - _, err = setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") + err := setupGitLabPipelineSchedules(ctx, fake, printer, "group", "project", "main") require.Error(t, err) assert.Contains(t, err.Error(), "creating poll schedule") } -func TestSetupGitLabPipelineSchedules_EnterpriseFastScheduleError(t *testing.T) { - ctx := context.Background() - - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": true}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - - fake := forge.NewFakeClient() - fake.Errors["CreatePipelineSchedule"] = fmt.Errorf("quota exceeded") - var buf bytes.Buffer - printer := ui.New(&buf) - - isEnterprise, err := setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") - require.Error(t, err) - assert.True(t, isEnterprise) - assert.Contains(t, err.Error(), "creating fast poll schedule") -} - func TestSetupGitLabPipelineSchedules_ListError(t *testing.T) { ctx := context.Background() - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": false}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - fake := forge.NewFakeClient() fake.Errors["ListPipelineSchedules"] = fmt.Errorf("forbidden") var buf bytes.Buffer printer := ui.New(&buf) - isEnterprise, err := setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") + err := setupGitLabPipelineSchedules(ctx, fake, printer, "group", "project", "main") require.NoError(t, err) - assert.False(t, isEnterprise) assert.Contains(t, buf.String(), "Could not list existing schedules") } @@ -492,17 +409,6 @@ func TestSetupGitLabBotToken_RevokesExistingBeforeCreate(t *testing.T) { func TestSetupGitLabPipelineSchedules_DeletesExisting(t *testing.T) { ctx := context.Background() - mux := http.NewServeMux() - mux.HandleFunc("/api/v4/metadata", func(w http.ResponseWriter, _ *http.Request) { - w.Header().Set("Content-Type", "application/json") - json.NewEncoder(w).Encode(map[string]any{"enterprise": false}) - }) - srv := httptest.NewServer(mux) - defer srv.Close() - - glClient, err := gitlab.New("test-token", gitlab.WithBaseURL(srv.URL)) - require.NoError(t, err) - fake := &forge.FakeClient{ PipelineSchedules: map[string][]forge.PipelineSchedule{ "group/project": { @@ -514,9 +420,8 @@ func TestSetupGitLabPipelineSchedules_DeletesExisting(t *testing.T) { var buf bytes.Buffer printer := ui.New(&buf) - isEnterprise, err := setupGitLabPipelineSchedules(ctx, fake, glClient, printer, "group", "project", "main") + err := setupGitLabPipelineSchedules(ctx, fake, printer, "group", "project", "main") require.NoError(t, err) - assert.False(t, isEnterprise) assert.Equal(t, []int64{5}, fake.DeletedScheduleIDs, "should delete existing fullsend schedule") require.Len(t, fake.CreatedSchedules, 1) } diff --git a/internal/poll/dispatch_test.go b/internal/poll/dispatch_test.go index 7e58522861..684c4fdd8d 100644 --- a/internal/poll/dispatch_test.go +++ b/internal/poll/dispatch_test.go @@ -364,7 +364,7 @@ func TestDispatch_CreatePipelineErrorPropagates(t *testing.T) { func TestRunCreatePipelineFailureDoesNotAdvanceWatermark(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.pipelineErr = fmt.Errorf("API error: 500 internal server error") @@ -395,7 +395,7 @@ func TestRunCreatePipelineFailureDoesNotAdvanceWatermark(t *testing.T) { func TestRunPartialDispatchFailure(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) // Fail after 1 successful CreatePipeline call. diff --git a/internal/poll/poll.go b/internal/poll/poll.go index a0bb7aa3ca..7e42f94165 100644 --- a/internal/poll/poll.go +++ b/internal/poll/poll.go @@ -2,26 +2,31 @@ package poll import ( "context" + "errors" "fmt" "log" "strings" "time" "github.com/fullsend-ai/fullsend/internal/dispatch" + "github.com/fullsend-ai/fullsend/internal/forge" ) +const defaultFullPollInterval = 15 * time.Minute + // Poller discovers GitLab events and dispatches agent stages. type Poller struct { - client GitLabClient - router dispatch.EventRouter - projectPath string - owner string - repo string - botUserID int - gitlabURL string - opts Options - dispatches []Dispatch - warnedNoHMAC bool + client GitLabClient + router dispatch.EventRouter + projectPath string + owner string + repo string + botUserID int + gitlabURL string + opts Options + dispatches []Dispatch + warnedNoHMAC bool + slashCommandsOnly bool } // New creates a Poller for the given project. @@ -48,11 +53,23 @@ const maxEventRetries = 3 // Run executes a single poll cycle: read watermark, discover events, // filter, deduplicate, convert to NormalizedEvent, route, dispatch, // and advance the watermark. +// +// Poll mode is determined automatically: if 15+ minutes have elapsed +// since the last full poll (or no full poll has ever run), a full poll +// is executed. Otherwise a fast poll (slash commands only) is used. func (p *Poller) Run(ctx context.Context) error { if p.client == nil { return fmt.Errorf("poller requires a GitLab client (Phase 1 wiring incomplete)") } + // Auto-promote to full poll when enough time has elapsed. + p.slashCommandsOnly = !p.shouldFullPoll(ctx) + if p.slashCommandsOnly { + log.Printf("poll mode: fast (slash commands only)") + } else { + log.Printf("poll mode: full (auto-promoted)") + } + lastPollAt, err := p.readWatermark(ctx, p.owner, p.repo) if err != nil { return fmt.Errorf("read watermark: %w", err) @@ -61,7 +78,7 @@ func (p *Poller) Run(ctx context.Context) error { var events []RoutableEvent var labelState LabelState var minSkippedAt time.Time - if p.opts.SlashCommandsOnly { + if p.slashCommandsOnly { events, minSkippedAt, err = p.discoverSlashCommands(ctx, p.owner, p.repo, lastPollAt) } else { events, labelState, minSkippedAt, err = p.discoverAllEvents(ctx, p.owner, p.repo, lastPollAt) @@ -269,6 +286,36 @@ func trackLabelFailure(failedLabelEvents map[int]map[string]bool, event Routable failedLabelEvents[event.IID][event.ChangedLabel] = true } +// shouldFullPoll checks whether enough time has elapsed since the last +// full poll to warrant a full discovery cycle. Returns true when a full +// poll should run (15+ min since last full poll or first run). +func (p *Poller) shouldFullPoll(ctx context.Context) bool { + interval := p.opts.FullPollInterval + if interval == 0 { + interval = defaultFullPollInterval + } + + val, err := p.client.GetCIVariable(ctx, p.owner, p.repo, "FULLSEND_LAST_POLL_AT_FULL") + if err != nil { + if errors.Is(err, forge.ErrNotFound) { + // First run — no full poll has ever completed. + return true + } + // Transient error reading the variable — default to full poll + // to avoid silently skipping events. + log.Printf("WARNING: could not read full-poll watermark: %v (defaulting to full poll)", err) + return true + } + + lastFull, err := time.Parse(time.RFC3339, val) + if err != nil { + log.Printf("WARNING: invalid full-poll watermark %q: %v (defaulting to full poll)", val, err) + return true + } + + return time.Since(lastFull) >= interval +} + // splitOwnerRepo splits "group/subgroup/project" into owner="group/subgroup" and repo="project". func splitOwnerRepo(projectPath string) (string, string) { idx := strings.LastIndex(projectPath, "/") diff --git a/internal/poll/poll_test.go b/internal/poll/poll_test.go index 78de2df547..6aaed50ab9 100644 --- a/internal/poll/poll_test.go +++ b/internal/poll/poll_test.go @@ -33,9 +33,8 @@ func TestSplitOwnerRepo(t *testing.T) { func TestNew(t *testing.T) { mc := newMockClient() p := New(mc, nil, "org/sub/project", Options{ - SlashCommandsOnly: true, - BotUserID: 42, - GitLabURL: "https://gitlab.example.com", + BotUserID: 42, + GitLabURL: "https://gitlab.example.com", }) if p.owner != "org/sub" { t.Errorf("owner = %q, want %q", p.owner, "org/sub") @@ -70,7 +69,7 @@ func (r *stubRouter) Route(_ *dispatch.NormalizedEvent) ([]string, error) { func TestRunEmptyPoll(t *testing.T) { now := time.Now() mc := newMockClient() - mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = now.Add(-10 * time.Minute).Format(time.RFC3339) + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = now.Add(-20 * time.Minute).Format(time.RFC3339) p := New(mc, nil, "org/project", Options{}) @@ -89,14 +88,15 @@ func TestRunEmptyPoll(t *testing.T) { } } -func TestRunSlashCommandsOnlyMode(t *testing.T) { +func TestRunAutoPromoteFastMode(t *testing.T) { + // When the full-poll watermark is recent (< 15 min), the poller + // should auto-select fast mode (slash commands only). now := time.Now() mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = now.Add(-5 * time.Minute).Format(time.RFC3339) mc.variables["FULLSEND_LAST_POLL_AT_FAST"] = now.Add(-5 * time.Minute).Format(time.RFC3339) - p := New(mc, nil, "org/project", Options{ - SlashCommandsOnly: true, - }) + p := New(mc, nil, "org/project", Options{}) err := p.Run(context.Background()) if err != nil { @@ -108,6 +108,42 @@ func TestRunSlashCommandsOnlyMode(t *testing.T) { } } +func TestRunAutoPromoteFullMode(t *testing.T) { + // When the full-poll watermark is old (>= 15 min), the poller + // should auto-promote to full mode. + now := time.Now() + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = now.Add(-20 * time.Minute).Format(time.RFC3339) + + p := New(mc, nil, "org/project", Options{}) + + err := p.Run(context.Background()) + if err != nil { + t.Fatalf("Run() error: %v", err) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FULL"]; !ok { + t.Error("full watermark not updated after auto-promotion") + } +} + +func TestRunAutoPromoteFirstRun(t *testing.T) { + // When no full-poll watermark exists (first run), the poller + // should default to full mode. + mc := newMockClient() + + p := New(mc, nil, "org/project", Options{}) + + err := p.Run(context.Background()) + if err != nil { + t.Fatalf("Run() error: %v", err) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FULL"]; !ok { + t.Error("full watermark not updated on first run") + } +} + func TestTrackFailure(t *testing.T) { var min time.Time t1 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) @@ -149,9 +185,83 @@ func TestTrackLabelFailure(t *testing.T) { } } +func TestShouldFullPoll_FirstRun(t *testing.T) { + mc := newMockClient() + // No FULLSEND_LAST_POLL_AT_FULL variable → first run. + p := newTestPoller(mc, Options{}) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll on first run") + } +} + +func TestShouldFullPoll_RecentFullPoll(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = time.Now().Add(-5 * time.Minute).Format(time.RFC3339) + + p := newTestPoller(mc, Options{}) + if p.shouldFullPoll(context.Background()) { + t.Error("expected fast poll when full poll was recent") + } +} + +func TestShouldFullPoll_StaleFullPoll(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = time.Now().Add(-20 * time.Minute).Format(time.RFC3339) + + p := newTestPoller(mc, Options{}) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll when 20 minutes have elapsed") + } +} + +func TestShouldFullPoll_ExactBoundary(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = time.Now().Add(-15 * time.Minute).Format(time.RFC3339) + + p := newTestPoller(mc, Options{}) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll at exactly 15-minute boundary") + } +} + +func TestShouldFullPoll_CustomInterval(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = time.Now().Add(-8 * time.Minute).Format(time.RFC3339) + + p := newTestPoller(mc, Options{FullPollInterval: 10 * time.Minute}) + if p.shouldFullPoll(context.Background()) { + t.Error("expected fast poll when only 8 min elapsed with 10 min interval") + } + + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = time.Now().Add(-12 * time.Minute).Format(time.RFC3339) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll when 12 min elapsed with 10 min interval") + } +} + +func TestShouldFullPoll_InvalidTimestamp(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = "not-a-timestamp" + + p := newTestPoller(mc, Options{}) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll when timestamp is unparseable") + } +} + +func TestShouldFullPoll_ReadError(t *testing.T) { + mc := newMockClient() + mc.variableErr["FULLSEND_LAST_POLL_AT_FULL"] = fmt.Errorf("network failure") + + p := newTestPoller(mc, Options{}) + if !p.shouldFullPoll(context.Background()) { + t.Error("expected full poll on transient read error") + } +} + func TestRunFullPollWithRouterAndDispatch(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -193,7 +303,7 @@ func TestRunFullPollWithRouterAndDispatch(t *testing.T) { func TestRunMultipleStages(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -236,7 +346,7 @@ func TestRunMultipleStages(t *testing.T) { func TestRunLabelEventThreadsActorID(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -275,7 +385,7 @@ func TestRunLabelEventThreadsActorID(t *testing.T) { func TestRunNoMatchingStages(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -302,7 +412,7 @@ func TestRunNoMatchingStages(t *testing.T) { func TestRunRouterError(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -324,7 +434,7 @@ func TestRunRouterError(t *testing.T) { func TestRunConversionErrorSkipsEvent(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -349,7 +459,7 @@ func TestRunConversionErrorSkipsEvent(t *testing.T) { func TestRunAllEventsFailWatermarkNotAdvanced(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -377,7 +487,7 @@ func TestRunAllEventsFailWatermarkNotAdvanced(t *testing.T) { func TestRunNilRouter(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -398,7 +508,7 @@ func TestRunNilRouter(t *testing.T) { func TestRunIdempotentSecondPoll(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -441,7 +551,7 @@ func TestRunEntityDedup_MultipleNotesOnSameIssue(t *testing.T) { // should dispatch only one pipeline. The second note is skipped by // entity-level deduplication (stage:issue-3). now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -473,7 +583,7 @@ func TestRunEntityDedup_MultipleNotesOnSameIssue(t *testing.T) { func TestRunEntityDedup_DifferentIssues(t *testing.T) { // Notes on different issues should each dispatch their own pipeline. now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -508,7 +618,7 @@ func TestRunEntityDedup_DifferentStagesSameIssue(t *testing.T) { // Two notes on the same issue routed to different stages should // each dispatch (entity key includes the stage). now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ @@ -536,7 +646,7 @@ func TestRunEntityDedup_DifferentStagesSameIssue(t *testing.T) { func TestRunLabelFailureRollback(t *testing.T) { now := time.Now().Truncate(time.Second) - since := now.Add(-10 * time.Minute) + since := now.Add(-20 * time.Minute) mc := newMockClient() mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) mc.issues = []Issue{ diff --git a/internal/poll/state.go b/internal/poll/state.go index a4a275391a..ca187b4550 100644 --- a/internal/poll/state.go +++ b/internal/poll/state.go @@ -31,7 +31,7 @@ func (p *Poller) readWatermark(ctx context.Context, owner, repo string) (time.Ti // watermarkVarName returns the CI variable name used for the poll watermark. // Slash-command-only mode uses a faster polling cadence with its own variable. func (p *Poller) watermarkVarName() string { - if p.opts.SlashCommandsOnly { + if p.slashCommandsOnly { return "FULLSEND_LAST_POLL_AT_FAST" } return "FULLSEND_LAST_POLL_AT_FULL" @@ -134,7 +134,7 @@ func (p *Poller) isIssueClosed(ctx context.Context, owner, repo string, iid int) // dispatchedKeysVarName returns the per-mode CI variable name for dispatched keys. func (p *Poller) dispatchedKeysVarName() string { - if p.opts.SlashCommandsOnly { + if p.slashCommandsOnly { return "FULLSEND_DISPATCHED_KEYS_FAST" } return "FULLSEND_DISPATCHED_KEYS_FULL" @@ -179,7 +179,7 @@ func (p *Poller) persistDispatchedKeys(ctx context.Context, owner, repo string, // failedKeysVarName returns the CI variable name for failed event retry counts. func (p *Poller) failedKeysVarName() string { - if p.opts.SlashCommandsOnly { + if p.slashCommandsOnly { return "FULLSEND_FAILED_KEYS_FAST" } return "FULLSEND_FAILED_KEYS_FULL" diff --git a/internal/poll/state_test.go b/internal/poll/state_test.go index 9ab534443d..8b14fbf2bf 100644 --- a/internal/poll/state_test.go +++ b/internal/poll/state_test.go @@ -82,7 +82,8 @@ func TestReadWatermark_ClientError(t *testing.T) { // --- watermarkVarName tests --- func TestWatermarkVarName_FastMode(t *testing.T) { - p := newTestPoller(nil, Options{SlashCommandsOnly: true}) + p := newTestPoller(nil, Options{}) + p.slashCommandsOnly = true got := p.watermarkVarName() if got != "FULLSEND_LAST_POLL_AT_FAST" { t.Errorf("got %q, want FULLSEND_LAST_POLL_AT_FAST", got) @@ -90,7 +91,8 @@ func TestWatermarkVarName_FastMode(t *testing.T) { } func TestWatermarkVarName_FullMode(t *testing.T) { - p := newTestPoller(nil, Options{SlashCommandsOnly: false}) + p := newTestPoller(nil, Options{}) + p.slashCommandsOnly = false got := p.watermarkVarName() if got != "FULLSEND_LAST_POLL_AT_FULL" { t.Errorf("got %q, want FULLSEND_LAST_POLL_AT_FULL", got) @@ -118,7 +120,8 @@ func TestUpdateWatermark_StoresRFC3339(t *testing.T) { func TestUpdateWatermark_UsesFastVarForSlashOnly(t *testing.T) { mc := newMockClient() - p := newTestPoller(mc, Options{SlashCommandsOnly: true}) + p := newTestPoller(mc, Options{}) + p.slashCommandsOnly = true ts := time.Date(2025, 7, 1, 14, 30, 0, 0, time.UTC) err := p.updateWatermark(context.Background(), "testgroup", "testrepo", ts) diff --git a/internal/poll/types.go b/internal/poll/types.go index 5ec797d161..b1eb082507 100644 --- a/internal/poll/types.go +++ b/internal/poll/types.go @@ -7,12 +7,12 @@ import ( // Options configures the poller. type Options struct { - SlashCommandsOnly bool - BotUserID int - GitLabURL string - PipelineRef string // git ref for API-triggered pipelines (required; resolved at CLI wiring time) - PollJobURL string // back-link to the poller CI job (optional; from CI_JOB_URL) - DispatchSecret string // HMAC shared secret for signing dispatch variables (optional; from FULLSEND_DISPATCH_SECRET) + BotUserID int + GitLabURL string + PipelineRef string // git ref for API-triggered pipelines (required; resolved at CLI wiring time) + PollJobURL string // back-link to the poller CI job (optional; from CI_JOB_URL) + DispatchSecret string // HMAC shared secret for signing dispatch variables (optional; from FULLSEND_DISPATCH_SECRET) + FullPollInterval time.Duration // interval between full polls; 0 uses defaultFullPollInterval (15m) } // RoutableEvent is an intermediate representation of a detected change,