[FAB-3368] Zero length CouchDB attachment delays

CouchDB attachments with zero length causes CouchDB to return
successfully but the connection is not available for future
connections.  This is a bug identified in CouchDB.

Zero length attachments are valid.  When an empty byte array
is saved, CouchDB stores the value as a zero length attachment.

The next subsequent client request over the same connection times
out reading headers from server.

This is the CouchDB JIRA item for tracking purposes.
https://issues.apache.org/jira/browse/COUCHDB-3394

The fix in this change checks to see if the attachment in
SaveDoc is zero length.  If the attachment is zero length,
then request close value is set to true.  This setting will
not allow the http connection to be reused and will prevent
connection timeouts.

Change-Id: I1a1a33c44a3b96dfd2bffaf1cec5fe9ece084817
Signed-off-by: Chris Elder <[email protected]>

Author: Chris Elder

core/ledger/util/couchdb/couchdb.go

@@ -268,7 +268,7 @@ func (dbclient *CouchDatabase) CreateDatabaseIfNotExist() (*DBOperationResponse,
 		maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
 		//process the URL with a PUT, creates the database
-		resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPut, connectURL.String(), nil, "", "", maxRetries)
+		resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPut, connectURL.String(), nil, "", "", maxRetries, true)
 		if err != nil {
 			return nil, err
 		}
@@ -309,7 +309,7 @@ func (dbclient *CouchDatabase) GetDatabaseInfo() (*DBInfo, *DBReturn, error) {
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodGet, connectURL.String(), nil, "", "", maxRetries)
+	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodGet, connectURL.String(), nil, "", "", maxRetries, true)
 	if err != nil {
 		return nil, couchDBReturn, err
 	}
