@@ -3,16 +3,18 @@ package pkg
33import (
44 "context"
55 "encoding/hex"
6+ "fmt"
67 "net/http"
78 "net/rpc"
9+ "strings"
810 "time"
911
12+ "github.com/Layr-Labs/eigensdk-go/crypto/bls"
13+ eigentypes "github.com/Layr-Labs/eigensdk-go/types"
14+ retry "github.com/yetanotherco/aligned_layer/core"
1015 "github.com/yetanotherco/aligned_layer/core/types"
1116)
1217
13- const waitForEventRetries = 50
14- const waitForEventSleepSeconds = 4 * time .Second
15-
1618func (agg * Aggregator ) ServeOperators () error {
1719 // Registers a new RPC server
1820 err := rpc .Register (agg )
@@ -50,22 +52,10 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
5052 "BatchIdentifierHash" , "0x" + hex .EncodeToString (signedTaskResponse .BatchIdentifierHash [:]),
5153 "operatorId" , hex .EncodeToString (signedTaskResponse .OperatorId [:]))
5254 taskIndex := uint32 (0 )
53- ok := false
5455
55- for i := 0 ; i < waitForEventRetries ; i ++ {
56- agg .taskMutex .Lock ()
57- agg .AggregatorConfig .BaseConfig .Logger .Info ("- Locked Resources: Starting processing of Response" )
58- taskIndex , ok = agg .batchesIdxByIdentifierHash [signedTaskResponse .BatchIdentifierHash ]
59- if ! ok {
60- agg .taskMutex .Unlock ()
61- agg .logger .Info ("- Unlocked Resources: Task not found in the internal map" )
62- time .Sleep (waitForEventSleepSeconds )
63- } else {
64- break
65- }
66- }
56+ taskIndex , err := agg .GetTaskIndex (signedTaskResponse .BatchIdentifierHash )
6757
68- if ! ok {
58+ if err != nil {
6959 agg .logger .Warn ("Task not found in the internal map, operator signature will be lost. Batch may not reach quorum" )
7060 * reply = 1
7161 return nil
@@ -82,7 +72,7 @@ func (agg *Aggregator) ProcessOperatorSignedTaskResponseV2(signedTaskResponse *t
8272
8373 agg .logger .Info ("Starting bls signature process" )
8474 go func () {
85- err := agg .blsAggregationService . ProcessNewSignature (
75+ err := agg .ProcessNewSignature (
8676 context .Background (), taskIndex , signedTaskResponse .BatchIdentifierHash ,
8777 & signedTaskResponse .BlsSignature , signedTaskResponse .OperatorId ,
8878 )
@@ -121,3 +111,48 @@ func (agg *Aggregator) ServerRunning(_ *struct{}, reply *int64) error {
121111 * reply = 1
122112 return nil
123113}
114+
115+ // |---RETRYABLE---|
116+
117+ /*
118+ - Errors:
119+ Permanent:
120+ - SignatureVerificationError: Verification of the sigature within the BLS Aggregation Service failed. (https://github.com/Layr-Labs/eigensdk-go/blob/dev/services/bls_aggregation/blsagg.go#L42).
121+ Transient:
122+ - All others.
123+ - Retry times (3 retries): 12 sec (1 Blocks), 24 sec (2 Blocks), 48 sec (4 Blocks)
124+ - NOTE: TaskNotFound errors from the BLS Aggregation service are Transient errors as block reorg's may lead to these errors being thrown.
125+ */
126+ func (agg * Aggregator ) ProcessNewSignature (ctx context.Context , taskIndex uint32 , taskResponse interface {}, blsSignature * bls.Signature , operatorId eigentypes.Bytes32 ) error {
127+ processNewSignature_func := func () error {
128+ err := agg .blsAggregationService .ProcessNewSignature (
129+ ctx , taskIndex , taskResponse ,
130+ blsSignature , operatorId ,
131+ )
132+ if err != nil {
133+ if strings .Contains (err .Error (), "Failed to verify signature" ) {
134+ err = retry.PermanentError {Inner : err }
135+ }
136+ }
137+ return err
138+ }
139+
140+ return retry .Retry (processNewSignature_func , retry .MinDelayChain , retry .RetryFactor , retry .NumRetries , retry .MaxIntervalChain , retry .MaxElapsedTime )
141+ }
142+
143+ func (agg * Aggregator ) GetTaskIndex (batchIdentifierHash [32 ]byte ) (uint32 , error ) {
144+ getTaskIndex_func := func () (uint32 , error ) {
145+ agg .taskMutex .Lock ()
146+ agg .AggregatorConfig .BaseConfig .Logger .Info ("- Locked Resources: Starting processing of Response" )
147+ taskIndex , ok := agg .batchesIdxByIdentifierHash [batchIdentifierHash ]
148+ if ! ok {
149+ agg .taskMutex .Unlock ()
150+ agg .logger .Info ("- Unlocked Resources: Task not found in the internal map" )
151+ return taskIndex , fmt .Errorf ("Task not found in the internal map" )
152+ } else {
153+ return taskIndex , nil
154+ }
155+ }
156+
157+ return retry .RetryWithData (getTaskIndex_func , retry .MinDelay , retry .RetryFactor , retry .NumRetries , retry .MaxInterval , retry .MaxElapsedTime )
158+ }
0 commit comments