csot_prose_test.go

Documentation: go.mongodb.org/mongo-driver/mongo/integration

     1  // Copyright (C) MongoDB, Inc. 2022-present.
     2  //
     3  // Licensed under the Apache License, Version 2.0 (the "License"); you may
     4  // not use this file except in compliance with the License. You may obtain
     5  // a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
     6  
     7  package integration
     8  
     9  import (
    10  	"context"
    11  	"strings"
    12  	"testing"
    13  	"time"
    14  
    15  	"go.mongodb.org/mongo-driver/bson"
    16  	"go.mongodb.org/mongo-driver/event"
    17  	"go.mongodb.org/mongo-driver/internal/assert"
    18  	"go.mongodb.org/mongo-driver/internal/integtest"
    19  	"go.mongodb.org/mongo-driver/mongo"
    20  	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
    21  	"go.mongodb.org/mongo-driver/mongo/options"
    22  )
    23  
    24  func TestCSOTProse(t *testing.T) {
    25  	mt := mtest.New(t, mtest.NewOptions().CreateClient(false))
    26  
    27  	mt.RunOpts("1. multi-batch writes", mtest.NewOptions().MinServerVersion("4.4").
    28  		Topologies(mtest.Single), func(mt *mtest.T) {
    29  		// Test that multi-batch writes do not refresh the Timeout between batches.
    30  
    31  		err := mt.Client.Database("db").Collection("coll").Drop(context.Background())
    32  		assert.Nil(mt, err, "Drop error: %v", err)
    33  
    34  		// Configure a fail point to block both inserts of the multi-write for 1010ms (2020ms total).
    35  		mt.SetFailPoint(mtest.FailPoint{
    36  			ConfigureFailPoint: "failCommand",
    37  			Mode: mtest.FailPointMode{
    38  				Times: 2,
    39  			},
    40  			Data: mtest.FailPointData{
    41  				FailCommands:    []string{"insert"},
    42  				BlockConnection: true,
    43  				BlockTimeMS:     1010,
    44  			},
    45  		})
    46  
    47  		// Use a separate client with 2s Timeout and a separate command monitor to run a multi-batch
    48  		// insert against db.coll.
    49  		var started []*event.CommandStartedEvent
    50  		cm := &event.CommandMonitor{
    51  			Started: func(_ context.Context, evt *event.CommandStartedEvent) {
    52  				started = append(started, evt)
    53  			},
    54  		}
    55  		cliOptions := options.Client().
    56  			SetTimeout(2 * time.Second).
    57  			SetMonitor(cm).
    58  			ApplyURI(mtest.ClusterURI())
    59  		integtest.AddTestServerAPIVersion(cliOptions)
    60  		cli, err := mongo.Connect(context.Background(), cliOptions)
    61  		assert.Nil(mt, err, "Connect error: %v", err)
    62  
    63  		// Insert 50 1MB documents (OP_MSG payloads can only fit 48MB in one batch).
    64  		var bigStringBuilder strings.Builder
    65  		for i := 0; i < 1024*1024; i++ {
    66  			bigStringBuilder.WriteByte('a')
    67  		}
    68  		bigString := bigStringBuilder.String()
    69  		var docs []interface{}
    70  		for i := 0; i < 50; i++ {
    71  			docs = append(docs, bson.D{{"1mb", bigString}})
    72  		}
    73  
    74  		// Expect a timeout error from InsertMany (from the second batch).
    75  		_, err = cli.Database("db").Collection("coll").InsertMany(context.Background(), docs)
    76  		assert.NotNil(mt, err, "expected error from InsertMany, got nil")
    77  		assert.True(mt, mongo.IsTimeout(err), "expected error to be a timeout, got %v", err)
    78  
    79  		// Expect that two 'insert's were sent.
    80  		assert.True(mt, len(started) == 2, "expected two started events, got %d", len(started))
    81  		assert.Equal(mt, started[0].CommandName,
    82  			"insert", "expected an insert event, got %v", started[0].CommandName)
    83  		assert.Equal(mt, started[1].CommandName,
    84  			"insert", "expected a second insert event, got %v", started[1].CommandName)
    85  	})
    86  	mt.Run("8. server selection", func(mt *mtest.T) {
    87  		cliOpts := options.Client().ApplyURI("mongodb://invalid/?serverSelectionTimeoutMS=100")
    88  		mtOpts := mtest.NewOptions().ClientOptions(cliOpts).CreateCollection(false)
    89  		mt.RunOpts("serverSelectionTimeoutMS honored if timeoutMS is not set", mtOpts, func(mt *mtest.T) {
    90  			mt.Parallel()
    91  
    92  			callback := func(ctx context.Context) {
    93  				err := mt.Client.Ping(ctx, nil)
    94  				assert.NotNil(mt, err, "expected Ping error, got nil")
    95  			}
    96  
    97  			// Assert that Ping fails within 150ms due to server selection timeout.
    98  			assert.Soon(mt, callback, 150*time.Millisecond)
    99  		})
   100  
   101  		cliOpts = options.Client().ApplyURI("mongodb://invalid/?timeoutMS=100&serverSelectionTimeoutMS=200")
   102  		mtOpts = mtest.NewOptions().ClientOptions(cliOpts).CreateCollection(false)
   103  		mt.RunOpts("timeoutMS honored for server selection if it's lower than serverSelectionTimeoutMS", mtOpts, func(mt *mtest.T) {
   104  			mt.Parallel()
   105  
   106  			callback := func(ctx context.Context) {
   107  				err := mt.Client.Ping(ctx, nil)
   108  				assert.NotNil(mt, err, "expected Ping error, got nil")
   109  			}
   110  
   111  			// Assert that Ping fails within 150ms due to timeout.
   112  			assert.Soon(mt, callback, 150*time.Millisecond)
   113  		})
   114  
   115  		cliOpts = options.Client().ApplyURI("mongodb://invalid/?timeoutMS=200&serverSelectionTimeoutMS=100")
   116  		mtOpts = mtest.NewOptions().ClientOptions(cliOpts).CreateCollection(false)
   117  		mt.RunOpts("serverSelectionTimeoutMS honored for server selection if it's lower than timeoutMS", mtOpts, func(mt *mtest.T) {
   118  			mt.Parallel()
   119  
   120  			callback := func(ctx context.Context) {
   121  				err := mt.Client.Ping(ctx, nil)
   122  				assert.NotNil(mt, err, "expected Ping error, got nil")
   123  			}
   124  
   125  			// Assert that Ping fails within 150ms due to server selection timeout.
   126  			assert.Soon(mt, callback, 150*time.Millisecond)
   127  		})
   128  
   129  		cliOpts = options.Client().ApplyURI("mongodb://invalid/?timeoutMS=0&serverSelectionTimeoutMS=100")
   130  		mtOpts = mtest.NewOptions().ClientOptions(cliOpts).CreateCollection(false)
   131  		mt.RunOpts("serverSelectionTimeoutMS honored for server selection if timeoutMS=0", mtOpts, func(mt *mtest.T) {
   132  			mt.Parallel()
   133  
   134  			callback := func(ctx context.Context) {
   135  				err := mt.Client.Ping(ctx, nil)
   136  				assert.NotNil(mt, err, "expected Ping error, got nil")
   137  			}
   138  
   139  			// Assert that Ping fails within 150ms due to server selection timeout.
   140  			assert.Soon(mt, callback, 150*time.Millisecond)
   141  		})
   142  	})
   143  }
   144
View as plain text