Skip to content

Commit de8c1e4

Browse files
FGasperFGasper
committed
save
1 parent c20c1ba commit de8c1e4

File tree

6 files changed

+180
-55
lines changed

6 files changed

+180
-55
lines changed

‎internal/verifier/check.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,14 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
150150
if err != nil {
151151
return err
152152
}
153+
154+
err = verifier.doInMetaTransaction(
155+
ctx,
156+
func(ctx context.Context, sCtx mongo.SessionContext) error {
157+
return verifier.ResetInProgressTasks(sCtx)
158+
},
159+
)
160+
153161
verifier.logger.Debug().Msg("Starting Check")
154162

155163
verifier.phase = Check

‎internal/verifier/migration_verifier_test.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@ import (
3131

3232
var macArmMongoVersions []string = []string{
3333
"6.2.0", "6.0.1",
34+
"7.0.15",
35+
"8.0.1",
3436
}
3537

3638
var preMacArmMongoVersions []string = []string{

‎internal/verifier/reset.go‎

Lines changed: 49 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -6,42 +6,64 @@ import (
66
"go.mongodb.org/mongo-driver/mongo"
77
)
88

9+
var defaultTaskUpdate = bson.M{
10+
"$set": bson.M{"status": verificationTaskAdded},
11+
"$unset": bson.M{"begin_time": 1},
12+
}
13+
914
func (verifier *Verifier) ResetInProgressTasks(ctx mongo.SessionContext) error {
10-
err := verifier.resetPrimaryTaskIfNeeded(ctx)
15+
didReset, err := verifier.handleIncompletePrimary(ctx)
16+
17+
if err == nil {
18+
if didReset {
19+
return nil
20+
}
1121

12-
if err != nil {
1322
err = verifier.resetCollectionTasksIfNeeded(ctx)
1423
}
1524

16-
if err != nil {
25+
if err == nil {
1726
err = verifier.resetPartitionTasksIfNeeded(ctx)
1827
}
1928

2029
return err
2130
}
2231

23-
func (verifier *Verifier) resetPrimaryTaskIfNeeded(ctx mongo.SessionContext) error {
32+
func (verifier *Verifier) handleIncompletePrimary(ctx mongo.SessionContext) (bool, error) {
2433
taskColl := verifier.verificationTaskCollection()
2534

26-
incompletePrimary, err := taskColl.CountDocuments(
35+
cursor, err := taskColl.Find(
2736
ctx,
2837
bson.M{
2938
"type": verificationTaskPrimary,
30-
"status": verificationTaskProcessing,
39+
"status": bson.M{"$ne": verificationTaskCompleted},
3140
},
3241
)
3342
if err != nil {
34-
return errors.Wrapf(err, "failed to count incomplete %#q tasks", verificationTaskPrimary)
43+
return false, errors.Wrapf(err, "failed to fetch incomplete %#q task", verificationTaskPrimary)
3544
}
3645

37-
switch incompletePrimary {
46+
var incompletePrimaries []VerificationTask
47+
err = cursor.All(ctx, &incompletePrimaries)
48+
if err != nil {
49+
return false, errors.Wrapf(err, "failed to read incomplete %#q task", verificationTaskPrimary)
50+
}
51+
52+
switch len(incompletePrimaries) {
3853
case 0:
3954
// Nothing to do.
4055
case 1:
56+
// Invariant: task status should be “added”.
57+
if incompletePrimaries[0].Status != verificationTaskAdded {
58+
verifier.logger.Panic().
59+
Interface("task", incompletePrimaries[0]).
60+
Msg("Primary task status has invalid state.")
61+
}
62+
4163
verifier.logger.Info().
42-
Msg("Previous verifier run left primary task incomplete. Resetting.")
64+
Msg("Previous verifier run left primary task incomplete. Deleting non-primary tasks.")
4365

44-
_, err := taskColl.DeleteMany(
66+
deleted, err := taskColl.DeleteMany(
4567
ctx,
4668
bson.M{
4769
"type": bson.M{
@@ -50,30 +72,21 @@ func (verifier *Verifier) resetPrimaryTaskIfNeeded(ctx mongo.SessionContext) err
5072
},
5173
)
5274
if err != nil {
53-
return errors.Wrapf(err, "failed to delete non-%#q tasks", verificationTaskPrimary)
75+
return false, errors.Wrapf(err, "failed to delete non-%#q tasks", verificationTaskPrimary)
5476
}
5577

56-
_, err = taskColl.UpdateOne(
57-
ctx,
58-
bson.M{
59-
"type": verificationTaskPrimary,
60-
},
61-
bson.M{
62-
"$set": bson.M{"status": verificationTaskAdded},
63-
},
64-
)
65-
if err != nil {
66-
return errors.Wrapf(err, "failed to reset %#q task", verificationTaskPrimary)
67-
}
78+
verifier.logger.Info().
79+
Int64("deletedTasksCount", deleted.DeletedCount).
80+
Msg("Found and deleted non-primary tasks.")
6881

69-
return nil
82+
return true, nil
7083
default:
7184
verifier.logger.Panic().
72-
Int64("incompletePrimary", incompletePrimary).
85+
Interface("tasks", incompletePrimaries).
7386
Msg("Found multiple incomplete primary tasks; there should only be 1.")
7487
}
7588

76-
return nil
89+
return false, nil
7790
}
7891

7992
func (verifier *Verifier) resetCollectionTasksIfNeeded(ctx mongo.SessionContext) error {
@@ -95,12 +108,18 @@ func (verifier *Verifier) resetCollectionTasksIfNeeded(ctx mongo.SessionContext)
95108
return errors.Wrapf(err, "failed to read incomplete %#q tasks", verificationTaskVerifyCollection)
96109
}
97110

111+
if len(incompleteCollTasks) > 0 {
112+
verifier.logger.Info().
113+
Int("count", len(incompleteCollTasks)).
114+
Msg("Previous verifier run left collection-level verification task(s) pending. Resetting.")
115+
}
116+
98117
for _, task := range incompleteCollTasks {
99118
_, err := taskColl.DeleteMany(
100119
ctx,
101120
bson.M{
102-
"type": verificationTaskVerifyDocuments,
103-
"query_filter.partition.namespace": task.QueryFilter.Namespace,
121+
"type": verificationTaskVerifyDocuments,
122+
"query_filter.namespace": task.QueryFilter.Namespace,
104123
},
105124
)
106125
if err != nil {
@@ -113,10 +132,7 @@ func (verifier *Verifier) resetCollectionTasksIfNeeded(ctx mongo.SessionContext)
113132
"type": verificationTaskVerifyCollection,
114133
"query_filter.namespace": task.QueryFilter.Namespace,
115134
},
116-
bson.M{
117-
"$set": bson.M{"status": verificationTaskAdded},
118-
// TODO unset
119-
},
135+
defaultTaskUpdate,
120136
)
121137
if err != nil {
122138
return errors.Wrapf(err, "failed to reset namespace %#q's %#q task", task.QueryFilter.Namespace, verificationTaskVerifyCollection)
@@ -135,10 +151,7 @@ func (verifier *Verifier) resetPartitionTasksIfNeeded(ctx mongo.SessionContext)
135151
"type": verificationTaskVerifyDocuments,
136152
"status": verificationTaskProcessing,
137153
},
138-
bson.M{
139-
"$set": bson.M{"status": verificationTaskAdded},
140-
// TODO unset
141-
},
154+
defaultTaskUpdate,
142155
)
143156
if err != nil {
144157
return errors.Wrapf(err, "failed to reset in-progress %#q tasks", verificationTaskVerifyDocuments)

‎internal/verifier/reset_test.go‎

Lines changed: 99 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,99 @@
1+
package verifier
2+
3+
import (
4+
"context"
5+
6+
"go.mongodb.org/mongo-driver/bson"
7+
"go.mongodb.org/mongo-driver/mongo"
8+
)
9+
10+
func (suite *MultiDataVersionTestSuite) TestResetPrimaryTask() {
11+
verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
12+
13+
created, err := verifier.CheckIsPrimary()
14+
suite.Require().NoError(err)
15+
suite.Require().True(created)
16+
17+
_, err = verifier.InsertCollectionVerificationTask("foo.bar")
18+
suite.Require().NoError(err)
19+
20+
ctx := context.Background()
21+
22+
err = verifier.doInMetaTransaction(
23+
ctx,
24+
func(_ context.Context, ctx mongo.SessionContext) error {
25+
return verifier.ResetInProgressTasks(ctx)
26+
},
27+
)
28+
suite.Require().NoError(err)
29+
30+
tasksColl := verifier.verificationTaskCollection()
31+
cursor, err := tasksColl.Find(ctx, bson.M{})
32+
suite.Require().NoError(err)
33+
var taskDocs []bson.M
34+
suite.Require().NoError(cursor.All(ctx, &taskDocs))
35+
36+
suite.Assert().Len(taskDocs, 1)
37+
}
38+
39+
func (suite *MultiDataVersionTestSuite) TestResetCollectionTasks() {
40+
verifier := buildVerifier(suite.T(), suite.srcMongoInstance, suite.dstMongoInstance, suite.metaMongoInstance)
41+
42+
// Create a primary task, and set it to complete.
43+
created, err := verifier.CheckIsPrimary()
44+
suite.Require().NoError(err)
45+
suite.Require().True(created)
46+
47+
suite.Require().NoError(verifier.UpdatePrimaryTaskComplete())
48+
49+
ns1 := "foo.bar"
50+
ns2 := "qux.quux"
51+
52+
// Create a collection-verification task, and set it to processing.
53+
collTask, err := verifier.InsertCollectionVerificationTask(ns1)
54+
suite.Require().NoError(err)
55+
56+
collTask.Status = verificationTaskProcessing
57+
58+
suite.Require().NoError(
59+
verifier.UpdateVerificationTask(collTask),
60+
)
61+
62+
// Create four partition tasks: three with the same namespace as the
63+
// collection-verification task (status=[added, processing, completed]),
64+
// and another for a different namespace that’s completed.
65+
for _, status := range []verificationTaskStatus{
66+
verificationTaskAdded,
67+
verificationTaskProcessing,
68+
verificationTaskCompleted,
69+
} {
70+
task, err := verifier.InsertPartitionVerificationTask(
71+
&partitions.Partition{
72+
Ns: &partition.NameSpace{
73+
DB: "foo",
74+
Coll: "bar",
75+
},
76+
},
77+
nil,
78+
"foo.bar",
79+
)
80+
}
81+
82+
ctx := context.Background()
83+
84+
err = verifier.doInMetaTransaction(
85+
ctx,
86+
func(_ context.Context, ctx mongo.SessionContext) error {
87+
return verifier.ResetInProgressTasks(ctx)
88+
},
89+
)
90+
suite.Require().NoError(err)
91+
92+
tasksColl := verifier.verificationTaskCollection()
93+
cursor, err := tasksColl.Find(ctx, bson.M{})
94+
suite.Require().NoError(err)
95+
var taskDocs []bson.M
96+
suite.Require().NoError(cursor.All(ctx, &taskDocs))
97+
98+
suite.Assert().Len(taskDocs, 1)
99+
}

‎internal/verifier/unit_test_util.go‎

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -91,7 +91,7 @@ func (suite *WithMongodsTestSuite) SetupSuite() {
9191
suite.Require().NoError(err)
9292
clientOpts = options.Client().ApplyURI("mongodb://localhost:" + suite.metaMongoInstance.port).SetAppName("Verifier Test Suite")
9393
suite.metaMongoClient, err = mongo.Connect(ctx, clientOpts)
94-
suite.startReplSet()
94+
suite.startReplSets()
9595
suite.Require().NoError(err)
9696
suite.initialDbNames = map[string]bool{}
9797
for _, client := range []*mongo.Client{suite.srcMongoClient, suite.dstMongoClient, suite.metaMongoClient} {
@@ -103,25 +103,28 @@ func (suite *WithMongodsTestSuite) SetupSuite() {
103103
}
104104
}
105105

106-
func (suite *WithMongodsTestSuite) startReplSet() {
106+
func (suite *WithMongodsTestSuite) startReplSets() {
107107
ctx := context.Background()
108-
clientOpts := options.
109-
Client().
110-
ApplyURI("mongodb://localhost:" + suite.srcMongoInstance.port).
111-
SetDirect(true).
112-
SetAppName("Verifier Test Suite")
113-
directClient, err := mongo.Connect(ctx, clientOpts)
114-
suite.Require().NoError(err)
115-
command := bson.M{
116-
"replSetInitiate": bson.M{
117-
"_id": replSet,
118-
"members": bson.A{
119-
bson.M{"_id": 0, "host": "localhost:" + suite.srcMongoInstance.port},
108+
109+
for _, instance := range []MongoInstance{suite.srcMongoInstance, suite.metaMongoInstance} {
110+
clientOpts := options.
111+
Client().
112+
ApplyURI("mongodb://localhost:" + instance.port).
113+
SetDirect(true).
114+
SetAppName("Verifier Test Suite")
115+
directClient, err := mongo.Connect(ctx, clientOpts)
116+
suite.Require().NoError(err)
117+
command := bson.M{
118+
"replSetInitiate": bson.M{
119+
"_id": replSet,
120+
"members": bson.A{
121+
bson.M{"_id": 0, "host": "localhost:" + instance.port},
122+
},
120123
},
121-
},
124+
}
125+
err = directClient.Database("admin").RunCommand(ctx, command).Err()
126+
suite.Require().NoError(err)
122127
}
123-
err = directClient.Database("admin").RunCommand(ctx, command).Err()
124-
suite.Require().NoError(err)
125128
}
126129

127130
func (suite *WithMongodsTestSuite) TearDownSuite() {
@@ -181,7 +184,7 @@ func startTestMongods(t *testing.T, srcMongoInstance *MongoInstance, dstMongoIns
181184
return err
182185
}
183186

184-
err = startOneMongod(t, metaMongoInstance)
187+
err = startOneMongod(t, metaMongoInstance, "--replSet", replSet)
185188
if err != nil {
186189
return err
187190
}

‎internal/verifier/util.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -166,7 +166,7 @@ func (verifier *Verifier) doInMetaTransaction(
166166

167167
session, err := verifier.metaClient.StartSession()
168168
if err != nil {
169-
verifier.logger.Fatal().Err(err).Msg("failed to start session")
169+
return errors.Wrap(err, "failed to start metadata session")
170170
}
171171

172172
defer session.EndSession(ctx)

0 commit comments

Comments
 (0)