FAB-10521 Block-cutter should refetch config

The block-cutter is currently instantiated with a reference to the
orderer channel config.  Once upon a time, each config element was its
own mutable state machine, so this worked fine for v1.0.x.  However, in
v1.1, the individual mutable configs were transformed to be a single
immutable config bundle.  When the config changes, an entirely new
immutable bundle is created and atomically swapped out.

Because of this, the block-cutter only retains the reference to the
initial immutable blob.  It needs to be fixed to pull a fresh reference
of the config through the atomic dereference at every block.  This CR
does just that.

Change-Id: I6be1b65cb8d8cc86d5e47da906a127da6651354e
Signed-off-by: Jason Yellick <jyellick@us.ibm.com>

Author: Jason Yellick

orderer/common/blockcutter/blockcutter.go

@@ -22,6 +22,10 @@ func init() {
 	logger = flogging.MustGetLogger(pkgLogID)
 }
 
+type OrdererConfigFetcher interface {
+	OrdererConfig() (channelconfig.Orderer, bool)
+}
+
 // Receiver defines a sink for the ordered broadcast messages
 type Receiver interface {
 	// Ordered should be invoked sequentially as messages are ordered
@@ -35,15 +39,15 @@ type Receiver interface {
 }
 
 type receiver struct {
-	sharedConfigManager   channelconfig.Orderer
+	sharedConfigFetcher   OrdererConfigFetcher
 	pendingBatch          []*cb.Envelope
 	pendingBatchSizeBytes uint32
 }
 
 // NewReceiverImpl creates a Receiver implementation based on the given configtxorderer manager
-func NewReceiverImpl(sharedConfigManager channelconfig.Orderer) Receiver {
+func NewReceiverImpl(sharedConfigFetcher OrdererConfigFetcher) Receiver {
 	return &receiver{
-		sharedConfigManager: sharedConfigManager,
+		sharedConfigFetcher: sharedConfigFetcher,
 	}
 }
 
@@ -64,9 +68,15 @@ func NewReceiverImpl(sharedConfigManager channelconfig.Orderer) Receiver {
 //
 // Note that messageBatches can not be greater than 2.
 func (r *receiver) Ordered(msg *cb.Envelope) (messageBatches [][]*cb.Envelope, pending bool) {
+	ordererConfig, ok := r.sharedConfigFetcher.OrdererConfig()
+	if !ok {
+		logger.Panicf("Could not retrieve orderer config to query batch parameters, block cutting is not possible")
+	}
+	batchSize := ordererConfig.BatchSize()
+
 	messageSizeBytes := messageSizeBytes(msg)
-	if messageSizeBytes > r.sharedConfigManager.BatchSize().PreferredMaxBytes {
-		logger.Debugf("The current message, with %v bytes, is larger than the preferred batch size of %v bytes and will be isolated.", messageSizeBytes, r.sharedConfigManager.BatchSize().PreferredMaxBytes)
+	if messageSizeBytes > batchSize.PreferredMaxBytes {
+		logger.Debugf("The current message, with %v bytes, is larger than the preferred batch size of %v bytes and will be isolated.", messageSizeBytes, batchSize.PreferredMaxBytes)
 
 		// cut pending batch, if it has any messages
 		if len(r.pendingBatch) > 0 {
@@ -80,7 +90,7 @@ func (r *receiver) Ordered(msg *cb.Envelope) (messageBatches [][]*cb.Envelope, p
 		return
 	}
 
-	messageWillOverflowBatchSizeBytes := r.pendingBatchSizeBytes+messageSizeBytes > r.sharedConfigManager.BatchSize().PreferredMaxBytes
+	messageWillOverflowBatchSizeBytes := r.pendingBatchSizeBytes+messageSizeBytes > batchSize.PreferredMaxBytes
 
 	if messageWillOverflowBatchSizeBytes {
 		logger.Debugf("The current message, with %v bytes, will overflow the pending batch of %v bytes.", messageSizeBytes, r.pendingBatchSizeBytes)
@@ -94,7 +104,7 @@ func (r *receiver) Ordered(msg *cb.Envelope) (messageBatches [][]*cb.Envelope, p
 	r.pendingBatchSizeBytes += messageSizeBytes
 	pending = true
 
-	if uint32(len(r.pendingBatch)) >= r.sharedConfigManager.BatchSize().MaxMessageCount {
+	if uint32(len(r.pendingBatch)) >= batchSize.MaxMessageCount {
 		logger.Debugf("Batch size met, cutting batch")
 		messageBatch := r.Cut()
 		messageBatches = append(messageBatches, messageBatch)

orderer/common/blockcutter/blockcutter_test.go

@@ -1,32 +1,33 @@
 /*
-Copyright IBM Corp. 2016 All Rights Reserved.
+Copyright IBM Corp. All Rights Reserved.
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-                 http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package blockcutter
 
 import (
 	"testing"
 
-	mockconfig "github.com/hyperledger/fabric/common/mocks/config"
+	"github.com/hyperledger/fabric/common/channelconfig"
+	"github.com/hyperledger/fabric/orderer/common/blockcutter/mock"
 	cb "github.com/hyperledger/fabric/protos/common"
 	ab "github.com/hyperledger/fabric/protos/orderer"
 
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/stretchr/testify/assert"
 )
 
+//go:generate counterfeiter -o mock/config_fetcher.go --fake-name OrdererConfigFetcher . ordererConfigFetcher
+type ordererConfigFetcher interface {
+	OrdererConfigFetcher
+}
+
+//go:generate counterfeiter -o mock/orderer_config.go --fake-name OrdererConfig . ordererConfig
+type ordererConfig interface {
+	channelconfig.Orderer
+}
+
 func init() {
 	flogging.SetModuleLevel(pkgLogID, "DEBUG")
 }
@@ -38,7 +39,17 @@ func TestNormalBatch(t *testing.T) {
 	maxMessageCount := uint32(2)
 	absoluteMaxBytes := uint32(1000)
 	preferredMaxBytes := uint32(100)
-	r := NewReceiverImpl(&mockconfig.Orderer{BatchSizeVal: &ab.BatchSize{MaxMessageCount: maxMessageCount, AbsoluteMaxBytes: absoluteMaxBytes, PreferredMaxBytes: preferredMaxBytes}})
+	mockConfig := &mock.OrdererConfig{}
+	mockConfig.BatchSizeReturns(&ab.BatchSize{
+		MaxMessageCount:   maxMessageCount,
+		AbsoluteMaxBytes:  absoluteMaxBytes,
+		PreferredMaxBytes: preferredMaxBytes,
+	})
+
+	mockConfigFetcher := &mock.OrdererConfigFetcher{}
+	mockConfigFetcher.OrdererConfigReturns(mockConfig, true)
+
+	r := NewReceiverImpl(mockConfigFetcher)
 
 	batches, pending := r.Ordered(tx)
 	assert.Nil(t, batches, "Should not have created batch")
@@ -58,7 +69,17 @@ func TestBatchSizePreferredMaxBytesOverflow(t *testing.T) {
 	// set message count > 9
 	maxMessageCount := uint32(20)
 
-	r := NewReceiverImpl(&mockconfig.Orderer{BatchSizeVal: &ab.BatchSize{MaxMessageCount: maxMessageCount, AbsoluteMaxBytes: preferredMaxBytes * 2, PreferredMaxBytes: preferredMaxBytes}})
+	mockConfig := &mock.OrdererConfig{}
+	mockConfig.BatchSizeReturns(&ab.BatchSize{
+		MaxMessageCount:   maxMessageCount,
+		AbsoluteMaxBytes:  preferredMaxBytes * 2,
+		PreferredMaxBytes: preferredMaxBytes,
+	})
+
+	mockConfigFetcher := &mock.OrdererConfigFetcher{}
+	mockConfigFetcher.OrdererConfigReturns(mockConfig, true)
+
+	r := NewReceiverImpl(mockConfigFetcher)
 
 	// enqueue 9 messages
 	for i := 0; i < 9; i++ {
@@ -89,7 +110,17 @@ func TestBatchSizePreferredMaxBytesOverflowNoPending(t *testing.T) {
 	// set message count > 1
 	maxMessageCount := uint32(20)
 
-	r := NewReceiverImpl(&mockconfig.Orderer{BatchSizeVal: &ab.BatchSize{MaxMessageCount: maxMessageCount, AbsoluteMaxBytes: preferredMaxBytes * 3, PreferredMaxBytes: preferredMaxBytes}})
+	mockConfig := &mock.OrdererConfig{}
+	mockConfig.BatchSizeReturns(&ab.BatchSize{
+		MaxMessageCount:   maxMessageCount,
+		AbsoluteMaxBytes:  preferredMaxBytes * 3,
+		PreferredMaxBytes: preferredMaxBytes,
+	})
+
+	mockConfigFetcher := &mock.OrdererConfigFetcher{}
+	mockConfigFetcher.OrdererConfigReturns(mockConfig, true)
+
+	r := NewReceiverImpl(mockConfigFetcher)
 
 	// submit normal message
 	batches, pending := r.Ordered(tx)
@@ -105,3 +136,9 @@ func TestBatchSizePreferredMaxBytesOverflowNoPending(t *testing.T) {
 		assert.Len(t, batch, 1, "Should have had one normal tx in batch %d", i)
 	}
 }
+
+func TestPanicOnMissingConfig(t *testing.T) {
+	mockConfigFetcher := &mock.OrdererConfigFetcher{}
+	r := NewReceiverImpl(mockConfigFetcher)
+	assert.Panics(t, func() { r.Ordered(tx) })
+}