monitor

package
v1.19.2 Latest Latest
Warning

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

 Go to latest Published: Jun 26, 2026 License: MIT Imports: 31 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var ErrAlreadyStopped = errors.New("already stopped")
View Source 
var ErrBlockNotContiguous = errors.New("blocks are not contiguous")
View Source 
var ErrBlockNotFound = errors.New("block not found")
View Source 
var ErrLogNotFound = errors.New("log not found")
View Source 
var ErrNotExecutingMessage = errors.New("not an executing message")

Functions

func Main 

func Main(version string) cliapp.LifecycleAction

Types

type BlockBuffer 

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

BlockBuffer is a circular buffer of seen blocks. It can be used as a fix-sized stack of blocks to ensure a canonical and contiguous view of the block history.

func NewBlockBuffer 

func NewBlockBuffer(size int) *BlockBuffer

NewBlockBuffer creates a new block buffer

func (*BlockBuffer) Add 

func (r *BlockBuffer) Add(block eth.BlockInfo)

Add adds a block to the buffer

func (*BlockBuffer) Peek 

func (r *BlockBuffer) Peek() eth.BlockInfo

Peek returns the last added block to the buffer if the buffer is empty, it returns nil if the buffer is not empty, it returns the last added block

func (*BlockBuffer) Pop 

func (r *BlockBuffer) Pop() (eth.BlockInfo, error)

func (*BlockBuffer) Reset 

func (r *BlockBuffer) Reset()

Reset resets the buffer to empty

type CLIConfig 

type CLIConfig struct {
	L2Rpcs            []string
	DependencySetPath string
	PollInterval      time.Duration

	// InteropFilterEndpoint, when set, enables a read-only cross-check against the op-interop-filter.
	InteropFilterEndpoint  string
	InteropFilterMinSafety string

	// SupernodeEndpoints, when set, enable read-only observation of op-supernode liveness,
	// per-chain heads, and cross-safety violations.
	SupernodeEndpoints []string

	RPCConfig     oprpc.CLIConfig
	LogConfig     oplog.CLIConfig
	MetricsConfig opmetrics.CLIConfig
	PprofConfig   oppprof.CLIConfig
}

func NewConfig 

func NewConfig(ctx *cli.Context) *CLIConfig

func (*CLIConfig) Check 

func (c *CLIConfig) Check() error

type CanonicalBlockSource added in v1.19.1 

type CanonicalBlockSource interface {
	InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error)
}

CanonicalBlockSource is the minimal execution-layer surface used to confirm a job's executing block is still canonical on its chain. Satisfied by sources.EthClient.

type FilterChecker added in v1.19.1 

type FilterChecker interface {
	// CheckMessage replays a single executing message as an access list to the filter.
	// A nil error means the filter considers the message valid at minSafety.
	CheckMessage(ctx context.Context, msg messages.Message, executingChain eth.ChainID, executingTimestamp uint64) error
	GetFailsafeEnabled(ctx context.Context) (bool, error)
	Close()
}

FilterChecker is the read-only interop-filter surface the observer needs.

type FilterClient added in v1.19.1 

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

FilterClient calls the op-interop-filter public RPC (read-only). It delegates the access-list check to the canonical sources.InteropFilterClient so the monitor stays in sync with the filter's RPC signature.

func NewFilterClient added in v1.19.1 

func NewFilterClient(endpoint string, minSafety safety.Level, log log.Logger) (*FilterClient, error)

func (*FilterClient) CheckMessage added in v1.19.1 

func (fc *FilterClient) CheckMessage(ctx context.Context, msg messages.Message, executingChain eth.ChainID, executingTimestamp uint64) error

CheckMessage builds the access-list for one executing message and delegates to the canonical interop-filter client's CheckAccessList. A nil error means the filter considers the message valid at minSafety; a non-nil error is the filter's rejection (or a transport error).

func (*FilterClient) Close added in v1.19.1 

func (fc *FilterClient) Close()

func (*FilterClient) GetFailsafeEnabled added in v1.19.1 

func (fc *FilterClient) GetFailsafeEnabled(ctx context.Context) (bool, error)

GetFailsafeEnabled reads the filter's failsafe state via admin_getFailsafeEnabled. This is an admin-namespace method, distinct from the interop_* query API the canonical filter client wraps, so it is called directly here.

type FilterObserver added in v1.19.1 

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

