Skip to content

[CRE] Remove legacy workflow spec factories, v1 artifacts and workflow job paths #23713

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pavel-raykov merged 1 commit into from
Sep 11, 2026
Merged

[CRE] Remove legacy workflow spec factories, v1 artifacts and workflow job paths #23713

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci-core.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,7 @@

- name: Resolve Spot Setting
id: spot
uses: $/.github/actions/setup-runner-spot

Check failure on line 165 in .github/workflows/ci-core.yml

View workflow job for this annotation

GitHub Actions / Validate Github Action Workflows 

[actionlint] reported by reviewdog 🐶
specifying action "$/.github/actions/setup-runner-spot" in invalid format because ref is missing. available formats are "{owner}/{repo}@{ref}" or "{owner}/{repo}/{path}@{ref}" [action]

Raw Output:
e:.github/workflows/ci-core.yml:165:15: specifying action "$/.github/actions/setup-runner-spot" in invalid format because ref is missing. available formats are "{owner}/{repo}@{ref}" or "{owner}/{repo}/{path}@{ref}" [action]

golangci:
name: GolangCI Lint
Expand Down Expand Up @@ -391,7 +391,7 @@
id: wasm-key
shell: bash
run: |
echo "key=${{ runner.os }}-${{ runner.arch }}-wasmcache-${{ hashFiles('go.mod', 'go.sum', 'core/internal/testutils/wasmtest/**', 'core/capabilities/compute/**', 'core/services/workflows/**') }}" >> "$GITHUB_OUTPUT"
echo "key=${{ runner.os }}-${{ runner.arch }}-wasmcache-${{ hashFiles('go.mod', 'go.sum', 'core/internal/testutils/wasmtest/**', 'core/services/workflows/**') }}" >> "$GITHUB_OUTPUT"

- name: Restore WASM test binaries cache
if: ${{ matrix.type.should-run == 'true' }}
Expand Down
12 changes: 5 additions & 7 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,13 +52,11 @@ cl_backup_*.tar.gz

# Test artifacts
core/cmd/TestClient_ImportExportP2PKeyBundle_test_key.json
race.*
*output.txt
/golangci-lint/
.covdata
core/services/job/testdata/wasm/testmodule.wasm
core/services/job/testdata/wasm/testmodule.br
temp-repo
race.*
*output.txt
/golangci-lint/
.covdata
temp-repo
diagnose-*/
diagnose-attempted-fixes-*.jsonl

Expand Down
124 changes: 5 additions & 119 deletions core/services/feeds/service.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -623,18 +623,11 @@ func (s *service) DeleteJob(ctx context.Context, args *DeleteJobArgs) (int64, er
return 0, errors.New("cannot delete a job proposal belonging to another feeds manager")
}

// Try to delete as workflow job first (with auto-cancellation), fallback to simple deletion if not applicable
deleted, err := s.tryDeleteWithWorkflowCancellation(ctx, proposal, logger)
if err != nil {
return 0, err
}

