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

mongo/integration/unified/client_entity.go

@@ -568,6 +568,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.Microsecond)
  case "heartbeatfrequencyms":
  clientOpts.SetHeartbeatInterval(time.Duration(value.(int32)) * time.Millisecond)
  case "loadbalanced":

mongo/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":

mongo/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")
  }

mongo/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 unmarshalling '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 <-ch:
+ return nil
+ 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)
  }
 }
 

mongo/integration/unified/unified_spec_runner.go

@@ -32,6 +32,10 @@ var (
  "Write commands with snapshot session do not affect snapshot reads": "Test fails frequently. See GODRIVER-2843",
  // TODO(GODRIVER-2943): Fix and unskip this test case.
  "Topology lifecycle": "Test times out. See GODRIVER-2943",
+
+ "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