Add context on few error logs + early return on encoding streamer (#372)

core/assignment.go

@@ -106,7 +106,7 @@ func (c *StdAssignmentCoordinator) GetAssignments(state *OperatorState, blobLeng
 		gammaChunkLength := big.NewInt(int64(info.ChunkLength) * int64((info.ConfirmationThreshold - info.AdversaryThreshold)))
 		denom := new(big.Int).Mul(gammaChunkLength, totalStakes)
 		if denom.Cmp(big.NewInt(0)) == 0 {
-			return nil, AssignmentInfo{}, fmt.Errorf("gammaChunkLength %d and total stake in quorum %d must be greater than 0", gammaChunkLength, totalStakes)
+			return nil, AssignmentInfo{}, fmt.Errorf("gammaChunkLength %d and total stake %d in quorum %d must be greater than 0", gammaChunkLength, totalStakes, quorum)
 		}
 		m := roundUpDivideBig(num, denom)
 

core/thegraph/state.go

@@ -132,7 +132,7 @@ func (ics *indexedChainState) GetIndexedOperatorState(ctx context.Context, block
 		}
 	}
 	if len(aggKeys) == 0 {
-		return nil, errors.New("no aggregate public keys found for any of the specified quorums")
+		return nil, fmt.Errorf("no aggregate public keys found for any of the specified quorums at block number %d", blockNumber)
 	}
 
 	indexedOperators, err := ics.getRegisteredIndexedOperatorInfo(ctx, uint32(blockNumber))
@@ -177,7 +177,7 @@ func (ics *indexedChainState) GetIndexedOperatorInfoByOperatorId(ctx context.Con
 	)
 	err := ics.querier.Query(context.Background(), &query, variables)
 	if err != nil {
-		ics.logger.Error("Error requesting for operator", "err", err)
+		ics.logger.Error("Error requesting info for operator", "err", err, "operatorId", operatorId.Hex(), "blockNumber", blockNumber)
 		return nil, err
 	}
 

disperser/batcher/encoding_streamer.go

@@ -314,7 +314,6 @@ func (e *EncodingStreamer) RequestEncodingForBlob(ctx context.Context, metadata
 			},
 			ChunkLength: chunkLength,
 		}
-
 		assignments, info, err := e.assignmentCoordinator.GetAssignments(state.OperatorState, blobLength, blobQuorumInfo)
 		if err != nil {
 			e.logger.Error("[RequestEncodingForBlob] error getting assignments", "err", err)
@@ -343,7 +342,6 @@ func (e *EncodingStreamer) RequestEncodingForBlob(ctx context.Context, metadata
 
 	// Execute the encoding requests
 	for ind := range pending {
-
 		res := pending[ind]
 
 		// Create a new context for each encoding request
@@ -502,6 +500,7 @@ func (e *EncodingStreamer) CreateBatch() (*batch, error) {
 		for _, quorum := range blobQuorums[blobKey] {
 			quorumPresent[quorum.QuorumID] = true
 		}
+		// Check if the blob has valid quorums. If any of the quorums are not valid, delete the blobKey
 		for _, quorum := range metadata.RequestMetadata.SecurityParams {
 			_, ok := quorumPresent[quorum.QuorumID]
 			if !ok {
@@ -513,6 +512,10 @@ func (e *EncodingStreamer) CreateBatch() (*batch, error) {
 		}
 	}
 
+	if len(metadataByKey) == 0 {
+		return nil, errNoEncodedResults
+	}
+
 	// Transform maps to slices so orders in different slices match
 	encodedBlobs := make([]core.EncodedBlob, len(metadataByKey))
 	blobHeaders := make([]*core.BlobHeader, len(metadataByKey))

disperser/cmd/batcher/main.go

@@ -228,6 +228,7 @@ func RunBatcher(ctx *cli.Context) error {
 		if err != nil {
 			return err
 		}
+		logger.Info("Initialized PrivateKey wallet", "address", address.Hex())
 	} else {
 		return errors.New("no wallet is configured. Either Fireblocks or PrivateKey wallet should be configured")
 	}