taskconsensus

Documentation

Index

• type OperatorRaftConfig
• type Service
  • func NewService[K any](addr string, onNewOperatorJoiningCluster onNewOperatorJoiningCluster, ...) *Service[K]
  • func (s *Service[K]) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (s *Service[K]) Start() error
• type SignedTaskResponse
• type TaskConsensusCallbacks
• type TaskConsensusEngine
  • func (f *TaskConsensusEngine[T, K, S]) Apply(l *raft.Log) interface{}
  • func (p *TaskConsensusEngine[T, K, S]) GetHttpBindingUrl() string
  • func (p *TaskConsensusEngine[T, K, S]) InitializeRaftRpcServer(shouldBootstrapCluster bool, operatorId string) error
  • func (p *TaskConsensusEngine[T, K, S]) IsLeader() (bool, string)
  • func (p *TaskConsensusEngine[T, K, S]) Join(operatorId, addr string) error
  • func (p *TaskConsensusEngine[T, K, S]) JoinExistingOperatorCluster(joinHttpUrl string, latestBlock uint64) error
  • func (p *TaskConsensusEngine[T, K, S]) LeaderSendTaskRequestToFollowers(taskRequest T) error
  • func (f *TaskConsensusEngine[T, K, S]) Restore(rc io.ReadCloser) error
  • func (f *TaskConsensusEngine[T, K, S]) Snapshot() (raft.FSMSnapshot, error)
  • func (p *TaskConsensusEngine[T, K, S]) SubmitTaskToLeader(request T, responses []K) error
  • func (p *TaskConsensusEngine[T, K, S]) TriggerElection()
• type TaskConsensusManager
  • func NewAVSConcensusEngine[T any, K any, S any](keyPair *bls.KeyPair, pk *ecdsa.PrivateKey, ...) (TaskConsensusManager[T, K, S], error)

Types

type OperatorRaftConfig

type OperatorRaftConfig struct {
	// Http url that the operator must provide and expose to create the needed custom http server
	HttpUrl string
	// Rpc url that the operator must provide and expose to connect to a raft cluster
	RpcUrl string
	// Path to directory where raft protocol distributed replicated log is stored
	FileStorageDirectory string
	OperatorId           sdktypes.OperatorId
}

type Service

type Service[K any] struct {
	// contains filtered or unexported fields
}

Type K is the task response submitted from followers to the leader

func NewService

func NewService[K any](addr string, onNewOperatorJoiningCluster onNewOperatorJoiningCluster, blsAggregationService blsagg.BlsAggregationService, ethClient eth.Client) *Service[K]

New returns an uninitialized HTTP service.

func (*Service[K]) ServeHTTP

func (s *Service[K]) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP allows Service to serve HTTP requests.

func (*Service[K]) Start

func (s *Service[K]) Start() error

Start starts the service.

type SignedTaskResponse

type SignedTaskResponse[K any] struct {
	TaskResponse []K
	BlsSignature []bls.Signature
	OperatorId   sdktypes.OperatorId
}

We provide a default signed response type (S) where the type K is the task response type that followers submit to the leader

type TaskConsensusCallbacks

type TaskConsensusCallbacks[T any, K any, S any] struct {
	// Method that is triggered when a follower receives a task request (T) from the current leader. The follower resonse (K) is returned
	OnTaskRequestFn onTaskRequest[T, K]

	// Method that is triggered when a follower want to sign their task response (K) with a BLS signature and submit that response to the leader
	OnTaskResponseFn onSubmitTaskToLeader[T, K, S]

	// Method that is used to verify that a given operator address
	IsValidOperator isRegisteredOperator

	// Method that is triggered when the current leader receives a task response (K) from a follower and generatesa taskDigest
	// The task digest is essentially an unsigned hash of the task fileds and values
	OnLeaderProcessTaskResponse onLeaderProcessTaskResponse[K]

	// Method that fetches the raftRpc and http urls for a given operator address
	// It is up to the AVS developers to implement how operator urls are discovered by other operators
	FetchOperatorUrl fetchOperatorUrl
}

Callbacks that must be implemented by AVS developers Type T is the task request type is sent from the leader to followers Type K is the task response type from followers to the leader Type S is the bls signed response type submitted to the leader

type TaskConsensusEngine

type TaskConsensusEngine[T any, K any, S any] struct {
	RaftDir      string // Directory for operator raft logs
	RaftRpcBind  string // rpc host:port used by the operator for raft protocol
	RaftHttpBind string // http host:port for custom server for custom raft logic
	// contains filtered or unexported fields
}

