From 0682cd502e54d1b401fef7d4e24d054dfca300fe Mon Sep 17 00:00:00 2001 From: Artyom Savchenko Date: Wed, 1 Jul 2026 16:29:35 +0700 Subject: [PATCH] Fix backup restore (#10943) * Fix backup clean blobs issue Signed-off-by: Andrey Sobolev * Restore accounts Signed-off-by: Andrey Sobolev * Allow skip queue for backup-restore Signed-off-by: Andrey Sobolev * Fix backup of wrong social ids Signed-off-by: Andrey Sobolev * Filter backup logs Signed-off-by: Andrey Sobolev * Fix doRestoreWorkspace signature after accounts-restore port The 'Restore accounts' cherry-pick updated the workspace-service restore call site to pass accountsDbUrl/accountsDbNs through to doRestoreWorkspace, but doRestoreWorkspace itself was never updated to accept them (this mismatch existed in the upstream fork too and only surfaced once types were rebuilt). Extend doRestoreWorkspace to open an AccountDB from the given URL/ns, mirroring the existing doBackup pattern, and thread it into restore() so the automatic workspace-service restore flow can also restore accounts, not just the manual dev-tool CLI restore. Signed-off-by: Artyom Savchenko --------- Signed-off-by: Andrey Sobolev Signed-off-by: Artyom Savchenko Co-authored-by: Andrey Sobolev --- dev/tool/src/db.ts | 4 +- dev/tool/src/index.ts | 16 ++- server/backup/src/backup.ts | 21 ++- server/backup/src/restore.ts | 174 +++++++++++++++++++++++- server/backup/src/service.ts | 15 +- server/workspace-service/src/service.ts | 2 + 6 files changed, 214 insertions(+), 18 deletions(-) diff --git a/dev/tool/src/db.ts b/dev/tool/src/db.ts index 201489a6f9..186d7a6c20 100644 --- a/dev/tool/src/db.ts +++ b/dev/tool/src/db.ts @@ -1755,7 +1755,7 @@ export async function restoreFromv6All ( } await sendTransactorEvent(uuid, 'force-maintenance') - await restore(ctx, pipeline, wsIds, storage, { + await restore(ctx, pipeline, wsIds, storage, undefined, { date: -1, merge: false, parallel: 1, @@ -1928,7 +1928,7 @@ export async function restoreTrustedV6Workspace ( } await sendTransactorEvent(uuid, 'force-maintenance') - await restore(ctx, pipeline, wsIds, backupWsStorage, { + await restore(ctx, pipeline, wsIds, backupWsStorage, undefined, { date: -1, merge: false, parallel: 1, diff --git a/dev/tool/src/index.ts b/dev/tool/src/index.ts index e748e621ba..2636f0262d 100644 --- a/dev/tool/src/index.ts +++ b/dev/tool/src/index.ts @@ -1136,6 +1136,8 @@ export function devTool ( .option('-i, --include ', 'A list of ; separated domain names to include during backup', '*') .option('-s, --skip ', 'A list of ; separated domain names to skip during backup', '') .option('--upgrade', 'Upgrade workspace', false) + .option('--noqueue', 'NoQueue', false) + .option('--accounts', 'Restore accounts (person/socialId) from backup', false) .option( '--history-file ', 'Store blob send info into file. Will skip already send documents.', @@ -1156,6 +1158,8 @@ export function devTool ( useStorage: string historyFile: string upgrade: boolean + noqueue: boolean + accounts: boolean } ) => { await withAccountDatabase(async (db) => { @@ -1174,10 +1178,10 @@ export function devTool ( const storage = await createFileBackupStorage(dirName) const storageConfig = storageConfigFromEnv() - const queue = getPlatformQueue('tool', ws.region) - const wsProducer = queue.getProducer(toolCtx, QueueTopic.Workspace) + const queue = !cmd.noqueue ? getPlatformQueue('tool', ws.region) : undefined + const wsProducer = queue?.getProducer(toolCtx, QueueTopic.Workspace) - await wsProducer.send(toolCtx, ws.uuid, [workspaceEvents.restoring()]) + await wsProducer?.send(toolCtx, ws.uuid, [workspaceEvents.restoring()]) const workspaceStorage: StorageAdapter = buildStorageFromConfig(storageConfig) @@ -1202,7 +1206,7 @@ export function devTool ( } await sendTransactorEvent(workspace, 'force-maintenance') - await restore(toolCtx, pipeline, wsIds, storage, { + await restore(toolCtx, pipeline, wsIds, storage, cmd.accounts ? db : undefined, { date: parseInt(date ?? '-1'), merge: cmd.merge, parallel: parseInt(cmd.parallel ?? '1'), @@ -1219,12 +1223,12 @@ export function devTool ( } console.log('workspace restored') - await wsProducer.send(toolCtx, ws.uuid, [workspaceEvents.restored()]) + await wsProducer?.send(toolCtx, ws.uuid, [workspaceEvents.restored()]) } catch (err) { toolCtx.error('failed to restore', { err }) } await pipeline?.close() - await queue.shutdown() + await queue?.shutdown() await workspaceStorage?.close() }) } diff --git a/server/backup/src/backup.ts b/server/backup/src/backup.ts index 6672bc6ac4..abf9db37cd 100644 --- a/server/backup/src/backup.ts +++ b/server/backup/src/backup.ts @@ -220,7 +220,7 @@ export async function backup ( // We need to perform compaction ctx.warn('Compacting backup') await compactBackup(ctx, storage, true, { - blobLimit: options.blobDownloadLimit, + blobLimit: options.blobDownloadLimit * 1024 * 1024, skipContentTypes: options.skipBlobContentTypes, msg: { workspaceId, url: wsIds.url } }) @@ -269,12 +269,12 @@ export async function backup ( it !== DOMAIN_MODEL_TX && it !== DOMAIN_TX && it !== DOMAIN_BLOB && - it !== ('fulltext-blob' as Domain) && - !options.skipDomains.includes(it) && - (options.include === undefined || options.include.has(it)) + it !== ('fulltext-blob' as Domain) ), ...accountDomains - ] + ].filter( + (it) => !options.skipDomains.includes(it) && (options.include === undefined || options.include.has(it)) + ) ctx.info('domains for dump', { domains: domains.length, workspace: workspaceId, url: wsIds.url }) @@ -1030,7 +1030,16 @@ export async function backup ( // 1. We need to include global records based on persons/socialIdentities info which are missing in digest // 2. We need to check updates for all records present in digest const batchSize = 1000 - const toLoad = new Set([...digest.keys(), ...affectedObjects]) as Set + const toLoad = new Set( + [...digest.keys(), ...affectedObjects].filter((it) => { + try { + BigInt(it) + return true + } catch (err: any) { + return false + } + }) + ) as Set if (toLoad.size === 0) { ctx.info('No records updates') return diff --git a/server/backup/src/restore.ts b/server/backup/src/restore.ts index d441dba9c7..8454182c3e 100644 --- a/server/backup/src/restore.ts +++ b/server/backup/src/restore.ts @@ -15,7 +15,10 @@ // import { Analytics } from '@hcengineering/analytics' +import { type Person as GlobalPerson, type SocialId, type AccountDB, Account } from '@hcengineering/account' import core, { + AccountRole, + AccountUuid, Doc, Domain, DOMAIN_BLOB, @@ -37,7 +40,7 @@ import { extract } from 'tar-stream' import { createGunzip, gunzipSync } from 'zlib' import { BackupStorage } from './storage' import type { BackupInfo } from './types' -import { doTrimHash, isAccountDomain, loadDigest, migradeBlobData } from './utils' +import { chunkArray, doTrimHash, isAccountDomain, loadDigest, migradeBlobData, toAccountDomain } from './utils' export * from './storage' const dataUploadSize = 2 * 1024 * 1024 @@ -55,6 +58,7 @@ export async function restore ( pipeline: Pipeline, wsIds: WorkspaceIds, storage: BackupStorage, + accountDb: AccountDB | undefined, opt: { date: number merge?: boolean @@ -488,7 +492,88 @@ export async function restore ( } async function processAccountDomain (c: Domain): Promise { - // TODO + if (accountDb === undefined) { + ctx.info('skipping account domain restore, no accountDb provided', { domain: c }) + return + } + + const isPersonDomain = c === toAccountDomain('person') + const changeset = await loadDigest(ctx, storage, snapshots, c, opt.date) + + if (changeset.size === 0) { + ctx.info('no account domain data to restore', { domain: c }) + return + } + + ctx.info('restoring account domain', { domain: c, total: changeset.size, workspace: workspaceId }) + + const processed = new Set() + const collectedObjects: any[] = [] + + // Collect all objects from backup snapshots + for (const s of rsnapshots) { + const d = s.domains[c] + if (d === undefined) continue + + for (const sf of d.storage ?? []) { + const readStream = await storage.load(sf) + const ex = extract() + + const endPromise = new Promise((resolve, reject) => { + ex.on('entry', (headers, stream, next) => { + const name = headers.name ?? '' + if (name.endsWith('.json')) { + const objKey = name.substring(0, name.length - 5) + if (changeset.has(objKey) && !processed.has(objKey)) { + const chunks: Buffer[] = [] + stream.on('data', (chunk) => { + chunks.push(chunk) + }) + stream.on('end', () => { + try { + const obj = JSON.parse(Buffer.concat(chunks as any).toString()) + processed.add(objKey) + collectedObjects.push(obj) + } catch (err) { + ctx.warn('failed to parse account object', { name, err }) + } + next() + }) + } else { + next() + } + } else { + next() + } + stream.resume() + }) + + ex.on('finish', () => { + resolve() + }) + + const unzip = createGunzip({ level: defaultLevel }) + readStream.on('end', () => { + readStream.destroy() + }) + readStream.pipe(unzip).on('error', (err) => { + readStream.destroy() + reject(err) + }) + unzip.pipe(ex) + }) + + await endPromise + } + } + + ctx.info('collected account objects', { domain: c, count: collectedObjects.length, workspace: workspaceId }) + + if (isPersonDomain) { + await restorePersons(ctx, accountDb, collectedObjects as GlobalPerson[], wsIds) + } else { + await restoreSocialIds(ctx, accountDb, collectedObjects as SocialId[]) + } } const limiter = new RateLimiter(opt.parallel ?? 1) @@ -538,3 +623,88 @@ export async function restore ( } return true } + +const accountBatchSize = 500 + +async function restorePersons ( + ctx: MeasureContext, + accountDb: AccountDB, + persons: GlobalPerson[], + wsIds: WorkspaceIds +): Promise { + const chunks = chunkArray(persons, accountBatchSize) + for (const chunk of chunks) { + const uuids = chunk.map((p) => p.uuid) + + // Find existing persons + const existingPersons = await accountDb.person.find({ uuid: { $in: uuids } }) + const existingPersonUuids = new Set(existingPersons.map((p) => p.uuid)) + + // Insert missing persons + const personsToInsert = chunk + .filter((p) => !existingPersonUuids.has(p.uuid)) + .map((it) => { + const { '%hash%': _, ...data } = it as any + return data + }) + if (personsToInsert.length > 0) { + await accountDb.person.insertMany(personsToInsert) + ctx.info('inserted persons', { count: personsToInsert.length }) + } + + // Find existing accounts + const accountUuids = uuids as AccountUuid[] + const existingAccounts = await accountDb.account.find({ uuid: { $in: accountUuids } }) + const existingAccountUuids = new Set(existingAccounts.map((a) => a.uuid)) + + // Insert missing accounts (without password) + const accountsToInsert: Account[] = chunk + .filter((p) => !existingAccountUuids.has(p.uuid as AccountUuid)) + .map((p) => ({ uuid: p.uuid as AccountUuid })) + if (accountsToInsert.length > 0) { + await accountDb.account.insertMany(accountsToInsert) + ctx.info('inserted accounts', { count: accountsToInsert.length }) + } + + // Assign workspace to all accounts + const workspaceRoles = new Map() + for (const uuid of accountUuids) { + const role = await accountDb.getWorkspaceRole(uuid, wsIds.uuid) + if (role !== null) { + workspaceRoles.set(uuid, role) + } + } + + const toAssign: [AccountUuid, typeof wsIds.uuid, AccountRole][] = accountUuids + .filter((uuid) => !workspaceRoles.has(uuid)) + .map((uuid) => [uuid, wsIds.uuid, AccountRole.User]) + + if (toAssign.length > 0) { + await accountDb.batchAssignWorkspace(toAssign) + ctx.info('assigned workspace to accounts', { count: toAssign.length, workspace: wsIds.uuid }) + } + } +} + +async function restoreSocialIds (ctx: MeasureContext, accountDb: AccountDB, socialIds: SocialId[]): Promise { + const chunks = chunkArray(socialIds, accountBatchSize) + for (const chunk of chunks) { + const ids = chunk.map((s) => s.key) + + // Find existing socialIds + const existingSocialIds = await accountDb.socialId.find({ key: { $in: ids } }) + const existingIds = new Set(existingSocialIds.map((s) => s.key)) + + // Insert missing socialIds + const socialIdsToInsert: SocialId[] = chunk + .filter((s) => !existingIds.has(s.key)) + .map((it) => { + const { '%hash%': _1, key: _2, ...data } = it as any + return data + }) + if (socialIdsToInsert.length > 0) { + await accountDb.socialId.insertMany(socialIdsToInsert) + ctx.info('inserted socialIds', { count: socialIdsToInsert.length }) + } + } +} diff --git a/server/backup/src/service.ts b/server/backup/src/service.ts index 76887d2d94..70f5ca659c 100644 --- a/server/backup/src/service.ts +++ b/server/backup/src/service.ts @@ -28,7 +28,7 @@ import { type WorkspaceIds, type WorkspaceInfoWithStatus } from '@hcengineering/core' -import { getAccountDB } from '@hcengineering/account' +import { getAccountDB, type AccountDB } from '@hcengineering/account' import { getAccountClient } from '@hcengineering/server-client' import { type DbConfiguration, @@ -168,6 +168,9 @@ class BackupWorker { const infoTo = setInterval(() => { const avgTime = this.allBackupTime / (this.processed + 1) + if (this.activeWorkspaces.size === 0 && this.workspacesToBackup.size === 0) { + return + } ctx.warn('********** backup info **********', { processed: this.processed, toGo: this.workspacesToBackup.size, @@ -449,6 +452,8 @@ export async function doRestoreWorkspace ( pipelineFactory: PipelineFactory, skipDomains: string[], cleanIndexState: boolean, + accountsDbUrl?: string, + accountsDbNs?: string, notify?: (progress: number) => Promise ): Promise { rootCtx.warn('\nRESTORE WORKSPACE ', { @@ -457,6 +462,7 @@ export async function doRestoreWorkspace ( }) const ctx = rootCtx.newChild('doRestore', {}, { span: false }) let pipeline: Pipeline | undefined + let closeAccountDB: (() => void) | undefined try { pipeline = await pipelineFactory( ctx, @@ -472,11 +478,15 @@ export async function doRestoreWorkspace ( } const restoreIds = { uuid: bucketName as WorkspaceUuid, dataId: bucketName as WorkspaceDataId, url: '' } const storage = await createStorageBackupStorage(ctx, backupAdapter, restoreIds, wsIds.dataId ?? wsIds.uuid) + let accountDB: AccountDB | undefined + if (accountsDbUrl !== undefined) { + ;[accountDB, closeAccountDB] = await getAccountDB(accountsDbUrl, accountsDbNs) + } const result: boolean = await ctx.with( 'restore', {}, (ctx) => - restore(ctx, pipeline as Pipeline, wsIds, storage, { + restore(ctx, pipeline as Pipeline, wsIds, storage, accountDB, { date: -1, skip: new Set(skipDomains), recheck: false, // Do not need to recheck @@ -492,6 +502,7 @@ export async function doRestoreWorkspace ( rootCtx.error('\n\nFAILED to RESTORE', { workspace: wsIds.uuid, err }) return false } finally { + closeAccountDB?.() if (pipeline !== undefined) { await pipeline.close() } diff --git a/server/workspace-service/src/service.ts b/server/workspace-service/src/service.ts index 236ea991f9..bd0f7d1b86 100644 --- a/server/workspace-service/src/service.ts +++ b/server/workspace-service/src/service.ts @@ -705,6 +705,8 @@ export class WorkspaceWorker { pipelineFactory, [DOMAIN_BLOB], true, + this.accountsDbUrl, + undefined, (_p: number) => { if (progress !== Math.round(_p)) { progress = Math.round(_p)