indexer_service.go

package indexer

import (
	"context"
	"time"

	abci "github.com/dashpay/tenderdash/abci/types"
	"github.com/dashpay/tenderdash/internal/eventbus"
	"github.com/dashpay/tenderdash/internal/pubsub"
	"github.com/dashpay/tenderdash/libs/log"
	"github.com/dashpay/tenderdash/libs/service"
	"github.com/dashpay/tenderdash/types"
)

// Service connects event bus, transaction and block indexers together in
// order to index transactions and blocks coming from the event bus.
type Service struct {
	service.BaseService
	logger log.Logger

	eventSinks []EventSink
	eventBus   *eventbus.EventBus
	metrics    *Metrics

	currentBlock struct {
		header types.EventDataNewBlockHeader
		height int64
		batch  *Batch
	}
}

// NewService constructs a new indexer service from the given arguments.
func NewService(args ServiceArgs) *Service {
	is := &Service{
		logger:     args.Logger,
		eventSinks: args.Sinks,
		eventBus:   args.EventBus,
		metrics:    args.Metrics,
	}
	if is.metrics == nil {
		is.metrics = NopMetrics()
	}
	is.BaseService = *service.NewBaseService(args.Logger, "IndexerService", is)
	return is
}

// publish publishes a pubsub message to the service. The service blocks until
// the message has been fully processed.
func (is *Service) publish(msg pubsub.Message) error {
	// Indexing has three states. Initially, no block is in progress (WAIT) and
	// we expect a block header. Upon seeing a header, we are waiting for zero
	// or more transactions (GATHER). Once all the expected transactions have
	// been delivered (in some order), we are ready to index. After indexing a
	// block, we revert to the WAIT state for the next block.

	if is.currentBlock.batch == nil {
		// WAIT: Start a new block.
		hdr := msg.Data().(types.EventDataNewBlockHeader)
		is.currentBlock.header = hdr
		is.currentBlock.height = hdr.Header.Height
		is.currentBlock.batch = NewBatch(hdr.NumTxs)

		if hdr.NumTxs != 0 {
			return nil
		}
		// If the block does not expect any transactions, fall through and index
		// it immediately.  This shouldn't happen, but this check ensures we do
		// not get stuck if it does.
	}

	curr := is.currentBlock.batch
	if curr.Pending != 0 {
		// GATHER: Accumulate a transaction into the current block's batch.
		txResult := msg.Data().(types.EventDataTx).TxResult
		if err := curr.Add(&txResult); err != nil {
			is.logger.Error("failed to add tx to batch",
				"height", is.currentBlock.height, "index", txResult.Index, "err", err)
		}

		// This may have been the last transaction in the batch, so fall through
		// to check whether it is time to index.
	}

	if curr.Pending == 0 {
		// INDEX: We have all the transactions we expect for the current block.
		for _, sink := range is.eventSinks {
			start := time.Now()
			if err := sink.IndexBlockEvents(is.currentBlock.header); err != nil {
				is.logger.Error("failed to index block header",
					"height", is.currentBlock.height, "err", err)
			} else {
				is.metrics.BlockEventsSeconds.Observe(time.Since(start).Seconds())
				is.metrics.BlocksIndexed.Add(1)
				is.logger.Debug("indexed block",
					"height", is.currentBlock.height, "sink", sink.Type())
			}

			if curr.Size() != 0 {
				start := time.Now()

				var err error
				curr.Ops, err = DeduplicateBatch(curr.Ops, sink)
				if err != nil {
					is.logger.Error("failed to deduplicate batch", "height", is.currentBlock.height, "error", err)
				}

				err = sink.IndexTxEvents(curr.Ops)
				if err != nil {
					is.logger.Error("failed to index block txs",
						"height", is.currentBlock.height, "err", err)
				} else {
					is.metrics.TxEventsSeconds.Observe(time.Since(start).Seconds())
					is.metrics.TransactionsIndexed.Add(float64(curr.Size()))
					is.logger.Debug("indexed txs",
						"height", is.currentBlock.height, "sink", sink.Type())
				}
			}
		}
		is.currentBlock.batch = nil // return to the WAIT state for the next block
	}

	return nil
}

// OnStart implements part of service.Service. It registers an observer for the
// indexer if the underlying event sinks support indexing.
//
// TODO(creachadair): Can we get rid of the "enabled" check?
func (is *Service) OnStart(ctx context.Context) error {
	// If the event sinks support indexing, register an observer to capture
	// block header data for the indexer.
	if IndexingEnabled(is.eventSinks) {
		err := is.eventBus.Observe(ctx, is.publish,
			types.EventQueryNewBlockHeader, types.EventQueryTx)
		if err != nil {
			return err
		}
	}
	return nil
}

// OnStop implements service.Service by closing the event sinks.
func (is *Service) OnStop() {
	for _, sink := range is.eventSinks {
		if err := sink.Stop(); err != nil {
			is.logger.Error("failed to close eventsink", "eventsink", sink.Type(), "err", err)
		}
	}
}

// ServiceArgs are arguments for constructing a new indexer service.
type ServiceArgs struct {
	Sinks    []EventSink
	EventBus *eventbus.EventBus
	Metrics  *Metrics
	Logger   log.Logger
}

// KVSinkEnabled returns the given eventSinks is containing KVEventSink.
func KVSinkEnabled(sinks []EventSink) bool {
	for _, sink := range sinks {
		if sink.Type() == KV {
			return true
		}
	}

	return false
}

// IndexingEnabled returns the given eventSinks is supporting the indexing services.
func IndexingEnabled(sinks []EventSink) bool {
	for _, sink := range sinks {
		if sink.Type() == KV || sink.Type() == PSQL {
			return true
		}
	}

	return false
}

// DeduplicateBatch consider the case of duplicate txs.
// if the current one under investigation is NOT OK, then we need to check
// whether there's a previously indexed tx.
// SKIP the current tx if the previously indexed record is found and successful.
func DeduplicateBatch(ops []*abci.TxResult, sink EventSink) ([]*abci.TxResult, error) {
	result := make([]*abci.TxResult, 0, len(ops))

	// keep track of successful txs in this block in order to suppress latter ones being indexed.
	var successfulTxsInThisBlock = make(map[string]struct{})

	for _, txResult := range ops {
		hash := types.Tx(txResult.Tx).Hash()

		if txResult.Result.IsOK() {
			successfulTxsInThisBlock[string(hash)] = struct{}{}
		} else {
			// if it already appeared in current block and was successful, skip.
			if _, found := successfulTxsInThisBlock[string(hash)]; found {
				continue
			}

			// check if this tx hash is already indexed
			old, err := sink.GetTxByHash(hash)

			// if db op errored
			// Not found is not an error
			if err != nil {
				return nil, err
			}

			// if it's already indexed in an older block and was successful, skip.
			if old != nil && old.Result.Code == abci.CodeTypeOK {
				continue
			}
		}

		result = append(result, txResult)
	}

	return result, nil
}