[5/N][multi quorum ejection] Compute num batches per quorum and block…

… range (#327)

disperser/dataapi/nonsigner_utils.go

@@ -1,8 +1,26 @@
 package dataapi
 
-import "fmt"
+import (
+	"fmt"
+	"sort"
+)
 
-// Representing an interval [StartBlock, EndBlock] (inclusive).
+// NumBatchesAtBlock represents the number of batches at current block.
+type NumBatchesAtBlock struct {
+	BlockNumber uint32
+	NumBatches  int
+}
+
+// QueryBatches represents number of batches at different block numbers, as well
+// as accumulated number of batches from the first block in NumBatches.
+// The NumBatches is in ascending order by NumBatchesAtBlock.BlockNumber, and
+// AccuBatches[i] is corresponding to NumBatches[i].
+type QueryBatches struct {
+	NumBatches  []*NumBatchesAtBlock
+	AccuBatches []int
+}
+
+// BlockInterval represents an interval [StartBlock, EndBlock] (inclusive).
 type BlockInterval struct {
 	StartBlock uint32
 	EndBlock   uint32
@@ -44,7 +62,7 @@ func (oqi OperatorQuorumIntervals) GetQuorums(operatorId string, blockNum uint32
 //     Requires: startBlock <= endBlock.
 //   - operatorInitialQuorum: the initial quorums at startBlock that operators were
 //     registered in.
-//     Requires: operatorInitialQuorum[op] is non-empty for each operator "op".
+//     Requires: operatorInitialQuorum contains all operators of interest (caller to ensure).
 //   - addedToQuorum, removedFromQuorum: a sequence of events that added/removed operators
 //     to/from quorums.
 //     Requires:
@@ -66,9 +84,6 @@ func CreateOperatorQuorumIntervals(
 	addedToQuorumErr := "cannot add operator %s to quorum %d at block number %d, " +
 		"the operator is already in the quorum since block number %d"
 	for op, initialQuorums := range operatorInitialQuorum {
-		if len(initialQuorums) == 0 {
-			return nil, fmt.Errorf("operator %s must be in at least one quorum at block %d", op, startBlock)
-		}
 		operatorQuorumIntervals[op] = make(map[uint8][]BlockInterval)
 		openQuorum := make(map[uint8]uint32)
 		for _, q := range initialQuorums {
@@ -180,3 +195,89 @@ func validateQuorumEvents(added []*OperatorQuorum, removed []*OperatorQuorum, st
 	}
 	return validate(removed)
 }
+
+// ComputeNumBatches returns the number of batches in the block interval [startBlock, endBlock].
+func ComputeNumBatches(queryBatches *QueryBatches, startBlock, endBlock uint32) int {
+	start := getLowerBoundIndex(queryBatches.NumBatches, startBlock)
+	end := getUpperBoundIndex(queryBatches.NumBatches, endBlock)
+	num := 0
+	if end > 0 {
+		num = queryBatches.AccuBatches[end-1]
+	}
+	if start > 0 {
+		num = num - queryBatches.AccuBatches[start-1]
+	}
+	return num
+}
+
+// CreatQuorumBatches returns quorumBatches, where quorumBatches[q] is a list of
+// QueryBatches in ascending order by block number.
+func CreatQuorumBatches(batches []*BatchNonSigningInfo) map[uint8]*QueryBatches {
+	quorumBatchMap := make(map[uint8]map[uint32]int)
+	for _, batch := range batches {
+		for _, q := range batch.QuorumNumbers {
+			if _, ok := quorumBatchMap[q]; !ok {
+				quorumBatchMap[q] = make(map[uint32]int)
+			}
+			quorumBatchMap[q][batch.ReferenceBlockNumber]++
+		}
+	}
+	quorumBatches := make(map[uint8]*QueryBatches)
+	for q, s := range quorumBatchMap {
+		numBatches := make([]*NumBatchesAtBlock, 0)
+		for block, num := range s {
+			element := &NumBatchesAtBlock{
+				BlockNumber: block,
+				NumBatches:  num,
+			}
+			numBatches = append(numBatches, element)
+		}
+		sort.SliceStable(numBatches, func(i, j int) bool {
+			// note: since it's created from a map with block number as key, all block
+			// numbers are different.
+			return numBatches[i].BlockNumber < numBatches[j].BlockNumber
+		})
+		accuBatches := make([]int, len(numBatches))
+		if len(numBatches) > 0 {
+			accuBatches[0] = numBatches[0].NumBatches
+		}
+		for i := 1; i < len(numBatches); i++ {
+			accuBatches[i] = numBatches[i].NumBatches + accuBatches[i-1]
+		}
+		quorumBatches[q] = &QueryBatches{
+			NumBatches:  numBatches,
+			AccuBatches: accuBatches,
+		}
+	}
+	return quorumBatches
+}
+
+// getLowerBoundIndex returns the index of the first element intervals[i] where the
+// intervals[i].BlockNumber is no less than the given "blockNum".
+func getLowerBoundIndex(intervals []*NumBatchesAtBlock, blockNum uint32) int {
+	low, high := 0, len(intervals)-1
+	for low <= high {
+		mid := low + (high-low)/2
+		if intervals[mid].BlockNumber < blockNum {
+			low = mid + 1
+		} else {
+			high = mid - 1
+		}
+	}
+	return high + 1
+}
+
+// getUpperBoundIndex returns the index of the first element intervals[i] where the
+// intervals[i].BlockNumber is greater than the given "blockNum".
+func getUpperBoundIndex(intervals []*NumBatchesAtBlock, blockNum uint32) int {
+	low, high := 0, len(intervals)-1
+	for low <= high {
+		mid := low + (high-low)/2
+		if intervals[mid].BlockNumber <= blockNum {
+			low = mid + 1
+		} else {
+			high = mid - 1
+		}
+	}
+	return high + 1
+}

disperser/dataapi/nonsigner_utils_test.go

@@ -19,21 +19,12 @@ func TestCreateOperatorQuorumIntervalsWithInvalidArgs(t *testing.T) {
 	addedQuorums := map[string][]*dataapi.OperatorQuorum{}
 	removedQuorums := map[string][]*dataapi.OperatorQuorum{}
 
-	// Empty initial quorums
-	operatorInitialQuorum := map[string][]uint8{
-		"operator-1": {},
-		"operator-2": {0x01},
-	}
-	_, err := dataapi.CreateOperatorQuorumIntervals(10, 25, operatorInitialQuorum, addedQuorums, removedQuorums)
-	assert.Error(t, err)
-	assert.True(t, strings.Contains(err.Error(), "must be in at least one quorum"))
-
 	// StartBlock > EndBlock
-	operatorInitialQuorum = map[string][]uint8{
+	operatorInitialQuorum := map[string][]uint8{
 		"operator-1": {0x00},
 		"operator-2": {0x00},
 	}
-	_, err = dataapi.CreateOperatorQuorumIntervals(100, 25, operatorInitialQuorum, addedQuorums, removedQuorums)
+	_, err := dataapi.CreateOperatorQuorumIntervals(100, 25, operatorInitialQuorum, addedQuorums, removedQuorums)
 	assert.Error(t, err)
 	assert.True(t, strings.Contains(err.Error(), "endBlock must be no less than startBlock"))
 
@@ -389,3 +380,108 @@ func TestCreateOperatorQuorumIntervals(t *testing.T) {
 	assert.ElementsMatch(t, []uint8{0x00, 0x02}, quorumIntervals.GetQuorums("operator-2", 22))
 	assert.ElementsMatch(t, []uint8{0x00, 0x02}, quorumIntervals.GetQuorums("operator-2", 25))
 }
+
+func TestComputeNumBatches(t *testing.T) {
+	queryBatches := &dataapi.QueryBatches{
+		NumBatches:  []*dataapi.NumBatchesAtBlock{},
+		AccuBatches: []int{},
+	}
+	assert.Equal(t, 0, dataapi.ComputeNumBatches(queryBatches, 1, 4))
+
+	numBatches := []*dataapi.NumBatchesAtBlock{
+		{
+			BlockNumber: 5,
+			NumBatches:  2,
+		},
+	}
+	queryBatches = &dataapi.QueryBatches{
+		NumBatches:  numBatches,
+		AccuBatches: []int{2},
+	}
+	assert.Equal(t, 0, dataapi.ComputeNumBatches(queryBatches, 1, 4))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 1, 5))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 5, 5))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 5, 6))
+
+	numBatches = []*dataapi.NumBatchesAtBlock{
+		{
+			BlockNumber: 5,
+			NumBatches:  2,
+		},
+		{
+			BlockNumber: 10,
+			NumBatches:  2,
+		},
+		{
+			BlockNumber: 15,
+			NumBatches:  2,
+		},
+		{
+			BlockNumber: 20,
+			NumBatches:  2,
+		},
+	}
+	queryBatches = &dataapi.QueryBatches{
+		NumBatches:  numBatches,
+		AccuBatches: []int{2, 4, 6, 8},
+	}
+
+	assert.Equal(t, 0, dataapi.ComputeNumBatches(queryBatches, 1, 4))
+	assert.Equal(t, 0, dataapi.ComputeNumBatches(queryBatches, 21, 22))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 1, 5))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 5, 5))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 5, 9))
+	assert.Equal(t, 4, dataapi.ComputeNumBatches(queryBatches, 5, 10))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 6, 10))
+	assert.Equal(t, 4, dataapi.ComputeNumBatches(queryBatches, 5, 14))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 6, 14))
+	assert.Equal(t, 6, dataapi.ComputeNumBatches(queryBatches, 5, 15))
+	assert.Equal(t, 8, dataapi.ComputeNumBatches(queryBatches, 5, 20))
+	assert.Equal(t, 8, dataapi.ComputeNumBatches(queryBatches, 5, 22))
+	assert.Equal(t, 8, dataapi.ComputeNumBatches(queryBatches, 1, 22))
+	assert.Equal(t, 6, dataapi.ComputeNumBatches(queryBatches, 6, 22))
+	assert.Equal(t, 4, dataapi.ComputeNumBatches(queryBatches, 11, 22))
+	assert.Equal(t, 2, dataapi.ComputeNumBatches(queryBatches, 16, 22))
+}
+
+func TestCreatQuorumBatches(t *testing.T) {
+	// The nonsigning info for a list of batches.
+	batchNonSigningInfo := []*dataapi.BatchNonSigningInfo{
+		{
+			QuorumNumbers:        []uint8{0, 1},
+			ReferenceBlockNumber: 2,
+		},
+		{
+			QuorumNumbers:        []uint8{0},
+			ReferenceBlockNumber: 2,
+		},
+		{
+			QuorumNumbers:        []uint8{1, 2},
+			ReferenceBlockNumber: 4,
+		},
+	}
+
+	quorumBatches := dataapi.CreatQuorumBatches(batchNonSigningInfo)
+
+	assert.Equal(t, 3, len(quorumBatches))
+
+	q0, ok := quorumBatches[0]
+	assert.True(t, ok)
+	assert.Equal(t, 1, len(q0.NumBatches))
+	assert.Equal(t, uint32(2), q0.NumBatches[0].BlockNumber)
+	assert.Equal(t, 2, q0.AccuBatches[0])
+
+	q1, ok := quorumBatches[1]
+	assert.True(t, ok)
+	assert.Equal(t, 2, len(q1.NumBatches))
+	assert.Equal(t, uint32(2), q1.NumBatches[0].BlockNumber)
+	assert.Equal(t, 1, q1.AccuBatches[0])
+	assert.Equal(t, uint32(4), q1.NumBatches[1].BlockNumber)
+	assert.Equal(t, 2, q1.AccuBatches[1])
+
+	q2, ok := quorumBatches[2]
+	assert.True(t, ok)
+	assert.Equal(t, 1, len(q2.NumBatches))
+	assert.Equal(t, uint32(4), q2.NumBatches[0].BlockNumber)
+	assert.Equal(t, 1, q2.AccuBatches[0])
+}