database

package
v0.1.118-dev Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jul 22, 2026 License: AGPL-3.0 Imports: 19 Imported by: 0

Documentation

Index

Constants

View Source
const (
	// Kinds are the datastore equivalent of tables.
	BlocksKind            = "blocks"
	BlockEventsKind       = "block_events"
	TransactionsKind      = "transactions"
	TransactionEventsKind = "transaction_events"
	PeersKind             = "peers"
	MaxAttempts           = 5
)

Variables

This section is empty.

Functions

This section is empty.

Types

type ClickHouse

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

ClickHouse implements the Database interface backed by a ClickHouse cluster. The table definitions this writer targets (and the block_first_seen materialized view) live in the sensor-network-tools repo (clickhouse_schema.sql), not this repo.

func (*ClickHouse) Close

func (c *ClickHouse) Close() error

Close stops the batcher goroutines, waits for their final drain flush to complete, and closes the connection. It is safe to call on a no-op instance (connection never established).

func (*ClickHouse) HasBlock

func (c *ClickHouse) HasBlock(ctx context.Context, hash common.Hash) bool

HasBlock reports whether the block already exists. Called once per new block (not per event), so an indexed point lookup is cheap. Without a connection it reports true so the sensor never attempts a backfill it could not persist.

func (*ClickHouse) MaxConcurrentWrites

func (c *ClickHouse) MaxConcurrentWrites() int

func (*ClickHouse) NodeList

func (c *ClickHouse) NodeList(ctx context.Context, limit int) ([]string, error)

func (*ClickHouse) ShouldWriteBlockEvents

func (c *ClickHouse) ShouldWriteBlockEvents() bool

func (*ClickHouse) ShouldWriteBlocks

func (c *ClickHouse) ShouldWriteBlocks() bool

func (*ClickHouse) ShouldWriteFirstBlockEvent

func (c *ClickHouse) ShouldWriteFirstBlockEvent() bool

func (*ClickHouse) ShouldWriteFirstTransactionEvent

func (c *ClickHouse) ShouldWriteFirstTransactionEvent() bool

func (*ClickHouse) ShouldWritePeers

func (c *ClickHouse) ShouldWritePeers() bool

func (*ClickHouse) ShouldWriteTransactionEvents

func (c *ClickHouse) ShouldWriteTransactionEvents() bool

func (*ClickHouse) ShouldWriteTransactions

func (c *ClickHouse) ShouldWriteTransactions() bool

func (*ClickHouse) WriteBlock

func (c *ClickHouse) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time)

func (*ClickHouse) WriteBlockBody

func (c *ClickHouse) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time)

func (*ClickHouse) WriteBlockEvents

func (c *ClickHouse) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

func (*ClickHouse) WriteBlockHashFirstSeen

func (c *ClickHouse) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time)

WriteBlockHashFirstSeen is a no-op: ClickHouse derives earliest first-seen at query time from the block_events stream (see the block_first_seen materialized view), so no per-block stamp is needed.

func (*ClickHouse) WriteBlockHeaders

func (c *ClickHouse) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool)

func (*ClickHouse) WritePeers

func (c *ClickHouse) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time)

func (*ClickHouse) WriteTransactionEvents

func (c *ClickHouse) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

func (*ClickHouse) WriteTransactions

func (c *ClickHouse) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time)

type ClickHouseOptions

type ClickHouseOptions struct {
	DSN                              string
	SensorID                         string
	ChainID                          uint64
	MaxConcurrency                   int
	ShouldWriteBlocks                bool
	ShouldWriteBlockEvents           bool
	ShouldWriteFirstBlockEvent       bool
	ShouldWriteTransactions          bool
	ShouldWriteTransactionEvents     bool
	ShouldWriteFirstTransactionEvent bool
	ShouldWritePeers                 bool
}

ClickHouseOptions is used when creating a NewClickHouse.

type Database

