Final completeness by time window: ranges keep their posting span, release_final() over blocks of done ranges
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
3d24db5565
commit
ce43099188
12 files changed
+259
-9
No files matched your search
@@ -95,6 +95,9 @@ class BatchStore {
|
|||||||
.update({
|
.update({
|
||||||
next_id: batch.next,
|
next_id: batch.next,
|
||||||
headers: db.raw('headers + ?', [batch.headers]),
|
headers: db.raw('headers + ?', [batch.headers]),
|
||||||
|
// The range's posting time span (release_final): the batches' trimmed spans, min and max (least and greatest skip nulls).
|
||||||
|
posted_low: db.raw('least(posted_low, to_timestamp(?::double precision / 1000))', [batch.postedLow ?? null]),
|
||||||
|
posted_high: db.raw('greatest(posted_high, to_timestamp(?::double precision / 1000))', [batch.postedHigh ?? null]),
|
||||||
heartbeat_at: new Date(),
|
heartbeat_at: new Date(),
|
||||||
})
|
})
|
||||||
if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) throw new StaleBatch(`Range ${rangeId} moved on meanwhile`)
|
if (Number(moved) !== 1 && !(Array.isArray(moved) && moved.length === 1)) throw new StaleBatch(`Range ${rangeId} moved on meanwhile`)
|
||||||
|
|||||||
@@ -0,0 +1,31 @@
|
|||||||
|
import db from '@adonisjs/lucid/services/db'
|
||||||
|
|
||||||
|
/**
|
||||||
|
* range_blocks (migration 3900): per group, each run of consecutive done ranges with its posting time span, read by
|
||||||
|
* release_final(). A few thousand rows from ~650k ranges, rebuilt whole in one statement each time; a run breaks at a
|
||||||
|
* range that isn't done or doesn't start where the one before ended.
|
||||||
|
*/
|
||||||
|
class RangeBlocks {
|
||||||
|
async rebuild(): Promise<number> {
|
||||||
|
return db.transaction(async (trx) => {
|
||||||
|
await trx.rawQuery(`DELETE FROM range_blocks`)
|
||||||
|
const result = await trx.rawQuery(
|
||||||
|
`WITH r AS (
|
||||||
|
SELECT group_id, start_id, end_id, status = 'done' AS done, posted_low, posted_high,
|
||||||
|
min(start_id) OVER (PARTITION BY group_id) AS group_first,
|
||||||
|
lag(status = 'done') OVER w AS done_before, lag(end_id) OVER w AS end_before
|
||||||
|
FROM ranges WINDOW w AS (PARTITION BY group_id ORDER BY start_id)),
|
||||||
|
runs AS (
|
||||||
|
SELECT *, sum(CASE WHEN done IS DISTINCT FROM done_before OR end_before IS DISTINCT FROM start_id - 1 THEN 1 ELSE 0 END)
|
||||||
|
OVER (PARTITION BY group_id ORDER BY start_id) AS run
|
||||||
|
FROM r)
|
||||||
|
INSERT INTO range_blocks (group_id, start_id, end_id, posted_low, posted_high, at_group_start)
|
||||||
|
SELECT group_id, min(start_id), max(end_id), min(posted_low), max(posted_high), min(start_id) = min(group_first)
|
||||||
|
FROM runs WHERE done GROUP BY group_id, run`
|
||||||
|
)
|
||||||
|
return result.rowCount ?? 0
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export default new RangeBlocks()
|
||||||
@@ -51,6 +51,8 @@ export interface ReleaseSummary {
|
|||||||
size: number
|
size: number
|
||||||
readable: boolean
|
readable: boolean
|
||||||
complete: boolean | null
|
complete: boolean | null
|
||||||
|
/** The complete verdict is definite: done ranges cover the release's posting times and 6 hours either side (release_final). */
|
||||||
|
final: boolean
|
||||||
category: number
|
category: number
|
||||||
settled: boolean
|
settled: boolean
|
||||||
/** The prober found encrypted archive entries or headers; null: not probed (yet). */
|
/** The prober found encrypted archive entries or headers; null: not probed (yet). */
|
||||||
@@ -92,7 +94,8 @@ class SearchService {
|
|||||||
}
|
}
|
||||||
if (params.readable) where.push('(r.readable OR r.inner_name IS NOT NULL)')
|
if (params.readable) where.push('(r.readable OR r.inner_name IS NOT NULL)')
|
||||||
if (params.complete) where.push('r.complete')
|
if (params.complete) where.push('r.complete')
|
||||||
if (params.notIncomplete) where.push('r.complete IS NOT FALSE')
|
// Incomplete only when final: a release next to unread ranges may yet get its missing files (release_final).
|
||||||
|
if (params.notIncomplete) where.push('(r.complete IS NOT FALSE OR NOT release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at))')
|
||||||
if (params.unlocked) where.push('r.passworded IS NOT TRUE')
|
if (params.unlocked) where.push('r.passworded IS NOT TRUE')
|
||||||
if (params.source) {
|
if (params.source) {
|
||||||
where.push('r.source = ?')
|
where.push('r.source = ?')
|
||||||
@@ -133,7 +136,8 @@ class SearchService {
|
|||||||
const countLimit = offset + (params.countLimit ?? COUNT_LIMIT)
|
const countLimit = offset + (params.countLimit ?? COUNT_LIMIT)
|
||||||
const [{ rows }, { rows: counted }] = await Promise.all([
|
const [{ rows }, { rows: counted }] = await Promise.all([
|
||||||
db.rawQuery(
|
db.rawQuery(
|
||||||
`SELECT r.id, r.name, r.source, r.poster, r.groups, r.posted_at, r.file_total, r.files, r.bytes, r.readable, r.complete, r.category, r.settled_at, r.passworded, r.inner_name
|
`SELECT r.id, r.name, r.source, r.poster, r.groups, r.posted_at, r.file_total, r.files, r.bytes, r.readable, r.complete, r.category, r.settled_at, r.passworded, r.inner_name,
|
||||||
|
release_final(r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at) AS final
|
||||||
FROM releases r WHERE ${whereSql} ORDER BY ${order} LIMIT ? OFFSET ?`,
|
FROM releases r WHERE ${whereSql} ORDER BY ${order} LIMIT ? OFFSET ?`,
|
||||||
[...bindings, perPage, offset]
|
[...bindings, perPage, offset]
|
||||||
),
|
),
|
||||||
@@ -158,6 +162,7 @@ class SearchService {
|
|||||||
size: Number(r.bytes),
|
size: Number(r.bytes),
|
||||||
readable: r.readable,
|
readable: r.readable,
|
||||||
complete: r.complete,
|
complete: r.complete,
|
||||||
|
final: r.final,
|
||||||
category: r.category,
|
category: r.category,
|
||||||
settled: r.settled_at !== null,
|
settled: r.settled_at !== null,
|
||||||
passworded: r.passworded,
|
passworded: r.passworded,
|
||||||
@@ -170,9 +175,10 @@ class SearchService {
|
|||||||
async release(id: number) {
|
async release(id: number) {
|
||||||
const r = await db.from('releases').where('id', id).first()
|
const r = await db.from('releases').where('id', id).first()
|
||||||
if (!r) return null
|
if (!r) return null
|
||||||
const [files, { rows: inspections }] = await Promise.all([
|
const [files, { rows: inspections }, { rows: finality }] = await Promise.all([
|
||||||
NzbService.files(id),
|
NzbService.files(id),
|
||||||
db.rawQuery(`SELECT kind, result, created_at FROM release_inspections WHERE release_id = ? ORDER BY candidate_id`, [id]),
|
db.rawQuery(`SELECT kind, result, created_at FROM release_inspections WHERE release_id = ? ORDER BY candidate_id`, [id]),
|
||||||
|
db.rawQuery(`SELECT release_final(?, ?, ?, ?, ?) AS final`, [r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at]),
|
||||||
])
|
])
|
||||||
return {
|
return {
|
||||||
id: Number(r.id),
|
id: Number(r.id),
|
||||||
@@ -187,6 +193,7 @@ class SearchService {
|
|||||||
segments: Number(r.segments),
|
segments: Number(r.segments),
|
||||||
readable: r.readable,
|
readable: r.readable,
|
||||||
complete: r.complete,
|
complete: r.complete,
|
||||||
|
final: finality[0].final,
|
||||||
hasPar2: r.has_par2,
|
hasPar2: r.has_par2,
|
||||||
category: r.category,
|
category: r.category,
|
||||||
firstSeenAt: r.first_seen_at,
|
firstSeenAt: r.first_seen_at,
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import { Prefetcher } from '#services/prefetcher'
|
|||||||
import Planner from '#services/planner'
|
import Planner from '#services/planner'
|
||||||
import FragmentMerger from '#services/fragment_merger'
|
import FragmentMerger from '#services/fragment_merger'
|
||||||
import Settler from '#services/settler'
|
import Settler from '#services/settler'
|
||||||
|
import RangeBlocks from '#services/range_blocks'
|
||||||
import NntpPool from '#services/nntp_pool'
|
import NntpPool from '#services/nntp_pool'
|
||||||
import coordinatorConfig from '#config/coordinator'
|
import coordinatorConfig from '#config/coordinator'
|
||||||
|
|
||||||
@@ -13,6 +14,8 @@ const PLAN_EVERY_MS = 60 * 60 * 1000
|
|||||||
const MERGE_EVERY_MS = 60 * 1000
|
const MERGE_EVERY_MS = 60 * 1000
|
||||||
const MERGE_BUDGET_MS = 30 * 1000
|
const MERGE_BUDGET_MS = 30 * 1000
|
||||||
const STOP_WAIT_MS = 15 * 1000
|
const STOP_WAIT_MS = 15 * 1000
|
||||||
|
/** range_blocks (release_final) after ranges finish. */
|
||||||
|
const BLOCKS_EVERY_MS = 5 * 60 * 1000
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* The coordinator's background work (REBUILD.md, part 1): the prefetcher,
|
* The coordinator's background work (REBUILD.md, part 1): the prefetcher,
|
||||||
@@ -60,8 +63,13 @@ export default class CoordinatorRun extends BaseCommand {
|
|||||||
let plannedAt = 0
|
let plannedAt = 0
|
||||||
let tailedAt = 0
|
let tailedAt = 0
|
||||||
let mergedAt = 0
|
let mergedAt = 0
|
||||||
|
let blocksAt = 0
|
||||||
while (!stopping) {
|
while (!stopping) {
|
||||||
try {
|
try {
|
||||||
|
if (main && Date.now() - blocksAt >= BLOCKS_EVERY_MS) {
|
||||||
|
blocksAt = Date.now()
|
||||||
|
await RangeBlocks.rebuild()
|
||||||
|
}
|
||||||
if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
|
if (main && Date.now() - plannedAt >= PLAN_EVERY_MS) {
|
||||||
plannedAt = Date.now()
|
plannedAt = Date.now()
|
||||||
const { groups, ranges } = await Planner.planAll()
|
const { groups, ranges } = await Planner.planAll()
|
||||||
|
|||||||
@@ -0,0 +1,45 @@
|
|||||||
|
import { BaseSchema } from '@adonisjs/lucid/schema'
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Final completeness by time window (LOG.md 2026-10-16 (67)). A release settles once quiet, but history is read
|
||||||
|
* newest first and in islands: one next to an unread range was "incomplete" twice as often (52.5% against 25.7%),
|
||||||
|
* its missing files only not read yet. An upload is posted within hours, so a release is final (its complete verdict
|
||||||
|
* definite) once done ranges cover its posting times and FINAL_MARGIN either side.
|
||||||
|
* - ranges.posted_low/high: the posting time span the workers report per batch (postedSpan: without the earliest and
|
||||||
|
* latest 1%), the min and max over the range's batches.
|
||||||
|
* - range_blocks: runs of consecutive done ranges per group with their span (RangeBlocks rebuilds it); at_group_start
|
||||||
|
* when the run starts at the group's first range (nothing older to wait for).
|
||||||
|
* - release_final(): whether one block holds the release's articles and reaches FINAL_MARGIN past its posting times.
|
||||||
|
* Computed when asked, so no release row is written (50M updates would double the table on the NVMe).
|
||||||
|
*/
|
||||||
|
export default class extends BaseSchema {
|
||||||
|
async up() {
|
||||||
|
this.schema.raw(`ALTER TABLE ranges ADD COLUMN IF NOT EXISTS posted_low timestamptz, ADD COLUMN IF NOT EXISTS posted_high timestamptz`)
|
||||||
|
this.schema.raw(`
|
||||||
|
CREATE TABLE range_blocks (
|
||||||
|
group_id integer NOT NULL,
|
||||||
|
start_id bigint NOT NULL,
|
||||||
|
end_id bigint NOT NULL,
|
||||||
|
posted_low timestamptz,
|
||||||
|
posted_high timestamptz,
|
||||||
|
at_group_start boolean NOT NULL,
|
||||||
|
PRIMARY KEY (group_id, start_id)
|
||||||
|
)`)
|
||||||
|
// FINAL_MARGIN: 6 hours.
|
||||||
|
this.schema.raw(`
|
||||||
|
CREATE FUNCTION release_final(g integer, first_article bigint, last_article bigint, posted timestamptz, last_posted timestamptz)
|
||||||
|
RETURNS boolean LANGUAGE sql STABLE AS $$
|
||||||
|
SELECT EXISTS (
|
||||||
|
SELECT 1 FROM range_blocks b
|
||||||
|
WHERE b.group_id = g AND b.start_id <= first_article AND b.end_id >= last_article
|
||||||
|
AND (b.at_group_start OR b.posted_low <= posted - interval '6 hours')
|
||||||
|
AND b.posted_high >= coalesce(last_posted, posted) + interval '6 hours')
|
||||||
|
$$`)
|
||||||
|
}
|
||||||
|
|
||||||
|
async down() {
|
||||||
|
this.schema.raw(`DROP FUNCTION IF EXISTS release_final(integer, bigint, bigint, timestamptz, timestamptz)`)
|
||||||
|
this.schema.raw(`DROP TABLE IF EXISTS range_blocks`)
|
||||||
|
this.schema.raw(`ALTER TABLE ranges DROP COLUMN IF EXISTS posted_low, DROP COLUMN IF EXISTS posted_high`)
|
||||||
|
}
|
||||||
|
}
|
||||||
+1
-1
@@ -52,7 +52,7 @@ const release = (r: Item, b: number) => {
|
|||||||
}
|
}
|
||||||
const ts: Record<string, Item[]> = { b: [], r: [], s: [], c: [], n: [], f: [] }
|
const ts: Record<string, Item[]> = { b: [], r: [], s: [], c: [], n: [], f: [] }
|
||||||
batches.forEach((batch, b) => {
|
batches.forEach((batch, b) => {
|
||||||
ts.b.push({ b, from: batch.from, next: batch.next, headers: batch.headers })
|
ts.b.push({ b, from: batch.from, next: batch.next, headers: batch.headers, postedLow: batch.postedLow, postedHigh: batch.postedHigh })
|
||||||
for (const r of batch.releases) ts.r.push(release({ upload: r.upload, key: r.key, name: r.name, source: r.source, fileTotal: r.fileTotal, poster: r.poster, groups: r.groups, postedAt: r.postedAt, postedLast: r.postedLast, files: r.files, segments: r.segments, bytes: r.bytes, firstArticle: r.firstArticle, lastArticle: r.lastArticle, readable: r.readable, hasPar2: r.hasPar2, payload: r.payload }, b))
|
for (const r of batch.releases) ts.r.push(release({ upload: r.upload, key: r.key, name: r.name, source: r.source, fileTotal: r.fileTotal, poster: r.poster, groups: r.groups, postedAt: r.postedAt, postedLast: r.postedLast, files: r.files, segments: r.segments, bytes: r.bytes, firstArticle: r.firstArticle, lastArticle: r.lastArticle, readable: r.readable, hasPar2: r.hasPar2, payload: r.payload }, b))
|
||||||
for (const s of batch.samples) ts.s.push({ b, key: s.key, upload: s.upload, rule: s.rule, ruleKey: s.ruleKey, segments: s.segments })
|
for (const s of batch.samples) ts.s.push({ b, key: s.key, upload: s.upload, rule: s.rule, ruleKey: s.ruleKey, segments: s.segments })
|
||||||
for (const c of batch.candidates) ts.c.push({ b, upload: c.upload, releaseKey: c.releaseKey, kind: c.kind, messageId: c.messageId, file: c.file })
|
for (const c of batch.candidates) ts.c.push({ b, upload: c.upload, releaseKey: c.releaseKey, kind: c.kind, messageId: c.messageId, file: c.file })
|
||||||
|
|||||||
@@ -78,6 +78,9 @@ pub struct Batch {
|
|||||||
pub from: i64,
|
pub from: i64,
|
||||||
pub next: i64,
|
pub next: i64,
|
||||||
pub headers: u64,
|
pub headers: u64,
|
||||||
|
/// postedLow / postedHigh: the batch's posting times without the earliest and latest 1% (postedSpan).
|
||||||
|
pub posted_low: Option<i64>,
|
||||||
|
pub posted_high: Option<i64>,
|
||||||
pub releases: Vec<ReleaseChunk>,
|
pub releases: Vec<ReleaseChunk>,
|
||||||
pub samples: Vec<(ReleaseChunk, &'static str, String)>,
|
pub samples: Vec<(ReleaseChunk, &'static str, String)>,
|
||||||
pub candidates: Vec<Candidate>,
|
pub candidates: Vec<Candidate>,
|
||||||
@@ -277,6 +280,16 @@ fn first_track(text: &str) -> Option<usize> {
|
|||||||
None
|
None
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// postedSpan: posting times without the earliest and latest 1% (floor(n / 100) at each end).
|
||||||
|
pub fn posted_span(mut dates: Vec<i64>) -> (Option<i64>, Option<i64>) {
|
||||||
|
if dates.is_empty() {
|
||||||
|
return (None, None);
|
||||||
|
}
|
||||||
|
dates.sort_unstable();
|
||||||
|
let k = dates.len() / 100;
|
||||||
|
(Some(dates[k]), Some(dates[dates.len() - 1 - k]))
|
||||||
|
}
|
||||||
|
|
||||||
/// nameBesideQuotes (lib/headers/subject.ts): a file name elsewhere when the quoted text has no extension.
|
/// nameBesideQuotes (lib/headers/subject.ts): a file name elsewhere when the quoted text has no extension.
|
||||||
pub fn name_beside_quotes(quoted: &str, before_counter: &str) -> Option<String> {
|
pub fn name_beside_quotes(quoted: &str, before_counter: &str) -> Option<String> {
|
||||||
if has_extension(quoted) {
|
if has_extension(quoted) {
|
||||||
@@ -633,6 +646,8 @@ pub struct RangeIndexer {
|
|||||||
count_order: Vec<(&'static str, String)>,
|
count_order: Vec<(&'static str, String)>,
|
||||||
completed: FxHashMap<String, Vec<Vec<Seg>>>,
|
completed: FxHashMap<String, Vec<Vec<Seg>>>,
|
||||||
completed_order: Vec<String>,
|
completed_order: Vec<String>,
|
||||||
|
/// Posting times of this batch's segments (posted_span at take).
|
||||||
|
dates: Vec<i64>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl RangeIndexer {
|
impl RangeIndexer {
|
||||||
@@ -653,6 +668,7 @@ impl RangeIndexer {
|
|||||||
count_order: Vec::new(),
|
count_order: Vec::new(),
|
||||||
completed: FxHashMap::default(),
|
completed: FxHashMap::default(),
|
||||||
completed_order: Vec::new(),
|
completed_order: Vec::new(),
|
||||||
|
dates: Vec::new(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -683,6 +699,7 @@ impl RangeIndexer {
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
self.batch.headers += headers;
|
self.batch.headers += headers;
|
||||||
|
self.dates.extend(segments.iter().filter_map(|s| s.posted_at));
|
||||||
for owner in crossposts {
|
for owner in crossposts {
|
||||||
self.count("crosspost", owner, 1, 0);
|
self.count("crosspost", owner, 1, 0);
|
||||||
}
|
}
|
||||||
@@ -725,6 +742,7 @@ impl RangeIndexer {
|
|||||||
self.scanner.files.clear();
|
self.scanner.files.clear();
|
||||||
let next = self.batch.next;
|
let next = self.batch.next;
|
||||||
let mut out = std::mem::replace(&mut self.batch, Batch { from: next, next, ..Default::default() });
|
let mut out = std::mem::replace(&mut self.batch, Batch { from: next, next, ..Default::default() });
|
||||||
|
(out.posted_low, out.posted_high) = posted_span(std::mem::take(&mut self.dates));
|
||||||
for id in self.count_order.drain(..) {
|
for id in self.count_order.drain(..) {
|
||||||
let (articles, files) = self.counts.remove(&id).unwrap();
|
let (articles, files) = self.counts.remove(&id).unwrap();
|
||||||
out.counts.push((id.0, id.1, articles, files));
|
out.counts.push((id.0, id.1, articles, files));
|
||||||
@@ -1015,3 +1033,18 @@ mod beside_quotes_tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod posted_span_tests {
|
||||||
|
use super::posted_span;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn without_the_earliest_and_latest_percent() {
|
||||||
|
assert_eq!(posted_span(vec![]), (None, None));
|
||||||
|
let mut dates: Vec<i64> = (0..200).map(|i| 1_000_000 + i).collect();
|
||||||
|
dates.push(5);
|
||||||
|
dates.push(9_999_999_999);
|
||||||
|
// 202 dates: 2 trimmed at each end, so the two wild ones go (postedSpan).
|
||||||
|
assert_eq!(posted_span(dates), (Some(1_000_001), Some(1_000_198)));
|
||||||
|
}
|
||||||
|
}
|
||||||
+5
-1
@@ -96,7 +96,11 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
|
|||||||
out.write_all(line.as_bytes()).unwrap();
|
out.write_all(line.as_bytes()).unwrap();
|
||||||
line.clear();
|
line.clear();
|
||||||
};
|
};
|
||||||
line.push_str(&format!("{{\"t\":\"b\",\"from\":{},\"next\":{},\"headers\":{}}}", batch.from, batch.next, batch.headers));
|
let opt = |t: Option<i64>| t.map_or("null".to_string(), |t| t.to_string());
|
||||||
|
line.push_str(&format!(
|
||||||
|
"{{\"t\":\"b\",\"from\":{},\"next\":{},\"headers\":{},\"postedLow\":{},\"postedHigh\":{}}}",
|
||||||
|
batch.from, batch.next, batch.headers, opt(batch.posted_low), opt(batch.posted_high)
|
||||||
|
));
|
||||||
emit(&mut line);
|
emit(&mut line);
|
||||||
let release = |r: &indexer::ReleaseChunk, kind: &str, line: &mut String| {
|
let release = |r: &indexer::ReleaseChunk, kind: &str, line: &mut String| {
|
||||||
line.push_str(&format!("{{\"t\":\"{kind}\",\"upload\":{},\"key\":", r.upload));
|
line.push_str(&format!("{{\"t\":\"{kind}\",\"upload\":{},\"key\":", r.upload));
|
||||||
|
|||||||
@@ -74,10 +74,13 @@ struct WireFragment<'a> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Serialize)]
|
#[derive(Serialize)]
|
||||||
|
#[serde(rename_all = "camelCase")]
|
||||||
struct WireBatch<'a> {
|
struct WireBatch<'a> {
|
||||||
from: i64,
|
from: i64,
|
||||||
next: i64,
|
next: i64,
|
||||||
headers: u64,
|
headers: u64,
|
||||||
|
posted_low: Option<i64>,
|
||||||
|
posted_high: Option<i64>,
|
||||||
releases: Vec<WireRelease<'a>>,
|
releases: Vec<WireRelease<'a>>,
|
||||||
samples: Vec<WireRelease<'a>>,
|
samples: Vec<WireRelease<'a>>,
|
||||||
candidates: Vec<WireCandidate<'a>>,
|
candidates: Vec<WireCandidate<'a>>,
|
||||||
@@ -134,6 +137,8 @@ pub fn to_json(batch: &Batch) -> Vec<u8> {
|
|||||||
from: batch.from,
|
from: batch.from,
|
||||||
next: batch.next,
|
next: batch.next,
|
||||||
headers: batch.headers,
|
headers: batch.headers,
|
||||||
|
posted_low: batch.posted_low,
|
||||||
|
posted_high: batch.posted_high,
|
||||||
releases: batch.releases.iter().map(|r| release(r, None)).collect(),
|
releases: batch.releases.iter().map(|r| release(r, None)).collect(),
|
||||||
samples: batch.samples.iter().map(|(r, rule, key)| release(r, Some((rule, key.as_str())))).collect(),
|
samples: batch.samples.iter().map(|(r, rule, key)| release(r, Some((rule, key.as_str())))).collect(),
|
||||||
candidates: batch.candidates.iter().map(candidate).collect(),
|
candidates: batch.candidates.iter().map(candidate).collect(),
|
||||||
|
|||||||
+1
-1
@@ -1,4 +1,4 @@
|
|||||||
export { RangeIndexer, chunkOf, candidatesOf, releaseKeyOf, uploadsOf, UPLOAD_GAP_MS, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js'
|
export { RangeIndexer, postedSpan, chunkOf, candidatesOf, releaseKeyOf, uploadsOf, UPLOAD_GAP_MS, type EdgeBatch, type ReleaseChunk, type DroppedChunk, type ProbeCandidate, type CandidateKind, type RuleCount, type Fragment, type RangeIndexerOptions } from './range_indexer.js'
|
||||||
export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js'
|
export { encodeFrame, decodeFrames, isKeepalive, KEEPALIVE, type Frame } from './stream.js'
|
||||||
export { toWire, fromWire, type WireBatch } from './wire.js'
|
export { toWire, fromWire, type WireBatch } from './wire.js'
|
||||||
export type { Lease } from './lease.js'
|
export type { Lease } from './lease.js'
|
||||||
|
|||||||
@@ -77,6 +77,13 @@ export interface EdgeBatch {
|
|||||||
/** First article number not covered yet. */
|
/** First article number not covered yet. */
|
||||||
next: number
|
next: number
|
||||||
headers: number
|
headers: number
|
||||||
|
/**
|
||||||
|
* The batch's posting times without its earliest and latest 1% (ms; null without a dated article): the range's
|
||||||
|
* time span is their min and max (headers dated years off don't stretch it). The settler's final completeness
|
||||||
|
* reads it (LOG.md 2026-10-16 (67)).
|
||||||
|
*/
|
||||||
|
postedLow: number | null
|
||||||
|
postedHigh: number | null
|
||||||
releases: ReleaseChunk[]
|
releases: ReleaseChunk[]
|
||||||
samples: DroppedChunk[]
|
samples: DroppedChunk[]
|
||||||
candidates: ProbeCandidate[]
|
candidates: ProbeCandidate[]
|
||||||
@@ -115,6 +122,8 @@ export class RangeIndexer {
|
|||||||
private aggregator = new FileAggregator<EdgeSegment>(UPLOAD_GAP_MS)
|
private aggregator = new FileAggregator<EdgeSegment>(UPLOAD_GAP_MS)
|
||||||
private scanner = new SegmentScanner()
|
private scanner = new SegmentScanner()
|
||||||
private segments: EdgeSegment[] = []
|
private segments: EdgeSegment[] = []
|
||||||
|
/** Posting times of this batch's segments (postedSpan at take). */
|
||||||
|
private dates: number[] = []
|
||||||
private rules: RuleSet
|
private rules: RuleSet
|
||||||
private batch: EdgeBatch
|
private batch: EdgeBatch
|
||||||
private counts = new Map<string, RuleCount>()
|
private counts = new Map<string, RuleCount>()
|
||||||
@@ -155,8 +164,10 @@ export class RangeIndexer {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
const segment = this.scanner.segment(articleNumber, subject, from, date, messageId, bytes, xref, group)
|
const segment = this.scanner.segment(articleNumber, subject, from, date, messageId, bytes, xref, group)
|
||||||
if (segment) this.segments.push(segment)
|
if (segment) {
|
||||||
else this.count('obfuscated', '', 1, 0)
|
this.segments.push(segment)
|
||||||
|
if (Number.isFinite(segment.postedAt)) this.dates.push(segment.postedAt)
|
||||||
|
} else this.count('obfuscated', '', 1, 0)
|
||||||
}
|
}
|
||||||
|
|
||||||
private endChunk(next: number) {
|
private endChunk(next: number) {
|
||||||
@@ -180,6 +191,8 @@ export class RangeIndexer {
|
|||||||
const out = this.batch
|
const out = this.batch
|
||||||
out.counts = [...this.counts.values()]
|
out.counts = [...this.counts.values()]
|
||||||
this.counts.clear()
|
this.counts.clear()
|
||||||
|
;[out.postedLow, out.postedHigh] = postedSpan(this.dates)
|
||||||
|
this.dates = []
|
||||||
this.batch = this.empty(out.next)
|
this.batch = this.empty(out.next)
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
@@ -290,10 +303,18 @@ export class RangeIndexer {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private empty(from: number): EdgeBatch {
|
private empty(from: number): EdgeBatch {
|
||||||
return { from, next: from, headers: 0, releases: [], samples: [], candidates: [], counts: [], fragments: [] }
|
return { from, next: from, headers: 0, postedLow: null, postedHigh: null, releases: [], samples: [], candidates: [], counts: [], fragments: [] }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Posting times without the earliest and latest 1% (floor(n / 100) at each end); nulls without any. */
|
||||||
|
export function postedSpan(dates: number[]): [number | null, number | null] {
|
||||||
|
if (dates.length === 0) return [null, null]
|
||||||
|
const sorted = Float64Array.from(dates).sort()
|
||||||
|
const k = Math.floor(sorted.length / 100)
|
||||||
|
return [sorted[k], sorted[sorted.length - 1 - k]]
|
||||||
|
}
|
||||||
|
|
||||||
/** The release key; a file without a recognisable release becomes a release of its own. */
|
/** The release key; a file without a recognisable release becomes a release of its own. */
|
||||||
export function releaseKeyOf(s: EdgeSegment): string {
|
export function releaseKeyOf(s: EdgeSegment): string {
|
||||||
return s.file.release?.key ?? hash('sha1', `file\n${s.fileKey}`)
|
return s.file.release?.key ?? hash('sha1', `file\n${s.fileKey}`)
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
import { test } from '@japa/runner'
|
||||||
|
import db from '@adonisjs/lucid/services/db'
|
||||||
|
import testUtils from '@adonisjs/core/services/test_utils'
|
||||||
|
import BatchStore from '#services/batch_store'
|
||||||
|
import RangeBlocks from '#services/range_blocks'
|
||||||
|
import SearchService from '#services/search_service'
|
||||||
|
import { RangeIndexer, postedSpan } from '#lib/edge/index'
|
||||||
|
import { DEFAULT_RULES } from '#lib/rules/index'
|
||||||
|
import type { Overview } from '#lib/nntp/index'
|
||||||
|
|
||||||
|
const GROUP = 'alt.binaries.final'
|
||||||
|
const at = (iso: string) => new Date(iso)
|
||||||
|
|
||||||
|
test.group('Final completeness', (group) => {
|
||||||
|
group.each.setup(() => testUtils.db().withGlobalTransaction())
|
||||||
|
|
||||||
|
test("a batch's posting times, without the earliest and latest 1%, become the range's span", async ({ assert }) => {
|
||||||
|
assert.deepEqual(postedSpan([]), [null, null])
|
||||||
|
const dates = Array.from({ length: 200 }, (_, i) => 1_000_000 + i)
|
||||||
|
dates.push(5, 9_999_999_999)
|
||||||
|
// 202 dates: 2 trimmed at each end, so the two wild ones go.
|
||||||
|
assert.deepEqual(postedSpan(dates), [1_000_001, 1_000_198])
|
||||||
|
|
||||||
|
const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
|
||||||
|
const [node] = await db.table('nodes').insert({ name: 'final-node' }).returning('id')
|
||||||
|
const [range] = await db
|
||||||
|
.table('ranges')
|
||||||
|
.insert({ group_id: g.id, start_id: 1, end_id: 1000, next_id: 1, status: 'leased', node_id: node.id, rule_version: 1 })
|
||||||
|
.returning('id')
|
||||||
|
const overviews = Array.from({ length: 10 }, (_, i) => {
|
||||||
|
const date = new Date(Date.parse('2026-01-01T10:00:00Z') + i * 60_000).toUTCString().replace('GMT', '+0000')
|
||||||
|
return [1 + i, { 'subject': `[1/1] - "Some.Thing.2026.rar" yEnc (${i + 1}/10)`, 'from': 'p <p@x>', 'date': date, 'message-id': `<f${i}@x>`, ':bytes': '1000', 'xref': `x ${GROUP}:${1 + i}` }] as Overview
|
||||||
|
})
|
||||||
|
const ix = new RangeIndexer({ group: GROUP, start: 1, end: 1000, rules: { ...DEFAULT_RULES, samplePercent: 0, droppedSignatures: [], priorities: { [GROUP]: 1 } } })
|
||||||
|
ix.add(overviews, 11)
|
||||||
|
await BatchStore.store(Number(range.id), Number(node.id), ix.take())
|
||||||
|
const stored = await db.from('ranges').where('id', range.id).firstOrFail()
|
||||||
|
assert.equal(new Date(stored.posted_low).toISOString(), '2026-01-01T10:00:00.000Z')
|
||||||
|
assert.equal(new Date(stored.posted_high).toISOString(), '2026-01-01T10:09:00.000Z')
|
||||||
|
})
|
||||||
|
|
||||||
|
test('a release is final once one block of done ranges covers its posting times and 6 hours either side', async ({ assert }) => {
|
||||||
|
const [g] = await db.table('groups').insert({ name: GROUP, priority: 1 }).returning('id')
|
||||||
|
const range = (start: number, status: string, low: string | null, high: string | null) =>
|
||||||
|
db.table('ranges').insert({ group_id: g.id, start_id: start, end_id: start + 99_999, next_id: start, status, posted_low: low && at(low), posted_high: high && at(high) })
|
||||||
|
await range(1, 'done', '2026-01-01T00:00:00Z', '2026-01-01T12:00:00Z')
|
||||||
|
await range(100_001, 'done', '2026-01-01T12:00:00Z', '2026-01-01T23:00:00Z')
|
||||||
|
await range(200_001, 'planned', null, null)
|
||||||
|
await range(300_001, 'done', '2026-01-02T10:00:00Z', '2026-01-02T20:00:00Z')
|
||||||
|
assert.equal(await RangeBlocks.rebuild(), 2)
|
||||||
|
|
||||||
|
let n = 0
|
||||||
|
const release = async (first: number, posted: string, extra: Record<string, unknown> = {}) => {
|
||||||
|
n++
|
||||||
|
const key = `fc${n}`.padEnd(40, '0')
|
||||||
|
const [row] = await db
|
||||||
|
.table('releases')
|
||||||
|
.insert({ release_key: key, identity_key: key, name: `Final.Test.${n}`, source: 'stem', groups: '[]', group_id: g.id, files: 3, file_total: 5, complete: false, first_article: first, last_article: first + 10, posted_at: at(posted), last_posted_at: at(posted), readable: true, settled_at: new Date(), ...extra })
|
||||||
|
.returning('id')
|
||||||
|
return Number(row.id)
|
||||||
|
}
|
||||||
|
const final = async (id: number) => {
|
||||||
|
const r = await db.from('releases').where('id', id).firstOrFail()
|
||||||
|
const { rows } = await db.rawQuery(`SELECT release_final(?, ?, ?, ?, ?) AS final`, [r.group_id, r.first_article, r.last_article, r.posted_at, r.last_posted_at])
|
||||||
|
return rows[0].final as boolean
|
||||||
|
}
|
||||||
|
|
||||||
|
// Mid-block: final either way.
|
||||||
|
const middle = await release(90_000, '2026-01-01T11:00:00Z')
|
||||||
|
// In the first range: nothing older to wait for, although 6 hours before it isn't covered.
|
||||||
|
const first = await release(1_000, '2026-01-01T02:00:00Z')
|
||||||
|
// Near the block's end, the next range unread: its missing files may be there.
|
||||||
|
const edge = await release(195_000, '2026-01-01T22:00:00Z')
|
||||||
|
// In the island after the gap: the gap is before it.
|
||||||
|
const island = await release(310_000, '2026-01-02T12:00:00Z')
|
||||||
|
assert.isTrue(await final(middle))
|
||||||
|
assert.isTrue(await final(first))
|
||||||
|
assert.isFalse(await final(edge))
|
||||||
|
assert.isFalse(await final(island))
|
||||||
|
|
||||||
|
// Search: an incomplete release that isn't final isn't left out as incomplete; the API says which is final.
|
||||||
|
const shown = await SearchService.search({ q: 'Final.Test', notIncomplete: true })
|
||||||
|
assert.sameMembers(shown.releases.map((r) => r.id), [edge, island])
|
||||||
|
assert.isTrue(shown.releases.every((r) => r.final === false))
|
||||||
|
|
||||||
|
// The gap read: one block, the edge release final (the island's 12:00 + 6 h is inside 20:00).
|
||||||
|
await db.from('ranges').where('start_id', 200_001).update({ status: 'done', posted_low: at('2026-01-01T23:00:00Z'), posted_high: at('2026-01-02T10:00:00Z') })
|
||||||
|
assert.equal(await RangeBlocks.rebuild(), 1)
|
||||||
|
assert.isTrue(await final(edge))
|
||||||
|
assert.isTrue(await final(island))
|
||||||
|
assert.equal((await SearchService.search({ q: 'Final.Test', notIncomplete: true })).total, 0)
|
||||||
|
})
|
||||||
|
})
|
||||||
Reference in new issue
Block a user