Query blocks with ChangeStream

[jinq0123]

We use the issue tracker to track bugs with mgo - if you have a usage question, it's best to try Stack Overflow :)

Replace this text with your description, and please answer the questions below before submitting your issue to help us out. Thanks!

---

What version of MongoDB are you using (mongod --version)?

db version v4.2.0
git version: a4b751dcf51dd249c5865812b390cfd1c0129c30
allocator: tcmalloc
modules: none
build environment:
    distmod: 2012plus
    distarch: x86_64
    target_arch: x86_64

What version of Go are you using (go version)?

go version go1.12.7 windows/amd64

What operating system and processor architecture are you using (go env)?

set GOARCH=amd64
set GOBIN=
set GOCACHE=C:\Users\jinqing\AppData\Local\go-build
set GOEXE=.exe
set GOFLAGS=
set GOHOSTARCH=amd64
set GOHOSTOS=windows
set GOOS=windows
set GOPATH=D:\gopath
set GOPROXY=https://goproxy.cn
set GORACE=
set GOROOT=D:\Go
set GOTMPDIR=
set GOTOOLDIR=D:\Go\pkg\tool\windows_amd64
set GCCGO=gccgo
set CC=gcc
set CXX=g++
set CGO_ENABLED=1
set GOMOD=
set CGO_CFLAGS=-g -O2
set CGO_CPPFLAGS=
set CGO_CXXFLAGS=-g -O2
set CGO_FFLAGS=-g -O2
set CGO_LDFLAGS=-g -O2
set PKG_CONFIG=pkg-config
set GOGCCFLAGS=-m64 -mthreads -fmessage-length=0 -fdebug-prefix-map=C:\Users\jinqing\AppData\Local\Temp\go-build075823922=/tmp/go-build -gno-record-gcc-switches

What did you do?

The program watches a change stream in backgroud, and the query will block.

Setup a local mongodb:

mkdir d:\data\db
D:\Tool\mongodb-win32-x86_64-2012plus-4.2.0\bin
λ mongo --nodb
MongoDB shell version v4.2.0
> cluster = new SharingTest({shards: 3})

Then run the test program:

package main

import (
    "fmt"
    "os"
    "time"

    "github.com/globalsign/mgo"
)

func main() {
    session, err := mgo.Dial("mongodb://localhost:20006/test")
    if err != nil {
        fmt.Println("Dial error: ", err)
        return
    }

    go loopWatch(session)
    for {
        time.Sleep(time.Second)
        query(session)
    }
}

func query(session *mgo.Session) {
    c := session.DB("").C("test1")
    result := struct{}{}

    startTime := time.Now()
    err := c.Find(nil).One(&result)
    cost := time.Since(startTime)
    fmt.Printf("Query takes: %v, result: %v, err: %v\n", cost, result, err)
}

func loopWatch(session *mgo.Session) {
    for {
        watch(session)
    }
}

func watch(session *mgo.Session) {
    coll := session.DB("").C("test2")
    changeStream, err := coll.Watch(nil, mgo.ChangeStreamOptions{
        MaxAwaitTimeMS: 5 * time.Second,
    })
    if err != nil {
        fmt.Println("failed to watch: ", err)
        os.Exit(1)
    }
    defer changeStream.Close()

    //Handling change stream in a cycle
    for {
        //making a struct for unmarshalling
        changeDoc := struct {
            OperationType string `json:"operationType" bson:"operationType"`
        }{}

        //getting next item from the steam
        fmt.Println("Change stream next...")
        ok := changeStream.Next(&changeDoc)
        if ok {
            fmt.Println("got change")
            continue
        }

        //if data from the stream wasn't unmarshaled, we get ok == false as a result
        //so we need to call Err() method to get info why
        //it'll be nil if we just have no data
        err := changeStream.Err()
        if err == nil {
            continue
        }

        //if err is not nil, it means something bad happened, let's finish our func
        return
    } // for
}

Output:

D:\jinqing\Test\go\mgochg
λ mgochg.exe
Change stream next...
Change stream next...
Query takes: 4.0112788s, result: {}, err: not found
Change stream next...
Query takes: 4.0022317s, result: {}, err: not found
Change stream next...
Query takes: 4.0022971s, result: {}, err: not found
Change stream next...
Query takes: 4.0003046s, result: {}, err: not found
Change stream next...
Query takes: 4.0032182s, result: {}, err: not found
Change stream next...
Query takes: 4.0043058s, result: {}, err: not found
...

Can you reproduce the issue on the latest development branch?

[fananchong]

The correct way to use should look like this:

// Copy the session - if needed this will dial a new connection which
// can later be reused.
//
// Calling close returns the connection to the pool.
conn := session.Copy()
defer conn.Close()

// Do something(s) with the connection
_, _ = conn.DB("").C("my_data").Count()

[jinq0123]

Same result using copied session in Eventual mode:

    session.SetMode(mgo.Eventual, true)