@@ -125,46 +125,70 @@ func (verifier *Verifier) GenerateRecheckTasks(ctx context.Context) error {
125125
126126 verifier .logger .Debug ().Msgf ("Creating recheck tasks from generation %d’s %s documents" , prevGeneration , recheckQueue )
127127
128- // We generate one recheck task per collection, unless
129- // 1) The size of the list of IDs would exceed 12MB (a very conservative way of avoiding
130- // the 16MB BSON limit)
131- // 2) The size of the data would exceed our desired partition size. This limits memory use
132- // during the recheck phase.
133- prevDBName , prevCollName := "" , ""
134- var idAccum []interface {}
135- var idLenAccum int
136- var dataSizeAccum int64
137- const maxIdsSize = 12 * 1024 * 1024
138- cursor , err := verifier .verificationDatabase ().Collection (recheckQueue ).Find (
139- ctx , bson.D {{"_id.generation" , prevGeneration }}, options .Find ().SetSort (bson.D {{"_id" , 1 }}))
140- if err != nil {
141- return err
142- }
143- defer cursor .Close (ctx )
144- // We group these here using a sort rather than using aggregate because aggregate is
145- // subject to a 16MB limit on group size.
146- for cursor .Next (ctx ) {
147- err := cursor .Err ()
148- if err != nil {
149- return err
150- }
151- var doc RecheckDoc
152- err = cursor .Decode (& doc )
153- if err != nil {
154- return err
155- }
156- idRaw := cursor .Current .Lookup ("_id" , "docID" )
157- idLen := len (idRaw .Value )
158-
159- verifier .logger .Debug ().Msgf ("Found persisted recheck doc for %s.%s" , doc .PrimaryKey .DatabaseName , doc .PrimaryKey .CollectionName )
160-
161- if doc .PrimaryKey .DatabaseName != prevDBName ||
162- doc .PrimaryKey .CollectionName != prevCollName ||
163- idLenAccum >= maxIdsSize ||
164- dataSizeAccum >= verifier .partitionSizeInBytes {
165- namespace := prevDBName + "." + prevCollName
128+ return verifier .doInMetaTransaction (
129+ ctx ,
130+ func (_ context.Context , metaCtx mongo.SessionContext ) error {
131+ // We generate one recheck task per collection, unless
132+ // 1) The size of the list of IDs would exceed 12MB (a very conservative way of avoiding
133+ // the 16MB BSON limit)
134+ // 2) The size of the data would exceed our desired partition size. This limits memory use
135+ // during the recheck phase.
136+ prevDBName , prevCollName := "" , ""
137+ var idAccum []interface {}
138+ var idLenAccum int
139+ var dataSizeAccum int64
140+ const maxIdsSize = 12 * 1024 * 1024
141+ cursor , err := verifier .verificationDatabase ().Collection (recheckQueue ).Find (
142+ ctx , bson.D {{"_id.generation" , prevGeneration }}, options .Find ().SetSort (bson.D {{"_id" , 1 }}))
143+ if err != nil {
144+ return err
145+ }
146+ defer cursor .Close (ctx )
147+ // We group these here using a sort rather than using aggregate because aggregate is
148+ // subject to a 16MB limit on group size.
149+ for cursor .Next (ctx ) {
150+ err := cursor .Err ()
151+ if err != nil {
152+ return err
153+ }
154+ var doc RecheckDoc
155+ err = cursor .Decode (& doc )
156+ if err != nil {
157+ return err
158+ }
159+ idRaw := cursor .Current .Lookup ("_id" , "docID" )
160+ idLen := len (idRaw .Value )
161+
162+ verifier .logger .Debug ().Msgf ("Found persisted recheck doc for %s.%s" , doc .PrimaryKey .DatabaseName , doc .PrimaryKey .CollectionName )
163+
164+ if doc .PrimaryKey .DatabaseName != prevDBName ||
165+ doc .PrimaryKey .CollectionName != prevCollName ||
166+ idLenAccum >= maxIdsSize ||
167+ dataSizeAccum >= verifier .partitionSizeInBytes {
168+ namespace := prevDBName + "." + prevCollName
169+ if len (idAccum ) > 0 {
170+ err := verifier .InsertFailedIdsVerificationTask (metaCtx , idAccum , types .ByteCount (dataSizeAccum ), namespace )
171+ if err != nil {
172+ return err
173+ }
174+ verifier .logger .Debug ().Msgf (
175+ "Created ID verification task for namespace %s with %d ids, " +
176+ "%d id bytes and %d data bytes" ,
177+ namespace , len (idAccum ), idLenAccum , dataSizeAccum )
178+ }
179+ prevDBName = doc .PrimaryKey .DatabaseName
180+ prevCollName = doc .PrimaryKey .CollectionName
181+ idLenAccum = 0
182+ dataSizeAccum = 0
183+ idAccum = []interface {}{}
184+ }
185+ idLenAccum += idLen
186+ dataSizeAccum += int64 (doc .DataSize )
187+ idAccum = append (idAccum , doc .PrimaryKey .DocumentID )
188+ }
166189 if len (idAccum ) > 0 {
167- err := verifier .InsertFailedIdsVerificationTask (ctx , idAccum , types .ByteCount (dataSizeAccum ), namespace )
190+ namespace := prevDBName + "." + prevCollName
191+ err := verifier .InsertFailedIdsVerificationTask (metaCtx , idAccum , types .ByteCount (dataSizeAccum ), namespace )
168192 if err != nil {
169193 return err
170194 }
@@ -173,26 +197,7 @@ func (verifier *Verifier) GenerateRecheckTasks(ctx context.Context) error {
173197 "%d id bytes and %d data bytes" ,
174198 namespace , len (idAccum ), idLenAccum , dataSizeAccum )
175199 }
176- prevDBName = doc .PrimaryKey .DatabaseName
177- prevCollName = doc .PrimaryKey .CollectionName
178- idLenAccum = 0
179- dataSizeAccum = 0
180- idAccum = []interface {}{}
181- }
182- idLenAccum += idLen
183- dataSizeAccum += int64 (doc .DataSize )
184- idAccum = append (idAccum , doc .PrimaryKey .DocumentID )
185- }
186- if len (idAccum ) > 0 {
187- namespace := prevDBName + "." + prevCollName
188- err := verifier .InsertFailedIdsVerificationTask (ctx , idAccum , types .ByteCount (dataSizeAccum ), namespace )
189- if err != nil {
190- return err
191- }
192- verifier .logger .Debug ().Msgf (
193- "Created ID verification task for namespace %s with %d ids, " +
194- "%d id bytes and %d data bytes" ,
195- namespace , len (idAccum ), idLenAccum , dataSizeAccum )
196- }
197- return nil
200+ return nil
201+ },
202+ )
198203}
0 commit comments