fix: properly close iterators (#9705)

aonnikov · web-flow · commit 33b597d5b470 · 2025-08-20T17:29:39.000+07:00
Signed-off-by: Alexander Onnikov &lt;Alexander.Onnikov@xored.com&gt;
diff --git a/dev/tool/src/clean.ts b/dev/tool/src/clean.ts
@@ -179,21 +179,25 @@ export async function fixMinioBW (
   const from = new Date(new Date().setDate(new Date().getDate() - 7)).getTime()
   const list = await storageService.listStream(ctx, wsIds)
   let removed = 0
-  while (true) {
-    const objs = await list.next()
-    if (objs.length === 0) {
-      break
-    }
-    for (const obj of objs) {
-      if (obj.modifiedOn < from) continue
-      if ((obj._id as string).includes('%preview%')) {
-        await storageService.remove(ctx, wsIds, [obj._id])
-        removed++
-        if (removed % 100 === 0) {
-          console.log('removed: ', removed)
+  try {
+    while (true) {
+      const objs = await list.next()
+      if (objs.length === 0) {
+        break
+      }
+      for (const obj of objs) {
+        if (obj.modifiedOn < from) continue
+        if ((obj._id as string).includes('%preview%')) {
+          await storageService.remove(ctx, wsIds, [obj._id])
+          removed++
+          if (removed % 100 === 0) {
+            console.log('removed: ', removed)
+          }
         }
       }
     }
+  } finally {
+    await list.close()
   }
   console.log('FINISH, removed: ', removed)
 }
diff --git a/dev/tool/src/storage.ts b/dev/tool/src/storage.ts
@@ -215,7 +215,7 @@ async function processAdapter (
     }
     printStats()
   } finally {
-    await iterator.close()
+    await Promise.all([iterator.close(), targetIterator.close()])
   }
 }
 
diff --git a/models/contact/src/migration.ts b/models/contact/src/migration.ts
@@ -402,11 +402,6 @@ async function ensureGlobalPersonsForLocalAccounts (client: MigrationClient): Pr
 async function createUserProfiles (client: MigrationClient): Promise<void> {
   client.logger.log('creating user profiles for persons...', {})
 
-  const personsIterator = await client.traverse<Person>(DOMAIN_CONTACT, {
-    _class: contact.class.Person,
-    profile: { $exists: false }
-  })
-
   const lastCard = (
     await client.find<Card>(
       DOMAIN_CARD,
@@ -416,6 +411,11 @@ async function createUserProfiles (client: MigrationClient): Promise<void> {
   )[0]
   let prevRank = lastCard?.rank
 
+  const personsIterator = await client.traverse<Person>(DOMAIN_CONTACT, {
+    _class: contact.class.Person,
+    profile: { $exists: false }
+  })
+
   try {
     while (true) {
       const docs = await personsIterator.next(200)
diff --git a/models/server-activity/src/migration.ts b/models/server-activity/src/migration.ts
@@ -119,121 +119,125 @@ async function createDocUpdateMessages (client: MigrationClient): Promise<void>
 
   async function generateFor (_class: Ref<Class<Doc>>, documents: MigrationIterator<Doc>): Promise<void> {
     const classNotFound = new Set<string>()
-    while (true) {
-      const docs = await documents.next(100)
 
-      if (docs == null || docs.length === 0) {
-        break
-      }
-      const allTransactions = await getAllObjectTransactions(
-        txClient,
-        _class,
-        docs.map((it) => it._id)
-      )
+    try {
+      while (true) {
+        const docs = await documents.next(100)
 
-      // We need to find parent collection objects if missing
-      const byClass = new Map<Ref<Class<Doc>>, Set<Ref<Doc>>>()
-      for (const vv of allTransactions.values()) {
-        for (const v of vv) {
-          try {
-            const _cl = client.hierarchy.getBaseClass(v.objectClass)
-            const s = byClass.get(_cl) ?? new Set()
-            s.add(v.objectId)
-            byClass.set(_cl, s)
-          } catch {
-            const has = classNotFound.has(v.objectClass)
-            if (!has) {
-              classNotFound.add(v.objectClass)
-              console.log('class not found:', v.objectClass)
-            }
-            continue
-          }
+        if (docs == null || docs.length === 0) {
+          break
+        }
+        const allTransactions = await getAllObjectTransactions(
+          txClient,
+          _class,
+          docs.map((it) => it._id)
+        )
 
-          if (v.attachedToClass !== undefined && v.attachedTo !== undefined) {
+        // We need to find parent collection objects if missing
+        const byClass = new Map<Ref<Class<Doc>>, Set<Ref<Doc>>>()
+        for (const vv of allTransactions.values()) {
+          for (const v of vv) {
             try {
-              const _cl = client.hierarchy.getBaseClass(v.attachedToClass)
+              const _cl = client.hierarchy.getBaseClass(v.objectClass)
               const s = byClass.get(_cl) ?? new Set()
-              s.add(v.attachedTo)
+              s.add(v.objectId)
               byClass.set(_cl, s)
             } catch {
-              const objClass = v.attachedToClass
-              const has = classNotFound.has(objClass)
+              const has = classNotFound.has(v.objectClass)
               if (!has) {
-                classNotFound.add(objClass)
-                console.log('class not found:', objClass)
+                classNotFound.add(v.objectClass)
+                console.log('class not found:', v.objectClass)
+              }
+              continue
+            }
+
+            if (v.attachedToClass !== undefined && v.attachedTo !== undefined) {
+              try {
+                const _cl = client.hierarchy.getBaseClass(v.attachedToClass)
+                const s = byClass.get(_cl) ?? new Set()
+                s.add(v.attachedTo)
+                byClass.set(_cl, s)
+              } catch {
+                const objClass = v.attachedToClass
+                const has = classNotFound.has(objClass)
+                if (!has) {
+                  classNotFound.add(objClass)
+                  console.log('class not found:', objClass)
+                }
               }
             }
           }
         }
-      }
 
-      const docIds: Map<Ref<Doc>, Doc | null> = toIdMap(docs)
+        const docIds: Map<Ref<Doc>, Doc | null> = toIdMap(docs)
 
-      for (const [_class, classDocs] of byClass.entries()) {
-        const ids: Ref<Doc>[] = Array.from(classDocs.values()).filter((it) => !docIds.has(it))
-        if (ids.length > 0) {
-          for (const di of ids) {
-            docIds.set(di, null)
-          }
-          const edocs = await txClient.findAll(txClient.ctx, _class, { _id: { $in: ids } })
-          for (const ed of edocs) {
-            docIds.set(ed._id, ed)
+        for (const [_class, classDocs] of byClass.entries()) {
+          const ids: Ref<Doc>[] = Array.from(classDocs.values()).filter((it) => !docIds.has(it))
+          if (ids.length > 0) {
+            for (const di of ids) {
+              docIds.set(di, null)
+            }
+            const edocs = await txClient.findAll(txClient.ctx, _class, { _id: { $in: ids } })
+            for (const ed of edocs) {
+              docIds.set(ed._id, ed)
+            }
           }
         }
-      }
 
-      const docCache = {
-        docs: docIds,
-        transactions: allTransactions
-      }
-      const txIds = new Set<Ref<Tx>>()
-      for (const d of docs) {
-        processed += 1
-        if (processed % 1000 === 0) {
-          console.log('processed', processed)
+        const docCache = {
+          docs: docIds,
+          transactions: allTransactions
         }
-        const transactions = allTransactions.get(d._id) ?? []
-        for (const tx of transactions) {
-          txIds.add(tx._id)
+        const txIds = new Set<Ref<Tx>>()
+        for (const d of docs) {
+          processed += 1
+          if (processed % 1000 === 0) {
+            console.log('processed', processed)
+          }
+          const transactions = allTransactions.get(d._id) ?? []
+          for (const tx of transactions) {
+            txIds.add(tx._id)
+          }
         }
-      }
 
-      const ids = (
-        await client.find<DocUpdateMessage>(
-          DOMAIN_ACTIVITY,
-          { _class: activity.class.DocUpdateMessage, txId: { $in: Array.from(txIds) as Ref<TxCUD<Doc>>[] } },
-          { projection: { _id: 1, txId: 1 } }
-        )
-      ).map((p) => p.txId as Ref<Tx>)
+        const ids = (
+          await client.find<DocUpdateMessage>(
+            DOMAIN_ACTIVITY,
+            { _class: activity.class.DocUpdateMessage, txId: { $in: Array.from(txIds) as Ref<TxCUD<Doc>>[] } },
+            { projection: { _id: 1, txId: 1 } }
+          )
+        ).map((p) => p.txId as Ref<Tx>)
 
-      const existsMessages = new Set(ids)
+        const existsMessages = new Set(ids)
 
-      for (const d of docs) {
-        processed += 1
-        if (processed % 1000 === 0) {
-          console.log('processed', processed)
-        }
-        const transactions = allTransactions.get(d._id) ?? []
-        for (const tx of transactions) {
-          if (!client.hierarchy.hasClass(tx.objectClass)) {
-            const objClass = tx.objectClass
-            const has = classNotFound.has(objClass)
-            if (!has) {
-              classNotFound.add(objClass)
-              console.log('class not found:', objClass)
-            }
-            continue
+        for (const d of docs) {
+          processed += 1
+          if (processed % 1000 === 0) {
+            console.log('processed', processed)
           }
+          const transactions = allTransactions.get(d._id) ?? []
+          for (const tx of transactions) {
+            if (!client.hierarchy.hasClass(tx.objectClass)) {
+              const objClass = tx.objectClass
+              const has = classNotFound.has(objClass)
+              if (!has) {
+                classNotFound.add(objClass)
+                console.log('class not found:', objClass)
+              }
+              continue
+            }
 
-          try {
-            await generateDocUpdateMessageByTx(tx, notificationControl, client, docCache, existsMessages)
-          } catch (e: any) {
-            console.error('error processing:', d._id, e.stack)
+            try {
+              await generateDocUpdateMessageByTx(tx, notificationControl, client, docCache, existsMessages)
+            } catch (e: any) {
+              console.error('error processing:', d._id, e.stack)
+            }
           }
         }
       }
+    } finally {
+      await documents.close()
     }
-    await documents.close()
   }
 
   for (const activityClass of activityDocClasses) {
diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts
@@ -182,25 +182,28 @@ export async function removeAllObjects (
   ctx.warn('removing all objects from workspace', wsIds)
   // We need to list all files and delete them
   const iterator = await storage.listStream(ctx, wsIds)
-  let bulk: string[] = []
-  while (true) {
-    const objs = await iterator.next()
-    if (objs.length === 0) {
-      break
-    }
-    for (const obj of objs) {
-      bulk.push(obj._id)
-      if (bulk.length > 50) {
-        await storage.remove(ctx, wsIds, bulk)
-        bulk = []
+  try {
+    let bulk: string[] = []
+    while (true) {
+      const objs = await iterator.next()
+      if (objs.length === 0) {
+        break
+      }
+      for (const obj of objs) {
+        bulk.push(obj._id)
+        if (bulk.length > 50) {
+          await storage.remove(ctx, wsIds, bulk)
+          bulk = []
+        }
       }
     }
+    if (bulk.length > 0) {
+      await storage.remove(ctx, wsIds, bulk)
+      bulk = []
+    }
+  } finally {
+    await iterator.close()
   }
-  if (bulk.length > 0) {
-    await storage.remove(ctx, wsIds, bulk)
-    bulk = []
-  }
-  await iterator.close()
 }
 
 export async function objectsToArray (
@@ -210,16 +213,19 @@ export async function objectsToArray (
 ): Promise<ListBlobResult[]> {
   // We need to list all files and delete them
   const iterator = await storage.listStream(ctx, wsIds)
-  const bulk: ListBlobResult[] = []
-  while (true) {
-    const obj = await iterator.next()
-    if (obj.length === 0) {
-      break
+  try {
+    const bulk: ListBlobResult[] = []
+    while (true) {
+      const obj = await iterator.next()
+      if (obj.length === 0) {
+        break
+      }
+      bulk.push(...obj)
     }
-    bulk.push(...obj)
+    return bulk
+  } finally {
+    await iterator.close()
   }
-  await iterator.close()
-  return bulk
 }
 
 export function getDataId (wsIds: WorkspaceIds): WorkspaceDataId {
diff --git a/server/minio/src/index.ts b/server/minio/src/index.ts
@@ -176,7 +176,7 @@ export class MinioService implements StorageAdapter {
     try {
       await removeAllObjects(ctx, this, wsIds)
     } catch (err: any) {
-      ctx.error('failed t oclean all objecrs', { error: err })
+      ctx.error('failed to clean all objects', { error: err })
     }
     if (this.opt.rootBucket === undefined) {
       // Also delete a bucket
diff --git a/server/s3/src/index.ts b/server/s3/src/index.ts
@@ -227,7 +227,7 @@ export class S3Service implements StorageAdapter {
     try {
       await removeAllObjects(ctx, this, wsIds)
     } catch (err: any) {
-      ctx.error('failed t oclean all objecrs', { error: err })
+      ctx.error('failed to clean all objects', { error: err })
     }
     if (this.opt.rootBucket === undefined) {
       // We should also delete bucket

Original file line number	Diff line number	Diff line change
`@@ -215,7 +215,7 @@ async function processAdapter (`
`215`	`215`	`}`
`216`	`216`	`printStats()`
`217`	`217`	`} finally {`
`218`		`- await iterator.close()`
	`218`	`+ await Promise.all([iterator.close(), targetIterator.close()])`
`219`	`219`	`}`
`220`	`220`	`}`
`221`	`221`
Original file line number	Diff line number	Diff line change
`@@ -176,7 +176,7 @@ export class MinioService implements StorageAdapter {`
`176`	`176`	`try {`
`177`	`177`	`await removeAllObjects(ctx, this, wsIds)`
`178`	`178`	`} catch (err: any) {`
`179`		`- ctx.error('failed t oclean all objecrs', { error: err })`
	`179`	`+ ctx.error('failed to clean all objects', { error: err })`
`180`	`180`	`}`
`181`	`181`	`if (this.opt.rootBucket === undefined) {`
`182`	`182`	`// Also delete a bucket`
Original file line number	Diff line number	Diff line change
`@@ -227,7 +227,7 @@ export class S3Service implements StorageAdapter {`
`227`	`227`	`try {`
`228`	`228`	`await removeAllObjects(ctx, this, wsIds)`
`229`	`229`	`} catch (err: any) {`
`230`		`- ctx.error('failed t oclean all objecrs', { error: err })`
	`230`	`+ ctx.error('failed to clean all objects', { error: err })`
`231`	`231`	`}`
`232`	`232`	`if (this.opt.rootBucket === undefined) {`
`233`	`233`	`// We should also delete bucket`