voyager 3: the window in the Rust edge

The indexer finishes files and releases by posting time instead of the
range path: the frontier is the median Date of the last 1,024 plausible
dates (within 7 days of it; a run of 1,024 agreeing dates moves it), and
a release is finished, judged and written once the frontier is 24 h past
its last part, flagged edge when posted within a day of the block's start.
At the block's end the rest is written provisionally as edges for the
block merger. Files short of parts stay in their release (partial: the
store writes complete = false). Gone: fragments and the cut-off,
uploads_of (online now), check.ts and the fixture comparison; CI replays
the fixtures through edge-bench instead. The wire carries finished, edge
and partial.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

edge/src/indexer.rs

@@ -1,6 +1,7 @@
-//! 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.
+//! The edge's indexing: lib/edge/scan.ts (SegmentScanner, scanOverviewText) and lib/files/aggregator.ts as in the
+//! TypeScript, then voyager 3's window (REBUILD.md, "Voyager 3: window indexing", 2): files and releases are
+//! finished when the frontier (the median Date of the last FRONTIER_ARTICLES plausible ones) is WINDOW_MS past
+//! their last part, or at the block's end (provisionally, as edges), and written once.
 
 use crate::codec::{encode_chunk, ChunkFile};
 use crate::js::{self, sha1_hex};
