feat: implement SLING_TIMEOUT functionality for long-running tasks and add corresponding tests

Author: flarco

cmd/sling/sling_run.go

@@ -891,7 +891,6 @@ func setTimeout(values ...string) (deadline time.Time) {
 		// only process first non-empty value
 		duration := time.Duration(cast.ToFloat64(timeout) * float64(time.Minute))
 		parent, cancel := context.WithTimeout(context.Background(), duration)
-		_ = cancel
 
 		ctx = g.NewContext(parent) // overwrite global context
 		time.AfterFunc(duration-time.Second, func() {
@@ -901,9 +900,9 @@ func setTimeout(values ...string) (deadline time.Time) {
 				content, _ := os.ReadFile(filePath)
 				env.Println(string(content))
 				panic(g.F("SLING_TIMEOUT = %s mins reached!", timeout))
-			} else {
-				g.Warn("SLING_TIMEOUT = %s mins reached!", timeout)
 			}
+			g.Warn("SLING_TIMEOUT = %s mins reached!", timeout)
+			cancel()
 		})
 
 		// set deadline for status setting later

cmd/sling/tests/pipelines/p.30.sling_timeout.yaml

@@ -0,0 +1,5 @@
+steps:
+  - type: query
+    id: long_sleep
+    connection: POSTGRES
+    query: select pg_sleep(15)

cmd/sling/tests/replications/r.102.sling_timeout.yaml

@@ -0,0 +1,21 @@
+# Test SLING_TIMEOUT triggers on a long-running replication.
+# Streams a very large generate_series so the run cannot finish before the
+# 0.1 minute (6s) timeout fires.
+
+source: postgres
+target: postgres
+
+defaults:
+  mode: full-refresh
+
+streams:
+  test_sling_timeout:
+    sql: |
+      SELECT
+        i AS id,
+        md5(i::text) AS hash_a,
+        md5((i + 1)::text) AS hash_b,
+        md5((i + 2)::text) AS hash_c,
+        repeat('x', 1000) AS padding
+      FROM generate_series(1, 200000000) AS s(i)
+    object: public.test_sling_timeout

cmd/sling/tests/suite.cli.yaml

@@ -2220,4 +2220,22 @@
   name: 'sling agent service CLI (help outputs + install error path)'
   run: 'sling run -p cmd/sling/tests/pipelines/p.29.agent_service_cli.yaml'
   output_contains:
-    - 'Agent Service: all checks passed'
+    - 'Agent Service: all checks passed'
+
+- id: 232
+  name: 'SLING_TIMEOUT triggers on long-running pipeline query step'
+  run: 'sling run -p cmd/sling/tests/pipelines/p.30.sling_timeout.yaml'
+  env:
+    SLING_TIMEOUT: '0.1'
+  err: true
+  output_contains:
+    - 'SLING_TIMEOUT = 0.1 mins reached!'
+
+- id: 233
+  name: 'SLING_TIMEOUT triggers on long-running postgres-to-postgres replication'
+  run: 'sling run -r cmd/sling/tests/replications/r.102.sling_timeout.yaml'
+  env:
+    SLING_TIMEOUT: '0.1'
+  err: true
+  output_contains:
+    - 'SLING_TIMEOUT = 0.1 mins reached!'

core/sling/pipeline.go

@@ -200,6 +200,17 @@ func (pl *Pipeline) Execute() (err error) {
 	// Execute each step
 	var lastErr error
 	for i := 0; i < len(pl.steps); i++ {
+		// honor context cancellation (e.g., SLING_TIMEOUT deadline) between steps
+		if pl.Context != nil {
+			select {
+			case <-pl.Context.Ctx.Done():
+				if lastErr == nil {
+					lastErr = g.Error("pipeline cancelled: %s", pl.Context.Ctx.Err())
+				}
+			default:
+			}
+		}
+
 		step := pl.steps[i]
 		step.SetContext(pl.Context) // update with latest context
 
@@ -317,6 +328,10 @@ func (pse *PipelineStepExecution) Execute(skip bool) (err error) {
 			if pse.Context() != nil {
 				select {
 				case <-pse.Context().Ctx.Done():
+					if isTimeoutDeadlinePassed(pse.Context()) {
+						pse.Status = ExecStatusTimedOut
+					}
+					pse.StateSet()
 					return
 				case <-ticker5s.C:
 					pse.StateSet()
@@ -369,6 +384,11 @@ retry:
 	// Handle errors
 	if err != nil {
 		pse.Err = err
+		// classify as timed-out if the step ended after the configured SLING_TIMEOUT deadline
+		if isTimeoutDeadlinePassed(pse.Context()) {
+			pse.Status = ExecStatusTimedOut
+			return g.Error(err, "step timed-out: %s", pse.Step.ID())
+		}
 		pse.Status = ExecStatusError
 		return g.Error(err, "error executing step: %s", pse.Step.ID())
 	}

core/sling/task_run.go

@@ -173,8 +173,7 @@ func (t *TaskExecution) Execute() error {
 	} else {
 
 		// check for timeout
-		deadline, ok := t.Context.Map.Get("timeout-deadline")
-		if ok && cast.ToInt64(deadline) <= (time.Now().Unix()+1) {
+		if isTimeoutDeadlinePassed(t.Context) {
 			t.SetProgress("execution failed (timed-out)")
 			newStatus = ExecStatusTimedOut
 		} else if newStatus != ExecStatusTerminated {
@@ -218,6 +217,16 @@ func (t *TaskExecution) Execute() error {
 	return t.Err
 }
 
+// isTimeoutDeadlinePassed reports whether SLING_TIMEOUT was configured and has elapsed.
+// Returns false when no deadline was set or when it has not yet been reached.
+func isTimeoutDeadlinePassed(c *g.Context) bool {
+	if c == nil {
+		return false
+	}
+	deadline, ok := c.Map.Get("timeout-deadline")
+	return ok && cast.ToInt64(deadline) <= (time.Now().Unix()+1)
+}
+
 func (t *TaskExecution) ExecuteHooks(stage HookStage) (err error) {
 	if t.Config == nil || t.Config.ReplicationStream == nil {
 		return nil