change-streams.yml

description: "timeoutMS behaves correctly for change streams"

schemaVersion: "1.9"

runOnRequirements:
  - minServerVersion: "4.4"
    topologies: ["replicaset", "sharded"]

createEntities:
  - client:
      id: &failPointClient failPointClient
      useMultipleMongoses: false
  - client:
      id: &client client
      useMultipleMongoses: false
      observeEvents:
        - commandStartedEvent
      # Drivers are not required to execute killCursors during resume attempts, so it should be ignored for command
      # monitoring assertions.
      ignoreCommandMonitoringEvents: ["killCursors"]
  - database:
      id: &database database
      client: *client
      databaseName: &databaseName test
  - collection:
      id: &collection collection
      database: *database
      collectionName: &collectionName coll

initialData:
  - collectionName: *collectionName
    databaseName: *databaseName
    documents: []

tests:
  - description: "error if maxAwaitTimeMS is greater than timeoutMS"
    operations:
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 5
          maxAwaitTimeMS: 10
        expectError:
          isClientError: true

  - description: "error if maxAwaitTimeMS is equal to timeoutMS"
    operations:
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 5
          maxAwaitTimeMS: 5
        expectError:
          isClientError: true

  - description: "timeoutMS applied to initial aggregate"
    operations:
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 1 }
            data:
              failCommands: ["aggregate"]
              blockConnection: true
              blockTimeMS: 250
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 200
        expectError:
          isTimeoutError: true
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }

  # If maxAwaitTimeMS is not set, timeoutMS should be refreshed for the getMore and the getMore should not have a
  # maxTimeMS field. This test requires a high timeout because the server applies a default 1000ms maxAwaitTime. To
  # ensure that the driver is refreshing the timeout between commands, the test blocks aggregate and getMore commands
  # for 30ms each and creates/iterates a change stream with timeoutMS=1050. The initial aggregate will block for 30ms
  # and the getMore will block for 1030ms.
  - description: "timeoutMS is refreshed for getMore if maxAwaitTimeMS is not set"
    operations:
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 2 }
            data:
              failCommands: ["aggregate", "getMore"]
              blockConnection: true
              blockTimeMS: 30
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 1050
        saveResultAsEntity: &changeStream changeStream
      - name: iterateOnce
        object: *changeStream
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }
          - commandStartedEvent:
              commandName: getMore
              databaseName: *databaseName
              command:
                getMore: { $$type: ["int", "long"] }
                collection: *collectionName
                maxTimeMS: { $$exists: false }

  # If maxAwaitTimeMS is set, timeoutMS should still be refreshed for the getMore and the getMore command should have a
  # maxTimeMS field.
  - description: "timeoutMS is refreshed for getMore if maxAwaitTimeMS is set"
    operations:
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 2 }
            data:
              failCommands: ["aggregate", "getMore"]
              blockConnection: true
              blockTimeMS: 150
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 200
          batchSize: 2
          maxAwaitTimeMS: 1
        saveResultAsEntity: &changeStream changeStream
      - name: iterateOnce
        object: *changeStream
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }
          - commandStartedEvent:
              commandName: getMore
              databaseName: *databaseName
              command:
                getMore: { $$type: ["int", "long"] }
                collection: *collectionName
                maxTimeMS: 1

  # If a resume is required for a next call on a change stream, the timeout MUST apply to the entirety of the initial
  # getMore and all commands sent as part of the resume attempt.
  - description: "timeoutMS applies to full resume attempt in a next call"
    operations:
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 200
        saveResultAsEntity: &changeStream changeStream
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 2 }
            data:
              failCommands: ["getMore", "aggregate"]
              blockConnection: true
              blockTimeMS: 120
              errorCode: 7 # HostNotFound - resumable but does not require an SDAM state change.
              # failCommand doesn't correctly add the ResumableChangeStreamError by default. It needs to be specified
              # manually here so the error is considered resumable. The failGetMoreAfterCursorCheckout fail point
              # would add the label without this, but it does not support blockConnection functionality.
              errorLabels: ["ResumableChangeStreamError"]
      - name: iterateUntilDocumentOrError
        object: *changeStream
        expectError:
          isTimeoutError: true
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }
          - commandStartedEvent:
              commandName: getMore
              databaseName: *databaseName
              command:
                getMore: { $$type: ["int", "long"] }
                collection: *collectionName
                maxTimeMS: { $$exists: false }
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }

  # [resumption] If a next call fails with a timeout error, drivers MUST NOT
  # invalidate the change stream. The subsequent next call MUST perform a resume
  # attempt to establish a new change stream on the server.
  #
  # [refresh] If a resume is required for a next call on a change stream, the
  # timeout MUST apply to the entirety of the initial getMore and all commands
  # sent as part of the resume attempt.
  - description: "change stream iteration succeeds after a timeout error"
    operations:
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          # Specify a short maxAwaitTimeMS because otherwise the getMore on the new cursor will wait for 1000ms and
          # time out.
          maxAwaitTimeMS: 1
          timeoutMS: 200
        saveResultAsEntity: &changeStream changeStream
      # Block getMore for 250ms to force the next() call to time out.
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 1 }
            data:
              failCommands: ["getMore"]
              blockConnection: true
              blockTimeMS: 250
      # Iterate until timeout.
      - name: iterateUntilDocumentOrError
        object: *changeStream
        expectError:
          isTimeoutError: true
      # Block the resume aggregate for 150ms. If the timeout were exhausted from
      # the previous call, this would fail immediately. With a fresh 200ms
      # timeout, the 150ms delay should be acceptable.
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 1 }
            data:
              failCommands: ["aggregate"]
              blockConnection: true
              blockTimeMS: 150
      # A final iteration should succeed because the timeout is refreshed and
      # the change stream was not invalidated.
      - name: iterateOnce
        object: *changeStream
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }
          # The first iterateUntilDocumentOrError sends a getMore that times
          # out.
          - commandStartedEvent:
              commandName: getMore
              databaseName: *databaseName
              command:
                getMore: { $$type: ["int", "long"] }
                collection: *collectionName
          # The iterateOnce operation re-creates the cursor via an aggregate
          # [resumption]. The aggregate is blocked for 150ms but succeeds
          # because the timeout was refreshed to a fresh 200ms, proving the
          # timeout is not shared with the previous timed-out call [refresh].
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }

  # The timeoutMS value should be refreshed for getMore's. This is a failure test. The createChangeStream operation
  # sets timeoutMS=200 and the getMore blocks for 250ms, causing iteration to fail with a timeout error.
  - description: "timeoutMS is refreshed for getMore - failure"
    operations:
      - name: failPoint
        object: testRunner
        arguments:
          client: *failPointClient
          failPoint:
            configureFailPoint: failCommand
            mode: { times: 1 }
            data:
              failCommands: ["getMore"]
              blockConnection: true
              blockTimeMS: 250
      - name: createChangeStream
        object: *collection
        arguments:
          pipeline: []
          timeoutMS: 200
        saveResultAsEntity: &changeStream changeStream
      # The first iteration should do a getMore
      - name: iterateUntilDocumentOrError
        object: *changeStream
        expectError:
          isTimeoutError: true
    expectEvents:
      - client: *client
        events:
          - commandStartedEvent:
              commandName: aggregate
              databaseName: *databaseName
              command:
                aggregate: *collectionName
                maxTimeMS: { $$type: ["int", "long"] }
          # The iterateUntilDocumentOrError operation should send a getMore.
          - commandStartedEvent:
              commandName: getMore
              databaseName: *databaseName
              command:
                getMore: { $$type: ["int", "long"] }
                collection: *collectionName