Type T is the task request that is sent from the leader to followers Type K is the task response submitted from followers to the leader Type S is the bls signed response type submitted to the leader

--------------------------------------------------------------

The task engine provides a minimal framework where AVS developers define the functionallity of task generation, aggregation and on-chain submission. This engine uses the raft protcol and ensures that there can only ever be 1 leader.

func (*TaskConsensusEngine[T, K, S]) Apply

func (f *TaskConsensusEngine[T, K, S]) Apply(l *raft.Log) interface{}

func (*TaskConsensusEngine[T, K, S]) GetHttpBindingUrl

func (p *TaskConsensusEngine[T, K, S]) GetHttpBindingUrl() string

func (*TaskConsensusEngine[T, K, S]) InitializeRaftRpcServer

func (p *TaskConsensusEngine[T, K, S]) InitializeRaftRpcServer(shouldBootstrapCluster bool, operatorId string) error

Operator initializes raft consenses server if enableSingle is set, and there are no existing peers, then this node becomes the first node, and therefore leader, of the cluster. operatorId should be the server identifier for this node.

func (*TaskConsensusEngine[T, K, S]) IsLeader

func (p *TaskConsensusEngine[T, K, S]) IsLeader() (bool, string)

Checks if an operator is the current leader of the raft cluster it is connected to This can be used to gate operator functionallity by leaders and followers

func (*TaskConsensusEngine[T, K, S]) Join

func (p *TaskConsensusEngine[T, K, S]) Join(operatorId, addr string) error

Join joins a node, identified by nodeID and located at addr, to this store. The node must be ready to respond to Raft communications at that address.

func (*TaskConsensusEngine[T, K, S]) JoinExistingOperatorCluster

func (p *TaskConsensusEngine[T, K, S]) JoinExistingOperatorCluster(joinHttpUrl string, latestBlock uint64) error

Operator attempts to join an existing raft cluster of operators. joinHttpUrl string: The url of an operator that is already connected to an existing raft cluster. It is up to the AVS developer to implement a way for urls to be discovered latestBlock uint64: the latest block the operator is aware of.

func (*TaskConsensusEngine[T, K, S]) LeaderSendTaskRequestToFollowers

func (p *TaskConsensusEngine[T, K, S]) LeaderSendTaskRequestToFollowers(taskRequest T) error

func (*TaskConsensusEngine[T, K, S]) Restore

func (f *TaskConsensusEngine[T, K, S]) Restore(rc io.ReadCloser) error

Restore stores the key-value store to a previous state.

func (*TaskConsensusEngine[T, K, S]) Snapshot

func (f *TaskConsensusEngine[T, K, S]) Snapshot() (raft.FSMSnapshot, error)

func (*TaskConsensusEngine[T, K, S]) SubmitTaskToLeader

func (p *TaskConsensusEngine[T, K, S]) SubmitTaskToLeader(request T, responses []K) error

func (*TaskConsensusEngine[T, K, S]) TriggerElection

func (p *TaskConsensusEngine[T, K, S]) TriggerElection()

Only the current leader can trigger a new election manually The raft protocol handles automatic re-elections is a leader goes offline

type TaskConsensusManager

type TaskConsensusManager[T any, K any, S any] interface {
	// must be called for an operator to initialize raft rpc server and decide if a new cluster should be bootstraped
	InitializeRaftRpcServer(shouldBootstrapCluster bool, operatorId string) error
	// attempts to join an existing raft cluster
	JoinExistingOperatorCluster(joinHttpUrl string, latestBlock uint64) error
	// Returns true if the operator is the leader on a cluster. False otherwise
	IsLeader() (bool, string)
	// Triggers a new leader election within the raft cluster. Only the current leader can trigger this
	TriggerElection()
	// Method to trigger the current leader of a cluster to send a task request to follower
	// Only the leader can call this method
	LeaderSendTaskRequestToFollowers(taskRequest T) error

	GetHttpBindingUrl() string
}

This represents an instance of the task engine once it is initiatied and all the methods avalible to developers when building their AVS

func NewAVSConcensusEngine

func NewAVSConcensusEngine[T any, K any, S any](keyPair *bls.KeyPair, pk *ecdsa.PrivateKey, blsAggregationService blsagg.BlsAggregationService, ethClient eth.Client, logger logging.Logger, callbacks TaskConsensusCallbacks[T, K, S], operatorRaftConfig OperatorRaftConfig) (TaskConsensusManager[T, K, S], error)

Type T is the task request type is sent from the leader to followers Type K is the task response type from followers to the leader Type S is the bls signed response type submitted to the leader