The Rust worker is the product: edge/, built and checked in CI
User: Rust is the product (option a). tools/edge-bench-rs moves to edge/
(crate edge; binaries edge-node, the worker, and edge-bench, the replay).
CI's new job edge runs cargo test and edge/check-fixtures.sh: both
RangeIndexers on three committed captures (edge/fixtures, 25k overview
lines each of a.b.models, a.b.ath and a.b.boneless in 5 chunks, 3.5 MB),
compared field by field with check.ts; deploy waits for it. The
TypeScript RangeIndexer stays as check.ts's reference. README: what it
is now; the benchmark findings as history.
Random names: looksRandom missed long tokens of capitals and digits
("51294P1CUKR2R35X", release 74539068, readable and a Newznab title).
12+ capitals and digits switching 6+ times between them are random
(COD4MW2REMASTERED switches 4 times, BEATLES1964REMASTER twice); both
implementations, identical on the fixtures.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Daan MeijerandClaude Opus 5.5 committed
1 parent
48f4e982f9
commit
5109481d67
38 files changed
+138
-54
No files matched your search
@@ -0,0 +1,562 @@
|
||||
//! The edge worker in Rust (REBUILD.md, part 2; commands/edge_node.ts): leases
|
||||
//! ranges from the coordinator, reads their XOVER chunks from its stream,
|
||||
//! indexes them (the same RangeIndexer as edge-bench, kept identical to
|
||||
//! the TypeScript by check.ts) and posts the batches. One range per thread.
|
||||
//!
|
||||
//! COORDINATOR_URL=… NODE_TOKEN=… edge-node [--threads N] [--batch-articles 500000] [--ranges N] [--max-mbit M]
|
||||
//!
|
||||
//! SIGTERM/SIGINT: each thread gives its range back after the current chunk
|
||||
//! (the next worker resumes at the last stored batch).
|
||||
|
||||
use edge::indexer::{Batch, RangeIndexer};
|
||||
use edge::names::Rules;
|
||||
use edge::wire;
|
||||
use flate2::write::GzEncoder;
|
||||
use serde::Deserialize;
|
||||
use std::collections::HashMap;
|
||||
use std::io::{Read, Write};
|
||||
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
|
||||
use std::sync::mpsc::{sync_channel, Receiver};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
#[global_allocator]
|
||||
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
|
||||
|
||||
/// A batch is posted once this many files (completed, or cut at an edge) wait, whatever its articles (commands/edge_node.ts).
|
||||
const MAX_PENDING_FILES: usize = 100_000;
|
||||
const RETRIES: u32 = 8;
|
||||
const IDLE: Duration = Duration::from_secs(30);
|
||||
/// Without a post for this long, the range's lease is kept alive by a heartbeat.
|
||||
const HEARTBEAT: Duration = Duration::from_secs(60);
|
||||
|
||||
fn log(level: &str, message: &str) {
|
||||
println!("[ {level} ] {message}");
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct LeaseJson {
|
||||
range: RangeJson,
|
||||
rules: RulesJson,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RangeJson {
|
||||
id: i64,
|
||||
group: String,
|
||||
start: i64,
|
||||
end: i64,
|
||||
next: i64,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct RulesJson {
|
||||
version: i64,
|
||||
priorities: HashMap<String, u32>,
|
||||
#[serde(default)]
|
||||
dropped_signatures: Vec<String>,
|
||||
#[serde(default)]
|
||||
dropped_tools: Option<Vec<String>>,
|
||||
#[serde(default)]
|
||||
per_article_max_parts: Option<u32>,
|
||||
#[serde(default)]
|
||||
sample_percent: Option<f64>,
|
||||
#[serde(default)]
|
||||
drop_executables: Option<bool>,
|
||||
}
|
||||
|
||||
impl RulesJson {
|
||||
fn rules(self) -> Rules {
|
||||
let mut rules = Rules::defaults(self.priorities.into_iter().collect());
|
||||
rules.version = self.version;
|
||||
rules.dropped_signatures = self.dropped_signatures.into_iter().collect();
|
||||
if let Some(tools) = self.dropped_tools {
|
||||
rules.dropped_tools = tools.into_iter().collect();
|
||||
}
|
||||
if let Some(n) = self.per_article_max_parts {
|
||||
rules.per_article_max_parts = n;
|
||||
}
|
||||
if let Some(p) = self.sample_percent {
|
||||
rules.sample_percent = p;
|
||||
}
|
||||
if let Some(e) = self.drop_executables {
|
||||
rules.drop_executables = e;
|
||||
}
|
||||
rules
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct StoredJson {
|
||||
releases: u64,
|
||||
candidates: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
enum ApiError {
|
||||
/// 409: the range is no longer this node's (or the batch was stale): stop it, don't give it back.
|
||||
LeaseLost(String),
|
||||
/// 413: the batch was over the server's body limit; the range is tried again in smaller batches.
|
||||
TooLarge(String),
|
||||
Other(String),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for ApiError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
|
||||
match self {
|
||||
ApiError::LeaseLost(m) | ApiError::TooLarge(m) | ApiError::Other(m) => f.write_str(m),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The coordinator's node API, with retries for network trouble and 5xx (as the TypeScript Api).
|
||||
struct Api {
|
||||
base: String,
|
||||
token: String,
|
||||
agent: ureq::Agent,
|
||||
}
|
||||
|
||||
impl Api {
|
||||
fn new(base: &str, token: &str) -> Self {
|
||||
let agent: ureq::Agent = ureq::Agent::config_builder()
|
||||
.http_status_as_error(false)
|
||||
.timeout_connect(Some(Duration::from_secs(30)))
|
||||
.timeout_recv_response(Some(Duration::from_secs(600)))
|
||||
.build()
|
||||
.into();
|
||||
Api { base: base.trim_end_matches('/').to_string(), token: token.to_string(), agent }
|
||||
}
|
||||
|
||||
/// A request, retried; `body`: JSON bytes (gzipped when `gzip`).
|
||||
fn request(&self, method: &str, path: &str, body: Option<&[u8]>, gzip: bool, accept: &str) -> Result<ureq::http::Response<ureq::Body>, ApiError> {
|
||||
let url = format!("{}/api/v1/node/{path}", self.base);
|
||||
let auth = format!("Bearer {}", self.token);
|
||||
for attempt in 1..=RETRIES {
|
||||
let result = match method {
|
||||
"GET" => self.agent.get(&url).header("Authorization", &auth).header("Accept", accept).call(),
|
||||
_ => {
|
||||
let mut request = self.agent.post(&url).header("Authorization", &auth).header("Accept", accept).header("Content-Type", "application/json");
|
||||
if gzip {
|
||||
request = request.header("Content-Encoding", "gzip");
|
||||
}
|
||||
request.send(body.unwrap_or(b"{}"))
|
||||
}
|
||||
};
|
||||
match result {
|
||||
Ok(mut response) => {
|
||||
let status = response.status().as_u16();
|
||||
if status == 409 {
|
||||
let text = response.body_mut().read_to_string().unwrap_or_default();
|
||||
let message = serde_json::from_str::<serde_json::Value>(&text).ok().and_then(|v| v["error"].as_str().map(str::to_string)).unwrap_or_else(|| "lease lost".into());
|
||||
return Err(ApiError::LeaseLost(message));
|
||||
}
|
||||
if status < 500 {
|
||||
return Ok(response);
|
||||
}
|
||||
if attempt == RETRIES {
|
||||
return Err(ApiError::Other(format!("{path}: HTTP {status}")));
|
||||
}
|
||||
}
|
||||
Err(error) => {
|
||||
if attempt == RETRIES {
|
||||
return Err(ApiError::Other(format!("{path}: {error}")));
|
||||
}
|
||||
}
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(60_000.min(1000 * 2u64.pow(attempt))));
|
||||
}
|
||||
unreachable!()
|
||||
}
|
||||
|
||||
fn lease(&self) -> Result<Option<LeaseJson>, ApiError> {
|
||||
let mut response = self.request("POST", "lease", None, false, "application/json")?;
|
||||
match response.status().as_u16() {
|
||||
204 => Ok(None),
|
||||
200 => {
|
||||
let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("lease: {e}")))?;
|
||||
serde_json::from_str(&text).map(Some).map_err(|e| ApiError::Other(format!("lease: {e}")))
|
||||
}
|
||||
status => Err(ApiError::Other(format!("lease: HTTP {status}"))),
|
||||
}
|
||||
}
|
||||
|
||||
fn stream(&self, range: i64) -> Result<Box<dyn Read + Send>, ApiError> {
|
||||
let response = self.request("GET", &format!("ranges/{range}/stream"), None, false, "application/octet-stream")?;
|
||||
if response.status().as_u16() != 200 {
|
||||
return Err(ApiError::Other(format!("stream: HTTP {}", response.status())));
|
||||
}
|
||||
Ok(Box::new(response.into_body().into_reader()))
|
||||
}
|
||||
|
||||
/// Posts a batch (gzipped JSON): the releases and candidates stored, and the bytes posted.
|
||||
fn batch(&self, range: i64, batch: &Batch) -> Result<(u64, u64, usize), ApiError> {
|
||||
let json = wire::to_json(batch);
|
||||
let mut gz = GzEncoder::new(Vec::with_capacity(json.len() / 3), flate2::Compression::new(1));
|
||||
gz.write_all(&json).unwrap();
|
||||
let body = gz.finish().unwrap();
|
||||
let mut response = self.request("POST", &format!("ranges/{range}/batches"), Some(&body), true, "application/json")?;
|
||||
if response.status().as_u16() == 413 {
|
||||
return Err(ApiError::TooLarge(format!("batch: HTTP {} ({} MB gzipped, {} MB of JSON)", response.status(), body.len() / 1_000_000, json.len() / 1_000_000)));
|
||||
}
|
||||
if response.status().as_u16() != 200 {
|
||||
return Err(ApiError::Other(format!("batch: HTTP {}", response.status())));
|
||||
}
|
||||
let text = response.body_mut().read_to_string().map_err(|e| ApiError::Other(format!("batch: {e}")))?;
|
||||
let stored: StoredJson = serde_json::from_str(&text).map_err(|e| ApiError::Other(format!("batch: {e}")))?;
|
||||
Ok((stored.releases, stored.candidates, body.len()))
|
||||
}
|
||||
|
||||
fn call(&self, range: i64, action: &str) -> Result<(), ApiError> {
|
||||
let response = self.request("POST", &format!("ranges/{range}/{action}"), None, false, "application/json")?;
|
||||
match response.status().as_u16() {
|
||||
200..=299 => Ok(()),
|
||||
status => Err(ApiError::Other(format!("{action}: HTTP {status}"))),
|
||||
}
|
||||
}
|
||||
|
||||
fn corrupt(&self, range: i64, from: i64) -> Result<(), ApiError> {
|
||||
let body = format!("{{\"from\":{from}}}");
|
||||
let response = self.request("POST", &format!("ranges/{range}/corrupt"), Some(body.as_bytes()), false, "application/json")?;
|
||||
match response.status().as_u16() {
|
||||
200..=299 => Ok(()),
|
||||
status => Err(ApiError::Other(format!("corrupt: HTTP {status}"))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One frame of the range's stream: a chunk inflated (or why it didn't), or a keep-alive.
|
||||
enum Frame {
|
||||
Chunk { from: i64, to: i64, text: Result<Vec<u8>, String> },
|
||||
KeepAlive,
|
||||
End,
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
/// Reads frames (8 B from, 8 B to, 4 B length, data) and inflates them, one frame ahead of the indexer.
|
||||
fn read_frames(mut stream: Box<dyn Read + Send>, bytes_per_second: Option<f64>) -> Receiver<Frame> {
|
||||
let (tx, rx) = sync_channel::<Frame>(2);
|
||||
std::thread::spawn(move || {
|
||||
let mut decompressor = libdeflater::Decompressor::new();
|
||||
let started = Instant::now();
|
||||
let mut read_bytes = 0f64;
|
||||
loop {
|
||||
let mut head = [0u8; 20];
|
||||
match read_full(&mut stream, &mut head) {
|
||||
Ok(0) => {
|
||||
let _ = tx.send(Frame::End);
|
||||
return;
|
||||
}
|
||||
Ok(20) => {}
|
||||
Ok(n) => {
|
||||
let _ = tx.send(Frame::Failed(format!("stream ended inside a frame ({n} bytes)")));
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = tx.send(Frame::Failed(format!("stream: {e}")));
|
||||
return;
|
||||
}
|
||||
}
|
||||
let from = u64::from_be_bytes(head[0..8].try_into().unwrap()) as i64;
|
||||
let to = u64::from_be_bytes(head[8..16].try_into().unwrap()) as i64;
|
||||
let length = u32::from_be_bytes(head[16..20].try_into().unwrap()) as usize;
|
||||
let mut data = vec![0u8; length];
|
||||
match read_full(&mut stream, &mut data) {
|
||||
Ok(n) if n == length => {}
|
||||
Ok(n) => {
|
||||
let _ = tx.send(Frame::Failed(format!("stream ended inside a frame ({} of {length} bytes)", n)));
|
||||
return;
|
||||
}
|
||||
Err(e) => {
|
||||
let _ = tx.send(Frame::Failed(format!("stream: {e}")));
|
||||
return;
|
||||
}
|
||||
}
|
||||
// --max-mbit: read no faster than the cap (TCP holds the coordinator back).
|
||||
if let Some(rate) = bytes_per_second {
|
||||
read_bytes += (20 + length) as f64;
|
||||
let due = Duration::from_secs_f64(((read_bytes - rate) / rate).max(0.0));
|
||||
let elapsed = started.elapsed();
|
||||
if due > elapsed {
|
||||
std::thread::sleep(due - elapsed);
|
||||
}
|
||||
}
|
||||
let frame = if length == 0 && from == 0 {
|
||||
Frame::KeepAlive
|
||||
} else {
|
||||
Frame::Chunk { from, to, text: inflate(&data, &mut decompressor) }
|
||||
};
|
||||
if tx.send(frame).is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
});
|
||||
rx
|
||||
}
|
||||
|
||||
/// Reads until `buf` is full or the stream ends: the bytes read.
|
||||
fn read_full(stream: &mut dyn Read, buf: &mut [u8]) -> std::io::Result<usize> {
|
||||
let mut at = 0;
|
||||
while at < buf.len() {
|
||||
match stream.read(&mut buf[at..]) {
|
||||
Ok(0) => break,
|
||||
Ok(n) => at += n,
|
||||
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
|
||||
Err(e) => return Err(e),
|
||||
}
|
||||
}
|
||||
Ok(at)
|
||||
}
|
||||
|
||||
/// A chunk as the news server sent it (zlib; gzip too), whole.
|
||||
fn inflate(raw: &[u8], d: &mut libdeflater::Decompressor) -> Result<Vec<u8>, String> {
|
||||
let mut out = vec![0u8; raw.len() * 4 + 4096];
|
||||
loop {
|
||||
let result = if raw.starts_with(&[0x1f, 0x8b]) { d.gzip_decompress(raw, &mut out) } else { d.zlib_decompress(raw, &mut out) };
|
||||
match result {
|
||||
Ok(n) => {
|
||||
out.truncate(n);
|
||||
return Ok(out);
|
||||
}
|
||||
Err(libdeflater::DecompressionError::InsufficientSpace) => out.resize(out.len() * 2, 0),
|
||||
Err(e) => return Err(e.to_string()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
struct Config {
|
||||
batch_articles: i64,
|
||||
bytes_per_second: Option<f64>,
|
||||
/// Ranges whose batch was too large (413), and how much smaller their batches are now (halved each time).
|
||||
smaller: Mutex<HashMap<i64, i64>>,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Articles and pending files a batch of this range is posted at.
|
||||
fn limits(&self, range: i64) -> (i64, usize) {
|
||||
let divisor = self.smaller.lock().unwrap().get(&range).copied().unwrap_or(1);
|
||||
(self.batch_articles / divisor, MAX_PENDING_FILES / divisor as usize)
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Totals {
|
||||
headers: u64,
|
||||
releases: u64,
|
||||
candidates: u64,
|
||||
bytes: usize,
|
||||
post_ms: u128,
|
||||
}
|
||||
|
||||
/// One leased range, start to finish (commands/edge_node.ts runRange).
|
||||
fn run_range(api: &Arc<Api>, lease: LeaseJson, config: &Config, stopping: &AtomicBool) {
|
||||
let range = lease.range;
|
||||
let started = Instant::now();
|
||||
let mut indexer = RangeIndexer::new(&range.group, range.start, range.end, range.next, lease.rules.rules(), 5);
|
||||
let totals = Arc::new(Mutex::new(Totals::default()));
|
||||
let post_error: Arc<Mutex<Option<ApiError>>> = Arc::new(Mutex::new(None));
|
||||
let last_contact = Arc::new(AtomicI64::new(0));
|
||||
let since = Instant::now();
|
||||
// Posts in order on their own thread; the indexer goes on meanwhile (one batch may wait).
|
||||
let (post_tx, post_rx) = sync_channel::<Batch>(1);
|
||||
let poster = {
|
||||
let (api, totals, post_error, last_contact) = (api.clone(), totals.clone(), post_error.clone(), last_contact.clone());
|
||||
std::thread::spawn(move || {
|
||||
for batch in post_rx {
|
||||
if post_error.lock().unwrap().is_some() {
|
||||
continue;
|
||||
}
|
||||
let at = Instant::now();
|
||||
match api.batch(range.id, &batch) {
|
||||
Ok((releases, candidates, bytes)) => {
|
||||
let mut t = totals.lock().unwrap();
|
||||
t.post_ms += at.elapsed().as_millis();
|
||||
t.bytes += bytes;
|
||||
t.headers += batch.headers;
|
||||
t.releases += releases;
|
||||
t.candidates += candidates;
|
||||
last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed);
|
||||
}
|
||||
Err(e) => *post_error.lock().unwrap() = Some(e),
|
||||
}
|
||||
}
|
||||
})
|
||||
};
|
||||
let frames = match api.stream(range.id) {
|
||||
Ok(stream) => read_frames(stream, config.bytes_per_second),
|
||||
Err(e) => {
|
||||
log("error", &format!("Range {} failed: {e}; giving it back.", range.id));
|
||||
let _ = api.call(range.id, "release");
|
||||
return;
|
||||
}
|
||||
};
|
||||
let (batch_articles, max_pending) = config.limits(range.id);
|
||||
let mut batch_from = range.next;
|
||||
let mut reached_end = false;
|
||||
let mut corrupt: Option<(i64, i64, String)> = None;
|
||||
let mut failure: Option<String> = None;
|
||||
for frame in frames.iter() {
|
||||
if stopping.load(Ordering::Relaxed) || post_error.lock().unwrap().is_some() {
|
||||
break;
|
||||
}
|
||||
// Keep the lease while no batch goes out (slow ranges wait for the spool).
|
||||
if since.elapsed().as_millis() as i64 - last_contact.load(Ordering::Relaxed) > HEARTBEAT.as_millis() as i64 {
|
||||
last_contact.store(since.elapsed().as_millis() as i64, Ordering::Relaxed);
|
||||
let _ = api.call(range.id, "heartbeat");
|
||||
}
|
||||
match frame {
|
||||
Frame::KeepAlive => continue,
|
||||
Frame::End => break,
|
||||
Frame::Failed(e) => {
|
||||
failure = Some(e);
|
||||
break;
|
||||
}
|
||||
Frame::Chunk { from, to, text } => {
|
||||
let data = match text {
|
||||
Ok(data) => data,
|
||||
Err(e) => {
|
||||
corrupt = Some((from, to, e));
|
||||
break;
|
||||
}
|
||||
};
|
||||
indexer.add_text(&String::from_utf8_lossy(&data), to + 1);
|
||||
reached_end = to >= range.end;
|
||||
if reached_end {
|
||||
indexer.finish();
|
||||
}
|
||||
if reached_end || to + 1 - batch_from >= batch_articles || indexer.pending() >= max_pending {
|
||||
if post_tx.send(indexer.take()).is_err() {
|
||||
break;
|
||||
}
|
||||
batch_from = to + 1;
|
||||
}
|
||||
if reached_end {
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(post_tx);
|
||||
let _ = poster.join();
|
||||
let post_error = post_error.lock().unwrap().take();
|
||||
if let Some((from, to, e)) = corrupt {
|
||||
// The server's compressed reply cut short: reported, fetched plain next time.
|
||||
log("warn", &format!("Range {}: chunk {from}–{to} doesn't inflate ({e}); reported, the range goes back", range.id));
|
||||
let _ = api.corrupt(range.id, from);
|
||||
return;
|
||||
}
|
||||
match post_error {
|
||||
Some(ApiError::LeaseLost(m)) => {
|
||||
log("warn", &format!("Range {}: {m}", range.id));
|
||||
return;
|
||||
}
|
||||
Some(ApiError::TooLarge(m)) => {
|
||||
// Batches already stored stay; from the resume point on, half the size (a batch is at least a chunk).
|
||||
let divisor = {
|
||||
let mut smaller = config.smaller.lock().unwrap();
|
||||
let d = smaller.entry(range.id).or_insert(1);
|
||||
*d = (*d * 2).min(64);
|
||||
*d
|
||||
};
|
||||
log("warn", &format!("Range {}: {m}; giving it back, its batches 1/{divisor} the size from now on.", range.id));
|
||||
let _ = api.call(range.id, "release");
|
||||
return;
|
||||
}
|
||||
Some(ApiError::Other(m)) => {
|
||||
log("error", &format!("Range {} failed: {m}; giving it back.", range.id));
|
||||
let _ = api.call(range.id, "release");
|
||||
return;
|
||||
}
|
||||
None => {}
|
||||
}
|
||||
if let Some(e) = failure {
|
||||
log("error", &format!("Range {} failed: {e}; giving it back.", range.id));
|
||||
let _ = api.call(range.id, "release");
|
||||
return;
|
||||
}
|
||||
if !reached_end {
|
||||
log("info", &format!("Range {}: giving it back.", range.id));
|
||||
let _ = api.call(range.id, "release");
|
||||
return;
|
||||
}
|
||||
match api.call(range.id, "finish") {
|
||||
Ok(()) => {
|
||||
let t = totals.lock().unwrap();
|
||||
let seconds = started.elapsed().as_secs_f64();
|
||||
log(
|
||||
"info",
|
||||
&format!(
|
||||
"{} {}–{}: {} headers in {:.0} s ({:.0}/s), {} release chunks, {} probe candidates; posted {:.1} MB in {:.1} s.",
|
||||
range.group,
|
||||
range.start,
|
||||
range.end,
|
||||
t.headers,
|
||||
seconds,
|
||||
t.headers as f64 / seconds,
|
||||
t.releases,
|
||||
t.candidates,
|
||||
t.bytes as f64 / 1e6,
|
||||
t.post_ms as f64 / 1000.0
|
||||
),
|
||||
);
|
||||
}
|
||||
Err(e) => log("warn", &format!("Range {}: {e}", range.id)),
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned();
|
||||
let url = std::env::var("COORDINATOR_URL").expect("COORDINATOR_URL is needed");
|
||||
let token = std::env::var("NODE_TOKEN").expect("NODE_TOKEN is needed");
|
||||
let threads: usize = flag("--threads")
|
||||
.or_else(|| std::env::var("EDGE_PROCESSES").ok())
|
||||
.map(|v| v.parse().unwrap())
|
||||
.filter(|&n| n > 0)
|
||||
.unwrap_or_else(|| std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1));
|
||||
let ranges: Option<usize> = flag("--ranges").map(|v| v.parse().unwrap());
|
||||
let max_mbit: Option<f64> = flag("--max-mbit").or_else(|| std::env::var("EDGE_MAX_MBIT").ok()).map(|v| v.parse().unwrap()).filter(|&m: &f64| m > 0.0);
|
||||
let config = Arc::new(Config {
|
||||
batch_articles: flag("--batch-articles").map(|v| v.parse().unwrap()).unwrap_or(500_000),
|
||||
// The node's cap, shared by its threads.
|
||||
bytes_per_second: max_mbit.map(|m| m * 1e6 / 8.0 / threads as f64),
|
||||
smaller: Mutex::new(HashMap::new()),
|
||||
});
|
||||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
signal_hook::flag::register(signal_hook::consts::SIGTERM, stopping.clone()).unwrap();
|
||||
signal_hook::flag::register(signal_hook::consts::SIGINT, stopping.clone()).unwrap();
|
||||
let api = Arc::new(Api::new(&url, &token));
|
||||
log("info", &format!("Starting {threads} worker threads{}.", max_mbit.map(|m| format!(" (download cap {m} Mbit/s)")).unwrap_or_default()));
|
||||
let workers: Vec<_> = (0..threads)
|
||||
.map(|_| {
|
||||
let (api, config, stopping) = (api.clone(), config.clone(), stopping.clone());
|
||||
std::thread::spawn(move || {
|
||||
let mut done = 0;
|
||||
while !stopping.load(Ordering::Relaxed) && ranges.is_none_or(|n| done < n) {
|
||||
match api.lease() {
|
||||
Ok(Some(lease)) => {
|
||||
run_range(&api, lease, &config, &stopping);
|
||||
done += 1;
|
||||
}
|
||||
Ok(None) if ranges.is_some() => break,
|
||||
Ok(None) => idle(&stopping),
|
||||
Err(e) => {
|
||||
log("warn", &format!("Lease failed: {e}"));
|
||||
idle(&stopping);
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
for worker in workers {
|
||||
let _ = worker.join();
|
||||
}
|
||||
}
|
||||
|
||||
fn idle(stopping: &AtomicBool) {
|
||||
let until = Instant::now() + IDLE;
|
||||
while Instant::now() < until && !stopping.load(Ordering::Relaxed) {
|
||||
std::thread::sleep(Duration::from_millis(500));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
//! lib/chunks/codec.ts encodeChunk, format v2: a version byte, the length of
|
||||
//! the compressed part, zstd(JSON head + text ids), then the nyuu and hex ids
|
||||
//! as raw bytes.
|
||||
|
||||
use crate::js::json_str;
|
||||
|
||||
pub struct ChunkFile<'a> {
|
||||
pub name: &'a str,
|
||||
pub poster: &'a str,
|
||||
pub subject: &'a str,
|
||||
pub posted_at: Option<i64>,
|
||||
pub groups: &'a [String],
|
||||
/// [part, message-id, bytes]
|
||||
pub segments: Vec<(u32, &'a str, i64)>,
|
||||
}
|
||||
|
||||
fn is_nyuu(id: &str, suffix: &str) -> bool {
|
||||
let b = id.as_bytes();
|
||||
if b.len() != 38 + suffix.len() || b[24] != b'-' || !id.ends_with(suffix) {
|
||||
return false;
|
||||
}
|
||||
(0..24).step_by(2).all(|i| b[i].is_ascii_uppercase() && b[i + 1].is_ascii_lowercase()) && b[25..38].iter().all(u8::is_ascii_digit)
|
||||
}
|
||||
|
||||
fn is_hex(id: &str, suffix: &str) -> bool {
|
||||
let b = id.as_bytes();
|
||||
b.len() == 32 + suffix.len() && id.ends_with(suffix) && b[..32].iter().all(|c| c.is_ascii_digit() || (b'a'..=b'f').contains(c))
|
||||
}
|
||||
|
||||
fn pack_letters(id: &[u8], out: &mut Vec<u8>) {
|
||||
let mut acc: u32 = 0;
|
||||
let mut bits = 0;
|
||||
for i in (0..24).step_by(2) {
|
||||
acc = (acc << 10) | ((id[i] - b'A') as u32 * 26 + (id[i + 1] - b'a') as u32);
|
||||
bits += 10;
|
||||
while bits >= 8 {
|
||||
bits -= 8;
|
||||
out.push(((acc >> bits) & 0xff) as u8);
|
||||
}
|
||||
acc &= (1 << bits) - 1;
|
||||
}
|
||||
}
|
||||
|
||||
fn hex_val(c: u8) -> u8 {
|
||||
if c.is_ascii_digit() {
|
||||
c - b'0'
|
||||
} else {
|
||||
c - b'a' + 10
|
||||
}
|
||||
}
|
||||
|
||||
fn push_num(out: &mut String, n: i64) {
|
||||
use std::fmt::Write;
|
||||
write!(out, "{n}").unwrap();
|
||||
}
|
||||
|
||||
pub fn encode_chunk(files: &[ChunkFile]) -> Vec<u8> {
|
||||
let mut head = String::with_capacity(256 * files.len());
|
||||
let mut text: Vec<&str> = Vec::new();
|
||||
let mut raw: Vec<u8> = Vec::new();
|
||||
head.push('[');
|
||||
for (fi, f) in files.iter().enumerate() {
|
||||
let ids = &f.segments;
|
||||
let code = match ids.first() {
|
||||
None => 't',
|
||||
Some(&(_, first, _)) => {
|
||||
let nyuu = first.get(38..).filter(|s| s.starts_with('@'));
|
||||
let hex = first.get(32..).filter(|s| s.starts_with('@'));
|
||||
if nyuu.is_some_and(|suffix| ids.iter().all(|&(_, id, _)| is_nyuu(id, suffix))) {
|
||||
'n'
|
||||
} else if hex.is_some_and(|suffix| ids.iter().all(|&(_, id, _)| is_hex(id, suffix))) {
|
||||
'h'
|
||||
} else {
|
||||
't'
|
||||
}
|
||||
}
|
||||
};
|
||||
if fi > 0 {
|
||||
head.push(',');
|
||||
}
|
||||
head.push_str("{\"n\":");
|
||||
json_str(f.name, &mut head);
|
||||
head.push_str(",\"p\":");
|
||||
json_str(f.poster, &mut head);
|
||||
head.push_str(",\"j\":");
|
||||
json_str(f.subject, &mut head);
|
||||
head.push_str(",\"d\":");
|
||||
match f.posted_at {
|
||||
Some(t) => push_num(&mut head, t),
|
||||
None => head.push_str("null"),
|
||||
}
|
||||
head.push_str(",\"g\":[");
|
||||
for (i, g) in f.groups.iter().enumerate() {
|
||||
if i > 0 {
|
||||
head.push(',');
|
||||
}
|
||||
json_str(g, &mut head);
|
||||
}
|
||||
head.push_str("],\"s\":[");
|
||||
for (i, &(_, _, b)) in ids.iter().enumerate() {
|
||||
if i > 0 {
|
||||
head.push(',');
|
||||
}
|
||||
push_num(&mut head, b);
|
||||
}
|
||||
head.push(']');
|
||||
if !ids.iter().enumerate().all(|(k, &(n, _, _))| n as usize == k + 1) {
|
||||
head.push_str(",\"k\":[");
|
||||
for (i, &(n, _, _)) in ids.iter().enumerate() {
|
||||
if i > 0 {
|
||||
head.push(',');
|
||||
}
|
||||
push_num(&mut head, n as i64);
|
||||
}
|
||||
head.push(']');
|
||||
}
|
||||
match code {
|
||||
'n' => {
|
||||
head.push_str(",\"c\":\"n\",\"x\":");
|
||||
json_str(&ids[0].1[38..], &mut head);
|
||||
head.push_str(",\"t\":[");
|
||||
let mut previous = 0i64;
|
||||
for (i, &(_, id, _)) in ids.iter().enumerate() {
|
||||
pack_letters(id.as_bytes(), &mut raw);
|
||||
let t: i64 = id[25..38].parse().unwrap();
|
||||
if i > 0 {
|
||||
head.push(',');
|
||||
}
|
||||
push_num(&mut head, t - previous);
|
||||
previous = t;
|
||||
}
|
||||
head.push(']');
|
||||
}
|
||||
'h' => {
|
||||
head.push_str(",\"c\":\"h\",\"x\":");
|
||||
json_str(&ids[0].1[32..], &mut head);
|
||||
for &(_, id, _) in ids {
|
||||
let b = id.as_bytes();
|
||||
for i in (0..32).step_by(2) {
|
||||
raw.push(hex_val(b[i]) << 4 | hex_val(b[i + 1]));
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => text.extend(ids.iter().map(|&(_, id, _)| id)),
|
||||
}
|
||||
head.push('}');
|
||||
}
|
||||
head.push(']');
|
||||
let mut plain = head.into_bytes();
|
||||
for id in text {
|
||||
plain.push(b'\n');
|
||||
plain.extend_from_slice(id.as_bytes());
|
||||
}
|
||||
let packed = zstd::bulk::compress(&plain, 3).unwrap();
|
||||
let mut out = Vec::with_capacity(5 + packed.len() + raw.len());
|
||||
out.push(2);
|
||||
out.extend_from_slice(&(packed.len() as u32).to_be_bytes());
|
||||
out.extend_from_slice(&packed);
|
||||
out.extend_from_slice(&raw);
|
||||
out
|
||||
}
|
||||
@@ -0,0 +1,797 @@
|
||||
//! lib/edge/scan.ts (SegmentScanner, scanOverviewText), lib/files/aggregator.ts
|
||||
//! and lib/edge/range_indexer.ts (RangeIndexer, chunkOf, candidatesOf), in the
|
||||
//! same order of operations, so the outputs can be compared one to one.
|
||||
|
||||
use crate::codec::{encode_chunk, ChunkFile};
|
||||
use crate::js::{self, sha1_hex};
|
||||
use crate::names::{self, is_random_file, JudgedFile, Release, Rules, Verdict};
|
||||
use rustc_hash::{FxHashMap, FxHashSet};
|
||||
use std::rc::Rc;
|
||||
|
||||
pub static mut PROF: [u64; 8] = [0; 8];
|
||||
|
||||
pub struct FileMeta {
|
||||
pub file_key: String,
|
||||
pub name: String,
|
||||
pub poster: String,
|
||||
pub subject: String,
|
||||
pub total_parts: u32,
|
||||
pub groups: Vec<String>,
|
||||
pub release: Release,
|
||||
}
|
||||
|
||||
pub struct Seg {
|
||||
pub file: Rc<FileMeta>,
|
||||
pub number: u32,
|
||||
pub message_id: String,
|
||||
pub bytes: i64,
|
||||
pub article: i64,
|
||||
pub posted_at: Option<i64>,
|
||||
}
|
||||
|
||||
pub struct ReleaseChunk {
|
||||
pub key: String,
|
||||
pub upload: usize,
|
||||
pub name: String,
|
||||
pub source: &'static str,
|
||||
pub file_total: Option<i64>,
|
||||
pub poster: String,
|
||||
pub groups: Vec<String>,
|
||||
pub posted_at: Option<i64>,
|
||||
pub posted_last: Option<i64>,
|
||||
pub files: usize,
|
||||
pub segments: usize,
|
||||
pub bytes: i64,
|
||||
pub first_article: i64,
|
||||
pub last_article: i64,
|
||||
pub readable: bool,
|
||||
pub has_par2: bool,
|
||||
pub payload: Vec<u8>,
|
||||
}
|
||||
|
||||
pub struct Candidate {
|
||||
pub release_key: String,
|
||||
pub upload: usize,
|
||||
pub kind: &'static str,
|
||||
pub message_id: String,
|
||||
pub file: String,
|
||||
}
|
||||
|
||||
pub struct Fragment {
|
||||
pub file_key: String,
|
||||
pub name: String,
|
||||
pub poster: String,
|
||||
pub subject: String,
|
||||
pub total_parts: u32,
|
||||
pub groups: Vec<String>,
|
||||
pub release: Release,
|
||||
pub posted_at: Option<i64>,
|
||||
pub segments: Vec<(u32, String, i64, i64)>,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
pub struct Batch {
|
||||
pub from: i64,
|
||||
pub next: i64,
|
||||
pub headers: u64,
|
||||
pub releases: Vec<ReleaseChunk>,
|
||||
pub samples: Vec<(ReleaseChunk, &'static str, String)>,
|
||||
pub candidates: Vec<Candidate>,
|
||||
/// (rule, key) → (articles, files)
|
||||
pub counts: Vec<(&'static str, String, u64, u64)>,
|
||||
pub fragments: Vec<Fragment>,
|
||||
}
|
||||
|
||||
struct Counter {
|
||||
index: usize,
|
||||
length: usize,
|
||||
segment: i64,
|
||||
total: i64,
|
||||
}
|
||||
|
||||
/// The last "(k/N)" after `after`, scanning back from the end (findCounter).
|
||||
fn find_counter(subject: &[u8], after: usize) -> Option<Counter> {
|
||||
let mut open = subject.len();
|
||||
loop {
|
||||
// lastIndexOf('(', open - 1): at most open - 1.
|
||||
if open == 0 {
|
||||
return None;
|
||||
}
|
||||
open = subject[..open].iter().rposition(|&c| c == b'(')?;
|
||||
if open <= after {
|
||||
return None;
|
||||
}
|
||||
let mut i = open + 1;
|
||||
let mut segment: i64 = 0;
|
||||
let mut digits = 0;
|
||||
while let Some(c) = subject.get(i).filter(|c| c.is_ascii_digit()) {
|
||||
segment = segment * 10 + (c - b'0') as i64;
|
||||
digits += 1;
|
||||
i += 1;
|
||||
}
|
||||
if digits == 0 || subject.get(i) != Some(&b'/') {
|
||||
continue;
|
||||
}
|
||||
i += 1;
|
||||
let mut total: i64 = 0;
|
||||
digits = 0;
|
||||
while let Some(c) = subject.get(i).filter(|c| c.is_ascii_digit()) {
|
||||
total = total * 10 + (c - b'0') as i64;
|
||||
digits += 1;
|
||||
i += 1;
|
||||
}
|
||||
if digits == 0 || subject.get(i) != Some(&b')') {
|
||||
continue;
|
||||
}
|
||||
return Some(Counter { index: open, length: i + 1 - open, segment, total });
|
||||
}
|
||||
}
|
||||
|
||||
fn xref_groups(xref: &str, group: &str) -> Vec<String> {
|
||||
let mut groups = Vec::new();
|
||||
for token in js::split_spaces(js::trim(xref)).into_iter().skip(1) {
|
||||
if let Some(colon) = token.rfind(':') {
|
||||
if colon > 0 {
|
||||
groups.push(token[..colon].to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
if groups.is_empty() {
|
||||
groups.push(group.to_string());
|
||||
}
|
||||
groups
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Scanner {
|
||||
files: FxHashMap<String, Option<Rc<FileMeta>>>,
|
||||
key: String,
|
||||
}
|
||||
|
||||
impl Scanner {
|
||||
fn owner<'a>(&self, xref: &'a str, group: &'a str, priorities: &FxHashMap<String, u32>) -> &'a str {
|
||||
let b = xref.as_bytes();
|
||||
let Some(mut at) = xref.find(' ') else { return group };
|
||||
let mut owner: Option<&str> = None;
|
||||
let mut best = u32::MAX;
|
||||
loop {
|
||||
while b.get(at) == Some(&b' ') {
|
||||
at += 1;
|
||||
}
|
||||
if at >= b.len() {
|
||||
break;
|
||||
}
|
||||
let end = xref[at..].find(' ').map(|i| i + at);
|
||||
// lastIndexOf(':', end ?? length): a colon at or before `end`.
|
||||
let limit = end.unwrap_or(b.len());
|
||||
let colon = xref[..limit.min(b.len())].rfind(':');
|
||||
if let Some(colon) = colon.filter(|&c| c > at) {
|
||||
let name = &xref[at..colon];
|
||||
if let Some(&p) = priorities.get(name) {
|
||||
if p < best {
|
||||
best = p;
|
||||
owner = Some(name);
|
||||
}
|
||||
}
|
||||
}
|
||||
match end {
|
||||
Some(e) => at = e,
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
owner.unwrap_or(group)
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn segment(&mut self, article: i64, subject: &str, from: &str, date: &str, message_id: &str, bytes: &str, xref: &str, group: &str) -> Option<Seg> {
|
||||
if subject.is_empty() || message_id.is_empty() {
|
||||
return None;
|
||||
}
|
||||
let sb = subject.as_bytes();
|
||||
let name_end = subject.rfind('"')?;
|
||||
if name_end == 0 {
|
||||
return None;
|
||||
}
|
||||
let counter = find_counter(sb, name_end)?;
|
||||
self.key.clear();
|
||||
self.key.push_str(from);
|
||||
self.key.push('\u{1}');
|
||||
self.key.push_str(&subject[..counter.index]);
|
||||
self.key.push('\u{0}');
|
||||
self.key.push_str(&subject[counter.index + counter.length..]);
|
||||
let file = match self.files.get(self.key.as_str()) {
|
||||
Some(f) => f.clone(),
|
||||
None => {
|
||||
let f = describe(subject, name_end, &counter, from, xref, group).map(Rc::new);
|
||||
self.files.insert(self.key.clone(), f.clone());
|
||||
f
|
||||
}
|
||||
};
|
||||
let file = file?;
|
||||
if counter.segment < 1 || counter.segment > file.total_parts as i64 {
|
||||
return None;
|
||||
}
|
||||
let id = if message_id.starts_with('<') { message_id.get(1..message_id.len().saturating_sub(1)).unwrap_or("") } else { message_id };
|
||||
Some(Seg {
|
||||
number: counter.segment as u32,
|
||||
message_id: id.to_string(),
|
||||
bytes: js::parse_int(bytes).unwrap_or(0),
|
||||
article,
|
||||
posted_at: js::parse_date(date),
|
||||
file,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
fn describe(subject: &str, name_end: usize, counter: &Counter, from: &str, xref: &str, group: &str) -> Option<FileMeta> {
|
||||
let name_start = subject[..name_end].rfind('"')?;
|
||||
let name = &subject[name_start + 1..name_end];
|
||||
if name.is_empty() || counter.total < 1 {
|
||||
return None;
|
||||
}
|
||||
let joined = format!("{}{}", &subject[..counter.index], &subject[counter.index + counter.length..]);
|
||||
let base = js::trim(&js::collapse_spaces(&joined)).to_string();
|
||||
let file_key = sha1_hex(&format!("{from}\n{base}\n{}", counter.total));
|
||||
let release = names::release_identity(from, &base, name);
|
||||
Some(FileMeta {
|
||||
file_key,
|
||||
name: name.to_string(),
|
||||
poster: from.to_string(),
|
||||
subject: base,
|
||||
total_parts: counter.total as u32,
|
||||
groups: xref_groups(xref, group),
|
||||
release,
|
||||
})
|
||||
}
|
||||
|
||||
/// One line of an XOVER reply: number, subject, from, date, message-id, bytes, xref (scanOverviewText).
|
||||
fn scan_overview_text(text: &str, mut line: impl FnMut(Option<i64>, &str, &str, &str, &str, &str, &str)) {
|
||||
let b = text.as_bytes();
|
||||
let mut start = 0;
|
||||
while start < b.len() {
|
||||
let end = memchr::memchr(b'\n', &b[start..]).map(|i| i + start).unwrap_or(b.len());
|
||||
let stop = if end > start && b[end - 1] == b'\r' { end - 1 } else { end };
|
||||
if stop > start {
|
||||
let l = &text[start..stop];
|
||||
let lb = l.as_bytes();
|
||||
let mut tabs = [usize::MAX; 9];
|
||||
let mut n = 0;
|
||||
for i in memchr::memchr_iter(b'\t', lb) {
|
||||
tabs[n] = i;
|
||||
n += 1;
|
||||
if n == 9 {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if n >= 5 {
|
||||
let field = |a: usize, b: usize| -> &str {
|
||||
if a == usize::MAX {
|
||||
""
|
||||
} else {
|
||||
&l[a + 1..if b == usize::MAX { l.len() } else { b }]
|
||||
}
|
||||
};
|
||||
let mut xref = if n >= 8 { field(tabs[7], tabs[8]) } else { "" };
|
||||
if let Some(rest) = xref.strip_prefix("Xref:") {
|
||||
xref = rest.strip_prefix(' ').unwrap_or(rest);
|
||||
}
|
||||
line(js::parse_int(&l[..tabs[0]]), &l[tabs[0] + 1..tabs[1]], field(tabs[1], tabs[2]), field(tabs[2], tabs[3]), field(tabs[3], tabs[4]), field(tabs[5], tabs[6]), xref);
|
||||
}
|
||||
}
|
||||
start = end + 1;
|
||||
}
|
||||
}
|
||||
|
||||
/// A re-upload's copy of a file is kept apart (FileAggregator, uploadGapMs).
|
||||
pub const UPLOAD_GAP_MS: i64 = 3 * 60 * 60 * 1000;
|
||||
|
||||
struct OpenFile {
|
||||
parts: Vec<Seg>,
|
||||
seen: FxHashSet<u32>,
|
||||
total: u32,
|
||||
touched: u64,
|
||||
/// Posting times seen.
|
||||
earliest: Option<i64>,
|
||||
latest: Option<i64>,
|
||||
}
|
||||
|
||||
impl OpenFile {
|
||||
fn distance(&self, t: Option<i64>) -> i64 {
|
||||
match (t, self.earliest, self.latest) {
|
||||
(Some(t), Some(e), Some(l)) => 0.max(e - t).max(t - l),
|
||||
_ => 0,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct Aggregator {
|
||||
/// Per file key its open copies, oldest first.
|
||||
open: FxHashMap<String, Vec<OpenFile>>,
|
||||
batch: u64,
|
||||
}
|
||||
|
||||
impl Aggregator {
|
||||
fn add(&mut self, segments: Vec<Seg>) -> Vec<Vec<Seg>> {
|
||||
self.batch += 1;
|
||||
let batch = self.batch;
|
||||
let mut touched: Vec<String> = Vec::new();
|
||||
let mut touched_set: FxHashSet<String> = FxHashSet::default();
|
||||
for s in segments {
|
||||
let key = s.file.file_key.clone();
|
||||
if !touched_set.contains(key.as_str()) {
|
||||
touched_set.insert(key.clone());
|
||||
touched.push(key.clone());
|
||||
}
|
||||
let copies = self.open.entry(key).or_default();
|
||||
let t = s.posted_at;
|
||||
// A copy lacking the part, the nearest in time.
|
||||
let mut chosen: Option<usize> = None;
|
||||
let mut best = i64::MAX;
|
||||
for (i, copy) in copies.iter().enumerate() {
|
||||
if copy.seen.contains(&s.number) {
|
||||
continue;
|
||||
}
|
||||
let d = copy.distance(t);
|
||||
if d < best {
|
||||
best = d;
|
||||
chosen = Some(i);
|
||||
}
|
||||
}
|
||||
let index = match chosen {
|
||||
Some(i) => i,
|
||||
None => {
|
||||
if let Some(near) = copies.iter_mut().find(|c| c.distance(t) <= UPLOAD_GAP_MS) {
|
||||
near.touched = batch;
|
||||
continue;
|
||||
}
|
||||
copies.push(OpenFile { parts: Vec::new(), seen: FxHashSet::default(), total: s.file.total_parts, touched: batch, earliest: None, latest: None });
|
||||
copies.len() - 1
|
||||
}
|
||||
};
|
||||
let file = &mut copies[index];
|
||||
file.seen.insert(s.number);
|
||||
if let Some(t) = t {
|
||||
if file.earliest.is_none_or(|e| t < e) {
|
||||
file.earliest = Some(t);
|
||||
}
|
||||
if file.latest.is_none_or(|l| t > l) {
|
||||
file.latest = Some(t);
|
||||
}
|
||||
}
|
||||
file.touched = batch;
|
||||
file.parts.push(s);
|
||||
}
|
||||
let mut complete = Vec::new();
|
||||
for key in touched {
|
||||
let copies = self.open.get_mut(&key).unwrap();
|
||||
let mut i = 0;
|
||||
while i < copies.len() {
|
||||
if (copies[i].parts.len() as u32) < copies[i].total {
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
let mut f = copies.remove(i);
|
||||
f.parts.sort_by_key(|s| s.number);
|
||||
complete.push(f.parts);
|
||||
}
|
||||
if copies.is_empty() {
|
||||
self.open.remove(&key);
|
||||
}
|
||||
}
|
||||
complete
|
||||
}
|
||||
|
||||
/// Files idle for `idle` batches (all when 0): their segments, per file key (its copies one after another).
|
||||
fn evict(&mut self, idle: u64) -> Vec<Vec<Seg>> {
|
||||
let batch = self.batch;
|
||||
let mut out = Vec::new();
|
||||
let keys: Vec<String> = self.open.keys().cloned().collect();
|
||||
for key in keys {
|
||||
let copies = self.open.get_mut(&key).unwrap();
|
||||
let mut gone: Vec<Seg> = Vec::new();
|
||||
let mut i = 0;
|
||||
while i < copies.len() {
|
||||
if idle > 0 && batch - copies[i].touched < idle {
|
||||
i += 1;
|
||||
continue;
|
||||
}
|
||||
gone.extend(copies.remove(i).parts);
|
||||
}
|
||||
if copies.is_empty() {
|
||||
self.open.remove(&key);
|
||||
}
|
||||
if !gone.is_empty() {
|
||||
out.push(gone);
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// uploadsOf: a key's files split where posting pauses for more than UPLOAD_GAP_MS after the latest article so far.
|
||||
fn uploads_of(files: Vec<Vec<Seg>>) -> Vec<Vec<Vec<Seg>>> {
|
||||
if files.len() < 2 {
|
||||
return vec![files];
|
||||
}
|
||||
let first: Vec<Option<i64>> = files.iter().map(|f| posted_of(f)).collect();
|
||||
let last: Vec<Option<i64>> = files.iter().map(|f| f.iter().filter_map(|s| s.posted_at).max()).collect();
|
||||
let mut dated: Vec<usize> = (0..files.len()).filter(|&i| first[i].is_some()).collect();
|
||||
if dated.len() < 2 {
|
||||
return vec![files];
|
||||
}
|
||||
dated.sort_by_key(|&i| (first[i].unwrap(), i));
|
||||
let mut upload = vec![0usize; files.len()];
|
||||
let mut current = 0;
|
||||
let mut end = last[dated[0]].unwrap();
|
||||
for &i in &dated[1..] {
|
||||
if first[i].unwrap() - end > UPLOAD_GAP_MS {
|
||||
current += 1;
|
||||
end = last[i].unwrap();
|
||||
} else if last[i].unwrap() > end {
|
||||
end = last[i].unwrap();
|
||||
}
|
||||
upload[i] = current;
|
||||
}
|
||||
if current == 0 {
|
||||
return vec![files];
|
||||
}
|
||||
let mut out: Vec<Vec<Vec<Seg>>> = (0..=current).map(|_| Vec::new()).collect();
|
||||
for (i, f) in files.into_iter().enumerate() {
|
||||
out[upload[i]].push(f);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
pub struct RangeIndexer {
|
||||
group: String,
|
||||
rules: Rules,
|
||||
start: i64,
|
||||
end: i64,
|
||||
priorities: FxHashMap<String, u32>,
|
||||
idle: u64,
|
||||
chunk_articles: i64,
|
||||
scanner: Scanner,
|
||||
aggregator: Aggregator,
|
||||
batch: Batch,
|
||||
counts: FxHashMap<(&'static str, String), (u64, u64)>,
|
||||
count_order: Vec<(&'static str, String)>,
|
||||
completed: FxHashMap<String, Vec<Vec<Seg>>>,
|
||||
completed_order: Vec<String>,
|
||||
}
|
||||
|
||||
impl RangeIndexer {
|
||||
/// `resume_at`: where this worker starts (a range taken over midway); its first batch starts there.
|
||||
pub fn new(group: &str, start: i64, end: i64, resume_at: i64, rules: Rules, idle: u64) -> Self {
|
||||
RangeIndexer {
|
||||
group: group.to_string(),
|
||||
start,
|
||||
end,
|
||||
priorities: rules.priorities.clone(),
|
||||
rules,
|
||||
idle,
|
||||
chunk_articles: 0,
|
||||
scanner: Scanner::default(),
|
||||
aggregator: Aggregator::default(),
|
||||
batch: Batch { from: resume_at, next: resume_at, ..Default::default() },
|
||||
counts: FxHashMap::default(),
|
||||
count_order: Vec::new(),
|
||||
completed: FxHashMap::default(),
|
||||
completed_order: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn add_text(&mut self, text: &str, next: i64) {
|
||||
let mut segments: Vec<Seg> = Vec::new();
|
||||
let mut crossposts: Vec<String> = Vec::new();
|
||||
let mut obfuscated = 0u64;
|
||||
let mut headers = 0u64;
|
||||
{
|
||||
let group = self.group.clone();
|
||||
let scanner = &mut self.scanner;
|
||||
let priorities = &self.priorities;
|
||||
scan_overview_text(text, |n, subject, from, date, message_id, bytes, xref| {
|
||||
let Some(n) = n else { return };
|
||||
if message_id.is_empty() {
|
||||
return;
|
||||
}
|
||||
headers += 1;
|
||||
let owner = if xref.is_empty() { group.as_str() } else { scanner.owner(xref, &group, priorities) };
|
||||
if owner != group {
|
||||
crossposts.push(owner.to_string());
|
||||
return;
|
||||
}
|
||||
match scanner.segment(n, subject, from, date, message_id, bytes, xref, &group) {
|
||||
Some(s) => segments.push(s),
|
||||
None => obfuscated += 1,
|
||||
}
|
||||
});
|
||||
}
|
||||
self.batch.headers += headers;
|
||||
for owner in crossposts {
|
||||
self.count("crosspost", owner, 1, 0);
|
||||
}
|
||||
if obfuscated > 0 {
|
||||
self.count("obfuscated", String::new(), obfuscated, 0);
|
||||
}
|
||||
self.end_chunk(segments, next);
|
||||
}
|
||||
|
||||
fn end_chunk(&mut self, segments: Vec<Seg>, next: i64) {
|
||||
let t = std::time::Instant::now();
|
||||
let done = self.aggregator.add(segments);
|
||||
unsafe { PROF[3] += t.elapsed().as_nanos() as u64; }
|
||||
self.files(done);
|
||||
if self.chunk_articles == 0 {
|
||||
self.chunk_articles = next - self.batch.from;
|
||||
}
|
||||
let t = std::time::Instant::now();
|
||||
let gone = self.aggregator.evict(self.idle);
|
||||
self.cut_off(gone, false);
|
||||
unsafe { PROF[4] += t.elapsed().as_nanos() as u64; }
|
||||
self.batch.next = next;
|
||||
}
|
||||
|
||||
pub fn finish(&mut self) {
|
||||
let gone = self.aggregator.evict(0);
|
||||
self.cut_off(gone, true);
|
||||
self.batch.next = self.end + 1;
|
||||
}
|
||||
|
||||
/// Files completed and fragments cut since the last take (RangeIndexer.pending).
|
||||
pub fn pending(&self) -> usize {
|
||||
self.batch.fragments.len() + self.completed.values().map(|l| l.len()).sum::<usize>()
|
||||
}
|
||||
|
||||
pub fn take(&mut self) -> Batch {
|
||||
let t = std::time::Instant::now();
|
||||
self.judge_releases();
|
||||
unsafe { PROF[5] += t.elapsed().as_nanos() as u64; }
|
||||
self.scanner.files.clear();
|
||||
let next = self.batch.next;
|
||||
let mut out = std::mem::replace(&mut self.batch, Batch { from: next, next, ..Default::default() });
|
||||
for id in self.count_order.drain(..) {
|
||||
let (articles, files) = self.counts.remove(&id).unwrap();
|
||||
out.counts.push((id.0, id.1, articles, files));
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn files(&mut self, files: Vec<Vec<Seg>>) {
|
||||
for file in files {
|
||||
let key = release_key_of(&file[0]);
|
||||
match self.completed.get_mut(&key) {
|
||||
Some(list) => list.push(file),
|
||||
None => {
|
||||
self.completed_order.push(key.clone());
|
||||
self.completed.insert(key, vec![file]);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn judge_releases(&mut self) {
|
||||
let order = std::mem::take(&mut self.completed_order);
|
||||
let mut by_release = std::mem::take(&mut self.completed);
|
||||
for key in order {
|
||||
for (upload, list) in uploads_of(by_release.remove(&key).unwrap()).into_iter().enumerate() {
|
||||
let judged: Vec<JudgedFile> = list
|
||||
.iter()
|
||||
.map(|f| JudgedFile { name: &f[0].file.name, total_parts: f[0].file.total_parts, groups: &f[0].file.groups, first_message_id: &f[0].message_id })
|
||||
.collect();
|
||||
match names::judge(&key, &judged, &self.rules) {
|
||||
Verdict::Keep => {
|
||||
let candidates = candidates_of(&key, &list, upload);
|
||||
self.batch.releases.push(chunk_of(&key, &list, upload));
|
||||
self.batch.candidates.extend(candidates);
|
||||
}
|
||||
Verdict::Drop { rule, key: rule_key, sample } => {
|
||||
let segments: usize = list.iter().map(|f| f.len()).sum();
|
||||
self.count(rule, rule_key.clone(), segments as u64, list.len() as u64);
|
||||
if sample {
|
||||
self.batch.samples.push((chunk_of(&key, &list, upload), rule, rule_key));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn cut_off(&mut self, files: Vec<Vec<Seg>>, at_end: bool) {
|
||||
let window = self.idle as i64 * self.chunk_articles.max(1);
|
||||
for list in files {
|
||||
if list.len() == 1 && is_random_file(&list[0].file.name) {
|
||||
let tool = names::upload_tool(&list[0].message_id);
|
||||
self.count("per-article", tool.to_string(), 1, 0);
|
||||
continue;
|
||||
}
|
||||
let min_part = list.iter().map(|s| s.number).min().unwrap();
|
||||
let max_part = list.iter().map(|s| s.number).max().unwrap();
|
||||
let first_article = list.iter().map(|s| s.article).min().unwrap();
|
||||
let cut_at_start = min_part > 1 && first_article < self.start + window;
|
||||
let cut_at_end = at_end && max_part < list[0].file.total_parts;
|
||||
if !cut_at_start && !cut_at_end {
|
||||
self.count("incomplete", String::new(), list.len() as u64, 1);
|
||||
continue;
|
||||
}
|
||||
let f = &list[0].file;
|
||||
let mut fragment = Fragment {
|
||||
file_key: f.file_key.clone(),
|
||||
name: f.name.clone(),
|
||||
poster: f.poster.clone(),
|
||||
subject: f.subject.clone(),
|
||||
total_parts: f.total_parts,
|
||||
groups: f.groups.clone(),
|
||||
release: f.release.clone(),
|
||||
posted_at: None,
|
||||
segments: Vec::with_capacity(list.len()),
|
||||
};
|
||||
for s in &list {
|
||||
fragment.segments.push((s.number, s.message_id.clone(), s.bytes, s.article));
|
||||
if let Some(t) = s.posted_at {
|
||||
if fragment.posted_at.is_none_or(|p| t < p) {
|
||||
fragment.posted_at = Some(t);
|
||||
}
|
||||
}
|
||||
}
|
||||
self.batch.fragments.push(fragment);
|
||||
}
|
||||
}
|
||||
|
||||
fn count(&mut self, rule: &'static str, key: String, articles: u64, files: u64) {
|
||||
match self.counts.get_mut(&(rule, key.clone())) {
|
||||
Some(c) => {
|
||||
c.0 += articles;
|
||||
c.1 += files;
|
||||
}
|
||||
None => {
|
||||
self.count_order.push((rule, key.clone()));
|
||||
self.counts.insert((rule, key), (articles, files));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn release_key_of(s: &Seg) -> String {
|
||||
// A named file always has a release (releaseIdentity isn't null unless weak).
|
||||
s.file.release.key.clone()
|
||||
}
|
||||
|
||||
fn posted_of(segments: &[Seg]) -> Option<i64> {
|
||||
segments.iter().filter_map(|s| s.posted_at).min()
|
||||
}
|
||||
|
||||
fn chunk_of(key: &str, files: &[Vec<Seg>], upload: usize) -> ReleaseChunk {
|
||||
let first = &files[0][0].file;
|
||||
let chunk_files: Vec<ChunkFile> = files
|
||||
.iter()
|
||||
.map(|f| {
|
||||
let m = &f[0].file;
|
||||
ChunkFile { name: &m.name, poster: &m.poster, subject: &m.subject, posted_at: posted_of(f), groups: &m.groups, segments: f.iter().map(|s| (s.number, s.message_id.as_str(), s.bytes)).collect() }
|
||||
})
|
||||
.collect();
|
||||
let mut segments = 0;
|
||||
let mut bytes = 0;
|
||||
let mut first_article = i64::MAX;
|
||||
let mut last_article = 0;
|
||||
let mut posted_at: Option<i64> = None;
|
||||
let mut posted_last: Option<i64> = None;
|
||||
for f in files {
|
||||
for s in f {
|
||||
if let Some(t) = s.posted_at {
|
||||
if posted_last.is_none_or(|p| t > p) {
|
||||
posted_last = Some(t);
|
||||
}
|
||||
}
|
||||
segments += 1;
|
||||
bytes += s.bytes;
|
||||
first_article = first_article.min(s.article);
|
||||
last_article = last_article.max(s.article);
|
||||
}
|
||||
if let Some(t) = posted_of(f) {
|
||||
if posted_at.is_none_or(|p| t < p) {
|
||||
posted_at = Some(t);
|
||||
}
|
||||
}
|
||||
}
|
||||
ReleaseChunk {
|
||||
key: key.to_string(),
|
||||
upload,
|
||||
name: first.release.name.clone(),
|
||||
source: first.release.source,
|
||||
file_total: files.iter().find_map(|f| f[0].file.release.file_total_truthy()),
|
||||
poster: first.poster.clone(),
|
||||
groups: first.groups.clone(),
|
||||
posted_at,
|
||||
posted_last,
|
||||
files: files.len(),
|
||||
segments,
|
||||
bytes,
|
||||
first_article,
|
||||
last_article,
|
||||
readable: chunk_files.iter().any(|f| !is_random_file(f.name)),
|
||||
has_par2: chunk_files.iter().any(|f| names::is_par2(f.name)),
|
||||
payload: encode_chunk(&chunk_files),
|
||||
}
|
||||
}
|
||||
|
||||
fn candidates_of(key: &str, files: &[Vec<Seg>], upload: usize) -> Vec<Candidate> {
|
||||
let mut out = Vec::new();
|
||||
let mut tail: Option<(i64, &Vec<Seg>)> = None;
|
||||
let c = |kind, s: &Seg| Candidate { release_key: key.to_string(), upload, kind, message_id: s.message_id.clone(), file: s.file.name.clone() };
|
||||
for f in files {
|
||||
let name = &f[0].file.name;
|
||||
if let Some(volume) = names::seven_zip_volume(name).filter(|&v| v > 1) {
|
||||
if tail.is_none_or(|(v, _)| volume > v) {
|
||||
tail = Some((volume, f));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if names::is_archive_start(name) {
|
||||
out.push(c("archive", &f[0]));
|
||||
} else if names::is_par2(name) && !names::is_later_volume(name) {
|
||||
out.push(c("par2", &f[0]));
|
||||
}
|
||||
}
|
||||
if let Some((_, f)) = tail {
|
||||
out.push(Candidate { release_key: key.to_string(), upload, kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone() });
|
||||
}
|
||||
if out.is_empty() && files.len() == 1 && is_random_file(&files[0][0].file.name) {
|
||||
out.push(c("lone", &files[0][0]));
|
||||
}
|
||||
out.truncate(4);
|
||||
out
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
const GROUP: &str = "alt.binaries.test";
|
||||
|
||||
fn line(n: i64, subject: &str, minutes: i64) -> String {
|
||||
let date = chrono::DateTime::from_timestamp_millis(1_791_194_400_000 + minutes * 60_000).unwrap().format("%a, %d %b %Y %H:%M:%S +0000");
|
||||
format!("{n}\t{subject}\tp <p@x>\t{date}\t<u{n}@test>\t\t1000\t10\tXref: news.example {GROUP}:{n}")
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn uploads_split_as_in_the_typescript() {
|
||||
let mut priorities = FxHashMap::default();
|
||||
priorities.insert(GROUP.to_string(), 1);
|
||||
let mut ix = RangeIndexer::new(GROUP, 1, 10_000, 1, Rules::defaults(priorities), 5);
|
||||
let mut lines: Vec<String> = (0..5).map(|i| line(100 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), i * 60)).collect();
|
||||
lines.push(line(200, r#"Big.Release [2/2] - "Big.Release.par2" yEnc (1/1)"#, 245));
|
||||
lines.extend((0..5).map(|i| line(300 + i, &format!(r#"Big.Release [1/2] - "Big.Release.mkv" yEnc ({}/5)"#, i + 1), 1440 + i)));
|
||||
ix.add_text(&lines.join("\r\n"), 10_001);
|
||||
ix.finish();
|
||||
let batch = ix.take();
|
||||
let got: Vec<(usize, usize)> = batch.releases.iter().map(|r| (r.upload, r.files)).collect();
|
||||
assert_eq!(got, vec![(0, 2), (1, 1)]);
|
||||
assert_eq!(batch.releases[0].posted_last.unwrap() - batch.releases[0].posted_at.unwrap(), 245 * 60_000);
|
||||
assert_eq!(batch.releases[0].key, batch.releases[1].key);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn copies_and_fills_as_in_the_typescript() {
|
||||
let hour = 3_600_000i64;
|
||||
let meta = Rc::new(FileMeta {
|
||||
file_key: "a".into(),
|
||||
name: "a.bin".into(),
|
||||
poster: "p".into(),
|
||||
subject: "s".into(),
|
||||
total_parts: 3,
|
||||
groups: vec![],
|
||||
release: names::release_identity("p", "s", "a.bin"),
|
||||
});
|
||||
let at = |number: u32, h: f64, id: &str| Seg { file: meta.clone(), number, message_id: id.into(), bytes: 100, article: 0, posted_at: Some((h * hour as f64) as i64) };
|
||||
let mut agg = Aggregator::default();
|
||||
assert!(agg.add(vec![at(1, 10.0, "a1"), at(2, 10.0, "a2"), at(1, 10.1, "resend")]).is_empty());
|
||||
assert!(agg.add(vec![at(1, 34.0, "b1"), at(2, 34.0, "b2")]).is_empty());
|
||||
assert_eq!(agg.open.get("a").unwrap().len(), 2);
|
||||
let second = agg.add(vec![at(3, 34.2, "b3")]);
|
||||
assert_eq!(second[0].iter().map(|s| s.message_id.as_str()).collect::<Vec<_>>(), vec!["b1", "b2", "b3"]);
|
||||
let first = agg.add(vec![at(3, 58.0, "a3")]);
|
||||
assert_eq!(first[0].iter().map(|s| s.message_id.as_str()).collect::<Vec<_>>(), vec!["a1", "a2", "a3"]);
|
||||
assert!(agg.open.is_empty());
|
||||
}
|
||||
}
|
||||
+257
@@ -0,0 +1,257 @@
|
||||
//! JavaScript semantics the TypeScript worker relies on: `\s` and `trim()`,
|
||||
//! UTF-16 lengths, `parseInt`, `Date.parse` (for Usenet dates) and sha1 hex.
|
||||
|
||||
use sha1::{Digest, Sha1};
|
||||
|
||||
/// JS `\s` / `trim()` whitespace (WhiteSpace and LineTerminator).
|
||||
pub fn is_space(c: char) -> bool {
|
||||
matches!(
|
||||
c,
|
||||
'\t' | '\n' | '\u{0B}' | '\u{0C}' | '\r' | ' ' | '\u{A0}' | '\u{1680}' | '\u{2000}'..='\u{200A}' | '\u{2028}' | '\u{2029}' | '\u{202F}' | '\u{205F}' | '\u{3000}' | '\u{FEFF}'
|
||||
)
|
||||
}
|
||||
|
||||
pub fn trim(s: &str) -> &str {
|
||||
s.trim_matches(is_space)
|
||||
}
|
||||
|
||||
/// `s.replace(/\s+/g, ' ')`.
|
||||
pub fn collapse_spaces(s: &str) -> String {
|
||||
let mut out = String::with_capacity(s.len());
|
||||
let mut in_space = false;
|
||||
for c in s.chars() {
|
||||
if is_space(c) {
|
||||
if !in_space {
|
||||
out.push(' ');
|
||||
in_space = true;
|
||||
}
|
||||
} else {
|
||||
out.push(c);
|
||||
in_space = false;
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// `s.split(/\s+/)`: like JS, a leading or trailing run gives an empty word.
|
||||
pub fn split_spaces(s: &str) -> Vec<&str> {
|
||||
let mut out = Vec::new();
|
||||
let mut start = 0;
|
||||
let mut in_space = false;
|
||||
let mut run_start = 0;
|
||||
for (i, c) in s.char_indices() {
|
||||
if is_space(c) {
|
||||
if !in_space {
|
||||
run_start = i;
|
||||
in_space = true;
|
||||
}
|
||||
} else if in_space {
|
||||
out.push(&s[start..run_start]);
|
||||
start = i;
|
||||
in_space = false;
|
||||
}
|
||||
}
|
||||
if in_space {
|
||||
out.push(&s[start..run_start]);
|
||||
out.push("");
|
||||
} else {
|
||||
out.push(&s[start..]);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// A string's `.length` in JavaScript (UTF-16 code units).
|
||||
pub fn len16(s: &str) -> usize {
|
||||
if s.is_ascii() {
|
||||
s.len()
|
||||
} else {
|
||||
s.encode_utf16().count()
|
||||
}
|
||||
}
|
||||
|
||||
/// `Number.parseInt(s, 10)`, None for NaN.
|
||||
pub fn parse_int(s: &str) -> Option<i64> {
|
||||
let s = s.trim_start_matches(is_space);
|
||||
let (neg, digits) = match s.as_bytes().first() {
|
||||
Some(b'-') => (true, &s[1..]),
|
||||
Some(b'+') => (false, &s[1..]),
|
||||
_ => (false, s),
|
||||
};
|
||||
let mut n: i64 = 0;
|
||||
let mut any = false;
|
||||
for b in digits.bytes() {
|
||||
if !b.is_ascii_digit() {
|
||||
break;
|
||||
}
|
||||
n = n.wrapping_mul(10).wrapping_add((b - b'0') as i64);
|
||||
any = true;
|
||||
}
|
||||
if any {
|
||||
Some(if neg { -n } else { n })
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
/// Days since 1970-01-01 of a proleptic Gregorian date (month 1..=12; a day past the month's end runs on, as in JS).
|
||||
fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
|
||||
let y = if m <= 2 { y - 1 } else { y };
|
||||
let era = y.div_euclid(400);
|
||||
let yoe = y - era * 400;
|
||||
let mp = (m + 9) % 12;
|
||||
let doy = (153 * mp + 2) / 5 + d - 1;
|
||||
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
|
||||
era * 146_097 + doe - 719_468
|
||||
}
|
||||
|
||||
/// The usual Date headers: [Day, ]D[D] Mon YY[YY] HH:MM:SS (GMT|UTC|UT|±hhmm). None for anything else.
|
||||
fn fast_date(s: &[u8]) -> Option<i64> {
|
||||
let mut i = 0;
|
||||
if s.len() > 4 && s[3] == b',' && s[..3].iter().all(u8::is_ascii_alphabetic) {
|
||||
i = 4;
|
||||
while s.get(i) == Some(&b' ') {
|
||||
i += 1;
|
||||
}
|
||||
}
|
||||
let num = |i: &mut usize, min: usize, max: usize| -> Option<i64> {
|
||||
let start = *i;
|
||||
while *i < s.len() && s[*i].is_ascii_digit() && *i - start < max {
|
||||
*i += 1;
|
||||
}
|
||||
if *i - start < min || s.get(*i).is_some_and(u8::is_ascii_digit) {
|
||||
return None;
|
||||
}
|
||||
Some(std::str::from_utf8(&s[start..*i]).ok()?.parse().ok()?)
|
||||
};
|
||||
let expect = |i: &mut usize, c: u8| -> Option<()> {
|
||||
(s.get(*i) == Some(&c)).then(|| *i += 1)
|
||||
};
|
||||
let day = num(&mut i, 1, 2)?;
|
||||
expect(&mut i, b' ')?;
|
||||
const MONTHS: [&[u8]; 12] = [b"jan", b"feb", b"mar", b"apr", b"may", b"jun", b"jul", b"aug", b"sep", b"oct", b"nov", b"dec"];
|
||||
let mon = s.get(i..i + 3)?;
|
||||
let month = MONTHS.iter().position(|m| mon.iter().zip(m.iter()).all(|(a, b)| a.to_ascii_lowercase() == *b))? as i64 + 1;
|
||||
i += 3;
|
||||
expect(&mut i, b' ')?;
|
||||
let year_start = i;
|
||||
let mut year = num(&mut i, 2, 4)?;
|
||||
match i - year_start {
|
||||
2 => year += if year < 50 { 2000 } else { 1900 },
|
||||
4 => {}
|
||||
_ => return None,
|
||||
}
|
||||
expect(&mut i, b' ')?;
|
||||
let hour = num(&mut i, 2, 2)?;
|
||||
expect(&mut i, b':')?;
|
||||
let minute = num(&mut i, 2, 2)?;
|
||||
expect(&mut i, b':')?;
|
||||
let second = num(&mut i, 2, 2)?;
|
||||
expect(&mut i, b' ')?;
|
||||
if day < 1 || day > 31 || hour > 23 || minute > 59 || second > 59 {
|
||||
return None;
|
||||
}
|
||||
let offset_minutes = match &s[i..] {
|
||||
b"GMT" | b"UTC" | b"UT" => 0,
|
||||
[sign @ (b'+' | b'-'), h1, h2, m1, m2] if [h1, h2, m1, m2].iter().all(|c| c.is_ascii_digit()) => {
|
||||
let minutes = ((h1 - b'0') as i64 * 10 + (h2 - b'0') as i64) * 60 + (m1 - b'0') as i64 * 10 + (m2 - b'0') as i64;
|
||||
if *sign == b'+' { minutes } else { -minutes }
|
||||
}
|
||||
_ => return None,
|
||||
};
|
||||
let days = days_from_civil(year, month, day);
|
||||
Some((((days * 24 + hour) * 60 + minute - offset_minutes) * 60 + second) * 1000)
|
||||
}
|
||||
|
||||
/// `Date.parse` for the Date header (RFC 5322 and the obsolete forms); None for NaN.
|
||||
pub fn parse_date(s: &str) -> Option<i64> {
|
||||
let s = trim(s);
|
||||
if let Some(t) = fast_date(s.as_bytes()) {
|
||||
return Some(t);
|
||||
}
|
||||
parse_date_chrono(s)
|
||||
}
|
||||
|
||||
fn parse_date_chrono(s: &str) -> Option<i64> {
|
||||
// "UTC" isn't an RFC 5322 zone (UT and GMT are), but Date.parse takes it (~a quarter of a.b.boneless).
|
||||
let owned;
|
||||
let s = match s.strip_suffix(" UTC") {
|
||||
Some(rest) => {
|
||||
owned = format!("{rest} GMT");
|
||||
owned.as_str()
|
||||
}
|
||||
None => s,
|
||||
};
|
||||
if let Ok(d) = chrono::DateTime::parse_from_rfc2822(s) {
|
||||
return Some(d.timestamp_millis());
|
||||
}
|
||||
// "Mon, 5 Oct 2026 10:00:00 +0000 (UTC)" and other trailing comments.
|
||||
if let Some(i) = s.find(" (") {
|
||||
if let Ok(d) = chrono::DateTime::parse_from_rfc2822(&s[..i]) {
|
||||
return Some(d.timestamp_millis());
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// sha1 of the UTF-8 text, as lowercase hex (node's `hash('sha1', text)`).
|
||||
pub fn sha1_hex(text: &str) -> String {
|
||||
let digest = Sha1::digest(text.as_bytes());
|
||||
let mut out = String::with_capacity(40);
|
||||
for b in digest {
|
||||
out.push(char::from_digit((b >> 4) as u32, 16).unwrap());
|
||||
out.push(char::from_digit((b & 15) as u32, 16).unwrap());
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// JSON.stringify of a string.
|
||||
pub fn json_str(s: &str, out: &mut String) {
|
||||
out.push('"');
|
||||
for c in s.chars() {
|
||||
match c {
|
||||
'"' => out.push_str("\\\""),
|
||||
'\\' => out.push_str("\\\\"),
|
||||
'\u{08}' => out.push_str("\\b"),
|
||||
'\u{0C}' => out.push_str("\\f"),
|
||||
'\n' => out.push_str("\\n"),
|
||||
'\r' => out.push_str("\\r"),
|
||||
'\t' => out.push_str("\\t"),
|
||||
c if (c as u32) < 0x20 => out.push_str(&format!("\\u{:04x}", c as u32)),
|
||||
c => out.push(c),
|
||||
}
|
||||
}
|
||||
out.push('"');
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn split_like_js() {
|
||||
assert_eq!(split_spaces("a b"), vec!["a", "b"]);
|
||||
assert_eq!(split_spaces(" a b "), vec!["", "a", "b", ""]);
|
||||
assert_eq!(split_spaces(""), vec![""]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_int_like_js() {
|
||||
assert_eq!(parse_int("123abc"), Some(123));
|
||||
assert_eq!(parse_int(" -5"), Some(-5));
|
||||
assert_eq!(parse_int("x"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn dates() {
|
||||
assert_eq!(parse_date("Mon, 05 Oct 2026 10:00:00 +0000"), Some(1_791_194_400_000));
|
||||
assert_eq!(parse_date("5 Oct 2026 10:00:00 GMT"), Some(1_791_194_400_000));
|
||||
assert_eq!(parse_date("Mon, 05 Oct 26 10:00:00 UTC"), Some(1_791_194_400_000));
|
||||
for d in [
|
||||
"Mon, 05 Oct 2026 10:00:00 +0000", "Mon, 5 Oct 2026 12:00:00 +0200", "05 Oct 2026 10:00:00 GMT", "Thu, 01 Jan 1970 00:00:00 GMT",
|
||||
"Sat, 29 Feb 2020 23:59:59 -0130", "Fri, 31 Dec 99 00:00:00 UTC", "Mon, 05 Oct 2026 10:00:00 UT",
|
||||
] {
|
||||
assert_eq!(fast_date(d.as_bytes()), parse_date_chrono(d), "{d}");
|
||||
}
|
||||
assert_eq!(fast_date(b"Mon, 05 Oct 2026 10:00:00 EST"), None);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
//! The edge worker's indexing (lib/edge RangeIndexer and what it uses) in
|
||||
//! Rust, kept identical to the TypeScript (check.ts): the edge-bench
|
||||
//! benchmark and the edge-node worker share it.
|
||||
|
||||
pub mod codec;
|
||||
pub mod indexer;
|
||||
pub mod js;
|
||||
pub mod names;
|
||||
pub mod wire;
|
||||
@@ -0,0 +1,243 @@
|
||||
//! The edge worker's indexing in Rust, on a range captured by `edge:bench
|
||||
//! --capture` (REBUILD.md, part 2: TypeScript or Rust). Same work as
|
||||
//! edge:bench's replay: inflate each chunk, scan it, aggregate files, cut off
|
||||
//! idle ones, and every N chunks judge and encode the releases. `--dump`
|
||||
//! writes every output as JSON lines for edge/check.ts.
|
||||
//!
|
||||
//! edge-bench <dir> [--batch-chunks 10] [--idle 5] [--dump out.jsonl] [--wire batches.jsonl]
|
||||
|
||||
use base64::Engine;
|
||||
use edge::indexer::{self, Batch, RangeIndexer};
|
||||
use edge::js::json_str;
|
||||
use edge::{names, wire};
|
||||
use rustc_hash::FxHashMap;
|
||||
use std::io::Write;
|
||||
use std::time::Instant;
|
||||
|
||||
// Many small strings (message-ids, names): the system allocator was a measurable share.
|
||||
#[global_allocator]
|
||||
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
|
||||
|
||||
/// A chunk as the news server sent it (zlib; gzip also taken), whole: libdeflate wants the output
|
||||
/// size, so it grows until the text fits (~3.6 times the compressed size on boneless).
|
||||
fn inflate(raw: &[u8], d: &mut libdeflater::Decompressor) -> Vec<u8> {
|
||||
let mut out = vec![0u8; raw.len() * 4 + 4096];
|
||||
loop {
|
||||
let result = if raw.starts_with(&[0x1f, 0x8b]) { d.gzip_decompress(raw, &mut out) } else { d.zlib_decompress(raw, &mut out) };
|
||||
match result {
|
||||
Ok(n) => {
|
||||
out.truncate(n);
|
||||
return out;
|
||||
}
|
||||
Err(libdeflater::DecompressionError::InsufficientSpace) => out.resize(out.len() * 2, 0),
|
||||
Err(e) => panic!("inflate: {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn json_field(json: &str, name: &str) -> String {
|
||||
let at = json.find(&format!("\"{name}\":")).unwrap() + name.len() + 3;
|
||||
let rest = json[at..].trim_start();
|
||||
if let Some(s) = rest.strip_prefix('"') {
|
||||
s[..s.find('"').unwrap()].to_string()
|
||||
} else {
|
||||
rest[..rest.find([',', '}']).unwrap()].trim().to_string()
|
||||
}
|
||||
}
|
||||
|
||||
fn peak_rss_mb() -> u64 {
|
||||
let status = std::fs::read_to_string("/proc/self/status").unwrap_or_default();
|
||||
status.lines().find(|l| l.starts_with("VmHWM:")).and_then(|l| l.split_whitespace().nth(1)).and_then(|k| k.parse::<u64>().ok()).unwrap_or(0) / 1024
|
||||
}
|
||||
|
||||
fn opt_num(n: Option<i64>, out: &mut String) {
|
||||
match n {
|
||||
Some(n) => out.push_str(&n.to_string()),
|
||||
None => out.push_str("null"),
|
||||
}
|
||||
}
|
||||
|
||||
fn dump_batch(batch: &Batch, out: &mut impl Write) {
|
||||
let b64 = base64::engine::general_purpose::STANDARD;
|
||||
let mut line = String::new();
|
||||
let mut emit = |line: &mut String| {
|
||||
line.push('\n');
|
||||
out.write_all(line.as_bytes()).unwrap();
|
||||
line.clear();
|
||||
};
|
||||
line.push_str(&format!("{{\"t\":\"b\",\"from\":{},\"next\":{},\"headers\":{}}}", batch.from, batch.next, batch.headers));
|
||||
emit(&mut line);
|
||||
let release = |r: &indexer::ReleaseChunk, kind: &str, line: &mut String| {
|
||||
line.push_str(&format!("{{\"t\":\"{kind}\",\"upload\":{},\"key\":", r.upload));
|
||||
json_str(&r.key, line);
|
||||
line.push_str(",\"name\":");
|
||||
json_str(&r.name, line);
|
||||
line.push_str(&format!(",\"source\":\"{}\",\"fileTotal\":", r.source));
|
||||
opt_num(r.file_total, line);
|
||||
line.push_str(",\"poster\":");
|
||||
json_str(&r.poster, line);
|
||||
line.push_str(",\"groups\":[");
|
||||
for (i, g) in r.groups.iter().enumerate() {
|
||||
if i > 0 {
|
||||
line.push(',');
|
||||
}
|
||||
json_str(g, line);
|
||||
}
|
||||
line.push_str("],\"postedAt\":");
|
||||
opt_num(r.posted_at, line);
|
||||
line.push_str(",\"postedLast\":");
|
||||
opt_num(r.posted_last, line);
|
||||
line.push_str(&format!(
|
||||
",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"payload\":\"{}\"",
|
||||
r.files,
|
||||
r.segments,
|
||||
r.bytes,
|
||||
r.first_article,
|
||||
r.last_article,
|
||||
r.readable,
|
||||
r.has_par2,
|
||||
b64.encode(&r.payload)
|
||||
));
|
||||
};
|
||||
for r in &batch.releases {
|
||||
release(r, "r", &mut line);
|
||||
line.push('}');
|
||||
emit(&mut line);
|
||||
}
|
||||
for (r, rule, key) in &batch.samples {
|
||||
release(r, "s", &mut line);
|
||||
line.push_str(&format!(",\"rule\":\"{rule}\",\"ruleKey\":"));
|
||||
json_str(key, &mut line);
|
||||
line.push('}');
|
||||
emit(&mut line);
|
||||
}
|
||||
for c in &batch.candidates {
|
||||
line.push_str(&format!("{{\"t\":\"c\",\"upload\":{},\"releaseKey\":", c.upload));
|
||||
json_str(&c.release_key, &mut line);
|
||||
line.push_str(&format!(",\"kind\":\"{}\",\"messageId\":", c.kind));
|
||||
json_str(&c.message_id, &mut line);
|
||||
line.push_str(",\"file\":");
|
||||
json_str(&c.file, &mut line);
|
||||
line.push('}');
|
||||
emit(&mut line);
|
||||
}
|
||||
for (rule, key, articles, files) in &batch.counts {
|
||||
line.push_str(&format!("{{\"t\":\"n\",\"rule\":\"{rule}\",\"key\":"));
|
||||
json_str(key, &mut line);
|
||||
line.push_str(&format!(",\"articles\":{articles},\"files\":{files}}}"));
|
||||
emit(&mut line);
|
||||
}
|
||||
for f in &batch.fragments {
|
||||
line.push_str("{\"t\":\"f\",\"fileKey\":");
|
||||
json_str(&f.file_key, &mut line);
|
||||
line.push_str(",\"name\":");
|
||||
json_str(&f.name, &mut line);
|
||||
line.push_str(&format!(",\"totalParts\":{},\"postedAt\":", f.total_parts));
|
||||
opt_num(f.posted_at, &mut line);
|
||||
line.push_str(",\"segments\":[");
|
||||
for (i, (n, id, bytes, article)) in f.segments.iter().enumerate() {
|
||||
if i > 0 {
|
||||
line.push(',');
|
||||
}
|
||||
line.push_str(&format!("[{n},"));
|
||||
json_str(id, &mut line);
|
||||
line.push_str(&format!(",{bytes},{article}]"));
|
||||
}
|
||||
line.push_str("]}");
|
||||
emit(&mut line);
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let args: Vec<String> = std::env::args().collect();
|
||||
let dir = args.get(1).expect("usage: edge-bench <dir> [--batch-chunks N] [--idle N] [--dump file]").clone();
|
||||
let flag = |name: &str| args.iter().position(|a| a == name).and_then(|i| args.get(i + 1)).cloned();
|
||||
let batch_chunks: usize = flag("--batch-chunks").map(|v| v.parse().unwrap()).unwrap_or(10);
|
||||
let idle: u64 = flag("--idle").map(|v| v.parse().unwrap()).unwrap_or(5);
|
||||
let mut dump = flag("--dump").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap()));
|
||||
// Each batch as the worker posts it (wire.rs), one JSON per line: check.ts compares it with toWire.
|
||||
let mut wire_out = flag("--wire").map(|p| std::io::BufWriter::new(std::fs::File::create(p).unwrap()));
|
||||
|
||||
let range = std::fs::read_to_string(format!("{dir}/range.json")).unwrap();
|
||||
let group = json_field(&range, "group");
|
||||
let start: i64 = json_field(&range, "start").parse().unwrap();
|
||||
let end: i64 = json_field(&range, "end").parse().unwrap();
|
||||
let mut files: Vec<String> = std::fs::read_dir(&dir).unwrap().map(|e| e.unwrap().file_name().into_string().unwrap()).filter(|f| f.ends_with(".gz")).collect();
|
||||
files.sort();
|
||||
|
||||
let mut priorities = FxHashMap::default();
|
||||
priorities.insert(group.clone(), 1);
|
||||
let mut indexer = RangeIndexer::new(&group, start, end, start, names::Rules::defaults(priorities), idle);
|
||||
|
||||
let (mut inflate_s, mut index_s) = (0.0, 0.0);
|
||||
let (mut compressed, mut text_bytes) = (0usize, 0usize);
|
||||
let mut out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, fragments, samples
|
||||
let mut keys = std::collections::HashSet::new();
|
||||
let mut decompressor = libdeflater::Decompressor::new();
|
||||
let wall = Instant::now();
|
||||
for (i, file) in files.iter().enumerate() {
|
||||
let raw = std::fs::read(format!("{dir}/{file}")).unwrap();
|
||||
compressed += raw.len();
|
||||
let t = Instant::now();
|
||||
let data = inflate(&raw, &mut decompressor);
|
||||
inflate_s += t.elapsed().as_secs_f64();
|
||||
text_bytes += data.len();
|
||||
let t = Instant::now();
|
||||
let text = String::from_utf8_lossy(&data);
|
||||
let next: i64 = file.split('-').nth(2).unwrap().trim_end_matches(".gz").parse::<i64>().unwrap() + 1;
|
||||
indexer.add_text(&text, next);
|
||||
let last = i + 1 == files.len();
|
||||
if last {
|
||||
indexer.finish();
|
||||
}
|
||||
if !last && (i + 1) % batch_chunks != 0 {
|
||||
index_s += t.elapsed().as_secs_f64();
|
||||
continue;
|
||||
}
|
||||
let batch = indexer.take();
|
||||
index_s += t.elapsed().as_secs_f64();
|
||||
out.0 += batch.headers;
|
||||
out.1 += batch.releases.len();
|
||||
for r in &batch.releases {
|
||||
keys.insert(r.key.clone());
|
||||
out.2 += r.files;
|
||||
out.3 += r.segments;
|
||||
out.4 += r.payload.len();
|
||||
}
|
||||
out.5 += batch.candidates.len();
|
||||
out.6 += batch.fragments.len();
|
||||
out.7 += batch.samples.len();
|
||||
if let Some(d) = dump.as_mut() {
|
||||
dump_batch(&batch, d);
|
||||
}
|
||||
if let Some(w) = wire_out.as_mut() {
|
||||
w.write_all(&wire::to_json(&batch)).unwrap();
|
||||
w.write_all(b"\n").unwrap();
|
||||
}
|
||||
}
|
||||
let total = inflate_s + index_s;
|
||||
#[allow(static_mut_refs)]
|
||||
let p = unsafe { indexer::PROF };
|
||||
eprintln!(" of which aggregate {:.2} s, evict+cutoff {:.2} s, judge+encode {:.2} s", p[3] as f64 / 1e9, p[4] as f64 / 1e9, p[5] as f64 / 1e9);
|
||||
println!("{group} {start}–{end}: {} chunks, {} MB compressed, {} MB text", files.len(), compressed / 1_000_000, text_bytes / 1_000_000);
|
||||
println!(
|
||||
"{} headers in {:.2} s ({:.0}/s on one core; wall {:.2} s)",
|
||||
out.0,
|
||||
total,
|
||||
out.0 as f64 / total,
|
||||
wall.elapsed().as_secs_f64()
|
||||
);
|
||||
println!(" inflate {inflate_s:.2} s, index {index_s:.2} s; peak RSS {} MB", peak_rss_mb());
|
||||
println!(
|
||||
" kept: {} release chunks ({} releases), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} fragments, {} samples",
|
||||
out.1,
|
||||
keys.len(),
|
||||
out.2,
|
||||
out.3,
|
||||
out.4 as f64 / 1e6,
|
||||
out.4 as f64 / out.3.max(1) as f64,
|
||||
out.5,
|
||||
out.6,
|
||||
out.7
|
||||
);
|
||||
}
|
||||
+1006
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,144 @@
|
||||
//! A batch as the coordinator takes it (lib/edge/wire.ts toWire): JSON with
|
||||
//! the TypeScript names, chunk payloads in base64.
|
||||
|
||||
use crate::indexer::{Batch, Candidate, Fragment, ReleaseChunk};
|
||||
use base64::Engine;
|
||||
use serde::Serialize;
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct WireRelease<'a> {
|
||||
key: &'a str,
|
||||
upload: usize,
|
||||
name: &'a str,
|
||||
source: &'a str,
|
||||
file_total: Option<i64>,
|
||||
poster: &'a str,
|
||||
groups: &'a [String],
|
||||
posted_at: Option<i64>,
|
||||
posted_last: Option<i64>,
|
||||
files: usize,
|
||||
segments: usize,
|
||||
bytes: i64,
|
||||
first_article: i64,
|
||||
last_article: i64,
|
||||
readable: bool,
|
||||
has_par2: bool,
|
||||
payload: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
rule: Option<&'a str>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
rule_key: Option<&'a str>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct WireCandidate<'a> {
|
||||
release_key: &'a str,
|
||||
upload: usize,
|
||||
kind: &'a str,
|
||||
message_id: &'a str,
|
||||
file: &'a str,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct WireCount<'a> {
|
||||
rule: &'a str,
|
||||
key: &'a str,
|
||||
articles: u64,
|
||||
files: u64,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct WireFragmentRelease<'a> {
|
||||
key: &'a str,
|
||||
name: &'a str,
|
||||
source: &'a str,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
file_total: Option<i64>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct WireFragment<'a> {
|
||||
file_key: &'a str,
|
||||
name: &'a str,
|
||||
poster: &'a str,
|
||||
subject: &'a str,
|
||||
total_parts: u32,
|
||||
groups: &'a [String],
|
||||
release: Option<WireFragmentRelease<'a>>,
|
||||
posted_at: Option<i64>,
|
||||
segments: Vec<(u32, &'a str, i64, i64)>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
struct WireBatch<'a> {
|
||||
from: i64,
|
||||
next: i64,
|
||||
headers: u64,
|
||||
releases: Vec<WireRelease<'a>>,
|
||||
samples: Vec<WireRelease<'a>>,
|
||||
candidates: Vec<WireCandidate<'a>>,
|
||||
counts: Vec<WireCount<'a>>,
|
||||
fragments: Vec<WireFragment<'a>>,
|
||||
}
|
||||
|
||||
fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRelease<'a> {
|
||||
WireRelease {
|
||||
key: &r.key,
|
||||
upload: r.upload,
|
||||
name: &r.name,
|
||||
source: r.source,
|
||||
file_total: r.file_total,
|
||||
poster: &r.poster,
|
||||
groups: &r.groups,
|
||||
posted_at: r.posted_at,
|
||||
posted_last: r.posted_last,
|
||||
files: r.files,
|
||||
segments: r.segments,
|
||||
bytes: r.bytes,
|
||||
first_article: r.first_article,
|
||||
last_article: r.last_article,
|
||||
readable: r.readable,
|
||||
has_par2: r.has_par2,
|
||||
payload: base64::engine::general_purpose::STANDARD.encode(&r.payload),
|
||||
rule: rule.map(|(rule, _)| rule),
|
||||
rule_key: rule.map(|(_, key)| key),
|
||||
}
|
||||
}
|
||||
|
||||
fn candidate(c: &Candidate) -> WireCandidate<'_> {
|
||||
WireCandidate { release_key: &c.release_key, upload: c.upload, kind: c.kind, message_id: &c.message_id, file: &c.file }
|
||||
}
|
||||
|
||||
fn fragment(f: &Fragment) -> WireFragment<'_> {
|
||||
WireFragment {
|
||||
file_key: &f.file_key,
|
||||
name: &f.name,
|
||||
poster: &f.poster,
|
||||
subject: &f.subject,
|
||||
total_parts: f.total_parts,
|
||||
groups: &f.groups,
|
||||
// A named file always has a release; fileTotal only when truthy, as fragmentOf.
|
||||
release: Some(WireFragmentRelease { key: &f.release.key, name: &f.release.name, source: f.release.source, file_total: f.release.file_total_truthy() }),
|
||||
posted_at: f.posted_at,
|
||||
segments: f.segments.iter().map(|(n, id, bytes, article)| (*n, id.as_str(), *bytes, *article)).collect(),
|
||||
}
|
||||
}
|
||||
|
||||
/// toWire(batch) as JSON bytes.
|
||||
pub fn to_json(batch: &Batch) -> Vec<u8> {
|
||||
let wire = WireBatch {
|
||||
from: batch.from,
|
||||
next: batch.next,
|
||||
headers: batch.headers,
|
||||
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(),
|
||||
candidates: batch.candidates.iter().map(candidate).collect(),
|
||||
counts: batch.counts.iter().map(|(rule, key, articles, files)| WireCount { rule, key, articles: *articles, files: *files }).collect(),
|
||||
fragments: batch.fragments.iter().map(fragment).collect(),
|
||||
};
|
||||
serde_json::to_vec(&wire).unwrap()
|
||||
}
|
||||
Reference in new issue
Block a user