Mega-Debrid: Link-Umwandlung parallel ueber mehrere Accounts (per-Account-Queue + Tiefen-Routing)
User-Wunsch: die Links eines Pakets parallel umwandeln statt seriell (single-flight),
damit sich die Download-Slots schneller fuellen. Gewaehlt: parallel ueber mehrere
Accounts (ein Login je Account laeuft parallel), NIE zwei gleichzeitig auf demselben
Account (Mega-Debrid-Sperr-Risiko).
- MegaWebFallback: globale Einzel-Warteschlange (this.queue) -> Warteschlange PRO Account
(this.queues: Map<login, Promise>). Gleicher Account serialisiert (kein Doppel-Login,
kein Hammern), verschiedene Accounts parallel. Key vor runExclusive berechnet.
- debrid.ts unrestrictWithAccounts: megaDebridInFlight zaehlt die LAUFENDE Tiefe pro
Account (Map<`${id}:${mode}`, number>). Kandidaten werden nach aufsteigender Tiefe
stabil sortiert (cursorOrder als Gleichstand-Tiebreak): gleichzeitige Aufloesungen
greifen den am wenigsten belegten Account -> auch bei mehr Links als Accounts
gleichmaessige Verteilung (4 Accounts, 8 parallel -> 2 je Account), statt sich hinter
dem Cursor-Account zu stauen. Sequenziell (alles Tiefe 0) bleibt es klebrig beim warmen
Account. add/inc vor dem try, dec/cleanup im finally (kein Leak).
- classifyAccountFailure: "Queue-Timeout" (lokaler Eigen-Stau) gibt jetzt cooldownMs:0 —
der warme Account wird nicht mehr faelschlich fuer Eigen-Stau mit Cooldown bestraft.
Vor Release adversarial per Multi-Agent-Workflow geprueft (4 Lenses + Verify). Der Review
fand genau die Ueberzahl-Stau-Schwaeche (binaeres belegt/frei staute >Accounts-Links hinter
einem Account) — daraufhin auf Tiefen-Zaehlung umgestellt. Sperr-Risiko strukturell
ausgeschlossen (per-Account-Queue serialisiert unabhaengig vom Set). 4 neue Tests
(gleicher Account 1 Login, verschiedene Accounts parallel, 4 gleichzeitig->4 Accounts,
8/2-Ueberzahl->4/4 ausgewogen). 813 Tests, tsc=6, self-check + build ok.
This commit is contained in:
Sucukdeluxe
parent
232af28715
commit
2c596bbc8d
@ -306,6 +306,13 @@ let megaDebridStickyCount = 0;
|
||||
// naechsten gewechselt. So bleibt es schnell UND ueber die Zeit kommen alle dran.
|
||||
export const MEGA_DEBRID_STICKY_LINKS = 25;
|
||||
|
||||
// Wie viele Umwandlungen pro Account (key `${id}:${mode}`) GERADE laufen/anstehen. Gleichzeitige
|
||||
// Aufloesungen waehlen den am WENIGSTEN belegten Account → verschiedene Accounts wandeln parallel
|
||||
// um (je eigene Queue in MegaWebFallback), und auch bei mehr gleichzeitigen Links als Accounts
|
||||
// verteilt es sich gleichmaessig statt sich hinter einem Account zu stauen. Tiefe (nicht nur
|
||||
// belegt/frei), damit ein frueh fertiger Erst-Job einen noch laufenden Folge-Job nicht "frei" meldet.
|
||||
const megaDebridInFlight = new Map<string, number>();
|
||||
|
||||
export function recordMegaDebridEmptyResponseStreak(accountId: string): number {
|
||||
const streak = (megaDebridEmptyResponseStreaks.get(accountId) || 0) + 1;
|
||||
megaDebridEmptyResponseStreaks.set(accountId, streak);
|
||||
@ -321,6 +328,7 @@ export function resetMegaDebridRuntimeStateForTests(): void {
|
||||
megaDebridEmptyResponseStreaks.clear();
|
||||
megaDebridRotationCursor = 0;
|
||||
megaDebridStickyCount = 0;
|
||||
megaDebridInFlight.clear();
|
||||
}
|
||||
|
||||
export function pruneExpiredMegaDebridRuntimeState(now = Date.now()): number {
|
||||
@ -1923,11 +1931,22 @@ class MegaDebridClient {
|
||||
// Schwung erfolgreicher Umwandlungen (MEGA_DEBRID_STICKY_LINKS) — sodass aufeinander
|
||||
// folgende Links auf demselben warmen Account laufen statt jeweils neu einzuloggen.
|
||||
const startOffset = ((megaDebridRotationCursor % accounts.length) + accounts.length) % accounts.length;
|
||||
const orderedEntries: { account: MegaDebridAccountEntry; idx: number }[] = [];
|
||||
const cursorOrder: { account: MegaDebridAccountEntry; idx: number }[] = [];
|
||||
for (let step = 0; step < accounts.length; step += 1) {
|
||||
const idx = (startOffset + step) % accounts.length;
|
||||
orderedEntries.push({ account: accounts[idx], idx });
|
||||
cursorOrder.push({ account: accounts[idx], idx });
|
||||
}
|
||||
// Parallel ueber mehrere Accounts: nach AKTUELLER Auslastung (in-flight-Tiefe) sortieren —
|
||||
// am wenigsten belegter Account zuerst, cursorOrder als stabiler Gleichstand-Tiebreak. So
|
||||
// verteilen sich auch MEHR gleichzeitige Aufloesungen als Accounts gleichmaessig (statt sich
|
||||
// alle hinter dem Cursor-Account zu stauen). Nichts in-flight (sequenziell) => alle Tiefe 0 =>
|
||||
// Reihenfolge == cursorOrder => bleibt klebrig beim warmen Account.
|
||||
const inFlightDepth = (entry: { account: MegaDebridAccountEntry }): number =>
|
||||
megaDebridInFlight.get(`${entry.account.id}:${mode}`) ?? 0;
|
||||
const orderedEntries = cursorOrder
|
||||
.map((entry, position) => ({ entry, position }))
|
||||
.sort((a, b) => (inFlightDepth(a.entry) - inFlightDepth(b.entry)) || (a.position - b.position))
|
||||
.map((wrapped) => wrapped.entry);
|
||||
|
||||
for (let orderPos = 0; orderPos < orderedEntries.length; orderPos += 1) {
|
||||
const entry = orderedEntries[orderPos];
|
||||
@ -1977,6 +1996,7 @@ class MegaDebridClient {
|
||||
const testStartedAt = Date.now();
|
||||
|
||||
usableAccountSeen = true;
|
||||
megaDebridInFlight.set(cooldownKey, (megaDebridInFlight.get(cooldownKey) ?? 0) + 1);
|
||||
try {
|
||||
const client = new MegaDebridClient(account.login, account.password, mode, allowApiFallback, megaWebUnrestrict);
|
||||
const result = await client.unrestrictLink(link, signal);
|
||||
@ -2078,6 +2098,13 @@ class MegaDebridClient {
|
||||
next: nextLabel,
|
||||
link: linkShort
|
||||
});
|
||||
} finally {
|
||||
const remainingInFlight = (megaDebridInFlight.get(cooldownKey) ?? 1) - 1;
|
||||
if (remainingInFlight <= 0) {
|
||||
megaDebridInFlight.delete(cooldownKey);
|
||||
} else {
|
||||
megaDebridInFlight.set(cooldownKey, remainingInFlight);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -2154,6 +2181,13 @@ class MegaDebridClient {
|
||||
};
|
||||
}
|
||||
|
||||
if (/queue.?timeout/i.test(errorText)) {
|
||||
// Lokaler Stau (die Queue dieses Accounts war zu lange belegt) — KEIN Signal fuer einen
|
||||
// ungesunden Account. Kein Cooldown, sonst wuerde der warme Account fuer Eigen-Stau bestraft;
|
||||
// der Link wird einfach erneut versucht und rotiert dann natuerlich weiter.
|
||||
return { fatal: false, cooldownMs: 0, message: errorText, category: "temporary" };
|
||||
}
|
||||
|
||||
if (isRetryableErrorText(errorText) || /timeout|network|fetch|socket/i.test(errorText)) {
|
||||
return {
|
||||
fatal: false,
|
||||
|
||||
@ -217,7 +217,11 @@ async function raceWithAbort<T>(promise: Promise<T>, signal?: AbortSignal): Prom
|
||||
}
|
||||
|
||||
export class MegaWebFallback {
|
||||
private queue: Promise<unknown> = Promise.resolve();
|
||||
// Pro Account eine eigene Warteschlange: Umwandlungen auf DEMSELBEN Account laufen
|
||||
// seriell (kein Doppel-Login, kein Hammern eines einzelnen Accounts), verschiedene
|
||||
// Accounts laufen parallel. So koennen die Links eines Pakets ueber mehrere Accounts
|
||||
// gleichzeitig umgewandelt werden statt global eine nach der anderen.
|
||||
private queues = new Map<string, Promise<unknown>>();
|
||||
|
||||
private getCredentials: () => MegaCredentials;
|
||||
|
||||
@ -233,15 +237,15 @@ export class MegaWebFallback {
|
||||
account?: { login: string; password: string }
|
||||
): Promise<UnrestrictedLink | null> {
|
||||
const overallSignal = withTimeoutSignal(signal, 180000);
|
||||
const creds = (account && account.login.trim() && account.password.trim())
|
||||
? account
|
||||
: this.getCredentials();
|
||||
if (!creds.login.trim() || !creds.password.trim()) {
|
||||
return null;
|
||||
}
|
||||
const key = creds.login.trim().toLowerCase();
|
||||
return this.runExclusive(async () => {
|
||||
throwIfAborted(overallSignal);
|
||||
const creds = (account && account.login.trim() && account.password.trim())
|
||||
? account
|
||||
: this.getCredentials();
|
||||
if (!creds.login.trim() || !creds.password.trim()) {
|
||||
return null;
|
||||
}
|
||||
const key = creds.login.trim().toLowerCase();
|
||||
let cookie = await this.ensureSession(key, creds.login, creds.password, overallSignal);
|
||||
|
||||
let generated = await this.generate(link, cookie, overallSignal);
|
||||
@ -259,7 +263,7 @@ export class MegaWebFallback {
|
||||
fileSize: null,
|
||||
retriesUsed: 0
|
||||
};
|
||||
}, overallSignal);
|
||||
}, key, overallSignal);
|
||||
}
|
||||
|
||||
private async ensureSession(key: string, login: string, password: string, signal?: AbortSignal): Promise<string> {
|
||||
@ -276,7 +280,7 @@ export class MegaWebFallback {
|
||||
this.sessions.clear();
|
||||
}
|
||||
|
||||
private async runExclusive<T>(job: () => Promise<T>, signal?: AbortSignal): Promise<T> {
|
||||
private async runExclusive<T>(job: () => Promise<T>, key: string, signal?: AbortSignal): Promise<T> {
|
||||
const queuedAt = Date.now();
|
||||
const QUEUE_WAIT_TIMEOUT_MS = 90000;
|
||||
const guardedJob = async (): Promise<T> => {
|
||||
@ -287,8 +291,9 @@ export class MegaWebFallback {
|
||||
}
|
||||
return job();
|
||||
};
|
||||
const run = this.queue.then(guardedJob, guardedJob);
|
||||
this.queue = run.then(() => undefined, () => undefined);
|
||||
const prev = this.queues.get(key) ?? Promise.resolve();
|
||||
const run = prev.then(guardedJob, guardedJob);
|
||||
this.queues.set(key, run.then(() => undefined, () => undefined));
|
||||
return raceWithAbort(run, signal);
|
||||
}
|
||||
|
||||
|
||||
@ -1460,6 +1460,77 @@ describe("debrid service", () => {
|
||||
expect(usedIds).toEqual(new Array(3).fill(getMegaDebridAccountId("user2")));
|
||||
}, 30000);
|
||||
|
||||
it("verteilt gleichzeitige Umwandlungen auf verschiedene Accounts (parallel, in-flight-Routing)", async () => {
|
||||
const settings = {
|
||||
...defaultSettings(),
|
||||
token: "", bestToken: "", allDebridToken: "",
|
||||
megaLogin: "user1", megaPassword: "pass1",
|
||||
megaCredentials: "user1:pass1\nuser2:pass2\nuser3:pass3\nuser4:pass4",
|
||||
megaDebridPreferApi: false,
|
||||
providerOrder: [] as const, providerPrimary: "megadebrid" as const,
|
||||
providerSecondary: "none" as const, providerTertiary: "none" as const,
|
||||
autoProviderFallback: false
|
||||
};
|
||||
|
||||
globalThis.fetch = (async () => new Response("error", { status: 500 })) as typeof fetch;
|
||||
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
const accountsSeen = new Set<string>();
|
||||
const allInFlight = new Promise<void>((resolve) => {
|
||||
const check = setInterval(() => { if (active >= 4) { clearInterval(check); resolve(); } }, 5);
|
||||
});
|
||||
const megaWeb = vi.fn(async (_link: string, _signal: AbortSignal | undefined, account?: { login: string; password: string }) => {
|
||||
active += 1;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
if (account) accountsSeen.add(account.login);
|
||||
await allInFlight;
|
||||
active -= 1;
|
||||
return { fileName: "ok.rar", directUrl: "https://mega-web.example/ok.rar", fileSize: null, retriesUsed: 0 };
|
||||
});
|
||||
|
||||
const service = new DebridService(settings, { megaWebUnrestrict: megaWeb });
|
||||
const results = await Promise.all([0, 1, 2, 3].map((i) => service.unrestrictLink(`https://rapidgator.net/file/conc-${i}`)));
|
||||
|
||||
expect(results.every((r) => Boolean((r as { directUrl?: string }).directUrl))).toBe(true);
|
||||
expect(accountsSeen.size).toBe(4);
|
||||
expect(maxActive).toBe(4);
|
||||
}, 15000);
|
||||
|
||||
it("verteilt Ueberzahl-Umwandlungen (mehr gleichzeitig als Accounts) gleichmaessig statt sie zu stapeln", async () => {
|
||||
const settings = {
|
||||
...defaultSettings(),
|
||||
token: "", bestToken: "", allDebridToken: "",
|
||||
megaLogin: "user1", megaPassword: "pass1",
|
||||
megaCredentials: "user1:pass1\nuser2:pass2",
|
||||
megaDebridPreferApi: false,
|
||||
providerOrder: [] as const, providerPrimary: "megadebrid" as const,
|
||||
providerSecondary: "none" as const, providerTertiary: "none" as const,
|
||||
autoProviderFallback: false
|
||||
};
|
||||
|
||||
globalThis.fetch = (async () => new Response("error", { status: 500 })) as typeof fetch;
|
||||
|
||||
let active = 0;
|
||||
const callsPerAccount = new Map<string, number>();
|
||||
const allInFlight = new Promise<void>((resolve) => {
|
||||
const check = setInterval(() => { if (active >= 8) { clearInterval(check); resolve(); } }, 5);
|
||||
});
|
||||
const megaWeb = vi.fn(async (_link: string, _signal: AbortSignal | undefined, account?: { login: string; password: string }) => {
|
||||
active += 1;
|
||||
if (account) callsPerAccount.set(account.login, (callsPerAccount.get(account.login) ?? 0) + 1);
|
||||
await allInFlight;
|
||||
return { fileName: "ok.rar", directUrl: "https://mega-web.example/ok.rar", fileSize: null, retriesUsed: 0 };
|
||||
});
|
||||
|
||||
const service = new DebridService(settings, { megaWebUnrestrict: megaWeb });
|
||||
await Promise.all(Array.from({ length: 8 }, (_unused, i) => service.unrestrictLink(`https://rapidgator.net/file/ov-${i}`)));
|
||||
|
||||
// 8 gleichzeitige Aufloesungen auf 2 Accounts → 4 je Account (statt 7/1 beim alten belegt/frei-Set).
|
||||
expect(callsPerAccount.get("user1")).toBe(4);
|
||||
expect(callsPerAccount.get("user2")).toBe(4);
|
||||
}, 15000);
|
||||
|
||||
it("rotates to the next Mega-Debrid account when one hits its daily limit (error-based)", async () => {
|
||||
const settings = {
|
||||
...defaultSettings(),
|
||||
|
||||
@ -202,6 +202,68 @@ describe("mega-web-fallback", () => {
|
||||
expect(result).toBeNull();
|
||||
});
|
||||
|
||||
it("serialisiert gleichzeitige Umwandlungen auf DEMSELBEN Account (kein Doppel-Login)", async () => {
|
||||
let loginCount = 0;
|
||||
globalThis.fetch = vi.fn(async (url: string | URL | Request) => {
|
||||
const u = String(url);
|
||||
if (u.includes("form=login")) {
|
||||
loginCount += 1;
|
||||
await new Promise((r) => setTimeout(r, 15));
|
||||
const headers = new Headers();
|
||||
headers.append("set-cookie", "session=c; path=/");
|
||||
return new Response("", { headers, status: 200 });
|
||||
}
|
||||
if (u.includes("page=debrideur")) return new Response('<form id="debridForm"></form>', { status: 200 });
|
||||
if (u.includes("form=debrid")) return new Response(`<div class="acp-box"><h3>Link: https://mega.debrid/l</h3><a href="javascript:processDebrid(1,'code',0)">d</a></div>`, { status: 200 });
|
||||
if (u.includes("ajax=debrid")) return new Response(JSON.stringify({ link: "https://mega.direct/ok" }), { status: 200 });
|
||||
return new Response("Not found", { status: 404 });
|
||||
}) as unknown as typeof fetch;
|
||||
|
||||
const fallback = new MegaWebFallback(() => ({ login: "same", password: "pw" }));
|
||||
const [r1, r2] = await Promise.all([
|
||||
fallback.unrestrict("https://mega.debrid/a", undefined, { login: "same", password: "pw" }),
|
||||
fallback.unrestrict("https://mega.debrid/b", undefined, { login: "same", password: "pw" })
|
||||
]);
|
||||
expect(r1?.directUrl).toBe("https://mega.direct/ok");
|
||||
expect(r2?.directUrl).toBe("https://mega.direct/ok");
|
||||
// Serialisiert auf demselben Account → der zweite nutzt die gecachte Session, kein zweiter Login.
|
||||
expect(loginCount).toBe(1);
|
||||
});
|
||||
|
||||
it("wandelt auf VERSCHIEDENEN Accounts parallel um (Logins laufen gleichzeitig)", async () => {
|
||||
let activeLogins = 0;
|
||||
let maxActiveLogins = 0;
|
||||
const bothInFlight = new Promise<void>((resolve) => {
|
||||
const check = setInterval(() => { if (activeLogins >= 2) { clearInterval(check); resolve(); } }, 5);
|
||||
});
|
||||
globalThis.fetch = vi.fn(async (url: string | URL | Request) => {
|
||||
const u = String(url);
|
||||
if (u.includes("form=login")) {
|
||||
activeLogins += 1;
|
||||
maxActiveLogins = Math.max(maxActiveLogins, activeLogins);
|
||||
await bothInFlight;
|
||||
activeLogins -= 1;
|
||||
const headers = new Headers();
|
||||
headers.append("set-cookie", "session=c; path=/");
|
||||
return new Response("", { headers, status: 200 });
|
||||
}
|
||||
if (u.includes("page=debrideur")) return new Response('<form id="debridForm"></form>', { status: 200 });
|
||||
if (u.includes("form=debrid")) return new Response(`<div class="acp-box"><h3>Link: https://mega.debrid/l</h3><a href="javascript:processDebrid(1,'code',0)">d</a></div>`, { status: 200 });
|
||||
if (u.includes("ajax=debrid")) return new Response(JSON.stringify({ link: "https://mega.direct/ok" }), { status: 200 });
|
||||
return new Response("Not found", { status: 404 });
|
||||
}) as unknown as typeof fetch;
|
||||
|
||||
const fallback = new MegaWebFallback(() => ({ login: "d", password: "p" }));
|
||||
const [r1, r2] = await Promise.all([
|
||||
fallback.unrestrict("https://mega.debrid/a", undefined, { login: "acc1", password: "p" }),
|
||||
fallback.unrestrict("https://mega.debrid/b", undefined, { login: "acc2", password: "p" })
|
||||
]);
|
||||
expect(r1?.directUrl).toBe("https://mega.direct/ok");
|
||||
expect(r2?.directUrl).toBe("https://mega.direct/ok");
|
||||
// Verschiedene Accounts → beide Logins gleichzeitig in-flight (sonst haengt es am bothInFlight-Barrier).
|
||||
expect(maxActiveLogins).toBe(2);
|
||||
}, 10000);
|
||||
|
||||
it("aborts pending Mega-Web polling when signal is cancelled", async () => {
|
||||
globalThis.fetch = vi.fn((url: string | URL | Request, init?: RequestInit): Promise<Response> => {
|
||||
const urlStr = String(url);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user