From 069c73de4fb3c3552a537142548d449e7d524cd9 Mon Sep 17 00:00:00 2001 From: Azzurriii Date: Fri, 29 May 2026 14:27:45 +0700 Subject: [PATCH] feat: add Solana fee conversion and improve handling of skipped slots in workers --- configs/config.example.yaml | 1 + internal/indexer/solana.go | 2 +- internal/indexer/solana_test.go | 1 + internal/rpc/solana/fee.go | 10 ++++++++ internal/worker/base.go | 7 +++++ internal/worker/catchup.go | 3 +-- internal/worker/regular.go | 6 +++++ internal/worker/regular_test.go | 45 +++++++++++++++++++++++++++++++++ internal/worker/rescanner.go | 3 +-- 9 files changed, 73 insertions(+), 5 deletions(-) create mode 100644 internal/rpc/solana/fee.go diff --git a/configs/config.example.yaml b/configs/config.example.yaml index a334d9f..0ce757f 100644 --- a/configs/config.example.yaml +++ b/configs/config.example.yaml @@ -191,6 +191,7 @@ chains: nodes: - url: "https://api.devnet.solana.com" - url: "https://solana-devnet.api.onfinality.io/public" + - url: "https://solana-devnet.gateway.tatum.io/" client: timeout: "12s" max_retries: 2 diff --git a/internal/indexer/solana.go b/internal/indexer/solana.go index 979d572..2b26ed1 100644 --- a/internal/indexer/solana.go +++ b/internal/indexer/solana.go @@ -477,7 +477,7 @@ func (s *SolanaIndexer) extractSolanaTransfers(networkID string, slot uint64, ts continue } txHash := tx.Transaction.Signatures[0] - fee := decimal.NewFromInt(int64(tx.Meta.Fee)) + fee := solana.FeeLamportsToSOL(tx.Meta.Fee) accountKeys := tx.Transaction.Message.AccountKeys // Build token-account -> (owner, mint) lookup from token balance metadata. diff --git a/internal/indexer/solana_test.go b/internal/indexer/solana_test.go index 91c455c..da24854 100644 --- a/internal/indexer/solana_test.go +++ b/internal/indexer/solana_test.go @@ -130,6 +130,7 @@ func TestSolanaBlockHashAndTransferIndex(t *testing.T) { for _, tx := range transfers { assert.Equal(t, blockHash, tx.BlockHash, "BlockHash should be propagated") assert.NotEmpty(t, tx.TransferIndex, "TransferIndex should be set") + assert.Equal(t, "0.000005", tx.TxFee.String(), "TxFee should be denominated in SOL") } // TransferIndexes should be unique diff --git a/internal/rpc/solana/fee.go b/internal/rpc/solana/fee.go new file mode 100644 index 0000000..4f64888 --- /dev/null +++ b/internal/rpc/solana/fee.go @@ -0,0 +1,10 @@ +package solana + +import "github.com/shopspring/decimal" + +const LamportsPerSOL = 1_000_000_000 + +func FeeLamportsToSOL(lamports uint64) decimal.Decimal { + return decimal.NewFromInt(int64(lamports)). + Div(decimal.NewFromInt(LamportsPerSOL)) +} diff --git a/internal/worker/base.go b/internal/worker/base.go index a36fb8a..cd836d4 100644 --- a/internal/worker/base.go +++ b/internal/worker/base.go @@ -11,6 +11,7 @@ import ( "github.com/fystack/multichain-indexer/internal/indexer" "github.com/fystack/multichain-indexer/internal/status" "github.com/fystack/multichain-indexer/pkg/common/config" + "github.com/fystack/multichain-indexer/pkg/common/enum" "github.com/fystack/multichain-indexer/pkg/common/logger" "github.com/fystack/multichain-indexer/pkg/common/types" "github.com/fystack/multichain-indexer/pkg/events" @@ -141,6 +142,12 @@ func (bw *BaseWorker) notifyObserver(blockNumber uint64, status BlockStatus) { } } +func (bw *BaseWorker) isSkippableNotFound(result indexer.BlockResult) bool { + return result.Error != nil && + result.Error.ErrorType == indexer.ErrorTypeBlockNotFound && + bw.chain.GetNetworkType() == enum.NetworkTypeSol +} + // handleBlockResult processes a block result and persists/forwards errors if needed. func (bw *BaseWorker) handleBlockResult(result indexer.BlockResult) bool { registry := status.EnsureStatusRegistry(bw.statusRegistry) diff --git a/internal/worker/catchup.go b/internal/worker/catchup.go index 2af7cdb..cd3951b 100644 --- a/internal/worker/catchup.go +++ b/internal/worker/catchup.go @@ -9,7 +9,6 @@ import ( "github.com/fystack/multichain-indexer/internal/indexer" "github.com/fystack/multichain-indexer/internal/status" "github.com/fystack/multichain-indexer/pkg/common/config" - "github.com/fystack/multichain-indexer/pkg/common/enum" "github.com/fystack/multichain-indexer/pkg/events" "github.com/fystack/multichain-indexer/pkg/infra" "github.com/fystack/multichain-indexer/pkg/store/blockstore" @@ -287,7 +286,7 @@ func (cw *CatchupWorker) processRange(r blockstore.CatchupRange, workerID int) e // Process results batchSuccess := current - 1 for _, res := range results { - if res.Error != nil && res.Error.ErrorType == indexer.ErrorTypeBlockNotFound && cw.chain.GetNetworkType() == enum.NetworkTypeSol { + if cw.isSkippableNotFound(res) { // Solana skipped slots are normal — the validator didn't produce a block // for this slot. Skip without retry. cw.logger.Debug("Solana skipped slot, no retry needed", "slot", res.Number) diff --git a/internal/worker/regular.go b/internal/worker/regular.go index 4d35d76..235d8b7 100644 --- a/internal/worker/regular.go +++ b/internal/worker/regular.go @@ -146,6 +146,12 @@ func (rw *RegularWorker) processRegularBlocks() error { stopTick, processErr = rw.processReorgCheckedBatch(results, end, &lastSuccess, &lastSuccessHash) } else { for _, res := range results { + if rw.isSkippableNotFound(res) { + rw.logger.Debug("Solana skipped slot, no retry needed", "slot", res.Number) + rw.notifyObserver(res.Number, BlockStatusNotFound) + lastSuccess = res.Number + continue + } if rw.handleBlockResult(res) { lastSuccess = res.Number lastSuccessHash = res.Block.Hash diff --git a/internal/worker/regular_test.go b/internal/worker/regular_test.go index 73573f1..30daf92 100644 --- a/internal/worker/regular_test.go +++ b/internal/worker/regular_test.go @@ -118,6 +118,51 @@ func TestRegularWorkerProcessRegularBlocksMarksUnresolvedGapFailed(t *testing.T) require.Equal(t, []uint64{100, 100}, chain.getBlockCalls) } +func TestRegularWorkerProcessRegularBlocksSkipsSolanaSkippedSlot(t *testing.T) { + t.Parallel() + + chain := &stubIndexer{ + name: "solana", + internalCode: "SOL_MAINNET", + networkType: enum.NetworkTypeSol, + latest: 102, + getBlocksFunc: func(context.Context, uint64, uint64, bool) ([]indexer.BlockResult, error) { + return []indexer.BlockResult{ + { + Number: 100, + Block: &types.Block{ + Number: 100, + Hash: "sol100", + }, + }, + { + Number: 101, + Error: &indexer.Error{ + ErrorType: indexer.ErrorTypeBlockNotFound, + Message: "block not found (skipped slot?)", + }, + }, + { + Number: 102, + Block: &types.Block{ + Number: 102, + Hash: "sol102", + }, + }, + }, nil + }, + } + store := &stubBlockStore{} + rw := newTestRegularWorker(chain, store, 100, 3) + + err := rw.processRegularBlocks() + require.NoError(t, err) + require.Equal(t, uint64(103), rw.currentBlock) + require.Equal(t, []uint64{102}, store.savedLatest) + require.Empty(t, store.failedBlocks) + require.Empty(t, chain.getBlockCalls) +} + func TestCheckContinuityReturnsFalseForNilBlocks(t *testing.T) { t.Parallel() diff --git a/internal/worker/rescanner.go b/internal/worker/rescanner.go index 72ff67e..88a2385 100644 --- a/internal/worker/rescanner.go +++ b/internal/worker/rescanner.go @@ -9,7 +9,6 @@ import ( "github.com/fystack/multichain-indexer/internal/indexer" "github.com/fystack/multichain-indexer/internal/status" "github.com/fystack/multichain-indexer/pkg/common/config" - "github.com/fystack/multichain-indexer/pkg/common/enum" "github.com/fystack/multichain-indexer/pkg/events" "github.com/fystack/multichain-indexer/pkg/infra" "github.com/fystack/multichain-indexer/pkg/store/blockstore" @@ -203,7 +202,7 @@ func (rw *RescannerWorker) processBatch(blocks []uint64) error { success := 0 for _, res := range results { - if res.Error != nil && res.Error.ErrorType == indexer.ErrorTypeBlockNotFound && rw.chain.GetNetworkType() == enum.NetworkTypeSol { + if rw.isSkippableNotFound(res) { // Solana skipped slots are normal. Do not retry them. toRemove = append(toRemove, res.Number) continue