42 lines
2.3 KiB
TypeScript
42 lines
2.3 KiB
TypeScript
import {setTimeout as sleep} from 'node:timers/promises';
|
|
import {transaction} from './store.ts';
|
|
|
|
type HostState={nextAt:number;blockedUntil:number};
|
|
export class HostCooldown extends Error{
|
|
constructor(until:number){super(`Запрос отложен: сервер ограничил доступ. Пауза до ${new Date(until).toISOString()}; доступность материала не подтверждена.`);}
|
|
}
|
|
// Persist reservations across the API and worker, including container restarts.
|
|
async function updateHost<T>(host:string,update:(state:HostState,now:number)=>T){
|
|
return transaction(async client=>{
|
|
const key='throttle:'+host;
|
|
await client.query('SELECT pg_advisory_xact_lock(hashtextextended($1,0))',[key]);
|
|
const {rows}=await client.query<{value:string}>('SELECT value FROM records WHERE key=$1',[key]);
|
|
const state:HostState=rows[0]?JSON.parse(rows[0].value):{nextAt:0,blockedUntil:0};
|
|
const result=update(state,Date.now());
|
|
await client.query('INSERT INTO records(key,value) VALUES($1,$2) ON CONFLICT(key) DO UPDATE SET value=excluded.value',[key,JSON.stringify(state)]);
|
|
return result;
|
|
});
|
|
}
|
|
export async function reserveRequest(host:string){
|
|
return updateHost(host,(state,now)=>{
|
|
if(state.blockedUntil>now)throw new HostCooldown(state.blockedUntil);
|
|
const at=Math.max(now,state.nextAt);state.nextAt=at+2000;return at;
|
|
});
|
|
}
|
|
export async function waitForHost(host:string,signal?:AbortSignal){
|
|
const at=await reserveRequest(host);await sleep(Math.max(0,at-Date.now()),undefined,{signal});
|
|
await updateHost(host,(state,now)=>{if(state.blockedUntil>now)throw new HostCooldown(state.blockedUntil);});
|
|
}
|
|
export function retryDelay(value:string|undefined,now:number){
|
|
const milliseconds=value&&/^\d+$/.test(value.trim())?Number(value)*1000:value?Date.parse(value)-now:0;
|
|
// Ignore malformed values; keep a minimum pause even if the server says zero.
|
|
return Math.min(8640000000000000-now,Math.max(60000,Number.isNaN(milliseconds)?0:milliseconds));
|
|
}
|
|
export async function limitHost(host:string,code:number,retryAfter?:string){
|
|
if(![403,429,503].includes(code))return;
|
|
await updateHost(host,(state,now)=>{
|
|
state.blockedUntil=Math.max(state.blockedUntil,now+(code===403?300000:retryDelay(retryAfter,now)));
|
|
state.nextAt=Math.max(state.nextAt,state.blockedUntil);
|
|
});
|
|
}
|