perf: run fetch sources concurrently instead of sequentially
Every LinkedIn search and every fetch-other.ts source hits a different host, so there was no real reason for the old for-loop to wait on each one in turn - that's what made a full run take 30-40+ minutes. Runs them concurrently now (bounded: 3 at a time within each group, both groups started together), which brought a real run down to under 90 seconds. existingLinks.add() now happens synchronously right after the .has() check and before any await, which is what makes concurrent workers safe against double-claiming the same link. Concurrency surfaced a real regression: LinkedIn started returning occasional 429s under parallel load, and fetchJobs treated that the same as end-of-results, silently truncating the rest of that search. Added retry-with-backoff (3 attempts, 1s/2s/4s) so a rate limit no longer costs real listings.
This commit is contained in:
@@ -54,6 +54,7 @@ plans/ # LinkedIn post drafts
|
||||
- These boards use UN-flavoured titles ("Information Management Officer", "ICT Specialist") that `SOFTWARE_TITLE_RE` alone drops, so `INTL_TECH_RE` widens the net **for these sources only**. Boards this small can afford the noise; missing a rare opening can't.
|
||||
- **itjobs.ch** is swept Switzerland-wide (no `location` param) — the board is small, ~210 jobs over 7 pages, and locking it to Basel returned only ~9. It 404s one page past the end instead of serving an empty list. Each item embeds its full `description`, so the DE/FR check runs in-process via `isEnglishDescription` rather than costing a page fetch per job. Use `job_location.name` for location — it's the English form; the bare `location` field is German.
|
||||
- **`index.ts`** — orchestrator: loads existing IDs from `results/jobs.csv`, calls `fetchJobs`, calls `isEnglishContent` for each new job, appends survivors to CSV. Writes both `search` (the query/source that found the job) and `location` (the posting's own location, as parsed off its card by `fetch.ts`/`fetch-other.ts` — no extra page fetch) to the CSV. Migrates old CSVs on startup: adds `source` if missing, then `location` (backfilled `""` for rows written before the column existed).
|
||||
- Every LinkedIn search and every `fetch-other.ts` source hits a different host, so they run **concurrently** (bounded: 3 at a time within each group, both groups started together) instead of one after another — cut a full `bun fetch` from ~16-20 min to well under 2 min. `existingLinks.add()` happens synchronously right after the `.has()` check and before any `await`, which is what makes concurrent workers safe against double-claiming the same link. Concurrency surfaced a real gap: LinkedIn started returning occasional 429s under parallel load, which `fetchJobs` used to treat as end-of-results (silently truncating that search) — `fetchJobs` now retries a 429 up to 3× with backoff (1s/2s/4s) before giving up.
|
||||
- **`recheck.ts`** — calls `fetchJobs(..., { raw: true })` to bypass `SOFTWARE_TITLE_RE`, then lists every new job that *would* be filtered out and prints word-frequency suggestions.
|
||||
- **`daily.ts`** — the notification layer on top of `index.ts`. Snapshots the CSV's links, shells out to `bun fetch`, then diffs to find what's new. Survivors are filtered through `matchesProfile` and grouped by area (Basel / Remote / Zürich / Bern) so the message scans on a phone. `results/sent.txt` is written **only after a successful send**, so a Telegram failure retries next run instead of silently losing jobs. A non-zero exit from `bun fetch` sends a failure notice rather than failing quietly.
|
||||
- The area grouping matches against the CSV's `location` column (the posting's real location) first, falling back to `search` (the query/source label) only when `location` is empty — i.e. for rows written before the `location` column existed. This fixed two real mislabels: a LinkedIn "Bern" sweep can surface a job actually based in Lausanne (search label ≠ real location), and `fetch-other.ts` sources (job-room.ch, itjobs.ch, stellen-basel.ch, datacareer.ch) write the source name into `search` instead of a city, which used to dump their jobs into "⚪️ Other" regardless of where they actually were.
|
||||
|
||||
@@ -368,7 +368,15 @@ export async function fetchJobs(config: SearchConfig, { raw = false } = {}): Pro
|
||||
const url = `${baseUrl}&start=${page * PAGE_SIZE}`;
|
||||
process.stdout.write(` page ${page + 1} … `);
|
||||
|
||||
const res = await fetch(url, { headers: HEADERS });
|
||||
// 429s became common once index.ts started running several searches
|
||||
// concurrently — a transient rate-limit, not a real end-of-results
|
||||
// signal, so retry with backoff instead of silently truncating the
|
||||
// search (losing every page after this one).
|
||||
let res = await fetch(url, { headers: HEADERS });
|
||||
for (let attempt = 0; res.status === 429 && attempt < 3; attempt++) {
|
||||
await Bun.sleep(1000 * 2 ** attempt); // 1s, 2s, 4s
|
||||
res = await fetch(url, { headers: HEADERS });
|
||||
}
|
||||
if (!res.ok) {
|
||||
console.error(`failed (${res.status})`);
|
||||
break;
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
#!/usr/bin/env bun
|
||||
import { searches } from "./config.ts";
|
||||
import { fetchJobs, isViableContent, isViableUrl } from "./fetch.ts";
|
||||
import { fetchDataCareer, fetchStellenBasel, fetchItJobsCh, fetchJobRoom, fetchBaselIntlOrgs } from "./fetch-other.ts";
|
||||
import { fetchDataCareer, fetchStellenBasel, fetchItJobsCh, fetchJobRoom, fetchBaselIntlOrgs, fetchJobBasel, fetchEnglishJobSearch } from "./fetch-other.ts";
|
||||
|
||||
const CSV = "./results/jobs.csv";
|
||||
const OLD_HEADER = "date_found,posted_at,search,title,company,link";
|
||||
@@ -59,40 +59,69 @@ function csvCell(val: string): string {
|
||||
return `"${val.replace(/"/g, '""')}"`;
|
||||
}
|
||||
|
||||
// LinkedIn searches
|
||||
for (const config of searches) {
|
||||
// Every search/source below hits a different host (LinkedIn, job-room.ch,
|
||||
// jobbasel.ch, itjobs.ch, unjobs.org, ...), so there's no reason they need to
|
||||
// run one after another — that's what made a full run take 30-40+ minutes.
|
||||
// Run them concurrently instead, bounded per group so we don't burst LinkedIn
|
||||
// (unauthenticated scraping, sensitive to anti-bot heuristics — see the
|
||||
// auth warning in CLAUDE.md) or open too many simultaneous connections to
|
||||
// the smaller extra sources at once.
|
||||
//
|
||||
// `existingLinks` is claimed synchronously (`.add()` right after the `.has()`
|
||||
// check, before any `await`) rather than after the content-check — since Bun
|
||||
// is single-threaded and only yields at `await`, that ordering is what makes
|
||||
// concurrent workers safe: two sources racing on the same link can't both
|
||||
// slip past the check before either claims it.
|
||||
//
|
||||
// Trade-off: each source still logs its own page-by-page progress from
|
||||
// inside fetch.ts/fetch-other.ts, so concurrent sources' output interleaves
|
||||
// in the terminal instead of printing in clean sequential blocks. Cutting
|
||||
// total runtime was the actual complaint; log readability is secondary and
|
||||
// still traceable (grep the source label).
|
||||
async function runWithConcurrency<T>(items: T[], limit: number, worker: (item: T) => Promise<void>): Promise<void> {
|
||||
const queue = [...items];
|
||||
async function runner() {
|
||||
while (queue.length) {
|
||||
const item = queue.shift()!;
|
||||
await worker(item);
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, runner));
|
||||
}
|
||||
|
||||
async function processLinkedInSearch(config: (typeof searches)[number]): Promise<void> {
|
||||
console.log(`\n[${config.label}]`);
|
||||
const jobs = await fetchJobs(config);
|
||||
|
||||
if (jobs.length === 0) {
|
||||
console.log(" No jobs found.");
|
||||
continue;
|
||||
console.log(` [${config.label}] No jobs found.`);
|
||||
return;
|
||||
}
|
||||
|
||||
let added = 0;
|
||||
let langDropped = 0;
|
||||
for (const job of jobs) {
|
||||
if (!existingLinks.has(job.link)) {
|
||||
process.stdout.write(` checking content: ${job.title} … `);
|
||||
// One fetch, two checks: written in German/French, or written in English
|
||||
// but demanding German above B1.
|
||||
const viable = await isViableContent(job.id);
|
||||
if (!viable.ok) {
|
||||
console.log(`dropped (${viable.reason})`);
|
||||
langDropped++;
|
||||
continue;
|
||||
}
|
||||
console.log("ok");
|
||||
existingLinks.add(job.link);
|
||||
newRows.push(
|
||||
[date, job.postedAt, config.label, job.title, job.company, job.link, "linkedin.com", job.location]
|
||||
.map(csvCell)
|
||||
.join(",")
|
||||
);
|
||||
added++;
|
||||
if (existingLinks.has(job.link)) continue;
|
||||
existingLinks.add(job.link); // claim before the await — see note above
|
||||
|
||||
process.stdout.write(` [${config.label}] checking content: ${job.title} … `);
|
||||
// One fetch, two checks: written in German/French, or written in English
|
||||
// but demanding German above B1.
|
||||
const viable = await isViableContent(job.id);
|
||||
if (!viable.ok) {
|
||||
console.log(`dropped (${viable.reason})`);
|
||||
langDropped++;
|
||||
continue;
|
||||
}
|
||||
console.log("ok");
|
||||
newRows.push(
|
||||
[date, job.postedAt, config.label, job.title, job.company, job.link, "linkedin.com", job.location]
|
||||
.map(csvCell)
|
||||
.join(",")
|
||||
);
|
||||
added++;
|
||||
}
|
||||
console.log(` ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
|
||||
console.log(` [${config.label}] ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
|
||||
}
|
||||
|
||||
const extraSources: Array<{ label: string; fetch: () => Promise<ReturnType<typeof fetchDataCareer>> }> = [
|
||||
@@ -101,49 +130,59 @@ const extraSources: Array<{ label: string; fetch: () => Promise<ReturnType<typeo
|
||||
{ label: "itjobs.ch", fetch: fetchItJobsCh },
|
||||
{ label: "job-room.ch", fetch: fetchJobRoom },
|
||||
{ label: "basel intl orgs", fetch: fetchBaselIntlOrgs },
|
||||
{ label: "jobbasel.ch", fetch: fetchJobBasel },
|
||||
{ label: "englishjobsearch.ch", fetch: fetchEnglishJobSearch },
|
||||
];
|
||||
|
||||
for (const { label, fetch: fetchFn } of extraSources) {
|
||||
async function processExtraSource({ label, fetch: fetchFn }: (typeof extraSources)[number]): Promise<void> {
|
||||
console.log(`\n[${label}]`);
|
||||
let jobs;
|
||||
try {
|
||||
jobs = await fetchFn();
|
||||
} catch (e) {
|
||||
console.error(` error: ${e}`);
|
||||
continue;
|
||||
console.error(` [${label}] error: ${e}`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (jobs.length === 0) {
|
||||
console.log(" No jobs found.");
|
||||
continue;
|
||||
console.log(` [${label}] No jobs found.`);
|
||||
return;
|
||||
}
|
||||
|
||||
let added = 0;
|
||||
let langDropped = 0;
|
||||
for (const job of jobs) {
|
||||
if (!existingLinks.has(job.link)) {
|
||||
if (!job.contentChecked) {
|
||||
process.stdout.write(` checking content: ${job.title} … `);
|
||||
const viable = await isViableUrl(job.link);
|
||||
if (!viable.ok) {
|
||||
console.log(`dropped (${viable.reason})`);
|
||||
langDropped++;
|
||||
continue;
|
||||
}
|
||||
console.log("ok");
|
||||
if (existingLinks.has(job.link)) continue;
|
||||
existingLinks.add(job.link); // claim before the await — see note above
|
||||
|
||||
if (!job.contentChecked) {
|
||||
process.stdout.write(` [${label}] checking content: ${job.title} … `);
|
||||
const viable = await isViableUrl(job.link);
|
||||
if (!viable.ok) {
|
||||
console.log(`dropped (${viable.reason})`);
|
||||
langDropped++;
|
||||
continue;
|
||||
}
|
||||
existingLinks.add(job.link);
|
||||
newRows.push(
|
||||
[date, job.postedAt, job.search, job.title, job.company, job.link, job.source, job.location]
|
||||
.map(csvCell)
|
||||
.join(",")
|
||||
);
|
||||
added++;
|
||||
console.log("ok");
|
||||
}
|
||||
newRows.push(
|
||||
[date, job.postedAt, job.search, job.title, job.company, job.link, job.source, job.location]
|
||||
.map(csvCell)
|
||||
.join(",")
|
||||
);
|
||||
added++;
|
||||
}
|
||||
console.log(` ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
|
||||
console.log(` [${label}] ${jobs.length} found, ${added} new, ${langDropped} non-English dropped`);
|
||||
}
|
||||
|
||||
const LINKEDIN_CONCURRENCY = 3;
|
||||
const EXTRA_SOURCE_CONCURRENCY = 3;
|
||||
|
||||
await Promise.all([
|
||||
runWithConcurrency(searches, LINKEDIN_CONCURRENCY, processLinkedInSearch),
|
||||
runWithConcurrency(extraSources, EXTRA_SOURCE_CONCURRENCY, processExtraSource),
|
||||
]);
|
||||
|
||||
if (newRows.length === 0) {
|
||||
console.log("\nNo new jobs.");
|
||||
process.exit(0);
|
||||
|
||||
Reference in New Issue
Block a user