Merge branch 'pr/79'

Change-Id: I0bce011cdc7d52009842a9aae97e2b1b03335ec1

Author: skriptble

core/command/command.go

@@ -47,6 +47,24 @@ func updateClusterTimes(sess *session.Client, clock *session.ClusterClock, respo
 	return nil
 }
 
+func updateOperationTime(sess *session.Client, response bson.Reader) error {
+	if sess == nil {
+		return nil
+	}
+
+	opTimeElem, err := response.Lookup("operationTime")
+	if err != nil {
+		// operationTime not included by the server
+		return nil
+	}
+
+	t, i := opTimeElem.Value().Timestamp()
+	return sess.AdvanceOperationTime(&bson.Timestamp{
+		T: t,
+		I: i,
+	})
+}
+
 func marshalCommand(cmd *bson.Document) (bson.Reader, error) {
 	if cmd == nil {
 		return bson.Reader{5, 0, 0, 0, 0}, nil
@@ -103,7 +121,7 @@ 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, rc *readconcern.ReadConcern) error {
+func addReadConcern(cmd *bson.Document, desc description.SelectedServer, rc *readconcern.ReadConcern, sess *session.Client) error {
 	if rc == nil {
 		return nil
 	}
@@ -113,11 +131,20 @@ func addReadConcern(cmd *bson.Document, rc *readconcern.ReadConcern) error {
 		return err
 	}
 
+	rcDoc := element.Value().MutableDocument()
+	if description.SessionsSupported(desc.WireVersion) && sess != nil && sess.Consistent && sess.OperationTime != nil {
+		rcDoc = rcDoc.Append(
+			bson.EC.Timestamp("afterClusterTime", sess.OperationTime.T, sess.OperationTime.I),
+		)
+	}
+
 	if _, err := cmd.LookupElementErr(element.Key()); err != nil {
 		cmd.Delete(element.Key())
 	}
 
-	cmd.Append(element)
+	if rcDoc.Len() != 0 {
+		cmd.Append(bson.EC.SubDocument("readConcern", rcDoc))
+	}
 	return nil
 }
 

core/command/read.go

@@ -187,7 +187,7 @@ func (r *Read) decodeOpReply(wm wiremessage.WireMessage) {
 // Encode will encode this command into a wire message for the given server description.
 func (r *Read) Encode(desc description.SelectedServer) (wiremessage.WireMessage, error) {
 	cmd := r.Command.Copy()
-	err := addReadConcern(cmd, r.ReadConcern)
+	err := addReadConcern(cmd, desc, r.ReadConcern, r.Session)
 	if err != nil {
 		return nil, err
 	}
@@ -228,7 +228,7 @@ func (r *Read) Decode(desc description.SelectedServer, wm wiremessage.WireMessag
 	}
 
 	_ = updateClusterTimes(r.Session, r.Clock, r.result)
-
+	_ = updateOperationTime(r.Session, r.result)
 	return r
 }
 

core/command/write.go

@@ -165,6 +165,11 @@ func (w *Write) Decode(desc description.SelectedServer, wm wiremessage.WireMessa
 	}
 
 	_ = updateClusterTimes(w.Session, w.Clock, w.result)
+
+	if writeconcern.AckWrite(w.WriteConcern) {
+		// don't update session operation time for unacknowledged write
+		_ = updateOperationTime(w.Session, w.result)
+	}
 	return w
 }
 

core/session/client_session.go

@@ -12,11 +12,13 @@ var ErrSessionEnded = errors.New("ended session was used")
 
 // Client is a session for clients to run commands.
 type Client struct {
-	ClientID    uuid.UUID
-	ClusterTime *bson.Document
-	SessionID   *bson.Document
-	SessionType Type
-	Terminated  bool
+	ClientID      uuid.UUID
+	ClusterTime   *bson.Document
+	Consistent    bool // causal consistency
+	OperationTime *bson.Timestamp
+	SessionID     *bson.Document
+	SessionType   Type
+	Terminated    bool
 
 	pool          *Pool
 	serverSession *Server
@@ -59,19 +61,31 @@ func MaxClusterTime(ct1 *bson.Document, ct2 *bson.Document) *bson.Document {
 }
 
 // NewClientSession creates a Client.
-func NewClientSession(pool *Pool, clientID uuid.UUID, sessionType Type) (*Client, error) {
+func NewClientSession(pool *Pool, clientID uuid.UUID, sessionType Type, opts ...ClientOptioner) (*Client, error) {
+	c := &Client{
+		Consistent:  true, // causal consistency defaults to true
+		ClientID:    clientID,
+		SessionType: sessionType,
+		pool:        pool,
+	}
+
+	var err error
+	for _, opt := range opts {
+		err = opt.Option(c)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	servSess, err := pool.GetSession()
 	if err != nil {
 		return nil, err
 	}
 
-	return &Client{
-		ClientID:      clientID,
-		SessionID:     servSess.SessionID,
-		SessionType:   sessionType,
-		pool:          pool,
-		serverSession: servSess,
-	}, nil
+	c.SessionID = servSess.SessionID
+	c.serverSession = servSess
+
+	return c, nil
 }
 
 // AdvanceClusterTime updates the session's cluster time.
@@ -83,6 +97,26 @@ func (c *Client) AdvanceClusterTime(clusterTime *bson.Document) error {
 	return nil
 }
 
+// AdvanceOperationTime updates the session's operation time.
+func (c *Client) AdvanceOperationTime(opTime *bson.Timestamp) error {
+	if c.Terminated {
+		return ErrSessionEnded
+	}
+
+	if c.OperationTime == nil {
+		c.OperationTime = opTime
+		return nil
+	}
+
+	if opTime.T > c.OperationTime.T {
+		c.OperationTime = opTime
+	} else if (opTime.T == c.OperationTime.T) && (opTime.I > c.OperationTime.I) {
+		c.OperationTime = opTime
+	}
+
+	return nil
+}
+
 // UpdateUseTime updates the session's last used time.
 // Must be called whenver this session is used to send a command to the server.
 func (c *Client) UpdateUseTime() error {

core/session/client_session_test.go

@@ -5,9 +5,20 @@ import (
 
 	"github.com/mongodb/mongo-go-driver/bson"
 	"github.com/mongodb/mongo-go-driver/core/uuid"
+	"github.com/mongodb/mongo-go-driver/internal/testutil/helpers"
 	"github.com/stretchr/testify/require"
 )
 
+func compareOperationTimes(t *testing.T, expected *bson.Timestamp, actual *bson.Timestamp) {
+	if expected.T != actual.T {
+		t.Fatalf("T value mismatch; expected %d got %d", expected.T, actual.T)
+	}
+
+	if expected.I != actual.I {
+		t.Fatalf("I value mismatch; expected %d got %d", expected.I, actual.I)
+	}
+}
+
 func TestClientSession(t *testing.T) {
 	var clusterTime1 = bson.NewDocument(bson.EC.SubDocument("$clusterTime",
 		bson.NewDocument(bson.EC.Timestamp("clusterTime", 10, 5))))
@@ -30,7 +41,7 @@ func TestClientSession(t *testing.T) {
 
 	t.Run("TestAdvanceClusterTime", func(t *testing.T) {
 		id, _ := uuid.New()
-		sess, err := NewClientSession(&Pool{}, id, Explicit)
+		sess, err := NewClientSession(&Pool{}, id, Explicit, OptCausalConsistency(true))
 		require.Nil(t, err, "Unexpected error")
 		err = sess.AdvanceClusterTime(clusterTime2)
 		require.Nil(t, err, "Unexpected error")
@@ -52,10 +63,48 @@ func TestClientSession(t *testing.T) {
 
 	t.Run("TestEndSession", func(t *testing.T) {
 		id, _ := uuid.New()
-		sess, err := NewClientSession(&Pool{}, id, Explicit)
+		sess, err := NewClientSession(&Pool{}, id, Explicit, OptCausalConsistency(true))
 		require.Nil(t, err, "Unexpected error")
 		sess.EndSession()
 		err = sess.UpdateUseTime()
 		require.NotNil(t, err, "Expected error, received nil")
 	})
+
+	t.Run("TestAdvanceOperationTime", func(t *testing.T) {
+		id, _ := uuid.New()
+		sess, err := NewClientSession(&Pool{}, id, Explicit, OptCausalConsistency(true))
+		require.Nil(t, err, "Unexpected error")
+
+		optime1 := &bson.Timestamp{
+			T: 1,
+			I: 0,
+		}
+		err = sess.AdvanceOperationTime(optime1)
+		testhelpers.RequireNil(t, err, "error updating first operation time: %s", err)
+		compareOperationTimes(t, optime1, sess.OperationTime)
+
+		optime2 := &bson.Timestamp{
+			T: 2,
+			I: 0,
+		}
+		err = sess.AdvanceOperationTime(optime2)
+		testhelpers.RequireNil(t, err, "error updating second operation time: %s", err)
+		compareOperationTimes(t, optime2, sess.OperationTime)
+
+		optime3 := &bson.Timestamp{
+			T: 2,
+			I: 1,
+		}
+		err = sess.AdvanceOperationTime(optime3)
+		testhelpers.RequireNil(t, err, "error updating third operation time: %s", err)
+		compareOperationTimes(t, optime3, sess.OperationTime)
+
+		err = sess.AdvanceOperationTime(&bson.Timestamp{
+			T: 1,
+			I: 10,
+		})
+		testhelpers.RequireNil(t, err, "error updating fourth operation time: %s", err)
+		compareOperationTimes(t, optime3, sess.OperationTime)
+		sess.EndSession()
+	})
 }

core/session/options.go

@@ -0,0 +1,15 @@
+package session
+
+// ClientOptioner is the interface implemented by types that can be used as options for configuring a client session.
+type ClientOptioner interface {
+	Option(*Client) error
+}
+
+// OptCausalConsistency specifies if a session should be causally consistent.
+type OptCausalConsistency bool
+
+// Option implements the ClientOptioner interface.
+func (opt OptCausalConsistency) Option(c *Client) error {
+	c.Consistent = bool(opt)
+	return nil
+}