tallship/FoundKey
0
0
Fork 0
forked from FoundKeyGang/FoundKey
Code Issues Pull requests Projects Releases Packages Wiki Activity

backend: simplify suspended and dead queries

Browse source 
This should also have better latency due to being a single query.
Furthermore, it's no longer a linear scan, since host is indexed.
Would be cool to simplify it further to a single query for blocks also...
Why exactly are blocks not in the db?
This commit is contained in:
Chloe Kudryavtsev 2022-10-16 09:22:05 -04:00
parent 91a4f38871
commit 21c1e5c06c
1 changed files with 8 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
packages/backend/src/queue/processors/deliver.ts
View file

@ -13,15 +13,13 @@ import { Instance } from '@/models/entities/instance.js';
import { StatusError } from '@/misc/fetch.js'; import { StatusError } from '@/misc/fetch.js';
import { DeliverJobData } from '@/queue/types.js'; import { DeliverJobData } from '@/queue/types.js';
import { LessThan } from 'typeorm'; import { LessThan } from 'typeorm';
import { DAY } from '@/const.ts';
const logger = new Logger('deliver'); const logger = new Logger('deliver');
let latest: string | null = null; let latest: string | null = null;
const suspendedHostsCache = new Cache<Instance[]>(1000 * 60 * 60); const deadThreshold = 30 * DAY;
// dead host list is a linear scan, so cache it longer
const deadHostsCache = new Cache<Instance[]>(1000 * 60 * 60 * 24);
const deadThreshold = 1000 * 60 * 60 * 24 * 30; // 1 month
export default async (job: Bull.Job<DeliverJobData>) => { export default async (job: Bull.Job<DeliverJobData>) => {
const { host } = new URL(job.data.to); const { host } = new URL(job.data.to);
@ -33,29 +31,12 @@ export default async (job: Bull.Job<DeliverJobData>) => {
return 'skip (blocked)'; return 'skip (blocked)';
} }
// isSuspended const isSuspendedOrDead = await Instances.countBy([
let suspendedHosts = suspendedHostsCache.get(null); { host: puny, isSuspended: true },
if (suspendedHosts == null) { { host: puny, lastCommunicatedAt: LessThan(deadTime) },
suspendedHosts = await Instances.findBy({ ]);
isSuspended: true, if (isSuspendedOrDead) {
}); return 'skip (suspended or dead)';
suspendedHostsCache.set(null, suspendedHosts);
}
if (suspendedHosts.some(x => x.host == puny)) {
return 'skip (suspended)';
}
// dead
let deadHosts = deadHostsCache.get(null);
if (deadHosts == null) {
const deadTime = new Date(Date.now() - deadThreshold);
deadHosts = await Instances.findBy({
lastCommunicatedAt: LessThan(deadTime),
});
deadHostsCache.set(null, deadHosts);
}
if (deadHosts.some(x => x.host == puny)) {
return 'skip (dead instance)';
} }
try { try {