@@ -46,6 +47,12 @@ pub struct ReleaseChunk {
     pub last_article: i64,
     pub readable: bool,
     pub has_par2: bool,
+    /// The window closed on it (false: still open at the block's end, written provisionally).
+    pub finished: bool,
+    /// Open at the block's end, or within the window of its start: the block merger's.
+    pub edge: bool,
+    /// A file misses parts.
+    pub partial: bool,
     pub payload: Vec<u8>,
     /// The first files' subjects (at most SUBJECTS): edge-bench's --releases dump for the grouping sweeps.
     pub subjects: Vec<String>,
@@ -61,18 +68,6 @@ pub struct Candidate {
     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,
@@ -86,7 +81,6 @@ pub struct Batch {
     pub candidates: Vec<Candidate>,
     /// (rule, key) → (articles, files)
     pub counts: Vec<(&'static str, String, u64, u64)>,
-    pub fragments: Vec<Fragment>,
 }
 
 struct Counter {
@@ -470,8 +464,108 @@ 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).
+/// A re-upload's copy of a file is kept apart (FileAggregator, uploadGapMs); a file joins the open release of its
+/// key posted within this gap of it, else starts one (uploadsOf, online).
 pub const UPLOAD_GAP_MS: i64 = 3 * 60 * 60 * 1000;
+/// The window (REBUILD.md, "Voyager 3: window indexing", 2): a file still missing parts, and a release, are finished
+/// once the frontier is this far past their last part.
+pub const WINDOW_MS: i64 = 24 * 60 * 60 * 1000;
+/// The frontier is the median Date of the last this many plausibly dated articles: enough that a minority of skewed
+/// dates can't move it, few enough that it trails the newest article by little even in a quiet group (half the
+/// ring: a day at 500 articles a day).
+pub const FRONTIER_ARTICLES: usize = 1024;
+/// A Date further than this from the frontier isn't believed (clock skew, bogus years); FRONTIER_ARTICLES of them
+/// in a row that agree among themselves move it (the group paused for longer).
+pub const PLAUSIBLE_MS: i64 = 7 * 24 * 60 * 60 * 1000;
+/// The frontier is established once this many dated articles were seen (a lone bogus date doesn't set it).
+const FRONTIER_MIN: usize = 64;
+
+fn median(dates: &mut [i64]) -> i64 {
+    let mid = dates.len() / 2;
+    *dates.select_nth_unstable(mid).1
+}
+
+/// Where the block is in posting time: the median Date of the last FRONTIER_ARTICLES plausible ones. Date headers
+/// alone aren't trusted: a date more than PLAUSIBLE_MS from the frontier is set aside, and only FRONTIER_ARTICLES
+/// such dates in a row, agreeing among themselves, move the frontier there.
+#[derive(Default)]
+pub struct Frontier {
+    ring: Vec<i64>,
+    at: usize,
+    value: Option<i64>,
+    /// The frontier when first established: the block's start in posting time (releases within WINDOW_MS of it are edges).
+    start: Option<i64>,
+    /// Dates set aside since the last plausible one.
+    aside: Vec<i64>,
+    pending: usize,
+}
+
+impl Frontier {
+    pub fn add(&mut self, t: i64) {
+        if let Some(f) = self.value {
+            if (t - f).abs() > PLAUSIBLE_MS {
+                self.aside.push(t);
+                if self.aside.len() >= FRONTIER_ARTICLES {
+                    let m = median(&mut self.aside.clone());
+                    if self.aside.iter().filter(|&&d| (d - m).abs() <= PLAUSIBLE_MS).count() * 2 >= self.aside.len() {
+                        self.ring = self.aside.iter().copied().filter(|&d| (d - m).abs() <= PLAUSIBLE_MS).collect();
+                        self.ring.truncate(FRONTIER_ARTICLES);
+                        self.at = self.ring.len() % FRONTIER_ARTICLES;
+                        self.value = Some(m);
+                    }
+                    self.aside.clear();
+                }
+                return;
+            }
+            self.aside.clear();
+        }
+        if self.ring.len() < FRONTIER_ARTICLES {
+            self.ring.push(t);
+        } else {
+            self.ring[self.at] = t;
+        }
+        self.at = (self.at + 1) % FRONTIER_ARTICLES;
+        self.pending += 1;
+    }
+
+    /// Recomputes the frontier (once per chunk: a median of FRONTIER_ARTICLES isn't for every article).
+    pub fn update(&mut self) {
+        if self.pending == 0 || (self.value.is_none() && self.ring.len() < FRONTIER_MIN) {
+            return;
+        }
+        self.pending = 0;
+        let m = median(&mut self.ring.clone());
+        self.value = Some(m);
+        if self.start.is_none() {
+            self.start = Some(m);
+        }
+    }
+
+    /// At the block's end: whatever dates there are set the frontier (a small block never reached FRONTIER_MIN).
+    pub fn settle(&mut self) {
+        if self.value.is_none() && !self.ring.is_empty() {
+            self.pending = 1;
+            let m = median(&mut self.ring.clone());
+            self.value = Some(m);
+            self.start = Some(m);
+        } else {
+            self.update();
+        }
+    }
+
+    pub fn value(&self) -> Option<i64> {
+        self.value
+    }
+
+    pub fn start(&self) -> Option<i64> {
+        self.start
+    }
+
+    /// The posting time before which a last part means the window closed.
+    fn closed_before(&self) -> Option<i64> {
+        self.value.map(|f| f - WINDOW_MS)
+    }
+}
 
 struct OpenFile {
     parts: Vec<Seg>,
@@ -570,105 +664,97 @@ impl Aggregator {
         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>> {
+    /// Files the window closed on, still missing parts: those whose last part is before `closed_before`, undated ones
+    /// idle for `idle` batches (all when `all`). Their segments, part order.
+    fn evict(&mut self, closed_before: Option<i64>, idle: u64, all: bool) -> 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 {
+                let c = &copies[i];
+                let closed = match c.latest {
+                    Some(l) => closed_before.is_some_and(|before| l < before),
+                    None => idle > 0 && batch - c.touched >= idle,
+                };
+                if !all && !closed {
                     i += 1;
                     continue;
                 }
-                gone.extend(copies.remove(i).parts);
+                let mut f = copies.remove(i);
+                f.parts.sort_by_key(|s| s.number);
+                out.push(f.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();
+/// A release still open in the window: the files of one upload of its key (one after another, as they finished).
+struct OpenRelease {
+    upload: usize,
+    files: Vec<Vec<Seg>>,
+    first: Option<i64>,
+    last: Option<i64>,
+    touched: u64,
+}
+
+impl OpenRelease {
+    /// How far a file's posting span is from the release's (0 when either is undated).
+    fn distance(&self, first: Option<i64>, last: Option<i64>) -> i64 {
+        match (first, last, self.first, self.last) {
+            (Some(ff), Some(fl), Some(rf), Some(rl)) => 0.max(rf - fl).max(ff - rl),
+            _ => 0,
         }
-        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,
+    frontier: Frontier,
+    /// Per release key its open uploads, oldest first.
+    open: FxHashMap<String, Vec<OpenRelease>>,
+    /// The next upload number per key (distinct within a batch: the store's slot is the key and the upload).
+    uploads: FxHashMap<String, usize>,
     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>,
     /// Posting times of this batch's segments (posted_span at take).
     dates: Vec<i64>,
+    pending_files: usize,
 }
 
 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 {
+    /// `resume_at`: where this worker starts (a range taken over midway); its first batch starts there. `idle`: batches
+    /// (chunks) after which an undated file or release is given up.
+    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(),
+            frontier: Frontier::default(),
+            open: FxHashMap::default(),
+            uploads: FxHashMap::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(),
             dates: Vec::new(),
+            pending_files: 0,
         }
     }
 
@@ -699,7 +785,10 @@ impl RangeIndexer {
             });
         }
         self.batch.headers += headers;
-        self.dates.extend(segments.iter().filter_map(|s| s.posted_at));
+        for t in segments.iter().filter_map(|s| s.posted_at) {
+            self.dates.push(t);
+            self.frontier.add(t);
+        }
         for owner in crossposts {
             self.count("crosspost", owner, 1, 0);
         }
@@ -714,31 +803,36 @@ impl RangeIndexer {
         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;
-        }
+        self.frontier.update();
         let t = std::time::Instant::now();
-        let gone = self.aggregator.evict(self.idle);
-        self.cut_off(gone, false);
+        let gone = self.aggregator.evict(self.frontier.closed_before(), self.idle, false);
+        self.files(gone);
+        self.close(false);
         unsafe { PROF[4] += t.elapsed().as_nanos() as u64; }
         self.batch.next = next;
     }
 
+    /// The block's end: every file and release left is written, finished when the window closed on it, else
+    /// provisionally and as an edge (the block merger finishes it with the neighbour block's part).
     pub fn finish(&mut self) {
-        let gone = self.aggregator.evict(0);
-        self.cut_off(gone, true);
+        self.frontier.settle();
+        let gone = self.aggregator.evict(None, 0, true);
+        self.files(gone);
+        self.close(true);
         self.batch.next = self.end + 1;
     }
 
-    /// Files completed and fragments cut since the last take (RangeIndexer.pending).
+    /// Files in the releases finished since the last take (RangeIndexer.pending).
     pub fn pending(&self) -> usize {
-        self.batch.fragments.len() + self.completed.values().map(|l| l.len()).sum::<usize>()
+        self.pending_files
+    }
+
+    /// Where the block is in posting time (None until FRONTIER_MIN dated articles were read).
+    pub fn frontier(&self) -> Option<i64> {
+        self.frontier.value()
     }
 
     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() });
@@ -747,87 +841,115 @@ impl RangeIndexer {
             let (articles, files) = self.counts.remove(&id).unwrap();
             out.counts.push((id.0, id.1, articles, files));
         }
+        self.pending_files = 0;
+        // Upload numbers only have to be distinct within a batch: those of keys with nothing open go.
+        let open = &self.open;
+        self.uploads.retain(|key, _| open.contains_key(key));
         out
     }
 
+    /// Finished files (complete, or the window closed on them) into the open release of their key and upload.
     fn files(&mut self, files: Vec<Vec<Seg>>) {
+        let batch = self.aggregator.batch;
         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, poster: &f[0].file.poster })
-                    .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);
+            if file.len() == 1 && (file[0].number != 1 || file[0].file.total_parts > 1) && is_random_file(&file[0].file.name) {
+                // One article of a multi-part random file (its upload per article: ngPost's, nyuu's): not a release.
+                let tool = names::upload_tool(&file[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()),
+            let key = release_key_of(&file[0]);
+            let first = posted_of(&file);
+            let last = file.iter().filter_map(|s| s.posted_at).max();
+            let uploads = self.open.entry(key.clone()).or_default();
+            let near = uploads.iter_mut().filter(|u| u.distance(first, last) <= UPLOAD_GAP_MS).min_by_key(|u| u.distance(first, last));
+            let release = match near {
+                Some(u) => u,
+                None => {
+                    let n = self.uploads.entry(key).or_insert(0);
+                    let upload = *n;
+                    *n += 1;
+                    uploads.push(OpenRelease { upload, files: Vec::new(), first: None, last: None, touched: batch });
+                    uploads.last_mut().unwrap()
+                }
             };
-            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);
-                    }
+            if let Some(f) = first {
+                if release.first.is_none_or(|r| f < r) {
+                    release.first = Some(f);
+                }
+            }
+            if let Some(l) = last {
+                if release.last.is_none_or(|r| l > r) {
+                    release.last = Some(l);
+                }
+            }
+            release.touched = batch;
+            release.files.push(file);
+        }
+    }
+
+    /// Releases the window closed on (their last part before the frontier less WINDOW_MS; undated ones idle for
+    /// `idle` batches): judged and written. At the end (`all`), the rest too: provisionally, as edges.
+    fn close(&mut self, all: bool) {
+        let before = self.frontier.closed_before();
+        let start = self.frontier.start();
+        let batch = self.aggregator.batch;
+        let idle = self.idle;
+        let mut done: Vec<(String, OpenRelease, bool)> = Vec::new();
+        let keys: Vec<String> = self.open.keys().cloned().collect();
+        for key in keys {
+            let uploads = self.open.get_mut(&key).unwrap();
+            let mut i = 0;
+            while i < uploads.len() {
+                let u = &uploads[i];
+                let closed = match u.last {
+                    Some(l) => before.is_some_and(|b| l < b),
+                    None => idle > 0 && batch - u.touched >= idle,
+                };
+                if !all && !closed {
+                    i += 1;
+                    continue;
+                }
+                done.push((key.clone(), uploads.remove(i), closed));
+            }
+            if uploads.is_empty() {
+                self.open.remove(&key);
+            }
+        }
+        for (key, u, closed) in done {
+            // An edge: still open at the block's end, or posted within the window of the block's start (its other
+            // part may be in the neighbour block); undated ones can't be placed, so they are edges too.
+            let near_start = match (start, u.first) {
+                (Some(s), Some(f)) => f < s + WINDOW_MS,
+                _ => true,
+            };
+            self.emit(&key, u, closed, !closed || near_start);
+        }
+    }
+
+    fn emit(&mut self, key: &str, release: OpenRelease, finished: bool, edge: bool) {
+        let list = release.files;
+        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, poster: &f[0].file.poster })
+            .collect();
+        match names::judge(key, &judged, &self.rules) {
+            Verdict::Keep => {
+                let candidates = candidates_of(key, &list, release.upload);
+                let mut chunk = chunk_of(key, &list, release.upload);
+                chunk.finished = finished;
+                chunk.edge = edge;
+                self.pending_files += chunk.files;
+                self.batch.releases.push(chunk);
+                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, release.upload), rule, rule_key));
                 }
             }
