func (blockchain *blockchain) persistRawBlock(block *protos.Block, blockNumber uint64) error {
	blockBytes, blockBytesErr := block.Bytes()
	if blockBytesErr != nil {
		return blockBytesErr
	}
	writeBatch := gorocksdb.NewWriteBatch()
	defer writeBatch.Destroy()
	writeBatch.PutCF(db.GetDBHandle().BlockchainCF, encodeBlockNumberDBKey(blockNumber), blockBytes)

	blockHash, err := block.GetHash()
	if err != nil {
		return err
	}

	// Need to check as we suport out of order blocks in cases such as block/state synchronization. This is
	// really blockchain height, not size.
	if blockchain.getSize() < blockNumber+1 {
		sizeBytes := encodeUint64(blockNumber + 1)
		writeBatch.PutCF(db.GetDBHandle().BlockchainCF, blockCountKey, sizeBytes)
		blockchain.size = blockNumber + 1
		blockchain.previousBlockHash = blockHash
	}

	if blockchain.indexer.isSynchronous() {
		blockchain.indexer.createIndexesSync(block, blockNumber, blockHash, writeBatch)
	}

	opt := gorocksdb.NewDefaultWriteOptions()
	defer opt.Destroy()
	err = db.GetDBHandle().DB.Write(opt, writeBatch)
	if err != nil {
		return err
	}
	return nil
}

func sendProducerBlockEvent(block *protos.Block) {

	// Remove payload from deploy transactions. This is done to make block
	// events more lightweight as the payload for these types of transactions
	// can be very large.
	blockTransactions := block.GetTransactions()
	for _, transaction := range blockTransactions {
		if transaction.Type == protos.Transaction_CHAINCODE_DEPLOY {
			deploymentSpec := &protos.ChaincodeDeploymentSpec{}
			err := proto.Unmarshal(transaction.Payload, deploymentSpec)
			if err != nil {
				ledgerLogger.Errorf("Error unmarshalling deployment transaction for block event: %s", err)
				continue
			}
			deploymentSpec.CodePackage = nil
			deploymentSpecBytes, err := proto.Marshal(deploymentSpec)
			if err != nil {
				ledgerLogger.Errorf("Error marshalling deployment transaction for block event: %s", err)
				continue
			}
			transaction.Payload = deploymentSpecBytes
		}
	}

	producer.Send(producer.CreateBlockEvent(block))
}

func (blockchain *blockchain) addPersistenceChangesForNewBlock(ctx context.Context,
	block *protos.Block, stateHash []byte, writeBatch *gorocksdb.WriteBatch) (uint64, error) {
	block = blockchain.buildBlock(block, stateHash)
	if block.NonHashData == nil {
		block.NonHashData = &protos.NonHashData{LocalLedgerCommitTimestamp: util.CreateUtcTimestamp()}
	} else {
		block.NonHashData.LocalLedgerCommitTimestamp = util.CreateUtcTimestamp()
	}
	blockNumber := blockchain.size
	blockHash, err := block.GetHash()
	if err != nil {
		return 0, err
	}
	blockBytes, blockBytesErr := block.Bytes()
	if blockBytesErr != nil {
		return 0, blockBytesErr
	}
	writeBatch.PutCF(db.GetDBHandle().BlockchainCF, encodeBlockNumberDBKey(blockNumber), blockBytes)
	writeBatch.PutCF(db.GetDBHandle().BlockchainCF, blockCountKey, encodeUint64(blockNumber+1))
	if blockchain.indexer.isSynchronous() {
		blockchain.indexer.createIndexesSync(block, blockNumber, blockHash, writeBatch)
	}
	blockchain.lastProcessedBlock = &lastProcessedBlock{block, blockNumber, blockHash}
	return blockNumber, nil
}

func TestLedgerPutRawBlock(t *testing.T) {
	ledgerTestWrapper := createFreshDBAndTestLedgerWrapper(t)
	ledger := ledgerTestWrapper.ledger
	block := new(protos.Block)
	block.PreviousBlockHash = []byte("foo")
	block.StateHash = []byte("bar")
	ledger.PutRawBlock(block, 4)
	testutil.AssertEquals(t, ledgerTestWrapper.GetBlockByNumber(4), block)

	ledger.BeginTxBatch(1)
	ledger.TxBegin("txUuid")
	ledger.SetState("chaincode1", "key1", []byte("value1"))
	ledger.TxFinished("txUuid", true)
	transaction, _ := buildTestTx(t)
	ledger.CommitTxBatch(1, []*protos.Transaction{transaction}, nil, []byte("proof"))

	previousHash, _ := block.GetHash()
	newBlock := ledgerTestWrapper.GetBlockByNumber(5)

	if !bytes.Equal(newBlock.PreviousBlockHash, previousHash) {
		t.Fatalf("Expected new block to properly set its previous hash")
	}

	// Assert that a non-existent block is nil
	testutil.AssertNil(t, ledgerTestWrapper.GetBlockByNumber(2))
}

