From 5edb45dc1a04971de8c061c41eaf594c6e05778d Mon Sep 17 00:00:00 2001 From: Zettat123 Date: Wed, 22 Jul 2026 14:54:51 -0600 Subject: [PATCH] fix(actions): fail unexpandable reusable workflow callers and decouple the job emitter's cross-run processing (#38565) ## Changes ### 1. Handle reusable workflow expansion failures If a reusable workflow caller job is invalid (e.g. uses a workflow with syntax error), the job emitter should mark it as failed instead of retrying. Related: https://github.com/go-gitea/gitea/pull/38518#discussion_r3608882095 ### 2. No longer process concurrent run inline **Before**: If a run(R1)'s status change unlocks another blocked run(R2) via concurrency group, the job emitter will process R2 in R1's transaction. **Current**: No longer process R2 inline and emit the ID of R2 to let another pass process it. --- models/actions/run_job.go | 8 ++ services/actions/job_emitter.go | 104 +++++++++++------- services/actions/job_emitter_test.go | 93 +++++++++++++++- services/actions/reusable_workflow.go | 55 +++++---- services/actions/reusable_workflow_test.go | 31 ++++++ .../actions_reusable_workflow_test.go | 43 ++++++++ 6 files changed, 270 insertions(+), 64 deletions(-) diff --git a/models/actions/run_job.go b/models/actions/run_job.go index 33696f7bee..f00c5da51b 100644 --- a/models/actions/run_job.go +++ b/models/actions/run_job.go @@ -333,6 +333,14 @@ func GetDirectChildJobsByParent(ctx context.Context, parentJob *ActionRunJob) (A return jobs, nil } +// DeleteDirectChildJobsByParent deletes the direct child jobs of a parent job. +func DeleteDirectChildJobsByParent(ctx context.Context, parentJob *ActionRunJob) error { + _, err := db.GetEngine(ctx). + Where("run_id=? AND parent_job_id=?", parentJob.RunID, parentJob.ID). + Delete(new(ActionRunJob)) + return err +} + // CollectAllDescendantJobs returns every job in `allJobs` that lives under parent's subtree (recursively), excluding `parent` itself func CollectAllDescendantJobs(parent *ActionRunJob, allJobs []*ActionRunJob) []*ActionRunJob { parents := map[int64]bool{parent.ID: true} diff --git a/services/actions/job_emitter.go b/services/actions/job_emitter.go index 1b45eab83f..4003a1ad1a 100644 --- a/services/actions/job_emitter.go +++ b/services/actions/job_emitter.go @@ -15,6 +15,7 @@ import ( "gitea.dev/modules/log" "gitea.dev/modules/queue" "gitea.dev/modules/setting" + "gitea.dev/modules/timeutil" "gitea.dev/modules/util" "xorm.io/builder" @@ -96,7 +97,7 @@ func checkJobsByRunID(ctx context.Context, runID int64) error { continue } if err := EmitJobsIfReadyByRun(rid); err != nil { - log.Error("re-emit run %d after caller expansion: %v", rid, err) + log.Error("re-emit run %d: %v", rid, err) } } NotifyWorkflowJobsAndRunsStatusUpdate(ctx, result.CancelledJobs) @@ -147,57 +148,66 @@ func createCommitStatusesForJobsByRun(ctx context.Context, jobs []*actions_model return nil } -// findBlockedRunIDByConcurrency finds a blocked concurrent run in a repo and returns 0 when there is no blocked run. -func findBlockedRunIDByConcurrency(ctx context.Context, repoID int64, concurrencyGroup string) (int64, error) { +// findConcurrencyWaiterToWake returns a run (other than excludeRunID) blocked on the group that can be woken now +func findConcurrencyWaiterToWake(ctx context.Context, repoID, excludeRunID int64, concurrencyGroup string) (int64, error) { if concurrencyGroup == "" { return 0, nil } + + // The slot should be free before any waiter can proceed. + holderAttempts, holderJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusRunning, actions_model.StatusCancelling}) + if err != nil { + return 0, fmt.Errorf("find concurrency-group holders: %w", err) + } + if len(holderAttempts) > 0 || len(holderJobs) > 0 { + return 0, nil + } + cAttempts, cJobs, err := actions_model.GetConcurrentRunAttemptsAndJobs(ctx, repoID, concurrencyGroup, []actions_model.Status{actions_model.StatusBlocked}) if err != nil { - return 0, fmt.Errorf("find concurrent runs and jobs: %w", err) + return 0, fmt.Errorf("find blocked concurrent runs: %w", err) } - - if len(cAttempts) > 0 { - return cAttempts[0].RunID, nil + for _, a := range cAttempts { + if a.RunID != excludeRunID { + return a.RunID, nil + } } - if len(cJobs) > 0 { - return cJobs[0].RunID, nil + for _, j := range cJobs { + if j.RunID != excludeRunID { + return j.RunID, nil + } } - return 0, nil } -func checkBlockedConcurrentRun(ctx context.Context, repoID, runID int64) (*jobsCheckResult, error) { - concurrentRun, err := actions_model.GetRunByRepoAndID(ctx, repoID, runID) - if err != nil { - return nil, fmt.Errorf("get run %d: %w", runID, err) - } - if concurrentRun.NeedApproval { - return &jobsCheckResult{}, nil - } - - return checkJobsOfCurrentRunAttempt(ctx, concurrentRun) -} - -// checkRunConcurrency rechecks runs blocked by concurrency that may become unblocked after the current run releases a workflow-level or job-level concurrency group. -// RunIDsToReEmit propagates from inner checkJobsOfCurrentRunAttempt calls; see that function's doc. +// checkRunConcurrency wakes a run blocked by concurrency that may become runnable now that +// the current run's activity may have freed a workflow-level or job-level concurrency group. func checkRunConcurrency(ctx context.Context, run *actions_model.ActionRun) (*jobsCheckResult, error) { result := &jobsCheckResult{} checkedConcurrencyGroup := make(container.Set[string]) collect := func(concurrencyGroup string) error { - concurrentRunID, err := findBlockedRunIDByConcurrency(ctx, run.RepoID, concurrencyGroup) - if err != nil { - return fmt.Errorf("find blocked run by concurrency: %w", err) - } - if concurrentRunID > 0 { - r, err := checkBlockedConcurrentRun(ctx, run.RepoID, concurrentRunID) - if err != nil { - return err - } - result.merge(r) - } checkedConcurrencyGroup.Add(concurrencyGroup) + + // Exclude run.ID: this run's own jobs are resolved by checkJobsOfCurrentRunAttempt, no need to re-emit. + concurrentRunID, err := findConcurrencyWaiterToWake(ctx, run.RepoID, run.ID, concurrencyGroup) + if err != nil { + return err + } + if concurrentRunID == 0 { + return nil + } + + concurrentRun, err := actions_model.GetRunByRepoAndID(ctx, run.RepoID, concurrentRunID) + if err != nil { + return fmt.Errorf("get concurrent run %d: %w", concurrentRunID, err) + } + // A run awaiting approval is not advanced by concurrency; ApproveRuns emits it once approved. + if concurrentRun.NeedApproval { + return nil + } + + result.RunIDsToReEmit = append(result.RunIDsToReEmit, concurrentRunID) return nil } @@ -282,10 +292,26 @@ func checkJobsOfCurrentRunAttempt(ctx context.Context, run *actions_model.Action switch status { case actions_model.StatusWaiting: if err := expandReusableWorkflowCaller(ctx, run, attempt, job, vars); err != nil { - return fmt.Errorf("trigger caller-ready %d: %w", job.ID, err) + // Terminal expansion failure (an invalid/unresolvable reusable workflow): fail this caller. + log.Warn("caller %d cannot be expanded: %v", job.ID, err) + job.Status = actions_model.StatusFailure + job.Stopped = timeutil.TimeStampNow() + if n, uerr := actions_model.UpdateRunJob(ctx, job, builder.Eq{"status": actions_model.StatusBlocked, "is_expanded": false}, "status", "stopped"); uerr != nil { + return fmt.Errorf("mark unexpandable caller %d failed: %w", job.ID, uerr) + } else if n == 1 { + log.Warn("unexpandable caller %d has been marked as failed", job.ID) + result.UpdatedJobs = append(result.UpdatedJobs, job) + // Re-emit so the failed caller's dependents get resolved on the next pass. + expandedAnyCaller = true + } else { + // A concurrent writer advanced the caller; restore the in-memory state. + log.Warn("unexpandable caller %d has been advanced by a concurrent writer, not marking it failed", job.ID) + job.Status = actions_model.StatusBlocked + job.Stopped = 0 + } + } else { + expandedAnyCaller = true } - // expandReusableWorkflowCaller inserts children as Blocked. They need a follow-up resolver pass. - expandedAnyCaller = true case actions_model.StatusSkipped: job.Status = actions_model.StatusSkipped if _, err := actions_model.UpdateRunJob(ctx, job, nil, "status"); err != nil { @@ -477,7 +503,7 @@ type jobsCheckResult struct { UpdatedJobs []*actions_model.ActionRunJob // CancelledJobs are jobs cancelled by job-level concurrency while preparing to start. CancelledJobs []*actions_model.ActionRunJob - // RunIDsToReEmit are runs whose newly expanded reusable workflow callers need another resolver pass. + // RunIDsToReEmit are runs that need another resolver pass in their own transaction. RunIDsToReEmit []int64 } diff --git a/services/actions/job_emitter_test.go b/services/actions/job_emitter_test.go index 6b15ecd346..dba21bfb44 100644 --- a/services/actions/job_emitter_test.go +++ b/services/actions/job_emitter_test.go @@ -203,7 +203,7 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) { assert.NoError(t, unittest.PrepareTestDatabase()) ctx := t.Context() - // Run A: the triggering run of attempt A + // Run A: the triggering run of attempt A. It is done, so it no longer holds "test-cg", which is what lets checkRunConcurrency wake the blocked waiter. runA := &actions_model.ActionRun{ RepoID: 4, OwnerID: 1, @@ -211,16 +211,16 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) { WorkflowID: "test.yml", Index: 9901, Ref: "refs/heads/main", - Status: actions_model.StatusRunning, + Status: actions_model.StatusSuccess, } assert.NoError(t, db.Insert(ctx, runA)) - // Attempt A: an attempt of run A with concurrency group "test-cg" + // Attempt A: a done attempt of run A with concurrency group "test-cg" runAAttempt := &actions_model.ActionRunAttempt{ RepoID: 4, RunID: runA.ID, Attempt: 1, - Status: actions_model.StatusRunning, + Status: actions_model.StatusSuccess, ConcurrencyGroup: "test-cg", } assert.NoError(t, db.Insert(ctx, runAAttempt)) @@ -283,9 +283,11 @@ func Test_checkRunConcurrency_NoDuplicateConcurrencyGroupCheck(t *testing.T) { result, err := checkRunConcurrency(ctx, runA) assert.NoError(t, err) - if assert.Len(t, result.Jobs, 1) { - assert.Equal(t, jobBBlocked.ID, result.Jobs[0].ID) + // "test-cg" is free, so the single blocked waiter (run B) is collected for re-emit. + if assert.Len(t, result.RunIDsToReEmit, 1) { + assert.Equal(t, runB.ID, result.RunIDsToReEmit[0]) } + assert.Empty(t, result.Jobs) } // Test_checkJobsOfCurrentRunAttempt_RunLevelConcurrencyKeepsJobsBlocked verifies that @@ -352,3 +354,82 @@ jobs: refreshed := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: blockedJob.ID}) assert.Equal(t, actions_model.StatusBlocked, refreshed.Status) } + +// Test_checkRunConcurrency_HeldGroupDoesNotWake verifies that only an unoccupied concurrency group can wake up a blocked run/job. +func Test_checkRunConcurrency_HeldGroupDoesNotWake(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + ctx := t.Context() + + // Run A holds "test-cg": its attempt is still running. + runA := &actions_model.ActionRun{ + RepoID: 4, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml", + Index: 9911, Ref: "refs/heads/main", Status: actions_model.StatusRunning, + } + assert.NoError(t, db.Insert(ctx, runA)) + runAAttempt := &actions_model.ActionRunAttempt{ + RepoID: 4, RunID: runA.ID, Attempt: 1, Status: actions_model.StatusRunning, ConcurrencyGroup: "test-cg", + } + assert.NoError(t, db.Insert(ctx, runAAttempt)) + _, err := db.Exec(ctx, "UPDATE `action_run` SET latest_attempt_id = ? WHERE id = ?", runAAttempt.ID, runA.ID) + assert.NoError(t, err) + + // Run B is blocked on the same group. + runB := &actions_model.ActionRun{ + RepoID: 4, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml", + Index: 9912, Ref: "refs/heads/main", Status: actions_model.StatusBlocked, + } + assert.NoError(t, db.Insert(ctx, runB)) + runBAttempt := &actions_model.ActionRunAttempt{ + RepoID: 4, RunID: runB.ID, Attempt: 1, Status: actions_model.StatusBlocked, ConcurrencyGroup: "test-cg", + } + assert.NoError(t, db.Insert(ctx, runBAttempt)) + _, err = db.Exec(ctx, "UPDATE `action_run` SET latest_attempt_id = ? WHERE id = ?", runBAttempt.ID, runB.ID) + assert.NoError(t, err) + + runA, _, _ = db.GetByID[actions_model.ActionRun](ctx, runA.ID) + result, err := checkRunConcurrency(ctx, runA) + assert.NoError(t, err) + + // The group is held by run A, so run B must not be woken; A will wake it when it releases the group. + assert.Empty(t, result.RunIDsToReEmit) +} + +// Test_findConcurrencyWaiterToWake covers the finder's contract: it skips the run being processed (excludeRunID), +// returns another blocked waiter when the group is free, and returns 0 while the group is still held. +func Test_findConcurrencyWaiterToWake(t *testing.T) { + assert.NoError(t, unittest.PrepareTestDatabase()) + ctx := t.Context() + + const repoID int64 = 4 + seed := func(index int64, group string, status actions_model.Status) *actions_model.ActionRun { + run := &actions_model.ActionRun{ + RepoID: repoID, OwnerID: 1, TriggerUserID: 1, WorkflowID: "test.yml", + Index: index, Ref: "refs/heads/main", Status: status, + } + assert.NoError(t, db.Insert(ctx, run)) + assert.NoError(t, db.Insert(ctx, &actions_model.ActionRunAttempt{ + RepoID: repoID, RunID: run.ID, Attempt: 1, Status: status, ConcurrencyGroup: group, + })) + return run + } + + // Free group "excl-cg" with two blocked runs: excluding self returns the other waiter, not self. + self := seed(99701, "excl-cg", actions_model.StatusBlocked) + other := seed(99702, "excl-cg", actions_model.StatusBlocked) + id, err := findConcurrencyWaiterToWake(ctx, repoID, self.ID, "excl-cg") + assert.NoError(t, err) + assert.Equal(t, other.ID, id) + + // Free group "solo-cg" with only self blocked: excluding it leaves no waiter. + solo := seed(99703, "solo-cg", actions_model.StatusBlocked) + id, err = findConcurrencyWaiterToWake(ctx, repoID, solo.ID, "solo-cg") + assert.NoError(t, err) + assert.Equal(t, int64(0), id) + + // Held group "held-cg" (a running holder) has a blocked waiter, but nothing is woken while held. + seed(99704, "held-cg", actions_model.StatusRunning) + seed(99705, "held-cg", actions_model.StatusBlocked) + id, err = findConcurrencyWaiterToWake(ctx, repoID, 0, "held-cg") + assert.NoError(t, err) + assert.Equal(t, int64(0), id) +} diff --git a/services/actions/reusable_workflow.go b/services/actions/reusable_workflow.go index 562e9ee201..345f6b039d 100644 --- a/services/actions/reusable_workflow.go +++ b/services/actions/reusable_workflow.go @@ -5,6 +5,7 @@ package actions import ( "context" + "errors" "fmt" "strings" @@ -236,30 +237,33 @@ func expandReusableWorkflowCaller(ctx context.Context, run *actions_model.Action return fmt.Errorf("build call payload: %w", err) } - // 8. Insert direct children of this caller. - existingChildren, err := actions_model.GetDirectChildJobsByParent(ctx, caller) - if err != nil { - return fmt.Errorf("get existing children of caller %d: %w", caller.ID, err) - } - if len(existingChildren) > 0 { - // Should not happen - child jobs cannot be expanded before the caller gets ready - return fmt.Errorf("invariant violation: caller %d has %d pre-existing children", caller.ID, len(existingChildren)) - } - if err := insertCallerChildren(ctx, run, attempt, caller, content, contentSourceRepoID, contentSourceCommitSHA, vars, workflowCallInputs); err != nil { - return err - } - - // 9. Update caller-related cols. - caller.CallPayload = string(callPayload) + // 8. Claim the expansion by flipping is_expanded false->true BEFORE inserting any children. + // Two concurrent expanders serialize on this row: exactly one winner matches (n==1) and owns the expansion. + // Children are only ever inserted by the claim winner, so no duplicate child rows can arise. caller.IsExpanded = true - n, err := actions_model.UpdateRunJob(ctx, caller, + n, err := actions_model.UpdateRunJob(ctx, caller, builder.And( builder.Eq{"is_expanded": false}, - "call_secrets", "reusable_workflow_content", "call_payload", "is_expanded") + builder.In("status", actions_model.StatusBlocked, actions_model.StatusWaiting), + ), "is_expanded") if err != nil { - return fmt.Errorf("commit caller %d expansion: %w", caller.ID, err) + caller.IsExpanded = false // the claim was not established + return fmt.Errorf("claim caller %d expansion: %w", caller.ID, err) } if n == 0 { - return fmt.Errorf("caller %d already expanded by another writer", caller.ID) + // Another writer won the expansion, or the caller has been moved to a terminal status (e.g. failed/cancelled). + return nil + } + + // 9. We own the expansion: insert the direct children. + if err := insertCallerChildren(ctx, run, attempt, caller, content, contentSourceRepoID, contentSourceCommitSHA, vars, workflowCallInputs); err != nil { + // On failure, undo the partial expansion so an error return always leaves the caller unexpanded and childless. + return errors.Join(err, undoExpansion(ctx, caller)) + } + + // 10. Persist the remaining caller metadata (the row is already ours via the claim above). + caller.CallPayload = string(callPayload) + if _, err := actions_model.UpdateRunJob(ctx, caller, nil, "call_secrets", "reusable_workflow_content", "call_payload"); err != nil { + return errors.Join(fmt.Errorf("persist caller %d expansion metadata: %w", caller.ID, err), undoExpansion(ctx, caller)) } return nil } @@ -375,3 +379,16 @@ func ResolveUses(ctx context.Context, uses string) (*jobparser.UsesRef, error) { } return ref, nil } + +// undoExpansion rolls back a partial expansion owned by the current transaction: +// it removes the inserted children and releases the is_expanded claim itself. +func undoExpansion(ctx context.Context, caller *actions_model.ActionRunJob) error { + if err := actions_model.DeleteDirectChildJobsByParent(ctx, caller); err != nil { + return fmt.Errorf("delete children of caller %d: %w", caller.ID, err) + } + caller.IsExpanded = false + if _, err := actions_model.UpdateRunJob(ctx, caller, nil, "is_expanded"); err != nil { + return fmt.Errorf("release caller %d expansion claim: %w", caller.ID, err) + } + return nil +} diff --git a/services/actions/reusable_workflow_test.go b/services/actions/reusable_workflow_test.go index 28b6696c36..8ed228a487 100644 --- a/services/actions/reusable_workflow_test.go +++ b/services/actions/reusable_workflow_test.go @@ -206,3 +206,34 @@ func TestResolveUses(t *testing.T) { assert.ErrorContains(t, err, "must point to this Gitea instance") }) } + +func TestUndoExpansion(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + ctx := t.Context() + + // A claimed caller with two children inserted by the aborted expansion, plus a sibling that must survive. + caller := &actions_model.ActionRunJob{ + RunID: 991, RepoID: 4, OwnerID: 1, JobID: "caller", Name: "caller", + Status: actions_model.StatusBlocked, IsReusableCaller: true, IsExpanded: true, + } + require.NoError(t, db.Insert(ctx, caller)) + for _, jobID := range []string{"child1", "child2"} { + require.NoError(t, db.Insert(ctx, &actions_model.ActionRunJob{ + RunID: 991, RepoID: 4, OwnerID: 1, JobID: jobID, Name: jobID, + Status: actions_model.StatusBlocked, ParentJobID: caller.ID, + })) + } + sibling := &actions_model.ActionRunJob{ + RunID: 991, RepoID: 4, OwnerID: 1, JobID: "sibling", Name: "sibling", + Status: actions_model.StatusBlocked, + } + require.NoError(t, db.Insert(ctx, sibling)) + + require.NoError(t, undoExpansion(ctx, caller)) + + assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRunJob{ParentJobID: caller.ID})) + assert.False(t, caller.IsExpanded) + refreshed := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: caller.ID}) + assert.False(t, refreshed.IsExpanded) + unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: sibling.ID}) +} diff --git a/tests/integration/actions_reusable_workflow_test.go b/tests/integration/actions_reusable_workflow_test.go index ab79fcfed6..03f0f1e99f 100644 --- a/tests/integration/actions_reusable_workflow_test.go +++ b/tests/integration/actions_reusable_workflow_test.go @@ -574,6 +574,49 @@ jobs: assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRun{RepoID: repo.ID})) }) + t.Run("Nested caller with missing callee fails instead of blocking", func(t *testing.T) { + // When the expansion hits a terminal error (e.g. missing callee), the emitter must fail the caller and let the run finish as failed, not retry the expansion forever. + apiRepo := createActionsTestRepo(t, user2Token, "nested-caller-missing-callee", false) + repo := unittest.AssertExistsAndLoadBean(t, &repo_model.Repository{ID: apiRepo.ID}) + + runner := newMockRunner() + runner.registerAsRepoRunner(t, repo.OwnerName, repo.Name, "mock-runner", []string{"ubuntu-latest"}, false) + + createRepoWorkflowFile(t, user2, user2Token, repo, ".gitea/workflows/caller.yaml", + `name: Caller +on: push +jobs: + plain_job: + runs-on: ubuntu-latest + steps: + - run: echo 'job' + bad_caller: + needs: plain_job + uses: ./.gitea/workflows/does-not-exist.yml +`) + + // plain_job runs first; bad_caller is Blocked on needs and is NOT expanded at creation. + plainTask := runner.fetchTask(t) + _, plainJob, run := getTaskAndJobAndRunByTaskID(t, plainTask.Id) + assert.Equal(t, "plain_job", plainJob.JobID) + badCallerPre := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{RunID: run.ID, JobID: "bad_caller"}) + assert.Equal(t, actions_model.StatusBlocked, badCallerPre.Status) + assert.False(t, badCallerPre.IsExpanded) + + runner.execTask(t, plainTask, &mockTaskOutcome{result: runnerv1.Result_RESULT_SUCCESS}) + + // The emitter now tries to expand bad_caller, hits the missing callee, and fails the caller. + badCaller := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRunJob{ID: badCallerPre.ID}) + assert.Equal(t, actions_model.StatusFailure, badCaller.Status) + // No children were inserted (the terminal error precedes the child inserts). + assert.Equal(t, 0, unittest.GetCount(t, &actions_model.ActionRunJob{ParentJobID: badCallerPre.ID})) + + finalRun := unittest.AssertExistsAndLoadBean(t, &actions_model.ActionRun{ID: run.ID}) + assert.Equal(t, actions_model.StatusFailure, finalRun.Status) + + runner.fetchNoTask(t) // no task scheduled for the failed caller; the run is not stuck + }) + t.Run("Fork PR with secrets: inherit does not leak base repo secrets", func(t *testing.T) { // user2 owns the base repo, configures a secret, and registers a reusable workflow that declares a required secret. // The caller workflow uses `secrets: inherit`.