handle connection errors in replication

- Improved error handling during replication to prevent cascading failures.
- Added `FailErr` field to `ReplicationConfig` to store the error encountered when a connection issue occurs.
- Modified `replicationRun` to set `FailErr` when a connection error is detected, stopping further execution.

cmd/sling/sling_run.go

@@ -357,15 +357,17 @@ func runTask(cfg *sling.Config, replication *sling.ReplicationConfig) (err error
 
 	if cast.ToBool(cfg.Env["SLING_DRY_RUN"]) || cast.ToBool(os.Getenv("SLING_DRY_RUN")) {
 		return nil
+	} else if replication.FailErr != "" {
+		task.Status = sling.ExecStatusError
+		task.Err = g.Error(replication.FailErr)
 	}
 
 	// set log sink
 	env.LogSink = func(ll *g.LogLine) {
 		task.AppendOutput(ll)
 	}
 
-	sling.StoreSet(task)       // set into store
-	defer sling.StoreSet(task) // set into store after
+	sling.StoreSet(task) // set into store
 
 	if task.Err != nil {
 		err = g.Error(task.Err)
@@ -375,6 +377,9 @@ func runTask(cfg *sling.Config, replication *sling.ReplicationConfig) (err error
 	// set context
 	task.Context = ctx
 
+	// set into store after
+	defer sling.StoreSet(task)
+
 	// run task
 	setTM()
 	err = task.Execute()
@@ -484,7 +489,7 @@ func replicationRun(cfgPath string, cfgOverwrite *sling.Config, selectStreams ..
 
 			// if a connection issue, stop
 			if e, ok := err.(*g.ErrType); ok && strings.Contains(e.Debug(), "Could not connect to ") {
-				break
+				replication.FailErr = g.ErrMsg(e)
 			}
 		} else {
 			successes++

core/sling/replication.go

@@ -27,6 +27,7 @@ type ReplicationConfig struct {
 	// Tasks are compiled tasks
 	Tasks    []*Config `json:"tasks"`
 	Compiled bool      `json:"compiled"`
+	FailErr  string    // error string to fail all (e.g. when the first tasks fails to connect)
 
 	streamsOrdered []string
 	originalCfg    string