From 90c95beefee96de3ea61e112dfd5833e1ecc7c2b Mon Sep 17 00:00:00 2001 From: yflory Date: Tue, 17 Dec 2024 16:05:28 +0100 Subject: [PATCH] Fix issues --- lib/archive-account.js | 18 ++- lib/eviction.js | 59 ---------- lib/storage/blob.js | 245 +++++++++++++++------------------------ lib/workers/db-worker.js | 6 - 4 files changed, 107 insertions(+), 221 deletions(-) diff --git a/lib/archive-account.js b/lib/archive-account.js index 43ff1357e..44928f6be 100644 --- a/lib/archive-account.js +++ b/lib/archive-account.js @@ -13,6 +13,7 @@ const Metadata = require("./commands/metadata"); const Meta = require("./metadata"); const Logger = require("./log"); const plugins = require("./plugin-manager"); +const HK = require('./hk-util'); let SSOUtils = plugins.SSO && plugins.SSO.utils; @@ -53,7 +54,13 @@ const init = (cb) => { Env.computeMetadata = function (channel, cb) { const ref = {}; const lineHandler = Meta.createLineHandler(ref, (err) => { console.log(err); }); - return void Env.store.readChannelMetadata(channel, lineHandler, function (err) { + + let f = Env.store.readChannelMetadata; + if (channel.length === HK.BLOB_ID_LENGTH) { + f = Env.blobStore.readMetadata; + } + + return void f(channel, lineHandler, function (err) { if (err) { // stream errors? return void cb(err); @@ -132,9 +139,12 @@ COMMANDS.start = (edPublic, blockId, reason) => { n = n((w) => { // Blobs if (Env.blobStore.isFileId(chanId)) { - return void Env.blobStore.isOwnedBy(safeKey, chanId, w((err, owned) => { - if (err || !owned) { return; } - blobsToArchive.push(chanId); + return Env.computeMetadata(chanId, w((e, md) => { + if (e || !md) { return; } + if (md && md.owners + && md.owners.includes(edPublic)) { + blobsToArchive.push(chanId); + } })); } // Pads diff --git a/lib/eviction.js b/lib/eviction.js index 95f430cdf..8975e3fdd 100644 --- a/lib/eviction.js +++ b/lib/eviction.js @@ -642,64 +642,6 @@ module.exports = function (Env, cb) { })); }; - var archiveInactiveBlobProofs = function (w) { - // iterate over blob proofs and remove them - // if they don't correspond to a pinned or active file - var removed = 0; - var total = 0; - - Log.info("EVICT_ARCHIVE_INACTIVE_BLOB_PROOFS_START", {}); - blobs.list.proofs(function (err, item, next) { - next = Util.mkAsync(next, THROTTLE_FACTOR); - if (err) { - return void Log.error("EVICT_BLOB_LIST_PROOFS_ERROR", err, next); - } - if (!item) { - return void Log.error('EVICT_BLOB_LIST_PROOFS_NO_ITEM', item, next); - } - total++; - - if (total % PROGRESS_FACTOR === 0) { - Log.info('EVICT_BLOB_PROOF_PROGRESS', { - proofs: total, - }); - } - - if (pinnedDocs.test(item.blobId)) { return void next(); } - if (item.mtime > inactiveTime) { return void next(); } - nThen(function (w) { - blobs.size(item.blobId, w(function (err, size) { - if (err && err === 'ENOENT') { return; } // XXX delete the proof - if (err) { - w.abort(); - return void Log.error("EVICT_BLOB_LIST_PROOFS_ERROR", err, next); - } - if (size !== 0) { - w.abort(); - next(); - } - })); - }).nThen(function () { - if (Env.DRY_RUN) { - removed++; - return void Log.info("EVICT_BLOB_PROOF_LONELY_DRY_RUN", item, next); - } - blobs.remove.proof(item.safeKey, item.blobId, function (err) { - if (err) { - return Log.error("EVICT_BLOB_PROOF_LONELY_ERROR", item, next); - } - removed++; - return Log.info("EVICT_BLOB_PROOF_LONELY", item, next); - }); - }); - }, w(function () { - Log.info("EVICT_BLOB_PROOFS_REMOVED", { - removed, - total, - }, w()); - })); - }; - var archiveInactiveChannels = function (w) { var channels = 0; var archived = 0; @@ -802,7 +744,6 @@ module.exports = function (Env, cb) { // (documents which are not in either bloom filter) .nThen(archiveInactiveBlobs) - .nThen(archiveInactiveBlobProofs) .nThen(archiveInactiveChannels) .nThen(function () { var runningTime = report.runningTime = msSinceStart(); diff --git a/lib/storage/blob.js b/lib/storage/blob.js index 8c6b9b2be..629dfd9d3 100644 --- a/lib/storage/blob.js +++ b/lib/storage/blob.js @@ -54,23 +54,10 @@ var makeStagePath = function (Env, safeKey) { return Path.join(Env.blobStagingPath, safeKey.slice(0, 2), safeKey); }; -// /blob//// -var makeProofPath = function (Env, safeKey, blobId) { - return Path.join(Env.blobPath, safeKey.slice(0, 3), safeKey, blobId.slice(0, 2), blobId); -}; - var mkPlaceholderPath = function (Env, blobId) { return makeBlobPath(Env, blobId) + '.placeholder'; }; -var parseProofPath = function (path) { - var parts = path.split('/'); - return { - blobId: parts[parts.length -1], - safeKey: parts[parts.length - 3], - }; -}; - // Placeholder for deleted files var addPlaceholder = function (Env, blobId, reason, cb) { if (!reason) { return cb(); } @@ -239,6 +226,11 @@ var archiveMetadata = (Env, blobId, cb) => { // if we fail to delete the metadata file, it can still be removed later by the eviction script Fse.move(path, archivePath, { overwrite: true }, cb); }; +var restoreActivity = function (Env, blobId, cb) { + var path = mkMetadataPath(Env, blobId); + var archivePath = prependArchive(Env, path); + Fse.move(archivePath, path, cb); +}; var readBlobMetadata = function (env, blobId, handler, _cb) { var metadataPath = mkMetadataPath(env, blobId); var stream = Fs.createReadStream(metadataPath, {start: 0}); @@ -403,7 +395,6 @@ var owned_upload_complete = function (Env, safeKey, id, cb) { var finalPath = makeBlobPath(Env, id); let unsafeKey = unescapeKeyCharacters(safeKey); - //var finalOwnPath = makeProofPath(Env, safeKey, id); // the user wants to move it into blob and create a metadata log with an owner @@ -465,19 +456,6 @@ var remove = function (Env, blobId, cb) { clearActivity(Env, blobId, () => {}); }; -// removeProof -var removeProof = function (Env, safeKey, blobId, cb) { - var proofPath = makeProofPath(Env, safeKey, blobId); - Fs.unlink(proofPath, cb); -}; - -// isOwnedBy(id, safeKey) -var isOwnedBy = function (Env, safeKey, blobId, cb) { - var proofPath = makeProofPath(Env, safeKey, blobId); - isFile(proofPath, cb); -}; - - // archiveBlob var archiveBlob = function (Env, blobId, reason, cb) { var blobPath = makeBlobPath(Env, blobId); @@ -499,29 +477,11 @@ var restoreBlob = function (Env, blobId, cb) { var blobPath = makeBlobPath(Env, blobId); var archivePath = prependArchive(Env, blobPath); Fse.move(archivePath, blobPath, cb); + restoreMetadata(Env, blobId, () => {}); restoreActivity(Env, blobId, () => {}); clearPlaceholder(Env, blobId, () => {}); }; -// archiveProof -var archiveProof = function (Env, safeKey, blobId, cb) { - var proofPath = makeProofPath(Env, safeKey, blobId); - var archivePath = prependArchive(Env, proofPath); - Fse.move(proofPath, archivePath, { overwrite: true }, cb); -}; - -var removeArchivedProof = function (Env, safeKey, blobId, cb) { - var archivedPath = prependArchive(Env, makeProofPath(Env, safeKey, blobId)); - Fs.unlink(archivedPath, cb); -}; - -// restoreProof -var restoreProof = function (Env, safeKey, blobId, cb) { - var proofPath = makeProofPath(Env, safeKey, blobId); - var archivePath = prependArchive(Env, proofPath); - Fse.move(archivePath, proofPath, cb); -}; - var makeWalker = function (n, handleChild, done) { if (!n || typeof(n) !== 'number' || n < 2) { n = 2; } @@ -553,7 +513,7 @@ var makeWalker = function (n, handleChild, done) { } if (!stats.isDirectory()) { w.abort(); - if (/\.activity$/.test(path)) { + if (/\.activity$/.test(path)) { // NOTE: some activity files were created for deleted blobs due to // a bug. We're going to detect them here in order to be able to clean // them. @@ -586,46 +546,6 @@ var makeWalker = function (n, handleChild, done) { return recurse; }; -var listProofs = function (root, handler, cb) { - Fs.readdir(root, function (err, dir) { - if (err) { return void cb(err); } - - var walk = makeWalker(20, function (err, path, next, loneActivity) { - if (loneActivity) { return void next(); } - // path is the path to a child node on the filesystem - - // next handles the next job in a queue - - // iterate over proofs - // check for presence of corresponding files - Fs.stat(path, function (err, stats) { - if (err) { - return void handler(err, void 0, next); - } - - var parsed = parseProofPath(path); - handler(void 0, { - path: path, - blobId: parsed.blobId, - safeKey: parsed.safeKey, - atime: stats.atime, - ctime: stats.ctime, - mtime: stats.mtime, - }, next); - }); - }, function () { - // called when there are no more directories or children to process - cb(); - }); - - dir.forEach(function (d) { - // ignore directories that aren't 3 characters long... - if (d.length !== 3) { return; } - walk(Path.join(root, d)); - }); - }); -}; - var getActivityStat = function (path, base, cb) { var suffix = base ? '' : '.activity'; Fs.stat(path+suffix, function (err, stats) { @@ -633,32 +553,91 @@ var getActivityStat = function (path, base, cb) { cb(err, stats); }); }; -var listBlobs = function (root, handler, cb) { - // iterate over files - Fs.readdir(root, function (err, dir) { - if (err) { return void cb(err); } - var walk = makeWalker(20, function (err, path, next, loneActivity) { - if (loneActivity) { return void next(); } - getActivityStat(path, false, function (err, stats) { - if (err) { - return void handler(err, void 0, next); - } - handler(void 0, { - blobId: Path.basename(path), - atime: stats.atime, - ctime: stats.ctime, - mtime: stats.mtime, - }, next); - }); - }, function () { - cb(); - }); +let blobRegex = /^[0-9a-fA-F]{48}(\.metadata)*(\.ndjson)*$/; +var listBlobs = function (root, handler, fast, cb) { + var dirList = []; - dir.forEach(function (d) { - if (d.length !== 2) { return; } - walk(Path.join(root, d)); + nThen(function (w) { + // the root of your datastore contains nested directories... + Fs.readdir(root, w(function (err, list) { + if (err) { + w.abort(); + // TODO check if we normally return strings or errors + return void cb(err); + } + dirList = list; + })); + }).nThen(function (waitFor) { + // search inside the nested directories + // stream it so you don't put unnecessary data in memory + var n = nThen; + dirList.forEach(function (dir) { + if (dir.length !== 2) { return; } + // Handle one directory at a time to save some memory + n = n(function (w) { + // do twenty things at a time + var sema = Semaphore.create(20); + var nestedDirPath = Path.join(root, dir); + Fs.readdir(nestedDirPath, w(function (err, list) { + if (err) { return void handler(err); } // Is this correct? + list.forEach(function (item) { + // ignore hidden files + if (/^\./.test(item)) { return; } + // ignore anything that isn't channel or metadata + if (!blobRegex.test(item)) { return; } + + var isLonelyMetadata = false; + var blobName; + var metadataName; + + // if the current file is not the channel data, then it must be metadata + if (!/^[0-9a-fA-F]{48}$/.test(item)) { + metadataName = item; + blobName = item.replace(/\.metadata\.ndjson/, ''); + // check if blob already exists + if (list.indexOf(blobName) !== -1) { return; } + // otherwise set a flag indicating that we should + // handle the metadata on its own + isLonelyMetadata = true; + } else { + blobName = item; + metadataName = blobName + '.metadata.ndjson'; + } + if (blobName.length !== 48) { return; } + + sema.take(function (give) { + var next = w(give()); + + if (fast) { + return void handler(void 0, { + blobId: blobName + }, next); + } + + var filePath = Path.join(nestedDirPath, blobName); + if (isLonelyMetadata) { + // Set time to 0 to delete this + // lonely metadata file + return void handler(void 0, { + blobId: blobName, + mtime: 0, + atime: 0, + ctime: 0 + }, next); + } + return void getActivityStat(filePath, false, (err, data) => { + data.blobId = blobName; + handler(err, data, next); + }); + }); + }); + })); + }).nThen; }); + n(waitFor()); + }).nThen(function () { + cb(); }); }; @@ -741,12 +720,6 @@ BlobStore.create = function (config, _cb) { upload_cancel(Env, safeKey, fileSize, cb); }, - isOwnedBy: function (safeKey, blobId, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } - isOwnedBy(Env, safeKey, blobId, cb); - }, - readMetadata: (blobId, handler, cb) => { if (!isValidId(blobId)) { return void cb("INVALID_ID"); } readBlobMetadata(Env, blobId, handler, cb); @@ -762,24 +735,12 @@ BlobStore.create = function (config, _cb) { if (!isValidId(blobId)) { return void cb("INVALID_ID"); } remove(Env, blobId, cb); }, - proof: function (safeKey, blobId, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } - if (!isValidId(blobId)) { return void cb("INVALID_ID"); } - removeProof(Env, safeKey, blobId, cb); - }, archived: { blob: function (blobId, _cb) { var cb = Util.once(Util.mkAsync(_cb)); if (!isValidId(blobId)) { return void cb("INVALID_ID"); } removeArchivedBlob(Env, blobId, cb); }, - proof: function (safeKey, blobId, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } - if (!isValidId(blobId)) { return void cb("INVALID_ID"); } - removeArchivedProof(Env, safeKey, blobId, cb); - }, }, loneActivity: function (_cb) { var cb = Util.once(Util.mkAsync(_cb)); @@ -793,12 +754,6 @@ BlobStore.create = function (config, _cb) { if (!isValidId(blobId)) { return void cb("INVALID_ID"); } archiveBlob(Env, blobId, reason, cb); }, - proof: function (safeKey, blobId, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } - if (!isValidId(blobId)) { return void cb("INVALID_ID"); } - archiveProof(Env, safeKey, blobId, cb); - }, }, restore: { @@ -807,12 +762,6 @@ BlobStore.create = function (config, _cb) { if (!isValidId(blobId)) { return void cb("INVALID_ID"); } restoreBlob(Env, blobId, cb); }, - proof: function (safeKey, blobId, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - if (!isValidSafeKey(safeKey)) { return void cb('INVALID_SAFEKEY'); } - if (!isValidId(blobId)) { return void cb("INVALID_ID"); } - restoreProof(Env, safeKey, blobId, cb); - }, }, isBlobAvailable: function (blobId, _cb) { @@ -865,22 +814,14 @@ BlobStore.create = function (config, _cb) { }, list: { - blobs: function (handler, _cb) { + blobs: function (handler, _cb, fast) { var cb = Util.once(Util.mkAsync(_cb)); - listBlobs(Env.blobPath, handler, cb); - }, - proofs: function (handler, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - listProofs(Env.blobPath, handler, cb); + listBlobs(Env.blobPath, handler, fast, cb); }, archived: { - proofs: function (handler, _cb) { + blobs: function (handler, _cb, fast) { var cb = Util.once(Util.mkAsync(_cb)); - listProofs(prependArchive(Env, Env.blobPath), handler, cb); - }, - blobs: function (handler, _cb) { - var cb = Util.once(Util.mkAsync(_cb)); - listBlobs(prependArchive(Env, Env.blobPath), handler, cb); + listBlobs(prependArchive(Env, Env.blobPath), handler, fast, cb); }, } }, diff --git a/lib/workers/db-worker.js b/lib/workers/db-worker.js index ddd49ccc0..d1a0fe962 100644 --- a/lib/workers/db-worker.js +++ b/lib/workers/db-worker.js @@ -570,12 +570,6 @@ const removeOwnedBlob = function (data, cb) { nThen(function (w) { // check if you have permissions - blobStore.isOwnedBy(safeKey, blobId, w(function (err, owned) { - if (err || !owned) { - w.abort(); - return void cb("INSUFFICIENT_PERMISSIONS"); - } - })); computeMetadata({channel: blobId}, w((err, meta) => { if (err || !meta) { w.abort();