@@ -347,7 +347,7 @@ func (couchInstance *CouchInstance) VerifyCouchConfig() (*ConnectionInfo, *DBRet
 	maxRetriesOnStartup := couchInstance.conf.MaxRetriesOnStartup
 
 	resp, couchDBReturn, err := couchInstance.handleRequest(http.MethodGet, connectURL.String(), nil,
-		couchInstance.conf.Username, couchInstance.conf.Password, maxRetriesOnStartup)
+		couchInstance.conf.Username, couchInstance.conf.Password, maxRetriesOnStartup, true)
 
 	if err != nil {
 		return nil, couchDBReturn, fmt.Errorf("Unable to connect to CouchDB, check the hostname and port: %s", err.Error())
@@ -396,7 +396,7 @@ func (dbclient *CouchDatabase) DropDatabase() (*DBOperationResponse, error) {
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodDelete, connectURL.String(), nil, "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodDelete, connectURL.String(), nil, "", "", maxRetries, true)
 	if err != nil {
 		return nil, err
 	}
@@ -436,7 +436,7 @@ func (dbclient *CouchDatabase) EnsureFullCommit() (*DBOperationResponse, error)
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, connectURL.String(), nil, "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, connectURL.String(), nil, "", "", maxRetries, true)
 	if err != nil {
 		logger.Errorf("Failed to invoke _ensure_full_commit Error: %s\n", err.Error())
 		return nil, err
@@ -500,6 +500,9 @@ func (dbclient *CouchDatabase) SaveDoc(id string, rev string, couchDoc *CouchDoc
 	//Set up a default boundary for use by multipart if sending attachments
 	defaultBoundary := ""
 
+	//Create a flag for shared connections.  This is set to false for zero length attachments
+	keepConnectionOpen := true
+
 	//check to see if attachments is nil, if so, then this is a JSON only
 	if couchDoc.Attachments == nil {
 
@@ -519,6 +522,13 @@ func (dbclient *CouchDatabase) SaveDoc(id string, rev string, couchDoc *CouchDoc
 			return "", err3
 		}
 
+		//If there is a zero length attachment, do not keep the connection open
+		for _, attach := range couchDoc.Attachments {
+			if attach.Length < 1 {
+				keepConnectionOpen = false
+			}
+		}
+
 		//Set the data buffer to the data from the create multi-part data
 		data = multipartData.Bytes()
 
@@ -531,7 +541,8 @@ func (dbclient *CouchDatabase) SaveDoc(id string, rev string, couchDoc *CouchDoc
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
 	//handle the request for saving the JSON or attachments
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPut, saveURL.String(), data, rev, defaultBoundary, maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPut, saveURL.String(), data,
+		rev, defaultBoundary, maxRetries, keepConnectionOpen)
 	if err != nil {
 		return "", err
 	}
@@ -662,7 +673,7 @@ func (dbclient *CouchDatabase) ReadDoc(id string) (*CouchDoc, string, error) {
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodGet, readURL.String(), nil, "", "", maxRetries)
+	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodGet, readURL.String(), nil, "", "", maxRetries, true)
 	if err != nil {
 		if couchDBReturn != nil && couchDBReturn.StatusCode == 404 {
 			logger.Debug("Document not found (404), returning nil value instead of 404 error")
@@ -815,7 +826,7 @@ func (dbclient *CouchDatabase) ReadDocRange(startKey, endKey string, limit, skip
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodGet, rangeURL.String(), nil, "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodGet, rangeURL.String(), nil, "", "", maxRetries, true)
 	if err != nil {
 		return nil, err
 	}
@@ -913,7 +924,7 @@ func (dbclient *CouchDatabase) DeleteDoc(id, rev string) error {
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodDelete, deleteURL.String(), nil, rev, "", maxRetries)
+	resp, couchDBReturn, err := dbclient.CouchInstance.handleRequest(http.MethodDelete, deleteURL.String(), nil, rev, "", maxRetries, true)
 	if err != nil {
 		fmt.Printf("couchDBReturn=%v", couchDBReturn)
 		if couchDBReturn != nil && couchDBReturn.StatusCode == 404 {
@@ -950,7 +961,7 @@ func (dbclient *CouchDatabase) QueryDocuments(query string) (*[]QueryResult, err
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, queryURL.String(), []byte(query), "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, queryURL.String(), []byte(query), "", "", maxRetries, true)
 	if err != nil {
 		return nil, err
 	}
@@ -1036,7 +1047,7 @@ func (dbclient *CouchDatabase) BatchRetrieveIDRevision(keys []string) ([]*DocMet
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, batchURL.String(), jsonKeys, "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, batchURL.String(), jsonKeys, "", "", maxRetries, true)
 	if err != nil {
 		return nil, err
 	}
@@ -1131,7 +1142,7 @@ func (dbclient *CouchDatabase) BatchUpdateDocuments(documents []*CouchDoc) ([]*B
 	//get the number of retries
 	maxRetries := dbclient.CouchInstance.conf.MaxRetries
 
-	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, batchURL.String(), jsonKeys, "", "", maxRetries)
+	resp, _, err := dbclient.CouchInstance.handleRequest(http.MethodPost, batchURL.String(), jsonKeys, "", "", maxRetries, true)
 	if err != nil {
 		return nil, err
 	}
@@ -1166,7 +1177,7 @@ func (dbclient *CouchDatabase) BatchUpdateDocuments(documents []*CouchDoc) ([]*B
 // if it returns an error, it ensures that the response body is closed, else it is the
 // callee's responsibility to close response correctly
 func (couchInstance *CouchInstance) handleRequest(method, connectURL string, data []byte, rev string,
-	multipartBoundary string, maxRetries int) (*http.Response, *DBReturn, error) {
+	multipartBoundary string, maxRetries int, keepConnectionOpen bool) (*http.Response, *DBReturn, error) {
 
 	logger.Debugf("Entering handleRequest()  method=%s  url=%v", method, connectURL)
 
@@ -1192,6 +1203,13 @@ func (couchInstance *CouchInstance) handleRequest(method, connectURL string, dat
 			return nil, nil, err
 		}
 
+		//set the request to close on completion if shared connections are not allowSharedConnection
+		//Current CouchDB has a problem with zero length attachments, do not allow the connection to be reused.
+		//Apache JIRA item for CouchDB   https://issues.apache.org/jira/browse/COUCHDB-3394
+		if !keepConnectionOpen {
+			req.Close = true
+		}
+
 		//add content header for PUT
 		if method == http.MethodPut || method == http.MethodPost || method == http.MethodDelete {
 

core/ledger/util/couchdb/couchdb_test.go

@@ -376,13 +376,119 @@ func TestDBCreateDatabaseAndPersist(t *testing.T) {
 			//Assert that the update was saved and retrieved
 			testutil.AssertEquals(t, assetResp.Owner, "bob")
 
-			//Drop the database
-			_, errdbdrop := db.DropDatabase()
-			testutil.AssertNoError(t, errdbdrop, fmt.Sprintf("Error dropping database"))
+			testBytes2 := []byte(`test attachment 2`)
 
-			//Retrieve the info for the new database and make sure the name matches
-			_, _, errdbinfo := db.GetDatabaseInfo()
-			testutil.AssertError(t, errdbinfo, fmt.Sprintf("Error should have been thrown for missing database"))
+			attachment2 := &Attachment{}
+			attachment2.AttachmentBytes = testBytes2
+			attachment2.ContentType = "application/octet-stream"
+			attachment2.Name = "data"
+			attachments2 := []*Attachment{}
+			attachments2 = append(attachments2, attachment2)
+
+			//Save the test document with an attachment
+			_, saveerr = db.SaveDoc("2", "", &CouchDoc{JSONValue: nil, Attachments: attachments2})
+			testutil.AssertNoError(t, saveerr, fmt.Sprintf("Error when trying to save a document"))
+
+			//Retrieve the test document with attachments
+			dbGetResp, _, geterr = db.ReadDoc("2")
+			testutil.AssertNoError(t, geterr, fmt.Sprintf("Error when trying to retrieve a document"))
+
+			//verify the text from the attachment is correct
+			testattach := dbGetResp.Attachments[0].AttachmentBytes
+			testutil.AssertEquals(t, testattach, testBytes2)
+
+			testBytes3 := []byte{}
+
+			attachment3 := &Attachment{}
+			attachment3.AttachmentBytes = testBytes3
+			attachment3.ContentType = "application/octet-stream"
+			attachment3.Name = "data"
+			attachments3 := []*Attachment{}
+			attachments3 = append(attachments3, attachment3)
+
+			//Save the test document with a zero length attachment
+			_, saveerr = db.SaveDoc("3", "", &CouchDoc{JSONValue: nil, Attachments: attachments3})
+			testutil.AssertNoError(t, saveerr, fmt.Sprintf("Error when trying to save a document"))
+
+			//Retrieve the test document with attachments
+			dbGetResp, _, geterr = db.ReadDoc("3")
+			testutil.AssertNoError(t, geterr, fmt.Sprintf("Error when trying to retrieve a document"))
+
+			//verify the text from the attachment is correct,  zero bytes
+			testattach = dbGetResp.Attachments[0].AttachmentBytes
+			testutil.AssertEquals(t, testattach, testBytes3)
+
+			testBytes4a := []byte(`test attachment 4a`)
+			attachment4a := &Attachment{}
+			attachment4a.AttachmentBytes = testBytes4a
+			attachment4a.ContentType = "application/octet-stream"
+			attachment4a.Name = "data1"
+
+			testBytes4b := []byte(`test attachment 4b`)
+			attachment4b := &Attachment{}
+			attachment4b.AttachmentBytes = testBytes4b
+			attachment4b.ContentType = "application/octet-stream"
+			attachment4b.Name = "data2"
+
+			attachments4 := []*Attachment{}
+			attachments4 = append(attachments4, attachment4a)
+			attachments4 = append(attachments4, attachment4b)
+
+			//Save the updated test document with multiple attachments
+			_, saveerr = db.SaveDoc("4", "", &CouchDoc{JSONValue: assetJSON, Attachments: attachments4})
+			testutil.AssertNoError(t, saveerr, fmt.Sprintf("Error when trying to save the updated document"))
+
+			//Retrieve the test document with attachments
+			dbGetResp, _, geterr = db.ReadDoc("4")
+			testutil.AssertNoError(t, geterr, fmt.Sprintf("Error when trying to retrieve a document"))
+
+			for _, attach4 := range dbGetResp.Attachments {
+
+				currentName := attach4.Name
+				if currentName == "data1" {
+					testutil.AssertEquals(t, attach4.AttachmentBytes, testBytes4a)
+				}
+				if currentName == "data2" {
+					testutil.AssertEquals(t, attach4.AttachmentBytes, testBytes4b)
+				}
+
+			}
+
+			testBytes5a := []byte(`test attachment 5a`)
+			attachment5a := &Attachment{}
+			attachment5a.AttachmentBytes = testBytes5a
+			attachment5a.ContentType = "application/octet-stream"
+			attachment5a.Name = "data1"
+
+			testBytes5b := []byte{}
+			attachment5b := &Attachment{}
+			attachment5b.AttachmentBytes = testBytes5b
+			attachment5b.ContentType = "application/octet-stream"
+			attachment5b.Name = "data2"
+
+			attachments5 := []*Attachment{}
+			attachments5 = append(attachments5, attachment5a)
+			attachments5 = append(attachments5, attachment5b)
+
+			//Save the updated test document with multiple attachments and zero length attachments
+			_, saveerr = db.SaveDoc("5", "", &CouchDoc{JSONValue: assetJSON, Attachments: attachments5})
+			testutil.AssertNoError(t, saveerr, fmt.Sprintf("Error when trying to save the updated document"))
+
+			//Retrieve the test document with attachments
+			dbGetResp, _, geterr = db.ReadDoc("5")
+			testutil.AssertNoError(t, geterr, fmt.Sprintf("Error when trying to retrieve a document"))
+
+			for _, attach5 := range dbGetResp.Attachments {
+
+				currentName := attach5.Name
+				if currentName == "data1" {
+					testutil.AssertEquals(t, attach5.AttachmentBytes, testBytes5a)
+				}
+				if currentName == "data2" {
+					testutil.AssertEquals(t, attach5.AttachmentBytes, testBytes5b)
+				}
+
+			}
 
 			//Attempt to save the document with an invalid id
 			_, saveerr = db.SaveDoc(string([]byte{0xff, 0xfe, 0xfd}), "", &CouchDoc{JSONValue: assetJSON, Attachments: nil})
@@ -392,6 +498,22 @@ func TestDBCreateDatabaseAndPersist(t *testing.T) {
 			_, _, readerr := db.ReadDoc(string([]byte{0xff, 0xfe, 0xfd}))
 			testutil.AssertError(t, readerr, fmt.Sprintf("Error should have been thrown when reading a document with an invalid ID"))
 
+			//Drop the database
+			_, errdbdrop := db.DropDatabase()
+			testutil.AssertNoError(t, errdbdrop, fmt.Sprintf("Error dropping database"))
+
+			//Make sure an error is thrown for getting info for a missing database
+			_, _, errdbinfo := db.GetDatabaseInfo()
+			testutil.AssertError(t, errdbinfo, fmt.Sprintf("Error should have been thrown for missing database"))
+
+			//Attempt to save a document to a deleted database
+			_, saveerr = db.SaveDoc("6", "", &CouchDoc{JSONValue: assetJSON, Attachments: nil})
+			testutil.AssertError(t, saveerr, fmt.Sprintf("Error should have been thrown while attempting to save to a deleted database"))
+
+			//Attempt to read from a deleted database
+			_, _, geterr = db.ReadDoc("6")
+			testutil.AssertNoError(t, geterr, fmt.Sprintf("Error should not have been thrown for a missing database, nil value is returned"))
+
 		}
 	}