pkg

Documentation

Index

• Constants
• type Aggregator
  • func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error)
  • func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32)
  • func (agg *Aggregator) ClearTasksFromMaps()
  • func (agg *Aggregator) GetTaskIndexRetryable(batchIdentifierHash [32]byte, config *retry.RetryParams) (uint32, error)
  • func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *types.SignedTaskResponse, reply *uint8) error
  • func (agg *Aggregator) ServeOperators() error
  • func (agg *Aggregator) ServerRunning(_ *struct{}, reply *int64) error
  • func (agg *Aggregator) Start(ctx context.Context) error
  • func (agg *Aggregator) SubscribeToNewTasks() error
• type BatchData
• type OperatorResponseMessage
• type QuorumReachedMessage
• type TaskErrorMessage
• type TaskGasPriceBumpMessage
• type TaskResponses
• type TaskSentToEthereumMessage
• type Telemetry
  • func NewTelemetry(serverAddress string, logger logging.Logger) *Telemetry
  • func (t *Telemetry) BumpedTaskGasPrice(batchMerkleRoot [32]byte, bumpedGasPrice string)
  • func (t *Telemetry) FinishTrace(batchMerkleRoot [32]byte)
  • func (t *Telemetry) InitNewTrace(batchMerkleRoot [32]byte)
  • func (t *Telemetry) LogOperatorResponse(batchMerkleRoot [32]byte, operatorId [32]byte)
  • func (t *Telemetry) LogQuorumReached(batchMerkleRoot [32]byte)
  • func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error)
  • func (t *Telemetry) TaskSentToEthereum(batchMerkleRoot [32]byte, txHash string)
• type TraceMessage

Constants

const MaxSentTxRetries = 5

const QUORUM_NUMBER = byte(0)

FIXME(marian): Read this from Aligned contract directly

const QUORUM_THRESHOLD = byte(67)

Types

type Aggregator

type Aggregator struct {
	AggregatorConfig *config.AggregatorConfig
	NewBatchChan     chan *servicemanager.ContractAlignedLayerServiceManagerNewBatchV3
	// contains filtered or unexported fields
}

func NewAggregator

func NewAggregator(aggregatorConfig config.AggregatorConfig) (*Aggregator, error)

func (*Aggregator) AddNewTask

func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32)

func (*Aggregator) ClearTasksFromMaps

func (agg *Aggregator) ClearTasksFromMaps()

Long-lived goroutine that periodically checks and removes old Tasks from stored Maps It runs every GarbageCollectorPeriod and removes all tasks older than GarbageCollectorTasksAge This was added because each task occupies memory in the maps, and we need to free it to avoid a memory leak

func (*Aggregator) GetTaskIndexRetryable

func (agg *Aggregator) GetTaskIndexRetryable(batchIdentifierHash [32]byte, config *retry.RetryParams) (uint32, error)

Checks Internal mapping for Signed Task Response, returns its TaskIndex. - All errors are considered Transient Errors - Retry times (3 retries): 1 sec, 2 sec, 4 sec TODO: We should refactor the retry duration considering extending it to a larger time or number of retries, at least somewhere between 1 and 2 blocks

func (*Aggregator) ProcessOperatorSignedTaskResponseV2

func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *types.SignedTaskResponse, reply *uint8) error

Aggregator Methods This is the list of methods that the Aggregator exposes to the Operator The Operator can call these methods to interact with the Aggregator This methods are automatically registered by the RPC server This takes a response an adds it to the internal. If reaching the quorum, it sends the aggregated signatures to ethereum Returns:

• 0: Success
• 1: Error

func (*Aggregator) ServeOperators

func (agg *Aggregator) ServeOperators() error

func (*Aggregator) ServerRunning

func (agg *Aggregator) ServerRunning(_ *struct{}, reply *int64) error

Dummy method to check if the server is running TODO: Remove this method in prod

func (*Aggregator) Start

func (agg *Aggregator) Start(ctx context.Context) error

func (*Aggregator) SubscribeToNewTasks

func (agg *Aggregator) SubscribeToNewTasks() error

type BatchData

type BatchData struct {
	BatchMerkleRoot [32]byte
	SenderAddress   [20]byte
}

BatchData stores the data of a batch, for use in map BatchIdentifierHash -> BatchData

type OperatorResponseMessage

type OperatorResponseMessage struct {
	MerkleRoot string `json:"merkle_root"`
	OperatorId string `json:"operator_id"`
}

type QuorumReachedMessage

type QuorumReachedMessage struct {
	MerkleRoot string `json:"merkle_root"`
}

type TaskErrorMessage

type TaskErrorMessage struct {
	MerkleRoot string `json:"merkle_root"`
	TaskError  string `json:"error"`
}

type TaskGasPriceBumpMessage

type TaskGasPriceBumpMessage struct {
	MerkleRoot     string `json:"merkle_root"`
	BumpedGasPrice string `json:"bumped_gas_price"`
}

type TaskResponses

type TaskResponses = []types.SignedTaskResponse

Aggregator stores TaskResponse for a task here

type TaskSentToEthereumMessage

type TaskSentToEthereumMessage struct {
	MerkleRoot string `json:"merkle_root"`
	TxHash     string `json:"tx_hash"`
}

type Telemetry

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

func NewTelemetry

func NewTelemetry(serverAddress string, logger logging.Logger) *Telemetry

func (*Telemetry) BumpedTaskGasPrice

func (t *Telemetry) BumpedTaskGasPrice(batchMerkleRoot [32]byte, bumpedGasPrice string)

func (*Telemetry) FinishTrace

func (t *Telemetry) FinishTrace(batchMerkleRoot [32]byte)

func (*Telemetry) InitNewTrace

func (t *Telemetry) InitNewTrace(batchMerkleRoot [32]byte)

func (*Telemetry) LogOperatorResponse

func (t *Telemetry) LogOperatorResponse(batchMerkleRoot [32]byte, operatorId [32]byte)

func (*Telemetry) LogQuorumReached

func (t *Telemetry) LogQuorumReached(batchMerkleRoot [32]byte)

func (*Telemetry) LogTaskError

func (t *Telemetry) LogTaskError(batchMerkleRoot [32]byte, taskError error)

func (*Telemetry) TaskSentToEthereum

func (t *Telemetry) TaskSentToEthereum(batchMerkleRoot [32]byte, txHash string)

type TraceMessage

type TraceMessage struct {
	MerkleRoot string `json:"merkle_root"`
}