rtb_vcs/git/
walk.rs

1//! `Repo::walk` — async commit-graph stream.
2//!
3//! v0.5 commit 2 of 7. Walks the commit graph reachable from a
4//! revspec (`HEAD`, `v0.4.0..HEAD`, etc.) and surfaces each commit
5//! as a [`CommitInfo`] value through an async [`Stream`] so very
6//! long histories don't materialise as a single `Vec`.
7//!
8//! The blocking gix walk runs on a `tokio::task::spawn_blocking`
9//! task that pipes commits through an `mpsc::channel` to the
10//! [`CommitWalk`] stream wrapper. Buffer capacity is 64 entries —
11//! enough to keep the gix side busy without unbounded memory growth.
12
13use std::pin::Pin;
14use std::task::{Context, Poll};
15
16use futures_core::Stream;
17use tokio::sync::mpsc;
18
19use super::{Repo, RepoError};
20
21/// Per-commit data surfaced by [`CommitWalk`].
22///
23/// Fields are owned (no gix references) so consumers can collect
24/// the stream across task boundaries.
25#[derive(Debug, Clone)]
26#[non_exhaustive]
27pub struct CommitInfo {
28    /// Commit object id in hex.
29    pub id: String,
30    /// First line of the commit message (the conventional summary).
31    pub summary: String,
32    /// Full commit message, including the body.
33    pub message: String,
34    /// Author display name.
35    pub author_name: String,
36    /// Author email address.
37    pub author_email: String,
38    /// Author timestamp in seconds since the Unix epoch. `0` when
39    /// gix could not parse the timestamp (rare on well-formed
40    /// repositories).
41    pub time_seconds: i64,
42}
43
44/// Async stream over the commits matched by a [`Repo::walk`] call.
45///
46/// Consume via the `futures::StreamExt` extension trait; the
47/// underlying gix walk runs on a `spawn_blocking` task that's
48/// released when this stream is dropped.
49pub struct CommitWalk {
50    rx: mpsc::Receiver<Result<CommitInfo, RepoError>>,
51}
52
53impl std::fmt::Debug for CommitWalk {
54    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        f.debug_struct("CommitWalk").field("rx", &"<mpsc::Receiver>").finish()
56    }
57}
58
59impl Stream for CommitWalk {
60    type Item = Result<CommitInfo, RepoError>;
61
62    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
63        self.rx.poll_recv(cx)
64    }
65}
66
67impl Repo {
68    /// Walk the commit graph from `revspec`.
69    ///
70    /// Supports `Include` (`HEAD`), `Range` (`A..B`), and `Merge`
71    /// (`A...B`) revspecs. The remaining gix-revision kinds
72    /// (`IncludeOnlyParents`, `ExcludeParents`, etc.) are reported
73    /// as [`RepoError::WalkFailed`] — they're rare and we add
74    /// support when a real consumer asks.
75    ///
76    /// # Errors
77    ///
78    /// - [`RepoError::RevspecNotFound`] — `revspec` did not resolve.
79    /// - [`RepoError::WalkFailed`] — gix could not initialise the
80    ///   walk, or `revspec` is a kind we don't yet support.
81    ///
82    /// Errors during walking (individual commits) are surfaced
83    /// in-band on the returned stream.
84    pub fn walk(&self, revspec: &str) -> Result<CommitWalk, RepoError> {
85        let inner = self.thread_safe();
86        let (tips, boundary) = {
87            let repo = inner.to_thread_local();
88            let spec = repo
89                .rev_parse(revspec)
90                .map_err(|_| RepoError::RevspecNotFound { revspec: revspec.to_string() })?;
91            tips_and_boundary(spec.detach())?
92        };
93
94        let (tx, rx) = mpsc::channel(64);
95        let inner_for_task = inner;
96        tokio::task::spawn_blocking(move || {
97            run_walk(&inner_for_task, tips, boundary, &tx);
98        });
99        Ok(CommitWalk { rx })
100    }
101}
102
103fn tips_and_boundary(
104    spec: gix::revision::plumbing::Spec,
105) -> Result<(Vec<gix::ObjectId>, Vec<gix::ObjectId>), RepoError> {
106    use gix::revision::plumbing::Spec as S;
107    match spec {
108        S::Include(oid) => Ok((vec![oid], vec![])),
109        S::Range { from, to } => Ok((vec![to], vec![from])),
110        S::Merge { theirs, ours } => Ok((vec![theirs, ours], vec![])),
111        other => {
112            Err(RepoError::WalkFailed { cause: format!("unsupported revspec kind: {other:?}") })
113        }
114    }
115}
116
117fn run_walk(
118    inner: &gix::ThreadSafeRepository,
119    tips: Vec<gix::ObjectId>,
120    boundary: Vec<gix::ObjectId>,
121    tx: &mpsc::Sender<Result<CommitInfo, RepoError>>,
122) {
123    let repo = inner.to_thread_local();
124    let mut platform = repo.rev_walk(tips);
125    if !boundary.is_empty() {
126        platform = platform.with_boundary(boundary);
127    }
128    let walk = match platform.all() {
129        Ok(w) => w,
130        Err(e) => {
131            let _ = tx
132                .blocking_send(Err(RepoError::WalkFailed { cause: format!("rev_walk init: {e}") }));
133            return;
134        }
135    };
136    for item in walk {
137        let info = match item {
138            Ok(i) => i,
139            Err(e) => {
140                let _ = tx
141                    .blocking_send(Err(RepoError::WalkFailed { cause: format!("walk item: {e}") }));
142                return;
143            }
144        };
145        let commit = match info.object() {
146            Ok(c) => c,
147            Err(e) => {
148                let _ = tx.blocking_send(Err(RepoError::WalkFailed {
149                    cause: format!("commit object: {e}"),
150                }));
151                return;
152            }
153        };
154        let payload = match make_commit_info(&commit) {
155            Ok(p) => p,
156            Err(e) => {
157                let _ = tx.blocking_send(Err(e));
158                return;
159            }
160        };
161        if tx.blocking_send(Ok(payload)).is_err() {
162            // Receiver dropped — stream consumer is gone; abandon walk.
163            return;
164        }
165    }
166}
167
168fn make_commit_info(commit: &gix::Commit<'_>) -> Result<CommitInfo, RepoError> {
169    let id = commit.id().to_string();
170
171    let message_ref = commit
172        .message()
173        .map_err(|e| RepoError::WalkFailed { cause: format!("commit message: {e}") })?;
174    let summary = message_ref.summary().to_string();
175    let message_full = message_ref.body.map_or_else(
176        || message_ref.title.to_string(),
177        |body| format!("{}\n\n{}", message_ref.title, body),
178    );
179
180    let author = commit
181        .author()
182        .map_err(|e| RepoError::WalkFailed { cause: format!("commit author: {e}") })?;
183    let author_name = author.name.to_string();
184    let author_email = author.email.to_string();
185    let time_seconds = author.time().ok().map_or(0, |t| t.seconds);
186
187    Ok(CommitInfo { id, summary, message: message_full, author_name, author_email, time_seconds })
188}
rtb_vcs/git/walk.rs

rtb_vcs/git/
walk.rs
