Merge branch 'dev' into fix_bls_aggregation_multiple_quorums

chainio/clients/elcontracts/writer.go

@@ -356,69 +356,3 @@ func (w *ChainWriter) ProcessClaim(
 
 	return receipt, nil
 }
-
-func (w *ChainWriter) ForceDeregisterFromOperatorSets(
-	ctx context.Context,
-	operator gethcommon.Address,
-	avs gethcommon.Address,
-	operatorSetIds []uint32,
-	operatorSignature avsdirectory.ISignatureUtilsSignatureWithSaltAndExpiry,
-	waitForReceipt bool,
-) (*gethtypes.Receipt, error) {
-	if w.avsDirectory == nil {
-		return nil, errors.New("AVSDirectory contract not provided")
-	}
-
-	noSendTxOpts, err := w.txMgr.GetNoSendTxOpts()
-	if err != nil {
-		return nil, utils.WrapError("failed to get no send tx opts", err)
-	}
-
-	tx, err := w.avsDirectory.ForceDeregisterFromOperatorSets(
-		noSendTxOpts,
-		operator,
-		avs,
-		operatorSetIds,
-		operatorSignature,
-	)
-
-	if err != nil {
-		return nil, utils.WrapError("failed to create ForceDeregisterFromOperatorSets tx", err)
-	}
-
-	receipt, err := w.txMgr.Send(ctx, tx, waitForReceipt)
-	if err != nil {
-		return nil, utils.WrapError("failed to send tx", err)
-	}
-
-	return receipt, nil
-}
-
-func (w *ChainWriter) SetOperatorCommissionBips(
-	ctx context.Context,
-	operatorSet rewardscoordinator.IAVSDirectoryOperatorSet,
-	rewardType uint8,
-	commissionBips uint16,
-	waitForReceipt bool,
-) (*gethtypes.Receipt, error) {
-	if w.rewardsCoordinator == nil {
-		return nil, errors.New("RewardsCoordinator contract not provided")
-	}
-
-	noSendTxOpts, err := w.txMgr.GetNoSendTxOpts()
-	if err != nil {
-		return nil, utils.WrapError("failed to get no send tx opts", err)
-	}
-
-	tx, err := w.rewardsCoordinator.SetOperatorCommissionBips(noSendTxOpts, operatorSet, rewardType, commissionBips)
-	if err != nil {
-		return nil, utils.WrapError("failed to create SetOperatorCommissionBips tx", err)
-	}
-
-	receipt, err := w.txMgr.Send(ctx, tx, waitForReceipt)
-	if err != nil {
-		return nil, utils.WrapError("failed to send tx", err)
-	}
-
-	return receipt, nil
-}

services/bls_aggregation/blsagg.go

