diff --git a/internal/cli/admin.go b/internal/cli/admin.go index fcc9af3fc..9a8a688e9 100644 --- a/internal/cli/admin.go +++ b/internal/cli/admin.go @@ -2543,53 +2543,36 @@ func saveRepoConfig(ctx context.Context, client forge.Client, printer *ui.Printe // config.yaml push, waits for it to complete, and prints any PR URLs from its // annotations. func awaitRepoMaintenance(ctx context.Context, client forge.Client, printer *ui.Printer, org string, dispatchTime time.Time) { - awaitRepoMaintenanceWithInterval(ctx, client, printer, org, dispatchTime, 5*time.Second, 36) + awaitRepoMaintenanceWithConfig(ctx, client, printer, org, dispatchTime, layers.DefaultPollConfig()) } func awaitRepoMaintenanceWithInterval(ctx context.Context, client forge.Client, printer *ui.Printer, org string, dispatchTime time.Time, pollInterval time.Duration, maxAttempts int) { + // Map legacy interval/attempts to a PollConfig and delegate. + cfg := layers.PollConfig{ + InitialInterval: pollInterval, + MaxInterval: pollInterval, // no backoff growth for legacy callers + Timeout: pollInterval * time.Duration(maxAttempts), + } + awaitRepoMaintenanceWithConfig(ctx, client, printer, org, dispatchTime, cfg) +} + +func awaitRepoMaintenanceWithConfig(ctx context.Context, client forge.Client, printer *ui.Printer, org string, dispatchTime time.Time, cfg layers.PollConfig) { printer.Blank() printer.StepStart("Watching repo-maintenance workflow") - // Poll for a workflow run created after dispatchTime. - var run *forge.WorkflowRun - for attempt := range maxAttempts { - select { - case <-ctx.Done(): - printer.StepWarn("context cancelled while waiting for workflow") - return - case <-time.After(pollInterval): - } - - runs, err := client.ListWorkflowRuns(ctx, org, forge.ConfigRepoName, "repo-maintenance.yml") - if err != nil { - printer.StepInfo(fmt.Sprintf("waiting for workflow run (attempt %d)...", attempt+1)) - continue - } + run, err := layers.AwaitWorkflowCompletion( + ctx, client, org, forge.ConfigRepoName, "repo-maintenance.yml", + dispatchTime, cfg, + func(msg string) { printer.StepInfo(msg) }, + ) - for i := range runs { - r := &runs[i] - runTime, parseErr := time.Parse(time.RFC3339, r.CreatedAt) - if parseErr != nil { - continue - } - if runTime.Before(dispatchTime) { - continue - } - if r.Status == "completed" { - run = r - break - } - printer.StepInfo(fmt.Sprintf("workflow run: %s (%s)", r.HTMLURL, r.Status)) - break // found our run, keep waiting - } - if run != nil { - break + if err != nil { + if ctx.Err() != nil { + printer.StepWarn("context cancelled while waiting for workflow") + } else { + printer.StepWarn("timed out waiting for repo-maintenance workflow") + printer.StepInfo("check the repo-maintenance workflow in .fullsend for results") } - } - - if run == nil { - printer.StepWarn("timed out waiting for repo-maintenance workflow") - printer.StepInfo("check the repo-maintenance workflow in .fullsend for results") return } diff --git a/internal/layers/enrollment.go b/internal/layers/enrollment.go index d418ec442..c15677589 100644 --- a/internal/layers/enrollment.go +++ b/internal/layers/enrollment.go @@ -105,39 +105,13 @@ func (l *EnrollmentLayer) Install(ctx context.Context) error { } // awaitWorkflowRun polls for a repo-maintenance workflow run created after -// dispatchTime and waits for it to complete. +// dispatchTime and waits for it to complete, using exponential backoff. func (l *EnrollmentLayer) awaitWorkflowRun(ctx context.Context, dispatchTime time.Time) (*forge.WorkflowRun, error) { - for attempt := range 36 { // 3 minutes max - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-time.After(5 * time.Second): - } - - runs, err := l.client.ListWorkflowRuns(ctx, l.org, forge.ConfigRepoName, repoMaintenanceWorkflow) - if err != nil { - l.ui.StepInfo(fmt.Sprintf("waiting for workflow run (attempt %d)...", attempt+1)) - continue - } - - for i := range runs { - run := &runs[i] - runTime, parseErr := time.Parse(time.RFC3339, run.CreatedAt) - if parseErr != nil { - continue - } - if runTime.Before(dispatchTime) { - continue - } - - if run.Status == "completed" { - return run, nil - } - l.ui.StepInfo(fmt.Sprintf("workflow run: %s (%s)", run.HTMLURL, run.Status)) - break // found our run, keep waiting - } - } - return nil, fmt.Errorf("timed out waiting for repo-maintenance workflow") + return AwaitWorkflowCompletion( + ctx, l.client, l.org, forge.ConfigRepoName, repoMaintenanceWorkflow, + dispatchTime, DefaultPollConfig(), + func(msg string) { l.ui.StepInfo(msg) }, + ) } // showWorkflowLogs fetches and displays workflow run logs locally so the user diff --git a/internal/layers/workflowpoll.go b/internal/layers/workflowpoll.go new file mode 100644 index 000000000..7cdadc6d1 --- /dev/null +++ b/internal/layers/workflowpoll.go @@ -0,0 +1,107 @@ +package layers + +import ( + "context" + "fmt" + "time" + + "github.com/fullsend-ai/fullsend/internal/forge" +) + +// PollConfig configures workflow polling behavior. +type PollConfig struct { + // InitialInterval is the delay before the first poll and the starting + // interval for exponential backoff. + InitialInterval time.Duration + + // MaxInterval caps how long the backoff can grow between polls. + MaxInterval time.Duration + + // Timeout is the maximum wall-clock time to wait for the workflow to + // complete before returning an error. + Timeout time.Duration +} + +// DefaultPollConfig returns the default polling configuration: +// 5s initial interval, 15s max interval, 3-minute timeout. +func DefaultPollConfig() PollConfig { + return PollConfig{ + InitialInterval: 5 * time.Second, + MaxInterval: 15 * time.Second, + Timeout: 3 * time.Minute, + } +} + +// ProgressFunc is an optional callback invoked with status messages during +// polling. Callers can use it to print progress to users. +type ProgressFunc func(msg string) + +// AwaitWorkflowCompletion polls for a workflow run created after dispatchTime +// and waits for it to complete, using exponential backoff. It returns the +// completed run or an error if the context is cancelled or the timeout expires. +// +// The onProgress callback, if non-nil, is called with status messages during +// polling (e.g., "waiting for workflow run" or intermediate run status). +func AwaitWorkflowCompletion( + parentCtx context.Context, + client forge.Client, + org, repo, workflowFile string, + dispatchTime time.Time, + cfg PollConfig, + onProgress ProgressFunc, +) (*forge.WorkflowRun, error) { + ctx, cancel := context.WithTimeout(parentCtx, cfg.Timeout) + defer cancel() + + interval := cfg.InitialInterval + + for attempt := 0; ; attempt++ { + select { + case <-ctx.Done(): + if parentCtx.Err() != nil { + return nil, parentCtx.Err() + } + return nil, fmt.Errorf("timed out waiting for %s workflow", workflowFile) + case <-time.After(interval): + } + + runs, err := client.ListWorkflowRuns(ctx, org, repo, workflowFile) + if err != nil { + if onProgress != nil { + onProgress(fmt.Sprintf("waiting for workflow run (attempt %d)...", attempt+1)) + } + interval = nextInterval(interval, cfg.MaxInterval) + continue + } + + for i := range runs { + run := &runs[i] + runTime, parseErr := time.Parse(time.RFC3339, run.CreatedAt) + if parseErr != nil { + continue + } + if runTime.Before(dispatchTime) { + continue + } + + if run.Status == "completed" { + return run, nil + } + if onProgress != nil { + onProgress(fmt.Sprintf("workflow run: %s (%s)", run.HTMLURL, run.Status)) + } + break // found our run, keep waiting + } + + interval = nextInterval(interval, cfg.MaxInterval) + } +} + +// nextInterval doubles the current interval, capping at maxInterval. +func nextInterval(current, maxInterval time.Duration) time.Duration { + next := current * 2 + if next > maxInterval { + return maxInterval + } + return next +} diff --git a/internal/layers/workflowpoll_test.go b/internal/layers/workflowpoll_test.go new file mode 100644 index 000000000..3a6623666 --- /dev/null +++ b/internal/layers/workflowpoll_test.go @@ -0,0 +1,207 @@ +package layers + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/fullsend-ai/fullsend/internal/forge" +) + +// errorCountClient wraps FakeClient, returning transient errors on the first +// N calls to ListWorkflowRuns before delegating to the embedded fake. +type errorCountClient struct { + *forge.FakeClient + errorsLeft int + mu sync.Mutex +} + +func (c *errorCountClient) ListWorkflowRuns(ctx context.Context, owner, repo, wf string) ([]forge.WorkflowRun, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.errorsLeft > 0 { + c.errorsLeft-- + return nil, fmt.Errorf("transient API error") + } + return c.FakeClient.ListWorkflowRuns(ctx, owner, repo, wf) +} + +func fastPollConfig() PollConfig { + return PollConfig{ + InitialInterval: 1 * time.Millisecond, + MaxInterval: 2 * time.Millisecond, + Timeout: 500 * time.Millisecond, + } +} + +func TestAwaitWorkflowCompletion_Success(t *testing.T) { + now := time.Now().UTC() + client := &forge.FakeClient{ + WorkflowRuns: map[string]*forge.WorkflowRun{ + "org/.fullsend/repo-maintenance.yml": { + ID: 1, + Status: "completed", + Conclusion: "success", + HTMLURL: "https://example.com/runs/1", + CreatedAt: now.Add(time.Minute).Format(time.RFC3339), + }, + }, + } + + run, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "repo-maintenance.yml", + now, fastPollConfig(), nil, + ) + + require.NoError(t, err) + assert.Equal(t, 1, run.ID) + assert.Equal(t, "success", run.Conclusion) +} + +func TestAwaitWorkflowCompletion_Timeout(t *testing.T) { + client := forge.NewFakeClient() + now := time.Now().UTC() + + _, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "repo-maintenance.yml", + now, fastPollConfig(), nil, + ) + + require.Error(t, err) + assert.Contains(t, err.Error(), "timed out") +} + +func TestAwaitWorkflowCompletion_ContextCancelled(t *testing.T) { + client := forge.NewFakeClient() + now := time.Now().UTC() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := AwaitWorkflowCompletion( + ctx, client, + "org", ".fullsend", "repo-maintenance.yml", + now, fastPollConfig(), nil, + ) + + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled) +} + +func TestAwaitWorkflowCompletion_ProgressCallback(t *testing.T) { + now := time.Now().UTC() + client := &forge.FakeClient{ + WorkflowRuns: map[string]*forge.WorkflowRun{ + "org/.fullsend/ci.yml": { + ID: 1, + Status: "completed", + HTMLURL: "https://example.com/runs/1", + CreatedAt: now.Add(time.Minute).Format(time.RFC3339), + }, + }, + } + + var messages []string + run, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "ci.yml", + now, fastPollConfig(), + func(msg string) { messages = append(messages, msg) }, + ) + + require.NoError(t, err) + assert.NotNil(t, run) +} + +func TestAwaitWorkflowCompletion_InProgressRunTimesOut(t *testing.T) { + now := time.Now().UTC() + client := &forge.FakeClient{ + WorkflowRuns: map[string]*forge.WorkflowRun{ + "org/.fullsend/ci.yml": { + ID: 1, + Status: "in_progress", + HTMLURL: "https://example.com/runs/1", + CreatedAt: now.Add(time.Minute).Format(time.RFC3339), + }, + }, + } + + var messages []string + _, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "ci.yml", + now, fastPollConfig(), + func(msg string) { messages = append(messages, msg) }, + ) + + require.Error(t, err) + assert.Contains(t, err.Error(), "timed out") + // Should have received at least one in-progress status message. + require.NotEmpty(t, messages) + assert.Contains(t, messages[0], "in_progress") +} + +func TestAwaitWorkflowCompletion_NoMatchingRunByDispatchTime(t *testing.T) { + now := time.Now().UTC() + // Run was created BEFORE dispatchTime — should not match. + client := &forge.FakeClient{ + WorkflowRuns: map[string]*forge.WorkflowRun{ + "org/.fullsend/ci.yml": { + ID: 1, + Status: "completed", + HTMLURL: "https://example.com/runs/1", + CreatedAt: now.Add(-time.Hour).Format(time.RFC3339), + }, + }, + } + + _, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "ci.yml", + now, fastPollConfig(), nil, + ) + + require.Error(t, err) + assert.Contains(t, err.Error(), "timed out") +} + +func TestAwaitWorkflowCompletion_ErrorThenSuccess(t *testing.T) { + now := time.Now().UTC() + inner := &forge.FakeClient{ + WorkflowRuns: map[string]*forge.WorkflowRun{ + "org/.fullsend/ci.yml": { + ID: 1, + Status: "completed", + Conclusion: "success", + HTMLURL: "https://example.com/runs/1", + CreatedAt: now.Add(time.Minute).Format(time.RFC3339), + }, + }, + } + client := &errorCountClient{FakeClient: inner, errorsLeft: 2} + + var messages []string + run, err := AwaitWorkflowCompletion( + context.Background(), client, + "org", ".fullsend", "ci.yml", + now, fastPollConfig(), + func(msg string) { messages = append(messages, msg) }, + ) + + require.NoError(t, err) + assert.Equal(t, 1, run.ID) + // The 2 transient errors should have produced progress messages. + assert.GreaterOrEqual(t, len(messages), 2) +} + +func TestNextInterval(t *testing.T) { + assert.Equal(t, 4*time.Second, nextInterval(2*time.Second, 15*time.Second)) + assert.Equal(t, 15*time.Second, nextInterval(10*time.Second, 15*time.Second)) + assert.Equal(t, 15*time.Second, nextInterval(15*time.Second, 15*time.Second)) +}