GODRIVER-603 changestream shouldn't panic when a user changes resumetoken

Change-Id: Ib495b53b1ca1867dc1c4f48a654e8f8405737839

Author: Isabella Siu

mongo/change_stream.go

@@ -197,7 +197,14 @@ func (cs *changeStream) DecodeBytes() (bson.Raw, error) {
 		return nil, ErrMissingResumeToken
 	}
 
-	cs.resumeToken, err = bsonx.ReadDoc(id.Document())
+	var idDoc bson.Raw
+	idDoc, ok := id.DocumentOK()
+	if !ok {
+		_ = cs.Close(context.Background())
+		return nil, ErrMissingResumeToken
+	}
+	cs.resumeToken, err = bsonx.ReadDoc(idDoc)
+
 	if err != nil {
 		_ = cs.Close(context.Background())
 		return nil, ErrMissingResumeToken

mongo/change_stream_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/google/go-cmp/cmp"
+	"github.com/mongodb/mongo-go-driver/bson/objectid"
 	"github.com/mongodb/mongo-go-driver/x/bsonx"
 	"github.com/mongodb/mongo-go-driver/x/network/command"
 	"github.com/stretchr/testify/require"
@@ -71,6 +72,44 @@ func TestChangeStream_firstStage(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestChangeStream_replaceRoot(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip()
+	}
+	skipIfBelow36(t)
+
+	if os.Getenv("TOPOLOGY") != "replica_set" {
+		t.Skip()
+	}
+
+	coll := createTestCollection(t, nil, nil)
+
+	// Ensure the database is created.
+	_, err := coll.InsertOne(context.Background(), bsonx.Doc{{"x", bsonx.Int32(7)}})
+	require.NoError(t, err)
+
+	pipeline := make(bsonx.Arr, 0)
+	pipeline = append(pipeline,
+		bsonx.Document(bsonx.Doc{{"$replaceRoot",
+			bsonx.Document(bsonx.Doc{{"newRoot",
+				bsonx.Document(bsonx.Doc{{"_id", bsonx.ObjectID(objectid.New())}, {"x", bsonx.Int32(1)}})}}),
+		}}))
+	changes, err := coll.Watch(context.Background(), pipeline)
+	require.NoError(t, err)
+
+	_, err = coll.InsertOne(context.Background(), bsonx.Doc{{"x", bsonx.Int32(4)}})
+	require.NoError(t, err)
+
+	getNextChange(changes)
+	var doc *bsonx.Doc
+
+	//Ensure the cursor returns an error when the resume token is changed.
+	err = changes.Decode(&doc)
+	require.Equal(t, err, ErrMissingResumeToken)
+}
+
 func TestChangeStream_noCustomStandaloneError(t *testing.T) {
 	t.Parallel()