kylefang/gaze-indexer
1
0
Fork 0
mirror of https://github.com/alexgo-io/gaze-indexer.git synced 2026-01-12 16:53:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: Generic Indexer (#14)

Browse Source 
* refactor(indexer): init generic indexer

* refactor(btc): updatet datasource

* refactor: remove old indexers pkg

* doc: update comment

---------

Co-authored-by: Gaze <gazenw@users.noreply.github.com>
Co-authored-by: Planxnx <thanee@cleverse.com>
This commit is contained in:
gazenw
2024-05-08 16:15:33 +07:00
committed by GitHub
parent 86716c1915
commit c9a5c6d217
9 changed files with 75 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
cmd/cmd_run.go
View File

@@ -16,7 +16,8 @@ import (
"github.com/cockroachdb/errors"
"github.com/gaze-network/indexer-network/common/errs"
"github.com/gaze-network/indexer-network/core/datasources"
"github.com/gaze-network/indexer-network/core/indexers"
"github.com/gaze-network/indexer-network/core/indexer"
"github.com/gaze-network/indexer-network/core/types"
"github.com/gaze-network/indexer-network/internal/config"
"github.com/gaze-network/indexer-network/internal/postgres"
"github.com/gaze-network/indexer-network/modules/bitcoin"
@@ -175,7 +176,7 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
if !opts.APIOnly {
processor := bitcoin.NewProcessor(conf, btcDB, indexerInfoDB)
datasource := datasources.NewBitcoinNode(client)
indexer := indexers.NewBitcoinIndexer(processor, datasource)
indexer := indexer.New(processor, datasource)
defer func() {
if err := indexer.ShutdownWithTimeout(shutdownTimeout); err != nil {
logger.ErrorContext(ctx, "Error during shutdown indexer", slogx.Error(err))
@@ -225,7 +226,7 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
default:
return errors.Wrapf(errs.Unsupported, "%q database for indexer is not supported", conf.Modules.Runes.Database)
}
var bitcoinDatasource indexers.BitcoinDatasource
var bitcoinDatasource datasources.Datasource[*types.Block]
var bitcoinClient btcclient.Contract
switch strings.ToLower(conf.Modules.Runes.Datasource) {
case "bitcoin-node":
@@ -250,8 +251,8 @@ func runHandler(opts *runCmdOptions, cmd *cobra.Command, _ []string) error {
}
if !opts.APIOnly {
processor := runes.NewProcessor(runesDg, indexerInfoDg, bitcoinClient, bitcoinDatasource, conf.Network, reportingClient)
indexer := indexers.NewBitcoinIndexer(processor, bitcoinDatasource)
processor := runes.NewProcessor(runesDg, indexerInfoDg, bitcoinClient, conf.Network, reportingClient)
indexer := indexer.New(processor, bitcoinDatasource)
defer func() {
if err := indexer.ShutdownWithTimeout(shutdownTimeout); err != nil {
logger.ErrorContext(ctx, "Error during shutdown indexer", slogx.Error(err))

2
core/datasources/bitcoin_node.go
View File

@@ -24,7 +24,7 @@ const (
)
// Make sure to implement the BitcoinDatasource interface
var _ Datasource[[]*types.Block] = (*BitcoinNodeDatasource)(nil)
var _ Datasource[*types.Block] = (*BitcoinNodeDatasource)(nil)
// BitcoinNodeDatasource fetch data from Bitcoin node for Bitcoin Indexer
type BitcoinNodeDatasource struct {

4
core/datasources/datasources.go
View File

@@ -10,7 +10,7 @@ import (
// Datasource is an interface for indexer data sources.
type Datasource[T any] interface {
Name() string
Fetch(ctx context.Context, from, to int64) (T, error)
FetchAsync(ctx context.Context, from, to int64, ch chan<- T) (*subscription.ClientSubscription[T], error)
Fetch(ctx context.Context, from, to int64) ([]T, error)
FetchAsync(ctx context.Context, from, to int64, ch chan<- []T) (*subscription.ClientSubscription[[]T], error)
GetBlockHeader(ctx context.Context, height int64) (types.BlockHeader, error)
}

93
core/indexers/bitcoin_indexer.go → core/indexer/indexer.go
View File

@@ -1,4 +1,4 @@
package indexers
package indexer
import (
"context"
@@ -16,20 +16,15 @@ import (
const (
maxReorgLookBack = 1000
// pollingInterval is the default polling interval for the indexer polling worker
pollingInterval = 15 * time.Second
)
type (
BitcoinProcessor Processor[[]*types.Block]
BitcoinDatasource datasources.Datasource[[]*types.Block]
)
// Make sure to implement the IndexerWorker interface
var _ IndexerWorker = (*BitcoinIndexer)(nil)
// BitcoinIndexer is the polling indexer for sync Bitcoin data to the database.
type BitcoinIndexer struct {
Processor BitcoinProcessor
Datasource BitcoinDatasource
// Indexer generic indexer for fetching and processing data
type Indexer[T Input] struct {
Processor Processor[T]
Datasource datasources.Datasource[T]
currentBlock types.BlockHeader
quitOnce sync.Once
@@ -37,9 +32,9 @@ type BitcoinIndexer struct {
done chan struct{}
}
// NewBitcoinIndexer create new BitcoinIndexer
func NewBitcoinIndexer(processor BitcoinProcessor, datasource BitcoinDatasource) *BitcoinIndexer {
return &BitcoinIndexer{
// New create new generic indexer
func New[T Input](processor Processor[T], datasource datasources.Datasource[T]) *Indexer[T] {
return &Indexer[T]{
Processor: processor,
Datasource: datasource,
@@ -48,21 +43,17 @@ func NewBitcoinIndexer(processor BitcoinProcessor, datasource BitcoinDatasource)
}
}
func (*BitcoinIndexer) Type() string {
return "bitcoin"
}
func (i *BitcoinIndexer) Shutdown() error {
func (i *Indexer[T]) Shutdown() error {
return i.ShutdownWithContext(context.Background())
}
func (i *BitcoinIndexer) ShutdownWithTimeout(timeout time.Duration) error {
func (i *Indexer[T]) ShutdownWithTimeout(timeout time.Duration) error {
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
return i.ShutdownWithContext(ctx)
}
func (i *BitcoinIndexer) ShutdownWithContext(ctx context.Context) (err error) {
func (i *Indexer[T]) ShutdownWithContext(ctx context.Context) (err error) {
i.quitOnce.Do(func() {
close(i.quit)
select {
@@ -76,12 +67,11 @@ func (i *BitcoinIndexer) ShutdownWithContext(ctx context.Context) (err error) {
return
}
func (i *BitcoinIndexer) Run(ctx context.Context) (err error) {
func (i *Indexer[T]) Run(ctx context.Context) (err error) {
defer close(i.done)
ctx = logger.WithContext(ctx,
slog.String("package", "indexers"),
slog.String("indexer", i.Type()),
slog.String("processor", i.Processor.Name()),
slog.String("datasource", i.Datasource.Name()),
)
@@ -114,15 +104,15 @@ func (i *BitcoinIndexer) Run(ctx context.Context) (err error) {
}
}
func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
func (i *Indexer[T]) process(ctx context.Context) (err error) {
// height range to fetch data
from, to := i.currentBlock.Height+1, int64(-1)
logger.InfoContext(ctx, "Start fetching bitcoin blocks", slog.Int64("from", from))
ch := make(chan []*types.Block)
logger.InfoContext(ctx, "Start fetching input data", slog.Int64("from", from))
ch := make(chan []T)
subscription, err := i.Datasource.FetchAsync(ctx, from, to, ch)
if err != nil {
return errors.Wrap(err, "failed to fetch data")
return errors.Wrap(err, "failed to fetch input data")
}
defer subscription.Unsubscribe()
@@ -130,21 +120,24 @@ func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
select {
case <-i.quit:
return nil
case blocks := <-ch:
// empty blocks
if len(blocks) == 0 {
case inputs := <-ch:
// empty inputs
if len(inputs) == 0 {
continue
}
firstInput := inputs[0]
firstInputHeader := firstInput.BlockHeader()
startAt := time.Now()
ctx := logger.WithContext(ctx,
slogx.Int64("from", blocks[0].Header.Height),
slogx.Int64("to", blocks[len(blocks)-1].Header.Height),
slogx.Int64("from", firstInputHeader.Height),
slogx.Int64("to", inputs[len(inputs)-1].BlockHeader().Height),
)
// validate reorg from first block
// validate reorg from first input
{
remoteBlockHeader := blocks[0].Header
remoteBlockHeader := firstInputHeader
if !remoteBlockHeader.PrevBlock.IsEqual(&i.currentBlock.Hash) {
logger.WarnContext(ctx, "Detected chain reorganization. Searching for fork point...",
slogx.String("event", "reorg_detected"),
@@ -210,33 +203,35 @@ func (i *BitcoinIndexer) process(ctx context.Context) (err error) {
}
}
// validate is block is continuous and no reorg
for i := 1; i < len(blocks); i++ {
if blocks[i].Header.Height != blocks[i-1].Header.Height+1 {
return errors.Wrapf(errs.InternalError, "block is not continuous, block[%d] height: %d, block[%d] height: %d", i-1, blocks[i-1].Header.Height, i, blocks[i].Header.Height)
// validate is input is continuous and no reorg
for i := 1; i < len(inputs); i++ {
header := inputs[i].BlockHeader()
prevHeader := inputs[i-1].BlockHeader()
if header.Height != prevHeader.Height+1 {
return errors.Wrapf(errs.InternalError, "input is not continuous, input[%d] height: %d, input[%d] height: %d", i-1, prevHeader.Height, i, header.Height)
}
if !blocks[i].Header.PrevBlock.IsEqual(&blocks[i-1].Header.Hash) {
logger.WarnContext(ctx, "Chain Reorganization occurred in the middle of batch fetching blocks, need to try to fetch again")
if !header.PrevBlock.IsEqual(&prevHeader.Hash) {
logger.WarnContext(ctx, "Chain Reorganization occurred in the middle of batch fetching inputs, need to try to fetch again")
// end current round
return nil
}
}
ctx = logger.WithContext(ctx, slog.Int("total_blocks", len(blocks)))
ctx = logger.WithContext(ctx, slog.Int("total_inputs", len(inputs)))
// Start processing blocks
logger.InfoContext(ctx, "Processing blocks")
if err := i.Processor.Process(ctx, blocks); err != nil {
// Start processing input
logger.InfoContext(ctx, "Processing inputs")
if err := i.Processor.Process(ctx, inputs); err != nil {
return errors.WithStack(err)
}
// Update current state
i.currentBlock = blocks[len(blocks)-1].Header
i.currentBlock = inputs[len(inputs)-1].BlockHeader()
logger.InfoContext(ctx, "Processed blocks successfully",
slogx.String("event", "processed_blocks"),
logger.InfoContext(ctx, "Processed inputs successfully",
slogx.String("event", "processed_inputs"),
slogx.Int64("current_block", i.currentBlock.Height),
slogx.Duration("duration", time.Since(startAt)),
)

20
core/indexers/indexers.go → core/indexer/interface.go
View File

@@ -1,30 +1,20 @@
package indexers
package indexer
import (
"context"
"time"
"github.com/gaze-network/indexer-network/core/types"
)
const (
// pollingInterval is the default polling interval for the indexer polling worker
pollingInterval = 15 * time.Second
)
type IndexerWorker interface {
Type() string
Run(ctx context.Context) error
Shutdown() error
ShutdownWithTimeout(timeout time.Duration) error
ShutdownWithContext(ctx context.Context) error
type Input interface {
BlockHeader() types.BlockHeader
}
type Processor[T any] interface {
type Processor[T Input] interface {
Name() string
// Process processes the input data and indexes it.
Process(ctx context.Context, inputs T) error
Process(ctx context.Context, inputs []T) error
// CurrentBlock returns the latest indexed block header.
CurrentBlock(ctx context.Context) (types.BlockHeader, error)

4
core/types/bitcoin_block.go
View File

@@ -38,6 +38,10 @@ type Block struct {
Transactions []*Transaction
}
func (b *Block) BlockHeader() BlockHeader {
return b.Header
}
func ParseMsgBlock(src *wire.MsgBlock, height int64) *Block {
hash := src.Header.BlockHash()
return &Block{

2
modules/bitcoin/btcclient/client_db.go
View File

@@ -24,7 +24,7 @@ const (
)
// Make sure to implement the BitcoinDatasource interface
var _ datasources.Datasource[[]*types.Block] = (*ClientDatabase)(nil)
var _ datasources.Datasource[*types.Block] = (*ClientDatabase)(nil)
// ClientDatabase is a client to connect to the bitcoin database.
type ClientDatabase struct {

6
modules/bitcoin/processor.go
View File

@@ -5,14 +5,14 @@ import (
"github.com/cockroachdb/errors"
"github.com/gaze-network/indexer-network/common/errs"
"github.com/gaze-network/indexer-network/core/indexers"
"github.com/gaze-network/indexer-network/core/indexer"
"github.com/gaze-network/indexer-network/core/types"
"github.com/gaze-network/indexer-network/internal/config"
"github.com/gaze-network/indexer-network/modules/bitcoin/datagateway"
)
// Make sure to implement the BitcoinProcessor interface
var _ indexers.BitcoinProcessor = (*Processor)(nil)
// Make sure to implement the Bitcoin Processor interface
var _ indexer.Processor[*types.Block] = (*Processor)(nil)
type Processor struct {
config config.Config

9
modules/runes/processor.go
View File

@@ -9,7 +9,7 @@ import (
"github.com/cockroachdb/errors"
"github.com/gaze-network/indexer-network/common"
"github.com/gaze-network/indexer-network/common/errs"
"github.com/gaze-network/indexer-network/core/indexers"
"github.com/gaze-network/indexer-network/core/indexer"
"github.com/gaze-network/indexer-network/core/types"
"github.com/gaze-network/indexer-network/modules/bitcoin/btcclient"
"github.com/gaze-network/indexer-network/modules/runes/datagateway"
@@ -22,13 +22,13 @@ import (
"github.com/samber/lo"
)
var _ indexers.BitcoinProcessor = (*Processor)(nil)
// Make sure to implement the Bitcoin Processor interface
var _ indexer.Processor[*types.Block] = (*Processor)(nil)
type Processor struct {
runesDg datagateway.RunesDataGateway
indexerInfoDg datagateway.IndexerInfoDataGateway
bitcoinClient btcclient.Contract
bitcoinDataSource indexers.BitcoinDatasource
network common.Network
reportingClient *reportingclient.ReportingClient
@@ -40,12 +40,11 @@ type Processor struct {
newRuneTxs []*entity.RuneTransaction
}
func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, bitcoinDataSource indexers.BitcoinDatasource, network common.Network, reportingClient *reportingclient.ReportingClient) *Processor {
func NewProcessor(runesDg datagateway.RunesDataGateway, indexerInfoDg datagateway.IndexerInfoDataGateway, bitcoinClient btcclient.Contract, network common.Network, reportingClient *reportingclient.ReportingClient) *Processor {
return &Processor{
runesDg: runesDg,
indexerInfoDg: indexerInfoDg,
bitcoinClient: bitcoinClient,
bitcoinDataSource: bitcoinDataSource,
network: network,
reportingClient: reportingClient,
newRuneEntries: make(map[runes.RuneId]*runes.RuneEntry),