fix(actions): fail unexpandable reusable workflow callers and decouple the job emitter's cross-run processing (#38565) (#38587)

Backport #38565 by @Zettat123

## 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.

Co-authored-by: Zettat123 <zettat123@gmail.com>

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
+}