Rebuild: re-uploads are releases of their own

A poster re-uploading under the same name made one release of all
uploads: in a sample, 24% of releases with 2+ files had a file name more
than once (4.3% of files extra copies), 70% of those copies posted a day
or more apart; their NZBs listed each copy.

- File assembly keeps a re-upload's copy apart: a part already in every
  open copy, posted more than 3 h from them, starts another copy; a
  missing part fills the nearest copy (a re-send of what was missing),
  a re-sent part near a copy that has it is ignored as before.
- A key's files in a batch split into uploads where posting pauses for
  more than 3 h, counted from the latest article posted so far (a file
  that takes hours to post doesn't split it); each upload is judged and
  chunked on its own (ReleaseChunk.upload, .postedLast; candidates carry
  their upload).
- The catalog matches a chunk to the release of its key whose articles
  the range's group owns and whose posting window is within 3 h, else
  starts one (key, key:2, …): releases get identity_key, group_id,
  last_posted_at. Keys are locked in order (advisory locks), so
  concurrent batches neither deadlock nor both start one; the merger
  writes in key order too. The same upload multi-posted to other groups
  is a release per group.
- The gap: 91% of uploads post within 3 h; 73% of re-uploaded copies
  come later (releases:upload-gaps). releases:duplicates measures the
  duplicate names.

Both implementations; check.ts compares the windows and uploads too
(identical on all captures; Rust tests mirror the TypeScript cases).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tools/edge-bench-rs/src/indexer.rs

@@ -31,12 +31,14 @@ pub struct Seg {
 
 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,
@@ -49,6 +51,7 @@ pub struct ReleaseChunk {
 
 pub struct Candidate {
     pub release_key: String,
+    pub upload: usize,
     pub kind: &'static str,
     pub message_id: String,
     pub file: String,
@@ -275,60 +278,167 @@ fn scan_overview_text(text: &str, mut line: impl FnMut(Option<i64>, &str, &str,
     }
 }
 
+/// 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 {
-    open: FxHashMap<String, OpenFile>,
+    /// 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;
-            let file = match self.open.get_mut(key.as_str()) {
-                Some(f) => f,
-                None => self.open.entry(key.clone()).or_insert(OpenFile { parts: Vec::new(), seen: FxHashSet::default(), total: s.file.total_parts, touched: self.batch }),
-            };
+            let key = s.file.file_key.clone();
             if !touched_set.contains(key.as_str()) {
                 touched_set.insert(key.clone());
                 touched.push(key.clone());
             }
-            file.touched = self.batch;
-            if file.seen.insert(s.number) {
-                file.parts.push(s);
+            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 f = self.open.get(&key).unwrap();
-            if (f.parts.len() as u32) < f.total {
-                continue;
+            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);
             }
-            let mut f = self.open.remove(&key).unwrap();
-            f.parts.sort_by_key(|s| s.number);
-            complete.push(f.parts);
         }
         complete
     }
 
-    /// Files idle for `idle` batches (all when 0): their segments, file by file in arrival order.
+    /// 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 gone: Vec<String> = self.open.iter().filter(|(_, f)| !(idle > 0 && batch - f.touched < idle)).map(|(k, _)| k.clone()).collect();
-        gone.into_iter().map(|k| self.open.remove(&k).unwrap().parts).collect()
+        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,
     start: i64,
@@ -452,22 +562,23 @@ impl RangeIndexer {
         let order = std::mem::take(&mut self.completed_order);
         let mut by_release = std::mem::take(&mut self.completed);
         for key in order {
-            let list = by_release.remove(&key).unwrap();
-            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) {
-                Verdict::Keep => {
-                    let candidates = candidates_of(&key, &list);
-                    self.batch.releases.push(chunk_of(&key, &list));
-                    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), rule, rule_key));
+            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) {
+                    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));
+                        }
                     }
                 }
             }
@@ -528,7 +639,7 @@ fn posted_of(segments: &[Seg]) -> Option<i64> {
     segments.iter().filter_map(|s| s.posted_at).min()
 }
 
-fn chunk_of(key: &str, files: &[Vec<Seg>]) -> ReleaseChunk {
+fn chunk_of(key: &str, files: &[Vec<Seg>], upload: usize) -> ReleaseChunk {
     let first = &files[0][0].file;
     let chunk_files: Vec<ChunkFile> = files
         .iter()
@@ -542,8 +653,14 @@ fn chunk_of(key: &str, files: &[Vec<Seg>]) -> ReleaseChunk {
     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);
@@ -557,12 +674,14 @@ fn chunk_of(key: &str, files: &[Vec<Seg>]) -> ReleaseChunk {
     }
     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,
@@ -574,10 +693,10 @@ fn chunk_of(key: &str, files: &[Vec<Seg>]) -> ReleaseChunk {
     }
 }
 
-fn candidates_of(key: &str, files: &[Vec<Seg>]) -> Vec<Candidate> {
+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(), kind, message_id: s.message_id.clone(), file: s.file.name.clone() };
+    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) {
@@ -593,7 +712,7 @@ fn candidates_of(key: &str, files: &[Vec<Seg>]) -> Vec<Candidate> {
         }
     }
     if let Some((_, f)) = tail {
-        out.push(Candidate { release_key: key.to_string(), kind: "7z-tail", message_id: f.last().unwrap().message_id.clone(), file: f[0].file.name.clone() });
+        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]));
@@ -601,3 +720,56 @@ fn candidates_of(key: &str, files: &[Vec<Seg>]) -> Vec<Candidate> {
     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, 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());
+    }
+}

tools/edge-bench-rs/src/main.rs

@@ -72,7 +72,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
     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}\",\"key\":"));
+        line.push_str(&format!("{{\"t\":\"{kind}\",\"upload\":{},\"key\":", r.upload));
         json_str(&r.key, line);
         line.push_str(",\"name\":");
         json_str(&r.name, line);
@@ -89,6 +89,8 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
         }
         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,
@@ -114,7 +116,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
         emit(&mut line);
     }
     for c in &batch.candidates {
-        line.push_str("{\"t\":\"c\",\"releaseKey\":");
+        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);