type Database interface {
	// WriteBlock will write the both the block and block event to the database
	// if ShouldWriteBlocks and ShouldWriteBlockEvents return true, respectively.
	WriteBlock(context.Context, *enode.Node, *types.Block, *big.Int, time.Time)

	// WriteBlockHeaders will write the block headers if ShouldWriteBlocks
	// returns true.
	WriteBlockHeaders(context.Context, []*types.Header, time.Time, bool)

	// WriteBlockEvents appends an inbound block event (peer, hash, time) for
	// each hash — one per peer we received the announcement from. The caller
	// decides which hashes to pass (every announcement for the full per-peer
	// stream, or just the first-seen ones); the backend only appends.
	WriteBlockEvents(context.Context, *enode.Node, []common.Hash, time.Time)

	// WriteBlockHashFirstSeen records the earliest sighting of a block hash on
	// the block entity itself (Datastore's TimeFirstSeenHash). Backends that
	// derive first-seen from the event stream (e.g. ClickHouse) treat it as a
	// no-op.
	WriteBlockHashFirstSeen(context.Context, *enode.Node, common.Hash, time.Time)

	// WriteBlockBody writes the transactions carried in the block body (the block
	// row itself comes from WriteBlock/WriteBlockHeaders). Backends with a
	// separate transactions table (e.g. ClickHouse) gate this on
	// ShouldWriteTransactions; the Datastore backend gates on ShouldWriteBlocks
	// because it links the transactions and uncles onto the block entity.
	WriteBlockBody(context.Context, *eth.BlockBody, common.Hash, time.Time)

	// WriteTransactions writes the transaction bodies if ShouldWriteTransactions
	// returns true. Transaction events are recorded separately via
	// WriteTransactionEvents at announcement time.
	WriteTransactions(context.Context, *enode.Node, []*types.Transaction, time.Time)

	// WriteTransactionEvents appends an inbound transaction event (peer, hash,
	// time) for each hash — the tx mirror of WriteBlockEvents. The caller
	// decides which hashes to pass (every announcement for the full per-peer
	// stream, or just the first-seen ones); the backend only appends.
	WriteTransactionEvents(context.Context, *enode.Node, []common.Hash, time.Time)

	// WritePeers will write the connected peers to the database.
	WritePeers(context.Context, []*p2p.Peer, time.Time)

	// HasBlock will return whether the block is in the database. If the database
	// client has not been initialized this will always return true.
	HasBlock(context.Context, common.Hash) bool

	MaxConcurrentWrites() int
	ShouldWriteBlocks() bool
	ShouldWriteBlockEvents() bool
	ShouldWriteFirstBlockEvent() bool
	ShouldWriteTransactions() bool
	ShouldWriteTransactionEvents() bool
	ShouldWriteFirstTransactionEvent() bool
	ShouldWritePeers() bool

	// NodeList will return a list of enode URLs.
	NodeList(ctx context.Context, limit int) ([]string, error)

	// Close flushes any buffered writes and releases the underlying database
	// connection. It blocks until in-flight writes have drained (or their
	// shutdown flush times out). Implementations with no resources to release
	// return nil.
	Close() error
}

Database represents a database solution to write block and transaction data to. To use another database solution, just implement these methods and update the sensor to use the new connection.

func NewClickHouse

func NewClickHouse(ctx context.Context, opts ClickHouseOptions) Database

NewClickHouse connects to ClickHouse, verifies connectivity, and starts the background batch flushers. Callers should defer Close to drain buffered rows on shutdown. If the connection cannot be established the returned Database no-ops all writes (mirroring the Datastore backend) so the sensor keeps running.

func NewDatastore

func NewDatastore(ctx context.Context, opts DatastoreOptions) Database

NewDatastore connects to datastore and creates the client. This should only be called once unless trying to write to different databases.

func NewJSONDatabase added in v0.1.88

func NewJSONDatabase(opts JSONDatabaseOptions) Database

NewJSONDatabase creates a new JSONDatabase instance.

func NoDatabase added in v0.1.88

func NoDatabase() Database

NoDatabase creates a new nodb instance.

type Datastore

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

Datastore wraps the datastore client, stores the sensorID, and other information needed when writing blocks and transactions.

func (*Datastore) Close

func (d *Datastore) Close() error

Close waits for in-flight async writes to finish, then releases the underlying Datastore client. Acquiring every slot in the jobs semaphore guarantees no write is in progress, since datastore.Client.Close must not be called concurrently with pending operations.

func (*Datastore) HasBlock

func (d *Datastore) HasBlock(ctx context.Context, hash common.Hash) bool

func (*Datastore) MaxConcurrentWrites

func (d *Datastore) MaxConcurrentWrites() int

func (*Datastore) NodeList

func (d *Datastore) NodeList(ctx context.Context, limit int) ([]string, error)

func (*Datastore) ShouldWriteBlockEvents

func (d *Datastore) ShouldWriteBlockEvents() bool

func (*Datastore) ShouldWriteBlocks

func (d *Datastore) ShouldWriteBlocks() bool

func (*Datastore) ShouldWriteFirstBlockEvent added in v0.1.108