-            self.batch.fragments.push(fragment);
         }
     }
 
@@ -904,6 +1026,9 @@ fn chunk_of(key: &str, files: &[Vec<Seg>], upload: usize) -> ReleaseChunk {
         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)),
+        finished: true,
+        edge: false,
+        partial: files.iter().any(|f| (f.len() as u32) < f[0].file.total_parts),
         payload: encode_chunk(&chunk_files),
         subjects: chunk_files.iter().take(SUBJECTS).map(|f| f.subject.to_string()).collect(),
     }
@@ -968,6 +1093,83 @@ mod tests {
         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);
+        // A day of posting in all: the window never closed on them, so at the end they are provisional edges.
+        assert!(batch.releases.iter().all(|r| !r.finished && r.edge && !r.partial));
+    }
+
+    /// `n` single-part filler files posted `minutes` apart from `start`: dated articles that move the frontier.
+    fn fillers(from: i64, n: i64, start: i64, minutes: i64) -> Vec<String> {
+        (0..n).map(|i| line(from + i, &format!(r#"Filler.{} [1/1] - "filler.{}.jpg" yEnc (1/1)"#, from + i, from + i), start + i * minutes)).collect()
+    }
+
+    #[test]
+    fn the_window_finishes_releases_and_keeps_incomplete_files() {
+        let mut priorities = FxHashMap::default();
+        priorities.insert(GROUP.to_string(), 1);
+        let mut ix = RangeIndexer::new(GROUP, 1, 1_000_000, 1, Rules::defaults(priorities), 5);
+        let day = 24 * 60;
+        // Chunk 1: a release at the block's start (two of three parts of its file), and 100 dated articles after it.
+        let mut lines = vec![
+            line(10, r#"Early.Show [1/1] - "Early.Show.mkv" yEnc (1/3)"#, 0),
+            line(11, r#"Early.Show [1/1] - "Early.Show.mkv" yEnc (2/3)"#, 1),
+        ];
+        lines.extend(fillers(100, 100, 2, 1));
+        ix.add_text(&lines.join("
+"), 1000);
+        assert!(ix.frontier().is_some());
+        let b1 = ix.take();
+        assert!(b1.releases.iter().all(|r| r.name.starts_with("Filler")));
+        // Chunk 2, three days on (more articles than before, so the median moves there): the window closed on the
+        // early show (its file still short a part) and on the first fillers.
+        let mut lines = vec![line(2000, r#"Later.Show [1/1] - "Later.Show.mkv" yEnc (1/1)"#, 3 * day)];
+        lines.extend(fillers(2100, 300, 3 * day + 1, 1));
+        ix.add_text(&lines.join("
+"), 3000);
+        let b2 = ix.take();
+        let early = b2.releases.iter().find(|r| r.name == "Early Show").expect("the early show is written");
+        assert!(early.finished && early.edge && early.partial, "finished {} edge {} partial {}", early.finished, early.edge, early.partial);
+        assert_eq!((early.files, early.segments), (1, 2));
+        assert!(!b2.releases.iter().any(|r| r.name == "Later Show"));
+        // Chunk 3, three more days: the later show is finished, and not an edge (a day past the block's start).
+        ix.add_text(&fillers(4000, 500, 6 * day, 1).join("
+"), 5000);
+        let b3 = ix.take();
+        let later = b3.releases.iter().find(|r| r.name == "Later Show").expect("the later show is written");
+        assert!(later.finished && !later.edge && !later.partial);
+        // The end: the last fillers are still inside the window, so they are provisional edges.
+        ix.finish();
+        let b4 = ix.take();
+        assert_eq!(b4.releases.len(), 500);
+        assert!(b4.releases.iter().all(|r| !r.finished && r.edge));
+        assert_eq!(b4.next, 1_000_001);
+    }
+
+    #[test]
+    fn the_frontier_ignores_implausible_dates_and_follows_a_pause() {
+        let hour = 3_600_000i64;
+        let mut f = Frontier::default();
+        for i in 0..200 {
+            f.add(i * hour / 10);
+        }
+        f.update();
+        let settled = f.value().unwrap();
+        assert!((settled - 10 * hour).abs() <= hour);
+        assert_eq!(f.start(), Some(settled));
+        // A bogus year, and one a year back: set aside.
+        for _ in 0..100 {
+            f.add(4_000_000_000_000);
+            f.add(-400 * 24 * hour);
+        }
+        f.update();
+        assert_eq!(f.value(), Some(settled));
+        // The group paused for a month: FRONTIER_ARTICLES agreeing dates move the frontier there.
+        let later = 30 * 24 * hour;
+        for i in 0..FRONTIER_ARTICLES as i64 {
+            f.add(later + i * 1000);
+        }
+        f.update();
+        assert!(f.value().unwrap() >= later);
+        assert_eq!(f.start(), Some(settled));
     }
 
     #[test]

edge/src/main.rs

@@ -123,7 +123,7 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
         line.push_str(",\"postedLast\":");
         opt_num(r.posted_last, line);
         line.push_str(&format!(
-            ",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"payload\":\"{}\"",
+            ",\"files\":{},\"segments\":{},\"bytes\":{},\"firstArticle\":{},\"lastArticle\":{},\"readable\":{},\"hasPar2\":{},\"finished\":{},\"edge\":{},\"partial\":{},\"payload\":\"{}\"",
             r.files,
             r.segments,
             r.bytes,
@@ -131,6 +131,9 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
             r.last_article,
             r.readable,
             r.has_par2,
+            r.finished,
+            r.edge,
+            r.partial,
             b64.encode(&r.payload)
         ));
     };
@@ -162,25 +165,6 @@ fn dump_batch(batch: &Batch, out: &mut impl Write) {
         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() {
@@ -208,7 +192,7 @@ fn main() {
 
     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 out = (0u64, 0usize, 0usize, 0usize, 0usize, 0usize, 0usize); // headers, releases, files, segments, payload, candidates, samples
     let mut keys = std::collections::HashSet::new();
     let mut decompressor = libdeflater::Decompressor::new();
     let wall = Instant::now();
@@ -242,8 +226,7 @@ fn main() {
             out.4 += r.payload.len();
         }
         out.5 += batch.candidates.len();
-        out.6 += batch.fragments.len();
-        out.7 += batch.samples.len();
+        out.6 += batch.samples.len();
         if let Some(d) = dump.as_mut() {
             dump_batch(&batch, d);
         }
@@ -269,7 +252,7 @@ fn main() {
     );
     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",
+        "  kept: {} release chunks ({} releases), {} files, {} segments, payload {:.1} MB ({:.1} B/segment); {} candidates, {} samples",
         out.1,
         keys.len(),
         out.2,
@@ -277,7 +260,6 @@ fn main() {
         out.4 as f64 / 1e6,
         out.4 as f64 / out.3.max(1) as f64,
         out.5,
-        out.6,
-        out.7
+        out.6
     );
 }

edge/src/wire.rs

@@ -1,7 +1,7 @@
 //! 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 crate::indexer::{Batch, Candidate, ReleaseChunk};
 use base64::Engine;
 use serde::Serialize;
 
@@ -24,6 +24,9 @@ struct WireRelease<'a> {
     last_article: i64,
     readable: bool,
     has_par2: bool,
+    finished: bool,
+    edge: bool,
+    partial: bool,
     payload: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     rule: Option<&'a str>,
@@ -49,30 +52,6 @@ struct WireCount<'a> {
     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)]
 #[serde(rename_all = "camelCase")]
 struct WireBatch<'a> {
@@ -85,7 +64,6 @@ struct WireBatch<'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> {
@@ -106,6 +84,9 @@ fn release<'a>(r: &'a ReleaseChunk, rule: Option<(&'a str, &'a str)>) -> WireRel
         last_article: r.last_article,
         readable: r.readable,
         has_par2: r.has_par2,
+        finished: r.finished,
+        edge: r.edge,
+        partial: r.partial,
         payload: base64::engine::general_purpose::STANDARD.encode(&r.payload),
         rule: rule.map(|(rule, _)| rule),
         rule_key: rule.map(|(_, key)| key),
@@ -116,21 +97,6 @@ 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 {
@@ -143,7 +109,6 @@ pub fn to_json(batch: &Batch) -> Vec<u8> {
         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()
 }