Posting-time windows into the corpus: ranges:dates, ranges:capture
ranges:dates samples each active group's posting dates from the news server (group_dates) and estimates every range's date (ranges.est_date). ranges:capture --from --to marks a window's ranges of every group for the corpus. The corpus fill takes capture ranges in priority order, done ones too, and remembers complete ones (ranges.stored_at). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
1 parent
0469fcd975
commit
137e931e9e
8 files changed
+683
-40
No files matched your search
@@ -0,0 +1,122 @@
|
||||
import type { Overview, XoverClient } from '#lib/nntp/index'
|
||||
|
||||
/** When an article of a group was posted: the median Date of a slice of articles at `article`. */
|
||||
export interface DateSample {
|
||||
article: number
|
||||
postedAt: Date
|
||||
}
|
||||
|
||||
/** Before this, a Date header is a broken clock (the oldest retention is 2008). */
|
||||
const OLDEST = Date.UTC(1995, 0, 1)
|
||||
const DAY = 86_400_000
|
||||
|
||||
/**
|
||||
* The median of the overviews' Date headers that parse and are plausible (1995 … a day from `now`), with at least
|
||||
* `min` of them; null otherwise (an expired or empty slice).
|
||||
*/
|
||||
export function medianDate(overviews: Overview[], now = new Date(), min = 3): Date | null {
|
||||
const latest = now.getTime() + DAY
|
||||
const times = overviews
|
||||
.map(([, fields]) => Date.parse(fields.date ?? ''))
|
||||
.filter((t) => t >= OLDEST && t <= latest)
|
||||
.sort((a, b) => a - b)
|
||||
if (times.length < min) return null
|
||||
return new Date(times[Math.floor(times.length / 2)])
|
||||
}
|
||||
|
||||
/** Where a group is sampled: `count` article numbers spread over low … high, the first at low and the last `size` before high. */
|
||||
export function samplePoints(low: number, high: number, count: number, size: number): number[] {
|
||||
const last = Math.max(low, high - size + 1)
|
||||
if (count <= 1 || last <= low) return [low]
|
||||
const points = new Set<number>()
|
||||
for (let i = 0; i < count; i++) points.add(Math.round(low + ((last - low) * i) / (count - 1)))
|
||||
return [...points].sort((a, b) => a - b)
|
||||
}
|
||||
|
||||
/**
|
||||
* The samples as a posting-time curve: in article order, a sample later than both of the next two by over a day
|
||||
* (a slice of one poster's wrong clock) is dropped, then each date is at least its predecessor's (articles arrive
|
||||
* in number order, bar stragglers and old crossposts).
|
||||
*/
|
||||
export function monotone(samples: DateSample[]): DateSample[] {
|
||||
const sorted = [...samples].sort((a, b) => a.article - b.article)
|
||||
const kept = sorted.filter((s, i) => {
|
||||
const next = sorted.slice(i + 1, i + 3)
|
||||
return next.length < 2 || next.some((n) => n.postedAt.getTime() + DAY >= s.postedAt.getTime())
|
||||
})
|
||||
let floor = Number.NEGATIVE_INFINITY
|
||||
return kept.map((s) => {
|
||||
floor = Math.max(floor, s.postedAt.getTime())
|
||||
return { article: s.article, postedAt: new Date(floor) }
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* When `article` was posted by the group's samples (monotone, in article order): interpolated between the two around
|
||||
* it, the first or last sample's date beyond them; null without samples.
|
||||
*/
|
||||
export function estimateDate(samples: DateSample[], article: number): Date | null {
|
||||
if (samples.length === 0) return null
|
||||
if (article <= samples[0].article) return samples[0].postedAt
|
||||
const last = samples[samples.length - 1]
|
||||
if (article >= last.article) return last.postedAt
|
||||
let lo = 0
|
||||
let hi = samples.length - 1
|
||||
while (hi - lo > 1) {
|
||||
const mid = (lo + hi) >> 1
|
||||
if (samples[mid].article <= article) lo = mid
|
||||
else hi = mid
|
||||
}
|
||||
const a = samples[lo]
|
||||
const b = samples[hi]
|
||||
const f = (article - a.article) / (b.article - a.article)
|
||||
return new Date(a.postedAt.getTime() + f * (b.postedAt.getTime() - a.postedAt.getTime()))
|
||||
}
|
||||
|
||||
/** How many samples a group gets: one per `every` articles of its span, at least `min` and at most `max`. */
|
||||
export function sampleCount(
|
||||
low: number,
|
||||
high: number,
|
||||
every: number,
|
||||
min = 16,
|
||||
max = 1000
|
||||
): number {
|
||||
return Math.min(max, Math.max(min, Math.ceil((high - low) / every) + 1))
|
||||
}
|
||||
|
||||
/**
|
||||
* Samples a group over one connection: GROUP, then an XOVER of `size` articles at each of `count` points (a slice
|
||||
* with too few articles that parse, expired or sparse, is widened to 32 × size once). The curve (monotone) and the
|
||||
* server's span.
|
||||
*/
|
||||
export async function sampleGroup(
|
||||
client: XoverClient,
|
||||
name: string,
|
||||
options: { every: number; size: number },
|
||||
now = new Date()
|
||||
): Promise<{ samples: DateSample[]; low: number; high: number; commands: number }> {
|
||||
const info = await client.group(name)
|
||||
const samples: DateSample[] = []
|
||||
let commands = 1
|
||||
for (const point of samplePoints(
|
||||
info.low,
|
||||
info.high,
|
||||
sampleCount(info.low, info.high, options.every),
|
||||
options.size
|
||||
)) {
|
||||
commands++
|
||||
let date = medianDate(
|
||||
await client.xover(point, Math.min(point + options.size - 1, info.high)),
|
||||
now
|
||||
)
|
||||
if (!date) {
|
||||
commands++
|
||||
date = medianDate(
|
||||
await client.xover(point, Math.min(point + options.size * 32 - 1, info.high)),
|
||||
now
|
||||
)
|
||||
}
|
||||
if (date) samples.push({ article: point, postedAt: date })
|
||||
}
|
||||
return { samples: monotone(samples), low: info.low, high: info.high, commands }
|
||||
}
|
||||
+47
-9
@@ -41,7 +41,12 @@ class Leases {
|
||||
AND l.heartbeat_at >= now() - make_interval(secs => ?)) < ${MAX_LEASES_PER_GROUP}
|
||||
ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
|
||||
RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
|
||||
[nodeId, rules.version, coordinatorConfig.leaseTtlSeconds, coordinatorConfig.leaseTtlSeconds]
|
||||
[
|
||||
nodeId,
|
||||
rules.version,
|
||||
coordinatorConfig.leaseTtlSeconds,
|
||||
coordinatorConfig.leaseTtlSeconds,
|
||||
]
|
||||
)
|
||||
let { rows } = await take(true)
|
||||
if (rows.length === 0) ({ rows } = await take(false))
|
||||
@@ -49,7 +54,16 @@ class Leases {
|
||||
if (!row) return null
|
||||
await db.from('nodes').where('id', nodeId).update({ last_seen_at: new Date() })
|
||||
const group = await db.from('groups').where('id', row.group_id).select('name').firstOrFail()
|
||||
return { range: { id: Number(row.id), group: group.name, start: Number(row.start_id), end: Number(row.end_id), next: Number(row.next_id) }, rules }
|
||||
return {
|
||||
range: {
|
||||
id: Number(row.id),
|
||||
group: group.name,
|
||||
start: Number(row.start_id),
|
||||
end: Number(row.end_id),
|
||||
next: Number(row.next_id),
|
||||
},
|
||||
rules,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -70,7 +84,8 @@ class Leases {
|
||||
|
||||
async leased(rangeId: number, nodeId: number) {
|
||||
const range = await db.from('ranges').where('id', rangeId).first()
|
||||
if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId) throw new LeaseLost(`Range ${rangeId} is not leased to this node`)
|
||||
if (!range || range.status !== 'leased' || Number(range.node_id) !== nodeId)
|
||||
throw new LeaseLost(`Range ${rangeId} is not leased to this node`)
|
||||
return range
|
||||
}
|
||||
|
||||
@@ -83,8 +98,14 @@ class Leases {
|
||||
/** The worker reached the range's end: done, and its spooled chunks go. */
|
||||
async finish(rangeId: number, nodeId: number) {
|
||||
const range = await this.leased(rangeId, nodeId)
|
||||
if (Number(range.next_id) <= Number(range.end_id)) throw new LeaseLost(`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`)
|
||||
await db.from('ranges').where('id', rangeId).update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() })
|
||||
if (Number(range.next_id) <= Number(range.end_id))
|
||||
throw new LeaseLost(
|
||||
`Range ${rangeId} is at ${range.next_id}, not past its end ${range.end_id}`
|
||||
)
|
||||
await db
|
||||
.from('ranges')
|
||||
.where('id', rangeId)
|
||||
.update({ status: 'done', done_at: new Date(), heartbeat_at: new Date() })
|
||||
await Spool.remove(rangeId)
|
||||
}
|
||||
|
||||
@@ -95,12 +116,19 @@ class Leases {
|
||||
*/
|
||||
async corrupt(rangeId: number, nodeId: number, from: number) {
|
||||
const range = await this.leased(rangeId, nodeId)
|
||||
const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find((s) => s.from === from)
|
||||
const span = Spool.chunksOf({ start: Number(range.start_id), end: Number(range.end_id) }).find(
|
||||
(s) => s.from === from
|
||||
)
|
||||
if (span) {
|
||||
await Spool.removeChunk(rangeId, span)
|
||||
if (range.capture) {
|
||||
const group = await db.from('groups').where('id', range.group_id).select('name').first()
|
||||
await Corpus.removeChunk({ group: group.name, start: Number(range.start_id), end: Number(range.end_id) }, span)
|
||||
await Corpus.removeChunk(
|
||||
{ group: group.name, start: Number(range.start_id), end: Number(range.end_id) },
|
||||
span
|
||||
)
|
||||
// Not complete in the corpus any more: the corpus fill fetches the chunk again.
|
||||
await db.from('ranges').where('id', rangeId).update({ stored_at: null })
|
||||
}
|
||||
}
|
||||
const plain: number[] = (range.plain_chunks ?? []).map(Number)
|
||||
@@ -108,11 +136,21 @@ class Leases {
|
||||
// refreshed yet) and spooled again (range 161820, LOG.md 2026-10-11), so a second try plain first.
|
||||
if (plain.filter((p) => p === from).length >= 2) {
|
||||
// Fetched plain (deflated here) and still not inflating: not the server's compression; set aside.
|
||||
await db.from('ranges').where('id', rangeId).update({ status: 'failed', node_id: null, error: `chunk ${from} doesn't inflate, also fetched plain` })
|
||||
await db
|
||||
.from('ranges')
|
||||
.where('id', rangeId)
|
||||
.update({
|
||||
status: 'failed',
|
||||
node_id: null,
|
||||
error: `chunk ${from} doesn't inflate, also fetched plain`,
|
||||
})
|
||||
return
|
||||
}
|
||||
// The server's compressed reply is cut short (the same bytes every time): fetched plain next time.
|
||||
await db.rawQuery(`UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`, [from, rangeId])
|
||||
await db.rawQuery(
|
||||
`UPDATE ranges SET status = 'planned', node_id = NULL, plain_chunks = array_append(plain_chunks, ?::bigint) WHERE id = ?`,
|
||||
[from, rangeId]
|
||||
)
|
||||
}
|
||||
|
||||
/** A stopping worker gives its range back; the next one resumes it. */
|
||||
|
||||
+130
-31
@@ -73,7 +73,13 @@ export class Prefetcher {
|
||||
private writingNow = 0
|
||||
private sampler?: NodeJS.Timeout
|
||||
/** Requests in flight: what, since when, on which connection (the report names the oldest). */
|
||||
private inFlight = new Set<{ rangeId: number; from: number; to: number; since: number; client: XoverClient }>()
|
||||
private inFlight = new Set<{
|
||||
rangeId: number
|
||||
from: number
|
||||
to: number
|
||||
since: number
|
||||
client: XoverClient
|
||||
}>()
|
||||
/** Capture ranges found complete in the corpus (not looked at again). */
|
||||
private stored = new Set<number>()
|
||||
|
||||
@@ -163,10 +169,14 @@ export class Prefetcher {
|
||||
}
|
||||
const corpusRange = { group: target.group, start: target.start, end: target.end }
|
||||
// A range in the corpus is read from disk (a reset re-indexes it without the news server).
|
||||
let data = target.capture && !target.corpusOnly ? await Corpus.read(corpusRange, span) : null
|
||||
let data =
|
||||
target.capture && !target.corpusOnly ? await Corpus.read(corpusRange, span) : null
|
||||
const stored = data !== null
|
||||
try {
|
||||
if (!data) data = target.plain.has(span.from) ? await NntpPool.usePlain(fetch) : await this.fetchHedged(target.id, target.group, span)
|
||||
if (!data)
|
||||
data = target.plain.has(span.from)
|
||||
? await NntpPool.usePlain(fetch)
|
||||
: await this.fetchHedged(target.id, target.group, span)
|
||||
} finally {
|
||||
this.fetchingNow--
|
||||
}
|
||||
@@ -176,7 +186,8 @@ export class Prefetcher {
|
||||
this.stats.chunks++
|
||||
this.stats.articles += span.to - span.from + 1
|
||||
this.stats.fetchMs += Date.now() - fetchedAt
|
||||
if (target.capture) await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter())
|
||||
if (target.capture)
|
||||
await Corpus.write(corpusRange, span, data, target.total, await this.corpusFilter())
|
||||
}
|
||||
if (target.corpusOnly) {
|
||||
this.stats.captured++
|
||||
@@ -193,7 +204,8 @@ export class Prefetcher {
|
||||
`UPDATE ranges SET spooled_chunks = spooled_chunks + 1 WHERE id = ? RETURNING spooled_chunks`,
|
||||
[target.id]
|
||||
)
|
||||
if (Number(rows[0]?.spooled_chunks) >= target.total) await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
|
||||
if (Number(rows[0]?.spooled_chunks) >= target.total)
|
||||
await db.from('ranges').where('id', target.id).update({ spooled_at: new Date() })
|
||||
} finally {
|
||||
this.writingNow--
|
||||
this.stats.writeMs += Date.now() - writtenAt
|
||||
@@ -215,7 +227,10 @@ export class Prefetcher {
|
||||
*/
|
||||
private async fetchHedged(rangeId: number, group: string, span: ChunkSpan): Promise<Buffer> {
|
||||
let age: number | undefined
|
||||
const request = (start: (work: (client: XoverClient) => Promise<Buffer>) => Promise<Buffer> | null, first: boolean): Request<Buffer> | null => {
|
||||
const request = (
|
||||
start: (work: (client: XoverClient) => Promise<Buffer>) => Promise<Buffer> | null,
|
||||
first: boolean
|
||||
): Request<Buffer> | null => {
|
||||
let client: XoverClient | undefined
|
||||
let cancelled = false
|
||||
const done = start(async (c) => {
|
||||
@@ -247,7 +262,9 @@ export class Prefetcher {
|
||||
data = await c.xoverRaw(span.from, span.to)
|
||||
} catch (error) {
|
||||
if (!late) throw error
|
||||
throw new Error(`Chunk ${span.from}–${span.to} of range ${rangeId} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB in a minute, ${Math.round(c.buffered / 1000)} kB in); fetching it again`)
|
||||
throw new Error(
|
||||
`Chunk ${span.from}–${span.to} of range ${rangeId} stalled (under ${coordinatorConfig.nntpStallBytesPerMinute / 1000} kB in a minute, ${Math.round(c.buffered / 1000)} kB in); fetching it again`
|
||||
)
|
||||
} finally {
|
||||
clearInterval(watch)
|
||||
this.inFlight.delete(entry)
|
||||
@@ -290,7 +307,8 @@ export class Prefetcher {
|
||||
|
||||
/** The longest-running request: range, chunk, how long, how much came in. */
|
||||
private oldest() {
|
||||
let oldest: { rangeId: number; from: number; to: number; since: number; client: XoverClient } | undefined
|
||||
let oldest:
|
||||
{ rangeId: number; from: number; to: number; since: number; client: XoverClient } | undefined
|
||||
for (const r of this.inFlight) if (!oldest || r.since < oldest.since) oldest = r
|
||||
if (!oldest) return 'none in flight'
|
||||
return `oldest request ${Math.round((Date.now() - oldest.since) / 1000)} s (range ${oldest.rangeId}, ${oldest.from}–${oldest.to}, ${Math.round(oldest.client.buffered / 1000)} kB in)`
|
||||
@@ -315,7 +333,12 @@ export class Prefetcher {
|
||||
const { chunks, articles, fetchMs, hedged, hedgeWon, byAge, samples, requests } = this.stats
|
||||
if (chunks > 0) {
|
||||
const ages = byAge
|
||||
.map((b, i) => `${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace('– min', '+ min'))
|
||||
.map((b, i) =>
|
||||
`${i === 0 ? '<' : `${AGE_BUCKETS[i - 1]}–`}${AGE_BUCKETS[i] === Infinity ? '' : AGE_BUCKETS[i]} min ${b.slow}/${b.requests}`.replace(
|
||||
'– min',
|
||||
'+ min'
|
||||
)
|
||||
)
|
||||
.join(', ')
|
||||
logger.info(
|
||||
`Prefetched ${chunks} chunks in ${seconds.toFixed(0)} s: ${Math.round(articles / seconds).toLocaleString('en')} articles/s, ` +
|
||||
@@ -341,13 +364,18 @@ export class Prefetcher {
|
||||
}
|
||||
|
||||
private peek(): boolean {
|
||||
return this.targets.some((t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`)))
|
||||
return this.targets.some(
|
||||
(t) => this.open(t) && t.missing.some((s) => !this.claimed.has(`${t.id}:${s.from}`))
|
||||
)
|
||||
}
|
||||
|
||||
/** Whether the target is under its range's and its group's limit of connections. */
|
||||
private open(target: Target) {
|
||||
const { perRange, perGroup } = coordinatorConfig.spool
|
||||
return (perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) && (perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup)
|
||||
return (
|
||||
(perRange <= 0 || (this.perRange.get(target.id) ?? 0) < perRange) &&
|
||||
(perGroup <= 0 || (this.perGroup.get(target.group) ?? 0) < perGroup)
|
||||
)
|
||||
}
|
||||
|
||||
private count(target: Target, by: number) {
|
||||
@@ -378,26 +406,58 @@ export class Prefetcher {
|
||||
this.sizedAt = Date.now()
|
||||
this.spoolBytes = await Spool.size()
|
||||
}
|
||||
const mine = (q: any) => (this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q)
|
||||
const rows = (q: any) => mine(q.from('ranges as r').join('groups as g', 'g.id', 'r.group_id').select('r.id', 'g.name as group', 'r.start_id', 'r.end_id', 'r.next_id', 'r.plain_chunks', 'r.capture'))
|
||||
const mine = (q: any) =>
|
||||
this.shard.count > 1 ? q.whereRaw('r.id % ? = ?', [this.shard.count, this.shard.index]) : q
|
||||
const rows = (q: any) =>
|
||||
mine(
|
||||
q
|
||||
.from('ranges as r')
|
||||
.join('groups as g', 'g.id', 'r.group_id')
|
||||
.select(
|
||||
'r.id',
|
||||
'g.name as group',
|
||||
'r.start_id',
|
||||
'r.end_id',
|
||||
'r.next_id',
|
||||
'r.plain_chunks',
|
||||
'r.capture'
|
||||
)
|
||||
)
|
||||
// Leased ranges, checked against the spool's files whatever spooled_at says: a worker waits for them.
|
||||
const leased = await rows(db.query()).where('r.status', 'leased').orderBy('r.leased_at')
|
||||
// Planned ranges ahead: those started already, then new ones up to SPOOL_AHEAD_PER_LEASE per lease.
|
||||
const ahead = Math.max(coordinatorConfig.spool.minAhead, coordinatorConfig.spool.aheadPerLease * leased.length)
|
||||
const started = await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', '>', 0).whereNull('r.spooled_at').orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
|
||||
const [{ ready }] = await mine(db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)).count('* as ready')
|
||||
const ahead = Math.max(
|
||||
coordinatorConfig.spool.minAhead,
|
||||
coordinatorConfig.spool.aheadPerLease * leased.length
|
||||
)
|
||||
const started = await rows(db.query())
|
||||
.where('r.status', 'planned')
|
||||
.where('r.spooled_chunks', '>', 0)
|
||||
.whereNull('r.spooled_at')
|
||||
.orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
|
||||
const [{ ready }] = await mine(
|
||||
db.from('ranges as r').where('r.status', 'planned').where('r.spooled_chunks', '>', 0)
|
||||
).count('* as ready')
|
||||
const room = this.spoolBytes < coordinatorConfig.spool.maxBytes
|
||||
// With at most perRange connections a range, enough ranges still being fetched for all loops: fully
|
||||
// spooled ones count as ready but need no connection (30 of 40 busy on small ranges; LOG.md 2026-10-10 (2)).
|
||||
// 8 more than that: ranges with a chunk or two left take fewer than perRange (with +2, 3 of 40 loops idle).
|
||||
const { perRange } = coordinatorConfig.spool
|
||||
const short = perRange > 0 ? Math.ceil(this.loops / perRange) + 8 - (leased.length + started.length) : 0
|
||||
const short =
|
||||
perRange > 0 ? Math.ceil(this.loops / perRange) + 8 - (leased.length + started.length) : 0
|
||||
const want = Math.max(ahead - Number(ready), short)
|
||||
// The catalog at its cap on the NVMe: no history fetched ahead (CatalogSpace).
|
||||
const full = await CatalogSpace.full()
|
||||
const fresh = room && want > 0
|
||||
? await rows(db.query()).where('r.status', 'planned').where('r.spooled_chunks', 0).whereRaw('r.next_id <= r.end_id').where((q: any) => (full ? q.where('r.priority', '<', 0) : q)).orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(want)
|
||||
: []
|
||||
const fresh =
|
||||
room && want > 0
|
||||
? await rows(db.query())
|
||||
.where('r.status', 'planned')
|
||||
.where('r.spooled_chunks', 0)
|
||||
.whereRaw('r.next_id <= r.end_id')
|
||||
.where((q: any) => (full ? q.where('r.priority', '<', 0) : q))
|
||||
.orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
|
||||
.limit(want)
|
||||
: []
|
||||
const targets: Target[] = []
|
||||
for (const row of [...leased, ...(room ? started : []), ...fresh]) {
|
||||
const start = Number(row.start_id)
|
||||
@@ -405,14 +465,25 @@ export class Prefetcher {
|
||||
const total = Spool.chunksOf({ start, end }).length
|
||||
const missing: ChunkSpan[] = []
|
||||
for (const span of Spool.chunksOf({ start, end }, Number(row.next_id))) {
|
||||
if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span))) missing.push(span)
|
||||
if (!this.claimed.has(`${row.id}:${span.from}`) && !(await Spool.has(Number(row.id), span)))
|
||||
missing.push(span)
|
||||
}
|
||||
if (missing.length > 0) {
|
||||
targets.push({ id: Number(row.id), group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total, start, end, capture: row.capture === true })
|
||||
targets.push({
|
||||
id: Number(row.id),
|
||||
group: row.group,
|
||||
plain: new Set((row.plain_chunks ?? []).map(Number)),
|
||||
missing,
|
||||
total,
|
||||
start,
|
||||
end,
|
||||
capture: row.capture === true,
|
||||
})
|
||||
}
|
||||
}
|
||||
// Last, for loops with nothing else to do: capture ranges not stored yet, into the corpus only.
|
||||
for (const target of await this.corpusTargets(rows, new Set(targets.map((t) => t.id)))) targets.push(target)
|
||||
for (const target of await this.corpusTargets(rows, new Set(targets.map((t) => t.id))))
|
||||
targets.push(target)
|
||||
return targets
|
||||
}
|
||||
|
||||
@@ -423,23 +494,38 @@ export class Prefetcher {
|
||||
if (!coordinatorConfig.corpus.filter) return undefined
|
||||
if (!this.filterRules || Date.now() - this.filterRules.at > 60_000) {
|
||||
const rules = await RuleSets.current()
|
||||
this.filterRules = { at: Date.now(), filter: { ruleVersion: rules.version, priorities: rules.priorities } }
|
||||
this.filterRules = {
|
||||
at: Date.now(),
|
||||
filter: { ruleVersion: rules.version, priorities: rules.priorities },
|
||||
}
|
||||
}
|
||||
return this.filterRules.filter
|
||||
}
|
||||
|
||||
/**
|
||||
* Up to corpus.ahead planned capture ranges, not leased soon, whose chunks aren't all in the corpus; as many of a
|
||||
* group as its connections allow (perGroup / perRange): in id order the 16 were one group's, which took 4 of the
|
||||
* loops and left the rest idle (30k articles/s; LOG.md 2026-10-23 (100)).
|
||||
* Up to corpus.ahead capture ranges whose chunks aren't all in the corpus, by priority: planned ones not started
|
||||
* (not leased soon), and done ones (a window of every group, ranges:capture, takes those indexed already too);
|
||||
* as many of a group as its connections allow (perGroup / perRange): in id order the 16 were one group's, which
|
||||
* took 4 of the loops and left the rest idle (30k articles/s; LOG.md 2026-10-23 (100)). A range found complete
|
||||
* gets stored_at and isn't read again.
|
||||
*/
|
||||
private async corpusTargets(rows: (q: any) => any, taken: Set<number>): Promise<Target[]> {
|
||||
const { dir, ahead } = coordinatorConfig.corpus
|
||||
if (!dir || ahead <= 0) return []
|
||||
const { perRange, perGroup } = coordinatorConfig.spool
|
||||
const ofGroup = perGroup > 0 && perRange > 0 ? Math.max(1, Math.floor(perGroup / perRange)) : Infinity
|
||||
const ofGroup =
|
||||
perGroup > 0 && perRange > 0 ? Math.max(1, Math.floor(perGroup / perRange)) : Infinity
|
||||
const groups = new Map<string, number>()
|
||||
const candidates = await rows(db.query()).where('r.status', 'planned').where('r.capture', true).where('r.spooled_chunks', 0).orderBy('r.id')
|
||||
const candidates = await rows(db.query())
|
||||
.where('r.capture', true)
|
||||
.whereNull('r.stored_at')
|
||||
.where((q: any) =>
|
||||
q
|
||||
.where((p: any) => p.where('r.status', 'planned').where('r.spooled_chunks', 0))
|
||||
.orWhere('r.status', 'done')
|
||||
)
|
||||
.orderBy([{ column: 'r.priority' }, { column: 'r.id' }])
|
||||
.limit(ahead * 16)
|
||||
const targets: Target[] = []
|
||||
for (const row of candidates) {
|
||||
if (targets.length >= ahead) break
|
||||
@@ -451,13 +537,26 @@ export class Prefetcher {
|
||||
const range = { group: row.group, start, end }
|
||||
if (await Corpus.complete(range)) {
|
||||
this.stored.add(id)
|
||||
await db.from('ranges').where('id', id).update({ stored_at: new Date() })
|
||||
continue
|
||||
}
|
||||
const chunks = Spool.chunksOf({ start, end })
|
||||
const missing: ChunkSpan[] = []
|
||||
for (const span of chunks) if (!this.claimed.has(`${id}:${span.from}`) && !(await Corpus.has(range, span))) missing.push(span)
|
||||
for (const span of chunks)
|
||||
if (!this.claimed.has(`${id}:${span.from}`) && !(await Corpus.has(range, span)))
|
||||
missing.push(span)
|
||||
if (missing.length > 0) {
|
||||
targets.push({ id, group: row.group, plain: new Set((row.plain_chunks ?? []).map(Number)), missing, total: chunks.length, start, end, capture: true, corpusOnly: true })
|
||||
targets.push({
|
||||
id,
|
||||
group: row.group,
|
||||
plain: new Set((row.plain_chunks ?? []).map(Number)),
|
||||
missing,
|
||||
total: chunks.length,
|
||||
start,
|
||||
end,
|
||||
capture: true,
|
||||
corpusOnly: true,
|
||||
})
|
||||
groups.set(row.group, (groups.get(row.group) ?? 0) + 1)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user