func (d *Datastore) ShouldWriteFirstBlockEvent() bool

func (*Datastore) ShouldWriteFirstTransactionEvent added in v0.1.112

func (d *Datastore) ShouldWriteFirstTransactionEvent() bool

func (*Datastore) ShouldWritePeers

func (d *Datastore) ShouldWritePeers() bool

func (*Datastore) ShouldWriteTransactionEvents

func (d *Datastore) ShouldWriteTransactionEvents() bool

func (*Datastore) ShouldWriteTransactions

func (d *Datastore) ShouldWriteTransactions() bool

func (*Datastore) WriteBlock

func (d *Datastore) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time)

WriteBlock writes the block and the block event to datastore.

func (*Datastore) WriteBlockBody

func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time)

WriteBlockBody will write the block bodies to datastore. It will not write block events because bodies will only be sent to the sensor when requested. The block events will be written when the hash is received instead. It will write the uncles and transactions to datastore if they don't already exist.

func (*Datastore) WriteBlockEvents

func (d *Datastore) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

WriteBlockEvents appends an inbound block event per hash for the given peer.

func (*Datastore) WriteBlockHashFirstSeen added in v0.1.97

func (d *Datastore) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time)

WriteBlockHashFirstSeen stamps the block entity's earliest hash-first-seen time. The block event itself is written separately via WriteBlockEvents.

func (*Datastore) WriteBlockHeaders

func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool)

WriteBlockHeaders will write the block headers to datastore. It will not write block events because headers will only be sent to the sensor when requested. The block events will be written when the hash is received instead. The isParent parameter indicates if these headers were fetched as parent blocks.

func (*Datastore) WritePeers

func (d *Datastore) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time)

WritePeers writes the connected peers to datastore.

func (*Datastore) WriteTransactionEvents

func (d *Datastore) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

WriteTransactionEvents appends an inbound transaction event per hash for the given peer.

func (*Datastore) WriteTransactions

func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time)

WriteTransactions writes the transaction bodies to datastore. Transaction events are recorded separately via WriteTransactionEvents.

type DatastoreBlock

type DatastoreBlock struct {
	*DatastoreHeader
	TotalDifficulty     string           `datastore:",noindex"`
	Transactions        []*datastore.Key `datastore:",noindex"`
	Uncles              []*datastore.Key `datastore:",noindex"`
	TimeFirstSeenHash   time.Time
	SensorFirstSeenHash string
}

DatastoreBlock represents a block stored in datastore.

type DatastoreEvent

type DatastoreEvent struct {
	SensorId string
	PeerId   string
	Hash     *datastore.Key
	Time     time.Time
	TTL      time.Time
}

DatastoreEvent can represent a peer sending the sensor a transaction hash or a block hash. In this implementation, the block and transactions are written to different tables by specifying a kind during key creation see writeEvents for more.

type DatastoreHeader

type DatastoreHeader struct {
	ParentHash      *datastore.Key
	UncleHash       string `datastore:",noindex"`
	Coinbase        string `datastore:",noindex"`
	Root            string `datastore:",noindex"`
	TxHash          string `datastore:",noindex"`
	ReceiptHash     string `datastore:",noindex"`
	Bloom           []byte `datastore:",noindex"`
	Difficulty      string `datastore:",noindex"`
	Number          string
	GasLimit        string `datastore:",noindex"`
	GasUsed         string
	Time            time.Time
	Extra           []byte `datastore:",noindex"`
	MixDigest       string `datastore:",noindex"`
	Nonce           string `datastore:",noindex"`
	BaseFee         string `datastore:",noindex"`
	TimeFirstSeen   time.Time
	TTL             time.Time
	IsParent        bool
	SensorFirstSeen string
}

DatastoreHeader stores the data in manner that can be easily written without loss of precision.

type DatastoreOptions

type DatastoreOptions struct {
	ProjectID                        string
	DatabaseID                       string
	SensorID                         string
	ChainID                          uint64
	MaxConcurrency                   int
	ShouldWriteBlocks                bool
	ShouldWriteBlockEvents           bool
	ShouldWriteFirstBlockEvent       bool
	ShouldWriteTransactions          bool
	ShouldWriteTransactionEvents     bool
	ShouldWriteFirstTransactionEvent bool
	ShouldWritePeers                 bool
	TTL                              time.Duration
}

DatastoreOptions is used when creating a NewDatastore.

type DatastorePeer

