Merge branch 'pr/81'

Change-Id: I36542fdacf1a5f2df21fd15aac1c9e08026453b2

Author: skriptble

core/command/abort_transaction.go

@@ -0,0 +1,82 @@
+package command
+
+import (
+	"context"
+
+	"github.com/mongodb/mongo-go-driver/bson"
+	"github.com/mongodb/mongo-go-driver/core/description"
+	"github.com/mongodb/mongo-go-driver/core/result"
+	"github.com/mongodb/mongo-go-driver/core/session"
+	"github.com/mongodb/mongo-go-driver/core/wiremessage"
+)
+
+// AbortTransaction represents the abortTransaction() command
+type AbortTransaction struct {
+	Session *session.Client
+	err     error
+	result  result.TransactionResult
+}
+
+// Encode will encode this command into a wiremessage for the given server description.
+func (at *AbortTransaction) Encode(desc description.SelectedServer) (wiremessage.WireMessage, error) {
+	cmd := at.encode(desc)
+	return cmd.Encode(desc)
+}
+
+func (at *AbortTransaction) encode(desc description.SelectedServer) *Write {
+	cmd := bson.NewDocument(bson.EC.Int32("abortTransaction", 1))
+	return &Write{
+		DB:           "admin",
+		Command:      cmd,
+		Session:      at.Session,
+		WriteConcern: at.Session.CurrentWc,
+	}
+}
+
+// Decode will decode the wire message using the provided server description. Errors during decoding are deferred until
+// either the Result or Err methods are called.
+func (at *AbortTransaction) Decode(desc description.SelectedServer, wm wiremessage.WireMessage) *AbortTransaction {
+	rdr, err := (&Write{}).Decode(desc, wm).Result()
+	if err != nil {
+		at.err = err
+		return at
+	}
+
+	return at.decode(desc, rdr)
+}
+
+func (at *AbortTransaction) decode(desc description.SelectedServer, rdr bson.Reader) *AbortTransaction {
+	at.err = bson.Unmarshal(rdr, &at.result)
+	if at.err == nil && at.result.WriteConcernError != nil {
+		at.err = Error{
+			Code:    int32(at.result.WriteConcernError.Code),
+			Message: at.result.WriteConcernError.ErrMsg,
+		}
+	}
+	return at
+}
+
+// Result returns the result of a decoded wire message and server description.
+func (at *AbortTransaction) Result() (result.TransactionResult, error) {
+	if at.err != nil {
+		return result.TransactionResult{}, at.err
+	}
+
+	return at.result, nil
+}
+
+// Err returns the error set on this command
+func (at *AbortTransaction) Err() error {
+	return at.err
+}
+
+// RoundTrip handles the execution of this command using the provided wiremessage.ReadWriter
+func (at *AbortTransaction) RoundTrip(ctx context.Context, desc description.SelectedServer, rw wiremessage.ReadWriter) (result.TransactionResult, error) {
+	cmd := at.encode(desc)
+	rdr, err := cmd.RoundTrip(ctx, desc, rw)
+	if err != nil {
+		return result.TransactionResult{}, err
+	}
+
+	return at.decode(desc, rdr).Result()
+}

core/command/aggregate.go

@@ -144,7 +144,14 @@ func (a *Aggregate) decode(desc description.SelectedServer, cb CursorBuilder, rd
 		opts = append(opts, curOpt)
 	}
 
-	a.result, a.err = cb.BuildCursor(rdr, a.Session, a.Clock, opts...)
+	labels, err := getErrorLabels(&rdr)
+	a.err = err
+
+	res, err := cb.BuildCursor(rdr, a.Session, a.Clock, opts...)
+	a.result = res
+	if err != nil {
+		a.err = Error{Message: err.Error(), Labels: labels}
+	}
 	return a
 }
 

core/command/command.go

@@ -73,8 +73,8 @@ func marshalCommand(cmd *bson.Document) (bson.Reader, error) {
 	return cmd.MarshalBSON()
 }
 
