feat: TipSetWorker accepts all storage systems (#1035)

- closes #1029

Author: frrist

chain/indexer/distributed/queue/tasks/gapfill.go

@@ -59,16 +59,24 @@ func NewGapFillTipSetTask(ctx context.Context, ts *types.TipSet, tasks []string)
 	return asynq.NewTask(TypeGapFillTipSet, payload), nil
 }
 
-type AsynqGapFillTipSetTaskHandler struct {
+func NewGapFillHandler(indexer indexer.Indexer, db *storage.Database) *GapFillTipSetHandler {
+	return &GapFillTipSetHandler{indexer: indexer, db: db}
+}
+
+type GapFillTipSetHandler struct {
 	indexer indexer.Indexer
 	db      *storage.Database
 }
 
-func NewGapFillHandler(indexer indexer.Indexer, db *storage.Database) *AsynqGapFillTipSetTaskHandler {
-	return &AsynqGapFillTipSetTaskHandler{indexer: indexer, db: db}
+func (gh *GapFillTipSetHandler) Handler() asynq.HandlerFunc {
+	return gh.HandleGapFillTipSetTask
+}
+
+func (gh *GapFillTipSetHandler) Type() string {
+	return TypeGapFillTipSet
 }
 
-func (gh *AsynqGapFillTipSetTaskHandler) HandleGapFillTipSetTask(ctx context.Context, t *asynq.Task) error {
+func (gh *GapFillTipSetHandler) HandleGapFillTipSetTask(ctx context.Context, t *asynq.Task) error {
 	var p GapFillTipSetPayload
 	if err := json.Unmarshal(t.Payload(), &p); err != nil {
 		return err

chain/indexer/distributed/queue/tasks/index.go

@@ -61,15 +61,23 @@ func NewIndexTipSetTask(ctx context.Context, ts *types.TipSet, tasks []string) (
 	return asynq.NewTask(TypeIndexTipSet, payload), nil
 }
 
-type AsynqTipSetTaskHandler struct {
+func NewIndexHandler(i indexer.Indexer) *TipSetTaskHandler {
+	return &TipSetTaskHandler{indexer: i}
+}
+
+type TipSetTaskHandler struct {
 	indexer indexer.Indexer
 }
 
-func NewIndexHandler(i indexer.Indexer) *AsynqTipSetTaskHandler {
-	return &AsynqTipSetTaskHandler{indexer: i}
+func (ih *TipSetTaskHandler) Type() string {
+	return TypeIndexTipSet
+}
+
+func (ih *TipSetTaskHandler) Handler() asynq.HandlerFunc {
+	return ih.HandleIndexTipSetTask
 }
 
-func (ih *AsynqTipSetTaskHandler) HandleIndexTipSetTask(ctx context.Context, t *asynq.Task) error {
+func (ih *TipSetTaskHandler) HandleIndexTipSetTask(ctx context.Context, t *asynq.Task) error {
 	var p IndexTipSetPayload
 	if err := json.Unmarshal(t.Payload(), &p); err != nil {
 		return err

chain/indexer/distributed/queue/worker.go

@@ -11,30 +11,30 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
-	"github.com/filecoin-project/lily/chain/indexer"
 	"github.com/filecoin-project/lily/chain/indexer/distributed"
 	"github.com/filecoin-project/lily/chain/indexer/distributed/queue/tasks"
 	"github.com/filecoin-project/lily/metrics"
-	"github.com/filecoin-project/lily/storage"
 )
 
 var log = logging.Logger("lily/distributed/worker")
 
 type AsynqWorker struct {
 	done chan struct{}
 
-	name   string
-	server *distributed.TipSetWorker
-	index  indexer.Indexer
-	db     *storage.Database
+	name     string
+	server   *distributed.TipSetWorker
+	handlers []TaskHandler
+}
+type TaskHandler interface {
+	Type() string
+	Handler() asynq.HandlerFunc
 }
 
-func NewAsynqWorker(name string, i indexer.Indexer, db *storage.Database, server *distributed.TipSetWorker) *AsynqWorker {
+func NewAsynqWorker(name string, server *distributed.TipSetWorker, handlers ...TaskHandler) *AsynqWorker {
 	return &AsynqWorker{
-		name:   name,
-		server: server,
-		index:  i,
-		db:     db,
+		name:     name,
+		server:   server,
+		handlers: handlers,
 	}
 }
 
@@ -43,8 +43,10 @@ func (t *AsynqWorker) Run(ctx context.Context) error {
 	defer close(t.done)
 
 	mux := asynq.NewServeMux()
-	mux.HandleFunc(tasks.TypeIndexTipSet, tasks.NewIndexHandler(t.index).HandleIndexTipSetTask)
-	mux.HandleFunc(tasks.TypeGapFillTipSet, tasks.NewGapFillHandler(t.index, t.db).HandleGapFillTipSetTask)
+	for _, handler := range t.handlers {
+		log.Infow("registered task handler", "type", handler.Type())
+		mux.HandleFunc(handler.Type(), handler.Handler())
+	}
 
 	t.server.ServerConfig.Logger = log.With("name", t.name)
 	t.server.ServerConfig.ErrorHandler = &WorkerErrorHandler{}

lens/lily/impl.go

@@ -25,6 +25,7 @@ import (
 	"github.com/filecoin-project/lily/chain/indexer"
 	"github.com/filecoin-project/lily/chain/indexer/distributed"
 	"github.com/filecoin-project/lily/chain/indexer/distributed/queue"
+	"github.com/filecoin-project/lily/chain/indexer/distributed/queue/tasks"
 	"github.com/filecoin-project/lily/chain/indexer/integrated"
 	"github.com/filecoin-project/lily/chain/indexer/integrated/tipset"
 	"github.com/filecoin-project/lily/chain/walk"
@@ -102,9 +103,20 @@ func (m *LilyNodeAPI) StartTipSetWorker(_ context.Context, cfg *LilyTipSetWorker
 		return nil, err
 	}
 
-	db, err := m.StorageCatalog.ConnectAsDatabase(ctx, cfg.JobConfig.Storage, md)
-	if err != nil {
-		return nil, err
+	handlers := []queue.TaskHandler{tasks.NewIndexHandler(im)}
+	// check if queue config contains configuration for gap fill tasks and if it expects the tasks to be processed. This
+	// is specified by giving the Fill queue a priority greater than 1.
+	priority, ok := worker.ServerConfig.Queues[indexer.Fill.String()]
+	if ok {
+		if priority > 0 {
+			// if gap fill tasks have a priority storage must be a database.
+			db, ok := strg.(*storage.Database)
+			if !ok {
+				return nil, fmt.Errorf("storage type (%T) is unsupported when %s queue is enable", strg, indexer.Fill.String())
+			}
+			//  add gap fill handler to set of worker handlers.
+			handlers = append(handlers, tasks.NewGapFillHandler(im, db))
+		}
 	}
 
 	res := m.Scheduler.Submit(&schedule.JobConfig{
@@ -114,7 +126,7 @@ func (m *LilyNodeAPI) StartTipSetWorker(_ context.Context, cfg *LilyTipSetWorker
 			"queue":   cfg.Queue,
 			"storage": cfg.JobConfig.Storage,
 		},
-		Job:                 queue.NewAsynqWorker(cfg.JobConfig.Name, im, db, worker),
+		Job:                 queue.NewAsynqWorker(cfg.JobConfig.Name, worker, handlers...),
 		RestartOnFailure:    cfg.JobConfig.RestartOnFailure,
 		RestartOnCompletion: cfg.JobConfig.RestartOnCompletion,
 		RestartDelay:        cfg.JobConfig.RestartDelay,