type DatastorePeer struct {
	Name         string
	Caps         []string `datastore:",noindex"`
	URL          string
	LastSeenBy   string
	TimeLastSeen time.Time
	TTL          time.Time
}

type DatastoreTransaction

type DatastoreTransaction struct {
	Data            []byte `datastore:",noindex"`
	From            string
	Gas             string `datastore:",noindex"`
	GasFeeCap       string `datastore:",noindex"`
	GasPrice        string `datastore:",noindex"`
	GasTipCap       string `datastore:",noindex"`
	Nonce           string `datastore:",noindex"`
	To              string
	Value           string `datastore:",noindex"`
	V, R, S         string `datastore:",noindex"`
	Time            time.Time
	TimeFirstSeen   time.Time
	TTL             time.Time
	Type            int16
	SensorFirstSeen string
}

DatastoreTransaction represents a transaction stored in datastore. Data is not indexed because there is a max sized for indexed byte slices, which Data will occasionally exceed.

type JSONBlock added in v0.1.88

type JSONBlock struct {
	Type            string    `json:"type"`
	SensorID        string    `json:"sensor_id"`
	Hash            string    `json:"hash"`
	ParentHash      string    `json:"parent_hash"`
	Number          uint64    `json:"number"`
	Timestamp       uint64    `json:"timestamp"`
	GasLimit        uint64    `json:"gas_limit"`
	GasUsed         uint64    `json:"gas_used"`
	Difficulty      string    `json:"difficulty,omitempty"`
	TotalDifficulty string    `json:"total_difficulty,omitempty"`
	BaseFee         string    `json:"base_fee,omitempty"`
	TxCount         int       `json:"tx_count"`
	UncleCount      int       `json:"uncle_count"`
	TimeFirstSeen   time.Time `json:"time_first_seen"`
	IsParent        bool      `json:"is_parent"`
}

JSONBlock represents a block in JSON format.

type JSONBlockEvent added in v0.1.88

type JSONBlockEvent struct {
	Type      string    `json:"type"`
	SensorID  string    `json:"sensor_id"`
	PeerID    string    `json:"peer_id"`
	Hash      string    `json:"hash"`
	Timestamp time.Time `json:"timestamp"`
}

JSONBlockEvent represents a block event in JSON format.

type JSONBlockHashFirstSeen added in v0.1.97

type JSONBlockHashFirstSeen struct {
	Type          string    `json:"type"`
	SensorID      string    `json:"sensor_id"`
	Hash          string    `json:"hash"`
	TimeFirstSeen time.Time `json:"time_first_seen"`
}

JSONBlockHashFirstSeen represents a block hash announcement in JSON format.

type JSONDatabase added in v0.1.88

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

JSONDatabase outputs data as JSON to stdout. Each record is output as a single line of JSON (newline-delimited JSON).

func (*JSONDatabase) Close

func (j *JSONDatabase) Close() error

Close does nothing; records are written to stdout unbuffered.

func (*JSONDatabase) HasBlock added in v0.1.88

func (j *JSONDatabase) HasBlock(ctx context.Context, hash common.Hash) bool

HasBlock always returns true to avoid unnecessary parent block fetching for JSON output.

func (*JSONDatabase) MaxConcurrentWrites added in v0.1.88

func (j *JSONDatabase) MaxConcurrentWrites() int

MaxConcurrentWrites returns the max concurrency.

func (*JSONDatabase) NodeList added in v0.1.88

func (j *JSONDatabase) NodeList(ctx context.Context, limit int) ([]string, error)

NodeList returns an empty list as JSON database doesn't store nodes.

func (*JSONDatabase) ShouldWriteBlockEvents added in v0.1.88

func (j *JSONDatabase) ShouldWriteBlockEvents() bool

ShouldWriteBlockEvents returns the configured value.

func (*JSONDatabase) ShouldWriteBlocks added in v0.1.88

func (j *JSONDatabase) ShouldWriteBlocks() bool

ShouldWriteBlocks returns the configured value.

func (*JSONDatabase) ShouldWriteFirstBlockEvent added in v0.1.108

func (j *JSONDatabase) ShouldWriteFirstBlockEvent() bool

ShouldWriteFirstBlockEvent returns false for JSON database.

func (*JSONDatabase) ShouldWriteFirstTransactionEvent added in v0.1.112

func (j *JSONDatabase) ShouldWriteFirstTransactionEvent() bool

ShouldWriteFirstTransactionEvent returns false for JSON database.

func (*JSONDatabase) ShouldWritePeers added in v0.1.88

