GODRIVER-2335 Preemptively cancel in progress operations when SDAM heartbeats timeout.

event/monitoring.go

@@ -106,8 +106,9 @@ type PoolEvent struct {
 	Reason       string              `json:"reason"`
 	// ServiceID is only set if the Type is PoolCleared and the server is deployed behind a load balancer. This field
 	// can be used to distinguish between individual servers in a load balanced deployment.
-	ServiceID *primitive.ObjectID `json:"serviceId"`
-	Error     error               `json:"error"`
+	ServiceID    *primitive.ObjectID `json:"serviceId"`
+	Interruption bool                `json:"interruptInUseConnections"`
+	Error        error               `json:"error"`
 }
 
 // PoolMonitor is a function that allows the user to gain access to events occurring in the pool

internal/eventtest/eventtest.go

@@ -76,3 +76,11 @@ func (tpm *TestPoolMonitor) IsPoolCleared() bool {
 	})
 	return len(poolClearedEvents) > 0
 }
+
+// Interruptions returns the number of interruptions in the events recorded by the testPoolMonitor.
+func (tpm *TestPoolMonitor) Interruptions() int {
+	interruptions := tpm.Events(func(evt *event.PoolEvent) bool {
+		return evt.Interruption
+	})
+	return len(interruptions)
+}

internal/integration/unified/client_entity.go

