Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 11 additions & 2 deletions apps/poller/src/index.js
Original file line number Diff line number Diff line change
Expand Up @@ -795,11 +795,18 @@ async function ringTick() {

try {
if (Date.now() - lastRingSeed >= ringSeedMs) {
// Stamped only once the pass has run: a pass that throws is retried
// ten minutes on, not six hours on, which is what the first deploy
// would have waited after the largest topic timed out.
const seeded = await seedTopRings(db, {
topics: ringTopics,
minMembers: 5,
onError: (topic, err) => log('ring-seed-error', { topic, message: String(err?.message ?? err) }),
});
lastRingSeed = Date.now();
const seeded = await seedTopRings(db, { topics: ringTopics, minMembers: 5 });
// Logged only when something changed: once the rings exist, a line
// every six hours saying "0 added" is not a log.
if (seeded.created || seeded.added) log('rings-seeded', seeded);
if (seeded.created || seeded.added || seeded.failed) log('rings-seeded', seeded);
}

const result = await verifyRingMembers(db, {
Expand All @@ -814,6 +821,8 @@ async function ringTick() {
if (result.checked) log('rings', result);
} catch (err) {
log('rings-error', { message: String(err?.message ?? err) });
// Try the seed again soon rather than at the next six-hour mark.
lastRingSeed = Math.min(lastRingSeed, Date.now() - ringSeedMs + 10 * 60 * 1000);
} finally {
ringing = false;
}
Expand Down
41 changes: 24 additions & 17 deletions packages/db/src/webrings.js
Original file line number Diff line number Diff line change
Expand Up @@ -224,18 +224,19 @@ export async function memberBySlug(db, ringSlug, memberSlug) {
* @returns {Promise<Array<{ id: string, slug: string, site_url: string, created_at: string }>>}
*/
export async function topicRingCandidates(db, topicSlug, limit) {
// Driven from feed_keywords, whose (slug, count) index makes this a range
// scan of one topic's rows, rather than from feeds, where the same filter
// is a walk of the whole directory with a subquery per row. Grouped on the
// feed because a feed can carry several spellings of one slug.
// Walked from feeds in admission order (feeds_created_idx) with a primary
// key probe into feed_keywords per row, so the statement stops the moment
// `limit` members are found. The other way round, a range scan of the
// topic's keyword rows joined, grouped and sorted before the limit applies,
// is a full pass over a big topic: the first seed in production timed out
// on the largest topic before anything was written. A small topic walks
// the directory with a point lookup per feed, which is the cheap case.
const { rows } = await db.execute({
sql: `select f.id, f.slug, f.site_url, f.created_at
from feed_keywords k
join feeds f on f.id = k.feed_id
where k.slug = ?
and f.status = 'active'
from feeds f
where f.status = 'active'
and f.site_url is not null and f.site_url <> ''
group by f.id
and exists (select 1 from feed_keywords k where k.feed_id = f.id and k.slug = ?)
order by f.created_at asc, f.id asc
limit ?`,
args: [topicSlug, limit],
Expand Down Expand Up @@ -352,15 +353,21 @@ export async function topRingTopics(db, opts = {}) {
* @param {string} topicSlug
* @returns {Promise<number>}
*/
export async function topicRingSize(db, topicSlug) {
export async function topicRingSize(db, topicSlug, opts = {}) {
// `cap` stops the count once it is high enough to answer the caller's
// question ("at least five?"), so the biggest topics cost the same as the
// smallest. Without it the count is exact.
const cap = Number(opts.cap) > 0 ? Number(opts.cap) : null;
const { rows } = await db.execute({
sql: `select count(distinct f.id) as n
from feed_keywords k
join feeds f on f.id = k.feed_id
where k.slug = ?
and f.status = 'active'
and f.site_url is not null and f.site_url <> ''`,
args: [topicSlug],
sql: `select count(*) as n from (
select f.id
from feeds f
where f.status = 'active'
and f.site_url is not null and f.site_url <> ''
and exists (select 1 from feed_keywords k where k.feed_id = f.id and k.slug = ?)
${cap ? 'limit ?' : ''}
)`,
args: cap ? [topicSlug, cap] : [topicSlug],
});
return Number(rows[0]?.n ?? 0);
}
Expand Down
2 changes: 2 additions & 0 deletions packages/db/test/webrings.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,8 @@ test('the top topics are the most covered ones with enough feeds to ring', async
assert.equal(await webrings.topicRingSize(db, 'physics'), 5, 'the feeds that can link, distinct');
assert.equal(await webrings.topicRingSize(db, 'chemistry'), 1);
assert.equal(await webrings.topicRingSize(db, 'nothing'), 0);
assert.equal(await webrings.topicRingSize(db, 'physics', { cap: 3 }), 3, 'a capped count stops at the cap');
assert.equal(await webrings.topicRingSize(db, 'chemistry', { cap: 3 }), 1, 'and is exact under it');
});

test('a check records status and stamp, and a descriptor sets made_by', async () => {
Expand Down
32 changes: 21 additions & 11 deletions packages/ingest/src/webring.js
Original file line number Diff line number Diff line change
Expand Up @@ -337,9 +337,14 @@ export async function verifyRingMembers(db, opts) {
* eligible feeds fall under `minMembers` is skipped rather than made into
* a ring of two, since a ring needs somewhere to hop to.
*
* A topic whose queries fail (the first production seed hit the request
* deadline on the largest topic) is counted as failed and reported through
* `onError`, and the pass goes on to the next one; one slow topic must not
* cost every other ring its seed.
*
* @param {import('@libsql/client').Client} db
* @param {{ topics?: number, minMembers?: number, limit?: number }} [opts]
* @returns {Promise<{ rings: number, created: number, added: number, skipped: number }>}
* @param {{ topics?: number, minMembers?: number, limit?: number, onError?: ((topic: string, err: unknown) => void)|null }} [opts]
* @returns {Promise<{ rings: number, created: number, added: number, skipped: number, failed: number }>}
*/
export async function seedTopRings(db, opts = {}) {
const topics = Math.max(1, Number(opts.topics ?? 20) || 20);
Expand All @@ -350,19 +355,24 @@ export async function seedTopRings(db, opts = {}) {
// on a topic, and a topic can be well covered by feeds that have no site
// to link from or that the crawler has given up on.
const candidates = await webrings.topRingTopics(db, { count: topics * 2, minFeeds: minMembers });
const tally = { rings: 0, created: 0, added: 0, skipped: 0 };
const tally = { rings: 0, created: 0, added: 0, skipped: 0, failed: 0 };

for (const topic of candidates) {
if (tally.rings >= topics) break;
const size = await webrings.topicRingSize(db, topic.slug);
if (size < minMembers) {
tally.skipped += 1;
continue;
try {
const size = await webrings.topicRingSize(db, topic.slug, { cap: minMembers });
if (size < minMembers) {
tally.skipped += 1;
continue;
}
const result = await webrings.seedTopicRing(db, topic.slug, { limit });
tally.rings += 1;
if (result.created) tally.created += 1;
tally.added += result.added;
} catch (err) {
tally.failed += 1;
opts.onError?.(topic.slug, err);
}
const result = await webrings.seedTopicRing(db, topic.slug, { limit });
tally.rings += 1;
if (result.created) tally.created += 1;
tally.added += result.added;
}

return tally;
Expand Down
Loading