The_Great_Archive/cryptpad
1
0
Fork 0
mirror of https://github.com/cryptpad/cryptpad.git synced 2026-09-12 19:49:59 +05:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1856 from cryptpad/eviction

Browse Source 
Fix eviction issues
This commit is contained in:
yflory 2025-04-16 15:38:43 +02:00 committed by GitHub
commit 1e8778fdf8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 86 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

127
lib/eviction.js
View File

@ -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
}));

30
lib/pins.js
View File

@ -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;
});

22
lib/storage/blob.js
View File

@ -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

1
lib/storage/file.js
View File

@ -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));
}

21
lib/workers/index.js
View File

@ -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);

2
scripts/check-account-deletion.js
View File

@ -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']

1
scripts/compare-pin-methods.js
View File

@ -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);

2
scripts/tests/test-pins.js
View File

@ -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) {