REALMC-8848: upgrade to latest go driver release (#4)

* REALMC-5872 merging topology changes to v1.4.1 driver (#1)

* Update version to v1.3.1+prerelease

* Change w value for unsatisfiable write concern in tests (mongodb#301)

- Addresses test failures that started after by SERVER-45920

* GODRIVER-895 check if strLength is 0 in ReadCodeWithScope (mongodb#311)

* GODRIVER-1487 fix adding UnknownTransactionCommitResult (mongodb#307)

* GODRIVER-1488 turn off DecodeDeepZeroInline for mgocompat.Registry (mongodb#312)

* Fix primitive.Regex Equal method (mongodb#318)

* GODRIVER-1507 Correctly pass URI to topology (mongodb#320)

* Update version to 1.3.1

* Update version to v1.3.2+prerelease

* GODRIVER-1522 Ignore read preference for aggregations with output stages (mongodb#327)

* GODRIVER-1520 Fix panics for lone scope (mongodb#331)

* GODRIVER-1532 Remove check when decoding invalid UTF-8 strings (mongodb#333)

* Fix typo in objectid.go (mongodb#335)

* GODRIVER-1502 Fix variable shadowing in parsing (mongodb#332)

* GODRIVER-1506 Fix error checking for invalid extjson timestamp values (mongodb#337)

* GODRIVER-1504 Restrict top-level keys from being parsed as extjson (mongodb#340)

* GODRIVER-1513 Disallow escaped single quotes in extjson keys (mongodb#342)

* GODRIVER-1535 Fix session IDs batching in Disconnect

* GODRIVER-1540 fix deadlock in connection (mongodb#348)

* GODRIVER-1549 Select all servers when an empty tag set is given (mongodb#352)

* GODRIVER-1431 Add explicit encryption examples (mongodb#350)

* Update version to v1.3.2

* STITCH-4296: Export contextWithSession

* STITCH-4296: Export sessionFromContext

* STITCH-4680 - expose topology consistency

* STITCH-4841: Add FullDocumentBeforeChange option to Mongo-go-driver

* GODRIVER-672 Change session IDs to be stored as bson.Raw (mongodb#339)

(cherry picked from commit d080bd0)

* do not leak cancelConnectContext

* REALMC-5872 renaming NewSessionContext to ContextWithSession

* REALMC-5872 adding back size check in pool test

* REALMC-5872 removing unused IsConsistent topology function and renaming changed vars

Co-authored-by: Divjot Arora <[email protected]>
Co-authored-by: iwysiu <[email protected]>
Co-authored-by: mengskysama <[email protected]>
Co-authored-by: Luca(Wei) Chen <[email protected]>
Co-authored-by: Haley Owen <[email protected]>
Co-authored-by: Eric Daniels <[email protected]>
Co-authored-by: Tyler Kaye <[email protected]>
Co-authored-by: Jonathan Reams <[email protected]>
Co-authored-by: mike o'brien <[email protected]>
(cherry picked from commit c59bda0)

* REALMC-6991: add topology check

Co-authored-by: Kush Patel <[email protected]>

Author: tsedgwick

mongo/change_stream.go

@@ -365,6 +365,14 @@ func (cs *ChangeStream) createPipelineOptionsDoc() bsoncore.Document {
 		plDoc = bsoncore.AppendStringElement(plDoc, "fullDocument", string(*cs.options.FullDocument))
 	}
 
+	if cs.options.FullDocumentBeforeChange != nil {
+		plDoc = bsoncore.AppendStringElement(
+			plDoc,
+			"fullDocumentBeforeChange",
+			string(*cs.options.FullDocumentBeforeChange),
+		)
+	}
+
 	if cs.options.ResumeAfter != nil {
 		var raDoc bsoncore.Document
 		raDoc, cs.err = transformBsoncoreDocument(cs.registry, cs.options.ResumeAfter, true, "resumeAfter")

mongo/client.go

@@ -926,3 +926,13 @@ func (c *Client) Watch(ctx context.Context, pipeline interface{},
 func (c *Client) NumberSessionsInProgress() int {
 	return c.sessionPool.CheckedOut()
 }
+
+// IsTopologyConsistent returns false when we have a replica set that claims to
+// have no primary but there exists a primary with all other nodes as secondaries. This
+// specifically works around HELP-13825.
+func (c *Client) IsTopologyConsistent() bool {
+	if topo, ok := c.deployment.(*topology.Topology); ok {
+		return topo.IsConsistent()
+	}
+	return true
+}

mongo/options/changestreamoptions.go

@@ -27,6 +27,9 @@ type ChangeStreamOptions struct {
 	// the updated document will not be included in the change notification.
 	FullDocument *FullDocument
 
+	// When set to ‘whenAvailable’ or ‘required’ will have both the delta and the full document before the change was applied
+	FullDocumentBeforeChange *FullDocumentBeforeChange
+
 	// The maximum amount of time that the server should wait for new documents to satisfy a tailable cursor query.
 	MaxAwaitTime *time.Duration
 
@@ -79,6 +82,12 @@ func (cso *ChangeStreamOptions) SetMaxAwaitTime(d time.Duration) *ChangeStreamOp
 	return cso
 }
 
+// SetFullDocumentBeforeChange sets the value for the FullDocumentBeforeChange field.
+func (cso *ChangeStreamOptions) SetFullDocumentBeforeChange(fdbc FullDocumentBeforeChange) *ChangeStreamOptions {
+	cso.FullDocumentBeforeChange = &fdbc
+	return cso
+}
+
 // SetResumeAfter sets the value for the ResumeAfter field.
 func (cso *ChangeStreamOptions) SetResumeAfter(rt interface{}) *ChangeStreamOptions {
 	cso.ResumeAfter = rt
@@ -114,6 +123,9 @@ func MergeChangeStreamOptions(opts ...*ChangeStreamOptions) *ChangeStreamOptions
 		if cso.FullDocument != nil {
 			csOpts.FullDocument = cso.FullDocument
 		}
+		if cso.FullDocumentBeforeChange != nil {
+			csOpts.FullDocumentBeforeChange = cso.FullDocumentBeforeChange
+		}
 		if cso.MaxAwaitTime != nil {
 			csOpts.MaxAwaitTime = cso.MaxAwaitTime
 		}

mongo/options/mongooptions.go

@@ -101,6 +101,23 @@ const (
 	UpdateLookup FullDocument = "updateLookup"
 )
 
+// FullDocumentBeforeChange specifies whether a change stream should include a copy of the entire document that was changed from
+// the time at which the update is applied
+type FullDocumentBeforeChange string
+
+const (
+	// Default / off does not include a document copy
+	FullDocumentBeforeChangeOff FullDocumentBeforeChange = "off"
+
+	// whenAvailable includes a delta describing the changes to the document and a copy of the entire document that
+	// was changed when it is available
+	FullDocumentBeforeChangeWhenAvailable FullDocumentBeforeChange = "whenAvailable"
+
+	// required includes a delta describing the changes to the document and a copy of the entire document that
+	// was changed
+	FullDocumentBeforeChangeRequired FullDocumentBeforeChange = "required"
+)
+
 // ArrayFilters is used to hold filters for the array filters CRUD option. If a registry is nil, bson.DefaultRegistry
 // will be used when converting the filter interfaces to BSON.
 type ArrayFilters struct {

x/mongo/driver/topology/topology.go

@@ -769,4 +769,40 @@ func (t *Topology) publishTopologyClosedEvent() {
 	if t.cfg.serverMonitor != nil && t.cfg.serverMonitor.TopologyClosed != nil {
 		t.cfg.serverMonitor.TopologyClosed(topologyClosed)
 	}
+// isTopologyConsistent returns false when we have a replica set that claims to
+// have no primary but there exists a primary with all other nodes as secondaries. This
+// specifically works around HELP-13825.
+func (t *Topology) IsConsistent() bool {
+	desc := t.Description()
+
+	if desc.Kind != description.ReplicaSetNoPrimary {
+		return true
+	}
+
+	t.serversLock.Lock()
+	serversCopy := make([]*Server, 0, len(t.servers))
+	for _, s := range t.servers {
+		serversCopy = append(serversCopy, s)
+	}
+	t.serversLock.Unlock()
+
+	if len(serversCopy) == 0 || len(serversCopy) == 1 {
+		return true
+	}
+	var foundPrimary bool
+	for _, server := range serversCopy {
+		serverDesc := server.Description()
+		switch serverDesc.Kind {
+		case description.RSPrimary:
+			if foundPrimary {
+				return true
+			}
+			foundPrimary = true
+		case description.RSSecondary:
+		default:
+			return true
+		}
+	}
+
+	return !foundPrimary
 }