FAB-14327 Race in BlockWriter corrupts config sequence

There is a race between the go-routine that submits a config-update
block to the internal thread that writes the (previous) block
asynchronously to the ledger.

The race may cause to back-to-back config-update transactions to have bad
block metadata - in the first the LAST_CONFIG will be ok, but in the
second the LAST_CONFIG will point to the first, rather then the second.
This may happen independently in different ledgers, creating a fork.

This fix avoid Bundle update before the go-routine in WriteBlock()
finished writing the previous block. We do this (in particular) to
prevent bw.support.Sequence() from advancing before the go-routine
in WriteBlock() reads it. In general, this prevents the StableBundle
from changing before the go-routine in WriteBlock() finishes.

The unit test replicates the condtions that cause the bug and verifies
it is fixed (without the fix it fails).

See respective JIRA for more details.

Change-Id: If26654e006b4fe6d48d2d8fc73894eaff2426dcd
Signed-off-by: Yoav Tock <tock@il.ibm.com>

orderer/common/multichannel/blockwriter.go

@@ -131,6 +131,11 @@ func (bw *BlockWriter) WriteConfigBlock(block *cb.Block, encodedMetadataValue []
 			logger.Panicf("Told to write a config block with a new config, but could not convert it to a bundle: %s", err)
 		}
 
+		// Avoid Bundle update before the go-routine in WriteBlock() finished writing the previous block.
+		// We do this (in particular) to prevent bw.support.Sequence() from advancing before the go-routine reads it.
+		// In general, this prevents the StableBundle from changing before the go-routine in WriteBlock() finishes.
+		bw.committingBlock.Lock()
+		bw.committingBlock.Unlock()
 		bw.support.Update(bundle)
 	default:
 		logger.Panicf("Told to write a config block with unknown header type: %v", chdr.Type)

orderer/common/multichannel/blockwriter_test.go

@@ -27,7 +27,9 @@ type mockBlockWriterSupport struct {
 	blockledger.ReadWriter
 }
 
-func (mbws mockBlockWriterSupport) Update(bundle *newchannelconfig.Bundle) {}
+func (mbws mockBlockWriterSupport) Update(bundle *newchannelconfig.Bundle) {
+	mbws.Validator.SequenceVal++
+}
 
 func (mbws mockBlockWriterSupport) CreateBundle(channelID string, config *cb.Config) (*newchannelconfig.Bundle, error) {
 	return nil, nil
@@ -186,3 +188,49 @@ func TestGoodWriteConfig(t *testing.T) {
 	omd := utils.GetMetadataFromBlockOrPanic(block, cb.BlockMetadataIndex_ORDERER)
 	assert.Equal(t, consenterMetadata, omd.Value)
 }
+
+func TestRaceWriteConfig(t *testing.T) {
+	confSys := configtxgentest.Load(genesisconfig.SampleInsecureSoloProfile)
+	genesisBlockSys := encoder.New(confSys).GenesisBlock()
+	_, l := newRAMLedgerAndFactory(10, genesisconfig.TestChainID, genesisBlockSys)
+
+	bw := &BlockWriter{
+		support: &mockBlockWriterSupport{
+			LocalSigner: mockCrypto(),
+			ReadWriter:  l,
+			Validator:   &mockconfigtx.Validator{},
+		},
+	}
+
+	ctx := makeConfigTx(genesisconfig.TestChainID, 1)
+	block1 := cb.NewBlock(1, genesisBlockSys.Header.Hash())
+	block1.Data.Data = [][]byte{utils.MarshalOrPanic(ctx)}
+	consenterMetadata1 := []byte("foo")
+
+	ctx = makeConfigTx(genesisconfig.TestChainID, 1)
+	block2 := cb.NewBlock(2, block1.Header.Hash())
+	block2.Data.Data = [][]byte{utils.MarshalOrPanic(ctx)}
+	consenterMetadata2 := []byte("bar")
+
+	bw.WriteConfigBlock(block1, consenterMetadata1)
+	bw.WriteConfigBlock(block2, consenterMetadata2)
+
+	// Wait for the commit to complete
+	bw.committingBlock.Lock()
+	bw.committingBlock.Unlock()
+
+	cBlock := blockledger.GetBlock(l, block1.Header.Number)
+	assert.Equal(t, block1.Header, cBlock.Header)
+	assert.Equal(t, block1.Data, cBlock.Data)
+	expectedLastConfigBlockNumber := block1.Header.Number
+	testLastConfigBlockNumber(t, block1, expectedLastConfigBlockNumber)
+
+	cBlock = blockledger.GetBlock(l, block2.Header.Number)
+	assert.Equal(t, block2.Header, cBlock.Header)
+	assert.Equal(t, block2.Data, cBlock.Data)
+	expectedLastConfigBlockNumber = block2.Header.Number
+	testLastConfigBlockNumber(t, block2, expectedLastConfigBlockNumber)
+
+	omd := utils.GetMetadataFromBlockOrPanic(block1, cb.BlockMetadataIndex_ORDERER)
+	assert.Equal(t, consenterMetadata1, omd.Value)
+}