A cap on the catalog's NVMe size: above it only the tail and the stored ranges
After reset 24 the catalog grew ~3.7 GB an hour (77 GB, root 31 GB free, 600k history ranges to go). CATALOG_MAX_NVME_GB (as voyager 1's BACKFILL_MAX_NVME_GB): once the default tablespace is over it, leases and the prefetcher take priority < 0 only (the tail, the stored ranges); measured once a minute. Off when unset; 90 on homelab. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
eddbe2863f
commit
256fb24c46
6 files changed
+74
-2
No files matched your search
@@ -0,0 +1,38 @@
|
|||||||
|
import db from '@adonisjs/lucid/services/db'
|
||||||
|
import logger from '@adonisjs/core/services/logger'
|
||||||
|
import coordinatorConfig from '#config/coordinator'
|
||||||
|
|
||||||
|
/** How often the catalog's size is measured (pg_tablespace_size walks its files). */
|
||||||
|
const CHECK_MS = 60_000
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The catalog's room on the NVMe (CATALOG_MAX_NVME_GB, like voyager 1's
|
||||||
|
* BACKFILL_MAX_NVME_GB): once its tablespace (the database's
|
||||||
|
* default_tablespace) is over the cap, only the tail and the stored ranges
|
||||||
|
* (priority below 0) are leased and prefetched, and history waits. After
|
||||||
|
* reset 24 the catalog grew ~3.7 GB an hour, root down to 31 GB free with
|
||||||
|
* 600k history ranges to go (LOG.md 2026-10-14 (53)). Off when unset.
|
||||||
|
*/
|
||||||
|
class CatalogSpace {
|
||||||
|
private checkedAt = 0
|
||||||
|
private over = false
|
||||||
|
|
||||||
|
async full(): Promise<boolean> {
|
||||||
|
const cap = coordinatorConfig.catalogMaxNvmeGb
|
||||||
|
if (!cap) return false
|
||||||
|
if (Date.now() - this.checkedAt < CHECK_MS) return this.over
|
||||||
|
this.checkedAt = Date.now()
|
||||||
|
const { rows } = await db.rawQuery(
|
||||||
|
`SELECT pg_tablespace_size(coalesce(nullif(current_setting('default_tablespace'), ''), 'pg_default')) AS bytes`
|
||||||
|
)
|
||||||
|
const bytes = Number(rows[0].bytes)
|
||||||
|
const was = this.over
|
||||||
|
this.over = bytes > cap * 1e9
|
||||||
|
if (this.over !== was) {
|
||||||
|
logger.info(`Catalog ${(bytes / 1e9).toFixed(1)} GB of ${cap}: history ${this.over ? 'paused (tail and stored ranges only)' : 'resumed'}.`)
|
||||||
|
}
|
||||||
|
return this.over
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export default new CatalogSpace()
|
||||||
@@ -3,6 +3,7 @@ import coordinatorConfig from '#config/coordinator'
|
|||||||
import RuleSets from '#services/rule_sets'
|
import RuleSets from '#services/rule_sets'
|
||||||
import Spool from '#services/spool'
|
import Spool from '#services/spool'
|
||||||
import Corpus from '#services/corpus'
|
import Corpus from '#services/corpus'
|
||||||
|
import CatalogSpace from '#services/catalog_space'
|
||||||
import type { Lease } from '#lib/edge/index'
|
import type { Lease } from '#lib/edge/index'
|
||||||
|
|
||||||
export class LeaseLost extends Error {}
|
export class LeaseLost extends Error {}
|
||||||
@@ -19,12 +20,14 @@ export class LeaseLost extends Error {}
|
|||||||
class Leases {
|
class Leases {
|
||||||
async lease(nodeId: number): Promise<Lease | null> {
|
async lease(nodeId: number): Promise<Lease | null> {
|
||||||
const rules = await RuleSets.current()
|
const rules = await RuleSets.current()
|
||||||
|
// The catalog at its cap on the NVMe: the tail and the stored ranges only (CatalogSpace).
|
||||||
|
const history = (await CatalogSpace.full()) ? 'AND priority < 0' : ''
|
||||||
const take = (spooled: boolean) =>
|
const take = (spooled: boolean) =>
|
||||||
db.rawQuery(
|
db.rawQuery(
|
||||||
`UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
|
`UPDATE ranges r SET status = 'leased', node_id = ?, leased_at = now(), heartbeat_at = now(), attempts = attempts + 1, rule_version = ?
|
||||||
WHERE r.id = (
|
WHERE r.id = (
|
||||||
SELECT id FROM ranges
|
SELECT id FROM ranges
|
||||||
WHERE ${spooled ? 'spooled_chunks > 0' : 'true'} AND next_id <= end_id
|
WHERE ${spooled ? 'spooled_chunks > 0' : 'true'} AND next_id <= end_id ${history}
|
||||||
AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
|
AND (status = 'planned' OR (status = 'leased' AND heartbeat_at < now() - make_interval(secs => ?)))
|
||||||
ORDER BY priority, id LIMIT 1 FOR UPDATE SKIP LOCKED)
|
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`,
|
RETURNING r.id, r.group_id, r.start_id, r.end_id, r.next_id`,
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import coordinatorConfig from '#config/coordinator'
|
|||||||
import NntpPool from '#services/nntp_pool'
|
import NntpPool from '#services/nntp_pool'
|
||||||
import Spool, { type ChunkSpan } from '#services/spool'
|
import Spool, { type ChunkSpan } from '#services/spool'
|
||||||
import Corpus from '#services/corpus'
|
import Corpus from '#services/corpus'
|
||||||
|
import CatalogSpace from '#services/catalog_space'
|
||||||
import { hedge, type Request, type XoverClient } from '#lib/nntp/index'
|
import { hedge, type Request, type XoverClient } from '#lib/nntp/index'
|
||||||
|
|
||||||
interface Target {
|
interface Target {
|
||||||
@@ -386,8 +387,10 @@ export class Prefetcher {
|
|||||||
const { perRange } = coordinatorConfig.spool
|
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)
|
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
|
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').orderBy([{ column: 'r.priority' }, { column: 'r.id' }]).limit(want)
|
? 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[] = []
|
const targets: Target[] = []
|
||||||
for (const row of [...leased, ...(room ? started : []), ...fresh]) {
|
for (const row of [...leased, ...(room ? started : []), ...fresh]) {
|
||||||
|
|||||||
@@ -77,6 +77,11 @@ export default {
|
|||||||
/** BODY commands in flight per connection (pipelined; XoverClient.bodies). */
|
/** BODY commands in flight per connection (pipelined; XoverClient.bodies). */
|
||||||
pipeline: env.get('PROBE_PIPELINE', 8),
|
pipeline: env.get('PROBE_PIPELINE', 8),
|
||||||
},
|
},
|
||||||
|
/**
|
||||||
|
* The catalog's tablespace may grow to this many GB (app/services/catalog_space.ts); above it only the tail and
|
||||||
|
* the stored ranges are leased and prefetched. 0: no cap.
|
||||||
|
*/
|
||||||
|
catalogMaxNvmeGb: env.get('CATALOG_MAX_NVME_GB', 0),
|
||||||
/**
|
/**
|
||||||
* Spotnet spots (app/services/spotter.ts, spot_inspector.ts): the group the spots are posted in, how many of
|
* Spotnet spots (app/services/spotter.ts, spot_inspector.ts): the group the spots are posted in, how many of
|
||||||
* its last articles the spotter reads the first time (~215 spots a day: ~3 months), how often it looks again,
|
* its last articles the spotter reads the first time (~215 spots a day: ~3 months), how often it looks again,
|
||||||
|
|||||||
@@ -61,6 +61,7 @@ export default await Env.create(new URL('../', import.meta.url), {
|
|||||||
LEASE_TTL_SECONDS: Env.schema.number.optional(),
|
LEASE_TTL_SECONDS: Env.schema.number.optional(),
|
||||||
SPOOL_DIR: Env.schema.string.optional(),
|
SPOOL_DIR: Env.schema.string.optional(),
|
||||||
CORPUS_DIR: Env.schema.string.optional(),
|
CORPUS_DIR: Env.schema.string.optional(),
|
||||||
|
CATALOG_MAX_NVME_GB: Env.schema.number.optional(),
|
||||||
SPOT_GROUP: Env.schema.string.optional(),
|
SPOT_GROUP: Env.schema.string.optional(),
|
||||||
SPOT_BACKFILL: Env.schema.number.optional(),
|
SPOT_BACKFILL: Env.schema.number.optional(),
|
||||||
SPOT_EVERY_SECONDS: Env.schema.number.optional(),
|
SPOT_EVERY_SECONDS: Env.schema.number.optional(),
|
||||||
|
|||||||
@@ -0,0 +1,22 @@
|
|||||||
|
import { test } from '@japa/runner'
|
||||||
|
import coordinatorConfig from '#config/coordinator'
|
||||||
|
|
||||||
|
test.group('CatalogSpace', () => {
|
||||||
|
test('is full over its cap, and off without one', async ({ assert }) => {
|
||||||
|
const before = coordinatorConfig.catalogMaxNvmeGb
|
||||||
|
try {
|
||||||
|
const { default: off } = await import('#services/catalog_space')
|
||||||
|
coordinatorConfig.catalogMaxNvmeGb = 0
|
||||||
|
assert.isFalse(await off.full())
|
||||||
|
// A fresh measurement with a cap far under any database's size.
|
||||||
|
;(off as any).checkedAt = 0
|
||||||
|
coordinatorConfig.catalogMaxNvmeGb = 0.000001
|
||||||
|
assert.isTrue(await off.full())
|
||||||
|
;(off as any).checkedAt = 0
|
||||||
|
coordinatorConfig.catalogMaxNvmeGb = 1_000_000
|
||||||
|
assert.isFalse(await off.full())
|
||||||
|
} finally {
|
||||||
|
coordinatorConfig.catalogMaxNvmeGb = before
|
||||||
|
}
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user