wieerwill/platform
1
0
Fork 0
mirror of https://github.com/hcengineering/platform.git synced 2025-04-14 12:25:17 +00:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

UBERF-4287: Fix Indexer peak memory usage (#3993)

Browse Source 
Signed-off-by: Andrey Sobolev <haiodo@gmail.com>
This commit is contained in:
Andrey Sobolev 2023-11-15 11:48:49 +07:00 committed by GitHub
parent 64ae21db07
commit 5d8c1a6b76
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 75 additions and 73 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

148
server/core/src/indexer/indexer.ts
View File

@ -559,91 +559,93 @@ export class FullTextIndexPipeline implements FullTextPipeline {
} }
async checkIndexConsistency (dbStorage: ServerStorage): Promise<void> { async checkIndexConsistency (dbStorage: ServerStorage): Promise<void> {
if (process.env.MODEL_VERSION !== undefined) { await rateLimitter.exec(async () => {
const modelVersion = (await this.model.findAll(core.class.Version, {}))[0] if (process.env.MODEL_VERSION !== undefined) {
if (modelVersion !== undefined) { const modelVersion = (await this.model.findAll(core.class.Version, {}))[0]
const modelVersionString = versionToString(modelVersion) if (modelVersion !== undefined) {
if (modelVersionString !== process.env.MODEL_VERSION) { const modelVersionString = versionToString(modelVersion)
console.error( if (modelVersionString !== process.env.MODEL_VERSION) {
`Indexer: Model version mismatch model: ${modelVersionString} env: ${process.env.MODEL_VERSION}` console.error(
) `Indexer: Model version mismatch model: ${modelVersionString} env: ${process.env.MODEL_VERSION}`
return )
return
}
} }
} }
}
this.hierarchy.domains() this.hierarchy.domains()
const allClasses = this.hierarchy.getDescendants(core.class.Doc) const allClasses = this.hierarchy.getDescendants(core.class.Doc)
for (const c of allClasses) { for (const c of allClasses) {
if (this.cancelling) {
return
}
if (!isClassIndexable(this.hierarchy, c)) {
// No need, since no indexable fields or attachments.
continue
}
console.log(this.workspace.name, 'checking index', c)
// All saved state documents
const states = (
await this.storage.findAll(core.class.DocIndexState, { objectClass: c }, { projection: { _id: 1 } })
).map((it) => it._id)
while (true) {
if (this.cancelling) { if (this.cancelling) {
return return
} }
const newDocs: DocIndexState[] = (
await dbStorage.findAll<Doc>( if (!isClassIndexable(this.hierarchy, c)) {
this.metrics, // No need, since no indexable fields or attachments.
c, continue
{ _class: c, _id: { $nin: states } }, }
{ limit: 1000, projection: { _id: 1, attachedTo: 1, attachedToClass: 1 } as any }
) console.log(this.workspace.name, 'checking index', c)
).map((it) => {
return createStateDoc(it._id, c, { // All saved state documents
stages: {}, const states = (
attributes: {}, await this.storage.findAll(core.class.DocIndexState, { objectClass: c }, { projection: { _id: 1 } })
removed: false, ).map((it) => it._id)
space: it.space,
attachedTo: (it as AttachedDoc)?.attachedTo ?? undefined, while (true) {
attachedToClass: (it as AttachedDoc)?.attachedToClass ?? undefined if (this.cancelling) {
return
}
const newDocs: DocIndexState[] = (
await dbStorage.findAll<Doc>(
this.metrics,
c,
{ _class: c, _id: { $nin: states } },
{ limit: 1000, projection: { _id: 1, attachedTo: 1, attachedToClass: 1 } as any }
)
).map((it) => {
return createStateDoc(it._id, c, {
stages: {},
attributes: {},
removed: false,
space: it.space,
attachedTo: (it as AttachedDoc)?.attachedTo ?? undefined,
attachedToClass: (it as AttachedDoc)?.attachedToClass ?? undefined
})
}) })
})
states.push(...newDocs.map((it) => it._id)) states.push(...newDocs.map((it) => it._id))
if (newDocs.length === 0) { if (newDocs.length === 0) {
// All updated for this class // All updated for this class
break break
} }
try { try {
await this.storage.upload(DOMAIN_DOC_INDEX_STATE, newDocs) await this.storage.upload(DOMAIN_DOC_INDEX_STATE, newDocs)
} catch (err: any) { } catch (err: any) {
console.error(err) console.error(err)
}
} }
const statesSet = new Set(states)
const docIds = (await dbStorage.findAll<Doc>(this.metrics, c, { _class: c }, { projection: { _id: 1 } }))
.filter((it) => !statesSet.has(it._id as Ref<DocIndexState>))
.map((it) => it._id)
await this.storage.clean(DOMAIN_DOC_INDEX_STATE, docIds)
} }
const statesSet = new Set(states)
const docIds = (await dbStorage.findAll<Doc>(this.metrics, c, { _class: c }, { projection: { _id: 1 } }))
.filter((it) => !statesSet.has(it._id as Ref<DocIndexState>))
.map((it) => it._id)
await this.storage.clean(DOMAIN_DOC_INDEX_STATE, docIds)
}
// Clean for non existing classes // Clean for non existing classes
const unknownClasses = ( const unknownClasses = (
await this.storage.findAll( await this.storage.findAll(
core.class.DocIndexState, core.class.DocIndexState,
{ objectClass: { $nin: allClasses } }, { objectClass: { $nin: allClasses } },
{ projection: { _id: 1 } } { projection: { _id: 1 } }
) )
).map((it) => it._id) ).map((it) => it._id)
if (unknownClasses.length > 0) { if (unknownClasses.length > 0) {
await this.storage.clean(DOMAIN_DOC_INDEX_STATE, unknownClasses) await this.storage.clean(DOMAIN_DOC_INDEX_STATE, unknownClasses)
} }
})
} }
} }