feat: stream newest pull requests by category

src/session.rs

@@ -26,6 +26,153 @@ use crate::{
 
 type DiffResult = (u64, Result<DiffOutcome, String>);
 
+#[derive(Clone, Copy)]
+struct PullRequestJob {
+    page: u32,
+    busy: bool,
+    complete: bool,
+    failed: bool,
+    retry: u64,
+}
+
+struct PullRequestWorker {
+    id: u64,
+    requests: Vec<Sender<u32>>,
+    threads: Vec<JoinHandle<()>>,
+    cancelled: Arc<AtomicBool>,
+    jobs: [PullRequestJob; 2],
+    priority: usize,
+}
+
+impl PullRequestWorker {
+    fn start(
+        id: u64,
+        root: &Path,
+        remote: String,
+        outgoing: Sender<DiffResult>,
+    ) -> Result<Self, String> {
+        let cancelled = Arc::new(AtomicBool::new(false));
+        let fetches = std::array::from_fn(|index| {
+            let root = root.to_path_buf();
+            let remote = remote.clone();
+            let cancelled = Arc::clone(&cancelled);
+            let mut repository = None;
+            move |page| {
+                if repository.is_none() {
+                    repository = Some(Repository::open(&root, Arc::clone(&cancelled))?);
+                }
+                repository
+                    .as_mut()
+                    .expect("repository opened")
+                    .pull_request_page(&remote, index == 1, page)
+            }
+        });
+        Self::start_with(id, outgoing, cancelled, fetches)
+    }
+
+    fn start_with(
+        id: u64,
+        outgoing: Sender<DiffResult>,
+        cancelled: Arc<AtomicBool>,
+        fetches: [impl FnMut(u32) -> Result<crate::diff::PullRequestPage, String> + Send + 'static;
+            2],
+    ) -> Result<Self, String> {
+        let mut worker = Self {
+            id,
+            requests: Vec::with_capacity(2),
+            threads: Vec::with_capacity(2),
+            cancelled,
+            jobs: [PullRequestJob {
+                page: 1,
+                busy: false,
+                complete: false,
+                failed: false,
+                retry: 0,
+            }; 2],
+            priority: 0,
+        };
+        for (index, mut fetch) in fetches.into_iter().enumerate() {
+            let (requests, incoming) = mpsc::channel();
+            let outgoing = outgoing.clone();
+            let cancelled = Arc::clone(&worker.cancelled);
+            let thread = thread::Builder::new()
+                .name(format!("vamp-pull-requests-{index}"))
+                .spawn(move || {
+                    while let Ok(page) = incoming.recv() {
+                        if cancelled.load(Ordering::Acquire) {
+                            break;
+                        }
+                        let result = fetch(page);
+                        if cancelled.load(Ordering::Acquire)
+                            || outgoing
+                                .send((
+                                    id,
+                                    Ok(DiffOutcome::PullRequestPage {
+                                        closed: index == 1,
+                                        result,
+                                    }),
+                                ))
+                                .is_err()
+                        {
+                            break;
+                        }
+                    }
+                })
+                .map_err(|error| format!("Cannot start pull request worker: {error}"))?;
+            worker.requests.push(requests);
+            worker.threads.push(thread);
+        }
+        worker.schedule()?;
+        Ok(worker)
+    }
+
+    fn update(&mut self, category: usize, retries: [u64; 2]) -> Result<(), String> {
+        self.priority = usize::from(category == 1);
+        for (job, retry) in self.jobs.iter_mut().zip(retries) {
+            if job.retry != retry {
+                job.retry = retry;
+                job.failed = false;
+            }
+        }
+        self.schedule()
+    }
+
+    fn complete(&mut self, closed: bool, result: &Result<crate::diff::PullRequestPage, String>) {
+        let job = &mut self.jobs[usize::from(closed)];
+        job.busy = false;
+        match result {
+            Ok(page) => {
+                job.complete = !page.has_more;
+                job.page += 1;
+            }
+            Err(_) => job.failed = true,
+        }
+    }
+
+    fn schedule(&mut self) -> Result<(), String> {
+        for index in [self.priority, 1 - self.priority] {
+            let job = &mut self.jobs[index];
+            if !job.busy && !job.complete && !job.failed {
+                self.requests[index]
+                    .send(job.page)
+                    .map_err(|_| "Pull request worker stopped unexpectedly")?;
+                job.busy = true;
+            }
+        }
+        Ok(())
+    }
+}
+
+impl Drop for PullRequestWorker {
+    fn drop(&mut self) {
+        self.cancelled.store(true, Ordering::Release);
+        self.requests.clear();
+        for thread in self.threads.drain(..) {
+            let _ = thread.join();
+        }
+    }
+}
+
 struct DiffWorker {
     requests: Option<Sender<(u64, DiffOperation)>>,
     results: Option<Receiver<DiffResult>>,
@@ -33,13 +180,18 @@ struct DiffWorker {
     stopped: Arc<AtomicBool>,
     cancelled_request: Arc<AtomicU64>,
     thread: Option<JoinHandle<()>>,
+    root: PathBuf,
+    outgoing: Sender<DiffResult>,
+    pull_requests: Option<PullRequestWorker>,
     pending: Option<u64>,
 }
 
 impl DiffWorker {
     fn start(root: PathBuf) -> Result<Self, String> {
         let (requests, incoming) = mpsc::channel::<(u64, DiffOperation)>();
-        let (outgoing, results) = mpsc::sync_channel(1);
+        let (outgoing, results) = mpsc::channel();
+        let sender = outgoing.clone();
+        let worker_root = root.clone();
         let cancelled = Arc::new(AtomicBool::new(false));
         let stopped = Arc::new(AtomicBool::new(false));
         let cancelled_request = Arc::new(AtomicU64::new(0));
@@ -63,16 +215,23 @@ impl DiffWorker {
                     }
                     let result = (|| {
                         if repository.is_none() {
-                            repository = Some(Repository::open(&root, Arc::clone(&cancellation))?);
+                            repository =
+                                Some(Repository::open(&worker_root, Arc::clone(&cancellation))?);
                         }
                         let repository = repository.as_mut().expect("repository opened");
                         match operation {
                             DiffOperation::Discover => {
                                 repository.catalog().map(DiffOutcome::Catalog)
                             }
-                            DiffOperation::PullRequests(remote) => {
-                                repository.pull_requests(&remote).map(DiffOutcome::Catalog)
+                            DiffOperation::PullRequests(_) => {
+                                unreachable!("listing uses category workers")
                             }
+                            DiffOperation::ResolvePullRequest(revision) => repository
+                                .resolve_pull_request(revision)
+                                .map(|(before, after)| DiffOutcome::ResolvedPullRequest {
+                                    before,
+                                    after,
+                                }),
                             DiffOperation::ListRemote(remote) => {
                                 repository.list_remote(&remote).map(DiffOutcome::Catalog)
                             }
@@ -119,15 +278,29 @@ impl DiffWorker {
             stopped,
             cancelled_request,
             thread: Some(thread),
+            root,
+            outgoing: sender,
+            pull_requests: None,
             pending: None,
         })
     }
 
     fn request(&mut self, id: u64, operation: DiffOperation) -> Result<(), String> {
+        self.pull_requests = None;
         if let Some(previous) = self.pending.take() {
             self.cancelled_request.store(previous, Ordering::Release);
             self.cancelled.store(true, Ordering::Release);
         }
+        if let DiffOperation::PullRequests(remote) = operation {
+            self.pull_requests = Some(PullRequestWorker::start(
+                id,
+                &self.root,
+                remote,
+                self.outgoing.clone(),
+            )?);
+            self.pending = Some(id);
+            return Ok(());
+        }
         self.requests
             .as_ref()
             .expect("live diff worker has a sender")
@@ -141,6 +314,7 @@ impl DiffWorker {
 impl Drop for DiffWorker {
     fn drop(&mut self) {
         self.stopped.store(true, Ordering::Release);
+        self.pull_requests = None;
         self.cancelled.store(true, Ordering::Release);
         self.requests.take();
         self.results.take();
@@ -401,6 +575,15 @@ impl EditorSession {
             let Some(diff) = self.editor.tab_diff(*tab) else {
                 return false;
             };
+            if let Some(listing) = &mut worker.pull_requests {
+                if let Err(error) = listing.update(diff.category, diff.pull_request_retries) {
+                    if let Some(id) = worker.pending.take() {
+                        self.editor.apply_diff_result(id, Err(error));
+                    }
+                    return false;
+                }
+                return true;
+            }
             if !diff.loading
                 && !diff.refresh_pending()
                 && let Some(id) = worker.pending.take()
@@ -447,12 +630,28 @@ impl EditorSession {
                     .try_recv();
                 match result {
                     Ok((id, result)) => {
-                        if worker.pending == Some(id) {
+                        if let Ok(DiffOutcome::PullRequestPage { closed, result }) = &result {
+                            if let Some(listing) = worker
+                                .pull_requests
+                                .as_mut()
+                                .filter(|listing| listing.id == id)
+                            {
+                                listing.complete(*closed, result);
+                            }
+                        } else if worker.pending == Some(id) {
                             worker.pending = None;
                         }
                         changed |= self.editor.apply_diff_result(id, result);
                     }
-                    Err(mpsc::TryRecvError::Empty) => return true,
+                    Err(mpsc::TryRecvError::Empty) => {
+                        if let Some(listing) = &mut worker.pull_requests
+                            && let Err(error) = listing.schedule()
+                        {
+                            changed |= self.editor.apply_diff_result(listing.id, Err(error));
+                            return false;
+                        }
+                        return true;
+                    }
                     Err(mpsc::TryRecvError::Disconnected) => {
                         if let Some(id) = worker.pending.take() {
                             changed |= self.editor.apply_diff_result(
@@ -1485,6 +1684,98 @@ mod tests {
         sync::atomic::{AtomicU64, Ordering},
     };
 
+    #[test]
+    fn pull_request_pages_stream_concurrently_and_retry_only_the_failed_category() {
+        use crate::diff::{PullRequestPage, RepoKind};
+        let (outgoing, results) = mpsc::channel();
+        let (started, starts) = mpsc::channel();
+        let (open, open_gate) = mpsc::channel::<Result<bool, String>>();
+        let (closed, closed_gate) = mpsc::channel::<Result<bool, String>>();
+        let mut gates = [Some(open_gate), Some(closed_gate)];
+        let cancelled = Arc::new(AtomicBool::new(false));
+        let fetches = std::array::from_fn(|index| {
+            let gate = gates[index].take().unwrap();
+            let started = started.clone();
+            let cancelled = Arc::clone(&cancelled);
+            move |page| {
+                started.send((index, page)).unwrap();
+                let has_more = loop {
+                    match gate.try_recv() {
+                        Ok(result) => break result?,
+                        Err(mpsc::TryRecvError::Empty) if !cancelled.load(Ordering::Acquire) => {
+                            thread::yield_now()
+                        }
+                        Err(_) => return Err("cancelled".into()),
+                    }
+                };
+                Ok(PullRequestPage {
+                    kind: RepoKind::Git,
+                    remote: "origin".into(),
+                    entries: vec![],
+                    has_more,
+                })
+            }
+        });
+        let mut worker = PullRequestWorker::start_with(7, outgoing, cancelled, fetches).unwrap();
+        let mut first = [
+            starts.recv_timeout(Duration::from_secs(2)).unwrap(),
+            starts.recv_timeout(Duration::from_secs(2)).unwrap(),
+        ];
+        first.sort_unstable();
+        assert_eq!(first, [(0, 1), (1, 1)]);
+        let receive = |worker: &mut PullRequestWorker| {
+            let (id, result) = results.recv_timeout(Duration::from_secs(2)).unwrap();
+            assert_eq!(id, 7);
+            let Ok(DiffOutcome::PullRequestPage { closed, result }) = result else {
+                panic!("expected independently published page");
+            };
+            worker.complete(closed, &result);
+            (closed, result)
+        };
+        closed.send(Ok(true)).unwrap();
+        let (category, page) = receive(&mut worker);
+        assert!(category && page.unwrap().has_more);
+        // Open's first request remains blocked while the active Closed category advances.
+        worker.update(1, [0, 0]).unwrap();
+        assert_eq!(starts.recv_timeout(Duration::from_secs(2)).unwrap(), (1, 2));
+        closed.send(Err("late failure".into())).unwrap();
+        assert_eq!(receive(&mut worker).1.unwrap_err(), "late failure");
+        worker.update(1, [0, 0]).unwrap();
+        assert!(starts.try_recv().is_err());
+        worker.update(1, [0, 1]).unwrap();
+        assert_eq!(starts.recv_timeout(Duration::from_secs(2)).unwrap(), (1, 2));
+        closed.send(Ok(false)).unwrap();
+        assert!(!receive(&mut worker).1.unwrap().has_more);
+        open.send(Ok(false)).unwrap();
+        assert!(!receive(&mut worker).0);
+        worker.update(2, [0, 1]).unwrap();
+        assert!(starts.try_recv().is_err());
+    }
+
+    #[test]
+    fn closing_pull_request_listing_cancels_both_outstanding_reads() {
+        let (outgoing, _results) = mpsc::channel();
+        let (started, starts) = mpsc::channel();
+        let cancelled = Arc::new(AtomicBool::new(false));
+        let fetches = std::array::from_fn(|index| {
+            let started = started.clone();
+            let cancelled = Arc::clone(&cancelled);
+            move |_| {
+                started.send(index).unwrap();
+                while !cancelled.load(Ordering::Acquire) {
+                    thread::yield_now();
+                }
+                Err("cancelled".into())
+            }
+        });
+        let worker =
+            PullRequestWorker::start_with(1, outgoing, Arc::clone(&cancelled), fetches).unwrap();
+        starts.recv_timeout(Duration::from_secs(2)).unwrap();
+        starts.recv_timeout(Duration::from_secs(2)).unwrap();
+        drop(worker);
+        assert!(cancelled.load(Ordering::Acquire));
+    }
+
     #[test]
     fn normal_picker_tracks_unsaved_text_and_head_with_line_gutter_disabled() {
         use crate::diff::FileStatus;