introduce a light weight mode for tracking block progression

Signed-off-by: Chengxuan Xing <chengxuan.xing@kaleido.io>

Author: Chengxuan

README.md

@@ -34,40 +34,63 @@ 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`
 - `eth_getTransactionByHash`
 - `eth_getTransactionReceipt`
 
 ### Query
+
 - `eth_call`
 - `eth_getBalance`
 - `eth_gasPrice`[^1]
-  
+
 ### Transaction submission
+
 - `eth_estimateGas`
 - `eth_sendTransaction`
 - `eth_getTransactionCount`
 - `eth_sendRawTransaction`[^2]
 
-
 [^1]: also used by Transaction submission if the handler is configured to get gas price using "connector".
 
 [^2]: only required by custom transaction handlers that supports pre-signing.
+
+### 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
+
+> Required when [connector.useGetBlockReceipts](./config.md#connector) is set to `true`
+
+- `eth_getBlockReceipts`

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`
+|blockTrackingMode|The block tracking mode for the block listener|`headBlockNumber` or `inMemoryPartialChain`|`inMemoryPartialChain`
 |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`

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.21
-	github.com/hyperledger/firefly-transaction-manager v1.4.4
+	github.com/hyperledger/firefly-transaction-manager v0.0.0-20260421111941-f20dee0fc832
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.8.0
 	github.com/stretchr/testify v1.9.0
@@ -99,3 +99,6 @@ require (
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
+
+// Pin kaleido-io/firefly-transaction-manager branch lightweight-confirmation-tracking until merged upstream.
+replace github.com/hyperledger/firefly-transaction-manager => github.com/kaleido-io/firefly-transaction-manager v0.0.0-20260421111941-f20dee0fc832

go.sum

@@ -104,8 +104,6 @@ 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.21 h1:r7cTOw6e/6AtiXLf84wZy6Z7zppzlc191HokW2hv4N4=
 github.com/hyperledger/firefly-signer v1.1.21/go.mod h1:axrlSQeKrd124UdHF5L3MkTjb5DeTcbJxJNCZ3JmcWM=
-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/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=
@@ -115,6 +113,8 @@ github.com/jarcoal/httpmock v1.2.0 h1:gSvTxxFR/MEMfsGrvRbdfpRUMBStovlSRLw0Ep1bww
 github.com/jarcoal/httpmock v1.2.0/go.mod h1:oCoTsnAz4+UoOUIf5lJOWV2QQIW5UoeUI6aM2YnWAZk=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
+github.com/kaleido-io/firefly-transaction-manager v0.0.0-20260421111941-f20dee0fc832 h1:zkq96Tp9IfMa7isUBfXP0OS3gvddx8+vDlnqGDTX+tk=
+github.com/kaleido-io/firefly-transaction-manager v0.0.0-20260421111941-f20dee0fc832/go.mod h1:1kbYt8ofDXqfwC02vwV/HoOjmiv0IuT9UkJ//bbrliE=
 github.com/karlseguin/ccache v2.0.3+incompatible h1:j68C9tWOROiOLWTS/kCGg9IcJG+ACqn5+0+t8Oh83UU=
 github.com/karlseguin/ccache v2.0.3+incompatible/go.mod h1:CM9tNPzT6EdRh14+jiW8mEF9mkNZuuE51qmgGYUB93w=
 github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE=

internal/ethereum/config.go

@@ -19,13 +19,15 @@ 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 (
 	ConfigGasEstimationFactor   = "gasEstimationFactor"
 	ConfigDataFormat            = "dataFormat"
 	BlockPollingInterval        = "blockPollingInterval"
 	BlockCacheSize              = "blockCacheSize"
+	BlockTrackingMode           = "blockTrackingMode"
 	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(BlockTrackingMode, ffcapi.BlockListenerTrackingModeInMemoryPartialChain)
 	conf.AddKnownKey(ConfigDataFormat, "map")
 	conf.AddKnownKey(ConfigGasEstimationFactor, DefaultGasEstimationFactor)
 	conf.AddKnownKey(EventsBlockTimestamps, true)

internal/ethereum/ethereum.go

@@ -43,8 +43,10 @@ import (
 )
 
 type ethConnector struct {
-	backend                    rpcbackend.Backend
-	wsBackend                  rpcbackend.WebSocketRPCClient
+	backend           rpcbackend.Backend
+	wsBackend         rpcbackend.WebSocketRPCClient
+	blockTrackingMode ffcapi.BlockListenerTrackingMode
+
 	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) {
+
+	blockListenerTrackingMode := ffcapi.BlockListenerTrackingMode(conf.GetString(BlockTrackingMode))
+	if blockListenerTrackingMode == "" {
+		blockListenerTrackingMode = ffcapi.BlockListenerTrackingModeInMemoryPartialChain
+	}
+	if blockListenerTrackingMode != ffcapi.BlockListenerTrackingModeHeadBlockNumber && blockListenerTrackingMode != ffcapi.BlockListenerTrackingModeInMemoryPartialChain {
+		return nil, i18n.NewError(ctx, msgs.MsgInvalidBlockListenerTrackingMode, blockListenerTrackingMode)
+	}
+
 	c := &ethConnector{
 		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),
+		blockTrackingMode:          blockListenerTrackingMode,
 		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),
+		TrackingMode:                  blockListenerTrackingMode,
 	}, c.backend, c.wsBackend); err != nil {
 		return nil, err
 	}
@@ -216,6 +229,7 @@ func (c *ethConnector) ReconcileConfirmationsForTransaction(ctx context.Context,
 			Rebuilt:                 ethrpcRes.Rebuilt,
 			NewFork:                 ethrpcRes.NewFork,
 			Confirmed:               ethrpcRes.Confirmed,
+			ActualConfirmationCount: ethrpcRes.ActualConfirmationCount,
 			TargetConfirmationCount: ethrpcRes.TargetConfirmationCount,
 		}
 		if ethrpcReceipt != nil {

internal/ethereum/new_block_listener.go

@@ -23,6 +23,10 @@ import (
 	"github.com/hyperledger/firefly-transaction-manager/pkg/ffcapi"
 )
 
+func (c *ethConnector) GetBlockListenerTrackingMode(_ context.Context) ffcapi.BlockListenerTrackingMode {
+	return c.blockTrackingMode
+}
+
 func (c *ethConnector) NewBlockListener(_ context.Context, req *ffcapi.NewBlockListenerRequest) (*ffcapi.NewBlockListenerResponse, ffcapi.ErrorReason, error) {
 	// Add the block consumer
 	c.blockListener.AddConsumer(req.ListenerContext, &ethblocklistener.BlockUpdateConsumer{

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.blockTrackingMode", "The block tracking mode for the block listener", "`headBlockNumber` or `inMemoryPartialChain`")
 )

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")
+	MsgInvalidBlockListenerTrackingMode         = ffe("FF23069", "Invalid block listener tracking mode '%s': must be 'headBlockNumber' or 'inMemoryPartialChain'")
+	MsgTransactionNotIncludedInChainHead        = ffe("FF23070", "Transaction '%s' cannot be reconciled in head-only mode: chain head %d is before receipt block %s")
 )

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"`
+	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
+	ActualConfirmationCount uint64                     `json:"actualConfirmationCount"` // the actual 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"`
+	TrackingMode                  ffcapi.BlockListenerTrackingMode `json:"trackingMode,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.TrackingMode == "" {
+		conf.TrackingMode = ffcapi.BlockListenerTrackingModeInMemoryPartialChain
+	}
 	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.TrackingMode == ffcapi.BlockListenerTrackingModeHeadBlockNumber {
+			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()