FilterObserver cross-checks the monitor's independent verdict against the op-interop-filter (read-only). It never gates monitor behaviour; it only emits divergence and failsafe metrics for observability.

func NewFilterObserver added in v1.19.1 

func NewFilterObserver(filter FilterChecker, m InteropMessageMetrics, log log.Logger) *FilterObserver

func (*FilterObserver) Observe added in v1.19.1 

func (o *FilterObserver) Observe(ctx context.Context, jobs map[JobID]*Job)

Observe replays each terminal job's executing message to the filter and records divergences.

func (*FilterObserver) PollFailsafe added in v1.19.1 

func (o *FilterObserver) PollFailsafe(ctx context.Context)

PollFailsafe reads the filter's failsafe state and records it as a gauge.

type FinalityCallback 

type FinalityCallback func(chainID eth.ChainID, block eth.BlockInfo)

FinalityCallback is a function to be called when the finality of jobs for this chain is updated

type Finder 

type Finder interface {
	Start(ctx context.Context) error
	Stop() error
}

Finders are responsible for finding new jobs from a chain for an Updater to track

type FinderClient 

type FinderClient interface {
	InfoByLabel(ctx context.Context, label eth.BlockLabel) (eth.BlockInfo, error)
	InfoByNumber(ctx context.Context, number uint64) (eth.BlockInfo, error)
	FetchReceiptsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Receipts, error)
}

FinderClient is a client that can be used to find new blocks and their receipts it is satisfied by the ethclient.Client type

type InteropMessageMetrics 

type InteropMessageMetrics interface {
	RecordMessageStatus(executingChainID string, initiatingChainID string, status string, count float64)
	RecordTerminalStatusChange(executingChainID string, initiatingChainID string, count float64)
	RecordExecutingBlockRange(chainID string, min uint64, max uint64)
	RecordInitiatingBlockRange(chainID string, min uint64, max uint64)
	RecordInitiatingReorg(executingChainID string, initiatingChainID string)
	RecordFilterDivergence(executingChainID string, initiatingChainID string, monitorStatus string, filterStatus string)
	RecordFilterFailsafe(enabled bool)
	RecordSupernodeUp(endpoint string, up bool)
	RecordSupernodeSafeHead(chainID string, level string, blockNumber uint64)
	RecordCrossSafetyViolation(executingChainID string, initiatingChainID string)
}

type InteropMonitorConfig 

type InteropMonitorConfig struct {
	PollInterval time.Duration
}

type InteropMonitorService 

type InteropMonitorService struct {
	Log     log.Logger
	Metrics metrics.Metricer

	InteropMonitorConfig

	Version string
	// contains filtered or unexported fields
}

func InteropMonitorServiceFromCLIConfig 

func InteropMonitorServiceFromCLIConfig(ctx context.Context, version string, cfg *CLIConfig, log log.Logger) (*InteropMonitorService, error)

func InteropMonitorServiceFromClients 

func InteropMonitorServiceFromClients(
	ctx context.Context,
	version string,
	cfg *CLIConfig,
	clients map[eth.ChainID]*sources.EthClient,
	log log.Logger,
) (*InteropMonitorService, error)

InteropMonitorServiceFromClients creates a new InteropMonitorService with pre-initialized clients

func (*InteropMonitorService) Kill 

func (ms *InteropMonitorService) Kill() error

func (*InteropMonitorService) RouteNewJob 

func (ms *InteropMonitorService) RouteNewJob(job *Job)

RouteNewJob routes a new job to the appropriate updater by simply enqueuing to the initiating chain's updater

func (*InteropMonitorService) SetExpiry 

func (ms *InteropMonitorService) SetExpiry(chainID eth.ChainID, expiry eth.BlockInfo)

SetExpiry sets the expiry for a chain ID

func (*InteropMonitorService) Start 

func (ms *InteropMonitorService) Start(ctx context.Context) error

func (*InteropMonitorService) Stop 

func (ms *InteropMonitorService) Stop(ctx context.Context) error

func (*InteropMonitorService) Stopped 

func (ms *InteropMonitorService) Stopped() bool

type Job 

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

Job is a job that is being tracked by the monitor it represents an executing message and initiating message pair it is used to track the status of the executing message over time along with pertinent metadata about the initiating message its getters and setters are thread safe

func BlockReceiptsToJobs 

