mon2

README

New onchain events monitor

• Reduce rpc calls by query logs on address without sepcifying topics
• Simpler APIs and config

Usage

• only need to create a single monitor object for each chain

mon, err := NewMonitor(ec, dal, perChainCfg)

• for each contract address, call MonAddr

mon.MonAddr(perAddrCfg, cbfn)

• CAUTION: MonAddr is blocking, use go mon.MonAddr(...) for async

Interfaces

• EthClient, support onchain rpc queries

type EthClient interface {
	// so we don't need chainid in func arg
	ChainID(ctx context.Context) (*big.Int, error)
	// get latest block number, available since geth 1.9.22
	BlockNumber(ctx context.Context) (uint64, error)
	// get logs, q will not set topics to get all events from the address
	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
}

• DAL, to persist log block and index to resume after restart. key is chainid-addr. addr has no 0x prefix and are all lower case. eg. 5-16649fc66e802151ad4fb37a5925c5e588b413e9. No longer support per addr + event as the block number and index are tracked at contract level, not event level.

type DAL interface {
	GetMonitorBlock(key string) (uint64, int64, bool, error)
	SetMonitorBlock(key string, blockNum uint64, blockIdx int64) error
}

• EventCallback, handle event, first arg is event name if abi is provided in PerAddrCfg. all events from the same address will be delivered in sequence. It's the callback func's responsibility to switch on event name and handle differently.

 func(string, types.Log)

Configs

• PerChainCfg, chain specific params for interval to update head block number and block delay etc to set proper from and to block in query filter.

type PerChainCfg struct {
	BlkIntv time.Duration // interval to call BlockNumber
	// below are params affecting from/to block in query filter
	BlkDelay, MaxBlkDelta, ForwardBlkDelay uint64
}

• PerAddrCfg, for each contract, mainly address, abi string and polling interval

type PerAddrCfg struct {
	Addr    common.Address // contract addr
	ChkIntv time.Duration  // interval to call FilterLogs
	AbiStr  string         // XxxABI or XxxMetaData.ABI abi string from this contract's go binding, needed to match log topic to event name, if empty string, evname in callback is also empty
	FromBlk uint64         // optional. if > 0, means ignore persisted blocknum and use this for FromBlk in queries, don't set unless you know what you're doing
}

Duplicated events

If program terminates/crashes while calling callback functions to handle event, there is no way for monitor code to know if the event has been processed. So to be on the safer side, on restart we'll fetch logs from last persisted blocknum and skip logs before persisted log index. This means callback may receive same event again. Application should have its own de-dup logic to ignore already processed events.

Documentation

Index

• func EventIDMap(abistr string) map[common.Hash]string
• type DAL
• type EthClient
• type EventCallback
• type LogEventID
  • func (le *LogEventID) CountSkip(logs []types.Log) (skipped int)
  • func (le *LogEventID) ShouldSkip(blknum uint64, idx uint) bool
• type Monitor
  • func NewMonitor(ec EthClient, dal DAL, cfg PerChainCfg) (*Monitor, error)
  • func (m *Monitor) CalcNextFromBlkNum(lastFrom, lastTo uint64) uint64
  • func (m *Monitor) CalcToBlkNum(from uint64) uint64
  • func (m *Monitor) Close()
  • func (m *Monitor) GetBlkNum() uint64
  • func (m *Monitor) MonAddr(cfg PerAddrCfg, cbfn EventCallback)
  • func (m *Monitor) StopMon()
• type PerAddrCfg
• type PerChainCfg

Functions

func EventIDMap

func EventIDMap(abistr string) map[common.Hash]string

parse abi and return map from event.ID to its name eg. Deposited if abistr is empty string, return empty map (not nil)

Types

type DAL

type DAL interface {
	GetMonitorBlock(key string) (uint64, int64, bool, error)
	SetMonitorBlock(key string, blockNum uint64, blockIdx int64) error
}

DAL is an interface to persist block info to resume note we no longer have event granularity instead it'll be contract level only. key will be prefixed with chainid. sprintf("%d-%x", chainid, addr)

type EthClient

type EthClient interface {
	// so we don't need chainid in func arg
	ChainID(ctx context.Context) (*big.Int, error)
	// new get latest block number, available since geth 1.9.22
	BlockNumber(ctx context.Context) (uint64, error)
	// get logs, q should not set topics to get all events from address
	FilterLogs(ctx context.Context, q ethereum.FilterQuery) ([]types.Log, error)
}

interfaces

type EventCallback

type EventCallback func(string, types.Log)

signature for event callback func, first arg is event name matched by abi, could be empty string if no match

type LogEventID

type LogEventID struct {
	BlkNum uint64 // Number of the block containing the event
	Index  int64  // Index of the event within the block, use int to support -1 in fast forward case
}

LogEventID tracks the position of an event by blocknum and its index in the block

func (*LogEventID) CountSkip

func (le *LogEventID) CountSkip(logs []types.Log) (skipped int)

go over logs and return how many should be skipped

func (*LogEventID) ShouldSkip

func (le *LogEventID) ShouldSkip(blknum uint64, idx uint) bool

compare w/ saved blknum/idx to decide if we already handled this log so skip

type Monitor

type Monitor struct {
	// contains filtered or unexported fields
}

func NewMonitor

func NewMonitor(ec EthClient, dal DAL, cfg PerChainCfg) (*Monitor, error)

cfg is not pointer to ensure value copy and avoid unexpected change by other code

func (*Monitor) CalcNextFromBlkNum

func (m *Monitor) CalcNextFromBlkNum(lastFrom, lastTo uint64) uint64

return next from blknum based on last To blknum and didn't see any event guarantee return value >= lastFrom

func (*Monitor) CalcToBlkNum

func (m *Monitor) CalcToBlkNum(from uint64) uint64

given from block, return toblocknum for query consider BlkDelay and MaxBlkDelta caller MUST check return value and only do query if returned >= from so from==to is allowed to query single block for logs

func (*Monitor) Close

func (m *Monitor) Close()

Stops all MonAddr and updating blk num to release all resources. No longer usable after this call

func (*Monitor) GetBlkNum

func (m *Monitor) GetBlkNum() uint64

return cached block number from last BlockNumber call, no rpc, return in memory value directly

func (*Monitor) MonAddr

func (m *Monitor) MonAddr(cfg PerAddrCfg, cbfn EventCallback)

start monitoring events from addr with configured interval, calls cbfn for each recived log BLOCKING until m is Closed

func (*Monitor) StopMon

func (m *Monitor) StopMon()

StopMon will quit all MonAddr loops, but keep updating blk num.

type PerAddrCfg

type PerAddrCfg struct {
	Addr    common.Address  // contract addr
	ChkIntv time.Duration   // interval to call FilterLogs
	AbiStr  string          // XxxABI or XxxMetaData.ABI abi string from this contract's go binding, needed to match log topic to event name, if empty string, evname in callback is also empty
	FromBlk uint64          // optional. if > 0, means ignore persisted blocknum and use this for FromBlk in queries, don't set unless you know what you're doing
	Topics  [][]common.Hash // optional. topic filters. position sensitive. keccak256 hashed values. for usage, see go-ethereum's FilterQuery https://pkg.go.dev/github.com/ethereum/go-ethereum#FilterQuery
}

mon config for each contract

type PerChainCfg

type PerChainCfg struct {
	BlkIntv time.Duration // interval to call BlockNumber
	// below are params affecting from/to block in query filter
	BlkDelay, MaxBlkDelta, ForwardBlkDelay uint64
}

per chain config to affect how often call BlockNumber and from/to block in FilterQuery