diff --git a/README.md b/README.md index 545b1dd..a13bff8 100644 --- a/README.md +++ b/README.md @@ -34,22 +34,23 @@ For a full list of configuration options see [config.md](./config.md) ```yaml connectors: -- type: ethereum - server: - port: 5102 - ethereum: - url: http://localhost:8545 + - type: ethereum + server: + port: 5102 + ethereum: + url: http://localhost:8545 ``` ## Blockchain node compatibility For EVM connector to function properly, you should check the blockchain node supports the following JSON-RPC Methods over HTTP: + ### Event tracking + - `eth_blockNumber` - `eth_newBlockFilter` - `eth_getFilterLogs` - `eth_getFilterChanges` -- `eth_getBlockByHash` - `eth_getLogs` - `eth_newFilter` - `eth_uninstallFilter` @@ -57,17 +58,34 @@ For EVM connector to function properly, you should check the blockchain node sup - `eth_getTransactionReceipt` ### Query + - `eth_call` - `eth_getBalance` -- `eth_gasPrice`[^1] - +- `eth_gasPrice` + ### Transaction submission + - `eth_estimateGas` -- `eth_sendTransaction` +- `eth_sendTransaction` / `eth_sendRawTransaction` - `eth_getTransactionCount` -- `eth_sendRawTransaction`[^2] +### Optional methods + +#### Transaction tracing + +> Required when [connector.traceTXForRevertReason](./config.md#connector) is set to `true` (default: `false`) + +- `debug_traceTransaction` + +#### Block listener + +> Required when [connector.blockTrackingMode](./config.md#connector) is set to `inMemoryPartialChain` + +- `eth_getBlockByHash` +- `eth_getBlockByNumber` + +#### Receipt fetching -[^1]: also used by Transaction submission if the handler is configured to get gas price using "connector". +> Required when [connector.useGetBlockReceipts](./config.md#connector) is set to `true` -[^2]: only required by custom transaction handlers that supports pre-signing. +- `eth_getBlockReceipts` diff --git a/config.md b/config.md index 9abc6df..40cdb58 100644 --- a/config.md +++ b/config.md @@ -67,6 +67,7 @@ |---|-----------|----|-------------| |blockCacheSize|Maximum of blocks to hold in the block info cache|`int`|`250` |blockPollingInterval|Interval for polling to check for new blocks|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s` +|chainTrackingMode|Tracking mode for connector block progression. light: fetches head block numbers only, does not download block details, disables block listener support, and confirmation results include only the confirmation count. full: fetches head block numbers and block details, maintains an in-memory partial chain, enables block listener support, and confirmation results include both confirmation count and block details.|`light` or `full`|`full` |connectionTimeout|The maximum amount of time that a connection is allowed to remain with no data transmitted|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s` |dataFormat|Configure the JSON data format for query output and events|map,flat_array,self_describing|`map` |expectContinueTimeout|See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport)|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s` diff --git a/go.mod b/go.mod index c57bdcb..2638b4c 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,7 @@ require ( github.com/hashicorp/golang-lru v1.0.2 github.com/hyperledger/firefly-common v1.5.9 github.com/hyperledger/firefly-signer v1.1.23-0.20260422080826-42345c6c6b85 - github.com/hyperledger/firefly-transaction-manager v1.4.4 + github.com/hyperledger/firefly-transaction-manager v1.4.5 github.com/sirupsen/logrus v1.9.3 github.com/spf13/cobra v1.8.0 github.com/stretchr/testify v1.9.0 diff --git a/go.sum b/go.sum index 9bba158..4515a30 100644 --- a/go.sum +++ b/go.sum @@ -104,8 +104,8 @@ github.com/hyperledger/firefly-common v1.5.9 h1:Z1+SuKNYJ8hPKQ5CvcsMg6r/E4RyW6wb github.com/hyperledger/firefly-common v1.5.9/go.mod h1:1Xawm5PUhxT7k+CL/Kr3i1LE3cTTzoQwZMLimvlW8rs= github.com/hyperledger/firefly-signer v1.1.23-0.20260422080826-42345c6c6b85 h1:gh3YhxUYYwOfBCsEJXFmWO7SFzFrNuNulXftOam2JRI= github.com/hyperledger/firefly-signer v1.1.23-0.20260422080826-42345c6c6b85/go.mod h1:cb40Xkm/t2+KH+V1q3/zxZPohBNEA0iOA7mcr9wyfzI= -github.com/hyperledger/firefly-transaction-manager v1.4.4 h1:cbG9FkQWriOcc1MMGaMqU7OpOwLloSV+PImOoaN0ckU= -github.com/hyperledger/firefly-transaction-manager v1.4.4/go.mod h1:1kbYt8ofDXqfwC02vwV/HoOjmiv0IuT9UkJ//bbrliE= +github.com/hyperledger/firefly-transaction-manager v1.4.5 h1:zBe8hbzv6lJEWD5Ypk6efO5WXFs4+pqIFLUu7zbdmsg= +github.com/hyperledger/firefly-transaction-manager v1.4.5/go.mod h1:1kbYt8ofDXqfwC02vwV/HoOjmiv0IuT9UkJ//bbrliE= github.com/imdario/mergo v0.3.11/go.mod h1:jmQim1M+e3UYxmgPu/WyfjB3N3VflVyUjjjwH0dnCYA= github.com/imdario/mergo v0.3.16 h1:wwQJbIsHYGMUyLSPrEq1CT16AhnhNJQ51+4fdHUnCl4= github.com/imdario/mergo v0.3.16/go.mod h1:WBLT9ZmE3lPoWsEzCh9LPo3TiwVN+ZKEjmz+hD27ysY= diff --git a/internal/ethereum/config.go b/internal/ethereum/config.go index 1fb14c4..c113c1c 100644 --- a/internal/ethereum/config.go +++ b/internal/ethereum/config.go @@ -19,6 +19,7 @@ package ethereum import ( "github.com/hyperledger/firefly-common/pkg/config" "github.com/hyperledger/firefly-common/pkg/wsclient" + "github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi" ) const ( @@ -26,6 +27,7 @@ const ( ConfigDataFormat = "dataFormat" BlockPollingInterval = "blockPollingInterval" BlockCacheSize = "blockCacheSize" + ChainTrackingMode = "chainTrackingMode" EventsCatchupPageSize = "events.catchupPageSize" EventsCatchupThreshold = "events.catchupThreshold" EventsCatchupDownscaleRegex = "events.catchupDownscaleRegex" @@ -70,6 +72,7 @@ func InitConfig(conf config.Section) { conf.AddKnownKey(WebSocketsEnabled, false) conf.AddKnownKey(BlockCacheSize, 250) conf.AddKnownKey(BlockPollingInterval, "1s") + conf.AddKnownKey(ChainTrackingMode, ffcapi.ChainTrackingModeFull) conf.AddKnownKey(ConfigDataFormat, "map") conf.AddKnownKey(ConfigGasEstimationFactor, DefaultGasEstimationFactor) conf.AddKnownKey(EventsBlockTimestamps, true) diff --git a/internal/ethereum/ethereum.go b/internal/ethereum/ethereum.go index 9ecb073..c266f99 100644 --- a/internal/ethereum/ethereum.go +++ b/internal/ethereum/ethereum.go @@ -43,8 +43,10 @@ import ( ) type ethConnector struct { - backend rpcbackend.Backend - wsBackend rpcbackend.WebSocketRPCClient + backend rpcbackend.Backend + wsBackend rpcbackend.WebSocketRPCClient + chainTrackingMode ffcapi.ChainTrackingMode + serializer *abi.Serializer gasEstimationFactor *big.Float catchupPageSize int64 @@ -78,6 +80,15 @@ type Connector interface { } func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connector, err error) { + + chainTrackingMode := ffcapi.ChainTrackingMode(conf.GetString(ChainTrackingMode)) + if chainTrackingMode == "" { + chainTrackingMode = ffcapi.ChainTrackingModeFull + } + if chainTrackingMode != ffcapi.ChainTrackingModeLight && chainTrackingMode != ffcapi.ChainTrackingModeFull { + return nil, i18n.NewError(ctx, msgs.MsgInvalidChainTrackingMode, chainTrackingMode) + } + c := ðConnector{ eventStreams: make(map[fftypes.UUID]*eventStream), catchupPageSize: conf.GetInt64(EventsCatchupPageSize), @@ -86,6 +97,7 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto eventBlockTimestamps: conf.GetBool(EventsBlockTimestamps), eventFilterPollingInterval: conf.GetDuration(EventsFilterPollingInterval), traceTXForRevertReason: conf.GetBool(TraceTXForRevertReason), + chainTrackingMode: chainTrackingMode, retry: retryutil.RetryWrapper{Retry: &retry.Retry{}}, } @@ -164,6 +176,7 @@ func NewEthereumConnector(ctx context.Context, conf config.Section) (cc Connecto BlockCacheSize: conf.GetInt(BlockCacheSize), MaxAsyncBlockFetchConcurrency: conf.GetInt(MaxAsyncBlockFetchConcurrency), UseGetBlockReceipts: conf.GetBool(UseGetBlockReceipts), + ChainTrackingMode: c.chainTrackingMode, }, c.backend, c.wsBackend); err != nil { return nil, err } @@ -212,11 +225,12 @@ func (c *ethConnector) ReconcileConfirmationsForTransaction(ctx context.Context, } if err == nil && ethrpcRes != nil { res = &ffcapi.ConfirmationUpdateResult{ - Confirmations: ethRPCtoFFCAPIConfirmations(ethrpcRes.Confirmations), - Rebuilt: ethrpcRes.Rebuilt, - NewFork: ethrpcRes.NewFork, - Confirmed: ethrpcRes.Confirmed, - TargetConfirmationCount: ethrpcRes.TargetConfirmationCount, + Confirmations: ethRPCtoFFCAPIConfirmations(ethrpcRes.Confirmations), + Rebuilt: ethrpcRes.Rebuilt, + NewFork: ethrpcRes.NewFork, + Confirmed: ethrpcRes.Confirmed, + CurrentConfirmationCount: ethrpcRes.CurrentConfirmationCount, + TargetConfirmationCount: ethrpcRes.TargetConfirmationCount, } if ethrpcReceipt != nil { res.Receipt = c.enrichTransactionReceipt(ctx, ethrpcReceipt) diff --git a/internal/ethereum/new_block_listener.go b/internal/ethereum/new_block_listener.go index 77a3e42..0fb315b 100644 --- a/internal/ethereum/new_block_listener.go +++ b/internal/ethereum/new_block_listener.go @@ -23,6 +23,10 @@ import ( "github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi" ) +func (c *ethConnector) GetChainTrackingMode(_ context.Context) ffcapi.ChainTrackingMode { + return c.chainTrackingMode +} + func (c *ethConnector) NewBlockListener(_ context.Context, req *ffcapi.NewBlockListenerRequest) (*ffcapi.NewBlockListenerResponse, ffcapi.ErrorReason, error) { // Add the block consumer c.blockListener.AddConsumer(req.ListenerContext, ðblocklistener.BlockUpdateConsumer{ diff --git a/internal/msgs/en_config_descriptions.go b/internal/msgs/en_config_descriptions.go index cd8f45b..f98a959 100644 --- a/internal/msgs/en_config_descriptions.go +++ b/internal/msgs/en_config_descriptions.go @@ -51,4 +51,5 @@ var ( _ = ffc("config.connector.traceTXForRevertReason", "Enable the use of transaction trace functions (e.g. debug_traceTransaction) to obtain transaction revert reasons. This can place a high load on the EVM client.", i18n.BooleanType) _ = ffc("config.connector.maxAsyncBlockFetchConcurrency", "Maximum concurrency when using asynchronous block downloading (minium 1)", i18n.IntType) _ = ffc("config.connector.useGetBlockReceipts", "When true, the eth_getBlockReceipts call is available for this connector to use", i18n.BooleanType) + _ = ffc("config.connector.chainTrackingMode", "Tracking mode for connector block progression. light: fetches head block numbers only, does not download block details, disables block listener support, and confirmation results include only the confirmation count. full: fetches head block numbers and block details, maintains an in-memory partial chain, enables block listener support, and confirmation results include both confirmation count and block details.", "`light` or `full`") ) diff --git a/internal/msgs/en_error_messages.go b/internal/msgs/en_error_messages.go index 08b32bf..73485fc 100644 --- a/internal/msgs/en_error_messages.go +++ b/internal/msgs/en_error_messages.go @@ -86,4 +86,6 @@ var ( MsgUnknownJSONFormatOptions = ffe("FF23066", "JSON formatting option unknown %s=%s") MsgObservedPanic = ffe("FF23067", "Observed panic: %v") MsgReturnedBlockHashMismatch = ffe("FF23068", "Returned block %d hash %s does not match requested hash %s") + MsgInvalidChainTrackingMode = ffe("FF23069", "Invalid chain tracking mode '%s': must be 'light' or 'full'") + MsgTransactionNotIncludedInChainHead = ffe("FF23070", "Transaction '%s' cannot be reconciled because chain head %d is before receipt block %s") ) diff --git a/mocks/fftmmocks/manager.go b/mocks/fftmmocks/manager.go index 40e8c7a..444bc8c 100644 --- a/mocks/fftmmocks/manager.go +++ b/mocks/fftmmocks/manager.go @@ -11,6 +11,8 @@ import ( ffcapi "github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi" + metric "github.com/hyperledger/firefly-common/pkg/metric" + mock "github.com/stretchr/testify/mock" mux "github.com/gorilla/mux" @@ -133,6 +135,26 @@ func (_m *Manager) GetTransactionByIDWithStatus(ctx context.Context, txID string return r0, r1 } +// MetricsRegistry provides a mock function with no fields +func (_m *Manager) MetricsRegistry() metric.MetricsRegistry { + ret := _m.Called() + + if len(ret) == 0 { + panic("no return value specified for MetricsRegistry") + } + + var r0 metric.MetricsRegistry + if rf, ok := ret.Get(0).(func() metric.MetricsRegistry); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(metric.MetricsRegistry) + } + } + + return r0 +} + // ReconcileConfirmationsForTransaction provides a mock function with given fields: ctx, txHash, existingConfirmations, targetConfirmationCount func (_m *Manager) ReconcileConfirmationsForTransaction(ctx context.Context, txHash string, existingConfirmations []*ffcapi.MinimalBlockInfo, targetConfirmationCount uint64) (*ffcapi.ConfirmationUpdateResult, error) { ret := _m.Called(ctx, txHash, existingConfirmations, targetConfirmationCount) diff --git a/pkg/ethblocklistener/blocklistener.go b/pkg/ethblocklistener/blocklistener.go index ddba6fb..3462aa3 100644 --- a/pkg/ethblocklistener/blocklistener.go +++ b/pkg/ethblocklistener/blocklistener.go @@ -49,21 +49,23 @@ import ( // // `rebuilt` will be true if an invalid confirmation list is detected by the reconciliation process type ConfirmationUpdateResult struct { - Confirmations []*ethrpc.MinimalBlockInfo `json:"confirmations,omitempty"` - Rebuilt bool `json:"rebuilt,omitempty"` // when true, it means the existing confirmations contained invalid blocks, the new confirmations are rebuilt from scratch - NewFork bool `json:"newFork,omitempty"` // when true, it means a new fork was detected based on the existing confirmations - Confirmed bool `json:"confirmed,omitempty"` // when true, it means the confirmation list is complete and the transaction is confirmed - TargetConfirmationCount uint64 `json:"targetConfirmationCount"` // the target number of confirmations for this reconcile request + Confirmations []*ethrpc.MinimalBlockInfo `json:"confirmations,omitempty"` // the confirmation list + Rebuilt bool `json:"rebuilt,omitempty"` // when true, it means the existing confirmations contained invalid blocks, the new confirmations are rebuilt from scratch + NewFork bool `json:"newFork,omitempty"` // when true, it means a new fork was detected based on the existing confirmations + Confirmed bool `json:"confirmed,omitempty"` // when true, it means the confirmation list is complete and the transaction is confirmed + TargetConfirmationCount uint64 `json:"targetConfirmationCount"` // the target number of confirmations for this reconcile request + CurrentConfirmationCount uint64 `json:"currentConfirmationCount"` // the current number of confirmations for this reconcile request } type BlockListenerConfig struct { - MonitoredHeadLength int `json:"monitoredHeadLength"` - BlockPollingInterval time.Duration `json:"blockPollingInterval"` - HederaCompatibilityMode bool `json:"hederaCompatibilityMode"` - BlockCacheSize int `json:"blockCacheSize"` - IncludeLogsBloom bool `json:"includeLogsBloom"` - UseGetBlockReceipts bool `json:"useGetBlockReceipts"` - MaxAsyncBlockFetchConcurrency int `json:"maxAsyncBlockFetchConcurrency"` + MonitoredHeadLength int `json:"monitoredHeadLength"` + BlockPollingInterval time.Duration `json:"blockPollingInterval"` + HederaCompatibilityMode bool `json:"hederaCompatibilityMode"` + BlockCacheSize int `json:"blockCacheSize"` + IncludeLogsBloom bool `json:"includeLogsBloom"` + UseGetBlockReceipts bool `json:"useGetBlockReceipts"` + MaxAsyncBlockFetchConcurrency int `json:"maxAsyncBlockFetchConcurrency"` + ChainTrackingMode ffcapi.ChainTrackingMode `json:"chainTrackingMode,omitempty"` } type BlockListener interface { @@ -126,6 +128,9 @@ type blockListener struct { canonicalChain *list.List highestBlockSet bool highestBlock uint64 + + // headBlockNumber mode: last head value sent on the block listener channel (only written from listenLoop) + currentChainHead uint64 } func NewBlockListener(ctx context.Context, retry *retry.Retry, conf *BlockListenerConfig, httpConf *ffresty.Config, wsConf *wsclient.WSConfig) (bl BlockListener, err error) { @@ -141,6 +146,9 @@ func NewBlockListenerSupplyBackend(ctx context.Context, retry *retry.Retry, conf if conf.MaxAsyncBlockFetchConcurrency <= 0 { conf.MaxAsyncBlockFetchConcurrency = 1 } + if conf.ChainTrackingMode == "" { + conf.ChainTrackingMode = ffcapi.ChainTrackingModeFull + } bl := &blockListener{ ctx: log.WithLogField(ctx, "role", "blocklistener"), retry: &retryutil.RetryWrapper{Retry: retry}, @@ -152,6 +160,7 @@ func NewBlockListenerSupplyBackend(ctx context.Context, retry *retry.Retry, conf newHeadsTap: make(chan struct{}), highestBlockSet: false, highestBlock: 0, + currentChainHead: 0, consumers: make(map[fftypes.UUID]*BlockUpdateConsumer), canonicalChain: list.New(), blockFetchConcurrencyThrottle: make(chan *blockReceiptRequest, conf.MaxAsyncBlockFetchConcurrency), @@ -248,6 +257,17 @@ func (bl *blockListener) establishBlockHeightWithRetry() error { }) } +// refreshHighestBlockFromRPC updates highestBlock from eth_blockNumber. Caller must not hold canonicalChainLock. +func (bl *blockListener) refreshHighestBlockFromRPC() (uint64, error) { + var hexBlockHeight ethtypes.HexInteger + rpcErr := bl.backend.CallRPC(bl.ctx, &hexBlockHeight, "eth_blockNumber") + if rpcErr != nil { + return 0, rpcErr.Error() + } + head := hexBlockHeight.BigInt().Uint64() + return head, nil +} + func (bl *blockListener) waitNextIteration() bool { select { case <-bl.ctx.Done(): @@ -313,6 +333,30 @@ func (bl *blockListener) listenLoop() { } log.L(bl.ctx).Debugf("Block filter received new block hashes: %+v", blockHashes) + if bl.ChainTrackingMode == ffcapi.ChainTrackingModeLight { + head, err := bl.refreshHighestBlockFromRPC() + if err != nil { + log.L(bl.ctx).Errorf("Failed to refresh chain head: %s", err) + failCount++ + continue + } + if head == bl.currentChainHead { + failCount = 0 + continue + } + bl.currentChainHead = head + update := &ffcapi.BlockHashEvent{GapPotential: false, Created: fftypes.Now(), HeadBlockNumber: bl.currentChainHead} + bl.consumerMux.Lock() + consumers := make([]*BlockUpdateConsumer, 0, len(bl.consumers)) + for _, c := range bl.consumers { + consumers = append(consumers, c) + } + bl.consumerMux.Unlock() + bl.dispatchToConsumers(consumers, update) + failCount = 0 + continue + } + update := &ffcapi.BlockHashEvent{GapPotential: gapPotential, Created: fftypes.Now()} var notifyPos *list.Element for _, h := range blockHashes { @@ -621,6 +665,10 @@ func (bl *blockListener) GetHighestBlock(ctx context.Context) (uint64, bool) { return highestBlock, true } +func (bl *blockListener) GetHeadBlockNumber(_ context.Context) uint64 { + return bl.currentChainHead +} + func (bl *blockListener) setHighestBlock(block uint64) { bl.canonicalChainLock.Lock() defer bl.canonicalChainLock.Unlock() diff --git a/pkg/ethblocklistener/blocklistener_test.go b/pkg/ethblocklistener/blocklistener_test.go index cc25395..e7999a9 100644 --- a/pkg/ethblocklistener/blocklistener_test.go +++ b/pkg/ethblocklistener/blocklistener_test.go @@ -1747,6 +1747,79 @@ func TestBlockListenerWaitUntilStartedOnlyReturnsAfterEstablishingBlockFilter(t mRPC.AssertExpectations(t) } +// TestBlockListenerHeadBlockNumber_DispatchesAndSkipsDuplicateHead exercises listenLoop head-only mode: +// eth_blockNumber refresh updates currentChainHead and dispatches BlockHashEvent with HeadBlockNumber; +// when the RPC head is unchanged, no event is sent. +func TestBlockListenerHeadBlockNumber_DispatchesAndSkipsDuplicateHead(t *testing.T) { + // Block the first getFilterChanges until after AddConsumer registers; otherwise markStarted() + // unblocks waitUntilStarted before the first head refresh+dispatch, so the first event can be + // sent with zero consumers. + startLatch := newTestLatch() + var bnCall int + _, bl, mRPC, done := newTestBlockListener(t, func(conf *BlockListenerConfig, mRPC *rpcbackendmocks.Backend, _ context.CancelFunc) { + conf.BlockPollingInterval = shortDelay + conf.ChainTrackingMode = ffcapi.ChainTrackingModeLight + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_blockNumber").Return(nil).Run(func(args mock.Arguments) { + bnCall++ + hbh := args[1].(*ethtypes.HexInteger) + var v uint64 + switch bnCall { + case 1: + v = 1000 // establishBlockHeightWithRetry + case 2: + v = 1000 // first refresh: currentChainHead was 0 → dispatch + case 3: + v = 1000 // second refresh: same as currentChainHead → no dispatch + case 4: + v = 1001 // head advanced → dispatch + default: + v = 1001 // stable after cancel + } + *hbh = *ethtypes.NewHexIntegerU64(v) + }).Maybe() + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_newBlockFilter").Return(nil).Run(func(args mock.Arguments) { + *args[1].(*string) = testBlockFilterID1 + }).Once() + + var getFilterCalls int + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getFilterChanges", testBlockFilterID1).Return(nil).Run(func(args mock.Arguments) { + getFilterCalls++ + if getFilterCalls == 1 { + startLatch.waitComplete() + } + hbh := args[1].(*[]ethtypes.HexBytes0xPrefix) + *hbh = nil + }).Maybe() + }) + defer done() + + updates := make(chan *ffcapi.BlockHashEvent, 16) + bl.AddConsumer(context.Background(), &BlockUpdateConsumer{ + ID: fftypes.NewUUID(), + Ctx: context.Background(), + Updates: updates, + }) + startLatch.complete() + + ev1 := <-updates + assert.Equal(t, uint64(1000), ev1.HeadBlockNumber) + assert.False(t, ev1.GapPotential) + assert.Empty(t, ev1.BlockHashes) + + ev2 := <-updates + assert.Equal(t, uint64(1001), ev2.HeadBlockNumber) + assert.False(t, ev2.GapPotential) + assert.Empty(t, ev2.BlockHashes) + + done() + <-bl.listenLoopDone + + assert.Equal(t, uint64(1001), bl.currentChainHead) + mRPC.AssertExpectations(t) +} + func TestBlockListenerDispatchStopped(t *testing.T) { _, bl, _, done := newTestBlockListener(t) done() diff --git a/pkg/ethblocklistener/confirmation_reconciler.go b/pkg/ethblocklistener/confirmation_reconciler.go index f6ec687..9769ac1 100644 --- a/pkg/ethblocklistener/confirmation_reconciler.go +++ b/pkg/ethblocklistener/confirmation_reconciler.go @@ -24,6 +24,7 @@ import ( "github.com/hyperledger/firefly-evmconnect/internal/msgs" "github.com/hyperledger/firefly-evmconnect/pkg/ethrpc" "github.com/hyperledger/firefly-signer/pkg/ethtypes" + "github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi" ) func toBlockInfoList(ffcapiBlocks []*ethrpc.MinimalBlockInfo) (blocks []*ethrpc.BlockInfoJSONRPC) { @@ -40,6 +41,32 @@ func toBlockInfoList(ffcapiBlocks []*ethrpc.MinimalBlockInfo) (blocks []*ethrpc. func (bl *blockListener) ReconcileConfirmationsForTransaction(ctx context.Context, txHash string, existingConfirmations []*ethrpc.MinimalBlockInfo, targetConfirmationCount uint64) (*ConfirmationUpdateResult, *ethrpc.TxReceiptJSONRPC, error) { + if bl.BlockListenerConfig.ChainTrackingMode == ffcapi.ChainTrackingModeLight { + // when chain head is the only thing that's being tracked, we only need to calculate the confirmation list based on the head block number + // get the transaction receipt only + txReceipt, err := bl.GetTransactionReceipt(ctx, txHash) + if err != nil { + log.L(ctx).Errorf("Failed to fetch transaction receipt using tx hash %s: %v", txHash, err) + return nil, nil, err + } + // compare it against the chain head + chainHead := bl.GetHeadBlockNumber(ctx) + if chainHead < txReceipt.BlockNumber.Uint64() { + // the transaction is not yet included in the chain head + return nil, nil, i18n.NewError(ctx, msgs.MsgTransactionNotIncludedInChainHead, txHash, chainHead, txReceipt.BlockNumber.String()) + } + currentConfirmationCount := chainHead - txReceipt.BlockNumber.Uint64() + if currentConfirmationCount > targetConfirmationCount { + currentConfirmationCount = targetConfirmationCount + } + return &ConfirmationUpdateResult{ + Confirmed: currentConfirmationCount == targetConfirmationCount, + // no support on fork detection in this mode from the connector although firefly transaction manager will inject `NewFork: true` if it detects a reduction in confirmationCount. + CurrentConfirmationCount: currentConfirmationCount, + TargetConfirmationCount: targetConfirmationCount, + }, txReceipt, nil + } + blockInfoExistingConfirmations := toBlockInfoList(existingConfirmations) // Fetch the block containing the transaction first so that we can use it to build the confirmation list @@ -56,6 +83,7 @@ func (bl *blockListener) ReconcileConfirmationsForTransaction(ctx context.Contex confirmationUpdateResult, err := bl.buildConfirmationList(ctx, blockInfoExistingConfirmations, txBlockInfo, targetConfirmationCount) if confirmationUpdateResult != nil { confirmationUpdateResult.TargetConfirmationCount = targetConfirmationCount + confirmationUpdateResult.CurrentConfirmationCount = uint64(len(confirmationUpdateResult.Confirmations)) - 1 // NOTE: This function does not do the full receipt decoding, for which there is a complex function for. // The "Receipt" object is left empty (but the JSON/RPC receipt is return to the caller for enrichment) } diff --git a/pkg/ethblocklistener/confirmation_reconciler_test.go b/pkg/ethblocklistener/confirmation_reconciler_test.go index aeb190f..fa7ce29 100644 --- a/pkg/ethblocklistener/confirmation_reconciler_test.go +++ b/pkg/ethblocklistener/confirmation_reconciler_test.go @@ -30,6 +30,7 @@ import ( "github.com/hyperledger/firefly-evmconnect/pkg/ethrpc" "github.com/hyperledger/firefly-signer/pkg/ethtypes" "github.com/hyperledger/firefly-signer/pkg/rpcbackend" + "github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" ) @@ -67,6 +68,22 @@ const sampleJSONRPCReceipt = `{ "type": "0x0" }` +// Transaction hash in sampleJSONRPCReceipt (block 0x7b9 = 1977). +const headModeSampleTxHash = "0x7d48ae971faf089878b57e3c28e3035540d34f38af395958d2c73c36c57c83a2" + +func headBlockNumberTestConf(conf *BlockListenerConfig, _ *rpcbackendmocks.Backend, _ context.CancelFunc) { + conf.ChainTrackingMode = ffcapi.ChainTrackingModeLight +} + +func mockHeadModeReceipt(t *testing.T, mRPC *rpcbackendmocks.Backend, txHash string) { + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getTransactionReceipt", txHash). + Return(nil). + Run(func(args mock.Arguments) { + err := json.Unmarshal([]byte(sampleJSONRPCReceipt), args[1]) + assert.NoError(t, err) + }) +} + // Tests of the reconcileConfirmationsForTransaction function func TestReconcileConfirmationsForTransaction_TransactionNotFound(t *testing.T) { @@ -91,6 +108,90 @@ func TestReconcileConfirmationsForTransaction_TransactionNotFound(t *testing.T) mRPC.AssertExpectations(t) } +func TestReconcileConfirmationsForTransaction_HeadBlockNumber_ChainHeadBehindReceipt(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, headBlockNumberTestConf) + defer done() + + mockHeadModeReceipt(t, mRPC, headModeSampleTxHash) + bl.currentChainHead = 1976 // receipt block is 1977 (0x7b9) + + result, receipt, err := bl.ReconcileConfirmationsForTransaction(context.Background(), headModeSampleTxHash, nil, 5) + assert.Error(t, err) + assert.Regexp(t, "FF23070", err) + assert.Nil(t, result) + assert.Nil(t, receipt) +} + +func TestReconcileConfirmationsForTransaction_HeadBlockNumber_PartialConfirmations(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, headBlockNumberTestConf) + defer done() + + mockHeadModeReceipt(t, mRPC, headModeSampleTxHash) + bl.currentChainHead = 1980 // 1980 - 1977 = 3 confirmations vs target 5 + + result, receipt, err := bl.ReconcileConfirmationsForTransaction(context.Background(), headModeSampleTxHash, nil, 5) + assert.NoError(t, err) + if assert.NotNil(t, receipt) { + assert.Equal(t, uint64(1977), receipt.BlockNumber.Uint64()) + } + if assert.NotNil(t, result) { + assert.False(t, result.Confirmed) + assert.Equal(t, uint64(3), result.CurrentConfirmationCount) + assert.Equal(t, uint64(5), result.TargetConfirmationCount) + } +} + +func TestReconcileConfirmationsForTransaction_HeadBlockNumber_FullyConfirmed(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, headBlockNumberTestConf) + defer done() + + mockHeadModeReceipt(t, mRPC, headModeSampleTxHash) + bl.currentChainHead = 1982 // 1982 - 1977 = 5 confirmations + + result, receipt, err := bl.ReconcileConfirmationsForTransaction(context.Background(), headModeSampleTxHash, nil, 5) + assert.NoError(t, err) + if assert.NotNil(t, receipt) && assert.NotNil(t, result) { + assert.True(t, result.Confirmed) + assert.Equal(t, uint64(5), result.CurrentConfirmationCount) + assert.Equal(t, uint64(5), result.TargetConfirmationCount) + } +} + +func TestReconcileConfirmationsForTransaction_HeadBlockNumber_ActualCountCappedAtTarget(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, headBlockNumberTestConf) + defer done() + + mockHeadModeReceipt(t, mRPC, headModeSampleTxHash) + bl.currentChainHead = 2500 // many blocks past receipt; cap at target + + result, _, err := bl.ReconcileConfirmationsForTransaction(context.Background(), headModeSampleTxHash, nil, 3) + assert.NoError(t, err) + if assert.NotNil(t, result) { + assert.True(t, result.Confirmed) + assert.Equal(t, uint64(3), result.CurrentConfirmationCount) + assert.Equal(t, uint64(3), result.TargetConfirmationCount) + } +} + +func TestReconcileConfirmationsForTransaction_HeadBlockNumber_ReceiptRPCError(t *testing.T) { + _, bl, mRPC, done := newTestBlockListener(t, headBlockNumberTestConf) + defer done() + + mRPC.On("CallRPC", mock.Anything, mock.Anything, "eth_getTransactionReceipt", headModeSampleTxHash). + Return(&rpcbackend.RPCError{Message: "pop"}). + Run(func(args mock.Arguments) { + err := json.Unmarshal([]byte("null"), args[1]) + assert.NoError(t, err) + }) + + bl.currentChainHead = 2000 + + result, receipt, err := bl.ReconcileConfirmationsForTransaction(context.Background(), headModeSampleTxHash, nil, 5) + assert.Error(t, err) + assert.Nil(t, result) + assert.Nil(t, receipt) +} + func TestReconcileConfirmationsForTransaction_ReceiptRPCCallError(t *testing.T) { _, bl, mRPC, done := newTestBlockListener(t)