// Functions for persisting and retrieving index data
func addIndexDataForPersistence(block *protos.Block, blockNumber uint64, blockHash []byte, writeBatch *gorocksdb.WriteBatch) error {
	openchainDB := db.GetDBHandle()
	cf := openchainDB.IndexesCF

	// add blockhash -> blockNumber
	indexLogger.Debugf("Indexing block number [%d] by hash = [%x]", blockNumber, blockHash)
	writeBatch.PutCF(cf, encodeBlockHashKey(blockHash), encodeBlockNumber(blockNumber))

	addressToTxIndexesMap := make(map[string][]uint64)
	addressToChaincodeIDsMap := make(map[string][]*protos.ChaincodeID)

	transactions := block.GetTransactions()
	for txIndex, tx := range transactions {
		// add TxID -> (blockNumber,indexWithinBlock)
		writeBatch.PutCF(cf, encodeTxIDKey(tx.Txid), encodeBlockNumTxIndex(blockNumber, uint64(txIndex)))

		txExecutingAddress := getTxExecutingAddress(tx)
		addressToTxIndexesMap[txExecutingAddress] = append(addressToTxIndexesMap[txExecutingAddress], uint64(txIndex))

		switch tx.Type {
		case protos.Transaction_CHAINCODE_DEPLOY, protos.Transaction_CHAINCODE_INVOKE:
			authroizedAddresses, chaincodeID := getAuthorisedAddresses(tx)
			for _, authroizedAddress := range authroizedAddresses {
				addressToChaincodeIDsMap[authroizedAddress] = append(addressToChaincodeIDsMap[authroizedAddress], chaincodeID)
			}
		}
	}
	for address, txsIndexes := range addressToTxIndexesMap {
		writeBatch.PutCF(cf, encodeAddressBlockNumCompositeKey(address, blockNumber), encodeListTxIndexes(txsIndexes))
	}
	return nil
}

func (blockchain *blockchain) getBlockchainInfoForBlock(height uint64, block *protos.Block) *protos.BlockchainInfo {
	hash, _ := block.GetHash()
	info := &protos.BlockchainInfo{
		Height:            height,
		CurrentBlockHash:  hash,
		PreviousBlockHash: block.PreviousBlockHash}

	return info
}

//send chaincode events created by transactions in the block
func sendChaincodeEvents(block *protos.Block) {
	nonHashData := block.GetNonHashData()
	if nonHashData != nil {
		trs := nonHashData.GetTransactionResults()
		for _, tr := range trs {
			if tr.ChaincodeEvent != nil {
				producer.Send(producer.CreateChaincodeEvent(tr.ChaincodeEvent))
			}
		}
	}
}

func (blockchain *blockchain) buildBlock(block *protos.Block, stateHash []byte) *protos.Block {
	block.SetPreviousBlockHash(blockchain.previousBlockHash)
	block.StateHash = stateHash
	return block
}

			Expect(ledgerPtr.BeginTxBatch(2)).To(BeNil())
			ledgerPtr.TxBegin("txUuid1")
			Expect(ledgerPtr.SetState("chaincode1", "key1", []byte("value1"))).To(BeNil())
			ledgerPtr.TxFinished("txUuid1", true)
		})
		It("should retrieve a delta state hash array of length 1", func() {
			_, txDeltaHashes, err := ledgerPtr.GetTempStateHashWithTxDeltaStateHashes()
			Expect(err).To(BeNil())
			Expect(len(txDeltaHashes)).To(Equal(1))
		})
	})

	Describe("Ledger PutRawBlock", func() {
		ledgerPtr := InitSpec()

		block := new(protos.Block)
		block.PreviousBlockHash = []byte("foo")
		block.StateHash = []byte("bar")
		It("creates a raw block and puts it in the ledger without error", func() {
			Expect(ledgerPtr.PutRawBlock(block, 4)).To(BeNil())
		})
		It("should return the same block that was stored", func() {
			Expect(ledgerPtr.GetBlockByNumber(4)).To(Equal(block))
		})
		It("creates, populates and finishes a transaction", func() {
			Expect(ledgerPtr.BeginTxBatch(1)).To(BeNil())
			ledgerPtr.TxBegin("txUuid")
			Expect(ledgerPtr.SetState("chaincode1", "key1", []byte("value1"))).To(BeNil())
			ledgerPtr.TxFinished("txUuid", true)
		})
		It("should commit the batch", func() {

// HashBlock returns the hash of the included block, useful for mocking
func (p *PeerImpl) HashBlock(block *pb.Block) ([]byte, error) {
	return block.GetHash()
}

func (mock *MockLedger) HashBlock(block *protos.Block) ([]byte, error) {
	return block.GetHash()
}

// HashBlock returns the hash of the included block, useful for mocking
func (h *Helper) HashBlock(block *pb.Block) ([]byte, error) {
	return block.GetHash()
}