fix (workflow): handle back pressure and other edge cases

server/pkg/workflow/config.go

@@ -2,14 +2,15 @@ package workflow
 
 import (
 	. "github.com/mickael-kerjean/filestash/server/pkg/config"
-	. "github.com/mickael-kerjean/filestash/server/pkg/kernel"
 	. "github.com/mickael-kerjean/filestash/server/pkg/core"
+	. "github.com/mickael-kerjean/filestash/server/pkg/kernel"
 )
 
 func init() {
 	Hooks.Register.Onload(func() {
 		PluginEnable()
 		PluginNumberWorker()
+		PluginMaxQueue()
 	})
 }
 
@@ -20,7 +21,7 @@ var PluginEnable = func() bool {
 		}
 		f.Name = "enable"
 		f.Type = "enable"
-		f.Target = []string{"workflow_workers"}
+		f.Target = []string{"workflow_workers", "workflow_max_queue"}
 		f.Description = "Enable/Disable workflows"
 		f.Default = true
 		return f
@@ -40,3 +41,17 @@ var PluginNumberWorker = func() int {
 		return f
 	}).Int()
 }
+
+var PluginMaxQueue = func() int {
+	return Config.Get("features.workflow.max_queue").Schema(func(f *FormElement) *FormElement {
+		if f == nil {
+			f = &FormElement{}
+		}
+		f.Id = "workflow_max_queue"
+		f.Name = "max_queue"
+		f.Type = "number"
+		f.Description = "Maximum number of queued jobs per workflow. Default: 5"
+		f.Default = 5
+		return f
+	}).Int()
+}

server/pkg/workflow/index.go

@@ -3,8 +3,8 @@ package workflow
 import (
 	"time"
 
-	. "github.com/mickael-kerjean/filestash/server/pkg/kernel"
 	. "github.com/mickael-kerjean/filestash/server/pkg/core"
+	. "github.com/mickael-kerjean/filestash/server/pkg/kernel"
 	. "github.com/mickael-kerjean/filestash/server/pkg/utils"
 	. "github.com/mickael-kerjean/filestash/server/pkg/workflow/model"
 )
@@ -28,7 +28,11 @@ func Init() error {
 		}
 		go func(t chan ITriggerEvent) {
 			for trigger := range t {
-				if err := CreateJob(trigger.WorkflowID(), trigger.Input()); err != nil {
+				workflowID := trigger.WorkflowID()
+				if err := CreateJob(workflowID, trigger.Input(), PluginMaxQueue()); err == ErrCongestion {
+					Log.Warning("[workflow] action=createJob workflow=%s err=queue_full", workflowID)
+					continue
+				} else if err != nil {
 					Log.Error("[workflow] action=createJob err=%s", err.Error())
 				}
 				select {

server/pkg/workflow/model/index.go

@@ -44,8 +44,8 @@ func InitState() (err error) {
 
 	db.Exec(`
 	UPDATE jobs
-		SET status = 'READY', updated_at = CURRENT_TIMESTAMP
-		WHERE status IN ('RUNNING', 'CLAIMED')`)
+		SET status = 'FAILURE', updated_at = CURRENT_TIMESTAMP
+		WHERE status IN ('READY', 'PENDING', 'CLAIMED', 'RUNNING')`)
 
 	return nil
 }

server/pkg/workflow/model/job.go

@@ -16,7 +16,7 @@ type Job struct {
 	UpdatedAt       string `json:"updated_at"`
 }
 
-func CreateJob(workflowID string, input map[string]string) error {
+func CreateJob(workflowID string, input map[string]string, maxQueue int) error {
 	workflow, err := GetWorkflow(workflowID)
 	if err != nil {
 		return err
@@ -34,6 +34,13 @@ func CreateJob(workflowID string, input map[string]string) error {
 		return err
 	}
 	defer tx.Rollback()
+	var queued int
+	if err = tx.QueryRow(
+		`SELECT COUNT(*) FROM jobs WHERE related_workflow = ? AND status = 'READY'`,
+		workflowID,
+	).Scan(&queued); err == nil && queued >= maxQueue {
+		return ErrCongestion
+	}
 	if _, err = tx.Exec(`
 		INSERT INTO jobs (related_workflow, status, steps, input)
 			VALUES (?, 'READY', ?, ?)