func BlockReceiptsToJobs(receipts []*types.Receipt, executingChain eth.ChainID) []*Job

BlockReceiptsToJobs converts a slice of receipts to a slice of jobs

func JobFromExecutingMessageLog 

func JobFromExecutingMessageLog(log *types.Log, executingChain eth.ChainID) (Job, error)

JobFromExecutingMessageLog converts a log to a job

func (*Job) AddInitiatingHash 

func (j *Job) AddInitiatingHash(hash common.Hash)

AddInitiatingHash adds a hash to the initiatingHash slice if it hasn't been seen before

func (*Job) CountReorgOnce added in v1.19.1 

func (j *Job) CountReorgOnce() bool

CountReorgOnce reports true exactly once, the first time it is called, so the initiating-reorg counter increments per job rather than per collection cycle.

func (*Job) CountViolationOnce added in v1.19.1 

func (j *Job) CountViolationOnce() bool

CountViolationOnce reports true exactly once, the first time it is called, so the cross-safety-violation counter increments per job rather than per cycle.

func (*Job) DidMetrics 

func (j *Job) DidMetrics() bool

DidMetrics returns true if the job has been used to update the metrics at least once

func (*Job) FilterCheckedFor added in v1.19.1 

func (j *Job) FilterCheckedFor(status jobStatus) bool

FilterCheckedFor reports whether the interop-filter has already been compared against this exact monitor status, so the observer skips re-querying it until the monitor's verdict changes.

func (*Job) ID 

func (j *Job) ID() JobID

ID returns the ID of the job

func (*Job) InitiatingHashes 

func (j *Job) InitiatingHashes() []common.Hash

InitiatingHashes returns a copy of the initiating hashes

func (*Job) LastEvaluated 

func (j *Job) LastEvaluated() time.Time

LastEvaluated returns the last evaluated time of the job

func (*Job) LatestStatus 

func (j *Job) LatestStatus() jobStatus

LatestStatus returns the latest status of the job

func (*Job) MarkFilterChecked added in v1.19.1 

func (j *Job) MarkFilterChecked(status jobStatus)

MarkFilterChecked records the monitor status that was compared against the filter.

func (*Job) SetDidMetrics 

func (j *Job) SetDidMetrics()

SetDidMetrics sets the did metrics flag of the job

func (*Job) Statuses 

func (j *Job) Statuses() []jobStatus

Statuses returns the states of the job

func (*Job) String 

func (j *Job) String() string

String returns a string representation of the job

func (*Job) TerminalAt 

func (j *Job) TerminalAt() time.Time

TerminalAt returns the time the job last transitioned to a terminal state

func (*Job) UpdateLastEvaluated 

func (j *Job) UpdateLastEvaluated(t time.Time)

UpdateLastEvaluated updates the last evaluated time of the job

func (*Job) UpdateStatus 

func (j *Job) UpdateStatus(status jobStatus)

UpdateStatus updates the status of the job

func (*Job) ViolationCounted added in v1.19.1 

func (j *Job) ViolationCounted() bool

ViolationCounted reports whether a cross-safety violation has already been counted for this job, so the observer can skip re-confirming it every cycle.

type JobFilter 

type JobFilter func(receipts []*types.Receipt, executingChain eth.ChainID) []*Job

JobFilter is a function that turns any executing messages from a slice of receipts into a slice of jobs which can be added to an Updater's inbox

type JobID 

type JobID string

func JobId 

func JobId(
	executingBlockNumber uint64,
	executingLogIndex uint,
	executingPayload common.Hash,
	executingChain eth.ChainID,
	intitiatingBlockNumber uint64,
	logIndex uint32,
	initiatingChain eth.ChainID,
) JobID

type MetricCollector 

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

func NewMetricCollector 

func NewMetricCollector(log log.Logger, m InteropMessageMetrics, updaters map[eth.ChainID]Updater) *MetricCollector

func (*MetricCollector) CollectMetrics 

func (m *MetricCollector) CollectMetrics()

CollectMetrics scans the jobMaps, consolidates them, and updates the metrics

func (*MetricCollector) Run 

func (m *MetricCollector) Run()

Run is the main loop for the metric collector

func (*MetricCollector) Start 

func (m *MetricCollector) Start(ctx context.Context) error

func (*MetricCollector) Stop 

func (m *MetricCollector) Stop() error