@@ -571,6 +571,8 @@ func setClientOptionsFromURIOptions(clientOpts *options.ClientOptions, uriOpts b
 		switch strings.ToLower(key) {
 		case "appname":
 			clientOpts.SetAppName(value.(string))
+		case "connecttimeoutms":
+			clientOpts.SetConnectTimeout(time.Duration(value.(int32)) * time.Millisecond)
 		case "heartbeatfrequencyms":
 			clientOpts.SetHeartbeatInterval(time.Duration(value.(int32)) * time.Millisecond)
 		case "loadbalanced":

internal/integration/unified/entity.go

@@ -136,6 +136,7 @@ type EntityMap struct {
 	successValues            map[string]int32
 	iterationValues          map[string]int32
 	clientEncryptionEntities map[string]*mongo.ClientEncryption
+	waitChans                map[string]chan error
 	evtLock                  sync.Mutex
 	closed                   atomic.Value
 	// keyVaultClientIDs tracks IDs of clients used as a keyVaultClient in ClientEncryption objects.
@@ -167,6 +168,7 @@ func newEntityMap() *EntityMap {
 		successValues:            make(map[string]int32),
 		iterationValues:          make(map[string]int32),
 		clientEncryptionEntities: make(map[string]*mongo.ClientEncryption),
+		waitChans:                make(map[string]chan error),
 		keyVaultClientIDs:        make(map[string]bool),
 	}
 	em.setClosed(false)
@@ -283,6 +285,8 @@ func (em *EntityMap) addEntity(ctx context.Context, entityType string, entityOpt
 		err = em.addCollectionEntity(entityOptions)
 	case "session":
 		err = em.addSessionEntity(entityOptions)
+	case "thread":
+		em.waitChans[entityOptions.ID] = make(chan error)
 	case "bucket":
 		err = em.addGridFSBucketEntity(entityOptions)
 	case "clientEncryption":

internal/integration/unified/event_verification.go

@@ -61,7 +61,8 @@ type cmapEvent struct {
 	ConnectionCheckedInEvent *struct{} `bson:"connectionCheckedInEvent"`
 
 	PoolClearedEvent *struct {
-		HasServiceID *bool `bson:"hasServiceId"`
+		HasServiceID              *bool `bson:"hasServiceId"`
+		InterruptInUseConnections *bool `bson:"interruptInUseConnections"`
 	} `bson:"poolClearedEvent"`
 }
 
@@ -361,6 +362,10 @@ func verifyCMAPEvents(client *clientEntity, expectedEvents *expectedEvents) erro
 					return newEventVerificationError(idx, client, "error verifying serviceID: %v", err)
 				}
 			}
+			if expectInterruption := evt.PoolClearedEvent.InterruptInUseConnections; expectInterruption != nil && *expectInterruption != actual.Interruption {
+				return newEventVerificationError(idx, client, "expected interruptInUseConnections %v, got %v",
+					expectInterruption, actual.Interruption)
+			}
 		default:
 			return newEventVerificationError(idx, client, "no expected event set on cmapEvent instance")
 		}

internal/integration/unified/testrunner_operation.go

@@ -50,10 +50,10 @@ func (lp *loopArgs) iterationsStored() bool {
 	return lp.IterationsEntityID != ""
 }
 
-func executeTestRunnerOperation(ctx context.Context, operation *operation, loopDone <-chan struct{}) error {
-	args := operation.Arguments
+func executeTestRunnerOperation(ctx context.Context, op *operation, loopDone <-chan struct{}) error {
+	args := op.Arguments
 
-	switch operation.Name {
+	switch op.Name {
 	case "failPoint":
 		clientID := lookupString(args, "client")
 		client, err := entities(ctx).client(clientID)
@@ -187,9 +187,34 @@ func executeTestRunnerOperation(ctx context.Context, operation *operation, loopD
 			}
 		}
 		return nil
+	case "runOnThread":
+		operationRaw, err := args.LookupErr("operation")
+		if err != nil {
+			return fmt.Errorf("'operation' argument not found in runOnThread operation")
+		}
+		threadOp := new(operation)
+		if err := operationRaw.Unmarshal(threadOp); err != nil {
+			return fmt.Errorf("error unmarshaling 'operation' argument: %v", err)
+		}
+		ch := entities(ctx).waitChans[lookupString(args, "thread")]
+		go func(op *operation) {
+			err := op.execute(ctx, loopDone)
+			ch <- err
+		}(threadOp)
+		return nil
+	case "waitForThread":
+		if ch, ok := entities(ctx).waitChans[lookupString(args, "thread")]; ok {
+			select {
+			case err := <-ch:
+				return err
+			case <-time.After(10 * time.Second):
+				return fmt.Errorf("timed out after 10 seconds")
+			}
+		}
+		return nil
 	case "waitForEvent":
 		var wfeArgs waitForEventArguments
-		if err := bson.Unmarshal(operation.Arguments, &wfeArgs); err != nil {
+		if err := bson.Unmarshal(op.Arguments, &wfeArgs); err != nil {
 			return fmt.Errorf("error unmarshalling event to waitForEventArguments: %v", err)
 		}
 
@@ -198,7 +223,7 @@ func executeTestRunnerOperation(ctx context.Context, operation *operation, loopD
 
 		return waitForEvent(wfeCtx, wfeArgs)
 	default:
-		return fmt.Errorf("unrecognized testRunner operation %q", operation.Name)
+		return fmt.Errorf("unrecognized testRunner operation %q", op.Name)
 	}
 }
 

internal/integration/unified/unified_spec_runner.go

@@ -30,6 +30,11 @@ var (
 		// TODO(GODRIVER-2843): Fix and unskip these test cases.
 		"Find operation with snapshot":                                      "Test fails frequently. See GODRIVER-2843",
 		"Write commands with snapshot session do not affect snapshot reads": "Test fails frequently. See GODRIVER-2843",
+		// The current logic, which was implemented with GODRIVER-2577, only clears pools and cancels in-progress ops if
+		// the heartbeat fails twice. Therefore, we skip the following spec tests, which requires canceling ops immediately.
+		"Connection pool clear uses interruptInUseConnections=true after monitor timeout":                      "Godriver clears after multiple timeout",
+		"Error returned from connection pool clear with interruptInUseConnections=true is retryable":           "Godriver clears after multiple timeout",
+		"Error returned from connection pool clear with interruptInUseConnections=true is retryable for write": "Godriver clears after multiple timeout",
 	}
 
 	logMessageValidatorTimeout = 10 * time.Millisecond

testdata/connection-monitoring-and-pooling/pool-clear-interrupt-immediately.json

@@ -0,0 +1,77 @@
+{
+  "version": 1,
+  "style": "unit",
+  "description": "Connections MUST be interrupted as soon as possible (interruptInUseConnections=true)",
+  "poolOptions": {
+    "backgroundThreadIntervalMS": 10000
+  },
+  "operations": [
+    {
+      "name": "ready"
+    },
+    {
+      "name": "checkOut"
+    },
+    {
+      "name": "checkOut",
+      "label": "conn"
+    },
+    {
+      "name": "clear",
+      "interruptInUseConnections": true
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionPoolCleared",
+      "count": 1,
+      "timeout": 1000
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionClosed",
+      "count": 2,
+      "timeout": 1000
+    },
+    {
+      "name": "close"
+    }
+  ],
+  "events": [
+    {
+      "type": "ConnectionCheckedOut",
+      "connectionId": 1,
+      "address": 42
+    },
+    {
+      "type": "ConnectionCheckedOut",
+      "connectionId": 2,
+      "address": 42
+    },
+    {
+      "type": "ConnectionPoolCleared",
+      "interruptInUseConnections": true
+    },
+    {
+      "type": "ConnectionClosed",
+      "reason": "stale",
+      "address": 42
+    },
+    {
+      "type": "ConnectionClosed",
+      "reason": "stale",
+      "address": 42
+    },
+    {
+      "type": "ConnectionPoolClosed",
+      "address": 42
+    }
+  ],
+  "ignore": [
+    "ConnectionCreated",
+    "ConnectionPoolReady",
+    "ConnectionReady",
+    "ConnectionCheckOutStarted",
+    "ConnectionPoolCreated",
+    "ConnectionCheckedIn"
+  ]
+}

testdata/connection-monitoring-and-pooling/pool-clear-interrupt-immediately.yml

@@ -0,0 +1,46 @@
+version: 1
+style: unit
+description: Connections MUST be interrupted as soon as possible (interruptInUseConnections=true)
+poolOptions:
+  # ensure it's not involved by default
+  backgroundThreadIntervalMS: 10000
+operations:
+  - name: ready
+  - name: checkOut
+  - name: checkOut
+    label: conn
+  - name: clear
+    interruptInUseConnections: true
+  - name: waitForEvent
+    event: ConnectionPoolCleared
+    count: 1
+    timeout: 1000
+  - name: waitForEvent
+    event: ConnectionClosed
+    count: 2
+    timeout: 1000
+  - name: close
+events:
+  - type: ConnectionCheckedOut
+    connectionId: 1
+    address: 42
+  - type: ConnectionCheckedOut
+    connectionId: 2
+    address: 42
+  - type: ConnectionPoolCleared
+    interruptInUseConnections: true
+  - type: ConnectionClosed
+    reason: stale
+    address: 42
+  - type: ConnectionClosed
+    reason: stale
+    address: 42
+  - type: ConnectionPoolClosed
+    address: 42
+ignore:
+  - ConnectionCreated
+  - ConnectionPoolReady
+  - ConnectionReady
+  - ConnectionCheckOutStarted
+  - ConnectionPoolCreated
+  - ConnectionCheckedIn

testdata/connection-monitoring-and-pooling/pool-clear-interrupting-pending-connections.json

@@ -0,0 +1,77 @@
+{
+  "version": 1,
+  "style": "integration",
+  "description": "clear with interruptInUseConnections = true closes pending connections",
+  "runOn": [
+    {
+      "minServerVersion": "4.9.0"
+    }
+  ],
+  "failPoint": {
+    "configureFailPoint": "failCommand",
+    "mode": "alwaysOn",
+    "data": {
+      "failCommands": [
+        "isMaster",
+        "hello"
+      ],
+      "closeConnection": false,
+      "blockConnection": true,
+      "blockTimeMS": 1000
+    }
+  },
+  "poolOptions": {
+    "minPoolSize": 0
+  },
+  "operations": [
+    {
+      "name": "ready"
+    },
+    {
+      "name": "start",
+      "target": "thread1"
+    },
+    {
+      "name": "checkOut",
+      "thread": "thread1"
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionCreated",
+      "count": 1
+    },
+    {
+      "name": "clear",
+      "interruptInUseConnections": true
+    },
+    {
+      "name": "waitForEvent",
+      "event": "ConnectionCheckOutFailed",
+      "count": 1
+    }
+  ],
+  "events": [
+    {
+      "type": "ConnectionCheckOutStarted"
+    },
+    {
+      "type": "ConnectionCreated"
+    },
+    {
+      "type": "ConnectionPoolCleared",
+      "interruptInUseConnections": true
+    },
+    {
+      "type": "ConnectionClosed"
+    },
+    {
+      "type": "ConnectionCheckOutFailed"
+    }
+  ],
+  "ignore": [
+    "ConnectionCheckedIn",
+    "ConnectionCheckedOut",
+    "ConnectionPoolCreated",
+    "ConnectionPoolReady"
+  ]
+}

testdata/connection-monitoring-and-pooling/pool-clear-interrupting-pending-connections.yml

@@ -0,0 +1,42 @@
+version: 1
+style: integration
+description: clear with interruptInUseConnections = true closes pending connections
+runOn:
+  -
+    minServerVersion: "4.9.0"
+failPoint:
+  configureFailPoint: failCommand
+  mode: "alwaysOn"
+  data:
+    failCommands: ["isMaster","hello"]
+    closeConnection: false
+    blockConnection: true
+    blockTimeMS: 1000
+poolOptions:
+  minPoolSize: 0
+operations:
+  - name: ready
+  - name: start
+    target: thread1
+  - name: checkOut
+    thread: thread1
+  - name: waitForEvent
+    event: ConnectionCreated
+    count: 1
+  - name: clear
+    interruptInUseConnections: true
+  - name: waitForEvent
+    event: ConnectionCheckOutFailed
+    count: 1
+events:
+  - type: ConnectionCheckOutStarted
+  - type: ConnectionCreated
+  - type: ConnectionPoolCleared
+    interruptInUseConnections: true
+  - type: ConnectionClosed
+  - type: ConnectionCheckOutFailed
+ignore:
+  - ConnectionCheckedIn
+  - ConnectionCheckedOut
+  - ConnectionPoolCreated
+  - ConnectionPoolReady