Merge "FAB-1291: Couch support for doing a savepoint."

core/ledger/kvledger/txmgmt/couchdbtxmgmt/couchdb_txmgmt_test.go

@@ -106,3 +106,33 @@ func TestDatabaseAutoCreate(t *testing.T) {
 	}
 
 }
+
+//TestSavepoint tests the recordSavepoint and GetBlockNumfromSavepoint methods for recording and reading a savepoint document
+func TestSavepoint(t *testing.T) {
+
+	//Only run the tests if CouchDB is explitily enabled in the code,
+	//otherwise CouchDB may not be installed and all the tests would fail
+	//TODO replace this with external config property rather than config within the code
+	if ledgerconfig.IsCouchDBEnabled() == true {
+
+		env := newTestEnv(t)
+
+		txMgr := NewCouchDBTxMgr(env.conf,
+			env.couchDBAddress,    //couchDB Address
+			env.couchDatabaseName, //couchDB db name
+			env.couchUsername,     //enter couchDB id
+			env.couchPassword)     //enter couchDB pw
+
+		// record savepoint
+		txMgr.blockNum = 5
+		err := txMgr.recordSavepoint()
+		testutil.AssertNoError(t, err, fmt.Sprintf("Error when saving recordpoint data"))
+
+		// read the savepoint
+		blockNum, err := txMgr.GetBlockNumFromSavepoint()
+		testutil.AssertNoError(t, err, fmt.Sprintf("Error when saving recordpoint data"))
+		testutil.AssertEquals(t, txMgr.blockNum, blockNum)
+
+		txMgr.Shutdown()
+	}
+}

core/ledger/kvledger/txmgmt/couchdbtxmgmt/couchdb_txmgr.go

@@ -17,6 +17,8 @@ limitations under the License.
 package couchdbtxmgmt
 
 import (
+	"encoding/json"
+	"errors"
 	"sync"
 
 	"github.com/golang/protobuf/proto"
@@ -48,6 +50,15 @@ type updateSet struct {
 	m map[string]*versionedValue
 }
 
+// Savepoint docid (key) for couchdb
+const savepointDocID = "statedb_savepoint"
+
+// Savepoint data for couchdb
+type couchSavepointData struct {
+	BlockNum  uint64 `json:"BlockNum"`
+	UpdateSeq string `json:"UpdateSeq"`
+}
+
 func newUpdateSet() *updateSet {
 	return &updateSet{make(map[string]*versionedValue)}
 }
@@ -72,6 +83,7 @@ type CouchDBTxMgr struct {
 	updateSet    *updateSet
 	commitRWLock sync.RWMutex
 	couchDB      *couchdb.CouchDBConnectionDef // COUCHDB new properties for CouchDB
+	blockNum     uint64                        // block number corresponding to updateSet
 }
 
 // CouchConnection provides connection info for CouchDB
@@ -119,6 +131,7 @@ func (txmgr *CouchDBTxMgr) ValidateAndPrepare(block *common.Block) (*common.Bloc
 	invalidTxs := []*pb.InvalidTransaction{}
 	var valid bool
 	txmgr.updateSet = newUpdateSet()
+	txmgr.blockNum = block.Header.Number
 	logger.Debugf("Validating a block with [%d] transactions", len(block.Data.Data))
 	for txIndex, envBytes := range block.Data.Data {
 		// extract actions from the envelope message
@@ -159,6 +172,7 @@ func (txmgr *CouchDBTxMgr) ValidateAndPrepare(block *common.Block) (*common.Bloc
 				Transaction: &pb.Transaction{ /* FIXME */ }, Cause: pb.InvalidTransaction_RWConflictDuringCommit})
 		}
 	}
+
 	logger.Debugf("===COUCHDB=== Exiting CouchDBTxMgr.ValidateAndPrepare()")
 	return block, invalidTxs, nil
 }
@@ -258,13 +272,87 @@ func (txmgr *CouchDBTxMgr) Commit() error {
 
 	}
 
+	// Record a savepoint
+	err := txmgr.recordSavepoint()
+	if err != nil {
+		logger.Errorf("===COUCHDB=== Error during recordSavepoint: %s\n", err.Error())
+		return err
+	}
+
 	logger.Debugf("===COUCHDB=== Exiting CouchDBTxMgr.Commit()")
 	return nil
 }
 