-// add a session ID to a BSON doc representing a command
-func addSessionID(cmd *bson.Document, desc description.SelectedServer, client *session.Client) error {
+// adds session related fields to a BSON doc representing a command
+func addSessionFields(cmd *bson.Document, desc description.SelectedServer, client *session.Client) error {
 	if client == nil || !description.SessionsSupported(desc.WireVersion) || desc.SessionTimeoutMinutes == 0 {
 		return nil
 	}
@@ -88,9 +88,27 @@ func addSessionID(cmd *bson.Document, desc description.SelectedServer, client *s
 	}
 
 	cmd.Append(bson.EC.SubDocument("lsid", client.SessionID))
+
+	if client.TransactionRunning() ||
+		client.RetryingCommit {
+		addTransaction(cmd, client)
+	}
+
+	client.ApplyCommand() // advance the state machine based on a command executing
+
 	return nil
 }
 
+// if in a transaction, add the transaction fields
+func addTransaction(cmd *bson.Document, client *session.Client) {
+	cmd.Append(bson.EC.Int64("txnNumber", client.TxnNumber))
+	if client.TransactionStarting() {
+		// When starting transaction, always transition to the next state, even on error
+		cmd.Append(bson.EC.Boolean("startTransaction", true))
+	}
+	cmd.Append(bson.EC.Boolean("autocommit", false))
+}
+
 func addClusterTime(cmd *bson.Document, desc description.SelectedServer, sess *session.Client, clock *session.ClusterClock) error {
 	if (clock == nil && sess == nil) || !description.SessionsSupported(desc.WireVersion) {
 		return nil
@@ -122,6 +140,16 @@ func addClusterTime(cmd *bson.Document, desc description.SelectedServer, sess *s
 
 // add a read concern to a BSON doc representing a command
 func addReadConcern(cmd *bson.Document, desc description.SelectedServer, rc *readconcern.ReadConcern, sess *session.Client) error {
+	// Starting transaction's read concern overrides all others
+	if sess != nil && sess.TransactionStarting() && sess.CurrentRc != nil {
+		rc = sess.CurrentRc
+	}
+
+	// start transaction must append afterclustertime IF causally consistent and operation time exists
+	if rc == nil && sess != nil && sess.TransactionStarting() && sess.Consistent && sess.OperationTime != nil {
+		rc = readconcern.New()
+	}
+
 	if rc == nil {
 		return nil
 	}
@@ -168,6 +196,25 @@ func addWriteConcern(cmd *bson.Document, wc *writeconcern.WriteConcern) error {
 	return nil
 }
 
+// Get the error labels from a command response
+func getErrorLabels(rdr *bson.Reader) ([]string, error) {
+	var labels []string
+	labelsElem, err := rdr.Lookup("errorLabels")
+	if err != bson.ErrElementNotFound {
+		return nil, err
+	}
+	if labelsElem != nil {
+		labelsIt, err := labelsElem.Value().ReaderArray().Iterator()
+		if err != nil {
+			return nil, err
+		}
+		for labelsIt.Next() {
+			labels = append(labels, labelsIt.Element().Value().StringValue())
+		}
+	}
+	return labels, nil
+}
+
 // Remove command arguments for insert, update, and delete commands from the BSON document so they can be encoded
 // as a Section 1 payload in OP_MSG
 func opmsgRemoveArray(cmdDoc *bson.Document) (*bson.Array, string) {

core/command/commit_transaction.go

@@ -0,0 +1,82 @@
+package command
+
+import (
+	"context"
+
+	"github.com/mongodb/mongo-go-driver/bson"
+	"github.com/mongodb/mongo-go-driver/core/description"
+	"github.com/mongodb/mongo-go-driver/core/result"
+	"github.com/mongodb/mongo-go-driver/core/session"
+	"github.com/mongodb/mongo-go-driver/core/wiremessage"
+)
+
+// CommitTransaction represents the commitTransaction() command
+type CommitTransaction struct {
+	Session *session.Client
+	err     error
+	result  result.TransactionResult
+}
+
+// Encode will encode this command into a wiremessage for the given server description.
+func (ct *CommitTransaction) Encode(desc description.SelectedServer) (wiremessage.WireMessage, error) {
+	cmd := ct.encode(desc)
+	return cmd.Encode(desc)
+}
+
+func (ct *CommitTransaction) encode(desc description.SelectedServer) *Write {
+	cmd := bson.NewDocument(bson.EC.Int32("commitTransaction", 1))
+	return &Write{
+		DB:           "admin",
+		Command:      cmd,
+		Session:      ct.Session,
+		WriteConcern: ct.Session.CurrentWc,
+	}
+}
+
+// Decode will decode the wire message using the provided server description. Errors during decoding are deferred until
+// either the Result or Err methods are called.
+func (ct *CommitTransaction) Decode(desc description.SelectedServer, wm wiremessage.WireMessage) *CommitTransaction {
+	rdr, err := (&Write{}).Decode(desc, wm).Result()
+	if err != nil {
+		ct.err = err
+		return ct
+	}
+
+	return ct.decode(desc, rdr)
+}
+
+func (ct *CommitTransaction) decode(desc description.SelectedServer, rdr bson.Reader) *CommitTransaction {
+	ct.err = bson.Unmarshal(rdr, &ct.result)
+	if ct.err == nil && ct.result.WriteConcernError != nil {
+		ct.err = Error{
+			Code:    int32(ct.result.WriteConcernError.Code),
+			Message: ct.result.WriteConcernError.ErrMsg,
+		}
+	}
+	return ct
+}
+
+// Result returns the result of a decoded wire message and server description.
+func (ct *CommitTransaction) Result() (result.TransactionResult, error) {
+	if ct.err != nil {
+		return result.TransactionResult{}, ct.err
+	}
+
+	return ct.result, nil
+}
+
+// Err returns the error set on this command
+func (ct *CommitTransaction) Err() error {
+	return ct.err
+}
+
+// RoundTrip handles the execution of this command using the provided wiremessage.ReadWriter
+func (ct *CommitTransaction) RoundTrip(ctx context.Context, desc description.SelectedServer, rw wiremessage.ReadWriter) (result.TransactionResult, error) {
+	cmd := ct.encode(desc)
+	rdr, err := cmd.RoundTrip(ctx, desc, rw)
+	if err != nil {
+		return result.TransactionResult{}, err
+	}
+
+	return ct.decode(desc, rdr).Result()
+}

core/command/count_documents.go

@@ -9,6 +9,7 @@ package command
 import (
 	"context"
 	"errors"
+
 	"github.com/mongodb/mongo-go-driver/bson"
 	"github.com/mongodb/mongo-go-driver/core/description"
 	"github.com/mongodb/mongo-go-driver/core/option"

core/command/errors.go

@@ -10,6 +10,8 @@ import (
 	"errors"
 	"fmt"
 
+	"strings"
+
 	"github.com/mongodb/mongo-go-driver/bson"
 )
 
@@ -25,8 +27,18 @@ var (
 	// ErrDocumentTooLarge occurs when a document that is larger than the maximum size accepted by a
 	// server is passed to an insert command.
 	ErrDocumentTooLarge = errors.New("an inserted document is too large")
+	// ErrNonPrimaryRP occurs when a nonprimary read preference is used with a transaction.
+	ErrNonPrimaryRP = errors.New("read preference in a transaction must be primary")
+	// UnknownTransactionCommitResult is an error label for unknown transaction commit results.
+	UnknownTransactionCommitResult = "UnknownTransactionCommitResult"
+	// TransientTransactionError is an error label for transient errors with transactions.
+	TransientTransactionError = "TransientTransactionError"
+	// NetworkError is an error label for network errors.
+	NetworkError = "NetworkError"
 )
 
+var retryableCodes = []int32{11600, 11602, 10107, 13435, 13436, 189, 91, 7, 6, 89, 9001}
+
 // QueryFailureError is an error representing a command failure as a document.
 type QueryFailureError struct {
 	Message  string
@@ -61,6 +73,7 @@ func (e ResponseError) Error() string {
 type Error struct {
 	Code    int32
 	Message string
+	Labels  []string
 	Name    string
 }
 
@@ -72,6 +85,37 @@ func (e Error) Error() string {
 	return e.Message
 }
 
+// HasErrorLabel returns true if the error contains the specified label.
+func (e Error) HasErrorLabel(label string) bool {
+	if e.Labels != nil {
+		for _, l := range e.Labels {
+			if l == label {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// Retryable returns true if the error is retryable
+func (e Error) Retryable() bool {
+	for _, label := range e.Labels {
+		if label == NetworkError {
+			return true
+		}
+	}
+	for _, code := range retryableCodes {
+		if e.Code == code {
+			return true
+		}
+	}
+	if strings.Contains(e.Message, "not master") || strings.Contains(e.Message, "node is recovering") {
+		return true
+	}
+
+	return false
+}
+
 // IsNotFound indicates if the error is from a namespace not being found.
 func IsNotFound(err error) bool {
 	e, ok := err.(Error)

core/command/find.go

@@ -115,7 +115,14 @@ func (f *Find) decode(desc description.SelectedServer, cb CursorBuilder, rdr bso
 		opts = append(opts, curOpt)
 	}
 
-	f.result, f.err = cb.BuildCursor(rdr, f.Session, f.Clock, opts...)
+	labels, err := getErrorLabels(&rdr)
+	f.err = err
+
+	res, err := cb.BuildCursor(rdr, f.Session, f.Clock, opts...)
+	f.result = res
+	if err != nil {
+		f.err = Error{Message: err.Error(), Labels: labels}
+	}
 	return f
 }
 

core/command/list_collections.go

@@ -89,7 +89,14 @@ func (lc *ListCollections) decode(desc description.SelectedServer, cb CursorBuil
 		opts = append(opts, curOpt)
 	}
 
-	lc.result, lc.err = cb.BuildCursor(rdr, lc.Session, lc.Clock, opts...)
+	labels, err := getErrorLabels(&rdr)
+	lc.err = err
+
+	res, err := cb.BuildCursor(rdr, lc.Session, lc.Clock, opts...)
+	lc.result = res
+	if err != nil {
+		lc.err = Error{Message: err.Error(), Labels: labels}
+	}
 
 	return lc
 }

core/command/list_indexes.go

@@ -85,7 +85,15 @@ func (li *ListIndexes) decode(desc description.SelectedServer, cb CursorBuilder,
 		opts = append(opts, curOpt)
 	}
 
-	li.result, li.err = cb.BuildCursor(rdr, li.Session, li.Clock, opts...)
+	labels, err := getErrorLabels(&rdr)
+	li.err = err
+
+	res, err := cb.BuildCursor(rdr, li.Session, li.Clock, opts...)
+	li.result = res
+	if err != nil {
+		li.err = Error{Message: err.Error(), Labels: labels}
+	}
+
 	return li
 }