openvibers/openvibe.trade Commits

The sitemaps' instrument list is built once for concurrent readers, 8 at a time, and reused for 10 minutes in production (#206)

OP OpenVibers committed 8 days ago · 80d673e3d1 · 4 files, +154 −14

Changes

ddiff --git a/.env.example b/.env.example
iindex b4be310..1c8c9e3 100644
---- a/.env.example
++++ b/.env.example
@@@ -61,6 +61,9 @@ OV_SOURCES_INTERNAL_URL=http://127.0.0.1:4720
#TRADE_FRESHNESS_INTERVAL_MS=60000
# Indexability: minimum words of reviewed context before an instrument page may be indexed
#TRADE_GATE_MIN_WORDS=60
++# How long the sitemaps' instrument list (every index candidate's gate decision) is reused, in ms (production
++# default 600000; 0 rebuilds on every read, the default elsewhere). Concurrent reads share one build either way.
++#TRADE_SITEMAP_TTL_MS=600000
# OpenVibe.Events — the outbox relay runs only when EVENTS_URL and OV_OAUTH_CLIENT_SECRET are set
#EVENTS_URL=http://127.0.0.1:4300
ddiff --git a/server/config.js b/server/config.js
iindex fb6e62d..63a5f56 100644
---- a/server/config.js
++++ b/server/config.js
@@@ -27,6 +27,10 @@ function load(env = process.env) {
isProduction,
// Public origin: canonical URLs, feeds, sitemaps and JSON-LD are built from it.
baseUrl,
++ // The sitemaps' and /llms-full.txt's instrument list (every index candidate's gate decision) is built at most
++ // once in this window and shared (one build serves concurrent readers either way). Each crawler fetch of
++ // /sitemap.xml weighed every candidate: a few queries each, all at once. 0 = no reuse.
++ sitemapTtlMs: int(env.TRADE_SITEMAP_TTL_MS, isProduction ? 10 * 60 * 1000 : 0),
trustProxy: env.TRUST_PROXY != null ? Number(env.TRUST_PROXY) : 2,
// Per-actor limits (server/http/actor-limits.js, roadmap WS-R task 4): the API reads one caller
// may make per minute and per hour. Writes set their own numbers there.
ddiff --git a/server/http/discovery.js b/server/http/discovery.js
iindex 7d7bda4..583d8ec 100644
---- a/server/http/discovery.js
++++ b/server/http/discovery.js
@@@ -15,8 +15,8 @@
* GET /feed.xml | /atom.xml | /feed.json recent public documents across instruments
* GET /i/:symbol/documents.xml|.atom|.json one instrument's documents
*
-- * Built from the database on every request, never from the viewer. Watchlists, alert rules,
-- * deliveries and the editor never appear in any of them.
++ * Built from the database, never from the viewer (the instrument list at most once per config.sitemapTtlMs, shared by
++ * concurrent readers). Watchlists, alert rules, deliveries and the editor never appear in any of them.
*/
const express = require('express');
const seo = require('openvibe-publishing/seo');
@@@ -25,6 +25,28 @@ const cache = require('openvibe-shared/cache-policy');
const { define } = require('./routes');
const { DISCLAIMER, SITE_NAME, SITE_SUMMARY } = require('../render/layout');
++// Instruments whose gate decision a list build reads at once (the pool has a handful of connections for everyone).
++const INSTRUMENTS_PARALLEL = 8;
++
++/** fn over items, `limit` at a time; results in the items' order. */
++async function mapLimit(items, limit, fn) {
++ const out = new Array(items.length);
++ let next = 0;
++ const worker = async () => { while (next < items.length) { const i = next++; out[i] = await fn(items[i]); } };
++ await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
++ return out;
++}
++
++/** One build at a time, shared by the readers that ask while it runs, reused for ttlMs (0: the next read rebuilds). */
++function shared(build, ttlMs, now) {
++ let value = null; let at = 0; let running = null;
++ return async () => {
++ if (value && now() - at < ttlMs) return value;
++ if (!running) running = build().then((v) => { value = v; at = now(); return v; }).finally(() => { running = null; });
++ return await running;
++ };
++}
++
/**
* The public page catalogue: /sitemaps/pages.xml lists the entries marked for the sitemap and
* /llms-full.txt describes every entry, so the two artifacts cannot drift apart. Entity pages are
@@@ -52,6 +74,13 @@ function createDiscovery(ctx) {
const xml = (res, body, type = 'application/xml') => res.type(type).set('Cache-Control', cache.htmlHeaders({ maxAge: 300 })).send(body);
const listing = (url) => seo.evaluate({ state: 'published', visibility: 'public', canonicalUrl: url, wordCount: 0 }, { policy: { minWords: 0 } });
++ // Every index candidate and its gate decision, for the sitemaps and /llms-full.txt: built INSTRUMENTS_PARALLEL at a
++ // time, once for concurrent readers, reused for config.sitemapTtlMs. All at once, each crawler fetch of
++ // /sitemap.xml weighed every candidate (a few queries each), and the SEC list keeps growing.
++ const instrumentEntries = shared(async () => mapLimit(await instruments.indexCandidates(), INSTRUMENTS_PARALLEL, async (i) => ({
++ instrument: i, decision: await indexing.decide(i),
++ })), Number(config.sitemapTtlMs) || 0, store.now);
++
function feedItem(d, instrument) {
const url = urls.documentAnchor(instrument, d);
const when = (ms) => (ms == null ? 'no date stated' : new Date(ms).toISOString());
@@@ -126,12 +155,11 @@ function createDiscovery(ctx) {
// page — the same gate the sitemap uses. The summary is the reviewed context, clipped; the
// source's own full text is never reproduced (llms.txt states the rule).
define(router, 'get', '/llms-full.txt', 'llmsFullTxt', async (_req, res) => {
-- const active = await instruments.indexCandidates();
-- const decisions = await Promise.all(active.map(async (i) => await indexing.decide(i)));
-- const instrumentsPages = await Promise.all(active.filter((_i, n) => decisions[n].indexable).map(async (i) => {
++ const entries = (await instrumentEntries()).filter(({ decision }) => decision.indexable);
++ const instrumentsPages = await mapLimit(entries, INSTRUMENTS_PARALLEL, async ({ instrument: i }) => {
const rev = await context.published(i);
return { title: `${i.symbol} — ${i.name}`, url: urls.instrument(i), text: rev ? String(rev.text || '').replace(/\s+/g, ' ').trim().slice(0, 300) : '' };
-- }));
++ });
const pages = [...PUBLIC_PAGES, ...legalPages()].map((p) => ({ url: abs(p.path), title: p.title, text: p.text }));
res.type('text/plain').set('Cache-Control', cache.htmlHeaders({ maxAge: 3600 })).send(sharedSeo.llmsFull({
site: SITE_NAME, summary: SITE_SUMMARY, base: config.baseUrl, maxBytes: 512 * 1024,
@@@ -149,10 +177,8 @@ function createDiscovery(ctx) {
const pageLastmod = (instrument, changes) => Math.max(instrument.updated_at || 0, instrument.context_published_at || 0, changes.get(instrument.id) || 0);
define(router, 'get', '/sitemap.xml', 'sitemapIndex', async (_req, res) => {
-- const active = await instruments.indexCandidates();
-- const decisions = await Promise.all(active.map(async (i) => await indexing.decide(i)));
-- const changes = await pageChanges();
-- const times = active.filter((i, n) => decisions[n].indexable).map((i) => pageLastmod(i, changes)).filter(Boolean);
++ const [entries, changes] = await Promise.all([instrumentEntries(), pageChanges()]);
++ const times = entries.filter(({ decision }) => decision.indexable).map(({ instrument }) => pageLastmod(instrument, changes)).filter(Boolean);
const newest = times.length ? new Date(Math.max(...times)).toISOString() : null;
const pagesNewest = (await documents.recent(1))[0];
xml(res, seo.sitemapIndex([
@@@ -171,13 +197,13 @@ function createDiscovery(ctx) {
});
define(router, 'get', '/sitemaps/instruments.xml', 'instrumentSitemap', async (_req, res) => {
-- const changes = await pageChanges();
// Without a published context an instrument page is noindex (indexing.js), so only the candidates are weighed.
-- const entries = (await Promise.all((await instruments.indexCandidates()).map(async (i) => ({
++ const [weighed, changes] = await Promise.all([instrumentEntries(), pageChanges()]);
++ const entries = weighed.map(({ instrument: i, decision }) => ({
loc: urls.instrument(i),
...(pageLastmod(i, changes) ? { lastmod: new Date(pageLastmod(i, changes)).toISOString() } : {}),
-- decision: await indexing.decide(i),
-- }))));
++ decision,
++ }));
xml(res, seo.sitemap(entries).files[0]);
});
ddiff --git a/test/sitemap-build.test.js b/test/sitemap-build.test.js
nnew file mode 100644
iindex 0000000..6b94ca0
---- /dev/null
++++ b/test/sitemap-build.test.js
@@@ -0,0 +1,107 @@
++'use strict';
++/**
++ * The sitemaps' instrument list (server/http/discovery.js): one build serves every request that arrives while it runs,
++ * at most 8 gate decisions at once, reused for TRADE_SITEMAP_TTL_MS (production: 10 minutes; 0 rebuilds on every
++ * read). Each crawler fetch of /sitemap.xml weighed every index candidate at once: a few queries each, all on the
++ * pool, and the SEC list keeps growing.
++ */
++const assert = require('assert');
++const { boot, check, done } = require('./helpers/boot');
++
++const WORDS = (n) => Array.from({ length: n }, (_, i) => `word${i}`).join(' ');
++
++/** A person-written context makes the instrument an index candidate (the gate in server/domain/indexing.js). */
++async function publish(t, instrument) {
++ return await t.ctx.context.propose({ kind: 'user', editor: true, subject: t.editor.subject }, instrument, {
++ body: `The source states this figure. ${WORDS(70)}`, publish: '1',
++ });
++}
++
++/**
++ * Count the queries the list build makes: indexCandidates() once per build and the newest-price lookup once per
++ * decision. `holdFirst` (ms) keeps the first candidates query open so concurrent HTTP requests reach the same build.
++ */
++function watch(db, { holdFirst = 0 } = {}) {
++ const seen = { builds: 0, decisions: 0, now: 0, max: 0 };
++ const query = db.query.bind(db);
++ db.query = async (text, values) => {
++ const sql = String(text);
++ if (/FROM trade_instruments i WHERE i\.status = 'active' AND/.test(sql)) {
++ seen.builds++;
++ if (seen.builds === 1 && holdFirst) await new Promise((r) => setTimeout(r, holdFirst));
++ return await query(text, values);
++ }
++ if (/FROM trade_market_observations WHERE instrument_id/.test(sql)) {
++ seen.decisions++; seen.now++; seen.max = Math.max(seen.max, seen.now);
++ try { await new Promise((r) => setImmediate(r)); return await query(text, values); } finally { seen.now--; }
++ }
++ return await query(text, values);
++ };
++ return seen;
++}
++
++(async () => {
++ const t = await boot();
++ try {
++ for (let i = 0; i < 20; i++) await publish(t, await t.instrument({ symbol: `T${i}`, name: `Test ${i}` }));
++ const seen = watch(t.ctx.store.db, { holdFirst: 100 });
++
++ await check('concurrent fetches share one build, at most 8 decisions at a time; without a TTL the next fetch rebuilds', async () => {
++ const [index, map, llms] = await Promise.all([
++ t.get('/sitemap.xml'), t.get('/sitemaps/instruments.xml'), t.get('/llms-full.txt'),
++ ]);
++ assert.deepStrictEqual([index.status, map.status, llms.status], [200, 200, 200]);
++ assert.strictEqual(seen.builds, 1, `three fetches, one build (${seen.builds})`);
++ assert.strictEqual(seen.decisions, 20, `each instrument weighed once (${seen.decisions})`);
++ assert.ok(seen.max >= 2 && seen.max <= 8, `at most 8 decisions at once (saw ${seen.max})`);
++ assert.strictEqual((map.text.match(/<loc>/g) || []).length, 20, 'every indexable instrument listed');
++ assert.ok(map.text.includes('https://openvibe.trade/i/T19'));
++ assert.ok(llms.text.includes('https://openvibe.trade/i/T19'));
++
++ // No TTL (tests, development): the next fetch rebuilds, and the bytes are the same for the same data.
++ const again = await t.get('/sitemaps/instruments.xml');
++ assert.strictEqual(seen.builds, 2, 'no TTL: rebuilt');
++ assert.strictEqual(seen.decisions, 40);
++ assert.strictEqual(again.text, map.text, 'the output is unchanged');
++ });
++ } finally { await t.close(); }
++
++ const c = await boot({ env: { TRADE_SITEMAP_TTL_MS: '600000' } });
++ try {
++ const first = await c.instrument({ symbol: 'ACME', name: 'Acme Corp' });
++ await publish(c, first);
++ const seen = watch(c.ctx.store.db, {});
++
++ await check('with TRADE_SITEMAP_TTL_MS the list is reused until it runs out, and the output is unchanged', async () => {
++ const map1 = await c.get('/sitemaps/instruments.xml');
++ assert.strictEqual(seen.builds, 1);
++ assert.strictEqual(seen.decisions, 1);
++ // The index and /llms-full.txt read the same list: no second build inside the window.
++ assert.strictEqual((await c.get('/sitemap.xml')).status, 200);
++ assert.strictEqual((await c.get('/llms-full.txt')).status, 200);
++ c.clock.advance(9 * 60 * 1000);
++ const map2 = await c.get('/sitemaps/instruments.xml');
++ assert.strictEqual(seen.builds, 1, 'reused inside ten minutes');
++ assert.strictEqual(seen.decisions, 1);
++ assert.strictEqual(map2.text, map1.text);
++
++ // Expired: one rebuild, and the bytes are the same while the data is the same.
++ c.clock.advance(2 * 60 * 1000);
++ const map3 = await c.get('/sitemaps/instruments.xml');
++ assert.strictEqual(seen.builds, 2, 'rebuilt after');
++ assert.strictEqual(seen.decisions, 2);
++ assert.strictEqual(map3.text, map1.text, 'the output is unchanged');
++
++ // A new indexable instrument appears at the next build, not before.
++ const second = await c.instrument({ symbol: 'OTHR', name: 'Other Co' });
++ await publish(c, second);
++ assert.strictEqual((await c.get('/sitemaps/instruments.xml')).text, map1.text, 'not inside the window');
++ c.clock.advance(11 * 60 * 1000);
++ const map4 = await c.get('/sitemaps/instruments.xml');
++ assert.strictEqual(seen.builds, 3);
++ assert.strictEqual((map4.text.match(/<loc>/g) || []).length, 2);
++ });
++ } finally { await c.close(); }
++
++ done();
++})();