Handle unauth scheduler ctx

Author: maggie-lou

deps/go_deps.MODULE.bazel

@@ -255,7 +255,7 @@ use_repo(
     "com_github_prometheus_client_model",
     "com_github_prometheus_common",
     "com_github_rantav_go_grpc_channelz",
-     "com_github_robfig_cron_v3",
+    "com_github_robfig_cron_v3",
     "com_github_roaringbitmap_roaring",
     "com_github_rs_zerolog",
     "com_github_shirou_gopsutil_v3",

enterprise/server/githubapp/githubapp.go

@@ -619,6 +619,22 @@ func (a *GitHubApp) GetInstallationTokenForStatusReportingOnly(ctx context.Conte
 	return tok, nil
 }
 
+func (a *GitHubApp) GetDefaultBranch(ctx context.Context, repoURL string, accessToken string) (string, error) {
+	parsedURL, err := gitutil.ParseGitHubRepoURL(repoURL)
+	if err != nil {
+		return "", status.InvalidArgumentErrorf("invalid repo URL %q: %s", repoURL, err)
+	}
+	client, err := a.newAuthenticatedClient(ctx, accessToken)
+	if err != nil {
+		return "", status.WrapError(err, "create GitHub client")
+	}
+	repo, _, err := client.Repositories.Get(ctx, parsedURL.Owner, parsedURL.Repo)
+	if err != nil {
+		return "", err
+	}
+	return repo.GetDefaultBranch(), nil
+}
+
 func (a *GitHubApp) GetRepositoryInstallationToken(ctx context.Context, groupID, repoURL string) (string, error) {
 	if err := authutil.AuthorizeGroupAccess(ctx, a.env, groupID); err != nil {
 		return "", err

enterprise/server/webhooks/webhook_data/webhook_data.go

@@ -9,16 +9,18 @@ import (
 var (
 	// EventName holds canonical webhook event name constants.
 	EventName struct {
-		Push           string
-		PullRequest    string
-		ManualDispatch string
+		Push              string
+		PullRequest       string
+		ManualDispatch    string
+		ScheduledDispatch string
 	}
 )
 
 func init() {
 	EventName.Push = "push"
 	EventName.PullRequest = "pull_request"
 	EventName.ManualDispatch = "manual_dispatch"
+	EventName.ScheduledDispatch = "scheduled"
 }
 
 func DebugString(wd *interfaces.WebhookData) string {

enterprise/server/workflow/config/config.go

@@ -435,8 +435,8 @@ func GetDefault(targetRepoDefaultBranch string) *BuildBuddyConfig {
 // MatchesAnyTrigger returns whether the action is triggered by the event
 // published to the given branch or tag.
 func MatchesAnyTrigger(action *Action, event, branch, tag string) bool {
-	// If user has manually requested action dispatch, always run it
-	if event == webhook_data.EventName.ManualDispatch {
+	// If user has manually or scheduled action dispatch, always run it
+	if event == webhook_data.EventName.ManualDispatch || event == webhook_data.EventName.ScheduledDispatch {
 		return true
 	}
 

enterprise/server/workflow/service/service.go

@@ -1854,11 +1854,15 @@ func (ws *workflowService) RunScheduledWorkflows(ctx context.Context) error {
 			log.CtxWarningf(ctx, "Failed to claim scheduled workflow: %s", err)
 			continue
 		}
+		// If there are no eligible scheduled workflows, stop polling until the next time the scheduler runs.
 		if scheduled == nil {
 			break
 		}
 		if err := ws.dispatchScheduledWorkflow(ctx, scheduled); err != nil {
 			log.CtxErrorf(ctx, "Failed to dispatch scheduled workflow %s: %s", scheduled.ScheduleID, err)
+			if err := ws.unclaimScheduledWorkflow(ctx, scheduled.ScheduleID); err != nil {
+				log.CtxWarningf(ctx, "Failed to unclaim scheduled workflow %s: %s", scheduled.ScheduleID, err)
+			}
 			continue
 		}
 	}
@@ -1877,7 +1881,7 @@ func (ws *workflowService) claimScheduledWorkflow(ctx context.Context) (*tables.
 			WHERE next_run_usec <= ?
 			  AND (lease_expires_usec = 0 OR lease_expires_usec <= ?)
 			ORDER BY next_run_usec ASC
-			LIMIT 1`+dbh.SelectForUpdateModifier(), nowUsec, nowUsec).Take(scheduled)
+			LIMIT 1 `+dbh.SelectForUpdateModifier(), nowUsec, nowUsec).Take(scheduled)
 		if err != nil {
 			if db.IsRecordNotFound(err) {
 				scheduled = nil
@@ -1925,30 +1929,68 @@ func (ws *workflowService) unclaimScheduledWorkflow(ctx context.Context, schedul
 	return nil
 }
 
+func (ws *workflowService) getWorkflowForScheduledDispatch(ctx context.Context, groupID, repoURL string) (*tables.Workflow, error) {
+	gitRepository := &tables.GitRepository{}
+	err := ws.env.GetDBHandle().NewQuery(ctx, "workflow_service_get_for_scheduled_dispatch").Raw(`
+		SELECT * FROM "GitRepositories"
+		WHERE group_id = ?
+		AND repo_url = ?
+	`, groupID, repoURL).Take(gitRepository)
+	if err != nil {
+		return nil, status.WrapErrorf(err, "fetch repo %q", repoURL)
+	}
+	parsedURL, err := gitutil.ParseGitHubRepoURL(repoURL)
+	if err != nil {
+		return nil, status.WrapErrorf(err, "invalid repo URL %q", repoURL)
+	}
+	app, err := ws.env.GetGitHubAppService().GetGitHubAppForOwner(ctx, parsedURL.Owner)
+	if err != nil {
+		return nil, status.WrapErrorf(err, "get GitHub app for owner %q", parsedURL.Owner)
+	}
+	// The cron scheduler does not use an authenticated context, so we use this method.
+	accessToken, err := app.GetInstallationTokenForStatusReportingOnly(ctx, parsedURL.Owner)
+	if err != nil {
+		return nil, status.WrapErrorf(err, "get installation token for owner %q", parsedURL.Owner)
+	}
+	return ws.gitRepositoryWorkflow(gitRepository, accessToken.GetToken()).Workflow, nil
+}
+
 func (ws *workflowService) dispatchScheduledWorkflow(ctx context.Context, scheduled *tables.ScheduledRun) error {
-	wfID := ws.GetLegacyWorkflowIDForGitRepository(scheduled.GroupID, scheduled.RepoURL)
-	rsp, err := ws.ExecuteWorkflow(ctx, &wfpb.ExecuteWorkflowRequest{
-		WorkflowId:    wfID,
-		PushedRepoUrl: scheduled.RepoURL,
-		PushedBranch:  scheduled.Branch,
-		Async:         true,
-		ActionNames:   []string{scheduled.ActionName},
-	})
+	wf, err := ws.getWorkflowForScheduledDispatch(ctx, scheduled.GroupID, scheduled.RepoURL)
 	if err != nil {
-		if err := ws.unclaimScheduledWorkflow(ctx, scheduled.ScheduleID); err != nil {
-			log.CtxWarningf(ctx, "Failed to unclaim scheduled workflow %s: %s", scheduled.ScheduleID, err)
-		}
 		return err
 	}
-	for _, actionStatus := range rsp.GetActionStatuses() {
-		if actionErr := gstatus.FromProto(actionStatus.GetStatus()).Err(); actionErr != nil {
-			if err := ws.unclaimScheduledWorkflow(ctx, scheduled.ScheduleID); err != nil {
-				log.CtxWarningf(ctx, "Failed to unclaim scheduled workflow %s: %s", scheduled.ScheduleID, err)
-			}
-			return status.WrapErrorf(actionErr, "failed to start scheduled workflow action %q", scheduled.ActionName)
-		}
+	defaultBranch, err := ws.getRepoDefaultBranch(ctx, scheduled.RepoURL, wf.AccessToken)
+	if err != nil {
+		return err
+	}
+	wd := &interfaces.WebhookData{
+		EventName:     webhook_data.EventName.ScheduledDispatch,
+		PushedRepoURL: scheduled.RepoURL,
+		PushedBranch:  defaultBranch,
+		TargetRepoURL: scheduled.RepoURL,
+		TargetBranch:  defaultBranch,
+	}
+	apiKey, err := ws.apiKeyForWorkflow(ctx, wf)
+	if err != nil {
+		return err
+	}
+	actions, err := ws.getActions(ctx, wf, wd, []string{scheduled.ActionName})
+	if err != nil {
+		return err
 	}
-	nextRunUsec, err := ws.calculateNextRunTimeUsec(scheduled.CronExpr, scheduled.NextRunUsec)
+	if len(actions) != 1 {
+		return status.InvalidArgumentErrorf("multiple actions named %q found", scheduled.ActionName)
+	}
+	action := actions[0]
+	invocationUUID, err := guuid.NewRandom()
+	if err != nil {
+		return err
+	}
+	if _, err := ws.executeWorkflowAction(ctx, apiKey, wf, wd, true /*isTrusted*/, action, invocationUUID.String(), nil /*extraCIRunnerArgs*/, nil /*env*/, true /*shouldRetry*/); err != nil {
+		return status.WrapErrorf(err, "failed to start scheduled workflow action %q", scheduled.ActionName)
+	}
+	nextRunUsec, err := ws.calculateNextRunTimeUsec(scheduled.CronExpr)
 	if err != nil {
 		alert.CtxUnexpectedEvent(ctx, "Failed to calculate next run time for scheduled workflow %s: %s", scheduled.ScheduleID, err)
 		return err
@@ -1960,12 +2002,27 @@ func (ws *workflowService) dispatchScheduledWorkflow(ctx context.Context, schedu
 	return nil
 }
 
-func (ws *workflowService) calculateNextRunTimeUsec(cronExpr string, lastRunTimeUsec int64) (int64, error) {
+func (ws *workflowService) getRepoDefaultBranch(ctx context.Context, repoURL string, accessToken string) (string, error) {
+	parsedURL, err := gitutil.ParseGitHubRepoURL(repoURL)
+	if err != nil {
+		return "", status.InvalidArgumentErrorf("invalid repo URL %q: %s", repoURL, err)
+	}
+	app, err := ws.env.GetGitHubAppService().GetGitHubAppForOwner(ctx, parsedURL.Owner)
+	if err != nil {
+		return "", status.WrapErrorf(err, "get GitHub app for owner %q", parsedURL.Owner)
+	}
+	return app.GetDefaultBranch(ctx, repoURL, accessToken)
+}
+
+// calculateNextRunTimeUsec uses the given cron expression to return the next
+// scheduled time, using the current time as the minimum.
+func (ws *workflowService) calculateNextRunTimeUsec(cronExpr string) (int64, error) {
 	sched, err := cronParser.Parse(cronExpr)
 	if err != nil {
 		return 0, err
 	}
-	return sched.Next(time.UnixMicro(lastRunTimeUsec)).UnixMicro(), nil
+	now := ws.env.GetClock().Now()
+	return sched.Next(now).UnixMicro(), nil
 }
 
 func (ws *workflowService) advanceWorkflowSchedule(ctx context.Context, scheduleID string, nextRunUsec int64) error {

enterprise/server/workflow/service/service_test.go

@@ -1090,12 +1090,12 @@ func TestAPIDispatch_ActionFiltering(t *testing.T) {
 func TestScheduledWorkflow(t *testing.T) {
 	ctx := context.Background()
 	te := newTestEnv(t)
-	ctx, _, gid := authenticate(t, ctx, te)
 	execClient := te.GetRemoteExecutionClient().(*fakeExecutionClient)
 	te.SetRemoteExecutionClient(execClient)
 	provider := setupFakeGitProvider(t, te)
 	repoURL := makeTempRepo(t)
 	_ = runBBServer(ctx, t, te)
+	authCtx, _, gid := authenticate(t, ctx, te)
 	createWorkflow(t, te, repoURL, gid, false)
 	provider.FileContents = map[string]string{
 		config.FilePath: `
@@ -1119,7 +1119,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Should Run",
-		Branch:     "main",
 		// Run every hour, on the hour.
 		CronExpr:    "0 * * * *",
 		NextRunUsec: now.UnixMicro(),
@@ -1130,7 +1129,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Should Run 2",
-		Branch:     "main",
 		// Run every hour, 5 minutes before the hour.
 		CronExpr:    "55 * * * *",
 		NextRunUsec: fiveMinutesAgo.UnixMicro(),
@@ -1141,7 +1139,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Not Time Yet",
-		Branch:     "main",
 		// Run every day at 7PM UTC.
 		CronExpr:    "0 19 * * *",
 		NextRunUsec: sevenPM.UnixMicro(),
@@ -1152,7 +1149,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Already Leased",
-		Branch:     "main",
 		// Run every hour, on the hour.
 		CronExpr:         "0 * * * *",
 		NextRunUsec:      now.UnixMicro(),
@@ -1164,22 +1160,23 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Expired Lease",
-		Branch:     "main",
 		// Run every hour, on the hour.
 		CronExpr:         "0 * * * *",
 		NextRunUsec:      now.UnixMicro(),
 		LeaseExpiresUsec: now.Add(-1 * time.Minute).UnixMicro(),
 	})
 
-	err := te.GetWorkflowService().RunScheduledWorkflows(ctx)
+	// Intentionally use a non-authenticated context, like the cron scheduler would.
+	err := te.GetWorkflowService().RunScheduledWorkflows(t.Context())
 	require.NoError(t, err)
 
 	expectedActionNames := []string{"Should Run", "Should Run 2", "Expired Lease"}
 	executedActionNames := make([]string, 0, len(expectedActionNames))
 	for range expectedActionNames {
 		select {
 		case execReq := <-execClient.executeRequests:
-			actionName := getExecutedActionName(t, ctx, te, execReq.Payload)
+			// When fetching the executions, make sure to use the authenticated context.
+			actionName := getExecutedActionName(t, authCtx, te, execReq.Payload)
 			executedActionNames = append(executedActionNames, actionName)
 		case <-time.After(5 * time.Second):
 			t.Fatal("timed out waiting for scheduled workflow execution")
@@ -1211,10 +1208,44 @@ actions:
 	require.Equal(t, int64(0), notTimeYet.LeaseExpiresUsec)
 }
 
+func TestScheduledWorkflow_NoEligibleSchedules(t *testing.T) {
+	ctx := context.Background()
+	te := newTestEnv(t)
+	_, _, gid := authenticate(t, ctx, te)
+	repoURL := makeTempRepo(t)
+	_ = runBBServer(ctx, t, te)
+	createWorkflow(t, te, repoURL, gid, false)
+
+	now := time.Date(2026, 1, 2, 15, 0, 0, 0, time.UTC)
+	te.SetClock(clockwork.NewFakeClockAt(now))
+
+	// All next run times are in the future.
+	insertScheduledRun(t, te, &tables.ScheduledRun{
+		ScheduleID:  "future-1",
+		GroupID:     gid,
+		RepoURL:     repoURL,
+		ActionName:  "Some Action",
+		CronExpr:    "0 19 * * *",
+		NextRunUsec: now.Add(1 * time.Hour).UnixMicro(),
+	})
+
+	// Intentionally use a non-authenticated context, like the cron scheduler would.
+	err := te.GetWorkflowService().RunScheduledWorkflows(t.Context())
+	require.NoError(t, err)
+
+	// No executions should have been started.
+	require.Zero(t, len(te.GetRemoteExecutionClient().(*fakeExecutionClient).executeRequests))
+
+	// Schedule rows should be untouched.
+	future1 := getScheduledRun(t, te, "future-1")
+	require.Equal(t, now.Add(1*time.Hour).UnixMicro(), future1.NextRunUsec)
+	require.Equal(t, int64(0), future1.LeaseExpiresUsec)
+}
+
 func TestScheduledWorkflow_ConcurrentServers(t *testing.T) {
 	ctx := context.Background()
 	te := newTestEnv(t)
-	ctx, _, gid := authenticate(t, ctx, te)
+	authCtx, _, gid := authenticate(t, ctx, te)
 	execClient := te.GetRemoteExecutionClient().(*fakeExecutionClient)
 	te.SetRemoteExecutionClient(execClient)
 	provider := setupFakeGitProvider(t, te)
@@ -1239,7 +1270,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Test",
-		Branch:     "main",
 		// Run every hour, on the hour.
 		CronExpr:    "0 * * * *",
 		NextRunUsec: now.UnixMicro(),
@@ -1250,7 +1280,6 @@ actions:
 		GroupID:    gid,
 		RepoURL:    repoURL,
 		ActionName: "Test 2",
-		Branch:     "main",
 		// Run every hour, on the hour.
 		CronExpr:    "0 * * * *",
 		NextRunUsec: now.UnixMicro(),
@@ -1263,7 +1292,8 @@ actions:
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			errCh <- te.GetWorkflowService().RunScheduledWorkflows(ctx)
+			// Intentionally use a non-authenticated context, like the cron scheduler would.
+			errCh <- te.GetWorkflowService().RunScheduledWorkflows(t.Context())
 		}()
 	}
 	wg.Wait()
@@ -1277,7 +1307,8 @@ actions:
 	for range 2 {
 		select {
 		case execReq := <-execClient.executeRequests:
-			actionName := getExecutedActionName(t, ctx, te, execReq.Payload)
+			// When fetching the executions, make sure to use the authenticated context.
+			actionName := getExecutedActionName(t, authCtx, te, execReq.Payload)
 			executedActionNames = append(executedActionNames, actionName)
 		case <-time.After(5 * time.Second):
 			t.Fatal("timed out waiting for scheduled workflow execution")

server/interfaces/interfaces.go

@@ -724,6 +724,9 @@ type GitHubApp interface {
 	// so should be used for status reporting only.
 	GetInstallationTokenForStatusReportingOnly(ctx context.Context, owner string) (*github.InstallationToken, error)
 
+	// GetDefaultBranch returns the default branch for the given repo URL.
+	GetDefaultBranch(ctx context.Context, repoURL string, accessToken string) (string, error)
+
 	// GetRepositoryInstallationToken returns an installation token for the given repo.
 	// The repo must've been imported to BuildBuddy (i.e. a GitRepository row was created).
 	GetRepositoryInstallationToken(ctx context.Context, groupID, repoURL string) (string, error)

server/tables/tables.go

@@ -1469,5 +1469,4 @@ func RegisterTables() {
 	registerTable("UU", &UserUserList{})
 	registerTable("UM", &UserListGroup{})
 	registerTable("WF", &Workflow{})
-	registerTable("SR", &ScheduledRun{})
 }

server/testutil/testgit/BUILD

@@ -12,5 +12,6 @@ go_library(
         "//server/testutil/testfs",
         "//server/testutil/testshell",
         "//server/util/status",
+        "@com_github_google_go_github_v59//github",
     ],
 )