if !deleted {
// For non-workflow jobs: simple proposal deletion (no cancellation, just job_proposal delete)
if err = s.deleteSimpleJobProposal(ctx, proposal, logger); err != nil {
return 0, err
}
if err = s.orm.DeleteProposal(ctx, proposal.ID); err != nil {
logger.Errorw("Failed to delete the proposal", "err", err)
return 0, fmt.Errorf("DeleteProposal failed: %w", err)
}
logger.Infow("Successfully deleted job proposal", "jobProposalID", proposal.ID)

if err = s.observeJobProposalCounts(ctx); err != nil {
logger.Errorw("Failed to push metrics for job proposal deletion", "err", err)
Expand Down Expand Up @@ -848,7 +841,7 @@ func (s *service) ProposeJob(ctx context.Context, args *ProposeJobArgs) (int64,
switch {
case err != nil:
logger.Errorw("Failed to validate spec while checking for workflow", "err", err)
case slices.Contains([]job.Type{job.Workflow, job.CRESettings}, jobType):
case jobType == job.CRESettings:
promWorkflowRequests.Inc()
promFeedsWorkflowRequests.Inc()
err = s.ApproveSpec(ctx, specID, true)
Expand Down Expand Up @@ -1035,15 +1028,6 @@ func (s *service) ApproveSpec(ctx context.Context, id int64, force bool) error {
return errors.Wrap(txerr, "FindOCR2JobIDByAddress failed")
}
}
case job.Workflow:
existingJobID, txerr = tx.jobORM.FindJobIDByWorkflow(ctx, *j.WorkflowSpec)
if txerr != nil {
// Return an error if the repository errors. If there is a not found
// error we want to continue with approving the job.
if !errors.Is(txerr, sql.ErrNoRows) {
return fmt.Errorf("failed while checking for existing workflow job: %w", txerr)
}
}
case job.CCIP:
existingJobID, txerr = tx.jobORM.FindJobIDByCapabilityNameAndVersion(ctx, *j.CCIPSpec)
// Return an error if the repository errors. If there is a not found
Expand Down Expand Up @@ -1900,101 +1884,3 @@ func (ns NullService) UpdateSpecDefinition(ctx context.Context, id int64, spec s
func (ns NullService) Unsafe_SetConnectionsManager(_ ConnectionsManager) {}

//revive:enable

// deleteSimpleJobProposal deletes a simple (non-workflow) job proposal.
// This only removes the proposal without any cancellation, unlike workflow jobs
func (s *service) deleteSimpleJobProposal(ctx context.Context, proposal *JobProposal, logger logger.Logger) error {
if err := s.orm.DeleteProposal(ctx, proposal.ID); err != nil {
logger.Errorw("Failed to delete the proposal", "err", err)
return fmt.Errorf("DeleteProposal failed: %w", err)
}

logger.Infow("Successfully deleted simple job proposal", "jobProposalID", proposal.ID)
return nil
}

// tryDeleteWithWorkflowCancellation attempts to delete a job as a workflow job.
// Returns true if the job was successfully deleted as a workflow, false if it's not a workflow job.
// Returns an error if deletion failed.
func (s *service) tryDeleteWithWorkflowCancellation(ctx context.Context, proposal *JobProposal, logger logger.Logger) (bool, error) {
// Early return if no external job ID (we won't find a job to delete without it)
if !proposal.ExternalJobID.Valid {
logger.Debugw("Proposal has no ExternalJobID, skipping workflow job deletion", "proposalID", proposal.ID)
return false, nil
}

// Try to find the job by external job ID
jobFound, err := s.jobORM.FindJobByExternalJobID(ctx, proposal.ExternalJobID.UUID)
if err != nil {
logger.Warnw("Failed to find job by external job ID, skipping workflow job deletion",
"externalJobID", proposal.ExternalJobID.UUID, "err", err)
return false, nil
}

// Check if this is actually a workflow job
if jobFound.WorkflowSpecID == nil {
logger.Debugw("Job is not a workflow job, skipping workflow job deletion",
"jobID", jobFound.ID, "jobType", jobFound.Type)
return false, nil
}

// Get the approved spec for workflow cancellation
jpSpec, err := s.orm.GetApprovedSpec(ctx, proposal.ID)
if err != nil {
logger.Errorw("GetApprovedSpec failed - cannot proceed with workflow job deletion",
"proposalID", proposal.ID, "err", err, "jobName", jobFound.Name)
return false, nil
}

// All validations passed - proceed with workflow job deletion
logger.Debugw("Proceeding with workflow job deletion",
"proposalID", proposal.ID, "jobID", jobFound.ID, "specID", jpSpec.ID)

return true, s.deleteWorkflowJobWithTransaction(ctx, *proposal, jobFound, *jpSpec, logger)
}

// deleteWorkflowJobWithTransaction performs workflow job deletion with auto-cancellation within a transaction.
func (s *service) deleteWorkflowJobWithTransaction(ctx context.Context, proposal JobProposal, job job.Job, jpSpec JobProposalSpec, logger logger.Logger) error {
if job.WorkflowSpecID == nil {
return errors.New("job WorkflowSpecID is nil, cannot delete workflow job")
}
jobSpecID := int64(*job.WorkflowSpecID)

fmsClient, err := s.connMgr.GetClient(proposal.FeedsManagerID)
if err != nil {
logger.Errorw("Failed to get FMS client", "jobProposalID", proposal.ID, "jobProposalSpecID", jpSpec.ID, "err", err, "name", job.Name)
return fmt.Errorf("failed to get FMS client for workflow spec cancellation: %w", err)
}

cancelLogger := logger.With("job_proposal_spec_id", jpSpec.ID, "jobSpecID", jobSpecID)

err = s.transact(ctx, func(tx datasources) error {
if txerr := tx.orm.DeleteProposal(ctx, proposal.ID); txerr != nil {
return fmt.Errorf("DeleteProposal failed: %w", txerr)
}

if txerr := tx.orm.CancelSpec(ctx, jpSpec.ID); txerr != nil {
return txerr
}

if serr := s.jobSpawner.DeleteJob(ctx, tx.ds, job.ID); serr != nil {
return fmt.Errorf("DeleteJob failed: %w", serr)
}

if _, err = fmsClient.CancelledJob(ctx, &pb.CancelledJobRequest{
Uuid: proposal.RemoteUUID.String(),
Version: int64(jpSpec.Version),
}); err != nil {
return err
}

return nil
})
if err != nil {
cancelLogger.Errorw("Failed to auto-cancel workflow spec", "err", err, "name", job.Name)
return fmt.Errorf("failed to auto-cancel workflow spec (job proposal spec ID: %d): %w", jpSpec.ID, err)
}

logger.Infow("Successfully auto-cancelled a workflow spec", "jobProposalID", proposal.ID, "jobProposalSpecID", jpSpec.ID, "jobSpecID", jobSpecID, "name", job.Name)
return nil
}
Loading
Loading