+// recordSavepoint Record a savepoint in statedb.
+// Couch parallelizes writes in cluster or sharded setup and ordering is not guaranteed.
+// Hence we need to fence the savepoint with sync. So ensure_full_commit is called before AND after writing savepoint document
+// TODO: Optimization - merge 2nd ensure_full_commit with savepoint by using X-Couch-Full-Commit header
+func (txmgr *CouchDBTxMgr) recordSavepoint() error {
+	var err error
+	var savepointDoc couchSavepointData
+	// ensure full commit to flush all changes until now to disk
+	dbResponse, err := txmgr.couchDB.EnsureFullCommit()
+	if err != nil || dbResponse.Ok != true {
+		logger.Errorf("====COUCHDB==== Failed to perform full commit\n")
+		return errors.New("Failed to perform full commit")
+	}
+
+	// construct savepoint document
+	// UpdateSeq would be useful if we want to get all db changes since a logical savepoint
+	dbInfo, _, err := txmgr.couchDB.GetDatabaseInfo()
+	if err != nil {
+		logger.Errorf("====COUCHDB==== Failed to get DB info %s\n", err.Error())
+		return err
+	}
+	savepointDoc.BlockNum = txmgr.blockNum
+	savepointDoc.UpdateSeq = dbInfo.UpdateSeq
+
+	savepointDocJSON, err := json.Marshal(savepointDoc)
+	if err != nil {
+		logger.Errorf("====COUCHDB==== Failed to create savepoint data %s\n", err.Error())
+		return err
+	}
+
+	// SaveDoc using couchdb client and use JSON format
+	_, err = txmgr.couchDB.SaveDoc(savepointDocID, "", savepointDocJSON, nil)
+	if err != nil {
+		logger.Errorf("====CouchDB==== Failed to save the savepoint to DB %s\n", err.Error())
+	}
+
+	// ensure full commit to flush savepoint to disk
+	dbResponse, err = txmgr.couchDB.EnsureFullCommit()
+	if err != nil || dbResponse.Ok != true {
+		logger.Errorf("====COUCHDB==== Failed to perform full commit\n")
+		return errors.New("Failed to perform full commit")
+	}
+	return nil
+}
+
+// GetBlockNumFromSavepoint Reads the savepoint from database and returns the corresponding block number.
+// If no savepoint is found, it returns 0
+func (txmgr *CouchDBTxMgr) GetBlockNumFromSavepoint() (uint64, error) {
+	var err error
+	savepointJSON, _, err := txmgr.couchDB.ReadDoc(savepointDocID)
+	if err != nil {
+		// TODO: differentiate between 404 and some other error code
+		logger.Errorf("====COUCHDB==== Failed to read savepoint data %s\n", err.Error())
+		return 0, err
+	}
+
+	savepointDoc := &couchSavepointData{}
+	err = json.Unmarshal(savepointJSON, &savepointDoc)
+	if err != nil {
+		logger.Errorf("====COUCHDB==== Failed to read savepoint data %s\n", err.Error())
+		return 0, err
+	}
+
+	return savepointDoc.BlockNum, nil
+}
+
 // Rollback implements method in interface `txmgmt.TxMgr`
 func (txmgr *CouchDBTxMgr) Rollback() {
 	txmgr.updateSet = nil
+	txmgr.blockNum = 0
 }
 
 func (txmgr *CouchDBTxMgr) getCommitedVersion(ns string, key string) (*version.Height, error) {

core/ledger/kvledger/txmgmt/lockbasedtxmgmt/lockbased_txmgmt_test.go

@@ -18,6 +18,7 @@ package lockbasedtxmgmt
 
 import (
 	"fmt"
+	"math"
 	"testing"
 
 	"github.com/hyperledger/fabric/core/ledger"
@@ -67,9 +68,13 @@ func TestTxSimulatorWithExistingData(t *testing.T) {
 	testutil.AssertNoError(t, err, fmt.Sprintf("Error in validateTx(): %s", err))
 	testutil.AssertSame(t, isValid, true)
 	txMgr.addWriteSetToBatch(txRWSet, version.NewHeight(1, 1))
+	txMgr.blockNum = math.MaxUint64
 	err = txMgr.Commit()
 	testutil.AssertNoError(t, err, fmt.Sprintf("Error while calling commit(): %s", err))
 
+	blockNum, err := txMgr.GetBlockNumFromSavepoint()
+	testutil.AssertEquals(t, blockNum, txMgr.blockNum)
+
 	// simulate tx2 that make changes to existing data
 	s2, _ := txMgr.NewTxSimulator()
 	value, _ := s2.GetState("ns1", "key1")
@@ -258,7 +263,7 @@ func testIterator(t *testing.T, numKeys int, startKeyNum int, endKeyNum int) {
 		keyNum := begin + count
 		k := kv.(*ledger.KV).Key
 		v := kv.(*ledger.KV).Value
-		t.Logf("Retrieved k=%s, v=%s", k, v)
+		t.Logf("Retrieved k=%s, v=%s at count=%d start=%s end=%s", k, v, count, startKey, endKey)
 		testutil.AssertEquals(t, k, createTestKey(keyNum))
 		testutil.AssertEquals(t, v, createTestValue(keyNum))
 		count++

core/ledger/kvledger/txmgmt/lockbasedtxmgmt/lockbased_txmgr.go

@@ -18,6 +18,8 @@ package lockbasedtxmgmt
 
 import (
 	"bytes"
+	"encoding/binary"
+	"reflect"
 	"sync"
 
 	"github.com/hyperledger/fabric/core/ledger"
@@ -50,6 +52,9 @@ type updateSet struct {
 	m map[string]*versionedValue
 }
 
+// savepoint key
+const savepointKey = "savepoint"
+
 func newUpdateSet() *updateSet {
 	return &updateSet{make(map[string]*versionedValue)}
 }
@@ -73,6 +78,7 @@ type LockBasedTxMgr struct {
 	db           *db.DB
 	updateSet    *updateSet
 	commitRWLock sync.RWMutex
+	blockNum     uint64
 }
 
 // NewLockBasedTxMgr constructs a `LockBasedTxMgr`
@@ -102,6 +108,7 @@ func (txmgr *LockBasedTxMgr) ValidateAndPrepare(block *common.Block) (*common.Bl
 	invalidTxs := []*pb.InvalidTransaction{}
 	var valid bool
 	txmgr.updateSet = newUpdateSet()
+	txmgr.blockNum = block.Header.Number
 	logger.Debugf("Validating a block with [%d] transactions", len(block.Data.Data))
 	for txIndex, envBytes := range block.Data.Data {
 		// extract actions from the envelope message
@@ -203,6 +210,15 @@ func (txmgr *LockBasedTxMgr) Commit() error {
 			batch.Put([]byte(k), encodeValue(v.value, v.version))
 		}
 	}
+
+	// record the savepoint along with batch
+	if txmgr.blockNum != 0 {
+		savepointValue := make([]byte, reflect.TypeOf(txmgr.blockNum).Size())
+		binary.LittleEndian.PutUint64(savepointValue, txmgr.blockNum)
+		// Need a composite key for iterator to function correctly - use separator itself as special/hidden namespace
+		batch.Put(constructCompositeKey(string(compositeKeySep), savepointKey), savepointValue)
+	}
+
 	txmgr.commitRWLock.Lock()
 	defer txmgr.commitRWLock.Unlock()
 	defer func() { txmgr.updateSet = nil }()
@@ -212,9 +228,24 @@ func (txmgr *LockBasedTxMgr) Commit() error {
 	return nil
 }
 
+// GetBlockNumFromSavepoint returns the block num recorded in savepoint,
+// returns 0 if NO savepoint is found
+func (txmgr *LockBasedTxMgr) GetBlockNumFromSavepoint() (uint64, error) {
+	var blockNum uint64
+	savepointValue, err := txmgr.db.Get(constructCompositeKey(string(compositeKeySep), savepointKey))
+	if err != nil {
+		return 0, err
+	}
+
+	// savepointValue is not encoded with version
+	blockNum = binary.LittleEndian.Uint64(savepointValue)
+	return blockNum, nil
+}
+
 // Rollback implements method in interface `txmgmt.TxMgr`
 func (txmgr *LockBasedTxMgr) Rollback() {
 	txmgr.updateSet = nil
+	txmgr.blockNum = 0
 }
 
 func (txmgr *LockBasedTxMgr) getCommitedVersion(ns string, key string) (*version.Height, error) {

core/ledger/util/couchdb/couchdb.go

@@ -230,6 +230,38 @@ func (dbclient *CouchDBConnectionDef) DropDatabase() (*DBOperationResponse, erro
 
 }
 
+// EnsureFullCommit calls _ensure_full_commit for explicit fsync
+func (dbclient *CouchDBConnectionDef) EnsureFullCommit() (*DBOperationResponse, error) {
+
+	logger.Debugf("===COUCHDB=== Entering EnsureFullCommit()")
+
+	url := fmt.Sprintf("%s/%s/_ensure_full_commit", dbclient.URL, dbclient.Database)
+
+	resp, _, err := dbclient.handleRequest(http.MethodPost, url, nil, "", "")
+	if err != nil {
+		logger.Errorf("====COUCHDB==== Failed to invoke _ensure_full_commit Error: %s\n", err.Error())
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	dbResponse := &DBOperationResponse{}
+	json.NewDecoder(resp.Body).Decode(&dbResponse)
+
+	if dbResponse.Ok == true {
+		logger.Debugf("===COUCHDB=== _ensure_full_commit database %s ", dbclient.Database)
+	}
+
+	logger.Debugf("===COUCHDB=== Exiting EnsureFullCommit()")
+
+	if dbResponse.Ok == true {
+
+		return dbResponse, nil
+
+	}
+
+	return dbResponse, fmt.Errorf("Error syncing database")
+}
+
 //SaveDoc method provides a function to save a document, id and byte array
 func (dbclient *CouchDBConnectionDef) SaveDoc(id string, rev string, bytesDoc []byte, attachments []Attachment) (string, error) {
 
@@ -497,7 +529,7 @@ func (dbclient *CouchDBConnectionDef) handleRequest(method, url string, data io.
 	}
 
 	//add content header for PUT
-	if method == http.MethodPut {
+	if method == http.MethodPut || method == http.MethodPost {
 
 		//If the multipartBoundary is not set, then this is a JSON and content-type should be set
 		//to application/json.   Else, this is contains an attachment and needs to be multipart
@@ -514,7 +546,7 @@ func (dbclient *CouchDBConnectionDef) handleRequest(method, url string, data io.
 	}
 
 	//add content header for PUT
-	if method == http.MethodPut {
+	if method == http.MethodPut || method == http.MethodPost {
 		req.Header.Set("Accept", "application/json")
 	}