[deliveryorderer] Process headers-only blocks only in the headers stream (#633)

#### Type of change

- Bug fix
 
#### Description

- Verify that all block fields exists as expected, for all delivery
methods
- Process blocks without data in the headers-only stream, regardless of
their data source.

#### Related issues

- resolves #632

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

service/sidecar/sidecar.go

@@ -573,7 +573,8 @@ func (s *Service) deliverBlocks(
 		if err != nil {
 			return common.Status_INTERNAL_SERVER_ERROR, errors.Wrap(err, "error sending response")
 		}
-		logger.Infof("Successfully sent block %d:%d to client.", block.Header.Number, len(block.Data.Data))
+		logger.Infof("Successfully sent block [%d] (size: %d) to client.",
+			block.Header.Number, len(block.Data.Data))
 
 		if stopNum == block.Header.Number {
 			break

utils/deliver/delivery.go

@@ -120,8 +120,8 @@ func toQueueWithoutReconnect(ctx context.Context, p *Parameters) error {
 		// We make minimal verifications to ensure we receive blocks in order.
 		// This allows us to restart the connection from the next expected block.
 		// We restart the connection upon failure.
-		if block == nil || block.Header == nil {
-			return errors.Join(backoff, errors.New("received nil block or with nil header"))
+		if block == nil || block.Header == nil || block.Metadata == nil || (!p.HeaderOnly && block.Data == nil) {
+			return errors.Join(backoff, errors.New("received nil block, header, metadata, or data"))
 		}
 		if block.Header.Number != p.NextBlockNum {
 			return errors.Join(backoff, errors.Errorf("received block number %d != %d (expected)",

utils/deliver/delivery_test.go

@@ -38,7 +38,11 @@ func TestDeliverToChannel(t *testing.T) {
 	// 10 times for the pulled blocks, 10 times for the blocks in the buffer, and 1 that blocks because
 	// the buffer is full.
 	s.EXPECT().RecvBlockOrStatus().MaxTimes(21).DoAndReturn(func() (*common.Block, *common.Status, error) {
-		b := &common.Block{Header: &common.BlockHeader{Number: nextBlockNum}}
+		b := &common.Block{
+			Header:   &common.BlockHeader{Number: nextBlockNum},
+			Metadata: &common.BlockMetadata{},
+			Data:     &common.BlockData{},
+		}
 		nextBlockNum++
 		return b, nil, nil
 	})
@@ -48,7 +52,11 @@ func TestDeliverToChannel(t *testing.T) {
 	outputBlock := channel.NewReader(t.Context(), e.outputBlock)
 	outputBlockWithSourceID := channel.NewReader(t.Context(), e.outputBlockWithSourceID)
 	for i := range uint64(10) {
-		expectedBlock := &common.Block{Header: &common.BlockHeader{Number: i}}
+		expectedBlock := &common.Block{
+			Header:   &common.BlockHeader{Number: i},
+			Metadata: &common.BlockMetadata{},
+			Data:     &common.BlockData{},
+		}
 
 		readBlock, ok := outputBlock.ReadWithTimeout(3 * time.Second)
 		require.True(t, ok)
@@ -109,23 +117,56 @@ func TestDeliverToChannelFailedStreamer(t *testing.T) {
 	t.Log("Receive bad block (nil header)")
 	s = NewMockStreamer(e.ctrl)
 	s.EXPECT().Send(gomock.Not(gomock.Nil())).Times(1).Return(nil)
-	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{}, nil, nil)
+	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{
+		Metadata: &common.BlockMetadata{},
+		Data:     &common.BlockData{},
+	}, nil, nil)
+	streamerQueue.Write(s)
+	_, ok = outputBlock.ReadWithTimeout(time.Second)
+	require.False(t, ok)
+
+	t.Log("Receive bad block (nil metadata)")
+	s = NewMockStreamer(e.ctrl)
+	s.EXPECT().Send(gomock.Not(gomock.Nil())).Times(1).Return(nil)
+	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{
+		Header: &common.BlockHeader{Number: 1},
+		Data:   &common.BlockData{},
+	}, nil, nil)
+	streamerQueue.Write(s)
+	_, ok = outputBlock.ReadWithTimeout(time.Second)
+	require.False(t, ok)
+
+	t.Log("Receive bad block (nil data)")
+	s = NewMockStreamer(e.ctrl)
+	s.EXPECT().Send(gomock.Not(gomock.Nil())).Times(1).Return(nil)
+	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{
+		Header:   &common.BlockHeader{Number: 1},
+		Metadata: &common.BlockMetadata{},
+	}, nil, nil)
 	streamerQueue.Write(s)
 	_, ok = outputBlock.ReadWithTimeout(time.Second)
 	require.False(t, ok)
 
 	t.Log("Receive bad block (wrong number 1 != 0)")
 	s = NewMockStreamer(e.ctrl)
 	s.EXPECT().Send(gomock.Not(gomock.Nil())).Times(1).Return(nil)
-	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{Header: &common.BlockHeader{Number: 1}}, nil, nil)
+	s.EXPECT().RecvBlockOrStatus().Times(1).Return(&common.Block{
+		Header:   &common.BlockHeader{Number: 1},
+		Metadata: &common.BlockMetadata{},
+		Data:     &common.BlockData{},
+	}, nil, nil)
 	streamerQueue.Write(s)
 	_, ok = outputBlock.ReadWithTimeout(time.Second)
 	require.False(t, ok)
 
 	t.Log("Correct block, then bad block (wrong number 0 != 1)")
 	s = NewMockStreamer(e.ctrl)
 	s.EXPECT().Send(gomock.Not(gomock.Nil())).Times(1).Return(nil)
-	correctBlock := &common.Block{Header: &common.BlockHeader{Number: 0}}
+	correctBlock := &common.Block{
+		Header:   &common.BlockHeader{Number: 0},
+		Metadata: &common.BlockMetadata{},
+		Data:     &common.BlockData{},
+	}
 	s.EXPECT().RecvBlockOrStatus().Times(2).Return(correctBlock, nil, nil)
 	streamerQueue.Write(s)
 	rb, ok := outputBlock.ReadWithTimeout(3 * time.Second)

utils/deliverorderer/orderer.go

@@ -280,12 +280,10 @@ func (d *ftDelivery) readNextBlock(ctx context.Context) *deliver.BlockWithSource
 }
 
 func (d *ftDelivery) getProcessingState(block *deliver.BlockWithSourceID) *blockVerificationStateMachine {
-	switch block.SourceID {
-	case d.curDataBlockSourceID:
+	if block.Block.Data != nil && block.SourceID == d.curDataBlockSourceID {
 		return &d.dataStream
-	default:
-		return &d.headerOnlyStream
 	}
+	return &d.headerOnlyStream
 }
 
 func (d *ftDelivery) checkNextStreamAction(state *blockVerificationStateMachine) error {

utils/deliverorderer/orderer_test.go

@@ -442,7 +442,9 @@ func TestOrdererDeliverBFT(t *testing.T) {
 			p1.HoldFromBlock.Store(expectedBlockNum)
 			p2.HoldFromBlock.Store(expectedBlockNum + 1)
 			p0.ReplaceBlock.Store(expectedBlockNum, &common.Block{
-				Header: &common.BlockHeader{Number: expectedBlockNum},
+				Header:   &common.BlockHeader{Number: expectedBlockNum},
+				Data:     &common.BlockData{},
+				Metadata: &common.BlockMetadata{},
 			})
 			b = e.submit(t)
 			require.EqualValues(t, 2, b.SourceID)

utils/deliverorderer/verify.go

@@ -150,6 +150,9 @@ func (s *blockVerificationStateMachine) verificationStepAndUpdateState(blk *deli
 // verifyBlockForm checks whether the block is well-formed.
 // It is only used internally.
 func (s *blockVerificationStateMachine) verifyBlockForm(block *common.Block) bool {
+	// The delivery already filters nil blocks, headers, metadata, or data.
+	// We only need to verify that the metadata is with valid form.
+	// We verify everything anyway to ensure correctness.
 	if block == nil || block.Header == nil || block.Metadata == nil {
 		return false
 	}