Add method to resume all workflow steps
This commit is contained in:
@@ -132,6 +132,24 @@ func (w *Workflow) Resume(ctx context.Context, workflowId string) error {
|
||||
return allErr
|
||||
}
|
||||
|
||||
// ResumeAll restarts all executions of this workflow across all steps.
|
||||
// For each step it resets any in-progress jobs to pending and failed jobs to
|
||||
// pending with NumberOfTries reset to 0, then starts the queue again if it is
|
||||
// configured as running.
|
||||
func (w *Workflow) ResumeAll(ctx context.Context) error {
|
||||
var allErr error
|
||||
for _, step := range w.AllSteps {
|
||||
if step.Queue == nil {
|
||||
// Safety: ensure queues are initialized
|
||||
step.initQueue()
|
||||
}
|
||||
if err := step.Queue.Resume(ctx); err != nil {
|
||||
allErr = errors.Join(allErr, fmt.Errorf("step %s: %w", step.Name, err))
|
||||
}
|
||||
}
|
||||
return allErr
|
||||
}
|
||||
|
||||
func NewCronTrigger(ctx context.Context, workflow *Workflow, d time.Duration) {
|
||||
go func() {
|
||||
ticker := time.NewTicker(d)
|
||||
|
||||
Reference in New Issue
Block a user