openvibers/openvibe.food Commits

Food: the USDA pull passes the food feed's backlog to its rows (#217)

OP OpenVibers committed 7 days ago · 96a8f03686 · 3 files, +36 −6

A new change cursor starts at 0, and the food category feed holds Open Food Facts' tens of thousands of prior changes. The ingest chassis' 25-page default would take days of six-hourly pulls to reach the 155 USDA rows; the pull now reads 500 a page up to 300 pages and logs when it is still behind.

Changes

ddiff --git a/README.md b/README.md
iindex f9565a6..31cf555 100644
---- a/README.md
++++ b/README.md
@@@ -172,7 +172,9 @@ retail prices for 155 fresh and processed fruits and vegetables (Fruit and Veget
public domain); 16 of Food's items are among them, and an item's page shows ERS's figure for the food's pack beside
BLS's where both price it ([server/food/usda-prices.js](server/food/usda-prices.js)). Food never calls ERS:
OpenVibe.Sources reads the two published CSV tables once a week (source `usda-ers-fruit-vegetable-prices`) and Food
--follows its change feed into `food_usda_prices`. The page gives the data year (2023) and repeats ERS's warning that the
++follows its change feed into `food_usda_prices`. The first pull reads the shared food feed deeply enough (500 items a
++page, up to 300 pages) to pass Open Food Facts' history, so the rows arrive in one pass rather than after days of small
++pages. The page gives the data year (2023) and repeats ERS's warning that the
data should not be used for making inferences about price changes over time; the annual figure is never shown as a
month's price or a trend.
ddiff --git a/server/food/usda-prices.js b/server/food/usda-prices.js
iindex 609b3aa..2baf9b1 100644
---- a/server/food/usda-prices.js
++++ b/server/food/usda-prices.js
@@@ -15,6 +15,11 @@
*
* Pulls a minute after start and then every six hours. Off without OV_SOURCES_INTERNAL_URL and the OAuth client secret
* (a test passes its own source).
++ *
++ * The food category feed is shared with Open Food Facts' long history, so a new cursor starts tens of thousands of
++ * items behind and the ingest chassis' 25-page default would take days of six-hourly pulls to reach this source. This
++ * consumer reads 500 items a page and up to 300 pages (150k items): the first pull passes the history, and after that
++ * the first page ends the pull. A pull still behind the tip logs a warning and resumes from its committed cursor.
*/
const ingest = require('openvibe-publishing/ingest');
@@@ -22,6 +27,9 @@ const SOURCE_KEY = 'usda-ers-fruit-vegetable-prices';
const CURSOR = 'sources.food.usda.after';
const EVERY_MS = 6 * 60 * 60 * 1000;
const FIRST_MS = 60 * 1000;
++// 500 is the Sources API's page cap; 300 pages is more than the whole food feed held on 2026-11-06 (about 79k items).
++const PAGE_SIZE = 500;
++const MAX_PAGES = 300;
// ERS's current tables use 2023 retail scanner data; the files were last updated 2025-12-09. ERS updates them each
// year, subject to data availability: re-check both when the source's page changes (the seed's terms_note says so too).
const VINTAGE = '2023';
@@@ -125,11 +133,11 @@ function createUsdaPrices({ config, s, fetchImpl = globalThis.fetch, log = conso
}
/** Follow Sources' change feed to its end. Never throws: the caller is a timer or a test. */
-- async function pull() {
++ async function pull({ pageSize = PAGE_SIZE, maxPages = MAX_PAGES } = {}) {
if (!client || running) return { skipped: true };
running = true;
try {
-- const out = await ingest.pullChanges({ db: s.db, cursor, source: client, name: CURSOR, apply });
++ const out = await ingest.pullChanges({ db: s.db, cursor, source: client, name: CURSOR, apply, pageSize, maxPages });
stats.pulls++;
stats.applied += out.applied;
stats.removed += out.removed;
@@@ -137,6 +145,7 @@ function createUsdaPrices({ config, s, fetchImpl = globalThis.fetch, log = conso
stats.last_error = null;
await load();
if (out.applied || out.removed) log.log(`[Prices] ${out.applied} USDA rows updated, ${out.removed} removed (Sources cursor ${out.after})`);
++ if (out.pages >= maxPages) log.warn(`[Prices] USDA pull is still behind after ${out.pages} pages (Sources cursor ${out.after}); the next pull continues`);
return out;
} catch (err) {
stats.last_error = text(err && err.message, 300);
ddiff --git a/test/usda-prices.test.js b/test/usda-prices.test.js
iindex d8adb6d..5c80411 100644
---- a/test/usda-prices.test.js
++++ b/test/usda-prices.test.js
@@@ -28,14 +28,20 @@ const blsItem = (series, value, n) => ({
});
function feed(items) {
-- return {
++ const f = {
items,
++ // A page cap smaller than the requested limit: a check can then exercise many pages without thousands of items.
++ perPage: null,
++ lastLimit: null,
async listItems({ after = 0, limit = 100 }) {
++ f.lastLimit = limit;
++ const take = f.perPage || limit;
const rest = items.filter((x) => x.change_seq > after).sort((a, b) => a.change_seq - b.change_seq);
-- const page = rest.slice(0, limit);
-- return { items: page, next_after: page.length ? page[page.length - 1].change_seq : after, more: rest.length > limit, sources: {} };
++ const page = rest.slice(0, take);
++ return { items: page, next_after: page.length ? page[page.length - 1].change_seq : after, more: rest.length > take, sources: {} };
},
};
++ return f;
}
(async () => {
@@@ -110,6 +116,19 @@ function feed(items) {
assert.ok(!tuna.text.includes('USDA average price'));
assert.ok(!tuna.text.includes('US average price'));
});
++
++ await check('a first pull reads past the shared food feed\'s 25-page default to reach the source\'s items', async () => {
++ // The food feed holds Open Food Facts' history before this source's rows: 29 items Food holds, then potatoes.
++ // One item a page makes the pull cross the ingest default of 25 pages (a real page holds 500).
++ for (let n = 10; n < 39; n++) usda.items.push(usdaItem('Kale | Fresh', 3.57, n, { id: `itm_junk${n}` }));
++ usda.items.push(usdaItem('Potatoes | Fresh', 0.921544199, 39));
++ usda.perPage = 1;
++ const out = await t.ctx.usdaPrices.pull();
++ assert.strictEqual(usda.lastLimit, 500, 'the Sources API page cap, not the chassis default of 200');
++ assert.deepStrictEqual([out.pages, out.applied, out.hold], [30, 1, 29]);
++ assert.strictEqual(t.ctx.usdaPrices.forFood('potatoes-5lb').value, 0.9215);
++ assert.strictEqual(Number(await t.ctx.s.db.value("SELECT cursor FROM food_ingest_cursor WHERE name = 'sources.food.usda.after'")), 39);
++ });
} finally {
await t.close();
}