feat: add new option for walk config (#1212)

* Add new option for walk config

---------

Co-authored-by: Terry <[email protected]>

Author: Terryhung

chain/walk/walker.go

@@ -17,28 +17,30 @@ import (
 
 var log = logging.Logger("lily/chain/walk")
 
-func NewWalker(obs indexer.Indexer, node lens.API, name string, tasks []string, minHeight, maxHeight int64, r *schedule.Reporter) *Walker {
+func NewWalker(obs indexer.Indexer, node lens.API, name string, tasks []string, minHeight, maxHeight int64, r *schedule.Reporter, stopOnError bool) *Walker {
 	return &Walker{
-		node:      node,
-		obs:       obs,
-		name:      name,
-		tasks:     tasks,
-		minHeight: minHeight,
-		maxHeight: maxHeight,
-		report:    r,
+		node:        node,
+		obs:         obs,
+		name:        name,
+		tasks:       tasks,
+		minHeight:   minHeight,
+		maxHeight:   maxHeight,
+		report:      r,
+		stopOnError: stopOnError,
 	}
 }
 
 // Walker is a job that indexes blocks by walking the chain history.
 type Walker struct {
-	node      lens.API
-	obs       indexer.Indexer
-	name      string
-	tasks     []string
-	minHeight int64 // limit persisting to tipsets equal to or above this height
-	maxHeight int64 // limit persisting to tipsets equal to or below this height}
-	done      chan struct{}
-	report    *schedule.Reporter
+	node        lens.API
+	obs         indexer.Indexer
+	name        string
+	tasks       []string
+	minHeight   int64 // limit persisting to tipsets equal to or above this height
+	maxHeight   int64 // limit persisting to tipsets equal to or below this height}
+	done        chan struct{}
+	report      *schedule.Reporter
+	stopOnError bool
 }
 
 // Run starts walking the chain history and continues until the context is done or
@@ -92,6 +94,7 @@ func (c *Walker) WalkChain(ctx context.Context, node lens.API, ts *types.TipSet)
 	defer span.End()
 
 	var err error
+	errs := []error{}
 	for int64(ts.Height()) >= c.minHeight && ts.Height() != 0 {
 		select {
 		case <-ctx.Done():
@@ -102,7 +105,15 @@ func (c *Walker) WalkChain(ctx context.Context, node lens.API, ts *types.TipSet)
 		c.report.UpdateCurrentHeight(int64(ts.Height()))
 		if success, err := c.obs.TipSet(ctx, ts, indexer.WithIndexerType(indexer.Walk), indexer.WithTasks(c.tasks)); err != nil {
 			span.RecordError(err)
-			return fmt.Errorf("notify tipset: %w", err)
+			err := fmt.Errorf("index tipset, height: %v, error: %v", ts.Height().String(), err)
+			log.Errorf("%v", err)
+			// collect error
+			errs = append(errs, err)
+
+			// return an error only if the "stopOnError" flag is set to true.
+			if c.stopOnError {
+				return err
+			}
 		} else if !success {
 			log.Errorw("walk incomplete", "height", ts.Height(), "tipset", ts.Key().String(), "reporter", c.name)
 		}
@@ -115,5 +126,9 @@ func (c *Walker) WalkChain(ctx context.Context, node lens.API, ts *types.TipSet)
 		}
 	}
 
+	if len(errs) > 0 {
+		return fmt.Errorf("errors: %v", errs)
+	}
+
 	return nil
 }

chain/walk/walker_test.go

@@ -56,7 +56,7 @@ func TestWalker(t *testing.T) {
 
 	t.Logf("initializing indexer")
 	reporter := &schedule.Reporter{}
-	idx := NewWalker(im, nodeAPI, t.Name(), []string{tasktype.BlocksTask}, 0, int64(head.Height()), reporter)
+	idx := NewWalker(im, nodeAPI, t.Name(), []string{tasktype.BlocksTask}, 0, int64(head.Height()), reporter, false)
 
 	t.Logf("indexing chain")
 	err = idx.WalkChain(ctx, nodeAPI, head)

commands/job/job.go

@@ -35,6 +35,7 @@ var JobRunCmd = &cli.Command{
 		RunRestartDelayFlag,
 		RunRestartFailure,
 		RunRestartCompletion,
+		StopOnError,
 	},
 	Subcommands: []*cli.Command{
 		WalkCmd,

commands/job/options.go

@@ -21,6 +21,7 @@ type runOpts struct {
 
 	RestartCompletion bool
 	RestartFailure    bool
+	StopOnError       bool
 }
 
 func (r runOpts) ParseJobConfig(kind string) lily.LilyJobConfig {
@@ -35,6 +36,7 @@ func (r runOpts) ParseJobConfig(kind string) lily.LilyJobConfig {
 		RestartOnFailure:    RunFlags.RestartFailure,
 		RestartOnCompletion: RunFlags.RestartCompletion,
 		RestartDelay:        RunFlags.RestartDelay,
+		StopOnError:         RunFlags.StopOnError,
 	}
 }
 
@@ -96,6 +98,14 @@ var RunRestartFailure = &cli.BoolFlag{
 	Destination: &RunFlags.RestartFailure,
 }
 
+var StopOnError = &cli.BoolFlag{
+	Name:        "stop-on-error",
+	Usage:       "Stop the job if it encounters an error.",
+	EnvVars:     []string{"LILY_JOB_STOP_ON_ERROR"},
+	Value:       false,
+	Destination: &RunFlags.StopOnError,
+}
+
 type notifyOps struct {
 	queue string
 }

lens/lily/api.go

@@ -94,6 +94,8 @@ type LilyJobConfig struct {
 	RestartOnFailure bool
 	// RestartOnCompletion when true will restart the job when it completes.
 	RestartOnCompletion bool
+	// RestartOnCompletion when true will restart the job when it completes.
+	StopOnError bool
 	// RestartDelay configures how long to wait before restarting the job.
 	RestartDelay time.Duration
 	// Storage is the name of the storage system the job will use, may be empty.

lens/lily/impl.go

@@ -328,7 +328,7 @@ func (m *LilyNodeAPI) LilyWalk(_ context.Context, cfg *LilyWalkConfig) (*schedul
 		RestartOnFailure:    cfg.JobConfig.RestartOnFailure,
 		RestartOnCompletion: cfg.JobConfig.RestartOnCompletion,
 		RestartDelay:        cfg.JobConfig.RestartDelay,
-		Job:                 walk.NewWalker(idx, m, cfg.JobConfig.Name, cfg.JobConfig.Tasks, cfg.From, cfg.To, reporter),
+		Job:                 walk.NewWalker(idx, m, cfg.JobConfig.Name, cfg.JobConfig.Tasks, cfg.From, cfg.To, reporter, cfg.JobConfig.StopOnError),
 		Reporter:            reporter,
 	}
 
@@ -356,7 +356,7 @@ func (m *LilyNodeAPI) LilyWalkNotify(_ context.Context, cfg *LilyWalkNotifyConfi
 		RestartOnFailure:    cfg.WalkConfig.JobConfig.RestartOnFailure,
 		RestartOnCompletion: cfg.WalkConfig.JobConfig.RestartOnCompletion,
 		RestartDelay:        cfg.WalkConfig.JobConfig.RestartDelay,
-		Job:                 walk.NewWalker(idx, m, cfg.WalkConfig.JobConfig.Name, cfg.WalkConfig.JobConfig.Tasks, cfg.WalkConfig.From, cfg.WalkConfig.To, reporter),
+		Job:                 walk.NewWalker(idx, m, cfg.WalkConfig.JobConfig.Name, cfg.WalkConfig.JobConfig.Tasks, cfg.WalkConfig.From, cfg.WalkConfig.To, reporter, cfg.WalkConfig.JobConfig.StopOnError),
 		Reporter:            reporter,
 	}
 	res := m.Scheduler.Submit(jobConfig)