func (*MetricCollector) Stopped 

func (m *MetricCollector) Stopped() bool

type NewCallback 

type NewCallback func(*Job)

NewCallback is a function to be called when a new job is created

type RPCFinder 

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

RPCFinder connects to an Ethereum chain and extracts receipts in order to create jobs

func NewFinder 

func NewFinder(chainID eth.ChainID,
	client FinderClient,
	toCases JobFilter,
	newCallback NewCallback,
	finalityCallback FinalityCallback,
	bufferSize int,
	log log.Logger) *RPCFinder

func (*RPCFinder) Run 

func (t *RPCFinder) Run(ctx context.Context)

func (*RPCFinder) Start 

func (t *RPCFinder) Start(ctx context.Context) error

func (*RPCFinder) Stop 

func (t *RPCFinder) Stop() error

func (*RPCFinder) Stopped 

func (t *RPCFinder) Stopped() bool

type RPCUpdater 

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

RPCFinder connects to an Ethereum chain and extracts receipts in order to create jobs

func NewUpdater 

func NewUpdater(
	chainID eth.ChainID,
	client UpdaterClient,
	finalized *locks.RWMap[eth.ChainID, eth.NumberAndHash],
	messageExpiryWindow uint64,
	log log.Logger) *RPCUpdater

func (*RPCUpdater) CollectForMetrics 

func (t *RPCUpdater) CollectForMetrics(jobs map[JobID]*Job) map[JobID]*Job

GetJobs adds all jobs to the provided map and returns it

func (*RPCUpdater) Enqueue 

func (t *RPCUpdater) Enqueue(job *Job)

todo: make this a priority queue

func (*RPCUpdater) Run 

func (t *RPCUpdater) Run(ctx context.Context)

func (*RPCUpdater) ShouldExpire 

func (t *RPCUpdater) ShouldExpire(job *Job) bool

ShouldExpire returns true if the job should be expired jobs should only be expired when *both components* exist in finalized blocks. That is: - the initiating block is finalized - the executing block is finalized Before this point, the job status could change if a reorg affects either the initiating or executing block. This also checks that the job has been evaluated at least once, and counted for metrics at least once.

func (*RPCUpdater) Start 

func (t *RPCUpdater) Start(ctx context.Context) error

func (*RPCUpdater) Stop 

func (t *RPCUpdater) Stop() error

TODO: add wait group to make Stop return sync

func (*RPCUpdater) Stopped 

func (t *RPCUpdater) Stopped() bool

func (*RPCUpdater) UpdateJob 

func (t *RPCUpdater) UpdateJob(job *Job) error

func (*RPCUpdater) UpdateJobStatus 

func (t *RPCUpdater) UpdateJobStatus(job *Job)

type SupernodeObserver added in v1.19.1 

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

SupernodeObserver watches one op-supernode (read-only). It records liveness, per-chain safe/finalized heads, and the highest-signal check: a bad executing message that the supernode has already promoted to cross-safe. It never gates monitor behaviour.

func NewSupernodeObserver added in v1.19.1 

func NewSupernodeObserver(endpoint string, c SupernodeObserverClient, els map[eth.ChainID]CanonicalBlockSource, m InteropMessageMetrics, log log.Logger) *SupernodeObserver

func (*SupernodeObserver) Observe added in v1.19.1 

func (o *SupernodeObserver) Observe(ctx context.Context, jobs map[JobID]*Job)

type SupernodeObserverClient added in v1.19.1 

type SupernodeObserverClient interface {
	// SyncStatus returns the supernode's aggregate per-chain sync status. A successful
	// call is also taken as the supernode's liveness signal.
	SyncStatus(ctx context.Context) (eth.SuperNodeSyncStatusResponse, error)
	Close()
}

SupernodeObserverClient is the read-only op-supernode surface the observer needs. It is satisfied by op-service/sources.SuperNodeClient.

type Updater 

type Updater interface {
	Start(ctx context.Context) error
	Enqueue(job *Job)
	Stop() error
	CollectForMetrics(jobs map[JobID]*Job) map[JobID]*Job
}

Updaters are responsible for updating jobs from a chain for the metric collector to track

type UpdaterClient 

type UpdaterClient interface {
	FetchReceiptsByNumber(ctx context.Context, number uint64) (eth.BlockInfo, types.Receipts, error)
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.