openvibers/openvibe.tours Commits
Places, routes and stays run as soon as cities land (production had 100 cities and no places) (#2)
Changes
@@@ -34,11 +34,22 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = consconst running = new Set();const timers = [];++ // A run produced something when any of its numbers is above zero (the error list and the catalogue aside).++ const producedOf = (counts) => Object.entries(counts || {}).some(([k, v]) => k !== 'errors' && k !== 'catalog' && typeof v === 'number' && v > 0);++ // The jobs that work through the cities: when the cities job adds some, they run at once instead of waiting a week++ // (2026-11-01: places, routes and stays ran 11 minutes before the first cities landed, found none, and counted as++ // fresh for seven days).++ const AFTER_CITIES = ['places', 'routes', 'stays'];++ let scheduled = false; // start() ran: only the scheduler follows the cities job (a test's runOnce does not)async function isFresh(name) {const rows = await store.latestJobRuns();const row = rows.find((r) => r.job === name);if (!row || !['ok', 'partial'].includes(row.status) || !row.finished_at) return false;++ // A run that found nothing to work on is not fresh: the next boot or the cities job runs it again.++ let counts = row.counts;++ if (typeof counts === 'string') { try { counts = JSON.parse(counts); } catch { counts = null; } }++ if (!producedOf(counts)) return false;return Date.now() - new Date(row.finished_at).getTime() < jobs[name].maxAgeMs;}@@@ -52,10 +63,11 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = constry {const counts = await job.run(opts);const errors = Array.isArray(counts && counts.errors) ? counts.errors : [];-- const produced = Object.entries(counts || {}).some(([k, v]) => k !== 'errors' && k !== 'catalog' && typeof v === 'number' && v > 0);++ const produced = producedOf(counts);const status = errors.length ? (produced ? 'partial' : 'failed') : 'ok';await store.recordJobFinish(id, { status, counts, error: errors.length ? errors.slice(0, 10).join('; ') : null });log.log(`[Tours ${name}] ${status} in ${Math.round((Date.now() - started) / 1000)}s ${JSON.stringify(counts)}`);++ if (name === 'cities' && produced && scheduled && !opts.noFollow) followCities();return { status, counts };} catch (err) {await store.recordJobFinish(id, { status: 'failed', counts: {}, error: (err && err.message) || String(err) });@@@ -66,14 +78,27 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons}}++ /** After cities land: each job that works through them runs now, unless it already has fresh work. Never awaited. */++ function followCities() {++ if (config.drill) return;++ const kick = setTimeout(async () => {++ for (const dep of AFTER_CITIES) {++ try { if (!await isFresh(dep)) await runOnce(dep); } catch (err) { log.warn(`[Tours ${dep}] ${err && err.message}`); }++ }++ }, 1000);++ kick.unref();++ timers.push(kick);++ }++async function runAll() {const out = {};-- for (const name of ['cities', 'places', 'routes', 'live', 'stays']) out[name] = await runOnce(name);++ for (const name of ['cities', 'places', 'routes', 'live', 'stays']) out[name] = await runOnce(name, { noFollow: true });return out;}function start() {if (config.drill || !config.jobs.enabled) return;++ scheduled = true;const names = Object.keys(jobs);names.forEach((name, i) => {const kick = setTimeout(async () => {@@@ -99,6 +99,16 @@ const { boot, check, done } = require('./helpers/boot');assert.match(stay.url, /^https:\/\/openvibe\.rent\/listings\/rnt_/);});++ await check('a run that found nothing to work on is not fresh: the next boot runs it again', async () => {++ // 2026-11-01: places, routes and stays ran before the first cities landed, found none, and counted as fresh++ // for a week. Now an empty run is not fresh, and the scheduler runs them as soon as cities are added.++ const db = t.ctx.s.db;++ await db.exec(`INSERT INTO tours_job_runs (job, started_at, finished_at, status, counts) VALUES ('routes', now(), now(), 'ok', $1::jsonb)`, [JSON.stringify({ cities: 0, routes: 0, errors: [] })]);++ assert.strictEqual(await t.ctx.jobs.isFresh('routes'), false, 'an empty run is not fresh');++ await db.exec(`INSERT INTO tours_job_runs (job, started_at, finished_at, status, counts) VALUES ('routes', now() + interval '1 second', now() + interval '1 second', 'ok', $1::jsonb)`, [JSON.stringify({ cities: 3, routes: 2, errors: [] })]);++ assert.strictEqual(await t.ctx.jobs.isFresh('routes'), true, 'a run that worked is');++ });++await check('every job left a receipt with its counts', async () => {const rows = await t.ctx.s.db.many("SELECT DISTINCT ON (job) job, status, counts FROM tours_job_runs ORDER BY job, started_at DESC");assert.deepStrictEqual(rows.map((r) => r.job).sort(), ['cities', 'live', 'places', 'routes', 'stays']);
@@ -34,11 +34,22 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
@@ -34,11 +34,22 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
const running = new Set();
const running = new Set();
const timers = [];
const timers = [];
+ // A run produced something when any of its numbers is above zero (the error list and the catalogue aside).
+ const producedOf = (counts) => Object.entries(counts || {}).some(([k, v]) => k !== 'errors' && k !== 'catalog' && typeof v === 'number' && v > 0);
+ // The jobs that work through the cities: when the cities job adds some, they run at once instead of waiting a week
+ // (2026-11-01: places, routes and stays ran 11 minutes before the first cities landed, found none, and counted as
+ // fresh for seven days).
+ const AFTER_CITIES = ['places', 'routes', 'stays'];
+ let scheduled = false; // start() ran: only the scheduler follows the cities job (a test's runOnce does not)
async function isFresh(name) {
async function isFresh(name) {
const rows = await store.latestJobRuns();
const rows = await store.latestJobRuns();
const row = rows.find((r) => r.job === name);
const row = rows.find((r) => r.job === name);
if (!row || !['ok', 'partial'].includes(row.status) || !row.finished_at) return false;
if (!row || !['ok', 'partial'].includes(row.status) || !row.finished_at) return false;
+ // A run that found nothing to work on is not fresh: the next boot or the cities job runs it again.
+ let counts = row.counts;
+ if (typeof counts === 'string') { try { counts = JSON.parse(counts); } catch { counts = null; } }
+ if (!producedOf(counts)) return false;
return Date.now() - new Date(row.finished_at).getTime() < jobs[name].maxAgeMs;
return Date.now() - new Date(row.finished_at).getTime() < jobs[name].maxAgeMs;
}
}
@@ -52,10 +63,11 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
@@ -52,10 +63,11 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
try {
try {
const counts = await job.run(opts);
const counts = await job.run(opts);
const errors = Array.isArray(counts && counts.errors) ? counts.errors : [];
const errors = Array.isArray(counts && counts.errors) ? counts.errors : [];
- const produced = Object.entries(counts || {}).some(([k, v]) => k !== 'errors' && k !== 'catalog' && typeof v === 'number' && v > 0);
+ const produced = producedOf(counts);
const status = errors.length ? (produced ? 'partial' : 'failed') : 'ok';
const status = errors.length ? (produced ? 'partial' : 'failed') : 'ok';
await store.recordJobFinish(id, { status, counts, error: errors.length ? errors.slice(0, 10).join('; ') : null });
await store.recordJobFinish(id, { status, counts, error: errors.length ? errors.slice(0, 10).join('; ') : null });
log.log(`[Tours ${name}] ${status} in ${Math.round((Date.now() - started) / 1000)}s ${JSON.stringify(counts)}`);
log.log(`[Tours ${name}] ${status} in ${Math.round((Date.now() - started) / 1000)}s ${JSON.stringify(counts)}`);
+ if (name === 'cities' && produced && scheduled && !opts.noFollow) followCities();
return { status, counts };
return { status, counts };
} catch (err) {
} catch (err) {
await store.recordJobFinish(id, { status: 'failed', counts: {}, error: (err && err.message) || String(err) });
await store.recordJobFinish(id, { status: 'failed', counts: {}, error: (err && err.message) || String(err) });
@@ -66,14 +78,27 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
@@ -66,14 +78,27 @@ function createJobs({ config, s, store, fetchImpl = globalThis.fetch, log = cons
}
}
}
}
+ /** After cities land: each job that works through them runs now, unless it already has fresh work. Never awaited. */
+ function followCities() {
+ if (config.drill) return;
+ const kick = setTimeout(async () => {
+ for (const dep of AFTER_CITIES) {
+ try { if (!await isFresh(dep)) await runOnce(dep); } catch (err) { log.warn(`[Tours ${dep}] ${err && err.message}`); }
+ }
+ }, 1000);
+ kick.unref();
+ timers.push(kick);
+ }
+
async function runAll() {
async function runAll() {
const out = {};
const out = {};
- for (const name of ['cities', 'places', 'routes', 'live', 'stays']) out[name] = await runOnce(name);
+ for (const name of ['cities', 'places', 'routes', 'live', 'stays']) out[name] = await runOnce(name, { noFollow: true });
return out;
return out;
}
}
function start() {
function start() {
if (config.drill || !config.jobs.enabled) return;
if (config.drill || !config.jobs.enabled) return;
+ scheduled = true;
const names = Object.keys(jobs);
const names = Object.keys(jobs);
names.forEach((name, i) => {
names.forEach((name, i) => {
const kick = setTimeout(async () => {
const kick = setTimeout(async () => {
@@ -99,6 +99,16 @@ const { boot, check, done } = require('./helpers/boot');
@@ -99,6 +99,16 @@ const { boot, check, done } = require('./helpers/boot');
assert.match(stay.url, /^https:\/\/openvibe\.rent\/listings\/rnt_/);
assert.match(stay.url, /^https:\/\/openvibe\.rent\/listings\/rnt_/);
});
});
+ await check('a run that found nothing to work on is not fresh: the next boot runs it again', async () => {
+ // 2026-11-01: places, routes and stays ran before the first cities landed, found none, and counted as fresh
+ // for a week. Now an empty run is not fresh, and the scheduler runs them as soon as cities are added.
+ const db = t.ctx.s.db;
+ await db.exec(`INSERT INTO tours_job_runs (job, started_at, finished_at, status, counts) VALUES ('routes', now(), now(), 'ok', $1::jsonb)`, [JSON.stringify({ cities: 0, routes: 0, errors: [] })]);
+ assert.strictEqual(await t.ctx.jobs.isFresh('routes'), false, 'an empty run is not fresh');
+ await db.exec(`INSERT INTO tours_job_runs (job, started_at, finished_at, status, counts) VALUES ('routes', now() + interval '1 second', now() + interval '1 second', 'ok', $1::jsonb)`, [JSON.stringify({ cities: 3, routes: 2, errors: [] })]);
+ assert.strictEqual(await t.ctx.jobs.isFresh('routes'), true, 'a run that worked is');
+ });
+
await check('every job left a receipt with its counts', async () => {
await check('every job left a receipt with its counts', async () => {
const rows = await t.ctx.s.db.many("SELECT DISTINCT ON (job) job, status, counts FROM tours_job_runs ORDER BY job, started_at DESC");
const rows = await t.ctx.s.db.many("SELECT DISTINCT ON (job) job, status, counts FROM tours_job_runs ORDER BY job, started_at DESC");
assert.deepStrictEqual(rows.map((r) => r.job).sort(), ['cities', 'live', 'places', 'routes', 'stays']);
assert.deepStrictEqual(rows.map((r) => r.job).sort(), ['cities', 'live', 'places', 'routes', 'stays']);