Validate batch merkle root

core/mock/validator.go

@@ -12,27 +12,22 @@ var (
 	ErrChunkLengthMismatch = errors.New("chunk length mismatch")
 )
 
-// MockChunkValidator is a mock implementation of ChunkValidator
-type MockChunkValidator struct {
+// MockDataValidator is a mock implementation of DataValidator
+type MockDataValidator struct {
 	mock.Mock
 }
 
-var _ core.ChunkValidator = (*MockChunkValidator)(nil)
+var _ core.DataValidator = (*MockDataValidator)(nil)
 
-func NewMockChunkValidator() *MockChunkValidator {
-	return &MockChunkValidator{}
+func NewMockDataValidator() *MockDataValidator {
+	return &MockDataValidator{}
 }
 
-func (v *MockChunkValidator) ValidateBatch(blobs []*core.BlobMessage, operatorState *core.OperatorState, pool common.WorkerPool) error {
+func (v *MockDataValidator) ValidateBatch(batchHeader *core.BatchHeader, blobs []*core.BlobMessage, operatorState *core.OperatorState, pool common.WorkerPool) error {
 	args := v.Called(blobs, operatorState, pool)
 	return args.Error(0)
 }
 
-func (v *MockChunkValidator) ValidateBlob(blob *core.BlobMessage, operatorState *core.OperatorState) error {
-	args := v.Called(blob, operatorState)
-	return args.Error(0)
-}
-
-func (v *MockChunkValidator) UpdateOperatorID(operatorID core.OperatorID) {
+func (v *MockDataValidator) UpdateOperatorID(operatorID core.OperatorID) {
 	v.Called(operatorID)
 }

core/serialization_test.go

@@ -193,3 +193,20 @@ func TestHashPubKeyG1(t *testing.T) {
 	hash := eth.HashPubKeyG1(pk)
 	assert.Equal(t, common.Bytes2Hex(hash[:]), "426d1a0363fbdcd0c8d33b643252164057193ca022958fa0da99d9e70c980dd7")
 }
+
+func TestParseOperatorSocket(t *testing.T) {
+	operatorSocket := "localhost:1234;5678"
+	host, dispersalPort, retrievalPort, err := core.ParseOperatorSocket(operatorSocket)
+	assert.NoError(t, err)
+	assert.Equal(t, "localhost", host)
+	assert.Equal(t, "1234", dispersalPort)
+	assert.Equal(t, "5678", retrievalPort)
+
+	_, _, _, err = core.ParseOperatorSocket("localhost:12345678")
+	assert.NotNil(t, err)
+	assert.Equal(t, "invalid socket address format, missing retrieval port: localhost:12345678", err.Error())
+
+	_, _, _, err = core.ParseOperatorSocket("localhost1234;5678")
+	assert.NotNil(t, err)
+	assert.Equal(t, "invalid socket address format: localhost1234;5678", err.Error())
+}

core/test/core_test.go

@@ -83,198 +83,171 @@ func makeTestBlob(t *testing.T, length int, securityParams []*core.SecurityParam
 
 // prepareBatch takes in multiple blob, encodes them, generates the associated assignments, and the batch header.
 // These are the products that a disperser will need in order to disperse data to the DA nodes.
-func prepareBatch(t *testing.T, cst core.IndexedChainState, blobs []core.Blob, quorumIndex uint, bn uint) ([]core.EncodedBlob, core.BatchHeader) {
+func prepareBatch(t *testing.T, operatorCount uint, blobs []core.Blob, bn uint) ([]core.EncodedBlob, core.BatchHeader, *mock.ChainDataMock) {
+
+	cst, err := mock.MakeChainDataMock(core.OperatorIndex(operatorCount))
+	assert.NoError(t, err)
 
 	batchHeader := core.BatchHeader{
 		ReferenceBlockNumber: bn,
 		BatchRoot:            [32]byte{},
 	}
 
 	numBlob := len(blobs)
-	var encodedBlobs []core.EncodedBlob = make([]core.EncodedBlob, numBlob)
+	encodedBlobs := make([]core.EncodedBlob, numBlob)
+	blobHeaders := make([]*core.BlobHeader, numBlob)
 
 	for z, blob := range blobs {
-		quorumID := blob.RequestHeader.SecurityParams[quorumIndex].QuorumID
-		quorums := []core.QuorumID{quorumID}
 
-		state, err := cst.GetOperatorState(context.Background(), bn, quorums)
-		if err != nil {
-			t.Fatal(err)
+		blobHeader := &core.BlobHeader{
+			QuorumInfos: make([]*core.BlobQuorumInfo, 0),
 		}
+		blobHeaders[z] = blobHeader
 
-		blobSize := uint(len(blob.Data))
-		blobLength := encoding.GetBlobLength(blobSize)
-
-		chunkLength, err := asn.CalculateChunkLength(state, blobLength, 0, blob.RequestHeader.SecurityParams[quorumIndex])
-		if err != nil {
-			t.Fatal(err)
+		encodedBlob := core.EncodedBlob{
+			BlobHeader:        blobHeader,
+			BundlesByOperator: make(map[core.OperatorID]core.Bundles),
 		}
+		encodedBlobs[z] = encodedBlob
 
-		quorumHeader := &core.BlobQuorumInfo{
-			SecurityParam: core.SecurityParam{
-				QuorumID:           quorumID,
-				AdversaryThreshold: blob.RequestHeader.SecurityParams[quorumIndex].AdversaryThreshold,
-				QuorumThreshold:    blob.RequestHeader.SecurityParams[quorumIndex].QuorumThreshold,
-			},
-			ChunkLength: chunkLength,
-		}
+		for _, securityParam := range blob.RequestHeader.SecurityParams {
 
-		assignments, info, err := asn.GetAssignments(state, blobLength, quorumHeader)
-		if err != nil {
-			t.Fatal(err)
-		}
+			quorumID := securityParam.QuorumID
+			quorums := []core.QuorumID{quorumID}
 
-		params := encoding.ParamsFromMins(chunkLength, info.TotalChunks)
+			state, err := cst.GetOperatorState(context.Background(), bn, quorums)
+			if err != nil {
+				t.Fatal(err)
+			}
 
-		commitments, chunks, err := p.EncodeAndProve(blob.Data, params)
-		if err != nil {
-			t.Fatal(err)
-		}
+			blobSize := uint(len(blob.Data))
+			blobLength := encoding.GetBlobLength(blobSize)
 
-		blobHeader := &core.BlobHeader{
-			BlobCommitments: encoding.BlobCommitments{
+			chunkLength, err := asn.CalculateChunkLength(state, blobLength, 0, securityParam)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			quorumHeader := &core.BlobQuorumInfo{
+				SecurityParam: core.SecurityParam{
+					QuorumID:           quorumID,
+					AdversaryThreshold: securityParam.AdversaryThreshold,
+					QuorumThreshold:    securityParam.QuorumThreshold,
+				},
+				ChunkLength: chunkLength,
+			}
+
+			assignments, info, err := asn.GetAssignments(state, blobLength, quorumHeader)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			params := encoding.ParamsFromMins(chunkLength, info.TotalChunks)
+
+			commitments, chunks, err := p.EncodeAndProve(blob.Data, params)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			blobHeader.BlobCommitments = encoding.BlobCommitments{
 				Commitment:       commitments.Commitment,
 				LengthCommitment: commitments.LengthCommitment,
 				LengthProof:      commitments.LengthProof,
 				Length:           commitments.Length,
-			},
-			QuorumInfos: []*core.BlobQuorumInfo{quorumHeader},
-		}
-
-		encodedBlob := core.EncodedBlob{
-			BundlesByOperator: make(map[core.OperatorID]core.Bundles),
-		}
-		for id, assignment := range assignments {
-			bundles := map[core.QuorumID]core.Bundle{
-				quorumID: chunks[assignment.StartIndex : assignment.StartIndex+assignment.NumChunks],
 			}
-			encodedBlob.BlobHeader = blobHeader
-			encodedBlob.BundlesByOperator[id] = bundles
-		}
-		encodedBlobs[z] = encodedBlob
 
-	}
+			blobHeader.QuorumInfos = append(blobHeader.QuorumInfos, quorumHeader)
 
-	return encodedBlobs, batchHeader
+			for id, assignment := range assignments {
+				_, ok := encodedBlob.BundlesByOperator[id]
+				if !ok {
+					encodedBlob.BundlesByOperator[id] = map[core.QuorumID]core.Bundle{
+						quorumID: chunks[assignment.StartIndex : assignment.StartIndex+assignment.NumChunks],
+					}
+				} else {
+					encodedBlob.BundlesByOperator[id][quorumID] = chunks[assignment.StartIndex : assignment.StartIndex+assignment.NumChunks]
+				}
+			}
 
-}
+		}
 
-// checkBatch runs the verification logic for each DA node in the current OperatorState, and returns an error if any of
-// the DA nodes' validation checks fails
-func checkBatch(t *testing.T, cst core.IndexedChainState, encodedBlob core.EncodedBlob, header core.BatchHeader) {
-	val := core.NewChunkValidator(v, asn, cst, [32]byte{})
+	}
 
-	quorums := []core.QuorumID{0}
-	state, _ := cst.GetIndexedOperatorState(context.Background(), header.ReferenceBlockNumber, quorums)
+	// Set the batch root
 
-	for id := range state.IndexedOperators {
-		val.UpdateOperatorID(id)
-		err := val.ValidateBlob(&core.BlobMessage{
-			BlobHeader: encodedBlob.BlobHeader,
-			Bundles:    encodedBlob.BundlesByOperator[id],
-		}, state.OperatorState)
-		assert.NoError(t, err)
+	_, err = batchHeader.SetBatchRoot(blobHeaders)
+	if err != nil {
+		t.Fatal(err)
 	}
 
+	return encodedBlobs, batchHeader, cst
+
 }
 
 // checkBatchByUniversalVerifier runs the verification logic for each DA node in the current OperatorState, and returns an error if any of
 // the DA nodes' validation checks fails
 func checkBatchByUniversalVerifier(t *testing.T, cst core.IndexedChainState, encodedBlobs []core.EncodedBlob, header core.BatchHeader, pool common.WorkerPool) {
-	val := core.NewChunkValidator(v, asn, cst, [32]byte{})
+	val := core.NewDataValidator(v, asn, cst, [32]byte{})
 
-	quorums := []core.QuorumID{0}
+	quorums := []core.QuorumID{0, 1}
 	state, _ := cst.GetIndexedOperatorState(context.Background(), header.ReferenceBlockNumber, quorums)
 	numBlob := len(encodedBlobs)
 
 	for id := range state.IndexedOperators {
 		val.UpdateOperatorID(id)
-		var blobMessages []*core.BlobMessage = make([]*core.BlobMessage, numBlob)
+		blobMessages := make([]*core.BlobMessage, numBlob)
 		for z, encodedBlob := range encodedBlobs {
 			blobMessages[z] = &core.BlobMessage{
 				BlobHeader: encodedBlob.BlobHeader,
 				Bundles:    encodedBlob.BundlesByOperator[id],
 			}
 		}
-		err := val.ValidateBatch(blobMessages, state.OperatorState, pool)
+		err := val.ValidateBatch(&header, blobMessages, state.OperatorState, pool)
 		assert.NoError(t, err)
 	}
 
 }
 
 func TestCoreLibrary(t *testing.T) {
 
-	numBlob := 1 // must be greater than 0
-	blobLengths := []int{1, 64, 1000}
 	operatorCounts := []uint{1, 2, 4, 10, 30}
 
+	numBlob := 3 // must be greater than 0
+	blobLengths := []int{1, 64, 1000}
+
 	securityParams := []*core.SecurityParam{
 		{
 			QuorumID:           0,
 			AdversaryThreshold: 50,
 			QuorumThreshold:    100,
 		},
 		{
-			QuorumID:           0,
+			QuorumID:           1,
 			AdversaryThreshold: 80,
 			QuorumThreshold:    90,
 		},
 	}
 
-	quorumIndex := uint(0)
 	bn := uint(0)
 
 	pool := workerpool.New(1)
 
 	for _, operatorCount := range operatorCounts {
-		cst, err := mock.MakeChainDataMock(core.OperatorIndex(operatorCount))
-		assert.NoError(t, err)
-		batches := make([]core.EncodedBlob, 0)
-		batchHeader := core.BatchHeader{
-			ReferenceBlockNumber: bn,
-			BatchRoot:            [32]byte{},
-		}
+
 		// batch can only be tested per operatorCount, because the assignment would be wrong otherwise
+		blobs := make([]core.Blob, 0)
 		for _, blobLength := range blobLengths {
-
-			for _, securityParam := range securityParams {
-
-				t.Run(fmt.Sprintf("blobLength=%v, operatorCount=%v, securityParams=%v", blobLength, operatorCount, securityParam), func(t *testing.T) {
-
-					blobs := make([]core.Blob, numBlob)
-					for i := 0; i < numBlob; i++ {
-						blobs[i] = makeTestBlob(t, blobLength, []*core.SecurityParam{securityParam})
-					}
-
-					batch, header := prepareBatch(t, cst, blobs, quorumIndex, bn)
-					batches = append(batches, batch...)
-
-					checkBatch(t, cst, batch[0], header)
-				})
+			for i := 0; i < numBlob; i++ {
+				blobs = append(blobs, makeTestBlob(t, blobLength, securityParams))
 			}
-
 		}
-		t.Run(fmt.Sprintf("universal verifier operatorCount=%v over %v blobs", operatorCount, len(batches)), func(t *testing.T) {
-			checkBatchByUniversalVerifier(t, cst, batches, batchHeader, pool)
-		})
-
-	}
 
-}
+		blobMessages, header, cst := prepareBatch(t, operatorCount, blobs, bn)
 
-func TestParseOperatorSocket(t *testing.T) {
-	operatorSocket := "localhost:1234;5678"
-	host, dispersalPort, retrievalPort, err := core.ParseOperatorSocket(operatorSocket)
-	assert.NoError(t, err)
-	assert.Equal(t, "localhost", host)
-	assert.Equal(t, "1234", dispersalPort)
-	assert.Equal(t, "5678", retrievalPort)
+		t.Run(fmt.Sprintf("universal verifier operatorCount=%v over %v blobs", operatorCount, len(blobs)), func(t *testing.T) {
+			checkBatchByUniversalVerifier(t, cst, blobMessages, header, pool)
+		})
 
-	_, _, _, err = core.ParseOperatorSocket("localhost:12345678")
-	assert.NotNil(t, err)
-	assert.Equal(t, "invalid socket address format, missing retrieval port: localhost:12345678", err.Error())
+	}
 
-	_, _, _, err = core.ParseOperatorSocket("localhost1234;5678")
-	assert.NotNil(t, err)
-	assert.Equal(t, "invalid socket address format: localhost1234;5678", err.Error())
 }