1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
// Copyright (C) MongoDB, Inc. 2017-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
package integration
import (
"bytes"
"testing"
"go.mongodb.org/mongo-driver/bson"
"go.mongodb.org/mongo-driver/internal/testutil/assert"
"go.mongodb.org/mongo-driver/mongo"
"go.mongodb.org/mongo-driver/mongo/integration/mtest"
"go.mongodb.org/mongo-driver/mongo/options"
"go.mongodb.org/mongo-driver/x/mongo/driver"
)
func TestRetryableWritesProse(t *testing.T) {
clientOpts := options.Client().SetRetryWrites(true).SetWriteConcern(mtest.MajorityWc).
SetReadConcern(mtest.MajorityRc)
mtOpts := mtest.NewOptions().ClientOptions(clientOpts).MinServerVersion("3.6").CreateClient(false)
mt := mtest.New(t, mtOpts)
defer mt.Close()
includeOpts := mtest.NewOptions().Topologies(mtest.ReplicaSet, mtest.Sharded).CreateClient(false)
mt.RunOpts("txn number included", includeOpts, func(mt *mtest.T) {
updateDoc := bson.D{{"$inc", bson.D{{"x", 1}}}}
insertOneDoc := bson.D{{"x", 1}}
insertManyOrderedArgs := bson.D{
{"options", bson.D{{"ordered", true}}},
{"documents", []interface{}{insertOneDoc}},
}
insertManyUnorderedArgs := bson.D{
{"options", bson.D{{"ordered", true}}},
{"documents", []interface{}{insertOneDoc}},
}
testCases := []struct {
operationName string
args bson.D
expectTxnNumber bool
}{
{"deleteOne", bson.D{}, true},
{"deleteMany", bson.D{}, false},
{"updateOne", bson.D{{"update", updateDoc}}, true},
{"updateMany", bson.D{{"update", updateDoc}}, false},
{"replaceOne", bson.D{}, true},
{"insertOne", bson.D{{"document", insertOneDoc}}, true},
{"insertMany", insertManyOrderedArgs, true},
{"insertMany", insertManyUnorderedArgs, true},
{"findOneAndReplace", bson.D{}, true},
{"findOneAndUpdate", bson.D{{"update", updateDoc}}, true},
{"findOneAndDelete", bson.D{}, true},
}
for _, tc := range testCases {
mt.Run(tc.operationName, func(mt *mtest.T) {
tcArgs, err := bson.Marshal(tc.args)
assert.Nil(mt, err, "Marshal error: %v", err)
crudOp := crudOperation{
Name: tc.operationName,
Arguments: tcArgs,
}
mt.ClearEvents()
runCrudOperation(mt, "", crudOp, crudOutcome{})
started := mt.GetStartedEvent()
assert.NotNil(mt, started, "expected CommandStartedEvent, got nil")
_, err = started.Command.LookupErr("txnNumber")
if tc.expectTxnNumber {
assert.Nil(mt, err, "expected txnNumber in command %v", started.Command)
return
}
assert.NotNil(mt, err, "did not expect txnNumber in command %v", started.Command)
})
}
})
errorOpts := mtest.NewOptions().Topologies(mtest.ReplicaSet, mtest.Sharded)
mt.RunOpts("wrap mmapv1 error", errorOpts, func(mt *mtest.T) {
res, err := mt.DB.RunCommand(mtest.Background, bson.D{{"serverStatus", 1}}).DecodeBytes()
assert.Nil(mt, err, "serverStatus error: %v", err)
storageEngine, ok := res.Lookup("storageEngine", "name").StringValueOK()
if !ok || storageEngine != "mmapv1" {
mt.Skip("skipping because storage engine is not mmapv1")
}
_, err = mt.Coll.InsertOne(mtest.Background, bson.D{{"x", 1}})
assert.Equal(mt, driver.ErrUnsupportedStorageEngine, err,
"expected error %v, got %v", driver.ErrUnsupportedStorageEngine, err)
})
standaloneOpts := mtest.NewOptions().Topologies(mtest.Single).CreateClient(false)
mt.RunOpts("transaction number not sent on writes", standaloneOpts, func(mt *mtest.T) {
mt.Run("explicit session", func(mt *mtest.T) {
// Standalones do not support retryable writes and will error if a transaction number is sent
sess, err := mt.Client.StartSession()
assert.Nil(mt, err, "StartSession error: %v", err)
defer sess.EndSession(mtest.Background)
mt.ClearEvents()
err = mongo.WithSession(mtest.Background, sess, func(ctx mongo.SessionContext) error {
doc := bson.D{{"foo", 1}}
_, err := mt.Coll.InsertOne(ctx, doc)
return err
})
assert.Nil(mt, err, "InsertOne error: %v", err)
_, wantID := sess.ID().Lookup("id").Binary()
command := mt.GetStartedEvent().Command
lsid, err := command.LookupErr("lsid")
assert.Nil(mt, err, "Error getting lsid: %v", err)
_, gotID := lsid.Document().Lookup("id").Binary()
assert.True(mt, bytes.Equal(wantID, gotID), "expected session ID %v, got %v", wantID, gotID)
txnNumber, err := command.LookupErr("txnNumber")
assert.NotNil(mt, err, "expected no txnNumber, got %v", txnNumber)
})
mt.Run("implicit session", func(mt *mtest.T) {
// Standalones do not support retryable writes and will error if a transaction number is sent
mt.ClearEvents()
doc := bson.D{{"foo", 1}}
_, err := mt.Coll.InsertOne(mtest.Background, doc)
assert.Nil(mt, err, "InsertOne error: %v", err)
command := mt.GetStartedEvent().Command
lsid, err := command.LookupErr("lsid")
assert.Nil(mt, err, "Error getting lsid: %v", err)
_, gotID := lsid.Document().Lookup("id").Binary()
assert.NotNil(mt, gotID, "expected session ID, got nil")
txnNumber, err := command.LookupErr("txnNumber")
assert.NotNil(mt, err, "expected no txnNumber, got %v", txnNumber)
})
})
}
|