func (j *JSONDatabase) ShouldWritePeers() bool

ShouldWritePeers returns the configured value.

func (*JSONDatabase) ShouldWriteTransactionEvents added in v0.1.88

func (j *JSONDatabase) ShouldWriteTransactionEvents() bool

ShouldWriteTransactionEvents returns the configured value.

func (*JSONDatabase) ShouldWriteTransactions added in v0.1.88

func (j *JSONDatabase) ShouldWriteTransactions() bool

ShouldWriteTransactions returns the configured value.

func (*JSONDatabase) Write added in v0.1.88

func (j *JSONDatabase) Write(v any)

Write safely outputs JSON to stdout.

func (*JSONDatabase) WriteBlock added in v0.1.88

func (j *JSONDatabase) WriteBlock(_ context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time)

WriteBlock writes the block and the block event as JSON.

func (*JSONDatabase) WriteBlockBody added in v0.1.88

func (j *JSONDatabase) WriteBlockBody(ctx context.Context, body *eth.BlockBody, hash common.Hash, tfs time.Time)

WriteBlockBody writes the block body as JSON.

func (*JSONDatabase) WriteBlockEvents

func (j *JSONDatabase) WriteBlockEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

WriteBlockEvents writes the block events as JSON.

func (*JSONDatabase) WriteBlockHashFirstSeen added in v0.1.97

func (j *JSONDatabase) WriteBlockHashFirstSeen(ctx context.Context, peer *enode.Node, hash common.Hash, tfsh time.Time)

WriteBlockHashFirstSeen writes a partial block entry with just the hash first seen time. For JSON output, this writes a separate record type.

func (*JSONDatabase) WriteBlockHeaders added in v0.1.88

func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool)

WriteBlockHeaders writes the block headers as JSON. The isParent parameter indicates if these headers were fetched as parent blocks.

func (*JSONDatabase) WritePeers added in v0.1.88

func (j *JSONDatabase) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.Time)

WritePeers writes the connected peers as JSON.

func (*JSONDatabase) WriteTransactionEvents

func (j *JSONDatabase) WriteTransactionEvents(ctx context.Context, peer *enode.Node, hashes []common.Hash, tfs time.Time)

func (*JSONDatabase) WriteTransactions added in v0.1.88

func (j *JSONDatabase) WriteTransactions(_ context.Context, peer *enode.Node, txs []*types.Transaction, tfs time.Time)

WriteTransactions writes the transactions and transaction events as JSON.

type JSONDatabaseOptions added in v0.1.88

type JSONDatabaseOptions struct {
	SensorID                     string
	ChainID                      uint64
	MaxConcurrency               int
	ShouldWriteBlocks            bool
	ShouldWriteBlockEvents       bool
	ShouldWriteTransactions      bool
	ShouldWriteTransactionEvents bool
	ShouldWritePeers             bool
}

JSONDatabaseOptions is used when creating a NewJSONDatabase.

type JSONPeer added in v0.1.88

type JSONPeer struct {
	Type         string    `json:"type"`
	SensorID     string    `json:"sensor_id"`
	ID           string    `json:"id"`
	Name         string    `json:"name"`
	URL          string    `json:"url"`
	Caps         []string  `json:"caps"`
	TimeLastSeen time.Time `json:"time_last_seen"`
}

JSONPeer represents a peer in JSON format.

type JSONTransaction added in v0.1.88

type JSONTransaction struct {
	Type          string    `json:"type"`
	SensorID      string    `json:"sensor_id"`
	Hash          string    `json:"hash"`
	From          string    `json:"from,omitempty"`
	To            string    `json:"to,omitempty"`
	Value         string    `json:"value"`
	Gas           uint64    `json:"gas"`
	GasPrice      string    `json:"gas_price"`
	GasFeeCap     string    `json:"gas_fee_cap,omitempty"`
	GasTipCap     string    `json:"gas_tip_cap,omitempty"`
	Nonce         uint64    `json:"nonce"`
	TxType        uint8     `json:"tx_type"`
	TimeFirstSeen time.Time `json:"time_first_seen"`
}

JSONTransaction represents a transaction in JSON format.

type JSONTransactionEvent added in v0.1.88

type JSONTransactionEvent struct {
	Type      string    `json:"type"`
	SensorID  string    `json:"sensor_id"`
	PeerID    string    `json:"peer_id"`
	Hash      string    `json:"hash"`
	Timestamp time.Time `json:"timestamp"`
}

JSONTransactionEvent represents a transaction event in JSON format.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL