Administrator/Multi-Hoster-Upload
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 47 Wiki Activity

Harden batch-wide recovery identity claims

Browse Source 
Scope remote code ownership to normalized hoster and account identities while retaining title-only recovery serialization and canonical Unicode title matching.

Mark post-upload ambiguity and duplicate identities as uncertain so retries, account fallback, and later same-title jobs fail closed instead of reporting unsafe success.

Acquire recovery title leases before hoster and global semaphores, revalidate failed-account overrides before upload, and clear claim state at batch boundaries.

Add deterministic concurrent coverage for same-code rejection, distinct-code parallel success, uncertainty propagation, semaphore fairness, account isolation, Unicode equivalence, and registry lifetime.
This commit is contained in:
Sucukdeluxe
2026-08-13 23:09:03 +02:00
parent dd14381e43
commit c78160a521
7 changed files with 640 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/upload-manager.js
+157 -73
View File
@@ -71,7 +71,7 @@ class UploadManager extends EventEmitter {
this._suspectGoodAccounts.clear();
this._doodApiKeyCache.clear();
this._baselineCache.clear();
this._recoveryClaims.clear();
if (!this.running) this._recoveryClaims.clear();
}
switchAccount(hoster, fallbackAccount) {
@@ -134,6 +134,41 @@ class UploadManager extends EventEmitter {
return true;
}
_swapFailedAccount(task, jobId, fileName) {
if (!task.accountId || !this._failedAccounts.has(task.hoster + ':' + task.accountId)) return false;
const override = this._accountOverrides.get(task.hoster);
if (override && !this._failedAccounts.has(task.hoster + ':' + override.id)) {
this._rotLog('pre-job-swap', {
jobId, hoster: task.hoster, fileName, fromAccountId: task.accountId, toAccountId: override.id
});
task.accountId = override.id;
task.username = override.username;
task.password = override.password;
task.apiKey = override.apiKey;
return true;
}
this._rotLog('pre-job-swap-blocked', {
jobId, hoster: task.hoster, fileName, accountId: task.accountId,
hasOverride: !!override,
overrideAlsoFailed: override ? this._failedAccounts.has(task.hoster + ':' + override.id) : false
});
return false;
}
async _coordinateAccountFailure(task, err, signal, jobId) {
if (!task.accountId || (err && err.remoteCommitUncertain === true)) return;
if (!this._shouldSkipRetryOnAccountError(err)) return;
const key = task.hoster + ':' + task.accountId;
if (this._failedAccounts.has(key)) return;
this._failedAccounts.set(key, true);
this._rotLog('mark-failed', {
jobId, hoster: task.hoster, fileName: path.basename(task.file),
accountId: task.accountId, lastError: err && err.message ? err.message : String(err)
});
this.emit('account-failed', { hoster: task.hoster, accountId: task.accountId });
await this._sleep(800, signal);
}
_rotLog(event, data) {
this.emit('rot-log', { ts: Date.now(), event, ...data });
}
@@ -161,6 +196,7 @@ class UploadManager extends EventEmitter {
// which takes priority in _shouldSkipRetryOnAccountError.
_isFileRejectedError(err) {
if (!err) return false;
if (err.remoteCommitUncertain === true) return false;
if (err.transientNetwork === true) return false;
if (err.accountError === true) return false; // explicit account-level wins
if (err.fileRejected === true) return true;
@@ -355,6 +391,7 @@ class UploadManager extends EventEmitter {
this._doodApiKeyCache.clear(); // re-derive doodstream keys fresh each batch
this._baselineCache.clear(); // re-fetch baselines per batch (a long batch could outlast remote-side relevance)
this._recoveryClaims.clear();
this._recoveryClaims = createRecoveryClaimRegistry();
this.semaphores = {};
this.globalSemaphore = null;
this.globalThrottle = null;
@@ -440,13 +477,12 @@ class UploadManager extends EventEmitter {
files
};
this._recoveryClaims.clear();
this.emit('batch-done', summary);
}
async _runJob(task, results, batchSignal) {
const settings = this._getSettings(task.hoster);
const hosterSemaphore = this._getSemaphore(task.hoster);
const globalSemaphore = this._getGlobalSemaphore();
const uploadId = crypto.randomBytes(8).toString('hex');
const jobId = task.jobId || uploadId;
const fileName = path.basename(task.file);
@@ -466,8 +502,6 @@ class UploadManager extends EventEmitter {
const { signal, cleanup: cleanupSignals } = this._combineSignals(batchSignal, jobAbortController.signal);
this.jobAbortControllers.set(jobId, jobAbortController);
let hosterSlotAcquired = false;
let globalSlotAcquired = false;
let finalResultRecorded = false;
let finalStatus = 'error';
let lastError = null;
@@ -549,9 +583,6 @@ class UploadManager extends EventEmitter {
// queueJobs array; the first event it actually needs from main is the
// 'getting-server' / 'uploading' transition for the jobs that the
// semaphore lets through.
await hosterSemaphore.acquire(signal);
hosterSlotAcquired = true;
let fileProbe = null;
try {
fileProbe = await probeFileHead(task.file, 512);
@@ -566,11 +597,6 @@ class UploadManager extends EventEmitter {
headHex: fileProbe && fileProbe.headHex ? fileProbe.headHex.slice(0, 32) : null
});
if (globalSemaphore) {
await globalSemaphore.acquire(signal);
globalSlotAcquired = true;
}
if (settings.timeIntervalSec > 0) {
await this._waitForInterval(task.hoster, settings.timeIntervalSec * 1000, signal);
}
@@ -580,24 +606,7 @@ class UploadManager extends EventEmitter {
// of burning a guaranteed-to-fail upload attempt. Critical at scale:
// with 500 queued jobs and 1 parallel slot, without this check every
// job still hits the original dead account first.
if (task.accountId && this._failedAccounts.has(task.hoster + ':' + task.accountId)) {
const override = this._accountOverrides.get(task.hoster);
if (override && !this._failedAccounts.has(task.hoster + ':' + override.id)) {
this._rotLog('pre-job-swap', {
jobId, hoster: task.hoster, fileName, fromAccountId: task.accountId, toAccountId: override.id
});
task.accountId = override.id;
task.username = override.username;
task.password = override.password;
task.apiKey = override.apiKey;
} else {
this._rotLog('pre-job-swap-blocked', {
jobId, hoster: task.hoster, fileName, accountId: task.accountId,
hasOverride: !!override,
overrideAlsoFailed: override ? this._failedAccounts.has(task.hoster + ':' + override.id) : false
});
}
}
this._swapFailedAccount(task, jobId, fileName);
// A previous file of at least this size already got a suspect rejection
// on this exact account — skip the guaranteed-to-fail multi-GB upload
@@ -735,7 +744,7 @@ class UploadManager extends EventEmitter {
} catch { /* progress callbacks must never throw — swallowing is correct, the stream keeps going */ }
};
const result = await this._executeUpload(task, progressCb, uploadSignalBundle.signal, throttle, fileProbe);
const result = await this._executeUploadWithAdmission(task, progressCb, uploadSignalBundle.signal, throttle, fileProbe, true, jobId);
if (signal.aborted || this.cancelledJobIds.has(jobId)) throw new Error('Aborted');
@@ -784,6 +793,11 @@ class UploadManager extends EventEmitter {
break;
}
if (err && err.remoteCommitUncertain === true) {
lastError = err;
break;
}
if (isSpeedRestart && attempt < maxAttempts) {
lastError = new Error('Geschwindigkeit zu niedrig - Neustart');
await this._sleep(3000, signal);
@@ -1047,7 +1061,7 @@ class UploadManager extends EventEmitter {
? { consume: async (bytes, sig) => { await hosterThrottle.consume(bytes, sig); await globalThrottle.consume(bytes, sig); } }
: hosterThrottle || globalThrottle;
const result = await this._executeUpload(task, progressCb, signal, throttle, fileProbe);
const result = await this._executeUploadWithAdmission(task, progressCb, signal, throttle, fileProbe, true, jobId);
if (signal.aborted || this.cancelledJobIds.has(jobId)) throw new Error('Aborted');
this.activeJobs.delete(uploadId);
this.sessionBytes += fileSize;
@@ -1104,9 +1118,6 @@ class UploadManager extends EventEmitter {
this.activeJobs.delete(uploadId);
this.jobAbortControllers.delete(jobId);
cleanupSignals();
// Release in reverse order of acquire (global first, then hoster)
if (globalSlotAcquired && globalSemaphore) globalSemaphore.release();
if (hosterSlotAcquired) hosterSemaphore.release();
this.emit('job-settled', {
jobId,
sourceCleanupToken: task.sourceCleanupToken || null,
@@ -1190,7 +1201,7 @@ class UploadManager extends EventEmitter {
? { consume: async (bytes, sig) => { await hosterThrottle.consume(bytes, sig); await globalThrottle.consume(bytes, sig); } }
: hosterThrottle || globalThrottle;
try {
const result = await this._executeUpload(task, progressCb, signal, throttle, fileProbe);
const result = await this._executeUploadWithAdmission(task, progressCb, signal, throttle, fileProbe, false, jobId);
if (signal.aborted || this.cancelledJobIds.has(jobId)) throw new Error('Aborted');
this.activeJobs.delete(uploadId);
this.sessionBytes += fileSize;
@@ -1210,6 +1221,7 @@ class UploadManager extends EventEmitter {
});
}
if (signal.aborted || this.stopAfterActive) break;
if (err && err.remoteCommitUncertain === true) throw err;
if (err && err.suspectReject === true) {
this._noteSuspectReject(task.hoster, account.id, fileSize);
}
@@ -1241,16 +1253,95 @@ class UploadManager extends EventEmitter {
return null;
}
async _executeUpload(task, progressCb, signal, throttle, fileProbe) {
const result = await this._executeUploadUnchecked(task, progressCb, signal, throttle, fileProbe);
return assertUploadConfirmation(result, task.hoster);
async _executeUploadWithAdmission(task, progressCb, signal, throttle, fileProbe, coordinateAccountFailure = true, jobId = task.jobId) {
while (true) {
const context = await this._createRecoveryContext(task);
let retryAdmission = false;
const operation = async () => {
const hosterSemaphore = this._getSemaphore(task.hoster);
const globalSemaphore = this._getGlobalSemaphore();
let hosterSlotAcquired = false;
let globalSlotAcquired = false;
try {
await hosterSemaphore.acquire(signal);
hosterSlotAcquired = true;
if (globalSemaphore) {
await globalSemaphore.acquire(signal);
globalSlotAcquired = true;
}
if (this._swapFailedAccount(task, jobId, path.basename(task.file))) {
retryAdmission = true;
return null;
}
try {
return await this._executeUpload(task, progressCb, signal, throttle, fileProbe, context);
} catch (err) {
if (coordinateAccountFailure) await this._coordinateAccountFailure(task, err, signal, jobId);
if (context.recoveryClaim && err && err.remoteCommitUncertain === true) {
throw context.recoveryClaim.markUncertain(err);
}
throw err;
}
} finally {
if (globalSlotAcquired && globalSemaphore) globalSemaphore.release();
if (hosterSlotAcquired) hosterSemaphore.release();
}
};
const result = context.recoveryClaim
? await context.recoveryClaim.runExclusive(operation, signal)
: await operation();
if (retryAdmission) continue;
return result;
}
}
async _executeUploadUnchecked(task, progressCb, signal, throttle, fileProbe) {
async _executeUpload(task, progressCb, signal, throttle, fileProbe, context) {
const result = await this._executeUploadUnchecked(task, progressCb, signal, throttle, fileProbe, context);
let confirmed;
try {
confirmed = assertUploadConfirmation(result, task.hoster);
} catch (err) {
if (context.recoveryClaim) throw context.recoveryClaim.markUncertain(err);
throw err;
}
return confirmed;
}
async _createRecoveryContext(task) {
const fileName = path.basename(task.file);
if ((task.hoster === 'vidmoly.me' || task.hoster === 'voe.sx') && task.username) {
const accountIdentity = task.accountId !== null && task.accountId !== undefined
? task.accountId
: String(task.username || '').trim().toLowerCase();
return {
recoveryClaim: this._recoveryClaims.forUpload(task.hoster, accountIdentity, fileName),
doodApiKey: null
};
}
if (task.hoster === 'doodstream.com' && task.username) {
const doodApiKey = await this._resolveDoodstreamApiKey(task);
const accountIdentity = doodApiKey || (task.accountId !== null && task.accountId !== undefined
? task.accountId
: String(task.username || '').trim().toLowerCase());
return {
recoveryClaim: this._recoveryClaims.forUpload(task.hoster, accountIdentity, fileName),
doodApiKey
};
}
if (task.hoster === 'byse.sx' || task.hoster === 'doodstream.com') {
return {
recoveryClaim: this._recoveryClaims.forUpload(task.hoster, task.apiKey, fileName),
doodApiKey: null
};
}
return { recoveryClaim: null, doodApiKey: null };
}
async _executeUploadUnchecked(task, progressCb, signal, throttle, fileProbe, context) {
if (task.hoster === 'vidmoly.me' && task.username) {
return this._executeRecoveryAwareLoginUpload(task, VidmolyUploader, progressCb, signal, throttle);
return this._executeRecoveryAwareLoginUpload(task, VidmolyUploader, progressCb, signal, throttle, context.recoveryClaim);
} else if (task.hoster === 'voe.sx' && task.username) {
return this._executeRecoveryAwareLoginUpload(task, VoeUploader, progressCb, signal, throttle);
return this._executeRecoveryAwareLoginUpload(task, VoeUploader, progressCb, signal, throttle, context.recoveryClaim);
} else if (task.hoster === 'doodstream.com' && task.username) {
// Login-path reliability fix: the web-form upload returns the filecode in
// an HTML form that comes back empty for large files (doodstream backend
@@ -1258,54 +1349,47 @@ class UploadManager extends EventEmitter {
// session ONCE per batch and upload via the official API instead — it
// returns result[0].filecode directly and has no empty-form failure mode.
// Falls back to the web-form upload if no valid key can be derived.
const apiKey = await this._resolveDoodstreamApiKey(task);
const apiKey = context.doodApiKey;
if (apiKey) {
this._rotLog('doodstream-via-api', { accountId: task.accountId, fileName: path.basename(task.file) });
return this._executeRecoveryAwareApiUpload('doodstream.com', task.file, apiKey, progressCb, signal, throttle, fileProbe);
return this._executeRecoveryAwareApiUpload('doodstream.com', task.file, apiKey, progressCb, signal, throttle, fileProbe, context.recoveryClaim);
}
this._rotLog('doodstream-via-web', { accountId: task.accountId, fileName: path.basename(task.file) });
const dood = new DoodstreamUploader();
await dood.login(task.username, task.password);
return dood.upload(task.file, progressCb, signal, throttle);
const result = await dood.upload(task.file, progressCb, signal, throttle);
if (result && result.file_code && !context.recoveryClaim.reserve(result.file_code)) {
const error = new Error('Doodstream Upload lieferte eine bereits zugeordnete Remote-Identität');
error.remoteIdentityClaimed = true;
throw context.recoveryClaim.markUncertain(error);
}
return result;
} else if (task.hoster === 'clouddrop.cc') {
const clouddrop = new ClouddropUploader(task.apiKey);
return clouddrop.upload(task.file, progressCb, signal, throttle);
} else {
if (task.hoster === 'byse.sx' || task.hoster === 'doodstream.com') {
return this._executeRecoveryAwareApiUpload(task.hoster, task.file, task.apiKey, progressCb, signal, throttle, fileProbe);
return this._executeRecoveryAwareApiUpload(task.hoster, task.file, task.apiKey, progressCb, signal, throttle, fileProbe, context.recoveryClaim);
}
return uploadFile(task.hoster, task.file, task.apiKey, progressCb, signal, throttle, {});
}
}
async _executeRecoveryAwareLoginUpload(task, UploaderClass, progressCb, signal, throttle) {
const accountIdentity = task.accountId !== null && task.accountId !== undefined
? task.accountId
: String(task.username || '').trim().toLowerCase();
const recoveryClaim = this._recoveryClaims.forUpload(
task.hoster,
accountIdentity,
path.basename(task.file)
);
return recoveryClaim.runExclusive(async () => {
const uploader = new UploaderClass(recoveryClaim);
await uploader.login(task.username, task.password);
return uploader.upload(task.file, progressCb, signal, throttle);
}, signal);
async _executeRecoveryAwareLoginUpload(task, UploaderClass, progressCb, signal, throttle, recoveryClaim) {
const uploader = new UploaderClass(recoveryClaim);
await uploader.login(task.username, task.password);
return uploader.upload(task.file, progressCb, signal, throttle);
}
async _executeRecoveryAwareApiUpload(hosterName, filePath, apiKey, progressCb, signal, throttle, fileProbe) {
const recoveryClaim = this._recoveryClaims.forUpload(hosterName, apiKey, path.basename(filePath));
return recoveryClaim.runExclusive(async () => {
const options = { recoveryClaim };
if (hosterName === 'byse.sx') {
options.byseBaseline = await this._getBaseline(hosterName, apiKey, signal);
if (fileProbe && fileProbe.ok !== false) options.probeIsVideoLike = fileProbe.isVideoLike === true;
} else {
options.doodBaseline = await this._getBaseline(hosterName, apiKey, signal);
}
return uploadFile(hosterName, filePath, apiKey, progressCb, signal, throttle, options);
}, signal);
async _executeRecoveryAwareApiUpload(hosterName, filePath, apiKey, progressCb, signal, throttle, fileProbe, recoveryClaim) {
const options = { recoveryClaim };
if (hosterName === 'byse.sx') {
options.byseBaseline = await this._getBaseline(hosterName, apiKey, signal);
if (fileProbe && fileProbe.ok !== false) options.probeIsVideoLike = fileProbe.isVideoLike === true;
} else {
options.doodBaseline = await this._getBaseline(hosterName, apiKey, signal);
}
return uploadFile(hosterName, filePath, apiKey, progressCb, signal, throttle, options);
}
_getBaseline(hosterName, apiKey, signal) {