@@ -81,12 +81,13 @@ type aggregatedOperators struct {
 // BlsAggregationService is the interface provided to avs aggregator code for doing bls aggregation
 // Currently its only implementation is the BlsAggregatorService, so see the comment there for more details
 type BlsAggregationService interface {
-	// InitializeNewTask should be called whenever a new task is created. ProcessNewSignature will return an error
-	// if the task it is trying to process has not been initialized yet.
-	// quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered complete, which
+	// InitializeNewTask creates a new task goroutine meant to process new signed task responses for that task
+	// (that are sent via ProcessNewSignature) and adds a channel to a.taskChans to send the signed task responses to
+	// it. The quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered
+	// complete, which
 	// happens when a particular TaskResponseDigest (received via the a.taskChans[taskIndex]) has been signed by signers
-	// whose stake
-	// in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in that quorum
+	// whose stake in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in
+	// that quorum
 	InitializeNewTask(
 		taskIndex types.TaskIndex,
 		taskCreatedBlock uint32,
@@ -95,6 +96,27 @@ type BlsAggregationService interface {
 		timeToExpiry time.Duration,
 	) error
 
+	// InitializeNewTaskWithWindow creates a new task goroutine meant to process new signed task responses for that task
+	// (that are sent via ProcessNewSignature) and adds a channel to a.taskChans to send the signed task responses to
+	// it. The quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered
+	// complete, which
+	// happens when a particular TaskResponseDigest (received via the a.taskChans[taskIndex]) has been signed by signers
+	// whose stake in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in
+	// that quorum.
+	// Once the quorum is reached, the task is still open for a window of `windowDuration` time to receive more
+	// signatures,
+	// before sending the aggregation response through the aggregatedResponsesC channel.
+	// If the task expiration is reached before the window finishes, the task response will still be sent to the
+	// aggregatedResponsesC channel.
+	InitializeNewTaskWithWindow(
+		taskIndex types.TaskIndex,
+		taskCreatedBlock uint32,
+		quorumNumbers types.QuorumNums,
+		quorumThresholdPercentages types.QuorumThresholdPercentages,
+		timeToExpiry time.Duration,
+		windowDuration time.Duration,
+	) error
+
 	// ProcessNewSignature processes a new signature over a taskResponseDigest for a particular taskIndex by a
 	// particular operator It verifies that the signature is correct and returns an error if it is not, and then
 	// aggregates the signature and stake of
@@ -185,17 +207,43 @@ func (a *BlsAggregatorService) GetResponseChannel() <-chan BlsAggregationService
 }
 
 // InitializeNewTask creates a new task goroutine meant to process new signed task responses for that task
-// (that are sent via ProcessNewSignature) and adds a channel to a.taskChans to send the signed task responses to it
-// quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered complete, which
-// happens
-// when a particular TaskResponseDigest (received via the a.taskChans[taskIndex]) has been signed by signers whose stake
-// in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in that quorum
+// (that are sent via ProcessNewSignature) and adds a channel to a.taskChans to send the signed task responses to it.
+// The quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered complete, which
+// happens when a particular TaskResponseDigest (received via the a.taskChans[taskIndex]) has been signed by signers
+// whose stake in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in
+// that quorum
 func (a *BlsAggregatorService) InitializeNewTask(
 	taskIndex types.TaskIndex,
 	taskCreatedBlock uint32,
 	quorumNumbers types.QuorumNums,
 	quorumThresholdPercentages types.QuorumThresholdPercentages,
 	timeToExpiry time.Duration,
+) error {
+	return a.InitializeNewTaskWithWindow(
+		taskIndex,
+		taskCreatedBlock,
+		quorumNumbers,
+		quorumThresholdPercentages,
+		timeToExpiry,
+		0,
+	)
+}
+
+// InitializeNewTaskWithWindow creates a new task goroutine meant to process new signed task responses for that task
+// (that are sent via ProcessNewSignature) and adds a channel to a.taskChans to send the signed task responses to it.
+// The quorumNumbers and quorumThresholdPercentages set the requirements for this task to be considered complete, which
+// happens when a particular TaskResponseDigest (received via the a.taskChans[taskIndex]) has been signed by signers
+// whose stake in each of the listed quorums adds up to at least quorumThresholdPercentages[i] of the total stake in
+// that quorum.
+// Once the quorum is reached, the task is still open for a window of `windowDuration` time to receive more signatures,
+// before sending the aggregation response through the aggregatedResponsesC channel.
+func (a *BlsAggregatorService) InitializeNewTaskWithWindow(
+	taskIndex types.TaskIndex,
+	taskCreatedBlock uint32,
+	quorumNumbers types.QuorumNums,
+	quorumThresholdPercentages types.QuorumThresholdPercentages,
+	timeToExpiry time.Duration,
+	windowDuration time.Duration,
 ) error {
 	a.logger.Debug(
 		"AggregatorService initializing new task",
@@ -225,6 +273,7 @@ func (a *BlsAggregatorService) InitializeNewTask(
 		quorumNumbers,
 		quorumThresholdPercentages,
 		timeToExpiry,
+		windowDuration,
 		signedTaskRespsC,
 	)
 	return nil
@@ -271,6 +320,7 @@ func (a *BlsAggregatorService) singleTaskAggregatorGoroutineFunc(
 	quorumNumbers types.QuorumNums,
 	quorumThresholdPercentages []types.QuorumThresholdPercentage,
 	timeToExpiry time.Duration,
+	windowDuration time.Duration,
 	signedTaskRespsC <-chan types.SignedTaskResponseDigest,
 ) {
 	a.logger.Debug("AggregatorService goroutine processing new task",
@@ -342,6 +392,13 @@ func (a *BlsAggregatorService) singleTaskAggregatorGoroutineFunc(
 	taskExpiredTimer := time.NewTimer(timeToExpiry)
 
 	aggregatedOperatorsDict := map[types.TaskResponseDigest]aggregatedOperators{}
+	// The windowTimer is initialized to be longer than the taskExpiredTimer as it will
+	// be overwritten once the stake threshold is met
+	windowTimer := time.NewTimer(timeToExpiry + 1*time.Second)
+	openWindow := false
+	var lastSignedTaskResponseDigest types.SignedTaskResponseDigest
+	var lastDigestAggregatedOperators aggregatedOperators
+	var lastTaskResponseDigest types.TaskResponseDigest
 	for {
 		select {
 		case signedTaskResponseDigest := <-signedTaskRespsC:
@@ -426,11 +483,17 @@ func (a *BlsAggregatorService) singleTaskAggregatorGoroutineFunc(
 					digestAggregatedOperators.signersTotalStakePerQuorum[quorumNum].Add(digestAggregatedOperators.signersTotalStakePerQuorum[quorumNum], stake)
 				}
 			}
+
+			// update the buffer variables to be used when the window timer fires
+			lastDigestAggregatedOperators = digestAggregatedOperators
+			lastTaskResponseDigest = taskResponseDigest
+			lastSignedTaskResponseDigest = signedTaskResponseDigest
+
 			// update the aggregatedOperatorsDict. Note that we need to assign the whole struct value at once,
 			// because of https://github.com/golang/go/issues/3117
 			aggregatedOperatorsDict[taskResponseDigest] = digestAggregatedOperators
 
-			if checkIfStakeThresholdsMet(
+			if !openWindow && checkIfStakeThresholdsMet(
 				a.logger,
 				digestAggregatedOperators.signersTotalStakePerQuorum,
 				totalStakePerQuorum,
@@ -440,67 +503,105 @@ func (a *BlsAggregatorService) singleTaskAggregatorGoroutineFunc(
 					"taskIndex", taskIndex,
 					"taskResponseDigest", taskResponseDigest)
 
-				nonSignersOperatorIds := []types.OperatorId{}
-				for operatorId := range operatorsAvsStateDict {
-					if _, operatorSigned := digestAggregatedOperators.signersOperatorIdsSet[operatorId]; !operatorSigned {
-						nonSignersOperatorIds = append(nonSignersOperatorIds, operatorId)
-					}
-				}
-
-				// the contract requires a sorted nonSignersOperatorIds
-				sort.SliceStable(nonSignersOperatorIds, func(i, j int) bool {
-					iOprInt := new(big.Int).SetBytes(nonSignersOperatorIds[i][:])
-					jOprInt := new(big.Int).SetBytes(nonSignersOperatorIds[j][:])
-					return iOprInt.Cmp(jOprInt) == -1
-				})
-
-				nonSignersG1Pubkeys := []*bls.G1Point{}
-				for _, operatorId := range nonSignersOperatorIds {
-					operator := operatorsAvsStateDict[operatorId]
-					nonSignersG1Pubkeys = append(nonSignersG1Pubkeys, operator.OperatorInfo.Pubkeys.G1Pubkey)
-				}
-
-				indices, err := a.avsRegistryService.GetCheckSignaturesIndices(
-					&bind.CallOpts{},
+				openWindow = true
+				windowTimer = time.NewTimer(windowDuration)
+				a.logger.Debug("Window timer started")
+			}
+		case <-taskExpiredTimer.C:
+			if openWindow {
+				a.sendAggregatedResponse(
+					operatorsAvsStateDict,
+					taskIndex,
 					taskCreatedBlock,
+					lastSignedTaskResponseDigest,
+					lastDigestAggregatedOperators,
 					quorumNumbers,
-					nonSignersOperatorIds,
+					lastTaskResponseDigest,
+					quorumApksG1,
 				)
-				if err != nil {
-					a.aggregatedResponsesC <- BlsAggregationServiceResponse{
-						Err:       utils.WrapError(errors.New("Failed to get check signatures indices"), err),
-						TaskIndex: taskIndex,
-					}
-					return
-				}
-
-				blsAggregationServiceResponse := BlsAggregationServiceResponse{
-					Err:                          nil,
-					TaskIndex:                    taskIndex,
-					TaskResponse:                 signedTaskResponseDigest.TaskResponse,
-					TaskResponseDigest:           taskResponseDigest,
-					NonSignersPubkeysG1:          nonSignersG1Pubkeys,
-					QuorumApksG1:                 quorumApksG1,
-					SignersApkG2:                 digestAggregatedOperators.signersApkG2,
-					SignersAggSigG1:              digestAggregatedOperators.signersAggSigG1,
-					NonSignerQuorumBitmapIndices: indices.NonSignerQuorumBitmapIndices,
-					QuorumApkIndices:             indices.QuorumApkIndices,
-					TotalStakeIndices:            indices.TotalStakeIndices,
-					NonSignerStakeIndices:        indices.NonSignerStakeIndices,
-				}
-				a.aggregatedResponsesC <- blsAggregationServiceResponse
-				taskExpiredTimer.Stop()
-				return
 			}
-		case <-taskExpiredTimer.C:
+
 			a.aggregatedResponsesC <- BlsAggregationServiceResponse{
 				Err:       TaskExpiredErrorFn(taskIndex),
 				TaskIndex: taskIndex,
 			}
 			return
+		case <-windowTimer.C:
+			a.logger.Debug("Window timer expired")
+			a.sendAggregatedResponse(
+				operatorsAvsStateDict,
+				taskIndex,
+				taskCreatedBlock,
+				lastSignedTaskResponseDigest,
+				lastDigestAggregatedOperators,
+				quorumNumbers,
+				lastTaskResponseDigest,
+				quorumApksG1,
+			)
+			return
 		}
 	}
+}
 
+func (a *BlsAggregatorService) sendAggregatedResponse(
+	operatorsAvsStateDict map[types.OperatorId]types.OperatorAvsState,
+	taskIndex types.TaskIndex,
+	taskCreatedBlock uint32,
+	signedTaskResponseDigest types.SignedTaskResponseDigest,
+	digestAggregatedOperators aggregatedOperators,
+	quorumNumbers types.QuorumNums,
+	taskResponseDigest types.TaskResponseDigest,
+	quorumApksG1 []*bls.G1Point,
+) {
+	nonSignersOperatorIds := []types.OperatorId{}
+	for operatorId := range operatorsAvsStateDict {
+		if _, operatorSigned := digestAggregatedOperators.signersOperatorIdsSet[operatorId]; !operatorSigned {
+			nonSignersOperatorIds = append(nonSignersOperatorIds, operatorId)
+		}
+	}
+
+	// the contract requires a sorted nonSignersOperatorIds
+	sort.SliceStable(nonSignersOperatorIds, func(i, j int) bool {
+		iOprInt := new(big.Int).SetBytes(nonSignersOperatorIds[i][:])
+		jOprInt := new(big.Int).SetBytes(nonSignersOperatorIds[j][:])
+		return iOprInt.Cmp(jOprInt) == -1
+	})
+
+	nonSignersG1Pubkeys := []*bls.G1Point{}
+	for _, operatorId := range nonSignersOperatorIds {
+		operator := operatorsAvsStateDict[operatorId]
+		nonSignersG1Pubkeys = append(nonSignersG1Pubkeys, operator.OperatorInfo.Pubkeys.G1Pubkey)
+	}
+
+	indices, err := a.avsRegistryService.GetCheckSignaturesIndices(
+		&bind.CallOpts{},
+		taskCreatedBlock,
+		quorumNumbers,
+		nonSignersOperatorIds,
+	)
+	if err != nil {
+		a.aggregatedResponsesC <- BlsAggregationServiceResponse{
+			Err:       utils.WrapError(errors.New("Failed to get check signatures indices"), err),
+			TaskIndex: taskIndex,
+		}
+		return
+	}
+
+	blsAggregationServiceResponse := BlsAggregationServiceResponse{
+		Err:                          nil,
+		TaskIndex:                    taskIndex,
+		TaskResponse:                 signedTaskResponseDigest.TaskResponse,
+		TaskResponseDigest:           taskResponseDigest,
+		NonSignersPubkeysG1:          nonSignersG1Pubkeys,
+		QuorumApksG1:                 quorumApksG1,
+		SignersApkG2:                 digestAggregatedOperators.signersApkG2,
+		SignersAggSigG1:              digestAggregatedOperators.signersAggSigG1,
+		NonSignerQuorumBitmapIndices: indices.NonSignerQuorumBitmapIndices,
+		QuorumApkIndices:             indices.QuorumApkIndices,
+		TotalStakeIndices:            indices.TotalStakeIndices,
+		NonSignerStakeIndices:        indices.NonSignerStakeIndices,
+	}
+	a.aggregatedResponsesC <- blsAggregationServiceResponse
 }
 
 // closeTaskGoroutine is run when the goroutine processing taskIndex's task responses ends (for whatever reason)