diff --git a/lib/eviction.js b/lib/eviction.js index b75fc78a6..41cc79b88 100644 --- a/lib/eviction.js +++ b/lib/eviction.js @@ -8,9 +8,7 @@ var Util = require("../lib/common-util"); var Pins = require("../lib/pins"); var Keys = require("./keys"); var Path = require('node:path'); -var config = require("./load-config"); var Fs = require("node:fs"); -var Fse = require("fs-extra"); var getNewestTime = function (stats) { return stats[['atime', 'ctime', 'mtime'].reduce(function (a, b) { @@ -76,103 +74,6 @@ var evictArchived = function (Env, cb) { blobs = Env.blobStore; }; - var migrateBlobRoot = function (from, to) { - // only migrate subpaths, leave everything else alone - if (!Path.dirname(from).startsWith(Path.dirname(to))) { return; } - - // expects a directory - var recurse = function (relativePath) { - var src = Path.join(from, relativePath); - var children; - try { - children = Fs.readdirSync(src); - } catch (err) { - if (err.code === 'ENOENT') { return; } - // if you can't read a directory's contents - // then nothing else will work, so just abort - Log.verbose("EVICT_ARCHIVED_NOT_DIRECTORY", { - error: err, - }); - return; - } - - var dest; - if (children.length === 0) { - try { - Fse.removeSync(src); - } catch (err2) { - Log.error('EVICT_ARCHIVED_EMPTY_DIR_REMOVAL', { - error: err2, - }); - // removal is non-essential, so we can continue - } - } else { - // make an equivalent path in the target directory - dest = Path.join(to, relativePath); - - try { - Fse.mkdirpSync(dest); - } catch (err3) { - Log.error("EVICT_ARCHIVED_BLOB_MIGRATION", { - error: err3, - }); - - // failure to create the host directory - // will cause problems when we try to move - // so bail out here - return; - } - } - - children.forEach(function (child) { - var childSrcPath = Path.join(src, child); - var stat = Fs.statSync(childSrcPath); - if (stat.isDirectory()) { - return void recurse(Path.join(relativePath, child)); - } - - var childDestPath = Path.join(dest, child); - - try { - Log.verbose("EVICT_ARCHIVED_MOVE_FROM_DEPRECATED_PATH", { - from: childSrcPath, - to: childDestPath, - }); - Fse.moveSync(childSrcPath, childDestPath, { - overwrite: false, - }); - } catch (err4) { - Log.error('EVICT_ARCHIVED_MOVE_FAILURE', { - error: err4, - }); - } - }); - }; - recurse(''); - }; - -/* In CryptPad 5.2.0 we merged a patch which converted - all of CryptPad's root filepaths to their absolute form, - rather than the relative paths we'd been using until then. - Unfortunately, we overlooked a case where two absolute - paths were concatenated together, resulting in blobs being - archived to an incorrect path. - - This migration detects evidence of incorrect archivals - and moves such archived files to their intended location - before continuing with the normal eviction procedure. -*/ - var migrateIncorrectBlobs = function () { - var incorrectPaths = [ - Path.join(Env.paths.archive, config.blobPath), - Path.join(Env.paths.archive, Path.resolve(config.blobPath)) - ]; - var correctPath = Path.join(Env.paths.archive, 'blob'); - incorrectPaths.forEach(root => { - migrateBlobRoot(root, correctPath); - }); - }; - var removeArchivedChannels = function (w) { // this block will iterate over archived channels and removes them // if they've been in cold storage for longer than your configured archive time @@ -236,6 +137,30 @@ var evictArchived = function (Env, cb) { store.listArchivedChannels(handler, w(done)); }; + // Blob proofs are no longer supported and can't be restored + // so we can delete them all + var removeArchivedBlobProofs = function (w) { + var archivePath = Path.join(Env.paths.archive, 'blob'); + const cb = Util.once(w()); + let i = 0; + nThen(w => { + Fs.readdir(archivePath, w((err, list) => { + if (err) { return; } + list.forEach(dir => { + // Look for 3 characters long folders + if (dir.length !== 3) { return; } + let path = Path.join(archivePath, dir); + Fs.rm(path, { recursive: true, force: true }, w(err => { + if (err) { return; } + i++; + })); + }); + })); + }).nThen(() => { + Log.info('EVICT_ARCHIVED_BLOB_PROOFS', i); + cb(); + }); + }; var removeArchivedBlobs = function (w) { if (typeof(Env.archiveRetentionTime) !== "number") { return; } // Iterate over archived blobs and remove them @@ -269,8 +194,8 @@ var evictArchived = function (Env, cb) { if (Env.DRY_RUN) { Env.Log.info('DRY RUN'); } nThen(loadStorage) - .nThen(migrateIncorrectBlobs) .nThen(removeArchivedChannels) + .nThen(removeArchivedBlobProofs) .nThen(removeArchivedBlobs) .nThen(function () { cb(void 0, report); @@ -588,7 +513,7 @@ module.exports = function (Env, cb) { if (newerItem && getNewestTime(newerItem) > retentionTime) { // it's actually active, so don't archive it. w.abort(); - cb(); + next(); } // else fall through to the archival })); diff --git a/lib/pins.js b/lib/pins.js index d2504077f..04eb3e85f 100644 --- a/lib/pins.js +++ b/lib/pins.js @@ -8,6 +8,7 @@ const Fs = require("fs"); const Path = require("path"); const Util = require("./common-util"); const Plan = require("./plan"); +const Store = require('./storage/file'); const Semaphore = require('saferphore'); const nThen = require('nthen'); @@ -258,8 +259,20 @@ Pins.load = function (cb, config) { var pinPath = config.pinPath || './pins'; var done = Util.once(cb); var handler = config.handler; + let store; nThen((waitFor) => { + Store.create({ + filePath: config.pinPath, + volumeId: 'pins' + }, waitFor((err, _) => { + if (err) { + waitFor.abort(); + return void done(err); + } + store = _; + })); + }).nThen((waitFor) => { // recurse over the configured pinPath, or the default Fs.readdir(pinPath, waitFor((err, list) => { if (err) { @@ -283,26 +296,29 @@ Pins.load = function (cb, config) { } list2.forEach((ff) => { if (config && config.exclude && config.exclude.indexOf(ff) > -1) { return; } - fileList.push(Path.join(pinPath, f, ff)); + fileList.push(ff.replace(/(\.ndjson)$/, '')); }); }))); }); }); }).nThen((waitFor) => { - fileList.forEach((f) => { + fileList.forEach((id) => { sema.take((returnAfter) => { var next = waitFor(returnAfter()); - Fs.readFile(f, (err, content) => { + var ref = {}; + var h = createLineHandler(ref, id); + store.readMessagesBin(id, 0, (msgObj, next) => { + h(msgObj.buff.toString('utf8')); + next(); + }, (err) => { if (err) { waitFor.abort(); return void done(err); } - var id = f.replace(/.*\/([^/]*).ndjson$/, (x, y)=>y); - var contentString = content.toString('utf8'); if (handler) { - return void handler(processPinFile(contentString, f), id, next); + return void handler(ref, id, next); } - const hashes = Pins.calculateFromLog(contentString, f); + const hashes = Object.keys(ref.pins); hashes.forEach((x) => { (pinned[x] = pinned[x] || {})[id] = 1; }); diff --git a/lib/storage/blob.js b/lib/storage/blob.js index 91ade5e2f..eec755b22 100644 --- a/lib/storage/blob.js +++ b/lib/storage/blob.js @@ -473,9 +473,29 @@ var archiveBlob = function (Env, blobId, reason, cb) { }; var removeArchivedBlob = function (Env, blobId, cb) { + var CB = Util.once(cb); var archivePath = prependArchive(Env, makeBlobPath(Env, blobId)); + var metadataPath = prependArchive(Env, mkMetadataPath(Env, blobId)); Fs.unlink(archivePath, cb); - removeArchivedActivity(Env, blobId, () => {}); + nThen(function (w) { + Fs.unlink(archivePath, w(function (err) { + if (err) { + if (err.code === "ENOENT") { return; } + w.abort(); + CB("E_ARCHIVED_BLOB_REMOVAL_"+ err.code); + } + })); + Fs.unlink(metadataPath, w(function (err) { + if (err) { + if (err.code === "ENOENT") { return; } + w.abort(); + CB("E_ARCHIVED_BLOBMD_REMOVAL_"+ err.code); + } + })); + removeArchivedActivity(Env, blobId, () => {}); + }).nThen(function () { + CB(); + }); }; // restoreBlob diff --git a/lib/storage/file.js b/lib/storage/file.js index 4ea2284f2..b6b42a1c1 100644 --- a/lib/storage/file.js +++ b/lib/storage/file.js @@ -573,6 +573,7 @@ var removeArchivedChannel = function (env, channelName, cb) { nThen(function (w) { Fs.unlink(channelPath, w(function (err) { if (err) { + if (err.code === "ENOENT") { return; } w.abort(); CB(labelError("E_ARCHIVED_CHANNEL_REMOVAL", err)); } diff --git a/lib/workers/index.js b/lib/workers/index.js index 4b38f3664..81fba424f 100644 --- a/lib/workers/index.js +++ b/lib/workers/index.js @@ -53,7 +53,7 @@ Workers.initialize = function (Env, config, _cb) { //return Object.keys(workers[index].tasks || {}).length; }; - const WORKER_TASK_LIMIT = 100000; // XXX + const WORKER_TASK_LIMIT = 250000; // XXX var workerOffset = -1; var queue = []; @@ -260,6 +260,16 @@ Workers.initialize = function (Env, config, _cb) { pid: worker.pid, // store the child process's id in an easily accessible location }; + let pid = worker.pid; + const onWorkerClosed = () => { + Object.keys(Env.plugins || {}).forEach(name => { + let plugin = Env.plugins[name]; + if (!plugin.onWorkerClosed) { return; } + try { plugin.onWorkerClosed("db-worker", pid); } + catch (e) {} + }); + }; + state.replaceWorker = () => { let index = workers.indexOf(state); if (index === -1) { return; } @@ -291,6 +301,7 @@ Workers.initialize = function (Env, config, _cb) { worker: state.worker.pid, count: state.count }); + onWorkerClosed(); delete state.worker; worker.kill(); }; @@ -323,14 +334,8 @@ Workers.initialize = function (Env, config, _cb) { handleResponse(state, res); }); - let pid = worker.pid; var substituteWorker = Util.once(function () { - Object.keys(Env.plugins || {}).forEach(name => { - let plugin = Env.plugins[name]; - if (!plugin.onWorkerClosed) { return; } - try { plugin.onWorkerClosed("db-worker", pid); } - catch (e) {} - }); + onWorkerClosed(); Env.Log.info("SUBSTITUTE_DB_WORKER", ''); var idx = workers.indexOf(state); diff --git a/scripts/check-account-deletion.js b/scripts/check-account-deletion.js index ef404279a..806c9d138 100644 --- a/scripts/check-account-deletion.js +++ b/scripts/check-account-deletion.js @@ -44,7 +44,7 @@ nThen((waitFor) => { pinned = Pins.calculateFromLog(content.toString('utf8'), f); })); }).nThen((waitFor) => { - Pins.list(waitFor((err, d) => { + Pins.load(waitFor((err, d) => { data = Object.keys(d); }), { exclude: [edPublic + '.ndjson'] diff --git a/scripts/compare-pin-methods.js b/scripts/compare-pin-methods.js index 299e6462f..222d9c30d 100644 --- a/scripts/compare-pin-methods.js +++ b/scripts/compare-pin-methods.js @@ -20,7 +20,6 @@ var compare = function () { Pins.list(w(function (err, p) { if (err) { throw err; } list = p; - console.log(p); console.log(list); console.log(); }), conf); diff --git a/scripts/tests/test-pins.js b/scripts/tests/test-pins.js index 67309b7c8..e180f0b56 100644 --- a/scripts/tests/test-pins.js +++ b/scripts/tests/test-pins.js @@ -33,7 +33,7 @@ var handler = function (ref, id /* safeKey */, pinned) { //console.log(ref, id); }; -Pins.list(function (err) { +Pins.load(function (err) { if (err) { return void console.error(err); } /* for (var id in pinned) {