blockwatch

README

blockwatch

Derived from 0x's blockwatch package with a few modifications to allow it to be integrated into the Livepeer node.

Documentation

Index

• Constants
• type Client
• type Config
• type Event
• type EventType
• type MiniHeader
• type MiniHeaderStore
• type RPCClient
  • func NewRPCClient(rpcURL string, requestTimeout time.Duration) (*RPCClient, error)
  • func (rc *RPCClient) FilterLogs(q ethereum.FilterQuery) ([]types.Log, error)
  • func (rc *RPCClient) HeaderByHash(hash common.Hash) (*MiniHeader, error)
  • func (rc *RPCClient) HeaderByNumber(number *big.Int) (*MiniHeader, error)
• type Stack
  • func NewStack(store MiniHeaderStore, limit int) *Stack
  • func (s *Stack) Inspect() ([]*MiniHeader, error)
  • func (s *Stack) Peek() (*MiniHeader, error)
  • func (s *Stack) Pop() (*MiniHeader, error)
  • func (s *Stack) Push(header *MiniHeader) error
• type Watcher
  • func New(config Config) *Watcher
  • func (w *Watcher) BackfillEvents(ctx context.Context, chainHead *MiniHeader) error
  • func (w *Watcher) GetLatestBlock() (*MiniHeader, error)
  • func (w *Watcher) InspectRetainedBlocks() ([]*MiniHeader, error)
  • func (w *Watcher) Subscribe(sink chan<- []*Event) event.Subscription
  • func (w *Watcher) Watch(ctx context.Context) error

Constants

const (
	FakeHash          = "0x6bbf9b6e836207ab25379c20e517a89090cbbaf8877746f6ed7fb6820770816b"
	FakeBlockNumber   = 30
	FakeL1BlockNumber = 50
)

Types

type Client

type Client interface {
	HeaderByNumber(number *big.Int) (*MiniHeader, error)
	HeaderByHash(hash common.Hash) (*MiniHeader, error)
	FilterLogs(q ethereum.FilterQuery) ([]types.Log, error)
}

Client defines the methods needed to satisfy the client expected when instantiating a Watcher instance.

type Config

type Config struct {
	Store               MiniHeaderStore
	PollingInterval     time.Duration
	StartBlockDepth     rpc.BlockNumber
	BlockRetentionLimit int
	WithLogs            bool
	Topics              []common.Hash
	Client              Client
}

Config holds some configuration options for an instance of BlockWatcher.

type Event

type Event struct {
	Type        EventType
	BlockHeader *MiniHeader
}

Event describes a block event emitted by a Watcher

type EventType

type EventType int

EventType describes the types of events emitted by blockwatch.Watcher. A block can be discovered and added to our representation of the chain. During a block re-org, a block previously stored can be removed from the list.

const (
	// Added indicates that an event was added to our representation of the chain
	Added EventType = iota
	// Removed indicates that an event was removed from our representation of the chain
	Removed
)

type MiniHeader

type MiniHeader struct {
	Hash          ethcommon.Hash
	Parent        ethcommon.Hash
	Number        *big.Int
	L1BlockNumber *big.Int
	Logs          []types.Log
}

MiniHeader is a succinct representation of an Ethereum block header

type MiniHeaderStore

type MiniHeaderStore interface {
	FindLatestMiniHeader() (*MiniHeader, error)
	FindAllMiniHeadersSortedByNumber() ([]*MiniHeader, error)
	InsertMiniHeader(header *MiniHeader) error
	DeleteMiniHeader(hash ethcommon.Hash) error
}

MiniHeaderStore is an interface for a store that manages the state of a MiniHeader collection

type RPCClient

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

RPCClient is a Client for fetching Ethereum blocks from a specific JSON-RPC endpoint.

func NewRPCClient

func NewRPCClient(rpcURL string, requestTimeout time.Duration) (*RPCClient, error)

NewRPCClient returns a new Client for fetching Ethereum blocks using the given ethclient.Client.

func (*RPCClient) FilterLogs

func (rc *RPCClient) FilterLogs(q ethereum.FilterQuery) ([]types.Log, error)

FilterLogs returns the logs that satisfy the supplied filter query.

func (*RPCClient) HeaderByHash

func (rc *RPCClient) HeaderByHash(hash common.Hash) (*MiniHeader, error)

HeaderByHash fetches a block header by its block hash. If no block exists with this number it will return a `ethereum.NotFound` error.

func (*RPCClient) HeaderByNumber

func (rc *RPCClient) HeaderByNumber(number *big.Int) (*MiniHeader, error)

HeaderByNumber fetches a block header by its number. If no `number` is supplied, it will return the latest block header. If no block exists with this number it will return a `ethereum.NotFound` error.

type Stack

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

Stack allows performing basic stack operations on a stack of MiniHeaders.

func NewStack

func NewStack(store MiniHeaderStore, limit int) *Stack

NewStack instantiates a new stack with the specified size limit. Once the size limit is reached, adding additional blocks will evict the deepest block.

func (*Stack) Inspect

func (s *Stack) Inspect() ([]*MiniHeader, error)

Inspect returns all the block headers currently on the stack

func (*Stack) Peek

func (s *Stack) Peek() (*MiniHeader, error)

Peek returns the latest block header from the block stack without removing it. It returns nil if the stack is empty.

func (*Stack) Pop

func (s *Stack) Pop() (*MiniHeader, error)

Pop removes and returns the latest block header on the block stack. It returns nil if the stack is empty.

func (*Stack) Push

func (s *Stack) Push(header *MiniHeader) error

Push pushes a block header onto the block stack. If the stack limit is reached, it will remove the oldest block header.

type Watcher

type Watcher struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Watcher maintains a consistent representation of the latest `blockRetentionLimit` blocks, handling block re-orgs and network disruptions gracefully. It can be started from any arbitrary block height, and will emit both block added and removed events.

func New

func New(config Config) *Watcher

New creates a new Watcher instance.

func (*Watcher) BackfillEvents

func (w *Watcher) BackfillEvents(ctx context.Context, chainHead *MiniHeader) error

BackfillEvents finds missed events and sends them to event subscribers. It blocks until it is done backfilling or the given context is canceled. Note that the latest block is never backfilled here from logs. It will be polled separately in syncToLatestBlock(). The reason for that is that we always need to propagate events from the latest block even if it does not contain events which are filtered out during the backfilling process.

func (*Watcher) GetLatestBlock

func (w *Watcher) GetLatestBlock() (*MiniHeader, error)

GetLatestBlock returns the latest block processed

func (*Watcher) InspectRetainedBlocks

func (w *Watcher) InspectRetainedBlocks() ([]*MiniHeader, error)

InspectRetainedBlocks returns the blocks retained in-memory by the Watcher instance. It is not particularly performant and therefore should only be used for debugging and testing purposes.

func (*Watcher) Subscribe

func (w *Watcher) Subscribe(sink chan<- []*Event) event.Subscription

Subscribe allows one to subscribe to the block events emitted by the Watcher. To unsubscribe, simply call `Unsubscribe` on the returned subscription. The sink channel should have ample buffer space to avoid blocking other subscribers. Slow subscribers are not dropped.

func (*Watcher) Watch

func (w *Watcher) Watch(ctx context.Context) error

Watch starts the Watcher. It will continuously look for new blocks and blocks until the given context is canceled. Typically, you want to call Watch inside a goroutine.