'use strict'; exports = module.exports = { download, upload, verify, getFileExtension, copy, _saveFsMetadata: saveFsMetadata, _restoreFsMetadata: restoreFsMetadata }; const assert = require('node:assert'), async = require('async'), backupSites = require('../backupsites.js'), BoxError = require('../boxerror.js'), DataLayout = require('../datalayout.js'), { DecryptStream } = require('../hush.js'), debug = require('debug')('box:backupformat/rsync'), { EncryptStream } = require('../hush.js'), fs = require('node:fs'), HashStream = require('../hash-stream.js'), hush = require('../hush.js'), path = require('node:path'), paths = require('../paths.js'), ProgressStream = require('../progress-stream.js'), promiseRetry = require('../promise-retry.js'), safe = require('safetydance'), shell = require('../shell.js')('backupformat/rsync'), stream = require('stream/promises'), syncer = require('../syncer.js'), util = require('node:util'); async function addFile(sourceFile, encryption, uploader, progressCallback) { assert.strictEqual(typeof sourceFile, 'string'); assert.strictEqual(typeof encryption, 'object'); assert.strictEqual(typeof uploader, 'object'); assert.strictEqual(typeof progressCallback, 'function'); // make sure file can be opened for reading before we start the pipeline. otherwise, we end up with // destinations dirs/file which are owned by root (this process id) and cannot be copied (run as normal user) const [openError, sourceHandle] = await safe(fs.promises.open(sourceFile, 'r')); if (openError) { debug(`addFile: ignoring disappeared file: ${sourceFile}`); return null; } const sourceStream = sourceHandle.createReadStream(sourceFile, { autoClose: true }); const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds ps.on('progress', function (progress) { const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024); if (!transferred && !speed) return progressCallback({ message: `Uploading ${sourceFile}` }); // 0M@0MBps looks wrong progressCallback({ message: `Uploading ${sourceFile}: ${transferred}M@${speed}MBps` }); // 0M@0MBps looks wrong }); const hash = new HashStream(); let pipeline = null; if (encryption) { const encryptStream = new EncryptStream(encryption); pipeline = safe(stream.pipeline(sourceStream, encryptStream, ps, hash, uploader.stream)); } else { pipeline = safe(stream.pipeline(sourceStream, ps, hash, uploader.stream)); } const [error] = await safe(pipeline); if (error && !error.message.includes('ENOENT')) throw new BoxError(BoxError.EXTERNAL_ERROR, `tarPack pipeline error: ${error.message}`); // ignore error if file disappears // debug(`addFile: pipeline finished: ${JSON.stringify(ps.stats())}`); await uploader.finish(); return { stats: ps.stats(), integrity: { size: ps.stats().transferred, sha256: hash.digest('hex') } }; } async function sync(backupSite, remotePath, dataLayout, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof remotePath, 'string'); assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout'); assert.strictEqual(typeof progressCallback, 'function'); // the number here has to take into account the s3.upload partSize (which is 10MB). So 20=200MB const concurrency = backupSite.limits?.syncConcurrency || (backupSite.provider === 's3' ? 20 : 10); const cacheFile = path.join(paths.BACKUP_INFO_DIR, backupSite.id, `${dataLayout.getBasename()}.sync.cache`); const { delQueue, addQueue, integrityMap } = await syncer.sync(dataLayout, cacheFile); // integrityMap is unchanged files debug(`sync: processing ${delQueue.length} deletes and ${addQueue.length} additions`); const aggregatedStats = { transferred: 0, size: [...integrityMap.values()].reduce((sum, { size }) => sum + size, 0), fileCount: addQueue.length + integrityMap.size, // final file count, not the transferred file count startTime: Date.now(), totalMsecs: 0 }; async function processSyncerChange(change) { debug('sync: processing task: %j', change); // the empty task.path is special to signify the directory const destPath = change.path && backupSite.encryption?.encryptedFilenames ? hush.encryptFilePath(change.path, backupSite.encryption) : change.path; const fullPath = path.join(remotePath, destPath); if (change.operation === 'removedir') { debug(`Removing directory ${fullPath}`); await backupSites.storageApi(backupSite).removeDir(backupSite.config, fullPath, progressCallback); } else if (change.operation === 'remove') { debug(`Removing ${fullPath}`); await backupSites.storageApi(backupSite).remove(backupSite.config, fullPath); } else if (change.operation === 'add') { await promiseRetry({ times: 5, interval: 20000, debug }, async (retryCount) => { progressCallback({ message: `Adding ${change.path}` + (retryCount > 1 ? ` (Try ${retryCount})` : '') }); debug(`Adding ${change.path} position ${change.position} try ${retryCount}`); const uploader = await backupSites.storageApi(backupSite).upload(backupSite.config, fullPath); const result = await addFile(dataLayout.toLocalPath('./' + change.path), backupSite.encryption, uploader, progressCallback); if (!result) return; // this can happen if the file disappeared on us integrityMap.set(destPath, result.integrity); aggregatedStats.transferred += result.stats.transferred; aggregatedStats.size += result.stats.transferred; }); } } const [delError] = await safe(async.eachLimit(delQueue, concurrency, async (change) => await processSyncerChange(change, backupSite, remotePath, dataLayout, progressCallback))); debug('sync: done processing deletes. error: %o', delError); if (delError) throw delError; const [addError] = await safe(async.eachLimit(addQueue, concurrency, async (change) => await processSyncerChange(change, backupSite, remotePath, dataLayout, progressCallback))); debug('sync: done processing adds. error: %o', addError); if (addError) throw addError; await syncer.finalize(integrityMap, cacheFile); return { stats: { ...aggregatedStats, totalMsecs: Date.now()-aggregatedStats.startTime }, integrityMap }; } // this is not part of 'snapshotting' because we need root access to traverse async function saveFsMetadata(dataLayout, metadataFile) { assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout'); assert.strictEqual(typeof metadataFile, 'string'); // contains paths prefixed with './' const metadata = { emptyDirs: [], execFiles: [], symlinks: [] }; // we assume small number of files. spawnSync will raise a ENOBUFS error after maxBuffer for (const lp of dataLayout.localPaths()) { const [emptyDirsError, emptyDirs] = await safe(shell.spawn('find', [lp, '-type', 'd', '-empty'], { encoding: 'utf8', maxLines: 50000 })); if (emptyDirsError && emptyDirsError.stdoutLineCount >= 50000) throw new BoxError(BoxError.FS_ERROR, `Too many empty directories. Run "find ${lp} -type d -empty" to investigate`); if (emptyDirsError) throw emptyDirsError; if (emptyDirs.length) metadata.emptyDirs = metadata.emptyDirs.concat(emptyDirs.trim().split('\n').map((ed) => dataLayout.toRemotePath(ed))); const [execFilesError, execFiles] = await safe(shell.spawn('find', [lp, '-type', 'f', '-executable'], { encoding: 'utf8', maxLines: 20000 })); if (execFilesError && execFilesError.stdoutLineCount >= 20000) throw new BoxError(BoxError.FS_ERROR, `Too many executable files. Run "find ${lp} -type f -executable" to investigate`); if (execFilesError) throw execFilesError; if (execFiles.length) metadata.execFiles = metadata.execFiles.concat(execFiles.trim().split('\n').map((ef) => dataLayout.toRemotePath(ef))); const [symlinkFilesError, symlinkFiles] = await safe(shell.spawn('find', [lp, '-type', 'l'], { encoding: 'utf8', maxLines: 20000 })); if (symlinkFilesError && symlinkFilesError.stdoutLineCount >= 20000) throw new BoxError(BoxError.FS_ERROR, `Too many symlinks. Run "find ${lp} -type l" to investigate`); if (symlinkFilesError) throw symlinkFilesError; if (symlinkFiles.length) metadata.symlinks = metadata.symlinks.concat(symlinkFiles.trim().split('\n').map((sl) => { const site = safe.fs.readlinkSync(sl); return { path: dataLayout.toRemotePath(sl), site }; })); } if (!safe.fs.writeFileSync(metadataFile, JSON.stringify(metadata, null, 2))) throw new BoxError(BoxError.FS_ERROR, `Error writing fs metadata: ${safe.error.message}`); } async function restoreFsMetadata(dataLayout, metadataFile) { assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout'); assert.strictEqual(typeof metadataFile, 'string'); debug(`Recreating empty directories in ${dataLayout.toString()}`); const metadataJson = safe.fs.readFileSync(metadataFile, 'utf8'); if (metadataJson === null) throw new BoxError(BoxError.EXTERNAL_ERROR, 'Error loading fsmetadata.json:' + safe.error.message); const metadata = safe.JSON.parse(metadataJson); if (metadata === null) throw new BoxError(BoxError.EXTERNAL_ERROR, 'Error parsing fsmetadata.json:' + safe.error.message); for (const emptyDir of metadata.emptyDirs) { const [mkdirError] = await safe(fs.promises.mkdir(dataLayout.toLocalPath(emptyDir), { recursive: true })); if (mkdirError) throw new BoxError(BoxError.FS_ERROR, `unable to create path: ${mkdirError.message}`); } for (const execFile of metadata.execFiles) { const [chmodError] = await safe(fs.promises.chmod(dataLayout.toLocalPath(execFile), parseInt('0755', 8))); if (chmodError) throw new BoxError(BoxError.FS_ERROR, `unable to chmod: ${chmodError.message}`); } for (const symlink of (metadata.symlinks || [])) { if (!symlink.site) continue; // the path may not exist if we had a directory full of symlinks const [mkdirError] = await safe(fs.promises.mkdir(path.dirname(dataLayout.toLocalPath(symlink.path)), { recursive: true })); if (mkdirError) throw new BoxError(BoxError.FS_ERROR, `unable to symlink (mkdir): ${mkdirError.message}`); const [symlinkError] = await safe(fs.promises.symlink(symlink.site, dataLayout.toLocalPath(symlink.path), 'file')); if (symlinkError) throw new BoxError(BoxError.FS_ERROR, `unable to symlink: ${symlinkError.message}`); } } async function downloadDir(backupSite, remotePath, dataLayout, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof remotePath, 'string'); assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout'); assert.strictEqual(typeof progressCallback, 'function'); const encryptedFilenames = backupSite.encryption?.encryptedFilenames || false; debug(`downloadDir: ${remotePath} to ${dataLayout.toString()}. encryption filenames: ${encryptedFilenames}. encrypted files: ${!!backupSite.encryption}`); async function downloadFile(entry) { let relativePath = path.relative(remotePath, entry.path); if (encryptedFilenames) { const { error, result } = hush.decryptFilePath(relativePath, backupSite.encryption); if (error) throw new BoxError(BoxError.CRYPTO_ERROR, 'Unable to decrypt file'); relativePath = result; } const destFilePath = dataLayout.toLocalPath('./' + relativePath); const [mkdirError] = await safe(fs.promises.mkdir(path.dirname(destFilePath), { recursive: true })); if (mkdirError) throw new BoxError(BoxError.FS_ERROR, mkdirError.message); await promiseRetry({ times: 3, interval: 20000 }, async function () { progressCallback({ message: `Downloading ${entry.path} to ${destFilePath}` }); const [downloadError, sourceStream] = await safe(backupSites.storageApi(backupSite).download(backupSite.config, entry.path)); if (downloadError) { progressCallback({ message: `Download ${entry.path} to ${destFilePath} errored: ${downloadError.message}` }); throw downloadError; } const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds ps.on('progress', function (progress) { const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024); if (!transferred && !speed) return progressCallback({ message: `Downloading ${entry.path}` }); // 0M@0MBps looks wrong progressCallback({ message: `Downloading ${entry.path}: ${transferred}M@${speed}MBps` }); }); const destStream = fs.createWriteStream(destFilePath); const streams = [ sourceStream, ps ]; if (backupSite.encryption) { const decryptStream = new DecryptStream(backupSite.encryption); streams.push(decryptStream); } streams.push(destStream); const [pipelineError] = await safe(stream.pipeline(streams)); if (pipelineError) { progressCallback({ message: `Download error ${entry.path} to ${destFilePath}: ${pipelineError.message}` }); throw pipelineError; } progressCallback({ message: `Download finished ${entry.path} to ${destFilePath}` }); }); } // https://www.digitalocean.com/community/questions/rate-limiting-on-spaces?answer=40441 const concurrency = backupSite.limits?.downloadConcurrency || (backupSite.provider === 's3' ? 30 : 10); let marker = null; while (true) { const batch = await backupSites.storageApi(backupSite).listDir(backupSite.config, remotePath, marker === null ? 1 : 1000, marker); // try with one file first. if that works out, we continue faster await async.eachLimit(batch.entries, concurrency, downloadFile); if (!batch.marker) break; marker = batch.marker; } } async function download(backupSite, remotePath, dataLayout, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof remotePath, 'string'); assert(dataLayout instanceof DataLayout, 'dataLayout must be a DataLayout'); assert.strictEqual(typeof progressCallback, 'function'); debug(`download: Downloading ${remotePath} to ${dataLayout.toString()}`); await downloadDir(backupSite, remotePath, dataLayout, progressCallback); await restoreFsMetadata(dataLayout, `${dataLayout.localRoot()}/fsmetadata.json`); } async function upload(backupSite, remotePath, dataLayout, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof remotePath, 'string'); assert.strictEqual(typeof dataLayout, 'object'); assert.strictEqual(typeof progressCallback, 'function'); debug(`upload: uploading to site ${backupSite.id} path ${remotePath} (encrypted: ${!!backupSite.encryption}) dataLayout ${dataLayout.toString()}`); await saveFsMetadata(dataLayout, `${dataLayout.localRoot()}/fsmetadata.json`); return await sync(backupSite, remotePath, dataLayout, progressCallback); // { stats, integrityMap } } async function copy(backupSite, fromPath, toPath, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof fromPath, 'string'); assert.strictEqual(typeof toPath, 'string'); assert.strictEqual(typeof progressCallback, 'function'); await backupSites.storageApi(backupSite).copyDir(backupSite.config, fromPath, toPath, progressCallback); } function getFileExtension(encryption) { assert.strictEqual(typeof encryption, 'boolean'); return ''; // this also signals to backupcleanear that we are dealing with directories } async function verify(backupSite, remotePath, integrityMap, progressCallback) { assert.strictEqual(typeof backupSite, 'object'); assert.strictEqual(typeof remotePath, 'string'); assert(util.types.isMap(integrityMap), 'integrityMap should be a Map'); assert.strictEqual(typeof progressCallback, 'function'); debug(`verify: Verifying ${remotePath}`); const encryptedFilenames = backupSite.encryption?.encryptedFilenames || false; let fileCount = 0; async function validateFile(entry) { let relativePath = path.relative(remotePath, entry.path); if (encryptedFilenames) { const { error, result } = hush.decryptFilePath(relativePath, backupSite.encryption); if (error) throw new BoxError(BoxError.CRYPTO_ERROR, 'Unable to decrypt file'); relativePath = result; } ++fileCount; const sourceStream = await backupSites.storageApi(backupSite).download(backupSite.config, entry.path); const ps = new ProgressStream({ interval: 10000 }); // display a progress every 10 seconds ps.on('progress', function (progress) { const transferred = Math.round(progress.transferred/1024/1024), speed = Math.round(progress.speed/1024/1024); if (!transferred && !speed) return progressCallback({ message: `Downloading ${entry.path}` }); // 0M@0MBps looks wrong progressCallback({ message: `Downloading ${entry.path}: ${transferred}M@${speed}MBps` }); }); const streams = [ sourceStream, ps ]; if (backupSite.encryption) { const decryptStream = new DecryptStream(backupSite.encryption); streams.push(decryptStream); } const hash = new HashStream(); streams.push(hash); await stream.pipeline(streams); const integrity = integrityMap.get(relativePath); if (ps.stats().transferred !== integrity.size) throw new BoxError(BoxError.BAD_STATE, `${entry.path} has size ${ps.stats().transferred}. Expecting ${integrity.size}`); if (hash.digest() !== integrity.sha256) throw new BoxError(BoxError.BAD_STATE, `${entry.path} has size ${hash.digest()}. Expecting ${integrity.sha256}`); } debug(integrityMap.entries()); // https://www.digitalocean.com/community/questions/rate-limiting-on-spaces?answer=40441 const concurrency = backupSite.limits?.downloadConcurrency || (backupSite.provider === 's3' ? 30 : 10); let marker = null; while (true) { const batch = await backupSites.storageApi(backupSite).listDir(backupSite.config, remotePath, marker === null ? 1 : 1000, marker); // try with one file first. if that works out, we continue faster await async.eachLimit(batch.entries, concurrency, validateFile); if (!batch.marker) break; marker = batch.marker; } const check = (x, y) => { return x === y ? { status: 'passed' } : { status: 'failed', message: `Expecting ${x} but got ${y}` }; }; if (integrityMap.size !== fileCount) throw new BoxError(BoxError.BAD_STATE, `Got ${fileCount} files. Expecting ${integrityMap.size()} files`); return { size: { status: 'passed' }, fileCount: check(integrityMap.size, fileCount), sha256: { status: 'passed' }, }; }