From 04a282c0f8d4c239d655ccde308c50fe3b7d4e14 Mon Sep 17 00:00:00 2001 From: Andrey Sobolev Date: Tue, 4 Feb 2025 22:29:27 +0700 Subject: [PATCH] Merge staging feb4 (#7910) --- .vscode/launch.json | 8 +- common/scripts/version.txt | 2 +- packages/core/src/__tests__/client.test.ts | 6 +- packages/core/src/__tests__/connection.ts | 6 +- packages/core/src/backup.ts | 2 + packages/core/src/client.ts | 4 + packages/core/src/server.ts | 2 + packages/query/src/__tests__/connection.ts | 5 ++ plugins/client-resources/src/connection.ts | 4 + server/backup/src/backup.ts | 97 +++++++++++++--------- server/backup/src/service.ts | 50 ++++++----- server/core/src/benchmark/index.ts | 6 ++ server/core/src/mem.ts | 6 +- server/core/src/storage.ts | 4 + server/core/src/types.ts | 2 + server/core/src/utils.ts | 35 +++++++- server/middleware/src/lowLevel.ts | 3 + server/mongo/src/storage.ts | 8 +- server/postgres/src/storage.ts | 8 +- server/server-pipeline/src/blobStorage.ts | 8 +- server/server/src/client.ts | 11 +++ 21 files changed, 205 insertions(+), 72 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index cdacb92688..cefaa28f64 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -295,7 +295,7 @@ "name": "Debug backup tool", "type": "node", "request": "launch", - "args": ["src/__start.ts", "backup-restore", "../../../hardware/dump/alex-staff-agency", "w-haiodo-alex-staff-c-673ee7ab-87df5406ea-2b8b4d", "--skip", "blob"], + "args": ["src/__start.ts", "backup", "../../../hardware/dump/githubcr", "w-haiodo-githubcr-67403799-de2a46aa46-beb3b2"], "env": { "MINIO_ACCESS_KEY": "minioadmin", "MINIO_SECRET_KEY": "minioadmin", @@ -305,12 +305,14 @@ "ACCOUNTS_URL": "http://localhost:3000", "TELEGRAM_DATABASE": "telegram-service" }, + "smartStep": true, + "sourceMapRenames": true, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "sourceMaps": true, "cwd": "${workspaceRoot}/dev/tool", "protocol": "inspector", "outputCapture": "std", - "runtimeVersion": "20", + "runtimeVersion": "22", "showAsyncStacks": true }, { @@ -401,7 +403,7 @@ "SECRET": "secret", "REGION": "cockroach", "BUCKET_NAME":"backups", - "INTERVAL":"43200" + "INTERVAL":"0" }, "runtimeArgs": ["--nolazy", "-r", "ts-node/register"], "showAsyncStacks": true, diff --git a/common/scripts/version.txt b/common/scripts/version.txt index 7fd9f34bf6..672c5d09b7 100644 --- a/common/scripts/version.txt +++ b/common/scripts/version.txt @@ -1 +1 @@ -"0.6.421" \ No newline at end of file +"0.6.421" diff --git a/packages/core/src/__tests__/client.test.ts b/packages/core/src/__tests__/client.test.ts index a906d4aaef..13f2ac36f9 100644 --- a/packages/core/src/__tests__/client.test.ts +++ b/packages/core/src/__tests__/client.test.ts @@ -24,7 +24,7 @@ import { ModelDb, TxDb } from '../memdb' import { TxOperations } from '../operations' import type { DocumentQuery, FindResult, SearchOptions, SearchQuery, SearchResult, TxResult } from '../storage' import { Tx, TxFactory, TxProcessor } from '../tx' -import { fillConfiguration, pluginFilterTx } from '../utils' +import { fillConfiguration, generateId, pluginFilterTx } from '../utils' import { connect } from './connection' import { genMinModel } from './minmodel' @@ -142,6 +142,10 @@ describe('client', () => { finished: true }) + async getDomainHash (domain: Domain): Promise { + return generateId() + } + async closeChunk (idx: number): Promise {} async loadDocs (domain: Domain, docs: Ref[]): Promise { return [] diff --git a/packages/core/src/__tests__/connection.ts b/packages/core/src/__tests__/connection.ts index ceb91c17ec..a9a831b81f 100644 --- a/packages/core/src/__tests__/connection.ts +++ b/packages/core/src/__tests__/connection.ts @@ -13,7 +13,7 @@ // limitations under the License. // -import { ClientConnectEvent, DocChunk } from '..' +import { ClientConnectEvent, DocChunk, generateId } from '..' import type { Class, Doc, Domain, Ref, Timestamp } from '../classes' import { ClientConnection } from '../client' import core from '../component' @@ -82,6 +82,10 @@ export async function connect (handler: (tx: Tx) => void): Promise { + return generateId() + } + async closeChunk (idx: number): Promise {} async loadDocs (domain: Domain, docs: Ref[]): Promise { return [] diff --git a/packages/core/src/backup.ts b/packages/core/src/backup.ts index b7be7cb074..32c7702219 100644 --- a/packages/core/src/backup.ts +++ b/packages/core/src/backup.ts @@ -26,5 +26,7 @@ export interface BackupClient { upload: (domain: Domain, docs: Doc[]) => Promise clean: (domain: Domain, docs: Ref[]) => Promise + getDomainHash: (domain: Domain) => Promise + sendForceClose: () => Promise } diff --git a/packages/core/src/client.ts b/packages/core/src/client.ts index c3d74131a2..70abad3f7d 100644 --- a/packages/core/src/client.ts +++ b/packages/core/src/client.ts @@ -174,6 +174,10 @@ class ClientImpl implements Client, BackupClient { return await this.conn.loadChunk(domain, idx) } + async getDomainHash (domain: Domain): Promise { + return await this.conn.getDomainHash(domain) + } + async closeChunk (idx: number): Promise { await this.conn.closeChunk(idx) } diff --git a/packages/core/src/server.ts b/packages/core/src/server.ts index 52f9df1955..2e9e72b2dc 100644 --- a/packages/core/src/server.ts +++ b/packages/core/src/server.ts @@ -95,6 +95,8 @@ export interface LowLevelStorage { query: DocumentQuery, options?: Pick, 'sort' | 'limit' | 'projection'> ) => Promise> + + getDomainHash: (ctx: MeasureContext, domain: Domain) => Promise } export interface Iterator { diff --git a/packages/query/src/__tests__/connection.ts b/packages/query/src/__tests__/connection.ts index b8b490c7a4..f6c154aebb 100644 --- a/packages/query/src/__tests__/connection.ts +++ b/packages/query/src/__tests__/connection.ts @@ -27,6 +27,7 @@ import core, { FindOptions, FindResult, FulltextStorage, + generateId, Hierarchy, LoadModelResponse, ModelDb, @@ -121,6 +122,10 @@ FulltextStorage & { } } + async getDomainHash (domain: Domain): Promise { + return generateId() + } + async loadModel (lastTxTime: Timestamp): Promise { return txes } diff --git a/plugins/client-resources/src/connection.ts b/plugins/client-resources/src/connection.ts index f5eb96954b..7e48b547b6 100644 --- a/plugins/client-resources/src/connection.ts +++ b/plugins/client-resources/src/connection.ts @@ -804,6 +804,10 @@ class Connection implements ClientConnection { return this.sendRequest({ method: 'loadChunk', params: [domain, idx] }) } + async getDomainHash (domain: Domain): Promise { + return await this.sendRequest({ method: 'getDomainHash', params: [domain] }) + } + closeChunk (idx: number): Promise { return this.sendRequest({ method: 'closeChunk', params: [idx] }) } diff --git a/server/backup/src/backup.ts b/server/backup/src/backup.ts index 5da6f90b23..2fd722e7e9 100644 --- a/server/backup/src/backup.ts +++ b/server/backup/src/backup.ts @@ -134,6 +134,9 @@ export interface BackupInfo { snapshots: BackupSnapshot[] snapshotsIndex?: number lastTxId?: string + + // A hash of current domain transactions, so we could skip all other checks if same. + domainHashes: Record } async function loadDigest ( @@ -143,15 +146,13 @@ async function loadDigest ( domain: Domain, date?: number ): Promise, string>> { - ctx = ctx.newChild('load digest', { domain, count: snapshots.length }) - ctx.info('loading-digest', { domain, snapshots: snapshots.length }) const result = new Map, string>() for (const s of snapshots) { const d = s.domains[domain] // Load old JSON snapshot if (d?.snapshot !== undefined) { - const dChanges: SnapshotV6 = JSON.parse(gunzipSync((await storage.loadFile(d.snapshot)) as any).toString()) + const dChanges: SnapshotV6 = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(d.snapshot))).toString()) for (const [k, v] of Object.entries(dChanges.added)) { result.set(k as Ref, v) } @@ -164,7 +165,7 @@ async function loadDigest ( } for (const snapshot of d?.snapshots ?? []) { try { - const dataBlob = gunzipSync((await storage.loadFile(snapshot)) as any) + const dataBlob = gunzipSync(new Uint8Array(await storage.loadFile(snapshot))) .toString() .split('\n') const addedCount = parseInt(dataBlob.shift() ?? '0') @@ -195,7 +196,6 @@ async function loadDigest ( break } } - ctx.end() ctx.info('load-digest', { domain, snapshots: snapshots.length, documents: result.size }) return result } @@ -326,7 +326,7 @@ async function verifyDigest ( } let lmodified = false try { - const dataBlob = gunzipSync(await storage.loadFile(snapshot)) + const dataBlob = gunzipSync(new Uint8Array(await storage.loadFile(snapshot))) .toString() .split('\n') const addedCount = parseInt(dataBlob.shift() ?? '0') @@ -389,7 +389,7 @@ async function write (chunk: any, stream: Writable): Promise { }) }) if (needDrain) { - await new Promise((resolve, reject) => stream.once('drain', resolve)) + await new Promise((resolve) => stream.once('drain', resolve)) } } @@ -738,7 +738,8 @@ export async function backup ( let backupInfo: BackupInfo = { workspace: workspaceId, version: '0.6.2', - snapshots: [] + snapshots: [], + domainHashes: {} } // Version 0.6.2, format of digest file is changed to @@ -746,12 +747,17 @@ export async function backup ( const infoFile = 'backup.json.gz' if (await storage.exists(infoFile)) { - backupInfo = JSON.parse(gunzipSync((await storage.loadFile(infoFile)) as any).toString()) + backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) } backupInfo.version = '0.6.2' backupInfo.workspace = workspaceId + if (backupInfo.domainHashes === undefined) { + // Migration + backupInfo.domainHashes = {} + } + let lastTx: Tx | undefined let lastTxChecked = false @@ -872,13 +878,6 @@ export async function backup ( let st = Date.now() let changed: number = 0 const needRetrieveChunks: Ref[][] = [] - // Load all digest from collection. - ctx.info('processed', { - processed, - digest: digest.size, - time: Date.now() - st, - workspace: workspaceId - }) const oldHash = new Map, string>() function removeFromNeedRetrieve (needRetrieve: Ref[], id: string): void { @@ -961,7 +960,7 @@ export async function backup ( needRetrieve = [] } if (currentChunk.finished) { - ctx.info('processed-end', { + ctx.info('processed', { processed, digest: digest.size, time: Date.now() - st, @@ -1015,6 +1014,11 @@ export async function backup ( removed: 0 } + const dHash = await connection.getDomainHash(domain) + if (backupInfo.domainHashes[domain] === dHash) { + ctx.info('no changes in domain', { domain }) + return + } // Cumulative digest const digest = await ctx.with('load-digest', {}, (ctx) => loadDigest(ctx, storage, backupInfo.snapshots, domain)) @@ -1034,11 +1038,6 @@ export async function backup ( if (progress !== undefined) { await progress(10) } - - if (needRetrieveChunks.length > 0) { - ctx.info('dumping domain...', { workspace: workspaceId, domain }) - } - const totalChunks = needRetrieveChunks.flatMap((it) => it.length).reduce((p, c) => p + c, 0) let processed = 0 let blobs = 0 @@ -1058,7 +1057,7 @@ export async function backup ( if (needRetrieve.length === 0) { continue } - ctx.info('Retrieve chunk', { + ctx.info('<<<< chunk', { needRetrieve: needRetrieveChunks.reduce((v, docs) => v + docs.length, 0), toLoad: needRetrieve.length, workspace: workspaceId, @@ -1066,7 +1065,7 @@ export async function backup ( }) let docs: Doc[] = [] try { - docs = await ctx.with('load-docs', {}, async (ctx) => await connection.loadDocs(domain, needRetrieve)) + docs = await ctx.with('<<<< load-docs', {}, async () => await connection.loadDocs(domain, needRetrieve)) lastSize = docs.reduce((p, it) => p + estimateDocSize(it), 0) if (docs.length !== needRetrieve.length) { const nr = new Set(docs.map((it) => it._id)) @@ -1115,7 +1114,6 @@ export async function backup ( _pack = pack() stIndex++ const storageFile = join(backupIndex, `${domain}-data-${snapshot.date}-${stIndex}.tar.gz`) - ctx.info('storing from domain', { domain, storageFile, workspace: workspaceId }) domainInfo.storage = [...(domainInfo.storage ?? []), storageFile] const tmpFile = join(tmpRoot, basename(storageFile) + '.tmp') const tempFile = createWriteStream(tmpFile) @@ -1146,7 +1144,7 @@ export async function backup ( }) // We need to upload file to storage - ctx.info('Upload pack file', { storageFile, size: sz, workspace: workspaceId }) + ctx.info('>>>> upload pack', { storageFile, size: sz, workspace: wsIds.url }) await storage.writeFile(storageFile, createReadStream(tmpFile)) await rm(tmpFile) @@ -1283,7 +1281,10 @@ export async function backup ( changed++ } - if (changed > 0) { + if (changed > 0 || backupInfo.domainHashes[domain] !== dHash) { + // Store domain hash, to be used on next time. + backupInfo.domainHashes[domain] = dHash + snapshot.domains[domain] = domainInfo domainInfo.added += processedChanges.added.size domainInfo.updated += processedChanges.updated.size @@ -1312,10 +1313,14 @@ export async function backup ( try { global.gc?.() } catch (err) {} - ctx.info('memory-stats', { + + const mm = { old: Math.round(oldUsed / (1024 * 1024)), current: Math.round(process.memoryUsage().heapUsed / (1024 * 1024)) - }) + } + if (mm.old > mm.current + mm.current / 10) { + ctx.info('memory-stats', mm) + } await ctx.with('process-domain', { domain }, async (ctx) => { await processDomain( ctx, @@ -1338,7 +1343,7 @@ export async function backup ( let sizeInfo: Record = {} if (await storage.exists(sizeFile)) { - sizeInfo = JSON.parse(gunzipSync((await storage.loadFile(sizeFile)) as any).toString()) + sizeInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(sizeFile))).toString()) } let processed = 0 @@ -1436,7 +1441,7 @@ export async function backupList (storage: BackupStorage): Promise { if (!(await storage.exists(infoFile))) { throw new Error(`${infoFile} should present to restore`) } - const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) for (const s of backupInfo.snapshots) { console.log('snapshot: id:', s.date, ' date:', new Date(s.date)) @@ -1452,7 +1457,7 @@ export async function backupRemoveLast (storage: BackupStorage, date: number): P if (!(await storage.exists(infoFile))) { throw new Error(`${infoFile} should present to restore`) } - const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) const old = backupInfo.snapshots.length backupInfo.snapshots = backupInfo.snapshots.filter((it) => it.date < date) @@ -1474,7 +1479,7 @@ export async function backupSize (storage: BackupStorage): Promise { } let size = 0 - const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) const addFileSize = async (file: string | undefined | null): Promise => { if (file != null && (await storage.exists(file))) { @@ -1513,12 +1518,12 @@ export async function backupDownload (storage: BackupStorage, storeIn: string): } let size = 0 - const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) let sizeInfo: Record = {} if (await storage.exists(sizeFile)) { - sizeInfo = JSON.parse(gunzipSync(await storage.loadFile(sizeFile)).toString()) + sizeInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(sizeFile))).toString()) } console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) @@ -1579,7 +1584,7 @@ export async function backupFind (storage: BackupStorage, id: Ref, domain?: if (!(await storage.exists(infoFile))) { throw new Error(`${infoFile} should present to restore`) } - const backupInfo: BackupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) console.log('workspace:', backupInfo.workspace ?? '', backupInfo.version) const toolCtx = new MeasureMetricsContext('', {}) @@ -1676,7 +1681,7 @@ export async function restore ( ctx.error('file not pressent', { file: infoFile }) throw new Error(`${infoFile} should present to restore`) } - const backupInfo: BackupInfo = JSON.parse(gunzipSync((await storage.loadFile(infoFile)) as any).toString()) + const backupInfo: BackupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) let snapshots = backupInfo.snapshots if (opt.date !== -1) { const bk = backupInfo.snapshots.findIndex((it) => it.date === opt.date) @@ -1688,6 +1693,10 @@ export async function restore ( } else { opt.date = snapshots[snapshots.length - 1].date } + + if (backupInfo.domainHashes === undefined) { + backupInfo.domainHashes = {} + } ctx.info('restore to ', { id: opt.date, date: new Date(opt.date).toDateString() }) const rsnapshots = Array.from(snapshots).reverse() @@ -1764,6 +1773,11 @@ export async function restore ( } async function processDomain (c: Domain): Promise { + const dHash = await connection.getDomainHash(c) + if (backupInfo.domainHashes[c] === dHash) { + ctx.info('no changes in domain', { domain: c }) + return + } const changeset = await loadDigest(ctx, storage, snapshots, c, opt.date) // We need to load full changeset from server const serverChangeset = new Map, string>() @@ -1772,7 +1786,10 @@ export async function restore ( try { global.gc?.() } catch (err) {} - ctx.info('memory-stats', { old: oldUsed / (1024 * 1024), current: process.memoryUsage().heapUsed / (1024 * 1024) }) + const mm = { old: oldUsed / (1024 * 1024), current: process.memoryUsage().heapUsed / (1024 * 1024) } + if (mm.old > mm.current + mm.current / 10) { + ctx.info('memory-stats', mm) + } let idx: number | undefined let loaded = 0 @@ -2177,7 +2194,7 @@ export async function compactBackup ( const infoFile = 'backup.json.gz' if (await storage.exists(infoFile)) { - backupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) } else { console.log('No backup found') return @@ -2514,7 +2531,7 @@ export async function checkBackupIntegrity (ctx: MeasureContext, storage: Backup const infoFile = 'backup.json.gz' if (await storage.exists(infoFile)) { - backupInfo = JSON.parse(gunzipSync(await storage.loadFile(infoFile)).toString()) + backupInfo = JSON.parse(gunzipSync(new Uint8Array(await storage.loadFile(infoFile))).toString()) } else { console.log('No backup found') return diff --git a/server/backup/src/service.ts b/server/backup/src/service.ts index 50466643ea..7a1a32535c 100644 --- a/server/backup/src/service.ts +++ b/server/backup/src/service.ts @@ -140,7 +140,7 @@ class BackupWorker { } const lastBackup = it.backupInfo?.lastBackup ?? 0 - if ((now - lastBackup) / 1000 < this.config.Interval) { + if ((now - lastBackup) / 1000 < this.config.Interval && this.config.Interval !== 0) { // No backup required, interval not elapsed skipped++ return false @@ -187,7 +187,7 @@ class BackupWorker { ctx.warn('Preparing for BACKUP', { total: workspaces.length, skipped, - workspaces: workspaces.map((it) => it.uuid) + workspaces: workspaces.map((it) => it.url) }) const part = workspaces.slice(0, 500) @@ -222,26 +222,34 @@ class BackupWorker { ETA: Math.round((workspaces.length - processed) * avgTime) }) }, 10000) - for (const ws of workspaces) { - await rateLimiter.add(async () => { - index++ - if (this.canceled || Date.now() - startTime > recheckTimeout) { - return // If canceled, we should stop - } - const st = Date.now() - const result = await this.doBackup(ctx, ws) - const totalTime = Date.now() - st - times.push(totalTime) - if (!result) { - failedWorkspaces.push(ws) - return - } - processed++ - }) - } - await rateLimiter.waitProcessing() - clearInterval(infoTo) + try { + for (const ws of workspaces) { + await rateLimiter.add(async () => { + try { + index++ + if (this.canceled || Date.now() - startTime > recheckTimeout) { + return // If canceled, we should stop + } + const st = Date.now() + const result = await this.doBackup(ctx, ws) + const totalTime = Date.now() - st + times.push(totalTime) + if (!result) { + failedWorkspaces.push(ws) + return + } + processed++ + } catch (err: any) { + ctx.error('Backup failed', { err }) + } + }) + } + + await rateLimiter.waitProcessing() + } finally { + clearInterval(infoTo) + } return { failedWorkspaces, processed, skipped: workspaces.length - processed } } diff --git a/server/core/src/benchmark/index.ts b/server/core/src/benchmark/index.ts index a1cf94cbcf..44f95a68aa 100644 --- a/server/core/src/benchmark/index.ts +++ b/server/core/src/benchmark/index.ts @@ -22,6 +22,7 @@ import core, { type Class, type Doc, type DocumentQuery, + type Domain, type FindOptions, type FindResult, type Hierarchy, @@ -90,6 +91,11 @@ class BenchmarkDbAdapter extends DummyDbAdapter { return toFindResult(result as T[]) } + getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + // Since benchmark coult not be changed. + return Promise.resolve('') + } + tx (ctx: MeasureContext, ...tx: Tx[]): Promise { if (benchData === '') { benchData = genData(1024 * 1024) diff --git a/server/core/src/mem.ts b/server/core/src/mem.ts index a1a544c26e..92cdaa3686 100644 --- a/server/core/src/mem.ts +++ b/server/core/src/mem.ts @@ -37,7 +37,6 @@ import core, { type WorkspaceIds } from '@hcengineering/core' import { type DbAdapter, type DbAdapterHandler, type DomainHelperOperations } from './adapter' - /** * @public */ @@ -102,6 +101,11 @@ export class DummyDbAdapter implements DbAdapter { async clean (ctx: MeasureContext, domain: Domain, docs: Ref[]): Promise {} + getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + // Return '' for empty documents content. + return Promise.resolve('') + } + async update( ctx: MeasureContext, domain: Domain, diff --git a/server/core/src/storage.ts b/server/core/src/storage.ts index 4dc8834f98..7f6abdb014 100644 --- a/server/core/src/storage.ts +++ b/server/core/src/storage.ts @@ -88,6 +88,10 @@ export class BackupClientOps { }) } + getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + return this.storage.getDomainHash(ctx, domain) + } + closeChunk (ctx: MeasureContext, idx: number): Promise { return ctx.with('close-chunk', {}, async () => { const chunk = this.chunkInfo.get(idx) diff --git a/server/core/src/types.ts b/server/core/src/types.ts index 14ad1179cd..50d239e363 100644 --- a/server/core/src/types.ts +++ b/server/core/src/types.ts @@ -572,6 +572,8 @@ export interface Session { searchFulltext: (ctx: ClientSessionCtx, query: SearchQuery, options: SearchOptions) => Promise tx: (ctx: ClientSessionCtx, tx: Tx) => Promise loadChunk: (ctx: ClientSessionCtx, domain: Domain, idx?: number) => Promise + + getDomainHash: (ctx: ClientSessionCtx, domain: Domain) => Promise closeChunk: (ctx: ClientSessionCtx, idx: number) => Promise loadDocs: (ctx: ClientSessionCtx, domain: Domain, docs: Ref[]) => Promise upload: (ctx: ClientSessionCtx, domain: Domain, docs: Doc[]) => Promise diff --git a/server/core/src/utils.ts b/server/core/src/utils.ts index 4b7a4a98a4..346494b2a9 100644 --- a/server/core/src/utils.ts +++ b/server/core/src/utils.ts @@ -31,7 +31,7 @@ import core, { type PersonUuid } from '@hcengineering/core' import { PlatformError, unknownError } from '@hcengineering/platform' -import { type Hash } from 'crypto' +import { createHash, type Hash } from 'crypto' import fs from 'fs' import type { DbAdapter } from './adapter' import { BackupClientOps } from './storage' @@ -262,6 +262,7 @@ export function wrapPipeline (ctx: MeasureContext, pipeline: Pipeline, wsIds: Wo getHierarchy: () => pipeline.context.hierarchy, getModel: () => pipeline.context.modelDb, loadChunk: (domain, idx) => backupOps.loadChunk(ctx, domain, idx), + getDomainHash: (domain) => backupOps.getDomainHash(ctx, domain), loadDocs: (domain, docs) => backupOps.loadDocs(ctx, domain, docs), upload: (domain, docs) => backupOps.upload(ctx, domain, docs), searchFulltext: async (query, options) => ({ docs: [], total: 0 }), @@ -310,6 +311,10 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap throw new Error('unsupported') } + async getDomainHash (domain: Domain): Promise { + return await storageAdapter.getDomainHash(ctx, domain) + } + async closeChunk (idx: number): Promise {} async loadDocs (domain: Domain, docs: Ref[]): Promise { @@ -328,3 +333,31 @@ export function wrapAdapterToClient (ctx: MeasureContext, storageAdapter: DbAdap } return new TestClientConnection() } + +export async function calcHashHash (ctx: MeasureContext, domain: Domain, adapter: DbAdapter): Promise { + const hash = createHash('sha256') + + const it = adapter.find(ctx, domain) + + try { + let count = 0 + while (true) { + const part = await it.next(ctx) + if (part.length === 0) { + break + } + count += part.length + for (const doc of part) { + hash.update(doc.id) + hash.update(doc.hash) + } + } + if (count === 0) { + // Use empty hash for empty documents. + return '' + } + return hash.digest('hex') + } finally { + await it.close(ctx) + } +} diff --git a/server/middleware/src/lowLevel.ts b/server/middleware/src/lowLevel.ts index fa8b86ec10..9aa6e6b3b6 100644 --- a/server/middleware/src/lowLevel.ts +++ b/server/middleware/src/lowLevel.ts @@ -75,6 +75,9 @@ export class LowLevelMiddleware extends BaseMiddleware implements Middleware { rawDeleteMany (domain, query) { return adapterManager.getAdapter(domain, true).rawDeleteMany(domain, query) }, + getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + return adapterManager.getAdapter(domain, false).getDomainHash(ctx, domain) + }, traverse( domain: Domain, query: DocumentQuery, diff --git a/server/mongo/src/storage.ts b/server/mongo/src/storage.ts index 6c989ab37a..14860a1ded 100644 --- a/server/mongo/src/storage.ts +++ b/server/mongo/src/storage.ts @@ -69,7 +69,8 @@ import { type DomainHelperOperations, type ServerFindOptions, type StorageAdapter, - type TxAdapter + type TxAdapter, + calcHashHash } from '@hcengineering/server-core' import { type AbstractCursor, @@ -1084,6 +1085,11 @@ abstract class MongoAdapterBase implements DbAdapter { return Date.now().toString(16) // Current hash value } + @withContext('get-domain-hash') + async getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + return await calcHashHash(ctx, domain, this) + } + strimSize (str?: string): string { if (str == null) { return '' diff --git a/server/postgres/src/storage.ts b/server/postgres/src/storage.ts index 602179200b..6450accfc1 100644 --- a/server/postgres/src/storage.ts +++ b/server/postgres/src/storage.ts @@ -65,7 +65,8 @@ import { type DbAdapterHandler, type DomainHelperOperations, type ServerFindOptions, - type TxAdapter + type TxAdapter, + calcHashHash } from '@hcengineering/server-core' import type postgres from 'postgres' import { createDBClient, createGreenDBClient, type DBClient } from './client' @@ -1439,6 +1440,11 @@ abstract class PostgresAdapterBase implements DbAdapter { return res } + @withContext('get-domain-hash') + async getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + return await calcHashHash(ctx, domain, this) + } + curHash (): string { return Date.now().toString(16) // Current hash value } diff --git a/server/server-pipeline/src/blobStorage.ts b/server/server-pipeline/src/blobStorage.ts index 2efdcf6281..d969708551 100644 --- a/server/server-pipeline/src/blobStorage.ts +++ b/server/server-pipeline/src/blobStorage.ts @@ -33,7 +33,8 @@ import { type TxResult, type Blob, type WorkspaceIds, - type WorkspaceDataId + type WorkspaceDataId, + generateId } from '@hcengineering/core' import { PlatformError, unknownError } from '@hcengineering/platform' import { @@ -123,6 +124,11 @@ class StorageBlobAdapter implements DbAdapter { return blobs } + getDomainHash (ctx: MeasureContext, domain: Domain): Promise { + // TODO: Check if possible to ask storage if there any changes. + return Promise.resolve(generateId()) + } + async upload (ctx: MeasureContext, domain: Domain, docs: Doc[]): Promise { // Nothing to do } diff --git a/server/server/src/client.ts b/server/server/src/client.ts index 843a90210e..5bc0336f34 100644 --- a/server/server/src/client.ts +++ b/server/server/src/client.ts @@ -250,6 +250,17 @@ export class ClientSession implements Session { } } + async getDomainHash (ctx: ClientSessionCtx, domain: Domain): Promise { + this.lastRequest = Date.now() + try { + const result = await this.getOps(ctx.pipeline).getDomainHash(ctx.ctx, domain) + await ctx.sendResponse(ctx.requestId, result) + } catch (err: any) { + await ctx.sendError(ctx.requestId, 'Failed to upload', unknownError(err)) + ctx.ctx.error('failed to getDomainHash', { domain, err }) + } + } + async closeChunk (ctx: ClientSessionCtx, idx: number): Promise { this.lastRequest = Date.now() await this.getOps(ctx.pipeline).closeChunk(ctx.ctx, idx)